LangGraph マルチエージェント協調実践：Supervisor パターンとタスク分散

先月、チームの Agent システムをリファクタリングしました。もともと 1 つの Agent に 12 個ものツールがぶら下がっていた——検索、コード実行、ドキュメント生成、メール送信……。結果はどうだったか。LLM がツール選びで迷子になり、検索すべき場面でコード実行ツールを使ってしまう。デバッグ時のログを見ても、どのツール呼び出しが問題なのかさっぱり分からない。システム全体がブラックボックスでした。

その後、アーキテクチャを Supervisor + Workers モードに分割。統括 Agent がルーティングを担当し、3 つの専門 Agent がそれぞれの役割を果たす形に。ツール選択エラーは約 3 分の 1 まで減り、各レイヤーを順にトレースできるようになり、デバッグもはるかにクリアになりました。

要するに、Agent が 10 個以上のツールを持つなら、単一 Agent アーキテクチャは限界に来ています。本記事では Supervisor パターンの設計思想から create_supervisor API の使い方まで、Research + Writing チームの実践例まで一通り扱います。コードはそのまま動かせます。GitHub リポジトリへのリンクも末尾に載せています。

一、なぜマルチ Agent システムが必要なのか

単一 Agent の 3 つの弱点

私が踏んだ穴、あなたも今まさに踏んでいるかもしれません。単一 Agent はシンプルに見えて、実は 3 つの致命的な問題を抱えています。

1 つ目：ツールが多すぎて、選択に迷う。

大げさではありません。1 つの Agent が 10 個を超えるツールを持つと、LLM のツール選択ミス率は目に見えて上がります。「モデルは賢くなっているのでは？」——確かに。ただ、ツール説明がプロンプトに詰め込まれ、10 個以上から正しい 1 つを選ぶ認知負荷は小さくありません。

以前のシステムでは、検索ツールとコード実行ツールの説明が重なっていた（どちらも「情報を探す」）。モデルが二者の間を行ったり来たりし、何ラウンドも無駄に消費していました。

2 つ目：コンテキストが積み上がり、ウィンドウが溢れる。

すべてのサブタスクの会話履歴が 1 つのコンテキストウィンドウに押し込まれます。数ラウンド走らせると、ウィンドウは tool call の履歴で埋まり、核心の指示は薄まっていきます。モデルは「物忘れ」し始め、最初のリクエストすら見失うことも。

体感としては、20 ラウンド走った Agent をデバッグしたとき、コンテキストには関数呼び出しの記録ばかり。最終出力は、ユーザーの最初の要望とほとんど関係なくなっていました。

3 つ目：デバッグ不能で、原因特定が困難。

問題が起きても、どのツール呼び出しが原因か分かりません。単一 Agent はブラックボックス。ログは tool call の羅列で、問題箇所を特定するには 1 行ずつ追うしかない。

Alibaba Cloud - 開発者向けクラウドの第一候補、Lighthouse サーバー15%OFF、プロジェクトの迅速な立ち上げを支援

15%割引を入手 →広告

Supervisor パターンならこれらを解消できます。統括 Agent が複数の専門 Agent を調整し、役割を分けて各々が担当領域に集中する——それが基本です。

Supervisor パターンの考え方

シンプルに言えば、分担です。

チームを想像してください。プロジェクトマネージャーが全体を調整し、研究員は調査、エンジニアは実装、ドキュメント担当はレポート執筆。各人が得意分野を持ち、マネージャーは「この仕事は誰に渡すか」だけ分かればよい。

Supervisor パターンも同じ発想です。

• Supervisor（統括 Agent）：実作業はせず、ルーティング・調整・結果統合だけ担当
• Worker Agents（専門 Agent）：各々 1 領域に集中。ツールセットは絞り、役割は明確

メリットは 3 つ。

ツール数は各 Worker に分散し、各 Agent は自分のツールセット内で選ぶだけ。コンテキストも分散し、各 Agent は自分の会話履歴だけ維持。デバッグ時はレイヤーごとにトレースでき、Supervisor がどの Worker に割り当て、Worker が何を実行したかが一目瞭然。

二、Supervisor パターンのアーキテクチャ

