1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Geminiエージェント×Spec Driven Developmentで企業向けLLMゲートウェイのバックエンドを一気に立ち上げた話

1
Posted at

はじめに

ふだんは

  • LLM / RAG / エージェントまわりの検証
  • バックエンド〜アーキテクチャ設計
  • 「人間とAIの協調開発」のワークフロー研究

などをしています。

今回は、AIソリューションアーキテクトとしてのポートフォリオを意識して、

  • 企業向けの LLMゲートウェイ(マルチテナント)
  • 要件〜設計〜タスクまでを Spec Driven Development(cc-sdd / AI-DLC)
  • 実装とリファクタリングを Geminiエージェント(Antigravity)で駆動
  • LangChain v1 を前提としたバックエンド
  • ちゃんと pytest が 50+ ケース通る状態 まで持っていく

というところまでやったので、そのプロセスとアーキテクチャをまとめます。

リポジトリはこれです:


作ったものの概要:DOM Enterprise Gateway P0 Core Chat

コンセプト

  • 企業内で複数の LLM を安全に使うための LLMゲートウェイ
  • テナントごとにナレッジ・設定・ログを分離
  • チャット + RAG + 長期メモリ + フィードバック + ヘルプ/オンボーディング まで含めた P0 PoC

今回フォーカスしている範囲(バックエンド)

  • フレームワーク: FastAPI

  • 言語: Python 3.12

  • LLM: Gemini (langchain-google-genai 経由) を前提とした抽象化

  • RAG: langchain-postgres + PGVector(PostgreSQL)

  • メモリ:

    • セッション要約ベースの エピソードメモリ
    • ユーザー単位の 構造化メモリ
  • 認証:

    • OIDC(Auth0 / Google / 企業IdP を想定)
    • 最初のユーザーを INITIAL_ADMIN_EMAIL で管理者に昇格
  • 非機能:

    • Docker Compose で backend + postgres + redis を起動
    • pytest + pytest-asyncio で 54テスト中 53 pass / 1 skip
    • WSL2 + Poetry 前提でのローカル実行

なぜ「AI × Spec Driven Development」にしたのか

普通の「AIコード生成」がつらい理由

経験ある方も多いと思いますが、

  • 「とりあえず FastAPI + RAG のサンプル書いて」→ 一見動きそうなコードが出る

  • でも実際に動かすと

    • バージョン不整合(LangChain, SQLAlchemy, Pydantic v2…)
    • import地獄・循環依存
    • テストなし・例外握りつぶし
  • 結果、人間がほぼ書き直しになる…

というパターンになりがちです。

今回採ったアプローチ

そこで、今回はいきなり「コードをお願いする」のではなく、

  1. 要件を全部テキストで固める

    • requirements_p0_core_chat.md
    • help_content_outline.md
    • DOM Enterprise Gateway のドメインメモ
  2. それを cc-sdd (Kiro-style Spec Driven Development) に食わせる

    • .kiro/steering/ にプロダクト・技術・構造を書いた
    • /kiro:spec-init/kiro:spec-requirements/kiro:spec-design/kiro:spec-tasks
  3. 生成された tasks.md を元に、実装フェーズで Geminiエージェント を呼ぶ

  4. 実際に WSL2 上で pytest を回しながら赤→緑 にしていく

という流れにしました。


アーキテクチャ概要

全体像

[Client (今はまだ仮)] 
      |
      v
[FastAPI Backend]  --- PostgreSQL (PGVector)
      |
      +-- Redis (セッション/一時データ)
      |
      +-- Google Vertex AI / Gemini (LLM, Embeddings)

