0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@jjking(Junichi Ito)

LlamaIndexとLangGraphで構築する「自己修正型RAG」：サブクエリ分解と自律評価による高度な外部知識統合

0
Posted at

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

LlamaIndex×LangGraph：ハイブリッド協調アーキテクチャの設計思想

前回の記事では、エージェントRAG（Agentic RAG）導入に向けたロードマップを俯瞰しました。今回は、その核心となる「外部知識の自律的取り込み」と「堅牢なワークフロー制御」を実現するために、LlamaIndex と LangGraph を組み合わせたハイブリッド協調アーキテクチャについて、技術的な詳細を深掘りします。

従来のRAG（Naive RAG）は、ユーザーのクエリに対して1度だけ検索を行い、取得した情報をそのままLLMに引き渡す「1ウェイ・プロセス」でした。しかし、この手法は複雑なビジネス要件や、ドキュメント間の競合、関連性の低い情報の混入に対して脆弱であり、ハルシネーションを容易に引き起こします。AIを単なる一過性の呼び出しツールとしてではなく、状態を持った「自律的なエコシステム」として機能させるために、以下の2つのフレームワークの融合が不可欠となります。

  • 知識・検索層（LlamaIndex）： ドキュメントの論理構造を崩さない「構造化チャンキング」や、子チャンクから親セクションへ自動的に復元する「Auto-merging retrieval」など、検索精度の最大化に特化しています。特に、1つの複雑な質問を複数の具体的な「サブ質問」に分解して各データソースへ自動ルーティングする Sub-Question Query Engine は、外部知識を正確に取り込むための強力なパーツです。
  • 意思決定・制御層（LangGraph）： エージェントの「状態（State）」を明示的に維持しながら、ループ処理や「条件付き分岐（Conditional Edges）」、さらには人間の承認を挟むプロセスなどをプログラムとして堅牢に制御できます。

「情報の探索・分解のプロ」であるLlamaIndexと、「ワークフローの司令塔」であるLangGraphを協調させることで、**「賢い知識分解 ✕ 堅牢な自己修正ループ」**というプロダクション環境に耐えうるエージェントRAGが完成します。

外部知識の取り込みと自己修正ワークフローの5ステップ

本アーキテクチャを具現化するための5つの開発ステップを定義します。

[ ユーザーの質問 ]
       │
       ▼
┌──────────────────────────────────────┐
│  Step 1 ~ 2: LlamaIndex               │
│  ・外部知識の読み込みとインデックス化 │
│  ・Sub-Question Query Engine による   │
│    クエリの多段階分解と検索          │
└──────────────────┬───────────────────┘
                   │
                   ▼ (検索結果を含む初期コンテキスト)
┌──────────────────────────────────────┐
│  Step 3: LangGraph 状態管理 (State)   │
│  ・AgentState に検索結果とループ回数を格納│
└──────────────────┬───────────────────┘
                   │
                   ▼ (状態の遷移)
┌──────────────────────────────────────┐
│  Step 4: 自己評価ノード (Grader)     │
│  ・回答の妥当性評価（Pass / Fail）    │
└──────────┬────────────────┬──────────┘
           │ (Fail / 3回未満)  │ (Pass / 3回以上)
           ▼                ▼
┌──────────────────┐  ┌──────────────────┐
│  再クエリ・再構成│  │  最終回答の出力  │
│  (Transform)     │  │  (END)           │
└──────────┬───────┘  └──────────────────┘
           │
           └───（ループバック）───┘

Step 1: データソースの整理と構造化チャンキング（LlamaIndex）

外部知識（社内規定、仕様書、財務データなど）を、LlamaIndexを用いてパースし、ベクトルデータベースにインデックス化します。メタ情報を付与し、セマンティックな検索を可能にします。

Step 2: サブクエリエンジンの構築（LlamaIndex）

ユーザーの曖昧な「複合質問」に対し、複数のインデックスへと自律的に検索を振り分けるための SubQuestionQueryEngine を初期化します。

Step 3: 状態（State）と自己修正ワークフローの定義（LangGraph）

ワークフロー全体で共有する状態オブジェクト AgentState（TypedDict）を定義します。クエリ、現在の回答、評価ステータス、そしてループ数を管理するための変数を格納します。

Step 4: 自己評価ロジック（Grader）の実装

生成された回答が「本当にユーザーの質問の意図を満たしているか」「データなし等の不十分な情報になっていないか」を評価するLLMベースのGrader（品質評価ノード）を定義します。

Step 5: 評価の統合と例外処理の設計

回答品質を検証すると同時に、エージェントが無限ループに陥るのを防止するため、一定回数で強制終了する「ガードレール（最大ループカウンタ）」を設計します。

【実践テンプレート】Pythonによる実装コード

以下は、LlamaIndexの SubQuestionQueryEngine による複雑なドキュメント検索と、LangGraphによる自己修正ループを統合した実践的な実装例です。

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: LangGraph によるエージェント状態(State)と自己修正ループ
# =====================================================================

# 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"--- 検索と回答生成を実行中 (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）」による判定を推奨します。
    # ここでは簡易判定として、情報欠落を示すキーワードや短すぎる文章を '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,                 # ワークフロー終了
        "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"])

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

本アーキテクチャを実務のプロダクション環境に適用する際、開発者が直面しやすい3つの技術的課題とその克服方法を解説します。

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

  • 課題： 評価器（Grader）の基準が厳格すぎると、LLMが何度も「回答不十分」と判定を繰り返し、同じ検索を無限ループしてコストが高騰します。
  • 対策： 状態管理オブジェクト（AgentState）の中に必ず loop_count などのカウンタを定義し、一定回数（例: 3回）に達した場合は、強制的にその時点での最善の回答を出力するか、「情報不足」であることを明確に記述して安全に終了（END）させるガードレールを設けてください。

2. 評価器（Grader）の判定ブレ

  • 課題： 「回答に必要な情報が十分に含まれているか？」という問いに対して、通常のLLMに自然言語で自由記述の判定をさせると、出力フォーマットが安定せず、分岐用の条件付きエッジ（Conditional Edges）が誤動作する原因になります。
  • 対策： OpenAIやAnthropic、Geminiが提供する 「Structured Outputs（構造化出力）」 機能、または Pydantic によるスキーマ定義を強制的に適用します。これにより、出力モデルの形式を {"grade": "pass" | "fail"} に固定し、例外エラーを排除します。

3. APIコストとレイテンシの増加

  • 課題： ループを何往復も重ね、各ステップで推論を繰り返すため、エンドユーザーの待ち時間が大幅に長くなります。
  • 対策： クエリの受付直後に、そもそもRAGを実行する必要のない挨拶や不要なノイズを除外する「トリアージ用の軽量ルーター」を前段に配置します。さらに、品質評価（Grader）の役割には「GPT-4o-mini」などの安価かつ高速なモデルを適用し、処理全体の速度を最大化するのがベストプラクティスです。
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?