Go・NATS・DagsterのAI実行基盤
Goの互換API、NATS JetStream、Dagster sensorでLLMを業務パイプラインにつなぎました。AIシステム開発向けに、RAG、イベント、3ホストの構成を説明します。
構築した基盤
LLM を業務パイプラインにつなぐため、Go で OpenAI/Anthropic 互換 API ゲートウェイを構築しました。リクエストの RAG 処理と、NATS JetStream への fire-and-forget publish を担当します。Dagster daemon の sensor がイベントを pull してジョブを実行し、Vector(Rust) が NATS から telemetry を subscribe して Prometheus/Loki に送ります。
gateway は publish のみを担当し、各ミドルウェアが独立して consume します。gateway から Dagster への依存を外し、別々にデプロイできる構成にしました。
初期の Rust(axum) プロトタイプから Go に移行しました。SSE 中継、NATS 購読、PG/Qdrant 書き込みを同時に扱うコードを、goroutine + channel で整理するためです。
使用した構成
- 言語: Go 1.25, Gin フレームワーク
- メッセージング: NATS 2.11 (JetStream有効)
- オーケストレーション: Dagster (Python) — daemon + sensor + asset materialization
- データストア: PostgreSQL 18 + pgvector (JIT有効)
- テレメトリ: Vector 0.45 (Rust) → Prometheus + Loki + Grafana
- 推論バックエンド: vLLM, llama.cpp, LM Studio
- エンベディング + リランカー: multi-bert-inference — Rust (Axum) + ONNX Runtime (INT8, ~17MB, ~10ms)。ColBERT 256d embed + 64d token MaxSim rerank。DB依存排除、gRPCで256d vec直接送信（~1KB/candidate）
- コンテナ: Podman rootless + podman-compose
- ホスト構成: 3台 (storage server / desktop mac / compute server)
設計判断
レイヤードアーキテクチャ
Clean Architecture の依存方向を前提に、Transport、Domain、Infra の3層に分けました。
Client (opencode CLI, API consumer)
|
v
Transport Layer (Gin) [internal/transport/http/]
|-- middleware: RequestContext -> Logger -> Recovery
|-- v1/ handlers (OpenAI/Anthropic 互換エンドポイント)
|-- リクエスト解析・バリデーション
+-- Presentation DTO への変換
|
| Presentation Layer [pkg/openai/, pkg/anthropic/]
| +-- OpenAI/Anthropic 互換 request/response DTO
|
v
Domain Layer [internal/domain/]
|-- knowledge/ RAG オーケストレーション
|-- llm/ マルチバックエンド LLM ルーティング
|-- vectorstore/ ベクトル検索 (pgvector ANN)
+-- pipeline/ パイプライン連携 (NATS publish, fire-and-forget)
|
v
Infrastructure Layer [internal/infra/]
|-- vllm/ vLLM クライアント
|-- llamacpp/ llama.cpp クライアント
|-- lmstudio/ LM Studio クライアント
|-- postgres/ PostgreSQL + pgvector
|-- nats/ NATS JetStream publish
|-- vector/ Vector テレメトリ publish
+-- reranker/ multi-bert-inference gRPC クライアント
Domain 層は外部依存を持ちません。LLM バックエンドやデータストアの変更は Infra 層で扱います。業務システム開発でも、業務ロジックと接続先を分けるための構成です。
gatewayのpublish-only設計
gateway は NATS に publish し、subscribe や trigger は行いません。Dagster sensor と Vector が独立して pull / subscribe します。
この分担で、次の点を狙いました。
- gateway に Dagster を依存させず、デプロイを分ける
- gateway を変えずに消費側を追加・変更する
- gateway とパイプラインの障害を分離する
JetStreamを読むDagster sensor
Dagster daemon は durable consumer として JetStream を pull subscribe します。sensor → RunRequest → ジョブ実行に統一し、GraphQL trigger bridge を外しました。
なぜ Dagster (Python) か
Dagster は実行順と状態を管理します。重い計算は Python の外で動かします。
| 実際の計算 | 実行エンジン | Python/Dagster の役割 |
|---|---|---|
| LLM 推論 | vLLM / llama.cpp (C++/CUDA) | API 呼び出しのみ |
| ベクトル検索 | PostgreSQL + pgvector (C) | SQL 発行のみ |
| エンベディング + リランキング | multi-bert-inference (Rust/ONNX Runtime) | gRPC 呼び出しのみ |
| テレメトリ変換 | Vector (Rust) | YAML 設定のみ |
この構成では、Dagster の lineage tracking、sensor、asset materialization を採用しました。Rust 製の代替は見つからず、I/O wait と状態管理が中心の処理では、CPU bound 処理やメモリ安全という Rust の強みより、これらの機能を優先しました。
LLM バックエンド選定
指定されたモデル名でバックエンドを選びます。
| バックエンド | デフォルト URL | モデル例 |
|---|---|---|
| vLLM | http://$COMPUTE_HOST:8000 | qwen3-next:80b, qwen3-coder:30b |
| llama.cpp | http://$COMPUTE_HOST:8081 | nemotron-3-nano:30b |
| LM Studio | http://$COMPUTE_HOST:1234 | lfm2.5-1.2b-instruct-mlx |
LLM ルーティングは gateway の goroutine が担当します。回答内容に応じて次のバックエンドを決めるには状態を持つ制御が必要なので、Vector には任せていません。
モデルの性能と品質を継続して評価しています。検証済みモデルは GGUF、NVFP4、ONNX などの形式で、storage server のコールドストレージに保管します。
Rust から Go への移行
初期には Rust(axum)でプロトタイプを実装しました。型による設計は有用でしたが、非同期処理の記述量が増えました。
- SSEの中継 + NATSの購読 + PG/pgvectorへの書き込みが同時に走る非同期コンテキストの管理がRustでは冗長
- goroutine + channelの方がこのパターンに素直にフィットする
- 実行時オーバーヘッドの差はこのユースケースでは無視できるレベル
Go への移行後も、OpenAI 互換エンドポイント、NATS Pub/Sub、trace_id 付きログ、冪等設計を維持しました。Rust 版の設計文書を仕様として使いました。
実装
エンドポイント
| Method | Path | 説明 |
|---|---|---|
| POST | /v1/chat/completions | チャット補完 (OpenAI互換) |
| POST | /v1/messages | Anthropic Messages API |
| POST | /v1/responses | Responses API |
| POST | /v1/embeddings | エンベディング |
| GET | /v1/models | モデル一覧 |
| GET | /healthz | ヘルスチェック |
ミドルウェアチェーン
Gin ミドルウェア: RequestContext → Logger → Recovery
- RequestContext —
X-Correlation-IDまたはX-Request-IDヘッダからリクエスト追跡IDを抽出。未指定時はタイムスタンプベースのIDを自動生成し、コンテキストとレスポンスヘッダに設定 - Logger — HTTPリクエストごとに method, path, status, duration を
slogで記録 - Recovery — Gin 標準のパニックリカバリ
RAG データフロー
Client
|
v
Chat Completion Request
|
v
Knowledge Service (RAG オーケストレーション)
|-- 1. ユーザクエリ抽出
|-- 2. エンベディング生成 (multi-bert-inference -> 256d ColBERT)
|-- 3. ベクトル検索 (pgvector HNSW ANN) + リランキング (ColBERT MaxSim)
|-- 4. コンテキスト付与
|-- 5. LLM バックエンドへ送信
|-- 6. NATS publish (pipeline.* + telemetry.*) <- fire-and-forget
+-- 7. レスポンス返却
システムトポロジー
NATS JetStream :4222
+----------------------+
| pipeline.* |
| telemetry.* |
+---+----------+--------+
publish ------+ | pull subscribe
(fire & forget) | (durable consumer)
|
+-- agent-gateway :8080 ------+ +-- Dagster daemon -----------------------+
| | | |
| Client Request | | sensor: nats_dagster_chat_persist |
| | | | sensor: nats_dagster_embedding |
| v | | sensor: nats_dagster_retrieve |
| Knowledge Service | | sensor: nats_dagster_compose |
| |-- embed (multi-bert) | | sensor: nats_dagster_flow_lineage |
| |-- vector search (pg) | | sensor: nats_dagster_tool_call |
| |-- rerank (ColBERT) | | | |
| |-- LLM inference -------|-->-| vLLM / llama.cpp / LM Studio |
| | | | | |
| |-- NATS pub: pipeline.* | | v |
| |-- NATS pub: telemetry.*| | RunRequest -> Dagster job 実行 |
| +-- Response | | |-- chat persist -> pg |
| | | |-- embedding event -> pg |
| LLM Backends | | |-- retrieve event -> pg |
| |-- vLLM :8000 | | |-- compose event -> pg |
| |-- llama.cpp :8081 | | |-- flow lineage -> pg |
| +-- LM Studio :1234 | | +-- tool call -> pg |
| | | |
+-----------------------------+ +-----------------------------------------+
|
+------------+
| pull subscribe
v
+-- Vector (Rust) ----------------------+
| source: nats (telemetry.*) |
| |-- sink: Prometheus exporter |
| +-- sink: Loki (CorrelationID) |
+--------------------------------------+
NATS イベント設計
同じ CorrelationID を、独立した2つの経路で使います。
Pipeline (Dagster sensor → ジョブ実行):
| トピック | Dagster ジョブ |
|---|---|
pipeline.knowledge.chat.persist | knowledge_chat_persist + knowledge_chat_pair_materialize |
pipeline.knowledge.embedding | knowledge_embedding |
pipeline.knowledge.retrieve | knowledge_retrieve |
pipeline.knowledge.compose | knowledge_compose |
pipeline.knowledge.flow.lineage | knowledge_flow_lineage |
pipeline.knowledge.tool_call | knowledge_tool_call |
Telemetry (Vector → Prometheus / Loki):
| トピック | 内容 |
|---|---|
telemetry.knowledge.chat | モデル, バックエンド, トークン使用量, ストリーミング有無 |
telemetry.knowledge.embedding | モデル, 入力数, 次元数 |
telemetry.knowledge.retrieve | TopK, ヒット数, スコア |
telemetry.knowledge.tool_call | 関数名, モデル |
JetStreamストリーム設定:
PIPELINE— subjects:pipeline.>, retention: limits, max-age: 72h, storage: fileTELEMETRY— subjects:telemetry.>, retention: limits, max-age: 24h, storage: file
Dagster Assets と Jobs
Dagster は asset materialization と sensor で構成しています。
sensor が NATS JetStream を pull subscribe し、イベントに応じて RunRequest を発行します。GraphQL API を経由せずジョブを実行します。
tool_call_record には、例えば function_name: tokei、model: minimax-m2.5 を記録します。モデルが呼んだツールを Dagster の lineage で追えます。
knowledge_chat_pair_materialize は chat_pair_record と chat_pair_dataset の2つのアセットを materialize します。prompt-response ペアを PostgreSQL に保存し、将来の fine-tuning 用データセットに使う設計です。
devstack コンテナ構成
podman-compose でミドルウェアを rootless コンテナとして起動します。
| サービス | イメージ | ポート | 役割 |
|---|---|---|---|
| NATS | nats:2.11-alpine | 4222, 8222 | JetStream メッセージング |
| PostgreSQL | postgres:18-jit-vector | 5432 | pgvector + JIT、agent_gateway / dagster DB |
| Dagster | カスタムビルド | 3000 | Webserver + Daemon + User-code gRPC |
| Vector | timberio/vector:0.45.0-alpine | 8686, 9598 | NATS テレメトリ → Prometheus/Loki |
| multi-bert-inference | Rust (Axum) + ONNX Runtime | 3001 (HTTP), gRPC | ColBERT embed (256d) + rerank (64d MaxSim) |
Lakehouse profile（Phase 2用、明示的に起動）:
- Nessie (19120) — Iceberg メタデータカタログ (git-like ブランチ)
- Trino (8081) — Iceberg カタログ付き SQL エンジン
- dbt-fusion — dbt CLI コンテナ
3ホスト ネットワークトポロジー
NATS を gateway と同居させ、左右の経路を1 hop 以内に収めています。
+-- storage server (observability) --+ +-- desktop mac (gateway) ----------+ +-- compute server (inference) ------+
| | | | | |
| Prometheus <-- scrape --------------|-- :9598 (Vector exporter) | | vLLM :8000 (GPU inference) |
| Grafana | | | | llama.cpp :8081 |
| Loki <-- push ----------------------- Vector (nats source) | | |
| Vector (aggregator) | | | | | Dagster :3000 |
| |-- nats source -------------------- NATS JetStream :4222 <---------|--|---- daemon sensor (pull subscribe) |
| |-- -> Loki sink | | ^ publish | | |-- webserver |
| +-- -> Prometheus exporter | | | (fire-and-forget) | | |-- daemon (sensor -> job) |
| MinIO :9000 | | | | | +-- user-code gRPC |
| | | agent-gateway :8080 | | |
| | | |-- proxy API | | PostgreSQL 18 :5432 |
| | | | (OpenAI/Anthropic/ | | +-- pgvector (ANN + full-text) |
| | | | Responses) | | |
| | | |-- knowledge orchestrator | | multi-bert-inference :3001 |
| | | +-- NATS pub ----------------|---> pipeline.* + telemetry.* |
| | | | | |
| | | LM Studio :1234 (CPU inference) | | Trino :8080 |
| | | | | Nessie :19120 |
+------------------------------------+ +-----------------------------------+ +-------------------------------------+
| ホスト | 役割 | 主要サービス | 稼働 |
|---|---|---|---|
| storage server | Observability + オブジェクトストレージ | Prometheus, Grafana, Loki, Vector, MinIO | 24/7 |
| desktop mac | Gateway + メッセージング | agent-gateway, NATS JetStream, LM Studio | 作業時のみ |
| compute server | GPU 推論 + データ基盤 | vLLM, llama.cpp, Dagster, PostgreSQL, Trino | 作業時のみ |
NATS は最大の publisher である gateway と同じ desktop mac に配置し、ホスト間の publish 通信をなくしました。desktop と compute は同時に起動・停止します。storage server は常時稼働し、Vector は接続断後に自動再接続します。
データ保管戦略
用途に応じて PostgreSQL (pgvector) と Iceberg (Parquet) を使い分けます。
PostgreSQL (pgvector) Iceberg (Parquet via Trino/Nessie)
-------------------- -----------------------------------
特性 構造化 + ベクトル検索 ファイル指向 + 大量蓄積
クエリ WHERE + ANN (近傍探索) SQL JOIN + time-travel
更新パターン UPSERT (行単位) append-only (immutable Parquet)
バージョン管理 なし (行履歴は自前) Nessie ブランチ + Iceberg snapshot
Phase 1 は全データを PostgreSQL に直接書き込みます。Phase 2 以降は、Dagster batch で PG → Parquet → Iceberg append を行い、dbt-fusion で分析用データセットを作る計画です。
PostgreSQL スキーマ
- document_chunks — 256次元ColBERTエンベディング + HNSW vectorインデックス
- chat_history — correlation_id別の会話履歴
- rerank_scores — クエリ × ドキュメント × モデル別のリランクスコア
- api_responses — Responses API 永続化
注意事項
- desktop mac / compute server 停止時はNATSも落ちるが、JetStreamの永続化データはdesktop macのディスクに保持され、次回起動時に未消費メッセージから再開する
- Vectorは接続断を検知して自動再接続するが、NATS停止中のtelemetryイベントは失われる（pipeline側はJetStreamのdurable consumerで保護される）
- PostgreSQL 18のJIT有効化は
shm_size: 4gbの設定が前提。不足するとOOM killのリスクがある
検証
- Dagster UIでasset materialization状態を確認: chat_pair, flow_lineage, tool_callが正常にMaterialized
- NATSモニタリング (
:8222) でstream状態とconsumer lagを監視 - Vector exporter (
:9598) → Prometheusでtelemetryメトリクスの到達を確認 - インテグレーションテスト:
go test -tags=integration ./internal/infra/integration ./internal/domain/pipeline
次のアクション
記事時点では Phase 1 のリアルタイム基盤が稼働しています。以降は次の計画です。
| Phase | 内容 | 状態 |
|---|---|---|
| 1 | gateway → NATS → Dagster/pg, Vector → Prometheus/Loki | 稼働中 |
| 2 | pg → Parquet → Iceberg バッチ, dbt-fusion 加工 | devstack済 |
| 3 | MCP server tool 定義, Trino クエリ tool, 仮説検証自動化 | 未着手 |
| 4 | fine-tuning パイプライン, MOE expert pruning, 合成データ増幅 | 未着手 |