マルチエージェント協調の実践：4 つのアーキテクチャパターン選択ガイド

Anthropic の調査データによると、マルチエージェントシステムはシングルエージェントより 90.2% も高いパフォーマンスを発揮します。シングルエージェントには 3 つの致命的な弱点があります。コンテキストの制限（200k トークンを超えると出力が崩れ始める）、能力の分散（スキルを詰め込みすぎて「器用貧乏」になる）、デバッグの難しさ（1 つの Agent が落ちても原因を特定できない）です。マルチエージェント協調システムは AI 界の「マイクロサービスアーキテクチャ」のようなもので、各 Agent は 1 つのことだけを担当し、明確なメッセージのやり取りで協力します。

4 つのコアアーキテクチャパターンには、それぞれ適した場面があります。Subagents（中央オーケストレーション）は複数の独立ドメインの並列処理に、Skills（オンデマンド読み込み）はシングルスレッドの多段階処理に、Handoffs（状態駆動）は多段階の対話に、Router（並列ディスパッチ）は複数データソースのクエリに向いています。この記事では、選択ロジックから本番環境での落とし穴、LangGraph のコード実装まで解説し、「Subagents と Skills のどちらを使うか」「Agent 間の状態をどう管理するか」を判断できるよう支援します。

なぜマルチエージェントシステムが必要なのか

正直なところ、シングル Agent の最大の問題は「使えるかどうか」ではなく「どれだけ長く使えるか」です。

こんな経験はありませんか。最初はきれいなコードを書いていた Agent が、突然めちゃくちゃな出力をし始める。あるいは、A だけを聞いたのに、B にも C にも話を広げ、最後には Z まで持ち出してくる。これは Agent が「バカ」だからではなく、コンテキストウィンドウを超えてしまったからです。

シングル Agent には 3 つの致命的な弱点があります。

コンテキストの制限。どんなに強力な Agent でも、200k トークンのコンテキストには限りがあります。コードレビュー、セキュリティ分析、パフォーマンス最適化を同時に処理させると、半分覚えていられれば上出来です。私はある Agent が同じ対話の中で、最初の 20 ターンは Python の話をしていたのに、21 ターン目から JavaScript を出力し始めるのを見たことがあります。自分が何をすべきか完全に忘れてしまったのです。

能力の分散。1 つの Agent にスキルを詰め込めば詰め込むほど、「器用貧乏」になっていきます。何でも少しはできるけれど、何も極められていない状態です。コードレビューを頼んだら単体テストを書いてきたり、ドキュメントを頼んだらついでにコードをリファクタリングしたり。方向感覚？そんなものはありません。

デバッグの難しさ。1 つの Agent が落ちても、どの段階で問題が起きたのか全くわかりません。プロンプトが長すぎる？ツール呼び出しが失敗した？それともコンテキストの汚染？原因究明は砂浜で針を探すようなものです。

マルチエージェントシステムは、ありていに言えば AI 界の「マイクロサービスアーキテクチャ」です。各 Agent は 1 つのことだけをし、それをしっかりやり遂げます。すべてを 1 つの「スーパー Agent」に詰め込むのではなく、明確なメッセージのやり取りで協力します。

Google、LangChain、Anthropic はみなこの考え方を推進しています。O’Reilly のレポートによると、2025 年の Agent 関連論文は年初の 820 本から 2500 本以上へと、3 倍に増えました。なぜでしょうか。シングル Agent で天下を取る時代が、終わったと誰もが気づいたからです。

4 つのコアアーキテクチャパターン

LangChain と Google は、いくつかの主流なマルチエージェントアーキテクチャをまとめています。私もいくつかのプロジェクトを回してみて、それぞれに適した場面があると気づきました。選択を誤ると「大砲で蚊を撃つ」か「箸でスープを飲む」状態になります。

Subagents（サブエージェント）- 中央オーケストレーションパターン

これは最も直感的なパターンです。1 つの「メイン Agent」が指揮官となり、その配下に「サブ Agent」の一群を従えます。サブ Agent はメイン Agent のツールであり、メイン Agent がいつ誰を呼び出すかを決めます。

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

価格を見る →広告

ユーザーリクエスト → メイン Agent（コーディネーター）→ サブ Agent A/B/C へディスパッチ → 結果を集約 → ユーザーへ返却

