On this page

agent-gatewayのRAGとイベント処理

OpenAI互換API、NATS、Dagster、pgvectorで組んだLLM・RAG基盤の記録です。LLM導入に向け、応答と保存の分離、埋め込みサービス、障害時の動作を整理しました。

技術メモとして残していた ObsidianノートにAIの要約を利用しています。

OpenAI互換APIとRAGの基盤

agent-gatewayは、LLMとRAGをOpenAI互換HTTP APIで提供するゲートウェイです。homelabのAI基盤として、3月初頭から中旬の約2週間でPhase 1を組みました。LLM導入で必要になる推論、検索、記録の処理を分けています。

この後、v3 リデザインでドメインを分割し、全レイヤーリファクタでClean Architectureを適用しました。ここでは、その前の構成を扱います。

アーキテクチャの全体像

Go + Ginのレイヤードアーキテクチャで、各層の責務を分けました。

  Transport Layer (Gin) [internal/transport/http/]
    -- middleware: RequestContext -> Logger -> Recovery
    -- v1/ handlers (OpenAI 互換エンドポイント)

Presentation Layer [pkg/openai/]
    -- OpenAI 互換 / Anthropic 互換 request/response DTO

Domain Layer [internal/domain/]
    |-- knowledge/  RAG オーケストレーション
    |-- llm/        マルチバックエンド LLM ルーティング
    |-- vectorstore/ ベクトル検索
    +-- pipeline/   パイプライン連携 (NATS pub/sub, Dagster トリガー)

Infrastructure Layer [internal/infra/]
    |-- vllm/       vLLM クライアント
    |-- llamacpp/   llama.cpp クライアント
    |-- lmstudio/   LM Studio クライアント
    |-- postgres/   PostgreSQL + pgvector
    |-- nats/       NATS JetStream pub/sub
    |-- dagster/    Dagster ジョブトリガー
    +-- reranker/   ColBERT リランカークライアント

3つのLLMバックエンドを設定で切り替え、用途に応じてGPU推論とCPU推論を使います。

バックエンド	ホスト	対応モデル例
vLLM	compute.home.arpa:8000	qwen3-next:80b, qwen3-coder:30b
llama.cpp	compute.home.arpa:8081	nemotron-3-nano:30b
LM Studio	compute.home.arpa:1234	lfm2.5-1.2b-instruct-mlx

40種以上のモデルをカタログで管理し、名前のパターンで振り分けます。qwen3-はvLLM、nemotron-はllama.cppです。

NATSでリクエスト処理と保存を分ける

chat completionの応答を、データ保存の完了待ちにしない構成にしました。

応答後にNATSへ非同期でpublishし、Dagsterがイベントを受け取って保存します。gateway側はfire-and-forgetです。

  ユーザ -> gateway -> LLM バックエンド -> レスポンス返却
                  \-> NATS publish (fire-and-forget)
                        \-> Dagster sensor -> JetStream pull -> PostgreSQL

publish するトピック:

  pipeline.knowledge.chat.persist    -- チャット永続化
pipeline.knowledge.embedding       -- 埋め込み生成イベント
pipeline.knowledge.retrieve        -- ベクトル検索イベント
pipeline.knowledge.flow.lineage    -- フロー DAG リネージ追跡
pipeline.knowledge.tool_call       -- ツールコールログ

テレメトリは別経路です。VectorがNATSをsubscribeし、PrometheusとLokiに転送します。

Dagster 側の NATS パイプライン

model-foundryのDagsterに、NATS JetStream consumerを使うアセットとジョブを実装しました。

センサー: NATS JetStream の durable consumer でトピックをポーリング
アセット: 受け取ったイベントを DuckDB テーブルとして materialization
ジョブ: knowledge_chat_persist、knowledge_embedding、knowledge_lineage

