LangGraphで堅牢なAIエージェントを構築する5つのデザインパターン

LangGraphでAIエージェントの無限ループやエラーに終止符を！堅牢なワークフローを構築する5つのデザインパターン

LLMアプリ開発において、LangChainやLangGraphを使ったAIエージェントの複雑なワークフロー構築で、「エージェントが意図しないループに陥る」「本番環境で予期せぬエラーで停止する」「状態管理が複雑すぎてデバッグできない」といった課題に直面したことはありませんか？

この記事では、LangGraph を活用し、本番運用を見据えた堅牢なAIエージェントを構築するための具体的なデザインパターンと実装例を、最新のLangGraph v1.2.0の知見も交えて解説します。読み終える頃には、あなたのAIエージェントがより安定し、保守性の高いシステムへと進化しているでしょう。

LangGraphとは？なぜ複雑なAIエージェント開発に不可欠なのか

LangGraphは、LangChainエコシステムの一部として開発された、ステートフルなAIエージェントを構築するための低レベルオーケストレーションフレームワークです。従来のLangChainの線形的なチェーンでは実現が難しかった、ループ、条件分岐、動的な実行パスといった複雑なAIワークフローを、グラフ構造としてモデル化することで可能にします。

LangChainとの違いとLangGraphの利点

LangChainが提供する高レベルな抽象化は迅速なプロトタイプ開発に適していますが、より複雑なロジックやインタラクティブな動作、永続的な状態管理が必要な場合、LangGraphはその真価を発揮します。

LangGraphの主な利点は以下の通りです。

• 複雑なワークフローの表現: ノード（タスクやアクション）とエッジ（実行フロー）を定義することで、人間による介入や動的な意思決定を含む、表現力豊かなエージェントワークフローを構築できます。
• ステートフルな実行: 共有状態（State）を通じて、エージェントは過去のコンテキストを保持し、会話履歴やタスクの進行状況に基づいて意思決定を行えます。
• 永続性と回復性: チェックポインターにより各ノードの実行後にグラフの状態が保存されるため、障害発生時の再開やHuman-in-the-Loop（HITL）ワークフローの実装が容易です。
• ストリーミング対応: ファーストクラスのストリーミングサポートにより、ユーザーエクスペリエンスを向上させるリアルタイムな応答が可能です。

LangGraphの主要な概念

LangGraph v1.2.0 (2024年5月時点) をベースに、そのコアコンセプトを理解しましょう。

• StateGraph: エージェントのロジックを定義する主要なクラスで、共有状態のスキーマを定義し、ノードとエッジを追加します。
• Nodes: グラフ内の個々のコンポーネントやエージェントを表すPython関数です。LLM呼び出し、ツール呼び出し、カスタムロジックなどがノードとして機能します。
• Edges: ノード間の接続を定義し、ワークフローの進行方法を制御します。add_edgeによる直接エッジと、add_conditional_edgesによる条件付きエッジがあります。
• State: アプリケーションの現在のスナップショットを表す共有データ構造です。TypedDictなどで定義され、すべてのノードが読み書きできます。Annotatedとoperator.addを組み合わせることで、リストなどの状態を累積的に更新できます。
• Checkpointer: 各ノードの実行後にグラフの状態を保存し、永続性と回復性を実現します。
• RetryPolicy / TimeoutPolicy: ノード実行時のエラー回復やタイムアウトを設定し、本番環境での堅牢性を高めます。

LangGraphの基本！「Hello World」とツール利用エージェントの実装

まずは、LangGraphの基本的な使い方を確認しましょう。インストールからシンプルなLLM呼び出し、そしてツール利用と条件分岐を含むエージェントの例を通して、LangGraphのコア機能を掴みます。

環境準備とインストール

uvまたはpipでlanggraphをインストールします。

uv add langgraph
# または
pip install langgraph

OpenAIモデルを使用するため、langchain-openaiもインストールし、OPENAI_API_KEY環境変数を設定してください。

シンプルなLLM呼び出しグラフ

この例では、ユーザーからのメッセージを受け取り、LLMに渡して応答を返すだけの最も基本的なAIエージェントを構築します。状態の定義、ノードの追加、エッジの設定、そしてグラフのコンパイルと実行の流れを理解します。

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator # Stateの更新にoperator.addを使用
import os

# OpenAI APIキーの設定
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 状態の定義
# AgentStateはTypedDictとして定義され、'messages'というキーを持つ。
# 'messages'はBaseMessageオブジェクトのリストであり、operator.addによって新しいメッセージが既存のリストに追加される。
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

