langgraph-agent-patterns

LangGraph エージェントパターン

LangGraph アプリケーション向けに、マルチエージェント連携パターンを実装し、設定します。

パターンの選択

連携のニーズに応じて適切なパターンを選択します。

パターン	最適な用途	使用する状況
Supervisor	複雑なワークフロー、動的なルーティング	エージェントが連携する必要がある場合、ルーティングがコンテキストに依存する場合
Router	単純な分類、独立したタスク	1 回限りのルーティング、決定論的な判断
Orchestrator-Worker	並列実行、高スループット	独立したサブタスク、結果を集約する必要がある場合
Handoffs	順次的なワークフロー、コンテキストの保持	明確な順序があり、各エージェントが前のエージェントの成果を引き継ぐ場合

クイック判断：

• 動的なルーティングが必要ですか？ → Supervisor
• タスクを並列実行できますか？ → Orchestrator-Worker
• 単純な分類ですか？ → Router
• 直線的な順序ですか？ → Handoffs

詳細な比較については、references/pattern-comparison.md を参照してください。

パターン実装ガイド

Supervisor-Subagent パターン

概要： 中央のコーディネーターが、コンテキストに応じて専門のサブエージェントに処理を委任します。

クイックスタート：

# Generate supervisor graph boilerplate
uv run scripts/generate_supervisor_graph.py my-team \
  --subagents "researcher,writer,reviewer"

# TypeScript
uv run scripts/generate_supervisor_graph.py my-team \
  --subagents "researcher,writer,reviewer" \
  --typescript

主な構成要素：

1. ルーティングを含む状態：ルーティングの判断に使用する next フィールド
2. Supervisor ノード：コンテキストに基づいてルーティングを判断します
3. サブエージェントノード：それぞれ異なる能力を持つ専門エージェント
4. 条件付きエッジ：Supervisor からサブエージェントへルーティングします

フローの例：

User Request → Supervisor → Researcher → Supervisor → Writer → Supervisor → FINISH

完全な実装については、 references/supervisor-subagent.md を参照してください。

Router パターン

概要： 初期リクエストに基づいて、専門エージェントへ 1 回だけルーティングします。

主な構成要素：

1. ルートを含む状態：ルーティング判断を格納する単一のフィールド
2. Router ノード：リクエストを分類します（キーワード、LLM、またはセマンティック）
3. 専門エージェント：各カテゴリに対応する独立したエージェント
4. 条件付きルーティング：エージェントにルーティングした後、END に進みます

フローの例：

User Request → Router → Sales Agent → END
                   ├→ Support Agent → END
                   └→ Billing Agent → END

ルーティング戦略：

• キーワードベース：高速でシンプルな文字列マッチング
• LLM ベース：柔軟なセマンティック理解
• 埋め込みベース：類似度マッチング
• モデルベース：ファインチューニング済み分類器

完全な実装については、 references/router-pattern.md を参照してください。

Orchestrator-Worker パターン

概要： タスクを並列のサブタスクに分解し、結果を集約します。

主な構成要素：

1. サブタスクを含む状態：タスクの分解と結果の蓄積
2. Orchestrator ノード：タスクを独立したサブタスクに分割します
3. Worker ノード：サブタスクを並列処理します
4. Aggregator ノード：結果を統合します
5. Send によるファンアウト：条件付きエッジから Send(...) オブジェクトを返し、リストリデューサー（例：Annotated[list[dict], operator.add]）を使用してワーカーの出力を蓄積します

フローの例：

Task → Orchestrator → Worker 1 ┐
                  → Worker 2  ├→ Aggregator → Result
                  → Worker 3 ┘

ベストプラクティス：

• サブタスクが独立していることを確認します
• ワーカーの障害を適切に処理します
• リソースを管理するため、同時に実行するワーカー数を制限します
• 結果の統合には LLM を使用します

完全な実装については、 references/orchestrator-worker.md を参照してください。

Handoffs パターン

概要： コンテキストを保持しながら、エージェント間で順次処理を引き継ぎます。

主な構成要素：

1. コンテキストを含む状態：引き継ぎを通じて共有するコンテキスト
2. エージェントノード：各エージェントが次のエージェントに処理を引き継ぎます
3. 引き継ぎロジック：明示的または条件付きの引き継ぎ
4. コンテキスト管理：情報を保持して引き渡します

フローの例：

Request → Researcher → Writer → Editor → FINISH
         (with context preservation)

引き継ぎ戦略：

• 明示的：エージェントが次のエージェントを指定します
• 条件付き：完了条件に基づきます
• 循環型：修正のため、エージェントが前のエージェントに処理を戻せます

完全な実装については、 references/handoffs.md を参照してください。

例

実行可能なミニプロジェクト（Python + JavaScript）：

• assets/examples/supervisor-example/
• assets/examples/router-example/
• assets/examples/orchestrator-example/
• assets/examples/handoff-example/

マルチエージェントパターンの状態設計

各パターンには、固有の状態スキーマ設計が必要です。

