LiteLLMとLangGraphでSupervisor・Hierarchical・Swarmの3パターンのマルチエージェントを構築してみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langgraph-supervisor 0.1.0
langgraph-swarm 0.0.13
langchain-litellm 0.6.4
langchain-core 1.3.2

$ uv pip install litellm langgraph langgraph-supervisor langgraph-swarm langchain-litellm langchain-core

$ export OPENAI_API_KEY="sk-..."
$ export ANTHROPIC_API_KEY="sk-ant-..."

import litellm

litellm.modify_params = True

import litellm
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph_supervisor import create_supervisor

litellm.modify_params = True

# =============================================
# ツール定義
# =============================================
@tool
def search_web(query: str) -> str:
    """Webから情報を検索します。"""
    # デモ用のダミー実装。実運用ではTavilyやSerperなどを使う
    fake_db = {
        "東京 観光": "2026年の東京観光トレンド: 浅草・豊洲・渋谷のナイトマーケット、お台場の没入型アート展示、神田の日本酒バー巡り",
        "京都 観光": "2026年の京都観光トレンド: 嵐山の早朝座禅体験、伏見の酒蔵ツアー、二条城のプロジェクションマッピング",
        "大阪 観光": "2026年の大阪観光トレンド: 万博会場跡地のリニューアル、道頓堀の屋形船ディナー、新世界のレトロ酒場巡り",
    }
    for key, value in fake_db.items():
        if all(k in query for k in key.split()):
            return value
    return f"'{query}' に関する情報は見つかりませんでした"

@tool
def get_weather(city: str) -> str:
    """指定された都市の天気を取得します。"""
    weather = {
        "東京": "晴れ、最高気温24°C、観光に最適",
        "京都": "曇りのち晴れ、最高気温22°C、屋外散策におすすめ",
        "大阪": "晴れ、最高気温25°C、屋外イベント日和",
    }
    return weather.get(city, f"{city}の天気データはありません")

# =============================================
# エージェントごとに最適なモデルを割り当てる
# =============================================
# Researcher: 大量の検索を捌く高速・安価なモデル
researcher_llm = ChatLiteLLM(model="openai/gpt-5-mini")
# Writer: 文章品質が重要なので高品質モデル（Anthropic）
writer_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")
# Supervisor: 全体の指揮を執るため賢いモデル
supervisor_llm = ChatLiteLLM(model="openai/gpt-5.4")

# =============================================
# 各エージェントの定義
# =============================================
researcher = create_agent(
    model=researcher_llm,
    tools=[search_web, get_weather],
    name="researcher",
    system_prompt=(
        "あなたは調査担当エージェントです。"
        "与えられたテーマについて search_web や get_weather ツールで情報を集め、"
        "事実ベースで簡潔に整理して報告してください。"
        "推測や創作は行わないでください。"
    ),
)

writer = create_agent(
    model=writer_llm,
    tools=[],
    name="writer",
    system_prompt=(
        "あなたは文章作成担当エージェントです。"
        "researcherから渡された情報をもとに、読みやすく構造化されたレポートを作成してください。"
        "見出し・箇条書き・見どころの3要素で整理してください。"
    ),
)

# =============================================
# Supervisorで束ねる
# =============================================
app = create_supervisor(
    agents=[researcher, writer],
    model=supervisor_llm,
    prompt=(
        "あなたは調査チームを統括するマネージャーです。"
        "ユーザーの依頼を受けたら、まず researcher に必要な情報を調査させ、"
        "その結果を writer に渡してレポートを作成させてください。"
        "両方の作業が完了したら、最終結果をユーザーに返してください。"
    ),
).compile()

# =============================================
# 実行
# =============================================
result = app.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "東京の最新の観光トレンドと現在の天気を踏まえて、週末旅行の簡潔なレポートを書いてください",
            }
        ]
    }
)

print("=== 実行ログ ===")
for msg in result["messages"]:
    name = getattr(msg, "name", None) or msg.type
    content = msg.content if isinstance(msg.content, str) else str(msg.content)
    print(f"\n[{name}]")
    print(content[:400])

$ python supervisor_basic.py
=== 実行ログ ===

[human]
東京の最新の観光トレンドと現在の天気を踏まえて、週末旅行の簡潔なレポートを書いてください

[supervisor]

[transfer_to_researcher]
Successfully transferred to researcher

[researcher]
要点まとめ（ウェブ検索と天気予報に基づく）

