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?

@fd_ai_teacher

評価駆動で高精度！LangGraphマルチエージェントRAGを実装する手順

0
Posted at

多くのLLMアプリケーション開発者が直面する課題、それは「複雑な推論を伴うタスクでの精度向上」ではないでしょうか。従来の単一エージェントRAGでは、複数の情報源を横断したり、多段階の思考プロセスを要する質問に答えるのが難しいケースが頻繁に発生します。

この記事では、この課題を解決するために、LangGraph を用いた マルチエージェントRAG の具体的な実装手順を解説します。さらに、本番運用を見据え、評価駆動 でシステムの精度と信頼性を担保する方法についても深掘りします。この記事を読めば、高精度なLLMアプリケーションを構築し、持続的に改善していくための実践的な知見が得られるでしょう。

なぜ今、LangGraphとマルチエージェントRAGなのか

単一のRAG（Retrieval Augmented Generation）システムでは、ユーザーの複雑なクエリに対して、単一の情報検索と生成プロセスで対応しようとします。しかし、現実世界の質問はしばしば多角的で、異なる専門知識を持つ複数のエージェントが協調して回答を導き出す方が効率的かつ高精度な場合があります。

LangGraphは、エージェント間の協調をグラフ構造で表現し、柔軟なワークフローを構築するためのフレームワークです。これにより、以下のようなメリットが得られます。

  • 複雑な推論の実現: 複数のエージェントがそれぞれ得意なタスク（情報検索、要約、検証など）を担当し、互いに情報を共有しながら段階的に問題を解決できます。
  • モジュール性と拡張性: 各エージェント（ノード）が独立したロジックを持つため、システムの変更や機能追加が容易です。
  • 耐久性と監視: LangGraphのチェックポイント機能やLangSmithとの連携により、エージェントの実行状態を永続化し、デバッグや監視が容易になります。

本記事では、このLangGraphの特性を活かし、情報検索と生成プロセスを複数のエージェントに分散させる マルチエージェントRAG システムの構築に焦点を当てます。

LangGraphの基礎: コアコンポーネントとグラフの定義

LangGraphを使ったマルチエージェントRAGを実装する前に、LangGraphの基本的な概念を理解しておきましょう。LangGraphはワークフローをグラフとしてモデル化し、以下の3つの主要コンポーネントで構成されます。

State（状態）

アプリケーションの現在のスナップショットを表す共有データ構造です。ノード間で情報を受け渡すための唯一の手段であり、TypedDict、Pydantic、dataclassesがサポートされています。状態を最小限に保ち、明示的に型付けすることがベストプラクティスです。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class AgentState(TypedDict):
    """
    エージェントの共有状態を定義します。
    messages: 会話履歴を保持し、新しいメッセージが追加されるたびに履歴に追加されます。
    """
    messages: Annotated[list[BaseMessage], add_messages]
    # 必要に応じて、RAGの検索結果や追加のフラグなどをここに追加できます。
    # 例えば、retrieval_result: str など。

Annotated[list[BaseMessage], add_messages] は、messages リストに新しい BaseMessage が追加された際に、既存のリストに追記する形で状態を更新するためのLangGraphのヘルパーです。

Nodes（ノード）

エージェントのロジックをエンコードする関数です。現在の状態を入力として受け取り、更新された状態を返します。各ノードは特定のタスクを実行する責務を持ちます。

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import os

# 環境変数設定 (OpenAI APIキー)
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# ツール定義の例
@tool
def search_wikipedia(query: str) -> str:
    """Wikipediaで情報を検索するツール"""
    print(f"Searching Wikipedia for: {query}")
    # 実際にはWikipedia APIなどを呼び出す
    return f"Result for '{query}': Example Wikipedia content."

tools = [search_wikipedia]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

# LLMを呼び出すノード
def call_llm(state: AgentState) -> AgentState:
    """LLMを呼び出し、応答を状態に保存するノード"""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# ツールを実行するノード (LangGraphのToolNodeを利用)
# ToolNodeは、LLMが生成したtool_callsを自動的に実行し、ToolMessageを状態に追加します。

Edges（エッジ）

