agent-gateway v3 のドメイン分割と MLflow 連携

変更した構成

agent-gateway のパイプラインを v3 に変更しました。pipeline.knowledge.* を llm、obsidian、mlflow の 3 ドメインに分け、NATS subject、Dagster sensor/job/asset、Vector のルーティングを揃えました。

開発中で本番データがなかったため、互換処理は設けず、旧トピックと旧 durable consumer を削除する方針にしました。6 ワークストリームと MinIO 外部化で計 8 コミットです。各段階で go build ./... が通り、Dagster はインポートを確認しました。

Obsidian vault と MLflow を追加すると、knowledge という分類だけではイベントの用途が分かりにくくなりました。LLM 導入で扱う会話、文書、実験管理を、名前から区別できる構成にしています。

この記事はGo + NATS + Dagster によるAIオーケストレーション基盤の続編にあたる。

前提

• 対象システム: agent-gateway — Go + Python 構成の自作 LLM オーケストレーション基盤
• ランタイム: EPYC 9175F (768GB DDR5, RTX 6000 Pro Max-Q)、Podman rootless
• メッセージング: NATS 2.11 (JetStream)
• オーケストレーション: Dagster (Python)
• テレメトリ: Vector (Rust) → Prometheus / Loki
• 新規統合対象: Obsidian vault、MLflow Model Registry
• 設計ドキュメント: docs/2nd-design-archi.md

設計判断

ドメイン分割の方針

旧設計では、chat persist、embedding、tool call がすべて pipeline.knowledge.* に含まれていました。

• pipeline.knowledge.embedding ではチャット用 embedding と vault indexing を区別できません。
• MLflow の実験管理イベントも knowledge に入っていました。
• Dagster sensor が subject 名から用途を判断できませんでした。
• テレメトリの分類も機能の境界と合わなくなっていました。

分割後の3ドメイン:

テレメトリも telemetry.llm.* / telemetry.obsidian.* / telemetry.mlflow.* に分けました。

後方互換性の排除

本番データがない段階だったため、マイグレーション処理は書かず、旧トピックと旧 durable consumer の削除で移行しました。互換処理を省いて、実装量を減らしています。

ワークストリーム構成

6 ワークストリームを依存順に進め、各 WS を 1 コミットで区切りました。

  WS1 (Lakehouse簡素化) → WS2 (NATSトピック移行) → WS3 + WS4 + WS5 (並行) → WS6 (Vector config)
  

WS3 は Go domain 層、WS4 は Go transport 層、WS5 は Python Dagster です。変更する層が異なるため、並行して進められます。

実装

WS1: Lakehouse 簡素化

以前の ELT 検証で導入した Nessie + Trino + dbt-fusion は削除しました。この段階では PyIceberg + JDBC catalog で必要な処理を行えると判断しました。

変更内容:

• podman-compose.yml から nessie、trino、dbt-fusion サービスと trino-warehouse ボリュームを削除
• internal/config/config.go から TrinoBaseURL、NessieBaseURL フィールドと env var ロードを削除
• devstack/postgres/init.sql に Iceberg JDBC catalog メタデータテーブル (iceberg_tables、iceberg_namespace_properties) を追加
• devstack/trino/ ディレクトリを丸ごと削除

WS2: NATS トピック移行

pipeline.knowledge.* のトピック定数を、新しいドメイン別の名前に置き換えました。変更は 5 ファイルです。

テレメトリ (telemetry.knowledge.* → telemetry.llm.* / telemetry.obsidian.*) と DAG 名 (knowledge.prompt_flow → llm.prompt_flow) も更新しました。

grep "pipeline\.knowledge\|telemetry\.knowledge" で、internal/ に旧トピックが残っていないことを確認しました。

WS3: domain/obsidian 新規作成

internal/domain/obsidian/ に 3 ファイルを追加しました。

repository.go は VaultRepository と、そのローカル実装 LocalVault です。resolve() が vault root 外のパスを拒否し、path traversal を防ぎます。

service.go — Service struct に5メソッド:

