この記事は 第2弾（Agentic RAG） です。第1弾「RAGを実装してわかった『1回の検索』という壁」と同じ質問セットで比較しています。ソースコード一式は記事末尾からダウンロードできます。
前回の結果をおさらい
第1弾で普通のRAGを動かしたとき、Q2でこういう結果が出ました。
Q「認証とSQLインジェクション対策を両方教えて」
→ SQLインジェクション対策はコード例付きで回答できた
→ ただしJWT等の認証詳細は「ドキュメントに記載がありません」で終わった
1回の検索で両方を完璧にカバーするのは難しい。検索結果が薄くても「これで行く」しかないのが、普通のRAGの構造的な限界です。
Agentic RAGはここに判断を加えます。取れた文書をLLMが評価して「認証の詳細が不足している」と判断したら、クエリを書き換えて再検索する。このループが複雑な質問への対応力を変えます。
全体のアーキテクチャ
LangGraphを使ってこういうグラフを組みます。
質問
↓
[analyze] クエリを検索向けに整理
↓
[retrieve] ChromaDBでベクトル検索
↓
[evaluate] 「これで足りる？」をClaudeが判断
↓
足りない && 上限未達 → クエリを書き換えて[retrieve]に戻る
足りる or 上限到達 ↓
[generate] 収集した文書でClaudeが回答生成
↓
[reflect] 「ちゃんと答えられてる？」を自己評価
↓
不十分 && 上限未達 → [retrieve]に戻る
十分 or 上限到達 ↓
最終回答
普通のRAGとの最大の違いは、検索と評価がループすることです。
実装
環境
pip install anthropic chromadb sentence-transformers langgraph langchain-core
export ANTHROPIC_API_KEY="your-api-key"
State — パイプライン全体で引き回す状態
LangGraphの設計で最初にやることはStateの定義です。各ノードはこの辞書を読み書きしながら処理を進めます。
from typing import TypedDict, List
class RAGState(TypedDict):
original_query: str # 元の質問（変わらない）
current_query: str # 検索に使うクエリ（書き換えられる場合あり）
retrieved_docs: List[dict] # 蓄積された文書（ラウンドをまたいで保持）
retrieval_count: int # 何回検索したか
relevance_score: float # 関連性スコア
answer: str
is_sufficient: bool
log: List[str] # デバッグ用
retrieved_docs がリスト形式で蓄積される点がポイントです。再検索のたびに新しい文書が追加されていき、最終的な回答生成では全ラウンドの文書を使います。
ノード実装
クエリ分析ノード
def analyze_query(state: RAGState, collection) -> RAGState:
result = call_claude_json(f"""
質問: {state['original_query']}
以下のJSONで返してください:
{{
"search_query": "ベクトル検索向けに最適化したクエリ",
"complexity": "simple か complex",
"topics": ["含まれているトピック"]
}}
""")
return {
**state,
"current_query": result.get("search_query", state["original_query"]),
"retrieved_docs": [],
"retrieval_count": 0,
}
検索ノード（文書を蓄積する）
def retrieve(state: RAGState, collection) -> RAGState:
results = collection.query(
query_texts=[state["current_query"]],
n_results=4,
include=["documents", "metadatas", "distances"]
)
new_docs = [
{"content": doc, "source": meta.get("source"), "similarity": round(1 - dist, 3)}
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)
]
# 重複を除いて既存文書にマージ
existing = {d["content"] for d in state["retrieved_docs"]}
merged = state["retrieved_docs"] + [d for d in new_docs if d["content"] not in existing]
return {
**state,
"retrieved_docs": merged,
"retrieval_count": state["retrieval_count"] + 1,
}
関連性評価ノード（ここが核心）
def evaluate_relevance(state: RAGState, collection) -> RAGState:
docs_text = "\n\n".join([
f"[{d['source']}]\n{d['content']}"
for d in state["retrieved_docs"]
])
result = call_claude_json(f"""
質問: {state['original_query']}
取得文書:
{docs_text}
以下のJSONで評価してください:
{{
"score": 0.0〜1.0,
"reason": "評価の理由",
"missing": "不足している情報",
"improved_query": "改善されたクエリ（不十分な場合）"
}}
""")
score = float(result.get("score", 0.5))
return {
**state,
"relevance_score": score,
"current_query": result.get("improved_query") if score < 0.7 else state["current_query"],
}
条件分岐（ルーティング関数）
def route_after_evaluate(state: RAGState) -> str:
if state["relevance_score"] >= 0.7:
return "generate"
elif state["retrieval_count"] >= 3: # 上限に達したら諦めて進む
return "generate"
else:
return "retrieve" # 再検索
グラフの組み立て
from langgraph.graph import StateGraph, END
workflow = StateGraph(RAGState)
workflow.add_node("analyze", lambda s: analyze_query(s, collection))
workflow.add_node("retrieve", lambda s: retrieve(s, collection))
workflow.add_node("evaluate", lambda s: evaluate_relevance(s, collection))
workflow.add_node("generate", lambda s: generate_response(s, collection))
workflow.add_node("reflect", lambda s: reflect(s, collection))
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "retrieve")
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges(
"evaluate",
route_after_evaluate,
{"retrieve": "retrieve", "generate": "generate"}
)
workflow.add_edge("generate", "reflect")
workflow.add_conditional_edges(
"reflect",
route_after_reflect,
{END: END, "retrieve": "retrieve"}
)
app = workflow.compile()
第1弾と同じ質問で比較してみた
Q1. 単純な質問
「Pythonのリストに要素を追加する方法を教えてください」
実行ログ（実際のもの）:
[analyze] complexity=simple, topics=['Python', 'リスト', 'append', 'insert', 'extend']
[retrieve #1] query='Pythonリスト要素追加方法', 新規=4件, 累計=4件
[evaluate] score=0.95, reason=文書1がappendとinsertの2つの方法をコード例で示している
[generate] 使用文書数=4件
[reflect] is_sufficient=False, missing=extend()と+=演算子が記載されていない
[retrieve #2] query='Pythonリスト要素追加方法', 新規=4件, 累計=4件
[evaluate] score=0.95, reason=質問に完全に答えられている
[generate] 使用文書数=4件
[reflect] is_sufficient=True
検索回数: 2回
ここで少し面白い動きが起きています。evaluateのスコアは0.95と高かったのに、reflectが「extend() と += が記載されていない」と指摘して再検索に戻りました。ドキュメント自体にextendの記述がないため2回目も同じ結果になり、2回目のreflectでようやく「十分」と判断しています。
普通のRAGと回答の質は同等ですが、単純な質問なのに2回検索が走ったという点は正直なコストです。
Q2. 複合質問
「WebアプリのAPIを作るとき、認証とSQLインジェクション対策について両方教えてください」
実行ログ（実際のもの）:
[analyze] complexity=complex, topics=['API認証', 'SQLインジェクション', 'セキュリティ', 'Web開発']
[retrieve #1] query='Web API認証とSQLインジェクション対策', 新規=4件, 累計=4件
[evaluate] score=0.75, reason=SQLインジェクション対策は具体的だが認証の実装方法が不十分
missing=JWT生成・検証、パスワードハッシング、リフレッシュトークンなど
[generate] 使用文書数=4件
[reflect] is_sufficient=False, missing=認証の実装詳細が不足
[retrieve #2] query='Web API認証とSQLインジェクション対策', 新規=4件, 累計=4件
[evaluate] score=0.75, reason=同様の判断
[generate] 使用文書数=4件
[reflect] is_sufficient=True
検索回数: 2回
回答（抜粋）:
1. 認証について（JWT）
JWTはヘッダー、ペイロード、署名の3つから構成されるトークンです。
Authorization: Bearer [トークン]で送信します。2. SQLインジェクション対策
# ✅ 安全（プレースホルダーを使用） cursor.execute("SELECT * FROM users WHERE email = %s", (email,))
JWTの概要とSQLインジェクション対策のコード例、両方が回答に含まれています。普通のRAGで「記載がありません」と返ってきた部分が補完されました。
ただしevaluateスコアが0.75止まりだったのは、Claudeが「JWTの詳細な実装（トークン生成・検証コード等）がドキュメントにない」と判断し続けたためです。これはドキュメント自体の情報量の限界で、何回再検索しても解決しません。ドキュメントにない情報はAgenticにしても取れない、というのが正直なところです。
Q3. 複数ドキュメントにまたがる質問
「データベース設計とAPIのエラーレスポンス設計を比較して、共通する『一貫性』の考え方を説明してください」
実行ログ（実際のもの）:
[analyze] complexity=complex, topics=['データベース設計', 'API設計', 'エラーレスポンス', '一貫性', '比較分析']
[retrieve #1] query='データベース設計 API エラーレスポンス 一貫性 設計原則', 新規=4件, 累計=4件
[evaluate] score=0.30, reason=エラーレスポンスのConsistencyとトランザクションのConsistencyは
全く異なる概念で、文書には両者の比較説明がない
[retrieve #2] query='データベースのConsistency原則とAPI設計における形式の一貫性の関係性...', 新規=4件, 累計=6件
[evaluate] score=0.30, reason=個別の説明はあるが両者を結びつける分析がない
[retrieve #3] query='データベース設計とAPI設計においてスキーマ一貫性、エラーハンドリング...', 新規=4件, 累計=6件
[evaluate] score=0.30, reason=直接的な比較分析がドキュメントに存在しない
[generate] 使用文書数=6件 ← 上限到達で強制生成
[reflect] is_sufficient=False
検索回数: 3回（上限到達）
回答（抜粋）:
両分野において、「一貫性」は信頼性と予測可能性を確保するための重要な設計原則となっています。
- データベース：BEGIN / COMMIT / ROLLBACKによる明確なトランザクション管理
- API：すべてのエラーが
error→code→message→detailsという同じ構造を持つ
スコアが0.30のまま改善せず、3回上限で強制的に生成に進みました。reflectも「不十分」と判断しています。
これはAgenticの正直な限界です。「DBのConsistencyとAPIのエラー形式の一貫性を比較せよ」という質問は、そもそもドキュメントに書かれていない「横断的な考察」を求めており、何回検索してもドキュメントからは取れません。ただし回答の質は普通のRAGより高く、両方のドキュメントから素材を集めた上でClaudeが自分で分析・統合しています。
比較まとめ
実際の実行結果をまとめます。
| 普通のRAG | Agentic RAG | |
|---|---|---|
| Q1（単純） | ✅ 1回・十分 | ✅ 2回・同等（reflectが1回無駄に再検索） |
| Q2（複合） | ⚠️ JWT概要のみ、認証詳細なし | ✅ 2回・JWTとSQL両方カバー |
| Q3（比較・統合） | ⚠️ 素材は揃うが統合が浅い | ⚠️ 3回上限・スコア0.30だが回答の深さは上 |
| 平均検索回数 | 1.0回 | 2.3回 |
| Claude呼び出し | 1回 | 6〜10回 |
| コスト感 | 低い | やや高い |
正直な評価： Q2では明確な改善がありました。Q3は「ドキュメントにない知識を求める質問」なのでAgenticにしても完全な解決はできませんが、それでも回答の質は上がっています。Q1は単純な質問なのに2回検索が走るという無駄も起きました。
実際に使って気づいたこと
reflectノードが厳しすぎる問題
Q1でextend()がないと指摘して無駄な再検索を発生させたように、reflectが「ドキュメントに書いていない情報の欠如」まで指摘してしまうことがあります。reflectのプロンプトに「ドキュメントの範囲内で答えられているかを評価してください」と制約を加えると改善できます。
スコアが上がらない質問は早めに諦めるべき
Q3のようにスコアが0.30のまま3回ループした場合、再検索を重ねてもほぼ改善しません。「2回連続でスコアが変わらない場合は打ち切る」という早期終了ロジックを加えるのが実用上は有効です。
使い分けの判断基準
普通のRAGで十分なケース:
- FAQシステムのように質問のパターンが決まっている
- レスポンスタイムを最優先したい
- コストを最小化したい
Agentic RAGが効くケース:
- ユーザーが複合的な質問をする可能性がある
- 答えが複数のドキュメントに分散している
- 回答品質 > スピードのシステム
ソースコード
agentic_rag.zip に実装一式が入っています。
agentic_rag/
├── agentic_rag.py # 今回の実装（LangGraph + ChromaDB + Claude）
├── requirements.txt
└── docs/ # normal_ragと同じテスト用ドキュメント
pip install -r requirements.txt
export ANTHROPIC_API_KEY="your-key"
python agentic_rag.py
実行すると各ノードのログが出力されるので、どのタイミングで再検索が走っているかが確認できます。第1弾のコードと並べて実行してみてください。