まず全体像を図で確認しましょう。

                    ┌─────────────────┐
                    │  ユーザーリクエスト │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │  (統括 Agent)    │
                    │                 │
                    │ ルーティング+調整 │
                    │  + 結果統合     │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
              ▼              ▼              ▼
       ┌──────────┐   ┌──────────┐   ┌──────────┐
       │ Research │   │   Math   │   │ Writing  │
       │  Agent   │   │  Agent   │   │  Agent   │
       │          │   │          │   │          │
       │ 検索ツール │   │ 計算ツール │   │ 生成ツール │
       └────┬─────┘   └────┬─────┘   └────┬─────┘
            │              │              │
            │   Worker     │   Worker     │   Worker
            │   実行結果   │   実行結果   │   実行結果
            │              │              │
            └──────────────┴──────────────┘
                           │
                           ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │   結果統合      │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   最終回答      │
                    └─────────────────┘

コアコンポーネントの役割

Supervisor が担う 3 つのこと：

1. ルーティング：ユーザーリクエストを分析し、どの Worker に渡すか判断
2. 調整：Worker 間のタスクの流れを管理
3. 統合：各 Worker の実行結果をまとめ、最終回答を出力

Worker Agent の役割分担：

各 Worker は専用ツールセットだけを持ちます。Research Agent なら検索と Web スクレイピング、Math Agent なら四則演算。ツールが減れば、選択精度は自然と上がります。

メッセージ受け渡しの仕組み

ここがポイント：グローバルグラフ状態（Global Graph State）。

すべての Agent が同じ状態オブジェクトを共有します。Worker がタスクを終えると、結果を状態の messages フィールドに append。Supervisor は新しい message を見て、次に誰へ渡すか決めます。

append-only の仕組み——メッセージは増えるだけで減らない。会話履歴の完全性が保たれます。

Fan-out / Fan-in

複雑なタスクでは複数 Worker の並列実行が必要になることも。例えば「A と B の 2 製品の市場データを比較して」と聞かれたら、Supervisor は 2 つの Research タスク（A 用・B 用）を同時に投げられます。これが fan-out。

両 Worker が結果を返したら、Supervisor が統合する。これが fan-in。

LangGraph はこの並列パターンをサポートしていますが、基礎編では深掘りしません。応用テクニックで詳しく触れます。

"LangGraph は、各 Agent が独自のツールセットと責務範囲を持ち、Supervisor が調整とタスク分散を行うマルチ Agent システムの構築方法を提供します。"

三、create_supervisor API 詳解

理論はここまで。コードに入りましょう。

インストールとインポート

pip install langgraph-supervisor langchain-openai

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

ツールの定義

各 Worker 用のツールを用意します。

KiloClaw - 企業向けマネージド OpenClaw、AI エージェントのオーケストレーションとスマートルーティングを加速

今すぐ始める →広告

from typing import Annotated

# 数学計算ツール
def add(
    a: Annotated[float, "1 つ目の数値"],
    b: Annotated[float, "2 つ目の数値"]
) -> float:
    """Add two numbers together."""
    return a + b

def multiply(
    a: Annotated[float, "1 つ目の数値"],
    b: Annotated[float, "2 つ目の数値"]
) -> float:
    """Multiply two numbers."""
    return a * b

# 検索ツール（モック）
def web_search(query: str) -> str:
    """Search the web for information."""
    # 本番では Tavily、Serper などに接続
    if "population" in query.lower():
        return "東京の人口は約 1,400 万人（2023 年推計）"
    elif "weather" in query.lower():
        return "東京は本日晴れ、気温 15〜25°C"
    else:
        return f"検索結果：{query}"

Python の Annotated 型ヒントで、各パラメータの意味をモデルに明確に伝えます。ツール関数の docstring も重要——モデルはこれでツールの機能を理解します。

Worker Agent の作成

model = ChatOpenAI(model="gpt-4o")

# 数学専門 Agent
math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="あなたは数学の専門家で、数値計算に特化しています。ユーザーが数学演算を必要とするとき、ツールを使ってタスクを完了してください。"
)

# リサーチ専門 Agent
research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="あなたはベテランのリサーチャーで、情報検索と整理が得意です。ユーザーが資料を調べる必要があるとき、検索ツールで回答を取得してください。"
)

注意点は 3 つ。

1. name フィールドが重要：Supervisor は名前で Worker を識別・呼び出す
2. prompt で役割を定義：この Agent の専門分野を明示
3. ツールセットは絞る：各 Agent に必要なツールだけ。多すぎず少なすぎず

Supervisor の作成

