LlamaIndex×LangGraphで作る「自己修正型エージェントRAG」実装ガイド：複数ソースのクエリ分解とループ制御

📖 この記事は、前回の記事「エージェントRAG（Agentic RAG）導入のための5ステップ：外部知識取り込みとワークフローの設計実践テンプレート」の続編です。
前回の内容をさらに深掘り・発展させてお届けします。まだお読みでない方は、先に前回の記事 エージェントRAG（Agentic RAG）導入のための5ステップ：外部知識取り込みとワークフローの設計実践テンプレート からご覧いただくのがおすすめです。

Naive RAGが直面する実務上の限界

従来の「質問に対して一発で検索をかけ、得られた結果から回答を生成する」という素朴なRAG（Naive RAG）は、実務における複雑な要件において限界を迎えつつあります。例えば、「A社とB社の最新の決算書を比較し、それぞれの成長戦略の違いを分析せよ」という指示に対して、一発の検索クエリでは適切なドキュメント群をすべて引き当てることは不可能です。

この課題を解決するのが、検索を単一のパイプラインではなく自律的に呼び出す「ツール」として再定義する**「エージェントRAG（Agentic RAG）」**です。本記事では、RAG特化のデータフレームワークである LlamaIndex と、状態管理に優れた LangGraph を組み合わせた協調アーキテクチャをベースに、中級者向けにその原理、設計思想、および具体的な実装方法を解説します。

---

1. 「LlamaIndex × LangGraph」協調アーキテクチャの設計思想

エージェントRAGの構築において、LlamaIndexとLangGraphを組み合わせるアプローチは極めて強力です。AIを単なるツールではなく、高度な状態管理を行う「エコシステム」として機能させるために、それぞれの強みを融合させます。

• LlamaIndex（データの接続とクエリ分解）：ドキュメントのチャンキングやメタデータ抽出、セマンティック検索のほか、複雑なクエリを細分化する「計画能力」に優れています。
• LangGraph（状態管理とループ制御）：アプリケーションを「状態機械（State Machine）」としてモデル化し、自律的な「検索 ⇄ 評価 ⇄ 再構成」のループを厳密に制御します。

この2つを協調させることで、性質の異なるデータソース（社内規定、財務データ、製品マニュアルなど）を個別のインデックスとして構築し、エージェントが自律的に使い分けるルーティングと、動的な自己修正が可能になります。

---

2. 多段階クエリ分解と自己修正ワークフローの実装

実際にLlamaIndexで複数ソースに対するプランニング層を定義し、それをLangGraphの自己修正ワークフロー内に組み込む実装例を以下に示します。

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

# ==========================================
# 1. 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
)

# ==========================================
# 2. LangGraphによる自己修正ワークフローの設計
# ==========================================

# エージェントの状態（State）を定義
class AgentState(TypedDict):
    query: str
    response: str
    grade: str
    loop_count: int

# [Node 1] データ検索・回答生成を行うノード
def query_node(state: AgentState) -> Dict:
    # SubQuestionQueryEngineを呼び出し、複数ソースから回答を統合
    response = sub_query_engine.query(state["query"])
    return {
        "response": str(response), 
        "loop_count": state.get("loop_count", 0) + 1
    }

# [Node 2] 回答の品質・整合性を自己評価（Grader）するノード
def grade_node(state: AgentState) -> Dict:
    response_text = state["response"]
    if "データなし" in response_text or len(response_text) < 50:
        return {"grade": "fail"}
    return {"grade": "pass"}

# [Edge Logic] 次のノード遷移を判定する条件付きエッジ
def decide_to_end(state: AgentState) -> str:
    # 評価が合格、またはループカウンタ上限（3回）に達した場合は終了
    if state["grade"] == "pass" or state["loop_count"] >= 3:
        return "end"
    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()

---

3. 本番構築における典型的なつまずきポイントとエラー対処

Agentic RAGは高度な自律性を持つ反面、本番運用において特有の課題が発生します。設計時点で以下の「2大対策」を織り込むことが不可欠です。

① 評価器（Grader）の「判定ブレ」対策

評価を担うLLMの出力が不安定であると、正常な回答を「不合格（fail）」と誤判定し、ループが無駄に空回りしてしまいます。これを防ぐために、単一のフリープロンプト評価ではなく、**Pydanticによる構造化出力（Structured Outputs）**を強制し、{"is_relevant": True/False, "reason": "..."} などの厳格なフォーマットでLLMに回答を判定させます。また、事前に Ragas などの評価フレームワークを用い、テストデータセットに対して適切な閾値を検証・調整します。

② ループ増加に伴う「APIコストとレイテンシ」対策

検索と評価を繰り返すため、APIコストの高騰やレスポンスの遅延が問題となります。対策として、質問の難易度を最初にLLM（または軽量モデル）に判定させる「ルーティング」を実装します。単純な事実確認などの「簡単な質問」は自律ループを通さず、従来のClassic RAG（1パス）へバイパスさせます。また、上記実装例のように、AgentStateに loop_count を設け、最大処理回数（例：3回）に達した場合は強制離脱するガードレールと、LangGraph標準の recursion_limit を必ず併用してください。