JetStreamのpull consumerは再配送する場合があるため、保存側で冪等性を確保します。

  INSERT INTO chat_pairs (correlation_id, prompt, response, model, created_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (correlation_id) DO NOTHING;

RAG オーケストレーションフロー

chat completion リクエストに対する RAG の流れ:

ユーザクエリ抽出
エンベディング生成（multi-bert-inference gRPC）
pgvector でベクトル検索 + ColBERT リランキング
検索結果をコンテキストとして付与
LLM バックエンドへ送信
NATS にパイプラインイベント発行
レスポンス返却

PostgreSQL + pgvectorを使います。document_chunksにはembedding vector(256)と、cosine距離のHNSWインデックスを置きました。

  CREATE TABLE document_chunks (
    id SERIAL PRIMARY KEY,
    document_id TEXT NOT NULL,
    chunk_index INT NOT NULL,
    content TEXT NOT NULL,
    embedding vector(256),
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_chunks_embedding ON document_chunks
    USING hnsw (embedding vector_cosine_ops);

cache hitでは既存の埋め込みを検索してコンテキストを付けます。missでは埋め込みを生成・保存してからLLMに渡します。

multi-bert-inference のコンテナ化

埋め込みとリランキングは、RustのgRPCサーバーmulti-bert-inferenceが担当します。desktop.home.arpaの単独プロセスから、devstackに組み込むコンテナへ変更しました。

3ステージ Containerfile

  # Stage 1: 依存のみをビルドしてキャッシュ
FROM rust:1.94.0-slim-trixie AS deps
RUN apt-get update && apt-get install -y \
    protobuf-compiler pkg-config libssl-dev g++
WORKDIR /app
COPY Cargo.toml Cargo.lock build.rs proto/ ./
RUN mkdir src && echo "fn main() {}" > src/main.rs && \
    cargo build --release && rm -rf src

# Stage 2: 実際のソースをビルド
FROM deps AS builder
COPY src/ src/
RUN cargo build --release && strip target/release/multi-bert-inference

# Stage 3: 最小ランタイム
FROM debian:trixie-slim
RUN apt-get update && apt-get install -y ca-certificates curl && \
    rm -rf /var/lib/apt/lists/*
RUN useradd -r -s /bin/false app
USER app
COPY --from=builder /app/target/release/multi-bert-inference /usr/local/bin/
EXPOSE 50051 3000
ENTRYPOINT ["multi-bert-inference"]

Stage 1ではダミーのmain.rsでCargo依存を先にビルドし、ソース変更時にキャッシュを使います。数GBのモデルはイメージに含めず、ボリュームで渡します。モデル更新でイメージを作り直さずに済むためです。

  podman build -t desktop.home.arpa/multi-bert-inference .

devstack への組み込み

devstack/desktop/podman-compose.yml に inference サービスとして追加:

  inference:
  image: desktop.home.arpa/multi-bert-inference
  ports:
    - "50051:50051"
    - "3800:3000"
  volumes:
    - /Users/ksh3/Development/multi-bert-inference/models:/app/models:ro,z
  environment:
    RUST_LOG: multi_bert_inference=info
  healthcheck:
    test: ["CMD-SHELL", "curl -sf http://localhost:3000/healthz || exit 1"]
    interval: 10s
    timeout: 3s
    retries: 3
  restart: unless-stopped

gRPCの50051とRESTヘルスチェックの3000を公開し、後者はホストの3800に割り当てます。モデルディレクトリはread-onlyです。

推論サービスに接続できない場合

internal/infra/inference/client.goで接続を管理し、3つのクライアントを提供します。

EmbeddingServiceClient: Embed()（256次元）と EmbedContext()（late chunking）
RerankServiceClient: Rerank()（ColBERT MaxSim）
SearchServiceClient: MaxSimSearch()

推論サービスへの接続は任意です。接続できなくてもgatewayは起動します。

  var inferClient *inference.Client
if strings.TrimSpace(cfg.InferenceGRPCAddr) != "" {
    var inferErr error
    inferClient, inferErr = inference.NewClient(ctx, cfg.InferenceGRPCAddr)
    if inferErr != nil {
        slog.Warn("inference gRPC unavailable", "err", inferErr, "addr", cfg.InferenceGRPCAddr)
    } else {
        slog.Info("inference gRPC connected", "addr", cfg.InferenceGRPCAddr)
        orc.SetRerankerRepository(inference.NewRerankRepository(inferClient))
        defer inferClient.Close()
    }
}

停止時はembed/rerankだけを無効にし、LLMプロキシと他のエンドポイントは動かします。開発時に全サービスを起動する必要もありません。

Dagster からの埋め込み呼び出し

Dagsterも同じgRPCを使います。EmbeddingResourceでprotobufを直接エンコード・デコードし、/search.EmbeddingService/Embedを呼びます。knowledge_eventsとchat_pairsには256次元の埋め込みを生成します。

col-bert-apiのINT8エラー

並行して、candle-onnxとaxumによるcol-bert-apiも動かしました。ここではINT8モデルの推論に失敗しました。

onnx/model_int8.onnxは読み込めても実行時にエラーになり、同じコードでonnx/model.onnxのFP32版は動きました。INT8が要求するi8/VNNI系命令にCPUが対応していなかったためです。

  // 変更前: INT8 優先フォールバック
// 1. onnx/model_int8.onnx -> 2. onnx/model_fp16.onnx -> 3. onnx/model.onnx

// 変更後: FP32 固定
let model = candle_onnx::read_file("onnx/model.onnx")?;

READMEの「Prefer provided INT8 models if available」もFP32を使う記述に変更しました。量子化モデルは、読み込みだけでなく対象CPUでの推論確認が必要です。

Phase 1 で完成した構成

実装済みエンドポイント

エンドポイント	状態
`POST /v1/chat/completions`	完了
`POST /v1/messages` (Anthropic API)	完了
`POST /v1/responses` + GET/DELETE	完了
`GET /v1/models` + `GET /v1/models/:id`	完了
`POST /v1/embeddings`	完了
`POST /v1/moderations`	完了
`GET /healthz`	完了

LLM パーサーチェーン

LLM出力は4段のパーサーで正規化します。

ApplyToolCallParser – ツールコール正規化
ApplyReasoningParser – thinking/推論タグ抽出
ApplyVisionParser – 画像 base64 処理
ApplyOutputParser – 出力フォーマット正規化

ミドルウェア

RequestContext -> Logger -> Recoveryの順で処理します。RequestContextはX-Correlation-IDを取得または生成し、以降のログに結びます。

サービストポロジ

  desktop.home.arpa
  -- agent-gateway :8080
  -- NATS :4222
  -- multi-bert-inference :50051

compute.home.arpa
  -- vLLM :8000
  -- llama.cpp :8081
  -- LM Studio :1234
  -- PostgreSQL :5432
  -- Dagster :3300

storage.home.arpa
  -- Prometheus :9090
  -- Loki :3100
  -- MinIO :9000

Phase 1では、次の処理を接続しました。

gateway -> NATS publish (fire-and-forget)
Dagster sensor -> JetStream pull -> PostgreSQL
Vector -> NATS subscribe -> Prometheus/Loki
pgvector ANN 検索 + ColBERT リランク

続くv3 リデザインでは、ドメインの分割を見直します。

llm-jpの翻訳を常駐からバッチへ

llm-jpのNVFP4を単GPUで測り、翻訳ロールをオンデ …

Gemma 4とBlackwell 2枚の推論基盤

Gemma …