0
1

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

LangGraphで堅牢なマルチエージェントRAGを実装する手順

「RAGの回答精度がいまいち上がらない」「複雑なタスクをLLMに自動で実行させたいけれど、途中でエラーになると困る」――そんな課題に直面しているエンジニアは多いのではないでしょうか？ 従来のRAGシステムやシンプルなLLMチェインでは、特定のシナリオで限界を感じることがあります。

この記事では、LangGraph を用いて、より堅牢でインテリジェントなマルチエージェントRAGシステムを構築する具体的な手順と、その設計思想を解説します。複雑なタスクを複数の専門エージェントに分割し、動的な意思決定と状態管理を組み合わせることで、RAGの精度向上と複雑なタスクの自動化を両立する道筋が見えてくるでしょう。本記事を読むことで、LangGraphの基本から実践的な設計、そして本番運用における考慮事項までを理解し、自身のLLMアプリケーション開発に活かせるようになります。

LangGraphとは？ なぜマルチエージェントRAGにLangGraphを使うのか

このセクションでは、LangGraphの基本的な概念と、なぜ複雑なマルチエージェントRAGシステムにおいてLangGraphが強力なツールとなるのかを解説します。

LangGraphは、LangChainエコシステムの一部として提供されるライブラリで、ステートフルなLLMアプリケーションを構築するためのフレームワークです。有向非巡回グラフ（DAG）の概念に基づき、複数のノード（計算ステップ）とエッジ（ノード間の遷移）を定義することで、複雑なワークフローを表現できます。

LangGraphがマルチエージェントRAGに適している理由は以下の点にあります。

  • 耐久性のある実行: ワークフローの途中でエラーが発生しても、チェックポインティング機能により中断した状態から再開できます。これは、長時間実行される複雑なRAGプロセスにおいて非常に重要です。出典: LangGraph Documentation - Why LangGraph?
  • 動的なルーティング: 条件付きエッジを使用することで、現在の状態（例: ユーザーのクエリの種類、RAG検索結果の品質）に基づいて、次にどのエージェント（ノード）を実行するかを動的に決定できます。これにより、固定的なチェインでは不可能な、柔軟な意思決定フローを実現します。出典: LangGraph Documentation - Conditional Edges
  • 包括的な状態管理: システム全体の「記憶」として機能する状態オブジェクトを定義し、ノード間で情報を共有・更新できます。これにより、複数のエージェントが協調して作業を進める際に、一貫したコンテキストを維持できます。出典: LangGraph Documentation - State
  • Human-in-the-Loop (HITL): 必要に応じて人間の介入をワークフローに組み込むことが可能です。高リスクな判断や機密情報の取り扱いにおいて、AIの判断を人間がレビュー・修正する仕組みを容易に実装できます。出典: LangGraph Documentation - Human-in-the-Loop

これらの機能により、単一のLLMやシンプルなチェインでは実現が難しい、複雑な意思決定、エラー耐性、そして人間との協調を必要とするマルチエージェントRAGシステムを構築する基盤を提供します。

前提知識と環境準備

このセクションでは、本記事の実装例を試すために必要な前提知識と環境構築について説明します。

前提知識

  • Pythonの基本的な文法: コード例はPythonで記述されています。
  • LangChainの基本: LangGraphはLangChainエコシステムの一部であるため、LLMやRetrieverといったLangChainのコンポーネントに関する基本的な理解があると、よりスムーズに理解できます。
  • RAG (Retrieval Augmented Generation) の概念: 外部知識ベースから情報を取得し、LLMの生成能力を強化するRAGの仕組みを理解していることが望ましいです。

環境準備

Python環境のセットアップと、必要なライブラリのインストールを行います。

# 仮想環境の作成とアクティベート
python -m venv .venv
source .venv/bin/activate # macOS/Linux
# .venv\Scripts\activate # Windows

# 必要なライブラリのインストール
pip install langchain langgraph langchain-openai "langchain_core>=0.1.13" # 最新のLangChainとLangGraph、OpenAI連携
pip install python-dotenv # 環境変数管理用
pip install pydantic # 状態定義の型チェック用 (LangChain内部で利用)
pip install graphviz # グラフの可視化用 (任意)

OPENAI_API_KEY を .env ファイルに設定しておきます。

OPENAI_API_KEY="YOUR_OPENAI_API_KEY"

LangGraphによるマルチエージェントRAGの構築

