LangGraphマルチエージェントオーケストレーション実戦：Pythonで本番級AIワークフローを構築する6つのコアパターン
AI与大数据
LangGraphマルチエージェントオーケストレーション：なぜ単一エージェントでは不十分なのか
単一のAIエージェントは簡単なタスクしか処理できません。マルチステップの意思決定、条件分岐、人間の承認、状態のロールバックが必要になると、コードはスパゲッティになります。LangGraphは有向グラフ（DAG）モデルに基づき、エージェント間のコラボレーションフローを宣言的に定義できます。各ノードはエージェントまたは関数、各エッジは状態転送パスです。2026年、LangGraphはStateGraph→条件ルーティング→Human-in-the-Loop→チェックポイント永続化→サブグラフネスト→ストリーミング出力のフルパイプラインをサポートしています。
本記事では6つのコアパターンから、グラフ定義→条件ルーティング→人間の承認→状態永続化→サブグラフ構成→ストリーミング出力のフルパイプライン実戦を解説します。
コア概念
| 概念 | 説明 |
|---|---|
| StateGraph | 状態ベースの有向グラフ。ノードが状態を処理、エッジが状態を転送 |
| Node | グラフ内の処理単位。状態を受け取り、状態の更新を返す |
| Edge | ノード間の接続。通常エッジと条件エッジに分かれる |
| Conditional Edge | 状態に基づいて次のノードを動的に選択するエッジ |
| Checkpoint | 状態スナップショット。一時停止/再開/ロールバックをサポート |
| Human-in-the-Loop | 人間の承認ノード。人間の入力を待つため実行を一時停止 |
| Subgraph | ネストされたサブグラフ。モジュラー構成をサポート |
| Stream Mode | ストリーミング出力モード。中間結果をリアルタイムで返す |
問題分析：マルチエージェントオーケストレーションが解決する5つの課題
- フローの制御不能：単一エージェントのReActループでは複雑なビジネスプロセスを表現できない
- 状態の喪失：長い会話でエージェントが以前の重要な決定を記憶できない
- 人間の承認の欠如：重要な操作を人間の確認のために一時停止できない
- エラーからの回復不能：実行失敗後に前の安定した状態にロールバックできない
- 協力の混乱：複数エージェント間の責任境界が曖昧、呼び出し順序が制御不能
ステップバイステップ：6つのLangGraphマルチエージェントオーケストレーションコアパターン
パターン1：StateGraphの基礎とグラフ定義
pip install langgraph==0.4.0 langchain-openai==0.3.0
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
user_intent: str
search_results: list[str]
final_answer: str
def classify_intent(state: AgentState) -> dict:
last_message = state["messages"][-1].content
if any(kw in last_message for kw in ["検索", "探す", "search"]):
return {"user_intent": "search"}
elif any(kw in last_message for kw in ["計算", "分析", "calculate"]):
return {"user_intent": "calculate"}
else:
return {"user_intent": "chat"}
def search_agent(state: AgentState) -> dict:
query = state["messages"][-1].content
results = [f"検索結果1: {query}の関連情報", f"検索結果2: {query}の詳細分析"]
return {"search_results": results}
def calculate_agent(state: AgentState) -> dict:
return {"final_answer": "計算結果: 42"}
def chat_agent(state: AgentState) -> dict:
return {"final_answer": "こんにちは！AIアシスタントです。何かお手伝いできますか？"}
def route_intent(state: AgentState) -> str:
return state["user_intent"]
graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("search", search_agent)
graph.add_node("calculate", calculate_agent)
graph.add_node("chat", chat_agent)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
"search": "search",
"calculate": "calculate",
"chat": "chat",
})
graph.add_edge("search", END)
graph.add_edge("calculate", END)
graph.add_edge("chat", END)
app = graph.compile()
result = app.invoke({
"messages": [{"role": "user", "content": "Pythonの最新機能を検索して"}],
"user_intent": "",
"search_results": [],
"final_answer": "",
})
print(result["search_results"])
パターン2：条件ルーティングとマルチターン会話
from typing import Literal
class ResearchState(TypedDict):
messages: Annotated[list, add_messages]
topic: str
outline: str
draft: str
review_feedback: str
revision_count: int
def researcher(state: ResearchState) -> dict:
topic = state.get("topic", state["messages"][-1].content)
outline = f"{topic}の研究アウトライン:\n1. 背景紹介\n2. コア概念\n3. 実践事例\n4. まとめと展望"
return {"topic": topic, "outline": outline}
def writer(state: ResearchState) -> dict:
draft = f"アウトライン『{state['outline']}』に基づく初稿..."
return {"draft": draft}
def reviewer(state: ResearchState) -> dict:
if state.get("revision_count", 0) >= 2:
return {"review_feedback": "approved"}
return {"review_feedback": "修正必要：より多くの事例とデータの裏付けを追加"}
def should_revise(state: ResearchState) -> Literal["revise", "publish"]:
if state.get("review_feedback") == "approved":
return "publish"
return "revise"
def reviser(state: ResearchState) -> dict:
revised = f"改訂版（修正#{state.get('revision_count', 0) + 1}）: {state['draft']}\n事例とデータを追加。"
return {"draft": revised, "revision_count": state.get("revision_count", 0) + 1}
def publisher(state: ResearchState) -> dict:
return {"final_answer": f"記事公開: {state['draft']}"}
research_graph = StateGraph(ResearchState)
research_graph.add_node("research", researcher)
research_graph.add_node("write", writer)
research_graph.add_node("review", reviewer)
research_graph.add_node("revise", reviser)
research_graph.add_node("publish", publisher)
research_graph.add_edge(START, "research")
research_graph.add_edge("research", "write")
research_graph.add_edge("write", "review")
research_graph.add_conditional_edges("review", should_revise, {
"revise": "revise",
"publish": "publish",
})
research_graph.add_edge("revise", "review")
research_graph.add_edge("publish", END)
research_app = research_graph.compile()
パターン3：Human-in-the-Loop
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
class ApprovalState(TypedDict):
messages: Annotated[list, add_messages]
request: str
risk_level: str
approved: bool
result: str
def risk_assessor(state: ApprovalState) -> dict:
request = state["request"]
if any(kw in request for kw in ["削除", "リセット", "クリア", "delete"]):
return {"risk_level": "high"}
elif any(kw in request for kw in ["変更", "更新", "update"]):
return {"risk_level": "medium"}
return {"risk_level": "low"}
def human_approval(state: ApprovalState) -> dict:
if state["risk_level"] == "high":
decision = interrupt(f"高リスク操作には承認が必要: {state['request']}")
return {"approved": decision.get("approved", False)}
return {"approved": True}
def executor(state: ApprovalState) -> dict:
if state["approved"]:
return {"result": f"実行成功: {state['request']}"}
return {"result": f"操作が拒否されました: {state['request']}"}
approval_graph = StateGraph(ApprovalState)
approval_graph.add_node("assess", risk_assessor)
approval_graph.add_node("approve", human_approval)
approval_graph.add_node("execute", executor)
approval_graph.add_edge(START, "assess")
approval_graph.add_edge("assess", "approve")
approval_graph.add_edge("approve", "execute")
approval_graph.add_edge("execute", END)
checkpointer = MemorySaver()
approval_app = approval_graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "approval-001"}}
result = approval_app.invoke(
{"request": "本番データベースの全テストデータを削除", "messages": [], "risk_level": "", "approved": False, "result": ""},
config=config,
)
for state in approval_app.get_state_history(config):
if state.next:
approval_app.invoke(
Command(resume={"approved": True}),
config=config,
)
break
パターン4：チェックポイント永続化と状態復元
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
class LongTaskState(TypedDict):
messages: Annotated[list, add_messages]
task_id: str
steps_completed: list[str]
current_step: str
error: str
def step_one(state: LongTaskState) -> dict:
return {"current_step": "step_one", "steps_completed": state.get("steps_completed", []) + ["step_one"]}
def step_two(state: LongTaskState) -> dict:
return {"current_step": "step_two", "steps_completed": state.get("steps_completed", []) + ["step_two"]}
def step_three(state: LongTaskState) -> dict:
return {"current_step": "step_three", "steps_completed": state.get("steps_completed", []) + ["step_three"]}
long_task_graph = StateGraph(LongTaskState)
long_task_graph.add_node("one", step_one)
long_task_graph.add_node("two", step_two)
long_task_graph.add_node("three", step_three)
long_task_graph.add_edge(START, "one")
long_task_graph.add_edge("one", "two")
long_task_graph.add_edge("two", "three")
long_task_graph.add_edge("three", END)
conn = sqlite3.connect(":memory:", check_same_thread=False)
sqlite_checkpointer = SqliteSaver(conn)
long_task_app = long_task_graph.compile(checkpointer=sqlite_checkpointer)
task_config = {"configurable": {"thread_id": "long-task-001"}}
long_task_app.invoke({"task_id": "task-1", "messages": [], "steps_completed": [], "current_step": "", "error": ""}, config=task_config)
state_snapshot = long_task_app.get_state(task_config)
print(f"完了ステップ: {state_snapshot.values.get('steps_completed', [])}")
for history_state in long_task_app.get_state_history(task_config):
print(f"ステップ: {history_state.values.get('current_step', 'N/A')}, 時間: {history_state.created_at}")
パターン5：サブグラフネストとモジュラー構成
class CodeReviewState(TypedDict):
messages: Annotated[list, add_messages]
code: str
lint_result: str
security_result: str
style_result: str
final_review: str
def linter(state: CodeReviewState) -> dict:
return {"lint_result": f"Lintチェック通過: {state['code'][:50]}..."}
def security_scanner(state: CodeReviewState) -> dict:
return {"security_result": "セキュリティスキャン: 脆弱性なし"}
def style_checker(state: CodeReviewState) -> dict:
return {"style_result": "コードスタイル: PEP8準拠"}
review_subgraph = StateGraph(CodeReviewState)
review_subgraph.add_node("lint", linter)
review_subgraph.add_node("security", security_scanner)
review_subgraph.add_node("style", style_checker)
review_subgraph.add_edge(START, "lint")
review_subgraph.add_edge(START, "security")
review_subgraph.add_edge(START, "style")
review_subgraph.add_edge("lint", END)
review_subgraph.add_edge("security", END)
review_subgraph.add_edge("style", END)
review_sub_app = review_subgraph.compile()
class DevPipelineState(TypedDict):
messages: Annotated[list, add_messages]
code: str
review_result: str
test_result: str
deploy_result: str
def code_review_node(state: DevPipelineState) -> dict:
review_output = review_sub_app.invoke({
"code": state["code"],
"messages": [],
"lint_result": "",
"security_result": "",
"style_result": "",
"final_review": "",
})
return {"review_result": f"Lint: {review_output['lint_result']} | Security: {review_output['security_result']} | Style: {review_output['style_result']}"}
def test_runner(state: DevPipelineState) -> dict:
return {"test_result": "全テスト通過 (42/42)"}
def deployer(state: DevPipelineState) -> dict:
return {"deploy_result": "デプロイ成功: v1.0.0 がライブ"}
pipeline_graph = StateGraph(DevPipelineState)
pipeline_graph.add_node("review", code_review_node)
pipeline_graph.add_node("test", test_runner)
pipeline_graph.add_node("deploy", deployer)
pipeline_graph.add_edge(START, "review")
pipeline_graph.add_edge("review", "test")
pipeline_graph.add_edge("test", "deploy")
pipeline_graph.add_edge("deploy", END)
pipeline_app = pipeline_graph.compile()
パターン6：ストリーミング出力とリアルタイムフィードバック
class StreamChatState(TypedDict):
messages: Annotated[list, add_messages]
query: str
thinking: str
response: str
def thinker(state: StreamChatState) -> dict:
return {"thinking": f"分析中: {state['query']}"}
def responder(state: StreamChatState) -> dict:
return {"response": f"'{state['query']}'に関する詳細な回答: LangGraphはストリーミング出力をサポートし、中間結果をリアルタイムで返します..."}
stream_graph = StateGraph(StreamChatState)
stream_graph.add_node("think", thinker)
stream_graph.add_node("respond", responder)
stream_graph.add_edge(START, "think")
stream_graph.add_edge("think", "respond")
stream_graph.add_edge("respond", END)
stream_app = stream_graph.compile()
for event in stream_app.stream({"query": "LangGraphの使い方は？", "messages": [], "thinking": "", "response": ""}):
for node_name, node_output in event.items():
print(f"[{node_name}] {node_output}")
よくある落とし穴
落とし穴1：状態スキーマ定義が不完全
# ❌ 間違い：フィールドのデフォルト値が不足
class BadState(TypedDict):
messages: Annotated[list, add_messages]
result: str # invoke時に必須、そうでないとKeyError
# ✅ 正しい：Optionalを使用するかデフォルト値を提供
from typing import Optional
class GoodState(TypedDict):
messages: Annotated[list, add_messages]
result: str
metadata: Optional[dict] # オプションフィールド
落とし穴2：条件ルーティングの戻り値が不一致
# ❌ 間違い：ルーターがマッピングテーブルにない値を返す
def bad_router(state) -> str:
return "unknown_node" # マッピングテーブルにないキー
# ✅ 正しい：すべての可能な戻り値がマッピングテーブルにあることを確認
def good_router(state) -> Literal["search", "calculate", "chat"]:
if "検索" in state["messages"][-1].content:
return "search"
elif "計算" in state["messages"][-1].content:
return "calculate"
return "chat"
落とし穴3：Checkpointerの忘れによる状態喪失
# ❌ 間違い：checkpointerなし、中断後に復元不可
app = graph.compile() # ステートレス
# ✅ 正しい：checkpointerで状態を永続化
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())
落とし穴4：interruptの不適切な使用
# ❌ 間違い：checkpointerなしのグラフでinterruptを使用
app = graph.compile() # checkpointerなし
# interrupt()が例外をスロー
# ✅ 正しい：checkpointerと組み合わせて使用
app = graph.compile(checkpointer=MemorySaver())
落とし穴5：サブグラフの状態が親グラフと不一致
# ❌ 間違い：サブグラフの状態フィールドが親と完全に異なる
class ParentState(TypedDict):
code: str
class ChildState(TypedDict):
data: bytes # 非互換フィールド
# ✅ 正しい：サブグラフの状態は親のサブセットまたは互換拡張
class ChildState(TypedDict):
code: str # 親フィールドに対応
extra_field: str
エラートラブルシューティング
| # | エラーメッセージ | 原因 | 解決方法 |
|---|---|---|---|
| 1 | KeyError: 'field_name' |
状態スキーマにフィールドが不足 | invoke時にすべての必須フィールドを提供 |
| 2 | InvalidEdgeError |
エッジが存在しないノードを指している | add_edgeのノード名がadd_node済みか確認 |
| 3 | GraphRecursionError |
グラフに無限ループが存在 | 最大反復回数または終了条件を追加 |
| 4 | interrupt() called without checkpointer |
checkpointerが未設定 | コンパイル時にcheckpointerパラメータを渡す |
| 5 | StateUpdateError: incompatible types |
状態更新の型が不一致 | ノードの戻り値dictのキーと値の型を確認 |
| 6 | NodeNotFoundError |
条件ルーティングが未登録ノードを返す | ルーティングの戻り値がマッピングにあることを確認 |
| 7 | CheckpointError: thread_id required |
thread_idが未提供 | invoke時にconfigurable.thread_idを渡す |
| 8 | SubgraphStateError |
サブグラフと親の状態が非互換 | 状態スキーマを統一または変換レイヤーを使用 |
| 9 | StreamTimeoutError |
ストリーミング出力がタイムアウト | 無限ループや長時間ブロッキングのノードを確認 |
| 10 | SerializationError |
状態にシリアライズ不可能なオブジェクトが含まれる | 状態にJSONシリアライズ可能な型のみ使用 |
高度な最適化
- カスタムReducer：
Annotated[type, custom_reducer]で複雑な状態マージロジックを実装 - 並列ノード：
add_edge(START, ["node_a", "node_b"])でノードの並列実行を実現 - グラフ可視化：
app.get_graph().draw_mermaid()でMermaidフローチャートを生成 - 本番級Checkpointer：MemorySaverの代わりにPostgreSQLやRedisを使用
- 非同期実行：
AsyncStateGraphとainvokeで非同期エージェントを実装
比較分析
| 次元 | LangGraph | CrewAI | AutoGen | LangChain Agent |
|---|---|---|---|---|
| グラフオーケストレーション | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐ |
| 状態永続化 | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐ | ⭐⭐ |
| Human-in-the-Loop | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐ |
| 条件ルーティング | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐ | ⭐ |
| サブグラフネスト | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐ | ⭐ |
| 学習曲線 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
まとめ：LangGraphは「命令型エージェント呼び出し」から「宣言型グラフオーケストレーション」へと進化させます。StateGraph→条件ルーティング→Human-in-the-Loop→チェックポイント永続化→サブグラフネスト→ストリーミング出力の六位一体、LangGraphは2026年のPythonマルチエージェントオーケストレーションの最適な選択です。コア原則：状態はデータフロー、ノードは処理単位、エッジは制御フロー。
オンラインツールおすすめ
- JSONフォーマッター：/ja/json/format
- Hash計算：/ja/encode/hash
- cURL→コード変換：/ja/dev/curl-to-code
ブラウザローカルツールを無料で試す →
#LangGraph#AI Agent#多Agent编排#Python AI#2026#AI与大数据