LangGraphで本番環境向けマルチエージェント(Multi-Agent)システムを構築する: 状態管理と循環グラフワークフロー

複雑なビジネスロジックを単一のLLMプロンプトや単純なリニアチェーンで処理するには限界があります。本番環境では、役割ごとに特化した複数のAIエージェントが協調して動作する マルチエージェント(Multi-Agent)システム が求められます。

LangGraph は、グラフベースの状態機械(State Machine)構造を採用し、ループ処理、フィードバック再試行、状態分岐、長期的状態管理を可能にします。

要約

StateGraphによる状態の一元化: 共通の TypedDict 状態を全ノードで共有。

循環グラフ(Cyclic Loops): add_conditional_edges により、エラー時の再試行ループをネイティブにサポート。

Checkpointer: 障害復旧や Human-in-the-loop (人間の承認による一示停止/再開) を実現。

Python実装例 (Supervisor パターン)

from typing import Annotated, TypedDict, Literal
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_step: str
    final_report: str

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

def researcher_node(state: AgentState):
    messages = state["messages"]
    prompt = "リサーチ担当エージェントです。情報を収集してください。"
    response = llm.invoke([HumanMessage(content=prompt)] + messages)
    return {"messages": [response], "next_step": "writer"}

def writer_node(state: AgentState):
    messages = state["messages"]
    prompt = "ライターエージェントです。レポートを作成してください。"
    response = llm.invoke([HumanMessage(content=prompt)] + messages)
    return {"messages": [response], "final_report": response.content, "next_step": "end"}

def supervisor_router(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
    next_step = state.get("next_step", "researcher")
    if next_step == "writer": return "writer"
    elif next_step == "end": return END
    return "researcher"

workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)

workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", supervisor_router)
workflow.add_conditional_edges("writer", supervisor_router)

app = workflow.compile(checkpointer=MemorySaver())

まとめ

LangGraph は明確な状態管理と循環グラフ構造により、マルチエージェントを本番環境で安全かつ堅牢に運用するためのフレームワークです。