現在の状態に基づいて次にどのノードを実行するかを決定する関数です。条件分岐や固定遷移が可能です。これにより、ワークフローの動的なルーティングが可能になります。

from typing import Literal

def should_continue(state: AgentState) -> Literal["tool_node", "end_node"]:
    """
    LLMからの最後のメッセージに基づいて、次に進むべきノードを決定する条件エッジ。
    ツール呼び出しがある場合は'tool_node'へ、そうでなければ'end_node'へ。
    """
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return "end_node"

グラフの構築例

これらのコンポーネントを組み合わせて、基本的なグラフを構築します。

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

# グラフの構築
workflow = StateGraph(AgentState)

# ノードの追加
workflow.add_node("llm_agent", call_llm)
workflow.add_node("tool_executor", ToolNode(tools)) # ツールを実行するノード

# エントリポイントとエッジの定義
workflow.set_entry_point("llm_agent")

# 条件付きエッジ: LLMがツールを呼び出すか否かで分岐
workflow.add_conditional_edges(
    "llm_agent",         # このノードの後に条件を評価
    should_continue,     # 評価関数
    {
        "tool_node": "tool_executor", # ツール呼び出しがあればtool_executorへ
        "end_node": END             # なければ終了
    }
)

# ツール実行後は再びLLMエージェントに戻り、ツール結果を元に思考させる（例）
workflow.add_edge("tool_executor", "llm_agent")

# グラフをコンパイル
app = workflow.compile()

# 実行例
print("--- ツール呼び出しを伴う実行 ---")
result_with_tool = app.invoke({"messages": [HumanMessage(content="Explain quantum entanglement using Wikipedia.")]})
print(result_with_tool)

print("\n--- ツール呼び出しを伴わない実行 ---")
result_no_tool = app.invoke({"messages": [HumanMessage(content="What is the capital of France?")]})
print(result_no_tool)

この例では、LLMエージェントがユーザーの質問を解析し、必要に応じてWikipedia検索ツールを呼び出します。ツールの実行後、その結果が再びLLMエージェントに渡され、最終的な回答が生成される、という簡単なマルチエージェントRAGの骨格ができています。

マルチエージェントRAGの実装パターン

より実用的なマルチエージェントRAGシステムを構築するには、複数のエージェントに異なる専門性を持たせ、協調させることが重要です。ここでは、一般的な実装パターンを紹介します。

1. ルーティングエージェントによる専門エージェントの振り分け

ユーザーのクエリを最初に受け取り、その内容に基づいて最適な専門エージェント（RAGエージェント、Web検索エージェント、要約エージェントなど）にタスクをルーティングするエージェントを配置します。

  • Router Agent: ユーザーの質問を分析し、どの専門エージェントが最も適切かを判断。
  • RAG Agent: 内部の知識ベース（ベクトルDBなど）から情報を検索し、回答を生成。
  • Web Search Agent: 最新の情報や一般的な知識が必要な場合にWeb検索を実行。
  • Summarizer Agent: 取得した情報を要約したり、複数の情報を統合したりする。

このパターンでは、LangGraphの条件付きエッジがルーティングの核となります。

2. 協調的な問題解決パターン

複数のエージェントが対話しながら問題を解決していくパターンです。例えば、RAGエージェントが情報を取得し、その情報が不十分であれば、さらにWeb検索エージェントが追加情報を収集し、最終的に統合エージェントが回答を生成する、といった流れです。

  • Query Rewriter Agent: 初期クエリをRAGに適した形に修正。
  • Retriever Agent: ベクトルデータベースから関連ドキュメントを取得。
  • Grader Agent: 取得したドキュメントがクエリに関連しているかを評価。関連性が低い場合はQuery Rewriterにフィードバック。
  • Generator Agent: 関連ドキュメントとクエリから回答を生成。
  • Fact Checker Agent: 生成された回答の事実関係を検証し、必要であれば追加の検索や修正を促す。

このパターンは、LangGraphのループ構造やHuman-in-the-Loop機能と相性が良いです。

