📖 この記事は、前回の記事「マルチソース＆クエリ分解で実現する高度なエージェントRAG実践ガイド」の続編です。
前回の内容をさらに深掘り・発展させてお届けします。まだお読みでない方は、先に前回の記事マルチソース＆クエリ分解で実現する高度なエージェントRAG実践ガイドからご覧いただくのがおすすめです。

前回、私たちはマルチソース検索と自律フローを可能にする「Agentic RAG」の概念とその利点を解説しました。しかし、実際にプロダクション環境で運用する場合、一度の検索（Naive RAG）で引き当てた情報が不十分であったり、ノイズが含まれていたりすることで、最終的な回答品質が低下する問題に直面します。

この課題を根本から解決するのが、「LlamaIndex」による高度な検索・クエリ分解能力と、「LangGraph」の強固な状態管理を組み合わせた「自己修正型エージェントRAG（Corrective RAG: CRAG）」です。本稿では、複雑な質問に対して自律的に「検索 ⇄ 評価 ⇄ 再構成（自己修正）」を繰り返すシステムの設計思想から、具体的なPythonコード、本番構築におけるつまずきポイントの回避策までを技術的に深掘りします。

設計思想：「LlamaIndex × LangGraph」協調アーキテクチャの原理

実務における「A社とB社の最新の決算書を比較し、成長戦略の違いを分析する」といった複雑な要求には、単一の検索クエリだけでは対応できません。本アーキテクチャでは、役割を明確に分担させる協調型のアプローチをとります。

1. 検索と分解を担う「LlamaIndex」

LlamaIndexは「知識・検索層」のプロフェッショナルです。複雑なクエリを、複数の実行可能な「サブクエリ（部分的な質問）」に分解し、それぞれに最適なデータソースに自動的にルーティングして並行検索を行う SubQuestionQueryEngine などを備えています。

2. 状態管理と意思決定を担う「LangGraph」

LangGraphは「意思決定・制御層」を司る司令塔です。アプリケーションを「状態機械（State Machine）」としてモデル化し、エージェントが検索、評価、再構成のループをたどる際、状態（State）を厳密に管理することで、無限ループの防止や条件分岐（Conditional Edges）を安全に制御します。

「自己修正型検索（CRAG）」の基本ループ

この2つの協調により、以下の自己修正ループ（CRAG）が実現します。

検索（Retrieve）: 複雑なクエリを分解し、複数ソースから並行して知識を検索します。
評価（Grade）: 取得した情報がユーザーの質問に答えるのに十分か（Pass/Fail）をLLM（Graderノード）に自己判定させます。
再構成（Re-write/Fallback）: 判定が「Fail」だった場合、クエリを書き換えて再検索するか、外部Web検索APIなどへフォールバックして不足情報を補完します。

5つのステップで進める自己修正RAGの実装手順

自己修正RAGの構築は、以下の5つの手順で進めます。

Step 1: データソースの整理とインデックス化
性質や文脈の異なる複数のデータ（社内規定、財務データなど）を個別のベクトルインデックスとして構築します。
Step 2: サブクエリエンジンの構築
構築した各インデックスを個別ツールとして定義し、それらを統合して自動クエリ分解を行う SubQuestionQueryEngine を初期化します。
Step 3: 状態（State）と回答ノードの定義
グラフ全体で共有する状態スキーマ（AgentState）を定義し、データ検索を行う query_node を実装します。
Step 4: 自己評価ロジック（Graderノード）の実装
出力された回答の品質を検証し、情報が不十分な場合に fail を返す評価ロジックを実装します。
Step 5: 状態遷移グラフの構築と無限ループ対策の導入
最大試行回数（ガードレール）を設け、条件付きエッジでループを制御するグラフを構築します。

【実践】Pythonによる自己修正RAGの実装コード

以下は、LlamaIndexの自動クエリ分解と、LangGraphによる自己修正ループ（CRAG）を統合した実践的な実装例です。

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

# =====================================================================
# Step 1 & 2: LlamaIndex による知識の構造化とサブクエリエンジンの構築
# =====================================================================

# 1. 異なるコンテキストデータ（例: A社、B社の資料）を読み込んでインデックス化
# ※ 実行環境に合わせて ./data/company_a, ./data/company_b にPDFやテキストを配置してください
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)

# 2. クエリエンジンを個別ツールとして定義
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社の決算書・事業情報")
    )
]

# 3. 複雑な問いを分解して横断検索する SubQuestionQueryEngine を初期化
sub_query_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=query_engine_tools
)


# =====================================================================
# Step 3, 4, 5: LangGraph によるエージェントRAGと自己修正ワークフローの構築
# =====================================================================