スーパーバイザーパターン：

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: Literal["agent1", "agent2", "FINISH"]
    current_agent: str

ルーターパターン：

class RouterState(TypedDict):
    messages: list[BaseMessage]
    route: Literal["category1", "category2"]

オーケストレーター・ワーカーパターン：

import operator
class OrchestratorState(TypedDict):
    task: str
    subtasks: list[dict]
    results: Annotated[list[dict], operator.add]

ハンドオフ：

class HandoffState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str
    context: dict

状態パターンの詳細： references/state-management-patterns.md を参照してください。

検証と可視化

グラフ構造を検証する

# Validate agent graph for issues
uv run scripts/validate_agent_graph.py path/to/graph.py:graph

# Checks for:
# - Unreachable nodes
# - Cycles without termination
# - Dead ends
# - Invalid routing

グラフを可視化する

# Generate Mermaid diagram
uv run scripts/visualize_graph.py path/to/graph.py:graph --output diagram.md

# View in browser or IDE with Mermaid support

よくあるパターンとアンチパターン

ベストプラクティス

1. エージェントの役割を明確にする

• 重複しない機能を定義する
• 各エージェントの目的を文書化する
• エージェントの重複を避ける

2. ループを防止する

• 状態で反復回数を追跡する
• 最大反復回数を設定する
• ループ検出を実装する

3. コンテキスト管理

• コンテキストが大きくなったら要約する
• 必要な情報だけを渡す
• 可能な場合は構造化されたコンテキストを使用する

4. エラー処理

• ルーティングの判断を検証する
• 無効なルートを適切に処理する
• 安全なフォールバックをデフォルトにする

避けるべきアンチパターン

1. 過剰な監督

# ❌ Bad: Supervisor for simple linear flow
User → Supervisor → Agent1 → Supervisor → Agent2 → Supervisor

# ✅ Good: Use handoffs instead
User → Agent1 → Agent2 → FINISH

2. 複雑なルーターロジック

# ❌ Bad: Complex routing rules in router
if complex_condition_A and (condition_B or condition_C):
    route = determine_complex_route()

# ✅ Good: Use supervisor with LLM
route = llm.invoke("Analyze and route: {query}")

3. 管理されていない状態の増大

# ❌ Bad: Accumulating all messages forever
messages: list[BaseMessage]  # Grows unbounded

# ✅ Good: Summarize or limit
if len(messages) > 20:
    messages = summarize_context(messages)

マルチエージェントシステムのデバッグ

1. エージェントのフローを追跡する

LangSmith を使用してエージェント間のやり取りを可視化します。

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "<your-api-key>"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-debug"

result = graph.invoke(input_state)

2. ルーティングの判断をログに記録する

ルーティングノードにログを追加します。

def supervisor_node(state: SupervisorState) -> dict:
    decision = make_routing_decision(state)

    print(f"Supervisor routing to: {decision}")
    print(f"Current state: {len(state['messages'])} messages")
    print(f"Iteration: {state.get('iteration', 0)}")

    return {"next": decision}

3. グラフ構造を検証する

# Detect common issues
uv run scripts/validate_agent_graph.py my_agent/graph.py:graph

# Check for:
# - Unreachable nodes
# - Infinite loops
# - Dead ends

4. フローを可視化する

# Generate diagram
uv run scripts/visualize_graph.py my_agent/graph.py:graph -o flow.md

パフォーマンスの最適化

レイテンシの最適化

スーパーバイザーパターン：

• ルーティングにはより高速なモデル（gpt-4o-mini）を使用する
• ルーティングの判断をキャッシュする
• 早期終了を実装する

ルーターパターン：

• 単純なケースにはキーワードマッチングを使用する
• 類似したクエリのルーティングをキャッシュする
• 可能な限り LLM の呼び出しを避ける

オーケストレーター・ワーカーパターン：

• 真の並列化がすでに最適です
• レート制限を避けるためにワーカー数を制限する
• 結果をアグリゲーターにストリーミングする

ハンドオフ：

• コンテキストのサイズを最小限に抑える
• 不要なハンドオフを省略する
• 適切な場合は、より低コストのモデルを使用する

コストの最適化

トークン使用量：

• コンテキストを定期的に要約する
• 信頼性を高めるために構造化出力を使用する
• 単純なタスクにはより低コストのモデルを使用する

LLM 呼び出し：

• ルーティングの判断をキャッシュする
• 可能な場合は決定論的なロジックを使用する
• 類似したリクエストをまとめて処理する

パターンの選択：

• ルーター < ハンドオフ < オーケストレーター < スーパーバイザー（コスト）

マルチエージェントパターンのテスト

ルーティングロジックの単体テスト

def test_supervisor_routing():
    """Test supervisor routes correctly."""
    state = {
        "messages": [HumanMessage(content="Need research")],
        "next": "",
        "current_agent": ""
    }

    result = supervisor_node(state)
    assert result["next"] == "researcher"

統合テスト

