langgraph-agents

Supervisor、Swarm、Handoff、Router、Master の各パターンを使用して、LangGraph で本番グレードのマルチエージェントシステムを構築します。マルチプロバイダールーティング（Claude、DeepSeek、Gemini — OpenAI は使用しません）、ガードレール、耐久性のある実行、可観測性、スケーラブルなエージェント連携を備えた、コスト最適化されたオーケストレーションを可能にします。

<quick_start> 状態スキーマ（基盤）:

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list, add_messages]  # Auto-merge
    next_agent: str  # For handoffs

パターンの選択:

パターン	使用する場面	エージェント数
Supervisor	明確な階層構造	3〜10
Swarm	ピア間の協調	5〜15
Handoff	逐次パイプライン	2〜5
Router	分類してディスパッチ	2〜10
Master	学習システム	10〜30+

API の選択: Graph API（明示的なノード／エッジ）か Functional API（@entrypoint／@task デコレーター）か

主要パッケージ: pip install langchain langgraph langgraph-supervisor langgraph-swarm langchain-mcp-adapters </quick_start>

<success_criteria> マルチエージェントシステムは、次の条件を満たす場合に成功です:

• 状態がメッセージを適切にマージするために Annotated[..., add_messages] を使用している
• 終了条件が無限ループを防いでいる
• ルーティングが条件付きエッジ（ハードコードされたパスではない）または Functional API のタスクを使用している
• コスト最適化: 単純なタスク → より安価なモデル（DeepSeek）
• 複雑な推論 → 高品質なモデル（Claude）
• どこにも OpenAI を使用していない
• コンテキスト保持のためにチェックポインターが有効になっている
• ヒューマン・イン・ザ・ループ: 承認ワークフローに interrupt() を使用している
• ガードレール: PII 検出、予算制限、呼び出し制限
• 適切な場合に MCP ツールが MultiServerMCPClient を介して標準化されている
• 可観測性: 本番環境で LangSmith トレーシングが有効になっている </success_criteria>

<core_content> LangGraph と LangChain を使用して、スケーラブルでコスト最適化されたマルチエージェントシステムを構築するための、本番環境で実証済みのパターンです。

このスキルを使用する場面

症状:

• 「エージェント間で状態が正しく更新されない」
• 「エージェントが適切に連携しない」
• 「LLM コストが制御不能に膨らんでいる」
• 「supervisor、swarm、handoff のどのパターンを選ぶべきかわからない」
• 「エージェントの状態スキーマをどのように構成すべきか不明確」
• 「エージェントがコンテキストを失ったり、作業を繰り返したりする」
• 「PII、予算、安全性のためのガードレールが必要」
• 「エージェントグラフをどのようにテストするか」
• 「クラッシュリカバリを備えた耐久性のある実行が必要」
• 「LangSmith トレーシング／可観測性のセットアップ」
• 「LangGraph を本番環境にデプロイする」

ユースケース:

• 3 つ以上の特化したエージェントによるマルチエージェントシステム
• オーケストレーションを必要とする複雑なワークフロー
• コストに敏感な本番デプロイ
• 自己学習型または適応型のエージェントシステム
• 複数の LLM プロバイダーを使用するエンタープライズアプリケーション

クイックリファレンス: オーケストレーションパターンの選択

パターン	使用する場面	複雑さ	リファレンス
Supervisor	明確な階層構造、集中型ルーティング	低〜中	reference/orchestration-patterns.md
Swarm	ピア間の協調、動的なハンドオフ	中	reference/orchestration-patterns.md
Handoff	逐次パイプライン、エスカレーション	低	reference/orchestration-patterns.md
Router	分類してディスパッチ、ファンアウト	低	reference/orchestration-patterns.md
Skills	段階的な開示、オンデマンド	低	reference/orchestration-patterns.md
Master	学習システム、複雑なワークフロー	高	reference/orchestration-patterns.md

コアパターン

1. 状態スキーマ（基盤）

from typing import TypedDict, Annotated, Dict, Any
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], add_messages]  # Auto-merge
    agent_type: str
    metadata: Dict[str, Any]
    next_agent: str  # For handoffs

詳細解説: reference/state-schemas.md（リデューサー、アノテーション、多階層の状態）

2. マルチプロバイダー構成（lang-core 経由）

# Use lang-core for unified provider access (NO OPENAI)
from lang_core.providers import get_llm_for_task, LLMPriority

llm_cheap = get_llm_for_task(priority=LLMPriority.COST)     # DeepSeek
llm_smart = get_llm_for_task(priority=LLMPriority.QUALITY)  # Claude
llm_fast = get_llm_for_task(priority=LLMPriority.SPEED)     # Cerebras
llm_local = get_llm_for_task(priority=LLMPriority.LOCAL)    # Ollama

詳細解説: reference/base-agent-architecture.md、reference/cost-optimization.md

3. Supervisor パターン

from langgraph_supervisor import create_supervisor  # pip install langgraph-supervisor
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(model, tools=research_tools, prompt="Research specialist")
writer_agent = create_react_agent(model, tools=writer_tools, prompt="Content writer")

supervisor = create_supervisor(agents=[research_agent, writer_agent], model=model)
result = supervisor.invoke({"messages": [("user", "Write article about LangGraph")]})

4. Swarm パターン

from langgraph_swarm import create_swarm, create_handoff_tool  # pip install langgraph-swarm

handoff_to_bob = create_handoff_tool(agent_name="Bob", description="Transfer for Python tasks")
alice = create_react_agent(model, tools=[query_db, handoff_to_bob], prompt="SQL expert")
bob = create_react_agent(model, tools=[execute_code], prompt="Python expert")

