はじめに
ふだんは
- LLM / RAG / エージェントまわりの検証
- バックエンド〜アーキテクチャ設計
- 「人間とAIの協調開発」のワークフロー研究
などをしています。
今回は、AIソリューションアーキテクトとしてのポートフォリオを意識して、
- 企業向けの LLMゲートウェイ(マルチテナント)
- 要件〜設計〜タスクまでを Spec Driven Development(cc-sdd / AI-DLC)
- 実装とリファクタリングを Geminiエージェント(Antigravity)で駆動
- LangChain v1 を前提としたバックエンド
- ちゃんと pytest が 50+ ケース通る状態 まで持っていく
というところまでやったので、そのプロセスとアーキテクチャをまとめます。
リポジトリはこれです:
- GitHub: DOM Enterprise Gateway
作ったものの概要: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地獄・循環依存
- テストなし・例外握りつぶし
-
結果、人間がほぼ書き直しになる…
というパターンになりがちです。
今回採ったアプローチ
そこで、今回はいきなり「コードをお願いする」のではなく、
-
要件を全部テキストで固める
requirements_p0_core_chat.mdhelp_content_outline.md-
DOM Enterprise Gatewayのドメインメモ
-
それを cc-sdd (Kiro-style Spec Driven Development) に食わせる
-
.kiro/steering/にプロダクト・技術・構造を書いた -
/kiro:spec-init→/kiro:spec-requirements→/kiro:spec-design→/kiro:spec-tasks
-
-
生成された tasks.md を元に、実装フェーズで Geminiエージェント を呼ぶ
-
実際に 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 にはメタデータだけ保存
- FastAPI の
-
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にマッピング
- OIDC の
-
初回ログインユーザー
- 環境変数
INITIAL_ADMIN_EMAILと一致したらadminロールを付与 - テナント (
Tenant) とユーザー (User) を必要に応じて自動プロビジョニング
- 環境変数
テスト戦略
-
test_auth_service.pyでは-
httpx.AsyncClientをAsyncMockして 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レスポンスのモック)
- Gemini / Embeddings (
-
スキップしているもの
-
test_rag_service.py::test_stream_rag_response- LangChain の
astream+RunnablePassthrough+StrOutputParserのモックがやや重いので、PoC ではスキップ
- LangChain の
-
「全部を完璧にモックする」のではなく、
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がない場合でもデフォルトで動くように調整 - テストでは
monkeypatchやoverride_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に書かせたコード」ではなく、
- アーキテクチャ設計
- 非機能要件(テナント分離・リセット戦略・ファイル制限)
- テスト戦略
を含めてポートフォリオとして説明できる形になったと思います。