実装時の考慮事項

  • 状態設計: 複数のエージェントが共有する情報は、AgentStateに適切に定義する必要があります。検索結果、中間的な推論、各エージェントの判断などを明確に型付けしましょう。
  • ノードの責務: 各ノード（エージェント）の責務を単一かつ明確にすることで、デバッグが容易になり、再利用性が高まります。
  • エラーハンドリング: ツール呼び出しの失敗やLLMの応答エラーなど、予期せぬ事態に備えたエラーハンドリングノードやリトライ機構を導入することが重要です。

評価駆動による精度向上: RAGとマルチエージェントの品質担保

LLMアプリケーション、特にRAGシステムにおいて、その品質を客観的に評価し、改善サイクルを回すことは非常に重要です。LangGraphのような複雑なマルチエージェントシステムでは、単一のエージェントよりも評価が難しくなりますが、その分効果も大きいです。

RAGシステムのための評価指標

RAGシステムの品質を測るには、主に以下の指標が用いられます。

  1. Faithfulness (忠実性): 生成された回答が、提供された情報源に基づいているか。幻覚（Hallucination）の有無を確認します。
  2. Context Precision (文脈の精度): 取得された情報源が、クエリに対してどの程度関連性が高いか。ノイズの少ない情報源が取得できているか。
  3. Context Recall (文脈の再現率): クエリに回答するために必要な情報が、取得された情報源にどの程度含まれているか。重要な情報が欠落していないか。
  4. Answer Relevance (回答の関連性): 生成された回答が、ユーザーのクエリに対して直接的かつ適切に回答しているか。
  5. Answer Correctness (回答の正確性): 生成された回答の事実関係が正しいか。

これらの指標は、Ragas, DeepEval, TruLens, LangSmithなどの評価フレームワークやツールを使って測定できます。

マルチエージェントシステムの評価

マルチエージェントシステムでは、上記のRAG評価指標に加えて、システム全体のタスク成功率や効率性も評価対象となります。

  • Task Success Rate (TSR): ユーザーの初期クエリに対して、システムが最終的に正しい回答を生成できた割合。
  • Efficiency: タスク完了までのステップ数、LLMのトークン使用量、実行時間など。
  • Robustness: エラー発生時の回復能力や、予期せぬ入力に対する耐性。

LangSmithを活用した評価駆動開発

LangSmithは、LangGraphアプリケーションの構築、テスト、監視のための統合開発プラットフォームです。特に評価フレームワークとトレース機能は、マルチエージェントRAGの精度向上に不可欠です。

  1. トレースとデバッグ: 各ノードの入力・出力、ツール呼び出し、LLMの応答などを詳細にトレースすることで、エージェント間の情報伝達や推論プロセスを可視化できます。これにより、問題発生時にどのエージェントで何が起きたのかを特定しやすくなります。
  2. データセットの作成と評価:
    • データセットの準備: 実際のユーザー質問と期待される回答、参照ドキュメントなどを含む評価データセットを作成します。
    • 評価の実行: LangSmith上で、このデータセットに対してLangGraphアプリケーションを実行し、RAG評価指標（Faithfulness, Context Relevanceなど）を自動的に計算します。
    • 結果の分析: 評価結果を分析し、どのケースでシステムが失敗したか、どのエージェントがボトルネックになっているかを特定します。
  3. A/Bテストと継続的改善: 異なるエージェントのロジックやプロンプト、RAGの検索戦略などを変更し、LangSmithで評価を実行することで、A/Bテスト的に改善の効果を検証できます。

Human-in-the-Loop (HITL) の導入

高精度なLLMアプリを目指す上で、人間の介入は最後の砦となります。LangGraphの割り込み機能とチェックポイントを活用することで、特定のツール呼び出しや不確実性の高い推論ステップで人間の承認を求めることができます。

# Human-in-the-Loopの実装 (再掲)
# config = {"configurable": {"thread_id": "some_id"}, "interrupt_on": ["sensitive_tool"]}
# app.stream(...) を使用して中断を検知し、人間の承認後に再開するロジックを実装

HITLは、特に初期段階でのシステム信頼性向上や、デリケートなタスクにおける安全性の確保に有効です。しかし、介入が頻繁すぎると効率が低下するため、介入のトリガー条件は慎重に設計する必要があります。