swarm = create_swarm(agents=[alice, bob], default_active_agent="Alice")

5. Functional API（Graph の代替）

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def research(query: str) -> str:
    return f"Results for: {query}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(query: str) -> dict:
    result = research(query).result()
    return {"output": result}

詳細解説: reference/functional-api.md（耐久性のある実行、タイムトラベル、テスト）

6. MCP ツールの統合

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
    {"tools": {"transport": "stdio", "command": "python", "args": ["./mcp_server.py"]}}
) as client:
    tools = await client.get_tools()
    agent = create_react_agent(model, tools=tools)

詳細解説: reference/mcp-integration.md

7. Deep Agents フレームワーク（本番環境）

from deep_agents import create_deep_agent
from deep_agents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend({
    "/workspace/": StateBackend(),      # Ephemeral
    "/memories/": StoreBackend()        # Persistent
})
agent = create_deep_agent(
    model=ChatAnthropic(model="claude-opus-4-6"),
    backend=backend,
    interrupt_on=["deploy", "delete"],
    skills_dirs=["./skills/"]
)

詳細解説: reference/deep-agents.md（サブエージェント、スキル、長期メモリ）

8. ガードレール

# Recursion limit prevents runaway agents (default: 25 steps)
config = {"recursion_limit": 25, "configurable": {"thread_id": "user-123"}}
result = graph.invoke(input_data, config=config)

# Add guardrail nodes for PII, safety checks, HITL — see reference

詳細解説: reference/guardrails.md (入力/出力の検証、トリップワイヤー、グラフノードのガードレール)

参照ファイル (14 の詳細解説)

アーキテクチャ:

• reference/state-schemas.md - TypedDict、Annotated レデューサー、マルチレベルの状態
• reference/base-agent-architecture.md - マルチプロバイダーのセットアップ、エージェントテンプレート
• reference/tools-organization.md - モジュール式のツール設計、InjectedState/InjectedStore

オーケストレーション:

• reference/orchestration-patterns.md - スーパーバイザー、スウォーム、ハンドオフ、ルーター、スキル、マスター、HITL
• reference/context-engineering.md - 3 種類のコンテキスト、メモリの圧縮、Anthropic のベストプラクティス
• reference/cost-optimization.md - プロバイダーのルーティング、キャッシュ、トークン予算、フォールバックチェーン

API:

• reference/functional-api.md - @entrypoint/@task、耐久性のある実行、タイムトラベル、テスト
• reference/mcp-integration.md - MultiServerMCPClient、非同期コンテキストマネージャー、ツールのコンポジション
• reference/deep-agents.md - ハーネス、バックエンド、サブエージェント、スキル、長期メモリ
• reference/streaming-patterns.md - 5 つのストリーミングモード、v2 フォーマット、カスタムストリーミング

本番環境:

• reference/guardrails.md - PII 検出、プロンプトインジェクション、予算トリップワイヤー、出力フィルタリング
• reference/testing-patterns.md - 単体/統合テスト、モッキング、スナップショットテスト、CI/CD
• reference/observability.md - LangSmith トレーシング、カスタムメトリクス、評価、モニタリング
• reference/deployment-patterns.md - アプリ構造、ローカルサーバー、LangGraph Platform、Docker

よくある落とし穴

問題	解決策
状態が更新されない	Annotated[..., add_messages] レデューサーを追加する
無限ループ	終了条件を追加するか、config で recursion_limit を設定する
コストが高い	単純なタスクはより安価なモデルにルーティングする。フォールバックチェーンを使用する
コンテキストの喪失	チェックポインターまたはメモリシステムを使用する
誤ったインポート	create_supervisor は langgraph_supervisor からインポートする（langgraph.prebuilt ではない）
誤ったインポート	create_swarm は langgraph_swarm からインポートする（langgraph.prebuilt ではない）
MCP API の不一致	get_langchain_tools() ではなく await client.get_tools() を使用する
PII の漏洩	PII 編集ガードノードを追加する（reference/guardrails.md を参照）
可観測性がない	ゼロコンフィグのトレーシングには LANGSMITH_TRACING=true を設定する
脆弱なエージェント	ガードレールを追加する: 呼び出し制限、予算トリップワイヤー、構造化出力

lang-core 統合

本番環境へのデプロイでは、以下に lang-core を使用します:

• ミドルウェア: コスト追跡、予算の適用、リトライ、キャッシュ、PII の安全性
• LangSmith: @traced_agent デコレーターによる統一トレーシング
• プロバイダー: get_llm_for_task(priority=...) による自動選択
• Celery: 進捗追跡を伴うバックグラウンドでのエージェント実行
• Redis: 分散ロック、レート制限、イベントの pub/sub

from lang_core import traced_agent, get_llm_for_task, LLMPriority
from lang_core.middleware import budget_enforcement_middleware

@traced_agent("QualificationAgent", tags=["sales"])
async def run_qualification(data):
    llm = get_llm_for_task(priority=LLMPriority.SPEED)
    # ... agent logic

成果サイドカーの出力

最終ステップとして、~/.claude/skill-analytics/last-outcome-langgraph-agents.json に書き込みます:

{"ts":"[UTC ISO8601]","skill":"langgraph-agents","version":"2.0.0","variant":"default",
 "status":"[success|partial|error]","runtime_ms":[estimated ms from start],
 "metrics":{"agents_created":[n],"nodes_configured":[n],"graphs_built":[n]},
 "error":null,"session_id":"[YYYY-MM-DD]"}

一部のステージが失敗しても結果が生成された場合は、ステータスに "partial" を使用します。出力がまったく生成されなかった場合にのみ "error" を使用します。