いつ使うか？ タスクが複数の独立した領域にまたがる場合です。たとえばカスタマーサポートシステム。あるサブ Agent は注文照会を、別のサブ Agent は返金を、また別のサブ Agent はクレーム対応を処理します。各領域は独自のナレッジベースとツールを持ち、メイン Agent は「振り分け」だけを担います。

コード例（LangGraph）：

from langgraph.prebuilt import create_react_agent

# サブ Agent を定義
order_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[query_order, update_order],
    prompt="あなたは注文の専門家です。注文に関する問題だけを処理してください。"
)

refund_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[check_refund_policy, process_refund],
    prompt="あなたは返金の専門家です。返金に関する問題だけを処理してください。"
)

# メイン Agent はサブ Agent をツールとして保持する
main_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[order_agent, refund_agent],  # サブ Agent がツールになる
    prompt="あなたはカスタマーサポートの統括です。ユーザーの問題に応じて適切な専門家へ振り分けてください。"
)

メリット：コンテキストの分離がきれいで、各サブ Agent は自分が見るべきものだけを見ます。並列実行の効率も高いです。

デメリット：各サブ Agent は独立した LLM 呼び出しなので、トークン消費が大きくなります。サブ Agent 同士で状態を共有したい場合は、一手間かかります。

Skills（スキル）- オンデマンド読み込みパターン

1 つの Agent に複数の「人格」を持たせます。Skills は本質的に動的に読み込まれるプロンプトテンプレートです。Agent はタスクに応じて「身分」を切り替えますが、終始同じ Agent のままです。

ユーザーリクエスト → 単一 Agent →「コードレビュー」Skill を読み込み → 実行 →「ドキュメント生成」Skill を読み込み → 実行

いつ使うか？ タスクを「シングルスレッド」で処理する必要があるが、段階ごとに異なる専門知識が必要な場合です。たとえばプログラミングアシスタント。コードを書くときは「開発者」モード、ドキュメントを書くときは「テクニカルライター」モードを使います。

コード例：

# Skills ディレクトリ構造
skills/
├── code_review.md      # コードレビューのプロンプト
├── doc_writer.md       # ドキュメント生成のプロンプト
└── security_audit.md   # セキュリティ監査のプロンプト

# Skill を動的に読み込む
def load_skill(skill_name: str) -> str:
    with open(f"skills/{skill_name}.md") as f:
        return f.read()

# 使用例
agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[...],
    prompt=load_skill("code_review")  # 実行時に切り替え
)

メリット：軽量で、追加の Agent 調整コストが不要です。トークン消費は Subagents より低くなります。

デメリット：コンテキストが蓄積します。Skill を 10 回切り替えると、それまでの 9 回分の Skill の内容がコンテキストに残り、どんどん散らかっていきます。

Handoffs（引き継ぎ）- 状態駆動パターン

Agent 同士がパスを回すようにタスクを引き継ぎます。Agent A が仕事を終えると、状態を Agent B に「放り投げ」、Agent B が続きを行います。リレー競走に少し似ています。

ユーザーリクエスト → Agent A（情報収集）→ 引き継ぎ → Agent B（問題分析）→ 引き継ぎ → Agent C（解決策の提示）

いつ使うか？ 多段階の対話シーンです。たとえばテクニカルサポートのフロー。まず問題を収集 → 問題を診断 → 解決策を提示 → 解決を確認。各段階で異なる専門知識が必要になる場合があります。

コード例：

from langchain_core.tools import tool

# 引き継ぎツールを定義
@tool
def handoff_to_diagnosis(issue_summary: str) -> str:
    """問題を診断の専門家へ引き継ぐ。"""
    return f"問題を受け取りました：{issue_summary}、診断を開始します..."

@tool
def handoff_to_solution(diagnosis_result: str) -> str:
    """診断結果を解決策の専門家へ引き継ぐ。"""
    return f"診断に基づき：{diagnosis_result}、解決策を策定中..."

# Agent チェーン
triage_agent = create_react_agent(
    tools=[handoff_to_diagnosis],
    prompt="あなたは問題の仕分け担当です。ユーザーの問題を収集し、診断の専門家へ引き継いでください。"
)

diagnosis_agent = create_react_agent(
    tools=[handoff_to_solution],
    prompt="あなたは診断の専門家です。問題の根本原因を分析し、解決策の専門家へ引き継いでください。"
)

メリット：対話の流れが自然で、人間の協力の直感に合っています。各 Agent は現在の段階だけに集中します。

デメリット：状態管理が複雑です。Agent A から Agent B へ渡すデータ形式が正しいことを保証しないと、チェーンが途切れてしまいます。