LLM は knowledge orchestrator 経由で呼びます。obsidian ドメインは vault I/O と embedding を担当し、pipeline.obsidian.* にイベント、telemetry.obsidian.* にテレメトリを送ります。

indexer.go — vault 走査 → チャンク分割 → embedding → pgvector upsert のパイプライン。

  vault files → scan → chunk (512 rune, 64 overlap) → embed → pgvector upsert
  

internal/config/config.go に ObsidianVaultDir と OBSIDIAN_VAULT_DIR を追加しました。cmd/server/main.go で obsidian.NewService を初期化しています。transport 層への接続は未実装です。

WS4: MLflow webhook エンドポイント

Go の transport 層に、MLflow 用の 2 ハンドラを追加しました。

webhook_mlflow.go — POST /v1/webhook/mlflow

MLflow Model Registry の webhook を受け、X-Webhook-Signature で HMAC-SHA256 を検証します。イベントタイプに応じて NATS に送ります。

pipeline_mlflow.go — 2エンドポイント:

server.go に WithPipelineService ServerOption を追加し、WithConversationRepository と同じ方法で DI しました。config.go には MLflowWebhookSecret と環境変数 MLFLOW_WEBHOOK_SECRET を追加しています。

WS5: Dagster sensor/job/asset 再構成

変更は 16 ファイル、+401 行 / -229 行でした。

6 個の per-topic sensor を、4 個の wildcard sensor に置き換えました。

各 sensor は wildcard subject を購読し、受信した subject の suffix で実行するジョブを選びます。

chat_persist.py、chat_pair_materialize.py、knowledge_events.py、tool_calls.py を削除し、llm_jobs.py、obsidian_jobs.py、mlflow_jobs.py に分けました。

asset の key_prefix もドメイン別に更新:

• ["knowledge", "embedding"] → ["obsidian", "embedding"]
• knowledge_chat_audit → llm_chat_audit
• knowledge_events → pipeline_events

WS6: Vector config 更新

devstack/vector/vector.yaml に、ドメイン別のルーティング transforms を追加しました。

  source: telemetry.> (wildcard)
    |
    v
route_by_domain (subject prefix で分岐)
    |-- telemetry.llm.*     → domain: "llm"
    |-- telemetry.obsidian.* → domain: "obsidian"
    |-- telemetry.mlflow.*  → domain: "mlflow"
    |
    v
add_domain_label (.domain フィールド付与)
    |
    +-- sink: console
    +-- sink: prometheus_exporter (ドメインラベル付き)
  

追加: MinIO 外部化

devstack の MinIO は、既存の storage.home.arpa:9000 に切り替えました。既存の archives、artifacts、datasets、lightdash、models バケットとは名前を分けています。

• podman-compose.yml から minio サービスと minio-data ボリュームを削除
• dagster-user-code、dagster-daemon の S3 エンドポイントを storage.home.arpa に更新
• minio-init コンテナで agw-mlflow と agw-iceberg バケットを agw- prefix で自動作成（既存バケットとの namespace 分離）
• MLflow の artifact root を s3://mlflow/ から s3://agw-mlflow/ に更新

注意事項

• obsidian domain の indexer は vault diff 検知が未実装。現状は全ファイル走査の IndexAll のみ
• NATS JetStream の旧 durable consumer 名が残骸として残る。nats-init での stream purge / 再作成が必要
• Dagster UI でのアセットグラフ動作確認が未実施
• obsidian domain の transport 層への接続は TODO のまま
• WS5 実装中にツールのコンテキストクリアが発生し defs.py と assets/__init__.py の再読み込みが必要になったが、実装自体への影響はなし
• indexer.go の初回生成時に encodeJSON ヘルパーが壊れたコードになっていた（var jsonMarshal を fmt.Sprintf にフォールバックする不自然なパターン）。json.Marshal 直接呼び出しに修正済み

検証

• 各ワークストリームで go build ./... を通過
• Dagster 側はインポート検証済み
• grep "pipeline\.knowledge\|telemetry\.knowledge" で internal/ 配下に旧トピック残留なし
• 全コミットは feature/redesin-devstack ブランチ上

次のアクション

コミット履歴