# LLMノードの定義
# 状態を受け取り、LLMを呼び出して応答を生成し、新しいメッセージとして状態に返す。
def call_llm(state: AgentState):
    """LLMを呼び出し、応答を生成するノード"""
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0) # temperature=0で応答の安定性を高める
    response = model.invoke(state["messages"])
    print(f"[LLM] Called with: {state['messages'][-1].content}")
    print(f"[LLM] Response: {response.content}")
    return {"messages": [response]} # 新しいメッセージをリストとして返す

# グラフの構築
graph = StateGraph(AgentState)
graph.add_node("chatbot", call_llm) # 'chatbot'という名前でcall_llmノードを追加
graph.add_edge(START, "chatbot") # グラフの開始から'chatbot'ノードへ
graph.add_edge("chatbot", END)   # 'chatbot'ノードからグラフの終了へ

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

# 実行
print("--- Simple LLM Graph Execution ---")
result = app.invoke({"messages": [HumanMessage(content="こんにちは！")]})
print(f"Final response: {result['messages'][-1].content}")
# 出力例: Final response: こんにちは！何かお手伝いできることはありますか？

ツール呼び出しと条件付きエッジを持つAIエージェント

より実用的なAIエージェントとして、LLMが外部ツールを呼び出す必要があるかどうかを判断し、必要に応じてツールを実行する例です。ここで、LangGraphの重要な機能である条件付きエッジが登場します。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
import operator
import os

# OpenAI APIキーの設定 (上記で設定済みであれば不要)
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 状態の定義 (上記と同じ)
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

# ツール定義
@tool
def search_web(query: str) -> str:
    """Perform a web search and return the results."""
    # 実際のWeb検索API呼び出しをここに実装
    print(f"[Tool] Executing search_web for: '{query}'")
    # 例としてモックデータを返す
    if "capital of France" in query.lower():
        return "The capital of France is Paris."
    elif "current weather in Paris" in query.lower():
        return "The current weather in Paris is sunny with a temperature of 25 degrees Celsius."
    return f"Search results for '{query}': No specific information found."