ここでは、LangGraphを用いたマルチエージェントRAGシステムの具体的な実装例を通じて、主要なコンポーネントの定義と連携方法を解説します。

システムのアーキテクチャ概要

今回のマルチエージェントRAGシステムは、以下のエージェント（ノード）で構成されます。

  1. Query Router Agent (llm_agent): ユーザーからの最初のクエリを受け取り、LLMの判断に基づいて、次にどの専門エージェントに処理を委譲するかを決定します。
    • クエリが簡単な挨拶であれば直接応答。
    • 外部知識が必要なRAGクエリであればRetriever Agentへ。
    • 一般的な知識で回答可能であればAnswer Generation Agentへ。
  2. Retriever Agent (retriever_agent): 外部のベクトルデータベースなどから、関連するドキュメントを検索します。
  3. Answer Generation Agent (answer_agent): 検索されたドキュメントと元のクエリに基づいて、最終的な回答を生成します。

このアーキテクチャにより、クエリの種類に応じて最適な処理フローを動的に選択し、効率的かつ高精度なRAGを実現します。

状態の定義

LangGraphでは、システム全体の状態をTypedDictで定義します。これにより、各ノード間で共有される情報が明示化され、型安全性が向上します。特にリストのようなフィールドはAnnotatedとReducer ("append") を使用し、追記動作を明示的に指定することがベストプラクティスです。出典: LangGraph Documentation - State

from typing import TypedDict, List, Annotated, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, END
import os
from dotenv import load_dotenv

load_dotenv()

# 状態の定義
class AgentState(TypedDict):
    """
    各エージェントが共有する状態を定義します。
    メッセージの履歴、検索されたドキュメント、元のクエリ、次に実行するエージェントの指示を含みます。
    """
    messages: Annotated[List[BaseMessage], "append"]
    documents: List[str] # RAGに必要な追加の状態（例：検索結果、ドキュメントなど）
    query: str
    # 次に実行するエージェントを決定するための状態。Literalで可能な値を制限。
    next_agent: Literal["retrieve_documents", "answer_generation", "end"]

messages: Annotated[List[BaseMessage], "append"] は、新しいメッセージが追加されるたびに既存のリストに追記されることを意味します。next_agentは、条件付きエッジでルーティングを行うための重要な状態です。

LLMとプロンプトの初期化

各エージェントで使用するLLMとプロンプトを準備します。

# LLMの初期化
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ルーターエージェント用のプロンプト
router_prompt = ChatPromptTemplate.from_messages([
    ("system", """あなたはユーザーのクエリを分析し、最適な次のステップを決定するルーターです。
    クエリが挨拶や一般的な会話であれば「end」を返してください。
    クエリが特定の情報検索を必要とするRAGクエリであれば「retrieve_documents」を返してください。
    クエリが一般的な知識で直接回答できるものであれば「answer_generation」を返してください。
    回答は次のいずれかの文字列のみでお願いします: "retrieve_documents", "answer_generation", "end"
    """),
    ("user", "{query}")
])
router_chain = router_prompt | llm | StrOutputParser()