Router（ルーター）- 並列ディスパッチパターン

1 つの「ルーター Agent」がリクエストを分析し、複数の専門 Agent を並列で呼び出して、最後に結果を統合します。

ユーザーリクエスト → Router（分類）→ Agent A/B/C を並列呼び出し → 結果の統合 → ユーザーへ返却

いつ使うか？ 1 つのリクエストで複数のデータソースを照会する必要がある場合です。たとえば企業ナレッジベースの Q&A。Router が問題のタイプを判断し、社内ドキュメント、外部 API、データベースを並列で照会して、最後に回答をまとめます。

コード例：

from langgraph.graph import StateGraph

# 並列実行ノードを定義
async def query_internal_docs(state):
    # 社内ドキュメントを照会
    return {"internal_results": [...]}

async def query_external_api(state):
    # 外部 API を照会
    return {"external_results": [...]}

async def query_database(state):
    # データベースを照会
    return {"db_results": [...]}

async def synthesize(state):
    # すべての結果を統合
    all_results = state["internal_results"] + state["external_results"] + state["db_results"]
    return {"final_answer": summarize(all_results)}

# 並列グラフを構築
graph = StateGraph(State)
graph.add_node("internal", query_internal_docs)
graph.add_node("external", query_external_api)
graph.add_node("database", query_database)
graph.add_node("synthesize", synthesize)

# 並列実行
graph.add_edge("router", ["internal", "external", "database"])
graph.add_edge(["internal", "external", "database"], "synthesize")

メリット：並列実行で、速度が最も速いです。ステートレスで、各クエリが独立しています。

Railway - モダンなデプロイプラットフォーム、煩わしい運用から解放、数分で公開

今すぐ始める →広告

デメリット：複数ターンの対話には向きません。各リクエストは新規で、Agent は前のターンで何を話したか覚えていません。

アーキテクチャ選択の意思決定フレームワーク

ここまで色々と話しましたが、結局どれを選べばいいのでしょうか。シンプルな意思決定フローを描いてみました。

あなたのニーズは何か？
         │
         ├─→ 複数の独立ドメインを並列処理する必要がある？
         │         │
         │         └─→ Subagents（中央オーケストレーション）
         │
         ├─→ シングル Agent で多段階のスキル切り替え？
         │         │
         │         └─→ Skills（オンデマンド読み込み）
         │
         ├─→ 順次的なワークフローで、バトンを次々渡す？
         │         │
         │         └─→ Handoffs（状態駆動）
         │
         └─→ 複数データソースのクエリを統合する必要がある？
                   │
                   └─→ Router（並列ディスパッチ）

フローだけでは直感的でないかもしれないので、比較表も整理しました。

この表はどう読むのか？

• 分散開発：チームが手分けして別々のモジュールを開発していますか？そうなら Subagents も Skills も適しており、各メンバーが 1 つのサブ Agent や Skill を担当できます。
• 並列化：速度を追求しますか？Router と Subagents は複数の Agent を並列で動かせて、効率が最も高くなります。
• マルチホップ対話：ユーザーは複数ターンのやり取りが必要ですか？Handoffs と Skills は対話フローを自然にサポートします。
• ユーザーとの直接対話：ユーザーがサブ Agent と直接話しますか？Skills と Handoffs はサポートしますが、Router はサポートしません。
• トークン消費：コストに敏感なら、Skills が最も節約でき、Router と Subagents が最も消費します。

私の経験では、シンプルに始めるのが鉄則です。まず Skills か Handoffs で MVP を動かし、ボトルネックを見つけてから Subagents や Router へアップグレードします。いきなり分散アーキテクチャに手を出さないこと。過剰設計の苦しみは、私もよくわかっています。

本番レベルの実装ポイント

デモから本番までの間には、太平洋ほどの隔たりがあります。これらの落とし穴は、私が全部踏み抜いてきました。

状態管理

複数 Agent が状態を共有するとき、最も問題が起きやすいのが「競合状態（レースコンディション）」です。2 つの Agent が同じ変数を同時に書き込んだら、最後にどちらが上書きするのでしょうか。

LangGraph の解決策は output_key です。各 Agent は自分専用のキーにしか書き込めません。

from langgraph.graph import StateGraph, MessagesState

class GraphState(MessagesState):
    security_result: str = ""   # セキュリティ Agent 専用
    style_result: str = ""      # スタイル Agent 専用
    perf_result: str = ""       # パフォーマンス Agent 専用