# LLMノード (上記と同じだが、ツール利用のためChatOpenAIのtool_modeを考慮)
def call_llm_with_tools(state: AgentState):
    """LLMを呼び出し、ツール呼び出しの提案または最終応答を生成するノード"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, tools=[search_web]) # ツールをLLMに渡す
    response = llm.invoke(state["messages"])
    print(f"[LLM] Called with last message: {state['messages'][-1].content[:50]}...")
    print(f"[LLM] Response type: {type(response)}")
    if response.tool_calls:
        print(f"[LLM] Tool calls suggested: {response.tool_calls}")
    else:
        print(f"[LLM] Final answer: {response.content[:50]}...")
    return {"messages": [response]}

# ツール実行ノード
# LangGraphのprebuiltにあるToolNodeを使うと、定義したツールを簡単に実行できる。
tools = [search_web]
tool_node = ToolNode(tools)

# 条件付きエッジの決定関数
# LLMの最後のメッセージを評価し、次にどのノードに進むべきかを決定する。
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    # LLMがツール呼び出しを提案した場合
    if last_message.tool_calls:
        print(f"[Router] LLM suggested tool calls. Routing to 'tools' node.")
        return "tools"
    # LLMが最終的な回答を生成した場合
    print(f"[Router] LLM provided final answer. Routing to 'end'.")
    return "end" # ツール呼び出しがない場合は終了

# グラフの構築
workflow = StateGraph(AgentState)
workflow.add_node("llm", call_llm_with_tools) # LLMノード
workflow.add_node("tools", tool_node)       # ツール実行ノード

# STARTからLLMノードへ
workflow.add_edge(START, "llm")

# LLMノードから条件付きでツールまたは終了へ
workflow.add_conditional_edges(
    "llm",
    should_continue, # 決定関数
    {
        "tools": "tools", # 決定関数が"tools"を返したらツールノードへ
        "end": END        # 決定関数が"end"を返したら終了
    }
)
# ツール実行後、再度LLMノードに戻り、ツール結果をLLMに渡して最終回答を生成させる
workflow.add_edge("tools", "llm")

app = workflow.compile()

# 実行例1: ツール呼び出しが発生しない場合
print("\n--- Execution 1: No tool call (Simple Greeting) ---")
result1 = app.invoke({"messages": [HumanMessage(content="こんにちは！元気ですか？")]})
print(f"Final response: {result1['messages'][-1].content}")
# 出力例: Final response: はい、元気です！何かお手伝いできることはありますか？

# 実行例2: ツール呼び出しが発生する場合
print("\n--- Execution 2: With tool call (Web Search) ---")
result2 = app.invoke({"messages": [HumanMessage(content="フランスの首都は何ですか？あと、パリの現在の天気も教えてください。")]})
print(f"Final response: {result2['messages'][-1].content}")
# 出力例: Final response: フランスの首都はパリです。パリの現在の天気は晴れで、気温は25度です。

この例から、LangGraphがどのように動的なワークフローを構築し、ツール利用のような複雑な挙動を実現するかが理解できたでしょう。

LangGraphで堅牢なAIエージェントを構築する5つのデザインパターン

本番運用を見据えたAIエージェントでは、単に機能するだけでなく、エラー耐性、保守性、拡張性が求められます。ここでは、LangGraphで堅牢なAIエージェントを構築するための主要なデザインパターンを5つ紹介します。

1. ルーティングパターン (Router Pattern)

ユーザーの意図や入力に基づいて、適切なツール、サブループ、または特定のエージェントに処理を振り分けるパターンです。複雑なタスクをモジュール化し、効率的に処理するために不可欠です。

なぜ必要か: 単一のLLMやエージェントでは対応しきれない多様なユーザー要求に対し、適切な専門エージェントを動的に選択することで、応答の精度と効率を向上させます。

実装のポイント:

• 条件付きエッジ (add_conditional_edges) を活用します。
• ルーティングを決定するノード（ルーターノード）は、ユーザーのクエリと利用可能なツール/エージェントのリストをLLMに渡し、最適な次ステップを決定させます。
• ルーターノードの出力に基づいて、異なるパスへと分岐させます。

コード例: 上記のツール呼び出しエージェントのshould_continue関数が、ルーティングパターンの最もシンプルな形です。ここでは、さらに複数のツールやサブエージェントへのルーティングを考えます。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
import operator
import os

# 状態の定義
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str # 次に実行すべきノード名を格納する

# ツール定義
@tool
def search_web(query: str) -> str:
    """Perform a web search and return the results."""
    print(f"[Tool] Executing search_web for: '{query}'")
    return f"Search results for '{query}': Some relevant information."

@tool
def calculate_expression(expression: str) -> str:
    """Evaluate a mathematical expression."""
    print(f"[Tool] Executing calculate_expression for: '{expression}'")
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error calculating '{expression}': {e}"

# LLMノード
def call_llm_for_routing(state: AgentState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, tools=[search_web, calculate_expression])
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ツール実行ノード
tools = [search_web, calculate_expression]
tool_node = ToolNode(tools)

# ルーティング決定関数
def route_agent(state: AgentState) -> Literal["tools", "llm_finish"]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        print("[Router] Routing to TOOLS due to tool_calls.")
        return "tools"
    else:
        print("[Router] Routing to LLM_FINISH (final answer).")
        return "llm_finish"

# グラフの構築
router_workflow = StateGraph(AgentState)
router_workflow.add_node("llm_router", call_llm_for_routing)
router_workflow.add_node("tools", tool_node)

router_workflow.add_edge(START, "llm_router")
router_workflow.add_conditional_edges(
    "llm_router",
    route_agent,
    {
        "tools": "tools",
        "llm_finish": END # LLMが最終回答を出したら終了
    }
)
router_workflow.add_edge("tools", "llm_router") # ツール実行後、再度LLMで結果を評価

router_app = router_workflow.compile()

print("\n--- Routing Pattern Execution (Search) ---")
result_search = router_app.invoke({"messages": [HumanMessage(content="現在の日本経済について調べてください。")]})
print(f"Final response: {result_search['messages'][-1].content}")
# 出力例: Final response: Search results for '現在の日本経済について': Some relevant information. (LLMの要約)

print("\n--- Routing Pattern Execution (Calculation) ---")
result_calc = router_app.invoke({"messages": [HumanMessage(content="123 + 456 はいくつですか？")]})
print(f"Final response: {result_calc['messages'][-1].content}")
# 出力例: Final response: 123 + 456 の結果は 579 です。

2. リフレクションパターン (Reflection Pattern)

エージェントが自身の出力や行動を評価し、必要に応じて修正・改善するためにループするパターンです。特に複雑なタスクやクリエイティブなタスクで、より高品質な結果を生成するために有効です。

なぜ必要か: LLMの出力は時に不正確であったり、指示から逸脱したりすることがあります。自己評価のプロセスを組み込むことで、エージェントは自身のミスを検出し、修正する能力を獲得します。

実装のポイント:

• 「計画」ノード、「実行」ノード、「評価（リフレクション）」ノードの3つの主要なノードで構成されます。
• 評価ノードは、LLMやルールベースのロジックを用いて、実行結果の品質や指示への適合性を判断します。
• 評価結果に基づいて、再度計画ノードに戻るか、終了するかを条件付きエッジで制御します。

コード例:
ここでは、「ユーザーの質問に対して回答を生成し、その回答が質問に適切かどうかを別のLLM（リフレクション）が評価する」というシンプルなリフレクションパターンを実装します。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
import operator
import os

class ReflectionState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    generation: str # LLMが生成した回答
    reflection: str # リフレクションLLMの評価
    iterations: int # ループ回数

# 回答生成ノード
def generate_answer(state: ReflectionState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # 創造性を出すためtemperatureを上げる
    user_message = state["messages"][-1].content
    prompt = f"ユーザーからの質問: {user_message}\n\nこの質問に回答してください。"
    response = llm.invoke([HumanMessage(content=prompt)])
    print(f"[Generate] Generated answer: {response.content[:50]}...")
    return {"messages": [response], "generation": response.content, "iterations": state.get("iterations", 0) + 1}

# リフレクションノード
def reflect_on_answer(state: ReflectionState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 評価は安定性を重視
    user_message = state["messages"][0].content # 元のユーザー質問
    generated_answer = state["generation"]

    reflection_prompt = (
        f"以下のユーザーの質問に対し、生成された回答が適切かどうかを評価してください。\n"
        f"質問: {user_message}\n"
        f"回答: {generated_answer}\n"
        f"評価結果を 'ACCEPT' または 'REVISE' で始めてください。REVISEの場合、改善点を具体的に記述してください。"
    )
    reflection_response = llm.invoke([HumanMessage(content=reflection_prompt)])
    print(f"[Reflect] Reflection: {reflection_response.content[:50]}...")
    return {"reflection": reflection_response.content}

# 条件付きエッジの決定関数
def should_revise(state: ReflectionState) -> Literal["revise", "accept"]:
    if "REVISE" in state["reflection"].upper() and state["iterations"] < 3: # 最大3回までリトライ
        print("[Router] Reflection indicates REVISE. Looping back to generate.")
        # 修正指示をメッセージに追加して、次の生成に活かす
        revise_message = HumanMessage(content=f"以前の回答: {state['generation']}\n\n上記の回答について、以下の改善点が指摘されました。これらを考慮して回答を修正してください。\n指摘: {state['reflection']}")
        state["messages"].append(revise_message) # 状態を直接更新
        return "revise"
    print("[Router] Reflection indicates ACCEPT or max iterations reached. Ending.")
    return "accept"

# グラフの構築
reflection_workflow = StateGraph(ReflectionState)
reflection_workflow.add_node("generate", generate_answer)
reflection_workflow.add_node("reflect", reflect_on_answer)

reflection_workflow.add_edge(START, "generate")
reflection_workflow.add_edge("generate", "reflect")
reflection_workflow.add_conditional_edges(
    "reflect",
    should_revise,
    {
        "revise": "generate", # 修正が必要なら再度生成ノードへ
        "accept": END         # 承認されたら終了
    }
)

reflection_app = reflection_workflow.compile()

print("\n--- Reflection Pattern Execution ---")
# 例: あえて不完全な回答を生成させるような質問
result_reflect = reflection_app.invoke({"messages": [HumanMessage(content="日本の人口は？小数点以下も含めて正確に教えて。")]})
print(f"Final answer (after reflection): {result_reflect['generation']}")
# この例ではLLMの出力に依存しますが、複数回のループでより良い回答を目指します。

3. エラーハンドリングパターン (Error Handling Pattern)

本番環境では、APIレート制限、ネットワーク障害、LLMの予期せぬ出力など、様々なエラーが発生します。これらのエラーに堅牢に対応するためのパターンです。

なぜ必要か: 適切なエラーハンドリングがないと、エージェントが停止したり、ユーザーエクスペリエンスが損なわれたりします。システム全体の信頼性を確保するために不可欠です。

実装のポイント:

• RetryPolicy: LangGraphのノードに設定できるRetryPolicyを利用し、一時的なエラーに対する自動リトライを設定します。
• TimeoutPolicy: ノードの実行時間やアイドル時間の上限を設定し、ハングアップを防ぎます。
• 専用のエラーハンドラーノード: リトライが尽きた後や、特定の致命的なエラーが発生した場合に実行される、クリーンアップ、アラート、フォールバックロジックを持つノードを定義します。
• LangSmith: エラー発生時のトレースとデバッグにLangSmithを統合します。

コード例:
ここでは、RetryPolicyと、エラー発生時に専用のエラーハンドラーノードにフォールバックする仕組みを実装します。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph.retry import RetryPolicy
import operator
import random
import os

class ErrorState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    error_count: int

# 意図的に失敗する可能性のあるノード
def unreliable_llm_call(state: ErrorState):
    print(f"[Unreliable LLM] Attempting LLM call (Error count: {state.get('error_count', 0)})")
    if random.random() < 0.6 and state.get("error_count", 0) < 2: # 60%の確率で失敗、2回目まではリトライ
        print("[Unreliable LLM] Simulating transient API error!")
        # エラーカウントをインクリメント
        current_error_count = state.get("error_count", 0) + 1
        return {"error_count": current_error_count, "messages": [AIMessage(content="APIエラーが発生しました。リトライします。")]}
        # raise ValueError("Simulated API error!") # これを有効にするとRetryPolicyが発動
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    response = llm.invoke(state["messages"])
    print(f"[Unreliable LLM] Successfully responded: {response.content[:50]}...")
    return {"messages": [response], "error_count": 0} # 成功したらエラーカウントをリセット

# エラーハンドラーノード
def handle_error(state: ErrorState):
    print(f"[Error Handler] Handling error after {state.get('error_count', 0)} retries.")
    # ここでエラー通知、ログ記録、フォールバック応答などを実装
    error_message = f"申し訳ありません、現在システムエラーが発生しており、ご要望にお応えできません。しばらくしてから再度お試しください。"
    return {"messages": [AIMessage(content=error_message)]}

# グラフの構築
error_workflow = StateGraph(ErrorState)
error_workflow.add_node("unreliable_node", unreliable_llm_call)
error_workflow.add_node("error_handler", handle_error)

# RetryPolicyの設定
# unreliable_nodeにRetryPolicyを設定
error_workflow.add_node(
    "unreliable_node",
    unreliable_llm_call,
    retry_policy=RetryPolicy(
        max_attempts=3, # 最大3回のリトライ（最初の試行＋2回のリトライ）
        initial_interval=1, # 最初の待ち時間1秒
        backoff_factor=2, # 待ち時間を2倍にする (1s, 2s, 4s)
        retry_on=[ValueError] # ValueErrorが発生した場合にリトライ
    )
)

error_workflow.add_edge(START, "unreliable_node")

# 条件付きエッジでエラーハンドリング
def check_for_error_fallback(state: ErrorState) -> Literal["success", "error"]:
    # 意図的に失敗するノードから、エラーカウントが最大リトライ回数を超えた場合にエラーハンドラーへ
    # この例では、ノード内でエラーカウントを更新しているため、直接チェック
    if state.get("error_count", 0) >= 3: # max_attempts=3なので、3回失敗したらエラーハンドラーへ
        print("[Router] Max retries exhausted. Routing to ERROR_HANDLER.")
        return "error"
    print("[Router] Not an error state or retries remaining. Routing to SUCCESS.")
    return "success"

error_workflow.add_conditional_edges(
    "unreliable_node",
    check_for_error_fallback,
    {
        "success": END,
        "error": "error_handler"
    }
)
error_workflow.add_edge("error_handler", END) # エラーハンドラーから終了

error_app = error_workflow.compile()

print("\n--- Error Handling Pattern Execution (May retry or fall back) ---")
# 失敗の確率を上げるため、複数回実行してみる
for i in range(3):
    print(f"\n--- Attempt {i+1} ---")
    result_error = error_app.invoke({"messages": [HumanMessage(content="今日のニュースを教えてください。")]})
    print(f"Final response: {result_error['messages'][-1].content}")
    # 結果はランダムに成功・リトライ・エラーハンドラーにフォールバックします

4. マルチエージェントコラボレーションパターン (Multi-Agent Collaboration Pattern)

複数の専門エージェントがそれぞれの得意分野を活かして連携し、複雑なタスクを解決するパターンです。人間社会の分業に似ています。

なぜ必要か: 単一のエージェントでは対応が難しい、多様な知識やスキルを要する複雑なタスクに対して、各エージェントが専門性を発揮することで、より高品質で効率的な解決策を生み出します。

実装のポイント:

• 各エージェントを独立したLangGraphのサブループまたはノードとして定義します。
• エージェント間のコミュニケーションは、共有状態（State）を通じて行われます。
• コーディネーターエージェント（ルーター）が、タスクの進行状況に応じて適切なエージェントに処理を委譲します。
• LangGraphのadd_subgraph機能を利用して、グラフの中に別のグラフを組み込むことも可能です。

コード例:
ここでは、ユーザーの質問に対して「Web検索担当エージェント」と「回答生成担当エージェント」が連携するシンプルなマルチエージェントシステムを実装します。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
import operator
import os

class MultiAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    task: str # 現在のタスク

# ツール定義 (Web検索エージェント用)
@tool
def search_web_agent_tool(query: str) -> str:
    """Perform a web search and return the results."""
    print(f"[Search Agent Tool] Executing search_web_agent_tool for: '{query}'")
    return f"Search results for '{query}': Example search result data."

# Web検索担当エージェントのノード
def search_agent_node(state: MultiAgentState):
    print("[Search Agent] Activating search agent.")
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, tools=[search_web_agent_tool])
    # 検索すべきクエリをLLMに考えさせるプロンプト
    current_question = state["messages"][-1].content
    search_prompt = f"ユーザーの質問に答えるために、どのようなWeb検索クエリが必要ですか？検索ツールを使ってください。\nユーザー質問: {current_question}"
    response = llm.invoke([HumanMessage(content=search_prompt)])
    print(f"[Search Agent] LLM suggested: {response.tool_calls}")
    return {"messages": [response]} # ツール呼び出しをメッセージとして返す

# 回答生成担当エージェントのノード
def answer_generation_agent_node(state: MultiAgentState):
    print("[Answer Agent] Activating answer generation agent.")
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    # これまでの会話履歴と検索結果（ToolMessage）を使って回答を生成
    response = llm.invoke(state["messages"])
    print(f"[Answer Agent] Generated answer: {response.content[:50]}...")
    return {"messages": [response]}

# ツール実行ノード (共有)
tools_for_multi_agent = [search_web_agent_tool]
multi_agent_tool_node = ToolNode(tools_for_multi_agent)

# メインのルーティングノード
def coordinator_router(state: MultiAgentState) -> Literal["search_agent", "answer_agent", "end"]:
    last_message = state["messages"][-1]
    if isinstance(last_message, HumanMessage):
        print("[Coordinator] Initial human message. Routing to SEARCH_AGENT.")
        return "search_agent" # 最初のユーザー質問は検索エージェントへ
    elif isinstance(last_message, AIMessage) and last_message.tool_calls:
        print("[Coordinator] LLM suggested tool calls. Routing to TOOLS.")
        return "tools" # ツール呼び出しがあればツール実行へ
    elif isinstance(last_message, ToolMessage):
        print("[Coordinator] Tool results received. Routing to ANSWER_AGENT.")
        return "answer_agent" # ツール結果が返ってきたら回答生成エージェントへ
    elif isinstance(last_message, AIMessage) and not last_message.tool_calls:
        print("[Coordinator] Final answer from LLM. Routing to END.")
        return "end" # 最終的な回答が得られたら終了
    print("[Coordinator] Unexpected state. Routing to END.")
    return "end"

# グラフの構築
multi_agent_workflow = StateGraph(MultiAgentState)
multi_agent_workflow.add_node("search_agent", search_agent_node)
multi_agent_workflow.add_node("answer_agent", answer_generation_agent_node)
multi_agent_workflow.add_node("tools", multi_agent_tool_node)

multi_agent_workflow.add_conditional_edges(
    START,
    coordinator_router,
    {
        "search_agent": "search_agent",
        "answer_agent": "answer_agent",
        "tools": "tools",
        "end": END
    }
)

multi_agent_workflow.add_conditional_edges(
    "search_agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "answer_agent", # 検索エージェントがツール呼び出しを提案したらツールへ、そうでなければ回答エージェントへ
    {"tools": "tools", "answer_agent": "answer_agent"} # ここは簡易的なルーター
)

multi_agent_workflow.add_edge("tools", "answer_agent") # ツール実行後は回答エージェントへ

multi_agent_workflow.add_conditional_edges(
    "answer_agent",
    lambda state: "end" if not state["messages"][-1].tool_calls else "tools", # 回答エージェントが最終回答を出したら終了、そうでなければツールを提案したとみなしツールへ
    {"end": END, "tools": "tools"} # ここも簡易的なルーター
)

multi_agent_app = multi_agent_workflow.compile()

print("\n--- Multi-Agent Collaboration Pattern Execution ---")
result_multi_agent = multi_agent_app.invoke({"messages": [HumanMessage(content="日本の現在の首相の名前と、その人物が所属する政党を教えてください。")]})
print(f"Final response: {result_multi_agent['messages'][-1].content}")
# 出力例: Final response: 日本の現在の首相は岸田文雄氏で、自由民主党に所属しています。(Web検索結果に基づいてLLMが生成)

5. Human-in-the-Loop (HITL) パターン

AIエージェントが自信のない出力や重要な決定を下す際に、人間の介入を組み込むパターンです。

なぜ必要か: 全自動のAIシステムでは、誤った情報生成、倫理的な問題、あるいは単に複雑すぎてAIだけでは判断できないケースが存在します。人間の専門知識や承認をプロセスに組み込むことで、システムの信頼性と安全性を向上させます。

実装のポイント:

• 特定の条件（例: LLMのconfidencスコアが低い、特定のキーワードが含まれる）で人間レビューノードにルーティングします。
• 人間のレビューアが結果を確認し、承認または修正指示をシステムにフィードバックするインターフェースが必要です。
• LangGraphのチェックポイント機能により、人間が介入している間も状態を永続化し、介入後に再開できます。

コード例:
ここでは、生成された回答の信頼度が低いと判断された場合に、人間によるレビューを挟むシンプルなHITLパターンを実装します。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
import operator
import os

class HITLState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    confidence: float # LLMの回答の自信度 (例として追加)
    human_feedback: str # 人間からのフィードバック
    reviewed: bool # 人間がレビュー済みかどうか

# 回答生成ノード (自信度も生成する例)
def generate_answer_with_confidence(state: HITLState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    user_question = state["messages"][-1].content
    prompt = f"ユーザーの質問に回答してください。回答の最後に、その回答に対するあなたの自信度を0.0から1.0の数値で示してください。\n質問: {user_question}"
    response_content = llm.invoke([HumanMessage(content=prompt)]).content

    # 自信度をパースする（簡略化のため、実際のLLM出力は整形が必要）
    confidence_score = 0.8 # デフォルト値
    if "自信度:" in response_content:
        try:
            confidence_str = response_content.split("自信度:")[-1].strip().split(" ")[0]
            confidence_score = float(confidence_str)
            response_content = response_content.replace(f"自信度: {confidence_str}", "").strip()
        except ValueError:
            pass

    print(f"[Generator] Generated answer with confidence {confidence_score}: {response_content[:50]}...")
    return {"messages": [AIMessage(content=response_content)], "confidence": confidence_score, "reviewed": False}

# 人間レビューノード
def human_review_node(state: HITLState):
    print("\n--- HUMAN INTERVENTION REQUIRED ---")
    print(f"User Question: {state['messages'][0].content}")
    print(f"Generated Answer: {state['messages'][-1].content}")
    print(f"Confidence: {state['confidence']:.2f}")

    # 本番ではWeb UIなどを通じて人間の入力を受け取る
    feedback = input("この回答を承認しますか？ (y/n) または修正点を入力してください: ")
    
    if feedback.lower() == 'y':
        print("[Human Review] Answer APPROVED.")
        return {"human_feedback": "APPROVED", "reviewed": True}
    else:
        print(f"[Human Review] Answer REJECTED/REVISED with feedback: {feedback}")
        return {"human_feedback": feedback, "reviewed": True}

# 最終回答生成または修正ノード
def finalize_or_revise_answer(state: HITLState):
    if state["human_feedback"] == "APPROVED":
        print("[Finalizer] Human approved. Returning final answer.")
        return state # 承認されたのでそのまま終了
    else:
        print("[Finalizer] Human rejected/revised. Attempting to revise answer.")
        llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
        user_question = state["messages"][0].content
        current_answer = state["messages"][-1].content
        feedback = state["human_feedback"]
        
        revise_prompt = (
            f"ユーザーの質問: {user_question}\n"
            f"現在の回答: {current_answer}\n"
            f"人間からのフィードバック: {feedback}\n"
            f"上記のフィードバックを考慮して、回答を修正してください。"
        )
        revised_response = llm.invoke([HumanMessage(content=revise_prompt)])
        print(f"[Finalizer] Revised answer: {revised_response.content[:50]}...")
        return {"messages": [revised_response], "confidence": 1.0, "reviewed": True} # 修正後は自信度を1.0とする

# 条件付きエッジの決定関数
def should_human_review(state: HITLState) -> Literal["review", "finalize"]:
    if state["confidence"] < 0.6 and not state["reviewed"]: # 自信度が低い、かつ未レビューの場合
        print("[Router] Low confidence and not reviewed. Routing to HUMAN_REVIEW.")
        return "review"
    print("[Router] High confidence or already reviewed. Routing to FINALIZE.")
    return "finalize"

# グラフの構築
hitl_workflow = StateGraph(HITLState)
hitl_workflow.add_node("generate", generate_answer_with_confidence)
hitl_workflow.add_node("human_review", human_review_node)
hitl_workflow.add_node("finalize_or_revise", finalize_or_revise_answer)

hitl_workflow.add_edge(START, "generate")
hitl_workflow.add_conditional_edges(
    "generate",
    should_human_review,
    {
        "review": "human_review",
        "finalize": "finalize_or_revise"
    }
)
hitl_workflow.add_edge("human_review", "finalize_or_revise")
hitl_workflow.add_edge("finalize_or_revise", END)

hitl_app = hitl_workflow.compile()

print("\n--- Human-in-the-Loop Pattern Execution ---")
# 自信度が低いと仮定される質問
result_hitl = hitl_app.invoke({"messages": [HumanMessage(content="将来のAI技術の発展について、まだ誰も知らないような画期的なアイデアを教えてください。")]})
print(f"Final system response: {result_hitl['messages'][-1].content}")
# 人間が「y」と入力すればそのまま、それ以外なら修正が試みられます。

本番運用を見据えたLangGraphのベストプラクティスとハマりどころ

これまで紹介したデザインパターンを効果的に活用し、LangGraphを本番環境で運用するために、いくつかのベストプラクティスと、よくあるハマりどころとその回避策をまとめます。

状態管理のベストプラクティスとハマりどころ

• ハマりどころ: TypedDictで定義された状態オブジェクトが肥大化したり、複数のノードで状態の更新が競合したりすると、デバッグが困難になります。
• 回避策:
  • 状態は最小限に保ち、明示的に型付けします。 一時的な値は状態にダンプせず、関数スコープで渡します。
  • 状態をデータベーススキーマのように扱い、ドキュメント化し、バージョン管理します。 どのキーがどのエージェントに属するかについて明確な規則を確立します。
  • 生データを状態に保存し、プロンプトのフォーマットはノード内でオンデマンドで行います。 これにより、プロンプトテンプレートの変更が状態スキーマに影響を与えなくなります。
  • 累積が必要な場合にのみ、operator.addのようなリデューサーヘルパーを使用します。

無限ループの回避策

• ハマりどころ: 特にリフレクションパターンやマルチエージェントコラボレーションパターンなど、エージェントが自身を修正したり、他のエージェントと対話したりする際に、意図しないループに陥ることがあります。
• 回避策:
  • 最大イテレーション回数を設定して、無限ループを防ぎます。 上記のリフレクションパターンの例のように、iterationsのような状態変数でカウントを管理し、上限に達したら強制的に終了させます。
  • 条件付きエッジのロジックを慎重に設計し、明確な終了条件を設けます。 各ループの出口を明確にし、エージェントが最終的にタスクを完了して終了できるパスを確保します。

エラーハンドリングの強化

• ハマりどころ: APIレート制限、ネットワーク障害、LLMの予期せぬ出力など、本番環境ではさまざまなエラーが発生します。適切なエラーハンドリングがないと、エージェントが停止したり、ユーザーエクスペリエンスが損なわれたりします。
• 回避策:
  • LangGraphのRetryPolicyとTimeoutPolicyを積極的に利用し、一時的なエラーからの回復とハングアップ防止を図ります。
  • 専用のerror_handlerノードを定義し、リトライが尽きた後に実行されるロジック（クリーンアップ、アラート、フォールバックなど）を実装します。
  • 予期せぬエラー（TypeError、KeyError、ロジックバグなど）は、あえてクラッシュさせて、LangSmithなどのトレーシングツールで詳細な情報を取得し、デバッグに役立てます。

トレーシングとデバッグの難しさへの対処

• ハマりどころ: 複雑なグラフ構造やマルチエージェントシステムでは、問題発生時にどのエージェントがどのように動作したかを追跡するのが難しい場合があります。
• 回避策:
  • LangSmithを統合します。 LANGSMITH_TRACING=trueとAPIキーを設定するだけで、エージェントの実行トレース、デバッグ、評価を詳細に行えます。これは本番運用における不可欠なツールです。
  • 各ノードの決定や主要なイベントをログに出力し、ワークフローのデバッグに役立てます。
  • 各関数だけでなく、グラフ全体をテストします。 ユニットテストと統合テストの両方を実施し、エッジケースもカバーします。

その他のベストプラクティス

• ノードの粒度: ワークフローを明確なステップに分割し、各ステップが1つの特定のことを行うノードとして機能するようにします。ノードが細かすぎると複雑になり、粗すぎると回復性が低下します。適切なバランスを見つけましょう。
• メモリと永続性: 本番環境ではPostgresなどのプロダクションチェックポインターを使用します。関連性のない履歴でプロンプトを過剰に詰め込まないように、要約や関連性フィルタリングを実装することも検討します。
• ストリーミングとパフォーマンス: UIの要件に合わせて適切なストリームモード（messages / updates / values / custom）を選択します。LangGraphはストリーミングワークフローを念頭に設計されており、オーバーヘッドは少ないです。
• 型ヒント: TypedDictやPydanticモデルを使用し、状態やノードの入出力に型ヒントを積極的に導入することで、コードの可読性、保守性、堅牢性を大幅に向上させます。

まとめ

この記事では、LangGraphを使って堅牢なAIエージェントを構築するための基本的な考え方から、5つの具体的なデザインパターン（ルーティング、リフレクション、エラーハンドリング、マルチエージェントコラボレーション、Human-in-the-Loop）とそれぞれの実装例を解説しました。

LangGraphは、複雑なAIワークフローをグラフとしてモデル化することで、従来の線形的なチェーンでは難しかった動的な意思決定、ループ、状態管理を可能にします。本番運用において、無限ループ、エラー耐性、デバッグの課題に直面した際には、今回紹介したデザインパターンとベストプラクティスが強力な手助けとなるでしょう。

LangGraphの可能性は非常に広く、これらのパターンを組み合わせることで、さらに高度で適応性の高いAIエージェントを構築できます。ぜひ、ご自身のプロジェクトにこれらの知見を適用し、より堅牢なAIシステム開発に挑戦してみてください。

次の一歩: 公式ドキュメントの「LangGraph Concepts」や「LangGraph Cookbook」で、さらに詳細な情報や発展的なトピックを探求することをお勧めします。