1) 最新の観光トレンド（ウェブ検索で確認）
- 浅草・豊洲・渋谷などでのナイトマーケットの開催が増加している。
- お台場の没入型アート展示（体験型・フォトジェニックな展示）が注目されている。
- 神田周辺での日本酒バー巡りなど、地元飲食体験の需要が高まっている。

2) 現在の天気（天気予報で確認）
- 晴れ。最高気温 24°C。屋外観光に適した気候。
（以下略）

[transfer_back_to_supervisor]
Successfully transferred back to supervisor

[supervisor]

[transfer_to_writer]
Successfully transferred to writer

[writer]
# 🗼 東京 週末旅行レポート
### ～最新トレンドと天気を活かした充実の2日間プラン～

## ☀️ 1. 現在の天気情報

| 項目 | 詳細 |
|------|------|
| 天候 | ☀️ 晴れ |
| 最高気温 | 24°C |
| 総合評価 | **屋外観光に最適なコンディション** |

## 🔥 2. 東京の最新観光トレンド

- 🎨 **没入型アート体験**：お台場を中心に、フォトジェニックで体験型の展示が急増中
- 🏮 **ナイトマーケットの台頭**：浅草・豊洲・渋谷などで夜の屋外マーケットが人気上昇
- 🍶 **地元飲食体験**：神田エリアの日本酒バー巡り
（以下略）

[transfer_back_to_supervisor]
Successfully transferred back to supervisor

[supervisor]
以下、東京の最新観光トレンドと現在の天気を踏まえた、週末旅行の簡潔なレポートです。
（以下略）

from typing import Literal

import litellm
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.graph import END, START, MessagesState, StateGraph

litellm.modify_params = True

# =============================================
# ツールとエージェント（前節と同じ）
# =============================================
@tool
def search_web(query: str) -> str:
    """Webから情報を検索します。"""
    return f"'{query}'の検索結果: 浅草・豊洲のナイトマーケットが人気"

@tool
def get_weather(city: str) -> str:
    """指定された都市の天気を取得します。"""
    return f"{city}: 晴れ、24°C"

researcher_llm = ChatLiteLLM(model="openai/gpt-5-mini")
writer_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")
supervisor_llm = ChatLiteLLM(model="openai/gpt-5.4")

researcher = create_agent(
    model=researcher_llm,
    tools=[search_web, get_weather],
    system_prompt="あなたは調査担当です。ツールで情報を集めて簡潔に報告してください。",
)

writer = create_agent(
    model=writer_llm,
    tools=[],
    system_prompt="渡された情報をもとに見出し付きの簡潔なレポートを作成してください。",
)

# =============================================
# Supervisor用のルーティングロジック
# =============================================
SUPERVISOR_PROMPT = (
    "あなたはチームのマネージャーです。次に呼ぶべきエージェントを決定してください。"
    "選択肢: 'researcher'（調査担当）、'writer'（執筆担当）、'FINISH'（完了）。"
    "ユーザーの依頼に対し、まずresearcherで情報収集し、次にwriterで執筆、"
    "完了したらFINISHを返してください。"
    "回答は 'researcher' / 'writer' / 'FINISH' のいずれか1語のみ。"
)

def supervisor_node(state: MessagesState) -> dict:
    """次に動くエージェントを決定する。"""
    messages = [{"role": "system", "content": SUPERVISOR_PROMPT}] + [
        {"role": m.type if m.type != "ai" else "assistant", "content": str(m.content)}
        for m in state["messages"]
    ]
    response = supervisor_llm.invoke(messages)
    decision = response.content.strip().lower()
    print(f"  [Supervisor判断] next = {decision}")
    return {"messages": [response]}

def researcher_node(state: MessagesState) -> dict:
    """調査エージェントを実行する。"""
    # 直前のユーザー指示を取得
    user_msg = next(
        (m for m in reversed(state["messages"]) if m.type == "human"),
        None,
    )
    query = user_msg.content if user_msg else ""
    result = researcher.invoke({"messages": [HumanMessage(content=query)]})
    final = result["messages"][-1]
    print(f"  [Researcher] {final.content[:80]}...")
    return {"messages": [final]}

def writer_node(state: MessagesState) -> dict:
    """執筆エージェントを実行する。"""
    # 直前の調査結果を渡す
    last = state["messages"][-1]
    result = writer.invoke({"messages": [HumanMessage(content=str(last.content))]})
    final = result["messages"][-1]
    print(f"  [Writer] {final.content[:80]}...")
    return {"messages": [final]}

def route_next(state: MessagesState) -> Literal["researcher", "writer", "__end__"]:
    """Supervisorの判断に応じて次のノードを決定する。"""
    last = state["messages"][-1]
    decision = str(last.content).strip().lower()
    if "researcher" in decision:
        return "researcher"
    if "writer" in decision:
        return "writer"
    return END

