多くのRAGシステム開発者が「複雑な推論や高度な情報処理が難しい」という壁にぶつかっています。単一のLLMエージェントだけでは、動的なクエリルーティングや多段階の評価・修正が困難で、結果としてRAGの精度や信頼性が頭打ちになることが少なくありません。
この記事では、LangGraphを用いたマルチエージェントRAGシステムの設計と実装を通じて、この課題を解決する具体的な手順を解説します。複数のエージェントを協調させ、洗練されたワークフロー自動化を実現することで、従来のRAGでは困難だった高度な情報検索や処理を可能にします。
LangGraphとは？複雑なワークフローをグラフで表現する利点
このセクションでは、LangGraphの基本的な概念と、なぜ複雑なエージェントワークフローに適しているのかを解説します。
LangGraphは、LangChainエコシステムの一部でありながら、エージェントのオーケストレーションに特化した低レベルのフレームワークです。その最大の特徴は、アプリケーションのロジックを「グラフ」としてモデル化する点にあります。これにより、以下の3つのコアコンポーネントを組み合わせて、エージェントの意思決定プロセスや情報フローを視覚的かつ構造的に定義できます。
-
State (状態): アプリケーションの現在のスナップショットを表す共有データ構造です。エージェント間で共有される情報や、ワークフローの進行状況を管理します。
TypedDictやPydanticで型付けすることで、状態の整合性を保ちやすくなります。 - Nodes (ノード): グラフ内で実行される処理単位です。LLMの呼び出し、外部APIの実行、データ処理、条件分岐の評価など、エージェントの具体的なロジックを実装します。現在の状態を入力として受け取り、計算や副作用を実行し、更新された状態を返します。
- Edges (エッジ): ノード間の遷移を定義します。現在の状態に基づいて次に実行するノードを決定する「条件付きエッジ」や、固定された順序で実行する「固定エッジ」があります。これにより、動的な意思決定や反復処理を持つマルチエージェントワークフローを柔軟に構築できます。
LangGraph v1は安定版であり、LangChain v1と連携するように設計されています。LangChainのcreate_agentはLangGraph上に構築されており、高レベルな開発から始めて、必要に応じて低レベルな制御に移行できるのが強みです。
なぜマルチエージェントRAGにLangGraphが適しているのか
従来のRAGシステムでは、単一のエージェントが「クエリの解釈」「ドキュメント検索」「回答生成」といった一連のタスクを順次実行することが一般的でした。しかし、複雑な質問や動的なコンテキストでは、この線形的なアプローチでは限界があります。
マルチエージェントRAGは、複数の専門エージェントが協調することで、この課題を克服します。例えば、以下のようなシナリオが考えられます。
- ルーターエージェント: ユーザーの質問を分析し、「事実質問」「意見質問」「ウェブ検索が必要な質問」などに分類。
- 検索エージェント: 分類された質問に基づいて、適切な検索戦略（ベクトルDB検索、キーワード検索、ウェブ検索など）を実行。
- 評価エージェント: 検索結果や生成された回答の関連性、正確性を評価し、必要に応じて修正を指示。
- 回答生成エージェント: 最終的な回答を生成。
LangGraphは、このようなエージェント間の複雑な連携や動的なルーティングを、明確なグラフ構造で表現・実行できるため、マルチエージェントRAGの構築に最適なフレームワークと言えます。
LangGraphの基本パターンでワークフローを構築する
ここからは、LangGraphの基本的な使い方をコード例を交えて解説します。まずはシンプルなグラフの構築から始めましょう。
状態の定義とノードの実装
LangGraphでワークフローを構築する最初のステップは、共有状態を定義し、個々の処理ロジックをノードとして実装することです。状態はTypedDictで型付けすることで、堅牢性と可読性を高めます。
from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, START, END
# 1. 状態の定義: アプリケーションの共有データをここに集約
class AgentState(TypedDict):
"""
RAGエージェントの状態を定義するTypedDict。
エージェント間で共有される情報や、ワークフローの進行状況を管理します。
"""
query: str # ユーザーからの元のクエリ
documents: List[str] # 検索されたドキュメントのリスト
answer: str # 生成された回答
history: List[Dict[str, Any]] # 会話履歴やステップ履歴など
# エラーハンドリング用のフィールドも追加可能
error_message: str | None
# 2. ノード関数の定義例: 各エージェントの処理ロジック
def retrieve_documents(state: AgentState) -> AgentState:
"""
クエリに基づいてドキュメントを検索するノード。
実際にはベクトルDB検索やキーワード検索などを実行します。
"""
print(f"--- Retrieving documents for query: {state['query']} ---")
# 実際にはここで検索ロジックを実装
retrieved_docs = [
f"Document 1 related to {state['query']}",
f"Document 2 related to {state['query']}"
]
return {"documents": retrieved_docs}
def generate_answer(state: AgentState) -> AgentState:
"""
検索されたドキュメントとクエリに基づいて回答を生成するノード。
実際にはLLMを呼び出して回答を生成します。
"""
print(f"--- Generating answer based on documents: {state['documents']} ---")
if not state.get("documents"):
return {"answer": "No documents found to generate an answer."}
# 実際にはLLM呼び出しロジック
generated_answer = (
f"Based on documents {state['documents']}, "
f"the answer to '{state['query']}' is a synthesized response."
)
return {"answer": generated_answer}
def final_output(state: AgentState) -> AgentState:
"""
最終的な出力を行うノード。
"""
print(f"--- Final Answer: {state['answer']} ---")
return state # 最終状態をそのまま返す
AgentStateは、複数のエージェントが共有する「状態」を定義しています。retrieve_documentsとgenerate_answerは、それぞれ特定のタスクを実行する「ノード」の例です。これらのノードはAgentStateを入力として受け取り、処理後に更新されたAgentStateを返します。
グラフの構築と実行
ノードが定義できたら、それらを組み合わせてグラフを構築します。StateGraphクラスを使用し、add_nodeでノードを追加し、add_edgeでノード間の遷移を定義します。STARTとENDは、グラフの開始と終了を示す特別なエッジです。
# 3. グラフの構築
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_documents)
workflow.add_node("generate", generate_answer)
workflow.add_node("final_output", final_output) # 最終出力ノードを追加
# エッジの定義: ワークフローの順序を決定
workflow.add_edge(START, "retrieve") # グラフ開始 -> ドキュメント検索
workflow.add_edge("retrieve", "generate") # ドキュメント検索 -> 回答生成
workflow.add_edge("generate", "final_output") # 回答生成 -> 最終出力
workflow.add_edge("final_output", END) # 最終出力 -> グラフ終了
# グラフのコンパイル
graph = workflow.compile()
# 4. グラフの実行
initial_state = {"query": "What is LangGraph?", "documents": [], "answer": "", "history": []}
result = graph.invoke(initial_state)
print("\n--- Execution Result ---")
print(f"Query: {result['query']}")
print(f"Documents: {result['documents']}")
print(f"Answer: {result['answer']}")
このコードを実行すると、retrieve_documentsノードが実行され、次にgenerate_answerノード、そしてfinal_outputノードが順に実行され、最終的な回答が出力されます。これは最もシンプルな線形的なワークフローの例です。
マルチエージェントRAGの高度なワークフロー設計と実装
より複雑な推論を必要とするRAGシステムでは、複数のエージェントが協調し、動的にタスクを委譲したり、結果を評価・修正したりするマルチエージェントワークフローが不可欠です。
条件付きエッジによる動的ルーティング
LangGraphの強力な機能の一つが「条件付きエッジ」です。これにより、現在の状態に基づいて次に実行するノードを動的に決定できます。これは、マルチエージェントシステムにおける「ルーターエージェント」の実装に非常に有効です。
例えば、ユーザーのクエリが「単純な質問」か「複雑な質問」かによって、異なる処理フローにルーティングするケースを考えてみましょう。
from typing import Literal # Python 3.8+ for Literal
from langgraph.graph import ConditionalEdge
# 状態定義はAgentStateを再利用
# 新しいノードの定義
def decide_route(state: AgentState) -> Literal["simple_qa", "complex_rag"]:
"""
クエリの複雑性に基づいて次のルートを決定するノード。
実際にはLLMを使ってクエリ分類を行う。
"""
print(f"--- Deciding route for query: {state['query']} ---")
if "complex" in state["query"].lower() or "?" not in state["query"]:
return "complex_rag" # 複雑なクエリは詳細なRAGへ
return "simple_qa" # 単純なクエリは直接回答へ
def simple_qa_agent(state: AgentState) -> AgentState:
"""
単純な質問に直接回答するエージェント（LLMのパラメトリック知識を利用）。
"""
print(f"--- Simple QA Agent processing: {state['query']} ---")
return {"answer": f"Simple answer to: {state['query']}"}
def complex_rag_coordinator(state: AgentState) -> AgentState:
"""
複雑なRAGワークフローを開始するコーディネーターノード。
ここでは便宜上、単なるマーカーとして機能。
"""
print(f"--- Complex RAG Coordinator started for: {state['query']} ---")
return state
# グラフの再構築
workflow_complex = StateGraph(AgentState)
workflow_complex.add_node("decide_route", decide_route)
workflow_complex.add_node("simple_qa", simple_qa_agent)
workflow_complex.add_node("retrieve", retrieve_documents) # 既存のノードを再利用
workflow_complex.add_node("generate", generate_answer) # 既存のノードを再利用
# 開始からルート決定へ
workflow_complex.add_edge(START, "decide_route")
# 条件付きエッジ: decide_routeの結果に基づいて遷移
workflow_complex.add_conditional_edges(
"decide_route",
decide_route, # ルーティング関数を再利用
{
"simple_qa": "simple_qa",
"complex_rag": "retrieve" # 複雑な場合はretrieveノードへ
}
)
# 単純QAの終了
workflow_complex.add_edge("simple_qa", END)
# 複雑RAGのフロー（既存のretrieve -> generate -> END）
workflow_complex.add_edge("retrieve", "generate")
workflow_complex.add_edge("generate", END) # generateから直接ENDへ
graph_complex = workflow_complex.compile()
# 実行例1: 単純なクエリ
print("\n--- Running with simple query ---")
result_simple = graph_complex.invoke({"query": "What is a cat?", "documents": [], "answer": "", "history": []})
print(f"Final Answer (Simple): {result_simple['answer']}")
# 実行例2: 複雑なクエリ
print("\n--- Running with complex query ---")
result_complex = graph_complex.invoke({"query": "Explain the complex history of quantum mechanics.", "documents": [], "answer": "", "history": []})
print(f"Final Answer (Complex): {result_complex['answer']}")
decide_routeノードが、クエリの内容に応じてsimple_qaまたはretrieveノードに処理をルーティングします。このように、条件付きエッジを使うことで、動的で柔軟なマルチエージェントワークフローを構築できます。
フォールトトレランスとエラーハンドリング
本番環境で堅牢なマルチエージェントRAGを運用するには、エラーハンドリングが不可欠です。LangGraphは、フォールトトレランスのための3つの強力なプリミティブを提供します。
-
RetryPolicy: 一時的なエラー（ネットワーク障害、レート制限など）に対して、自動的にノードの実行をリトライします。バックオフ（再試行間隔の延長）やジッター（ランダムな遅延）を設定できます。 -
TimeoutPolicy: ノードの実行時間に対して上限を設定します。長時間の処理やハングアップを防ぎます。 -
error_handler: リトライが尽きた後や、特定のタイプのエラーが発生した場合に実行されるノードです。エラー情報を状態に記録したり、別のエージェントに通知したりするのに使用できます。
これらの機能を活用して、LLM呼び出し時のエラーをハンドリングする例を見てみましょう。
from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeError
import random
# AgentStateにerror_messageフィールドがあることを前提とする
def call_llm_with_retry(state: AgentState) -> AgentState:
"""
LLM呼び出しを模倣し、ランダムにエラーを発生させるノード。
"""
print(f"--- Calling LLM for query: {state['query']} ---")
if random.random() < 0.6: # 60%の確率で一時的なエラー
raise ValueError("LLM API call failed temporarily due to network instability!")
return {"answer": f"LLM successfully generated response for: {state['query']}"}
def handle_model_failure(error: NodeError, state: AgentState) -> AgentState:
"""
LLM呼び出しノードでエラーが発生した場合に実行されるハンドラ。
エラーメッセージを状態に記録し、人間による介入を促す。
"""
print(f"!!! Error in node '{error.node}': {error.error} !!!")
return {"error_message": f"Failed to call LLM after multiple retries. Error: {str(error.error)}"}
# グラフの再構築（エラーハンドリング版）
workflow_error = StateGraph(AgentState)
workflow_error.add_node("call_llm_robust",
call_llm_with_retry,
retry_policy=RetryPolicy(max_attempts=3, backoff_factor=1.5), # 最大3回リトライ、1.5倍のバックオフ
timeout=TimeoutPolicy(run_timeout=15), # 15秒でタイムアウト
error_handler=handle_model_failure # エラーハンドラを指定
)
workflow_error.add_edge(START, "call_llm_robust")
# エラーハンドリング後にグラフを終了させるか、別のノードに遷移させるかを決定
# ここではエラーハンドラが実行されたら、最終出力へ
def check_error_status(state: AgentState) -> Literal["has_error", "no_error"]:
if state.get("error_message"):
return "has_error"
return "no_error"
workflow_error.add_conditional_edges(
"call_llm_robust",
check_error_status,
{
"has_error": END, # エラーがあれば終了
"no_error": END # 成功しても終了 (本来は次の処理へ)
}
)
graph_error = workflow_error.compile()
# 実行例: 複数回実行してリトライとエラーハンドリングの動作を確認
print("\n--- Running with error handling (may retry or fail) ---")
for i in range(3):
print(f"\n--- Test Run {i+1} ---")
initial_state_err = {"query": "Explain something important.", "documents": [], "answer": "", "history": [], "error_message": None}
result_err = graph_error.invoke(initial_state_err)
if result_err.get("error_message"):
print(f"Handled error: {result_err['error_message']}")
else:
print(f"Success: {result_err['answer']}")
この例では、call_llm_with_retryノードでランダムにエラーが発生した場合、RetryPolicyに従って最大3回リトライが試みられます。それでも解決しない場合はhandle_model_failureが実行され、エラーメッセージが状態に記録されます。これにより、一時的な障害に対する耐性を高め、ワークフローの信頼性を向上させることができます。
LangGraph CLIとLangSmithによるデバッグと運用
複雑なマルチエージェントワークフローを開発する上で、デバッグと監視は非常に重要です。LangGraphは、開発効率を高めるためのCLIツールと、LangSmithとの連携による強力なデバッグ機能を提供します。
LangGraph CLIで開発サーバーを起動
langgraph-cliを使用すると、開発サーバーを起動し、コードの変更を自動的にリロードしながらグラフのUIをブラウザで確認できます。
# langgraph-cli[inmem] をインストール
pip install -U "langgraph-cli[inmem]"
# (推奨) 仮想環境を使用し、依存関係をインストール
# プロジェクトのルートディレクトリで実行
python -m venv .venv
source .venv/bin/activate
pip install . # プロジェクトの依存関係をインストール
# LangGraph CLIの設定ファイル (langgraph.json) を作成
# 例えば、以下の内容でプロジェクトルートに保存
# {
# "entrypoint": "your_module:graph_error", # ここは上記のgraph_errorオブジェクトを指すように変更
# "graph_type": "compiled_runnable",
# "input_schema": {
# "type": "object",
# "properties": {
# "query": {"type": "string"},
# "documents": {"type": "array", "items": {"type": "string"}},
# "answer": {"type": "string"},
# "history": {"type": "array", "items": {"type": "object"}},
# "error_message": {"type": ["string", "null"]}
# }
# },
# "output_schema": {
# "type": "object",
# "properties": {
# "query": {"type": "string"},
# "documents": {"type": "array", "items": {"type": "string"}},
# "answer": {"type": "string"},
# "history": {"type": "array", "items": {"type": "object"}},
# "error_message": {"type": ["string", "null"]}
# }
# }
# }
# 開発サーバーを起動
# 例えば、上記のグラフを定義したPythonファイルが `my_graph.py` で、
# `graph_error` という変数名でコンパイルされたグラフがある場合:
# langgraph dev --config ./langgraph.json --entrypoint my_graph:graph_error
langgraph devコマンドを実行すると、ローカルサーバーが起動し、ブラウザでグラフの実行状況をリアルタイムに確認できます。これは、複雑なマルチエージェントワークフローの構造を理解し、デバッグする上で非常に役立ちます。
LangSmithによる詳細なトレースとデバッグ
LangSmithは、LangChainおよびLangGraphアプリケーションのデバッグ、テスト、監視のためのプラットフォームです。LangGraphの実行はLangSmithに自動的にトレースされ、各ノードの入力、出力、実行時間、エラーなどを詳細に確認できます。
- トレースの視覚化: グラフの各ステップで何が起こったのか、どのエージェントがどのツールを呼び出したのかを視覚的に追跡できます。
- 状態の変化の監視: 各ノードの実行前後の状態の変化を比較し、意図しない状態変更やデータ損失を特定できます。
-
無限ループの検出: LangSmithは、グラフが無限ループに陥っていることを検出し、どのノードでループが発生しているかを特定するのに役立ちます。
recursion_limitを設定することで、無限ループを早期に停止させることも可能です。
LangSmithを統合するには、環境変数にLANGCHAIN_TRACING_V2=trueとLANGCHAIN_API_KEY、LANGCHAIN_PROJECTを設定するだけです。
まとめ：LangGraphで実現する高度なマルチエージェントRAG
この記事では、LangGraphを用いたマルチエージェントRAGの設計とワークフロー自動化について、その基本的な概念から高度な実装、そしてデバッグ方法までを解説しました。
重要なポイントをまとめると以下のようになります。
- LangGraphは、状態、ノード、エッジを組み合わせることで、複雑なエージェントワークフローをグラフとして表現します。これにより、マルチエージェントRAGの動的なルーティングや協調動作を柔軟に構築できます。
- 条件付きエッジを活用することで、ユーザーのクエリや中間結果に基づいて、エージェントが動的に処理フローを決定する「ルーターエージェント」を実装できます。
- **
RetryPolicy、TimeoutPolicy、error_handler**といった組み込みのフォールトトレランス機能により、本番環境で堅牢なRAGシステムを運用するためのエラーハンドリングが可能です。 - LangGraph CLIとLangSmithは、開発中のデバッグや本番環境での監視において、ワークフローの可視化と問題特定に強力な支援を提供します。
LangGraphの低レベルな設計思想は、高い柔軟性を提供し、アダプティブRAG、修正RAG、自己反省RAGといった高度なRAGパターンを実装する基盤となります。ぜひ、このフレームワークを活用して、あなたのRAGシステムを次のレベルへと進化させてください。
さらに深く学びたい方は、LangGraphの公式ドキュメントを参照し、より複雑なエージェントパターンや永続化の仕組みについて探求することをお勧めします。