# Supervisor システムを作成
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model,
    prompt="""あなたはチームリーダーで、各専門 Agent を調整します。

ユーザーリクエストに応じて、タスクを誰に渡すか決めてください：
- 数学計算が必要 → math_expert
- 資料検索が必要 → research_expert
- タスク完了 → そのままユーザーに回答

複数の専門家の連携が必要なら、合理的な順序で順に呼び出してください。"""
)

# 実行可能なアプリとしてコンパイル
app = supervisor.compile()

create_supervisor の 3 つの主要パラメータ：

• agents：Worker Agent のリスト
• model：Supervisor 自身が使う LLM
• prompt：タスク分配のルールを Supervisor に伝える

実行例

from langchain_core.messages import HumanMessage

# 数学問題のテスト
result = app.invoke({
    "messages": [HumanMessage(content="123 足す 456 はいくつ？")]
})
print(result["messages"][-1].content)
# 出力：123 足す 456 は 579 です

# 検索問題のテスト
result = app.invoke({
    "messages": [HumanMessage(content="東京の人口は？")]
})
print(result["messages"][-1].content)
# 出力：検索結果によると、東京の人口は約 1,400 万人です

Supervisor がリクエスト種別を自動判断し、正しい Worker にルーティング。ユーザーからは透過的で、裏で複数 Agent が動いていることは意識する必要がありません。

四、実践例：Research + Writing チームの構築

ここまでが基本例。次はより完成度の高いシステム——技術記事を自動で調査・執筆するチームです。

シナリオ

ユーザーが技術テーマを入力すると、システムが自動で：

1. 関連資料を調査
2. 記事アウトラインを生成
3. 本文を執筆
4. 校閲・レビュー

3 つの専門 Agent の連携が必要です。

ツールセットの定義

from typing import TypedDict, List
import json

# モック検索ツール
def tech_search(query: str) -> str:
    """技術資料とドキュメントを検索する。"""
    # 本番では Tavily や Serper に接続
    database = {
        "langgraph": "LangGraph は LangChain が提供する Agent フレームワークで、状態管理と循環グラフ構造をサポートします。",
        "supervisor": "Supervisor パターンはマルチ Agent システムの中核アーキテクチャで、統括 Agent が複数の専門 Agent を調整します。",
        "multi-agent": "マルチ Agent システムはタスク分散と協調により、単一 Agent のツール過多とコンテキスト爆発を解決します。"
    }

    results = []
    for key, value in database.items():
        if key in query.lower():
            results.append(value)

    return json.dumps(results) if results else "関連資料が見つかりません。検索範囲を広げることを推奨します"

# アウトライン生成ツール
def generate_outline(topic: str) -> str:
    """テーマに基づいて記事アウトラインを生成する。"""
    return json.dumps({
        "title": f"{topic} 完全ガイド",
        "sections": [
            "1. 概要と背景",
            "2. コア概念",
            "3. 実践例",
            "4. ベストプラクティス",
            "5. まとめ"
        ]
    }, ensure_ascii=False)

# 本文生成ツール
def write_section(section_title: str, context: str) -> str:
    """タイトルとコンテキストに基づいて段落を生成する。"""
    # 本番ではここで LLM を呼び出す
    return f"## {section_title}\n\n調査資料に基づき、{section_title} の要点は以下のとおり……\n\n"

# レビューツール
def review_content(content: str) -> str:
    """内容の正確性と可読性をレビューする。"""
    issues = []
    if len(content) < 100:
        issues.append("内容が短すぎます。拡充を推奨")
    if "TODO" in content:
        issues.append("未完了の TODO マークが残っています")

    return json.dumps({
        "passed": len(issues) == 0,
        "issues": issues,
        "suggestion": "品質良好、公開可能" if not issues else "指摘事項を修正して再提出してください"
    }, ensure_ascii=False)

Worker Agent の作成

# リサーチャー Agent
researcher = create_react_agent(
    model=model,
    tools=[tech_search],
    name="researcher",
    prompt="""あなたはベテランの技術リサーチャーで、新技術の素早い調査と理解が得意です。

役割：
1. 調査テーマを受け取る
2. 検索ツールで関連資料を探す
3. 構造化された調査レポートにまとめる

注意：調査だけ行い、執筆はしない。結果は writer に渡す。"""
)

# ライター Agent
writer = create_react_agent(
    model=model,
    tools=[generate_outline, write_section],
    name="writer",
    prompt="""あなたは技術ライターで、複雑な概念を分かりやすい記事に変換するのが得意です。

役割：
1. 調査レポートを受け取る
2. 記事アウトラインを生成
3. 各セクションの本文を執筆

注意：初稿完成後、reviewer にレビューを依頼する。"""
)