# =============================================
# グラフ構築
# =============================================
graph = StateGraph(MessagesState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    route_next,
    {"researcher": "researcher", "writer": "writer", END: END},
)
# 各Workerが終わったら必ずSupervisorに戻る
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

# =============================================
# 実行
# =============================================
result = app.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "東京の観光トレンドを調べてレポートを書いてください",
            }
        ]
    }
)

print("\n=== 最終出力 ===")
print(result["messages"][-1].content)

$ python supervisor_manual.py
  [Supervisor判断] next = researcher
  [Researcher] 調査報告（概要） — 東京の観光トレンド（2023–2024、要点）
注記：今回の自動検索ツールからは有意な統計ページが取得できなかったため、報告は私の保有知...
  [Supervisor判断] next = finish

=== 最終出力 ===
FINISH

import litellm
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph_supervisor import create_supervisor

litellm.modify_params = True

# =============================================
# Worker レベルのツール
# =============================================
@tool
def search_news(query: str) -> str:
    """ニュースを検索します。"""
    return f"{query}に関する最新ニュース: AI業界で大規模モデルの統合が進行中"

@tool
def search_papers(query: str) -> str:
    """論文を検索します。"""
    return f"{query}に関する論文: マルチエージェントの協調学習に関する研究が多数発表"

@tool
def write_summary(text: str) -> str:
    """テキストを要約します。"""
    return f"要約: {text[:30]}..."

@tool
def write_blog(text: str) -> str:
    """ブログ記事を作成します。"""
    return f"ブログ記事ドラフト: {text[:30]}..."

# =============================================
# モデルの定義
# =============================================
worker_llm = ChatLiteLLM(model="openai/gpt-5-mini")
mid_llm = ChatLiteLLM(model="openai/gpt-5.1")
top_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

# =============================================
# リサーチチーム: news_searcher + paper_searcher
# =============================================
news_searcher = create_agent(
    model=worker_llm,
    tools=[search_news],
    name="news_searcher",
    system_prompt="ニュース検索担当。search_newsで最新ニュースを取得します。",
)
paper_searcher = create_agent(
    model=worker_llm,
    tools=[search_papers],
    name="paper_searcher",
    system_prompt="論文検索担当。search_papersで関連論文を取得します。",
)
research_team = create_supervisor(
    agents=[news_searcher, paper_searcher],
    model=mid_llm,
    prompt="リサーチチームのリーダー。ニュースと論文を両方調査してください。",
    supervisor_name="research_lead",
).compile(name="research_team")

# =============================================
# 執筆チーム: summarizer + blogger
# =============================================
summarizer = create_agent(
    model=worker_llm,
    tools=[write_summary],
    name="summarizer",
    system_prompt="要約担当。write_summaryで短くまとめます。",
)
blogger = create_agent(
    model=worker_llm,
    tools=[write_blog],
    name="blogger",
    system_prompt="ブログ記事作成担当。write_blogで記事を作成します。",
)
writing_team = create_supervisor(
    agents=[summarizer, blogger],
    model=mid_llm,
    prompt="執筆チームのリーダー。要約とブログ記事の両方を作成してください。",
    supervisor_name="writing_lead",
).compile(name="writing_team")

# =============================================
# トップレベルのSupervisor
# =============================================
app = create_supervisor(
    agents=[research_team, writing_team],
    model=top_llm,
    prompt=(
        "あなたは編集長です。"
        "research_teamに調査を依頼し、その結果をwriting_teamに渡して記事化してください。"
    ),
).compile()

# =============================================
# 実行
# =============================================
result = app.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "マルチエージェントの最新動向について記事を作ってください",
            }
        ]
    }
)
print(result["messages"][-1].content)

$ python hierarchical.py
両チームから成果物が上がってきました。以下が最終的な記事です！

# 📰 マルチエージェントの最新動向

## 🔍 要約

| ポイント | 内容 |
|------|------|
| **なぜ今？** | LLMとの融合・大規模化で第二の波が到来 |
| **主な技術** | LLM×RL・CTDE・GNN・ポピュレーション学習・発現的コミュニケーション |
| **応用分野** | ロボット群・自動運転・スマートグリッド・ゲームAI・市場シミュレーション |
| **主な課題** | 非定常性・クレジット割当・スケール・LLMコスト・倫理ガバナンス |
| **今後** | ハイブリッドエージェントの標準化・ガバナンス整備・社会インフラへの実装 |

## 1. なぜ今マルチエージェントなのか
（中略）