バックエンドの主要コンポーネント

  • app/main.py

    • FastAPI アプリ本体
    • ルーターのマウント: /api/auth, /api/chat, /api/files, /api/admin, /api/feedback
  • app/core/config.py

    • pydantic-settings ベースの環境変数管理
    • 例:DATABASE_URL, OIDC_ISSUER, OIDC_CLIENT_ID, INITIAL_ADMIN_EMAIL, アップロード制限など
  • app/core/database.py

    • create_async_engine / async_sessionmaker による非同期DB接続
    • テスト用に置き換えやすいように設計
  • app/models/*.py

    • User, Tenant, ChatSession, ChatMessage, KnowledgeDocument, Feedback など
  • app/services/*

    • AuthService: OIDC / JWT 検証・ユーザープロビジョニング
    • ChatService: チャット履歴管理 + リセット処理
    • DomOrchestratorService: LLM・RAG・メモリ・AnswerComposer のオーケストレーション
    • RagService: PGVector + LangChain v1 での RAG チェーン
    • MemoryService: 構造化メモリ・エピソードメモリのCRUD
    • FileService: ファイルアップロード/制限・一時RAGインデックス登録
    • AnswerComposerService: IC-5 ライト(Decision / Why / Next 3 Actions)形式への整形
  • app/api/endpoints/*

    • FastAPI のエンドポイント層(認証・バリデーション・DI)

LangChain v1 前提のセットアップ

pyproject.toml はこんな感じです(一部抜粋):

[tool.poetry.dependencies]
python = "^3.12"

fastapi = "^0.115.0"
langchain = ">=1.1.0,<2.0.0"
langchain-postgres = ">=0.0.16,<0.1.0"
sqlalchemy = {extras = ["asyncio"], version = "^2.0.30"}
alembic = "^1.13.1"
uvicorn = "^0.30.1"
python-dotenv = "^1.0.1"
psycopg2-binary = "^2.9.9"
redis = "^5.0.5"
authlib = "^1.3.1"
python-jose = {extras = ["cryptography"], version = "^3.3.0"}
asyncpg = ">=0.30.0,<1.0.0"
langchain-google-genai = ">=3.2.0,<4.0.0"
pydantic-settings = "^2.6.1"
email-validator = "^2.2.0"

ポイント

  • langchain は v1 系を明示
  • langchain-google-genai で Gemini / Embeddings をラップ
  • langchain-postgres で PGVector を扱う
  • Pydantic v2 + pydantic-settings で設定管理を統一

ファイルアップロードと Ephemeral RAG

要件として、

  • チャットに最大 10 ファイルまで添付
  • 合計サイズに上限(例:30MB)
  • P0 からエンタープライズを意識して「ファイルは勝手にプロダクションナレッジには入れない」

というものがありました。

実装の考え方

  • /api/files/upload

    • FastAPI の UploadFile で受け取り
    • 拡張子チェック (ALLOWED_FILE_EXTENSIONS)
    • サイズチェック(MAX_FILE_SIZE_MB と合算制限)
    • DB にはメタデータだけ保存
  • RAG 側

    • 「Knowledge ベース」用の PGVector と
    • 「セッション限定 Ephemeral インデックス」を分離
    • アップロードされたファイルは セッションID紐付きの一時インデックス に格納

これにより、PoC 時点でも「勝手に全テナント共有ナレッジに入ってしまう事故」を避けつつ
ファイル添付チャットの体験は実現しています。


セッションリセットとメモリの設計(Resetインバリアント)

チャットが長くなったときの /reset の仕様もかなりこだわりました。

要件として決めたこと

  • セッションをリセットする際、

    • そのセッションの要約を LLM に生成させる
    • エピソードメモリとして DB に保存
    • 保存が成功したら初めて短期メモリをクリア
  • もし要約生成や保存でエラーになった場合

    • 「リセットできませんでした」とユーザーに返す
    • それでもダメな場合に備えた「強制リセット」オプションも検討

実装のイメージ

  • ChatService.reset_session(session_id: UUID, force: bool = False)

    • force=False の場合: 要約生成 → EpisodicMemory 保存 → 成功したら ChatMessage/Context クリア
    • force=True の場合: ユーザー同意を前提に、保存を諦めてクリア

これにより、

「調子悪くなったからセッションを消したいけど、
大事な会話も同時に全部消えるのはイヤ」

というユーザー体験と、

「要約にも失敗してリセットもできない」

という最悪パターンを両方ケアできるようにしました。


Auth / OIDC 周り

Auth まわりはかなりエラーが出やすいところでした。

やっていること

  • AuthService

    • OIDC の /.well-known/openid-configuration から jwks_uri を取得
    • authlib / python-jose を組み合わせて JWT 検証
    • sub, email, name などを抽出して AuthenticatedUser にマッピング
  • 初回ログインユーザー

    • 環境変数 INITIAL_ADMIN_EMAIL と一致したら admin ロールを付与
    • テナント (Tenant) とユーザー (User) を必要に応じて自動プロビジョニング

テスト戦略

  • test_auth_service.py では

    • httpx.AsyncClientAsyncMock して JWKS 取得をモック
    • JsonWebKey / JsonWebToken の部分もユニットテストで差し替え
    • 新規ユーザ・既存ユーザ・管理者昇格パターンを網羅

実際のテスト結果:

53 passed, 1 skipped, 5 warnings in 1.79s

テストと TESTING_NOTES.md

外部サービス依存が多いので、テストでどこまでモックしているかを明示するために
backend/TESTING_NOTES.md も生成しています。

  • モックしているもの

    • Gemini / Embeddings (langchain-google-genai)
    • PostgreSQL / PGVector(必要なところだけモック)
    • OIDC / Auth0 (get_current_user 依存関係の上書き、JWKSレスポンスのモック)
  • スキップしているもの

    • test_rag_service.py::test_stream_rag_response

      • LangChain の astream + RunnablePassthrough + StrOutputParser のモックがやや重いので、PoC ではスキップ

「全部を完璧にモックする」のではなく、
PoCの目的に対してどこまでテストでカバーするかを明文化する、というスタンスです。


WSL2 + Poetry + Docker の開発手順(ざっくり)

自分の環境は Windows なので、

  • コードとライブラリは WSL2 (Ubuntu) 側に置く
  • npx / node / poetry / docker は全部 WSL 側

という構成にしています。

セットアップ例

# リポジトリ取得
git clone https://github.com/KanadeYumesaki/dom-enterprise-gateway.git
cd dom-enterprise-gateway/backend

# Poetry 仮想環境作成 & 依存関係インストール
poetry install

# テスト実行
poetry run pytest app/tests

# Docker で DB/Redis を起動(将来的には)
cd ..
docker compose up -d

AIエージェント開発でハマったところ

1. cc-sdd で生成されたコードは「そのままでは動かない」

  • LangChain や SQLAlchemy、Pydantic のバージョン差
  • import 循環(dependencies.py での DI の順番など)
  • authlib の API 変更(JWTBearer が無い など)

など、「AIとしては筋が良いけど、現実のライブラリとは微妙にズレる」コードが大量に出てきました。

→ 対応として、

  • 1テストファイルずつ pytest を回しながら修正する

  • Antigravity / Gemini エージェントに対しては

    • 「編集してよいファイル」
    • 「絶対に壊してはいけない仕様」
    • 「テストを必ず通してから終了すること」

を明示してプロンプトを書くようにしました。

2. 環境変数まわり(pydantic-settings)

  • 最初は Settings() 生成時に ValidationError が多発
  • Optional[str] にして .env がない場合でもデフォルトで動くように調整
  • テストでは monkeypatchoverride_settings_for_tests 的なアプローチで上書き

3. AIに任せる範囲の線引き

  • 「一気に全部直して」ではなく、

    • このタスクだけ
    • このファイルだけ
    • このテストが通るところまで
  • という 小さめの単位に切ること が重要でした。


これからやりたいこと

  • フロントエンド(Angular or Next.js)実装

    • チャットUI / ファイルアップロードUI
    • 管理者向けナレッジ管理画面
    • 設定&ヘルプ画面(ヘルプセンターのアウトラインは既に仕様に入っている)
  • Agentic Research の高度化

    • リサーチモード用に、検索ループ・証拠パネルを整備
  • 監査ログ・レートリミット・テナントごとのクォータ管理

  • 実際の企業IdP(Azure AD / Okta 等)との接続検証


まとめ

  • Spec Driven Development (cc-sdd) で要件・設計・タスクを固めてから Geminiエージェント + Antigravity に実装とリファクタリングを手伝ってもらい最終的に pytest で 50+ テストが通る FastAPI バックエンド を構築しました。

ただの「AIに書かせたコード」ではなく、

  • アーキテクチャ設計
  • 非機能要件(テナント分離・リセット戦略・ファイル制限)
  • テスト戦略
    を含めてポートフォリオとして説明できる形になったと思います。
1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?