# レビュアー Agent
reviewer = create_react_agent(
    model=model,
    tools=[review_content],
    name="reviewer",
    prompt="""あなたは厳格なレビュアーで、記事の品質と正確性を担保します。

役割：
1. 内容の完全性と正確性を確認
2. 可読性と論理構成を評価
3. 修正提案または公開承認

注意：問題があれば writer に差し戻す。"""
)

Supervisor ロジックの構築

# StateGraph でカスタム Supervisor を構築
from langgraph.graph import StateGraph, END
from typing import Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

# 状態の定義
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_agent: str

# Supervisor の意思決定ロジック
def supervisor_node(state: AgentState) -> dict:
    """現在の進捗に基づき、次に実行する Agent を決定する。"""
    messages = state["messages"]

    # LLM で意思決定
    decision = model.invoke([
        {"role": "system", "content": """あなたはチームリーダーです。会話履歴に基づき次のステップを決めてください：

- 調査資料がまだない → 'researcher' を返す
- 調査資料はあるが記事がまだ → 'writer' を返す
- 記事はあるが未レビュー → 'reviewer' を返す
- レビュー通過 → 'FINISH' を返す

Agent 名だけを返し、それ以外は出力しないこと。"""},
        *messages
    ])

    next_agent = decision.content.strip()

    # 正しい Agent 名にマッピング
    agent_map = {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "FINISH": END
    }

    return {"next_agent": agent_map.get(next_agent, "researcher")}

# グラフの構築
workflow = StateGraph(AgentState)

# ノードの追加
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

# 条件付きエッジ（Supervisor のルーティング）
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        END: END
    }
)

# すべての Agent 完了後は Supervisor に戻る
for agent in ["researcher", "writer", "reviewer"]:
    workflow.add_edge(agent, "supervisor")

# エントリーポイントの設定
workflow.set_entry_point("supervisor")

# コンパイル
app = workflow.compile()

このアーキテクチャにはループ機構があります。Worker がタスクを終えるたび Supervisor に戻り、Supervisor が次の Worker へ渡すか終了するかを決めます。

実行フロー

result = app.invoke({
    "messages": [HumanMessage(content="LangGraph Supervisor パターンに関する技術記事を書いて")]
})

# 最終結果の確認
print(result["messages"][-1].content)

# 実行トレースの確認
for i, msg in enumerate(result["messages"]):
    print(f"{i+1}. {msg.__class__.__name__}: {msg.content[:100]}...")

実行フローの概略：

ユーザーリクエスト → Supervisor が分析 → Researcher に分配 →
Researcher が調査 → Supervisor に戻る → Writer に分配 →
Writer が執筆 → Supervisor に戻る → Reviewer に分配 →
Reviewer がレビュー → Supervisor に戻る → 完了確認 → 結果出力

各レイヤーをトレースでき、デバッグは格段に楽になります。

五、応用テクニック

メッセージ転送の最適化：create_forward_message_tool

気づいたかもしれません。Worker がタスクを終えると、返却メッセージは Supervisor が受け取り、言い換えられることがある。トークンの無駄遣い。情報が薄まることも。

LangGraph は create_forward_message_tool でこれを解決します。

from langgraph_supervisor.handoff import create_forward_message_tool

# 転送ツールの作成
forward_tool = create_forward_message_tool("supervisor")

# Supervisor 作成時に渡す
supervisor = create_supervisor(
    agents=[researcher, writer, reviewer],
    model=model,
    tools=[forward_tool]  # 転送ツールを追加
)

Supervisor が Worker の応答を要約せず、そのままユーザーに転送できます。トークン節約。効率も上がります。

階層チームアーキテクチャ

プロジェクトがより複雑なら、多層 Supervisor も構築できます。

                    ┌──────────────┐
                    │ Top Supervisor│
                    └──────┬───────┘
                           │
           ┌───────────────┼───────────────┐
           │               │               │
           ▼               ▼               ▼
    ┌────────────┐  ┌────────────┐  ┌────────────┐
    │Research    │  │ Writing    │  │ QA         │
    │Team        │  │ Team       │  │ Team       │
    │Supervisor  │  │ Supervisor │  │ Supervisor │
    └─────┬──────┘  └─────┬──────┘  └─────┬──────┘
          │               │               │
     ┌────┼────┐    ┌────┼────┐    ┌────┼────┐
     │    │    │    │    │    │    │    │    │
   Web  Doc  API   Out  Cont Rev   Test Code Audit
 Search Scrp Parse line ent  iew