## 5. 実務で使い始めるためのヒント

- ✅ まず「何を最適化するか」を決める（個別最適 vs 全体最適）
- ✅ 既存ベースライン（MAPPO, QMIX, PettingZoo）から始める
- ✅ まず5〜10体の小規模プロトタイプで課題を把握してからスケール
- ✅ 評価指標は「報酬」だけでなく、公平性・安全性・説明可能性も含める
（以下略）

import litellm
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph_swarm import create_handoff_tool, create_swarm

litellm.modify_params = True

# =============================================
# 業務固有ツール
# =============================================
@tool
def get_pricing(plan: str) -> str:
    """料金プランの情報を取得します。"""
    pricing = {
        "basic": "Basicプラン: 月額1,000円、ユーザー5名まで",
        "pro": "Proプラン: 月額5,000円、ユーザー50名まで",
        "enterprise": "Enterpriseプラン: 個別見積もり、無制限ユーザー",
    }
    return pricing.get(plan.lower(), f"プラン '{plan}' は見つかりませんでした")

@tool
def diagnose_error(error_code: str) -> str:
    """エラーコードから原因と対処法を返します。"""
    errors = {
        "E001": "認証エラー: APIキーが正しいか確認してください",
        "E002": "レート制限: しばらく待ってから再試行してください",
        "E003": "通信エラー: ネットワーク接続を確認してください",
    }
    return errors.get(error_code.upper(), f"エラーコード '{error_code}' は不明です")

# =============================================
# ハンドオフ用ツール
# =============================================
to_tech = create_handoff_tool(
    agent_name="tech_support",
    description="技術的な質問・エラー対応の場合に技術サポートへ引き継ぐ",
)
to_sales = create_handoff_tool(
    agent_name="sales",
    description="料金・契約に関する質問の場合に営業へ引き継ぐ",
)

# =============================================
# 各エージェント
# =============================================
sales_llm = ChatLiteLLM(model="openai/gpt-5.4")
tech_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

sales = create_agent(
    model=sales_llm,
    tools=[get_pricing, to_tech],
    name="sales",
    system_prompt=(
        "あなたは営業担当です。"
        "料金プランや契約に関する質問に答えてください。"
        "技術的な質問・エラー対応が来たら、必ず tech_support にハンドオフしてください。"
    ),
)

tech_support = create_agent(
    model=tech_llm,
    tools=[diagnose_error, to_sales],
    name="tech_support",
    system_prompt=(
        "あなたは技術サポート担当です。"
        "エラーや技術的な質問に答えてください。"
        "料金・契約に関する質問が来たら、必ず sales にハンドオフしてください。"
    ),
)

# =============================================
# Swarm を構築
# =============================================
swarm = create_swarm(
    agents=[sales, tech_support],
    default_active_agent="sales",
)
# Swarmは「現在の担当エージェント」を状態として持つため checkpointer が必須
checkpointer = InMemorySaver()
app = swarm.compile(checkpointer=checkpointer)

# =============================================
# 実行: 同じスレッドで担当が切り替わる様子を観察
# =============================================
config = {"configurable": {"thread_id": "user-001"}}

queries = [
    "Proプランの料金を教えてください",  # → sales
    "エラーコード E002 が出ています",  # → sales が tech_support にハンドオフ
    "Enterpriseプランの料金は？",  # → tech_support が sales にハンドオフ
]

for query in queries:
    print(f"\n{'=' * 60}")
    print(f"USER: {query}")
    print(f"{'=' * 60}")
    result = app.invoke(
        {"messages": [{"role": "user", "content": query}]},
        config=config,
    )
    last = result["messages"][-1]
    active = result.get("active_agent", "?")
    print(f"[現在の担当: {active}]")
    print(f"AI: {last.content}")

$ python swarm.py

============================================================
USER: Proプランの料金を教えてください
============================================================
[現在の担当: ?]
AI: Proプランは月額5,000円、ユーザー50名までご利用いただけます。

============================================================
USER: エラーコード E002 が出ています
============================================================
[現在の担当: tech_support]
AI: エラーコード **E002** の診断結果をお伝えします。

---

**原因：レート制限**
短時間に多数のリクエストが送信されたため、レート制限がかかっています。

**対処法：**
しばらく時間をおいてから、再度お試しください。

---

改善しない場合や、さらにご不明な点がございましたら、お気軽にお知らせください！

============================================================
USER: Enterpriseプランの料金は？
============================================================
[現在の担当: sales]
AI: Enterpriseプランは**個別見積もり**、**ユーザー数は無制限**です。