def test_full_workflow():
    """Test complete multi-agent workflow."""
    graph = create_supervisor_graph()

    result = graph.invoke({
        "messages": [HumanMessage(content="Write article about AI")]
    })

    # Verify agents were called in correct order
    assert "researcher" in result["agent_history"]
    assert "writer" in result["agent_history"]

グラフ構造のテスト

# Validate before deployment
python3 scripts/validate_agent_graph.py graph.py:graph

パターン間の移行

ルーターからスーパーバイザーへ

ルーティングロジックが複雑になった場合：

# Before: Complex router
def route(query):
    if complex_rules(query):
        return category

# After: Supervisor with LLM
def supervisor(state):
    return llm_routing_decision(state)

ハンドオフからスーパーバイザーへ

動的なルーティングが必要になった場合：

# Before: Fixed sequence
Agent1 → Agent2 → Agent3

# After: Dynamic routing
Supervisor ⇄ Agent1/Agent2/Agent3

逐次処理から並列処理へ

タスクが独立して実行できるようになった場合：

# Before: Sequential
Agent1 → Agent2 → Agent3

# After: Parallel
Orchestrator → [Agent1, Agent2, Agent3] → Aggregator

よくあるユースケース

カスタマーサポートシステム

パターン： ルーター + スーパーバイザー

Router → Sales Supervisor → Sales Agents
     ↓
     Support Supervisor → Support Agents

リサーチと執筆のパイプライン

パターン： スーパーバイザーまたはハンドオフ

Supervisor ⇄ Researcher
         ⇄ Writer
         ⇄ Editor

データ分析パイプライン

パターン： オーケストレーター・ワーカー

Orchestrator → Data Collectors → Aggregator

ドキュメント処理

パターン： オーケストレーター・ワーカー + スーパーバイザー

Router → PDF Orchestrator → Workers → Aggregator
     ↓
     DOCX Orchestrator → Workers → Aggregator

スクリプトのリファレンス

generate_supervisor_graph.py

スーパーバイザーとサブエージェントのボイラープレートを生成します。

uv run scripts/generate_supervisor_graph.py <name> [options]

Options:
  --subagents AGENTS    Comma-separated list (default: researcher,writer,reviewer)
  --output DIR          Output directory (default: current directory)
  --typescript          Generate TypeScript instead of Python

validate_agent_graph.py

グラフ構造を検証します。

uv run scripts/validate_agent_graph.py <module_path>

Format: path/to/module.py:graph_name

Checks:
  - Unreachable nodes
  - Cycles
  - Dead ends
  - Invalid routing

visualize_graph.py

Mermaid 図を生成します。

uv run scripts/visualize_graph.py <module_path> [options]

Options:
  --output FILE         Output file (default: stdout)
  --diagram-only        Skip documentation, output diagram only

トラブルシューティング

「エージェントが正しく連携しない」

確認事項：

1. 状態スキーマが使用するパターンをサポートしている（state-management-patterns.md を参照）
2. ルーティングロジックが正しく検証される
3. エージェント間でコンテキストが保持される

「無限ループが検出される」

解決策：

1. 状態に反復カウンターを追加する
2. 最大反復回数の制限を実装する
3. ループ検出ロジックを追加する
4. validate_agent_graph.py で検証する

「ルーティングの判断が適切でない」

解決策：

1. エージェントの説明を明確にし、スーパーバイザープロンプトを改善する
2. 信頼性を高めるために構造化出力を使用する
3. ルーティングプロンプトに例を追加する
4. ルーティングの判断には、より優れたモデルを使用する

「高レイテンシ」

解決策：

1. 単純なケースではルーターパターンを検討します
2. ルーティングにはより高速なモデルを使用します
3. 可能な場合は並列実行を実装します
4. ルーティングの判断をキャッシュします

「トークン使用量が多い」

解決策：

1. コンテキストを定期的に要約します
2. 単純なタスクにはより低コストのモデルを使用します
3. コンテキストウィンドウイングを実装します
4. より効率的なパターンを選択します

追加リソース

• パターンの詳細：

  • スーパーバイザー：references/supervisor-subagent.md
  • ルーター：references/router-pattern.md
  • オーケストレーター・ワーカー：references/orchestrator-worker.md
  • ハンドオフ：references/handoffs.md

• 状態管理： references/state-management-patterns.md

• パターンの比較： references/pattern-comparison.md

• 実際の例： assets/examples/

• LangGraph ドキュメント：

  • マルチエージェントパターン： https://docs.langchain.com/oss/python/langchain/multi-agent/subagents-personal-assistant; https://docs.langchain.com/oss/python/langchain/multi-agent/handoffs-customer-support; https://docs.langchain.com/oss/python/langchain/multi-agent/router-knowledge-base
  • 条件付きエッジ： https://docs.langchain.com/oss/python/langgraph/graph-api
  • Map-Reduce： https://docs.langchain.com/oss/python/langgraph/graph-api#map-reduce-and-the-send-api