従来のRAGが抱える課題とエージェントRAGの必要性
LLMに外部知識を組み込む手法として「RAG（検索拡張生成）」は広く普及しています。しかし、従来の「Naive RAG」には大きな課題があります。それは「質問に対して一度だけ検索を行い、その結果から直接回答を生成する」という静的な一方向処理である点です。
検索された情報が不十分であったり、的外れであったりした場合、従来のシステムは不正確な回答（ハルシネーション）を出力するか、「分かりません」と答えるしかありません。この課題を解決するために注目されているのが、自律的なループ制御を行う「エージェントRAG」です。
自己修正型検索（Corrective RAG: CRAG）のブレイクスルー
「自己修正型検索（Self-RAG / Corrective RAG: CRAG）」は、検索結果の質をLLM自身に評価させ、必要に応じて動的に検索プロセスを修正する高度なアプローチです。
CRAGは以下の3つのステップで実行されます。
- 検索： ベクトルデータベースや外部データソースから関連情報を取得します。
- 評価/判定： 取得した情報がユーザーの質問に回答する上で十分（Correct/Pass）か、不十分（Incorrect/Fail）かをLLM（Graderノード）が判定します。
- 再構成・フォールバック： 判定が「Pass」であればそのまま回答を生成します。「Fail」の場合は、クエリを検索に適した形に書き直して再検索（再クエリ）するか、Web検索（Tavilyなどの検索APIやMCP経由のWeb検索サーバー）にフォールバックして不足した知識を補完します。
「LlamaIndex × LangGraph」による協調アーキテクチャ
この自己修正型検索を安定して動作させるには、異なる強みを持つ2つのフレームワークの協調が不可欠です。
| フレームワーク | 担当レイヤー | 主な強み・役割 |
|---|---|---|
| LlamaIndex | 知識・検索層（Data Integration） | 「LlamaParse」による高度なドキュメント解析、構造化チャンキング、および複雑な質問を複数の具体的な質問に分解して並行検索する「Sub-Question Query Engine」などの強力な検索技術を提供します。 |
| LangGraph | 意思決定・制御層（Orchestration） | アプリケーションを「状態機械（State Machine）」として定義します。状態（State）を厳密に管理しながら、条件付きエッジ（Conditional Edges）によるループ制御や人間による介入（Human-in-the-loop）を安全に制御します。 |
LlamaIndexを「情報の探索・クエリ分解の専門家」、LangGraphを「状態を監視する司令塔」として組み合わせることで、堅牢な自己修正システムを構築できます。
MCP（Model Context Protocol）による外部ツールの統合
エージェントが外部リソースへシームレスにアクセスするための標準化プロトコルとして、「MCP（Model Context Protocol）」の統合が有効です。
-
LlamaIndexでの利用：
llama-index-tools-mcpパッケージを使用し、外部のMCPサーバーが提供するAPIやツールをLlamaIndexの「ツール」に動的変換してエージェントへロードします。 -
LangGraphでの利用：
langchain-mcp-adaptersを用いて、MCPツールをLangGraphのToolNodeに組み込みます。
これにより、Web検索（Google Search MCPなど）やローカルデータベース（PostgreSQL MCPなど）を利用する際、エージェント側で個別のAPI連携コードを都度書き下す必要がなくなり、一元的に外部知識を拡張可能になります。
【実践】自己修正型RAGの構築ロードマップ
実際に「LlamaIndex × LangGraph」を組み合わせた自己修正型RAGの実装例を示します。このコードは、LlamaIndexの SubQuestionQueryEngine でドキュメントを並行検索し、LangGraphで検索結果の自己評価と修正ループを実行する仕組みを再現しています。
必要ライブラリのインストール
pip install llama-index langgraph langchain-openai
実装コード
import os
from typing import Dict, TypedDict
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.query_engine import SubQuestionQueryEngine
from langgraph.graph import StateGraph, END
# LlamaIndex による知識の構造化とサブクエリエンジンの構築
doc_a = SimpleDirectoryReader("./data/company_a").load_data()
doc_b = SimpleDirectoryReader("./data/company_b").load_data()
index_a = VectorStoreIndex.from_documents(doc_a)
index_b = VectorStoreIndex.from_documents(doc_b)
query_engine_tools = [
QueryEngineTool(
query_engine=index_a.as_query_engine(),
metadata=ToolMetadata(name="company_a_docs", description="A社の決算書・事業情報")
),
QueryEngineTool(
query_engine=index_b.as_query_engine(),
metadata=ToolMetadata(name="company_b_docs", description="B社の決算書・事業情報")
)
]
sub_query_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=query_engine_tools
)
# LangGraph による状態（State）の定義と自己修正ワークフロー
class AgentState(TypedDict):
query: str # ユーザーの質問
response: str # 最終回答
grade: str # 評価結果 ('pass' or 'fail')
loop_count: int # 無限ループ防止用のカウンター
def query_node(state: AgentState) -> Dict:
current_try = state.get("loop_count", 0) + 1
print(f"--- 検索と回答生成を実行中 (試行回数: {current_try}) ---")
response = sub_query_engine.query(state["query"])
return {
"response": str(response),
"loop_count": current_try
}
def grade_node(state: AgentState) -> Dict:
print("--- 検索結果・回答の品質を評価中 ---")
response_text = state["response"]
fail_keywords = ["データなし", "答えられません", "情報がありません"]
if any(kw in response_text for kw in fail_keywords) or len(response_text) < 50:
print("➔ 判定：[FAIL] 情報が不十分です。")
return {"grade": "fail"}
print("➔ 判定：[PASS] 十分な品質が確保されています。")
return {"grade": "pass"}
def decide_to_end(state: AgentState) -> str:
if state["grade"] == "pass" or state["loop_count"] >= 3:
return "end"
print("➔ 警告：回答品質が低いため、再検索プロセスに入ります。")
return "re_query"
workflow = StateGraph(AgentState)
workflow.add_node("query_data", query_node)
workflow.add_node("grade_data", grade_node)
workflow.set_entry_point("query_data")
workflow.add_edge("query_data", "grade_data")
workflow.add_conditional_edges(
"grade_data",
decide_to_end,
{
"end": END,
"re_query": "query_data"
}
)
app = workflow.compile()
if __name__ == "__main__":
inputs = {
"query": "A社とB社の直近の成長戦略の共通点と違いを分析してください。",
"loop_count": 0,
"grade": "fail"
}
for output in app.stream(inputs):
for key, value in output.items():
print(f"Node '{key}' が終了しました。")
final_state = app.get_state(config={"configurable": {"thread_id": "1"}})
print("\n--- 最終回答 ---")
print(final_state.values.get("response"))
本番運用における課題と対策
この自己修正型システムを実際のプロダクション環境に導入する際、開発者がつまずきやすい2つのポイントがあります。
1. 無限ループとAPIコストの高騰
社内ドキュメントや外部ソースに本当に該当する情報がない場合、エージェントは正しい回答を探し求めて無限ループに陥ります。これにより、LLMのAPIコストが急激に跳ね上がります。
-
対策： サンプルコードに示しているように、
AgentStateにloop_countを設け、最大試行回数（例：3回）に達した場合は強制的にワークフローを終了させる（またはWeb検索等の代替ルートへフォールバックさせる）ガードレールを必ず実装してください。
2. 自己評価器（Grader）の判定ブレ
簡易的なキーワードマッチを用いた評価では、十分な情報が含まれているにもかかわらず誤って「Fail」と判定されるなど、判定精度にブレが生じます。
-
対策： 品質を評価する
grade_nodeの内部ロジックにおいて、LLMの「Structured Outputs（構造化出力）」やPydanticスキーマを活用します。事実性やユーザー質問への網羅性を構造化された評価基準として定義し、LLMに客観的に判定させる仕組みにアップグレードしてください。
次世代RAGアーキテクチャの展望
検索、評価、そして再検索を自律的に繰り返す「自己修正型検索」は、RAGの精度を飛躍的に高める鍵となります。LlamaIndexによる高度なデータ構造化と、LangGraphによる厳密な状態管理、さらにMCPによる標準化されたツール統合を組み合わせることで、安定的で拡張性の高いエンタープライズ向けエージェントRAGの構築が可能になります。