# セキュリティ Agent は security_result だけを書き込む
async def security_agent(state: GraphState):
    result = await analyze_security(state["messages"])
    return {"security_result": result}  # このキーだけを書き込む

# スタイル Agent は style_result だけを書き込む
async def style_agent(state: GraphState):
    result = await analyze_style(state["messages"])
    return {"style_result": result}

こうすれば、並列でも直列でも、各 Agent は自分の領分だけを動かし、互いに干渉しません。

もう 1 つよくある問題が「コンテキストの汚染」です。Agent A の出力が Agent B に読まれてしまうが、B にはその情報が全く必要ない、というケースです。私のやり方は、状態に relevant_keys フィールドを追加し、各 Agent が自分に必要なキーだけを読むようにすることです。

パフォーマンス最適化

マルチエージェントシステムのトークン消費は底なし沼です。トークンを節約するコツをいくつか紹介します。

1. Subagents は Skills より 67% トークンを節約（マルチドメインの場合）

LangChain のテストデータによると、1 つのタスクが 3 つの独立領域にまたがる場合、Subagents のトークン消費は Skills の 3 分の 1 です。なぜでしょうか。Subagents はコンテキストが分離されていて、各サブ Agent は自分の領域の内容だけを見るからです。Skills の場合は、すべての Skill のコンテキストが累積し、どんどん大きくなっていきます。

2. ステートフルなパターンで重複呼び出しを 40〜50% 節約

タスクに大量の重複クエリがある場合（同じ問題を 10 回聞くなど）、ステートフルな Handoffs パターンを使えば、Agent は以前の回答を覚えていられます。LangChain のデータでは、ステートフルはステートレスより LLM 呼び出しを半分近く節約します。

3. リフレクションパターンで反復回数を制限する

多くの人は Agent に「リフレクション（内省）」能力を持たせたがります。自分で出力をチェックし、問題を発見し、再生成させるのです。これ自体は良いのですが、無限ループに陥りやすい欠点があります。私はいつも max_iterations=2 か 3 に制限し、それを超えたら強制終了させます。

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

15%割引を入手 →広告

from langgraph.checkpoint.memory import MemorySaver

# 反復の上限を設定
graph = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[...],
    checkpointer=MemorySaver(),
    config={"configurable": {"max_iterations": 3}}  # 最大 3 回まで内省
)

よくある落とし穴

無限ループ：Agent が自分を呼び出し、自分が自分を呼び、自分が自分を呼び……果てしなく続きます。解決策：max_iterations と明確な終了条件を設定します。

def should_continue(state):
    if state["iteration_count"] >= 3:
        return "end"
    if "done" in state["messages"][-1].content:
        return "end"
    return "continue"

コンテキストの肥大化：Agent がどんどん「バカ」になり、出力が短くなっていきます。多くの場合、コンテキストに詰め込みすぎが原因です。解決策：Blackboard パターン（共有黒板）を使い、必要なコンテキストだけを残して定期的に整理します。

調整税（コーディネーションタックス）：Agent の数が増えると、通信オーバーヘッドが指数関数的に増大します。私はあるシステムをテストしたことがあります。3 個の Agent から 10 個に増やしたところ、応答時間が 2 秒から 15 秒になりました。解決策：役割が近い Agent を統合し、Agent 数を 5 個以内に抑えます。

完全な実装例

理論ばかり話してきたので、実際のものを見てみましょう。私はコードレビューのマルチエージェントシステムを組んでみました。使ったのは Router + ParallelAgent パターンです。

アーキテクチャ：Router がコードの言語とタイプを判断 → セキュリティ監査、スタイルチェック、パフォーマンス分析の 3 つの Agent を並列呼び出し → 結果を統合してレポートを出力。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

# 状態を定義
class CodeReviewState(TypedDict):
    code: str
    language: str
    security_issues: list
    style_issues: list
    perf_issues: list
    final_report: str

# LLM を初期化
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# Router: 言語を判断
async def route_code(state: CodeReviewState) -> dict:
    code = state["code"]
    # 簡易判断。実際には LLM で分類してもよい
    if "def " in code or "import " in code:
        language = "python"
    elif "function" in code or "const " in code:
        language = "javascript"
    else:
        language = "unknown"
    return {"language": language}