# 回答生成エージェント用のプロンプト
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", """あなたは優秀なアシスタントです。
    以下の「コンテキスト」と「ユーザーの質問」に基づいて、簡潔かつ正確に回答を生成してください。
    コンテキストが回答に十分でない場合でも、既知の知識で補足せず、「十分な情報がありません」といった旨を伝えてください。

    コンテキスト:
    {documents}
    """),
    ("user", "{query}")
])
answer_chain = answer_prompt | llm | StrOutputParser()

ノード関数の定義

各エージェントのロジックは、状態を受け取り、新しい状態の一部を返す関数として定義します。

# ノード関数の例
def call_router_agent(state: AgentState) -> dict:
    """
    ユーザーのクエリを分析し、次に実行するエージェントを決定するルーターエージェント。
    """
    print("---Query Router Agent実行中---")
    query = state["messages"][-1].content # 最新のユーザーメッセージをクエリとする
    
    # ルーターLLMを呼び出し、次のエージェントを決定
    next_agent_decision = router_chain.invoke({"query": query})
    
    print(f"ルーターの決定: {next_agent_decision}")

    # 応答メッセージを追加（ここではルーターの決定をメッセージに含める）
    new_messages = [AIMessage(content=f"ルーター: 次は {next_agent_decision} を実行します。")]
    
    # next_agentの状態を更新して返す
    return {"messages": new_messages, "next_agent": next_agent_decision, "query": query}

def retrieve_documents_agent(state: AgentState) -> dict:
    """
    RAGの検索ロジックを実装するエージェント。
    ここではダミーのドキュメントを返すが、実際にはベクトルDBからの検索を行う。
    """
    print("---Retriever Agent実行中---")
    query = state["query"]
    
    # 実際にはここでRetrieverを呼び出し、関連ドキュメントを取得
    # 例: retriever = create_retriever(...)
    # retrieved_docs = retriever.invoke(query)
    # documents_content = [doc.page_content for doc in retrieved_docs]

    # デモンストレーションのため、固定値を返す
    documents_content = [
        f"LangGraphは、有向非巡回グラフ（DAG）を用いてLLMアプリケーションを構築するためのフレームワークです。",
        f"LangGraphの主要な機能には、状態管理、条件付きエッジ、チェックポインティング、Human-in-the-Loopがあります。",
        f"LangGraphはLangChainと密接に統合されており、複雑なエージェントワークフローを構築するのに適しています。"
    ]
    
    new_messages = [AIMessage(content=f"リトリーバー: {len(documents_content)}件のドキュメントを検索しました。")]
    
    # 検索結果と次のエージェントを更新して返す
    return {"messages": new_messages, "documents": documents_content, "next_agent": "answer_generation"}

def answer_generation_agent(state: AgentState) -> dict:
    """
    検索されたドキュメントとクエリに基づいて最終的な回答を生成するエージェント。
    """
    print("---Answer Generation Agent実行中---")
    query = state["query"]
    documents = "\n".join(state["documents"])
    
    # 回答生成LLMを呼び出す
    final_answer = answer_chain.invoke({"query": query, "documents": documents})
    
    new_messages = [AIMessage(content=f"回答生成: {final_answer}")]
    
    # 最終的な回答と次のエージェントを更新して返す
    # ここでnext_agentを"end"に設定することで、グラフの終了を指示
    return {"messages": new_messages, "next_agent": "end"}

def final_response_agent(state: AgentState) -> dict:
    """
    最終的な応答をユーザーに返す前に、必要に応じて整形するエージェント。
    """
    print("---Final Response Agent実行中---")
    # ここでは最後のAIメッセージを最終応答と仮定
    final_message = state["messages"][-1].content
    print(f"最終応答: {final_message}")
    return {"messages": [AIMessage(content=final_message)]} # 最終応答のみを含む状態に更新

グラフの構築と条件付きエッジの設定

StateGraphを使用してノードとエッジを定義し、ワークフローを構築します。特に、ルーターエージェントからの出力に基づいて動的にパスを切り替えるために、条件付きエッジを使用します。

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

# ノードの追加
workflow.add_node("llm_agent", call_router_agent) # Query Router Agent
workflow.add_node("retriever_agent", retrieve_documents_agent) # Retriever Agent
workflow.add_node("answer_agent", answer_generation_agent) # Answer Generation Agent
workflow.add_node("final_response", final_response_agent) # 最終応答

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

# 条件付きエッジの定義
def route_agent(state: AgentState) -> Literal["retriever_agent", "answer_agent", END]:
    """
    状態のnext_agentに基づいて、次に実行するエージェントを決定するルーター関数。
    """
    print(f"---ルーター関数実行: next_agent={state['next_agent']}---")
    if state["next_agent"] == "retrieve_documents":
        return "retriever_agent"
    elif state["next_agent"] == "answer_generation":
        return "answer_agent"
    return END # next_agentが"end"の場合、または予期しない値の場合

workflow.add_conditional_edges(
    "llm_agent", # このノードの後に条件を評価
    route_agent, # ルーター関数
    {
        "retrieve_documents": "retriever_agent",
        "answer_generation": "answer_agent",
        END: END, # "end"が返された場合はグラフを終了
    },
)

# 固定エッジの定義
# Retriever Agent の後は必ず Answer Generation Agent へ
workflow.add_edge("retriever_agent", "answer_agent")
# Answer Generation Agent の後は最終応答へ、その後グラフ終了
workflow.add_edge("answer_agent", "final_response")
workflow.add_edge("final_response", END)

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

# グラフの可視化 (Graphvizがインストールされている場合)
# from IPython.display import Image, display
# try:
#     display(Image(app.get_graph().draw_png()))
# except Exception as e:
#     print(f"グラフの可視化に失敗しました。Graphvizが正しくインストールされているか確認してください: {e}")

# 実行例
print("\n---グラフ実行開始: RAGクエリ---")
initial_state_rag = {"messages": [HumanMessage(content="LangGraphの主要な機能について教えてください")], "documents": [], "query": "", "next_agent": ""}
for s in app.stream(initial_state_rag):
    print(s)
    print("---")
print("---グラフ実行終了: RAGクエリ---")

print("\n---グラフ実行開始: 一般知識クエリ---")
initial_state_general = {"messages": [HumanMessage(content="日本の首都はどこですか？")], "documents": [], "query": "", "next_agent": ""}
for s in app.stream(initial_state_general):
    print(s)
    print("---")
print("---グラフ実行終了: 一般知識クエリ---")

print("\n---グラフ実行開始: 挨拶クエリ---")
initial_state_greeting = {"messages": [HumanMessage(content="こんにちは")], "documents": [], "query": "", "next_agent": ""}
for s in app.stream(initial_state_greeting):
    print(s)
    print("---")
print("---グラフ実行終了: 挨拶クエリ---")

このコードを実行すると、各クエリがルーターエージェントによって適切にルーティングされ、RAGクエリの場合はretriever_agentとanswer_agentが順に実行される様子が確認できます。

よくあるエラー・ハマりどころと回避策

LangGraphでの開発は強力である一方で、いくつかの落とし穴があります。ここでは、特に遭遇しやすい問題とその解決策を解説します。

1. 状態スキーマの不整合や予期せぬ変更

  • ハマりどころ: TypedDictで定義した状態スキーマと異なる型のデータをノードが返したり、意図しないキーを上書きしたりすると、予期せぬエラーや動作が発生します。特にリストのような可変オブジェクトを直接変更しようとすると問題が起こりやすいです。
  • 回避策:
    • 状態スキーマはTypedDictで明示的に定義し、各フィールドの型を厳密に指定します。
    • ノード関数は、現在の状態を直接変更するのではなく、部分的な状態更新を辞書として返すようにします。LangGraphがこれを既存の状態にマージします。
    • リストのようなフィールドにはAnnotated[List[..., "append"]]のようなReducerを使用し、追記動作を明示的に指定します。これにより、並列処理や再試行時にデータの破損を防げます。出典: LangGraph Documentation - State
    • ノードの境界で入力/出力状態の検証を実装し、スキーマチェックやガードを設けることで、下流での「謎のエラー」を防ぎます。

2. 条件付きエッジのルーティングミス

  • ハマりどころ: 条件付きエッジのルーター関数が返す文字列が、定義されたノード名やENDと一致しない場合、グラフが意図しないパスに進んだり、停止したりします。特にタイポは発見しにくいサイレントなルーティングミスを引き起こします。
  • 回避策:
    • ルーター関数の戻り値にLiteral型ヒントを使用し、可能なパスを明示的に指定します（例: -> Literal["node_name_1", "node_name_2", END]）。これにより、静的解析ツールやIDEがタイポを検出できるようになります。
    • path_mapをadd_conditional_edgesに渡すことで、ルーター関数の戻り値とノード名のマッピングを明確にします。
    • LangSmithなどの可視化ツールを活用し、グラフの実行パスを視覚的に確認して、ルーティングロジックが期待通りに機能しているかをデバッグします。出典: LangGraph Documentation - LangSmith

3. 本番環境での状態の永続化とスケーリング

  • ハマりどころ: プロトタイプではMemorySaverを使用することが多いですが、本番環境ではプロセス再起動で状態が失われたり、複数のワーカー間で状態を共有できなかったりします。また、状態が肥大化するとパフォーマンス問題を引き起こす可能性があります。
  • 回避策:
    • 本番環境では、PostgresSaverやRedisなどの永続的なチェックポインターを使用します。これにより、クラッシュや再デプロイ後も状態が維持され、複数のワーカー間で状態を共有できます。出典: LangGraph Documentation - Checkpointing
    • 状態スキーマは最小限に保ち、必要な情報のみを保存します。
    • thread_idを使用して会話のターンをリンクし、チェックポインターのライフサイクルを適切に管理します。
    • 状態のサイズを制御し、古い入力の切り捨てやトークンカウンターの使用を検討します。
    • 水平スケーリングを考慮したアーキテクチャ（共有チェックポイントストレージ、ワーカー間の調整など）を設計します。出典: LangGraph Documentation - Deployment

設計上のトレードオフとベストプラクティス

LangGraphを効果的に活用するためには、設計上のトレードオフを理解し、ベストプラクティスに従うことが重要です。

トレードオフ

  • 柔軟性 vs 複雑性: LangGraphは低レベルでカスタマイズ性が高いため、複雑なワークフローを柔軟に構築できますが、その分学習曲線が急で、より多くのコード記述が必要になります。シンプルで固定的なユースケースであれば、LangChainのより高レベルな抽象化で十分な場合もあります。
  • 単一エージェント vs マルチエージェント: 単一エージェントと適切なプロンプトエンジニアリングから始めるのが良いですが、限界に達した場合は、特定の失敗を診断し、最もシンプルなマルチエージェントパターンを実装します。マルチエージェントシステムは複雑な問題を分担して解決できますが、エージェント間の調整、状態管理、デバッグがより複雑になります。

ベストプラクティス

  1. 状態設計の原則:
    • 状態オブジェクトは最小限に、明示的に、型付けされたものに保ちます。必要な情報のみを保存し、不要な情報の肥大化を防ぎましょう。
    • AnnotatedとReducerを使用して、リストのようなフィールドの更新（"append"など）を適切に処理し、並列ブランチや再呼び出しでのサイレントな上書きを防ぎます。出典: LangGraph Documentation - State
    • 状態を機密情報として扱い、PII（個人を特定できる情報）のサニタイズ、暗号化、ログのスクラブを行います。
  2. ノード関数の設計:
    • 各ノードを純粋な関数のように扱い、入力の変更ではなく部分的な状態更新を辞書として返すようにします。これにより、テストが容易になり、エッジルーティングが予測可能になります。
    • ノードの境界で入力/出力状態の検証を実装し、シンプルなスキーマチェックやガードを設けることで、下流での「謎のエラー」を防ぎます。
  3. 条件付きエッジの活用:
    • ルーティングの決定を個々のノード内ではなく、グラフ構造自体に移すことで、各ノードの焦点を絞り、全体的なフローを理解しやすく、保守しやすくします。出典: LangGraph Documentation - Conditional Edges
    • 動的なルーティング、エラーハンドリング、ユーザー入力に基づくパス、データ駆動型ロジックに活用します。
  4. Human-in-the-Loop (HITL):
    • 人間の判断が価値を付加する場所（機密性の高いアクション、PIIの使用、高リスクなツール呼び出しなど）で動的な中断を使用します。
    • 中断後、最小限のコンテキストを復元し、決定論的に続行できるように再開パスを設計します。LangGraphのチェックポインティングシステムはHITLを可能にするために構築されています。出典: LangGraph Documentation - Human-in-the-Loop
  5. エラーハンドリング戦略:
    • LangGraphの組み込みのリトライポリシー、タイムアウト、エラーハンドラーを活用し、堅牢なエージェントを構築します。
    • エラーを型付けされたオブジェクトとして状態に埋め込み、エラーハンドリングノードを通じて構造化された状態遷移を行います。出典: LangGraph Documentation - Error Handling
  6. モニタリングとデバッグ:
    • LangSmithのようなツールを使用して、複雑なエージェントの動作を可視化し、実行パスをトレースし、状態遷移をキャプチャし、詳細なランタイムメトリックを提供します。出典: LangGraph Documentation - LangSmith

まとめ

本記事では、LangGraphを用いて堅牢なマルチエージェントRAGシステムを実装する手順を解説しました。

  • LangGraphが提供する状態管理、条件付きエッジ、チェックポインティングといった機能が、複雑なLLMアプリケーション開発においていかに強力であるかを理解しました。
  • 具体的なコード例を通じて、ルーターエージェント、リトリーバーエージェント、回答生成エージェントが連携し、ユーザーのクエリに応じて動的に最適な処理フローを選択するマルチエージェントRAGシステムを構築する方法を学びました。
  • 開発中に遭遇しやすいハマりどころとその回避策、そして本番運用を見据えた設計上のトレードオフとベストプラクティスについても触れました。

LangGraphは、単なるLLMチェインでは実現が難しい、高度な意思決定能力とエラー耐性を持つAIエージェントを構築するための強力なフレームワークです。この記事が、あなたのLLMアプリ開発における次の一歩を踏み出す助けとなれば幸いです。

さらに深く学びたい場合は、LangGraphの公式ドキュメントを参照し、より複雑なエージェントパターンやデプロイメント戦略について探求することをお勧めします。

0
1
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
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?