よくあるハマりどころと回避策

LangGraphを用いたマルチエージェントシステムは強力ですが、実装時にはいくつかの課題に直面する可能性があります。

1. 状態の肥大化 (State Explosion)

問題: 状態オブジェクトに不要な情報を詰め込みすぎると、パフォーマンスの低下やメモリ消費の増加、デバッグの困難につながります。
回避策:

  • 状態は最小限に保ち、明示的に型付け (TypedDict, Pydantic) します。
  • 一時的な値は関数スコープ内で渡し、状態には永続化すべき重要な情報のみを格納します。
  • add_messagesのようなリデューサーヘルパーは、本当に累積が必要な場合にのみ使用し、不要な履歴の蓄積を避けます。

2. ツールリトライストーム (Tool Retry Storms)

問題: ツール呼び出しが一時的なエラーで失敗した場合に、無限ループでリトライを繰り返してしまうことがあります。
回避策:

  • すべてのリトライラッパー（tenacityなど）に最大試行回数を設定します。
  • グラフの条件付きルーティングで、リトライが上限に達した場合にエラーハンドラーノードにルーティングする明示的な分岐を設けます。これにより、エラーを検知し、適切なリカバリ処理を実行できます。

3. チェックポイントドリフト (Checkpoint Drift)

問題: 長時間実行されるスレッドでチェックポイントのオーバーヘッドが増大し、パフォーマンスに影響を与えることがあります。
回避策:

  • DeltaChannelのような新しいチャネルタイプを使用し、各ステップで増分デルタのみを保存するようにします。これにより、変更された部分のみが永続化され、オーバーヘッドが削減されます。
  • snapshot_frequency=Kを設定してKステップごとに完全なスナップショットを書き込み、読み取りレイテンシを制限します。

4. 非同期コンテキストの伝播問題

問題: LangChain/LangGraphアプリケーションは非同期タスクに依存することが多く、OpenTelemetryのコンテキストが非同期境界を越えて自動的に伝播しないため、並行して実行されるタスクのスパンが親スパンから切断されることがあります。LangSmithでのトレースが不完全になる原因となります。
回避策:

  • OpenTelemetryのコンテキスト伝播メカニズムを正しく設定し、非同期タスク間でコンテキストが維持されるようにします。Pythonでは contextvars や opentelemetry-instrumentation-asyncio を適切に利用する必要があります。

5. LangGraphノードが表示されない/トレースが不完全

問題: 生のノード関数を記述したり、ルーティングロジックを構築したりすると、LangSmithのトレースにスパンが欠落したり、孤立したノード実行が発生したりする可能性があります。
回避策:

  • LangGraphは内部的にLangChainのRunnableプロトコルを使用しています。ノード関数をRunnableとしてラップすると、より良いトレースが得られます。
  • LangSmithなどのツールを使用して、LangGraphの実行を完全に可視化し、ノードの遷移とタイミングを追跡することで、問題の特定に役立ちます。

まとめと次の一歩

この記事では、LangGraphを用いたマルチエージェントRAGシステムの構築方法と、評価駆動による精度向上、そして本番環境での課題と回避策について解説しました。

  • LangGraph は、グラフ構造でエージェントのワークフローをモデル化し、複雑な推論を可能にします。
  • マルチエージェントRAG は、複数のエージェントが協調することで、単一エージェントでは難しい複雑な質問に対応できます。
  • 評価駆動開発 は、LangSmithなどのツールを活用し、FaithfulnessやContext Precisionといった指標を用いてシステムの品質を客観的に測定し、継続的に改善するための鍵です。

LangGraphはまだ発展途上のフレームワークであり、新しい機能が頻繁に追加されています。本番環境への導入を検討する際は、LangGraph公式ドキュメント の最新情報を常に確認し、ベストプラクティスに従うことを強くお勧めします。特に、本記事で言及したLangGraphのバージョン情報（v0.0.x系が現状の安定版であること、v1.0のリリース予定など）は、公式ドキュメントで最新を確認してください。

高精度なLLMアプリケーションを構築し、ユーザーに価値を届けるために、ぜひLangGraphと評価駆動開発の手法を取り入れてみてください。

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?