# セキュリティ監査 Agent
async def security_audit(state: CodeReviewState) -> dict:
    code = state["code"]
    prompt = f"""あなたはセキュリティ監査の専門家です。以下のコードのセキュリティ問題を確認してください：
- SQL インジェクションのリスク
- XSS 脆弱性
- 機密情報の漏洩
- 安全でない依存関係

コード：
{code}

JSON リスト形式で問題を出力してください。各項目には line（行番号）、severity（深刻度）、description（説明）を含めます。
"""
    response = await llm.ainvoke(prompt)
    # 結果を解析...
    return {"security_issues": []}

# スタイルチェック Agent
async def style_check(state: CodeReviewState) -> dict:
    code = state["code"]
    language = state["language"]
    prompt = f"""あなたはコードスタイルの専門家です。以下の {language} コードのスタイル問題を確認してください：
- 命名規約
- コードフォーマット
- コメントの完全性

コード：
{code}

JSON リスト形式で問題を出力してください。
"""
    response = await llm.ainvoke(prompt)
    return {"style_issues": []}

# パフォーマンス分析 Agent
async def perf_analysis(state: CodeReviewState) -> dict:
    code = state["code"]
    prompt = f"""あなたはパフォーマンス分析の専門家です。以下のコードのパフォーマンス問題を確認してください：
- 時間計算量が高すぎる
- 不要なループ
- メモリリークのリスク

コード：
{code}

JSON リスト形式で問題を出力してください。
"""
    response = await llm.ainvoke(prompt)
    return {"perf_issues": []}

# 統合レポート
async def generate_report(state: CodeReviewState) -> dict:
    security = state.get("security_issues", [])
    style = state.get("style_issues", [])
    perf = state.get("perf_issues", [])

    total_issues = len(security) + len(style) + len(perf)

    report = f"""# コードレビューレポート

## 概要
- 言語：{state['language']}
- 問題総数：{total_issues}

## セキュリティ問題 ({len(security)} 件)
{format_issues(security)}

## スタイル問題 ({len(style)} 件)
{format_issues(style)}

## パフォーマンス問題 ({len(perf)} 件)
{format_issues(perf)}

## 提案
以上の分析に基づき、セキュリティ問題を優先的に修正することをおすすめします...
"""
    return {"final_report": report}

# グラフを構築
graph = StateGraph(CodeReviewState)
graph.add_node("router", route_code)
graph.add_node("security", security_audit)
graph.add_node("style", style_check)
graph.add_node("perf", perf_analysis)
graph.add_node("report", generate_report)

# フロー：Router → 3 つのチェックを並列実行 → レポート生成
graph.set_entry_point("router")
graph.add_edge("router", "security")
graph.add_edge("router", "style")
graph.add_edge("router", "perf")
graph.add_edge("security", "report")
graph.add_edge("style", "report")
graph.add_edge("perf", "report")
graph.add_edge("report", END)

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

# 使用
async def review_code(code: str):
    result = await app.ainvoke({"code": code})
    return result["final_report"]

この例を動かしてみると、100 行のコードを 3 つの Agent が並列実行し、だいたい 3〜5 秒で結果が出ます。直列実行なら、少なくとも 10 秒はかかります。

もちろん、これは基礎的なバージョンにすぎません。本番環境では、さらにキャッシュ（同じコードを重複してレビューしない）、増分レビュー（変更部分だけを見る）、人間のフィードバック（ユーザーに誤検出をマークさせる）を追加する必要があります。しかしこれらの拡張は、すべてこのアーキテクチャの上に積み上げるものです。

結論

ここまで色々話しましたが、要は 3 つの言葉に尽きます。

根底のロジック：パターンの選択はフレームワークの選択より重要です。LangGraph、AutoGen、CrewAI はどれも良いツールですが、Handoffs が必要な問題を Router パターンで解こうとすれば、どんなに良いフレームワークでも救ってはくれません。

中間の戦略：シンプルに始め、段階的にアップグレードする。まず Skills か Handoffs の MVP を動かし、ボトルネックを見つけてから Subagents や Router を検討します。過剰設計こそ最大の落とし穴です。私も踏みました。あなたは踏まないでください。

最上層の実践：本番環境では状態管理、パフォーマンス、コストに注目します。トークン消費、無限ループ、コンテキストの汚染。この 3 つの問題を片付ければ、あなたのマルチエージェントシステムは安定して動くようになります。

次のアクション：LangGraph のドキュメントを開き、1 つのパターンを選んで、50 行のコードで最もシンプルなマルチエージェントシステムを実装してみましょう。考えすぎず、まず動かすことです。