# 1. エージェントの状態を定義する TypedDict
class AgentState(TypedDict):
    query: str       # ユーザーからの入力クエリ
    response: str    # LLM / LlamaIndex からの回答
    grade: str       # 評価結果（'pass' or 'fail'）
    loop_count: int  # ループ回数（無限ループ防止ガードレール）

# 2. 検索・回答生成ノード
def query_node(state: AgentState) -> Dict:
    print(f"\n--- 検索と回答生成を実行中 (Try: {state.get('loop_count', 0) + 1}) ---")
    
    # LlamaIndexのエンジンが自動でサブ質問に分解・各ソースへ並行検索して統合回答を生成
    response = sub_query_engine.query(state["query"])
    return {
        "response": str(response),
        "loop_count": state.get("loop_count", 0) + 1
    }

# 3. 品質評価ノード（自己判定ロジック）
def grade_node(state: AgentState) -> Dict:
    print("--- 検索結果・回答の品質を評価中 ---")
    response_text = state["response"]
    
    # ※ 本番環境ではLLMを用いた「構造化出力（Structured Outputs: Pydantic）」による判定を推奨します。
    # ここでは簡易的に、情報欠落キーワードや極端に短い文章を 'fail' とするモックロジックを適用
    if "データなし" in response_text or "答えられません" in response_text or len(response_text) < 50:
        print("➔ 判定：[FAIL] 情報が不十分です。再検索（再試行）を要求します。")
        return {"grade": "fail"}
        
    print("➔ 判定：[PASS] 十分な回答品質が確保されています。")
    return {"grade": "pass"}

# 4. 条件付きエッジのルーティングロジック
def decide_to_end(state: AgentState) -> str:
    # 評価を通過した、もしくは最大試行回数（3回）に達した場合は終了
    if state["grade"] == "pass" or state["loop_count"] >= 3:
        return "end"
    # 評価を通過できず、まだ試行回数に余裕がある場合は再クエリ・再検索を実行
    return "re_query"

# 5. 状態遷移グラフ（StateGraph）の構築
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,              # ワークフロー終了（ENDノードへ）
        "re_query": "query_data" # 自己修正を伴う再検索へループ
    }
)

# グラフをコンパイルして実行可能なアプリケーションにする
app = workflow.compile()


# =====================================================================
# 実行例
# =====================================================================
if __name__ == "__main__":
    inputs = {
        "query": "A社とB社の最新の決算書を比較し、成長性の違いを分析せよ",
        "loop_count": 0
    }
    final_state = app.invoke(inputs)
    
    print("\n================ [ 最終回答 ] ================")
    print(final_state["response"])

本番構築における典型的な「つまずきポイント」とエラー回避策

高度なエージェントRAGをプロダクション環境に適用する際、中級者以上が直面しやすい代表的な課題と対策は以下の通りです。

① 無限ループの発生とAPIコストの高騰

課題: 評価器（Grader）が「Fail」判定を出し続け、検索クエリが何度も書き直されてLLMの無限ループが発生し、APIトークン料金が跳ね上がるケースがあります。
対策: AgentState 内に必ず loop_count などの状態カウンタを持たせ、最大試行回数（例: 3回）に達した時点で「現状得られている最善の情報を統合して回答する」もしくは「人間の確認を挟むフロー（Human-in-the-Loop）へ逃がす」ガードレールを設計してください。

② 評価器（Grader）の「判定ブレ」による不安定性

課題: Graderに単純なテキスト判定や曖昧なプロンプトを使用すると、同じ検索結果であっても評価の合否（Pass/Fail）が毎回ブレてしまい、全体のワークフローが不安定になります。
対策: LLMにJSON形式で確実に回答させる「構造化出力（Structured Outputs / Tool Calling）」や、Pydanticを用いたスキーマ定義を強制します。判定基準のシステムプロンプトにはFew-shot（良質な判定例をいくつか提示する）を適用し、判定ロジックのロバストネス（堅牢性）を高めてください。

③ ネットワークレイテンシとコストの増加

課題: クエリ分解によって並行検索や多数のサブ質問が発生すると、LLMの呼び出し回数が数倍になり、回答がユーザーへ返るまでの待機時間（レイテンシ）が大幅に長くなります。
対策: SubQuestionQueryEngine や LangGraph のノード実行は、非同期処理（Async / asyncio）を活用して並行して処理させます。また、判定を行う Grader ノードには、推論能力が高く安価かつ高速な小型モデル（例: Claude 4.5 Haikuなど）を部分的に採用し、役割に応じた「モデルの適材適所」のハイブリッド設計を行うことが有効です。