各サブチームに Supervisor があり、上に総 Supervisor が調整。大規模プロジェクト向けで、責務の切り分けがより細かくなります。

エラーハンドリング

Worker の実行が失敗したら？

from langgraph.pregel import RetryPolicy

# リトライポリシーの設定
retry_policy = RetryPolicy(
    max_attempts=3,
    initial_interval=1.0,
    backoff_factor=2.0
)

app = workflow.compile(retry_policy=retry_policy)

Supervisor の prompt にエラー処理ロジックを加えることもできます。

Vultr - 高性能NVMeクラウドサーバー、世界32拠点、Dockerワンクリック展開対応

価格を見る →広告

ある Agent の実行が失敗した場合：
1. エラー情報を記録
2. 予備 Agent の呼び出しを試行
3. 複数回失敗したら、ユーザーに問題を報告

状態の永続化

マルチターン対話には状態保存が必要。LangGraph は Checkpointer を提供します。

from langgraph.checkpoint.memory import MemorySaver

# メモリ保存（開発環境）
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 実行時に thread_id を指定
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [HumanMessage(content="...")]}, config=config)

# 後続の会話でコンテキストを保持
result2 = app.invoke({"messages": [HumanMessage(content="前回のタスクを続けて")]}, config=config)

本番環境では Redis や PostgreSQL を Checkpointer に使えます。

"Hierarchical Agent Teams では、より複雑なマルチ Agent 協調を実現する多層 Supervisor アーキテクチャの構築方法を示しています。"

六、本番デプロイの提案

監視とデバッグ

LangSmith は LangChain 公式の監視プラットフォーム。各ステップの実行詳細を追跡できます。

import os

os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-project"

設定後、実行のたびに LangSmith に完全なトレースが残ります。

• 各 Agent の入出力
• ツール呼び出しの引数と戻り値
• トークン消費統計
• 実行時間の分析

デバッグに特に有用。ログを 1 行ずつ追う必要がなくなります。

トークンコスト管理

Supervisor パターンは優れていますが、マルチ Agent はトークン消費を増やします。最適化の提案：

1. Supervisor の prompt を絞る：必要なルーティングロジックだけ
2. forward_message_tool を使う：重複要約を避ける
3. ツールを適切に分配：各 Worker に必要なツールだけ
4. 会話ラウンド数を制限：最大ラウンド数を設定

# 最大ラウンド数の設定
app = workflow.compile(
    checkpointer=memory,
    interrupt_after=20  # 最大 20 ステップ
)

AWS Bedrock との統合

AWS 上のプロジェクトなら Bedrock のモデルが使えます。

from langchain_aws import ChatBedrock

model = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    region_name="us-east-1"
)

# 他のコードはそのまま、model を差し替えるだけ
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model
)

ベストプラクティスまとめ

実戦から得た教訓をまとめます。

1. 小さく始める：まず 2〜3 Agent から。徐々に増やす
2. 役割を明確に：各 Worker の領域をはっきりさせ、重複を避ける
3. LangSmith で監視：開発段階から接続。デバッグが楽になる
4. トークンコストに注意：マルチ Agent は消費を増幅。最適化が必要
5. forward_message_tool を活用：トークンをかなり節約できる
6. 公式リポジトリを参照：langgraph-supervisor-py に完全な例あり

まとめ

Supervisor パターンの本質は分担協調——大きなタスクを小さく分け、各 Agent が得意分野に集中する。

単一 Agent の 3 つの弱点（ツール選択の混乱、コンテキスト爆発、デバッグ困難）から、Supervisor パターンの解決策まで、本記事で一通り押さえました。create_supervisor API 自体は難しくありません。大事なのは裏にあるアーキテクチャの考え方です。

小さなプロジェクトから試すのがおすすめ。2 Agent システム（検索 + 要約など）を組んで動かし、慣れたら段階的に拡張。LangSmith の監視は必須級——デバッグ時に助かります。

完全なコード例は GitHub リポジトリ：langgraph-supervisor-py。公式チュートリアルも読む価値あり：Hierarchical Agent Teams。

参考資料

• LangGraph Supervisor Reference
• Hierarchical Agent Teams Tutorial
• LangGraph Multi-Agent Workflows Blog
• GitHub - langgraph-supervisor-py
• Build Multi-Agent Systems with AWS Bedrock