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

深夜3時、私のシングルエージェントシステムが壊れた。クラッシュしたわけではなく、3ページもの無意味な出力を吐き出したのだ。「コードスタイルをチェックして」と頼んだだけなのに。画面を見つめながら思った。このシステム、私の話好きな同僚よりも長文を書くんじゃないか？

これは特殊なケースではない。過去1年間、エージェント関連の論文数は820本から2500本以上に急増し、誰もが同じ疑問を抱いている。シングルエージェントで本当に十分なのか？

Anthropicの研究が衝撃的な事実を突きつけた。マルチエージェントシステムはシングルエージェントより90.2%も性能が高い。ほぼ2倍だ。「一人のスーパーエージェントで全てをこなす」という考えが、「一人でチーム全体の仕事をこなす」と同様に非現実的だと気づかされた。

この記事では、マルチエージェント協調システムの設計について解説する。4つのコアアーキテクチャパターンの選択ロジックから、本番環境での教訓、そしてすぐに実行できるコード実装まで。もしあなたも「Subagentsを使うべきかSkillsを使うべきか」「エージェント間の状態をどう管理するか」で悩んでいるなら、この記事でかなりの時間を節約できるはずだ。

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

正直なところ、シングルエージェントの最大の問題は「使えるかどうか」ではなく「どれくらい使えるか」にある。

こんな経験はないだろうか。あるエージェントが最初は良いコードを書いていたのに、突然めちゃくちゃな出力を始めた。あるいは、Aについて聞いたのに、B、C、そして最終的にZまで話題が逸れてしまった。これはエージェントが「愚か」なのではなく、コンテキストウィンドウが限界を超えているのだ。

シングルエージェントには3つの致命的な問題がある。

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

能力分散。エージェントに多くのスキルを与えれば与えるほど、「何でも屋」になってしまう。何でも少しはできるが、何も深くはできない。コードレビューを頼んだのにユニットテストを書き、ドキュメントを頼んだのにコードをリファクタリングする。方向性？存在しない。

デバッグ困難。エージェントが失敗した時、どの段階で問題が起きたのか分からない。プロンプトが長すぎたのか？ツール呼び出しが失敗したのか？それともコンテキスト汚染か？調査は大海から針を探すようなものだ。

マルチエージェントシステムとは、AI界の「マイクロサービスアーキテクチャ」だ。各エージェントは一つのことだけを行い、それをうまく行う。エージェント間は明確なメッセージパッシングで協調し、「スーパーエージェント」に全てを詰め込むことはしない。

Google、LangChain、Anthropicがこのアプローチを推進している。O’Reillyのレポートによると、2025年のエージェント関連論文は年初の820本から2500本以上に増加し、3倍になった。なぜか？シングルエージェントで世界を征服する時代が終わったと気づいたからだ。

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

LangChainとGoogleはいくつかの主流なマルチエージェントアーキテクチャをまとめた。いくつかのプロジェクトを試した後、それぞれに適したシナリオがあることが分かった。選択を間違えると「大砲で蚊を叩く」か「箸でスープを飲む」ことになる。

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

最も直感的なパターンだ。「メインエージェント」が指揮官となり、「サブエージェント」を束ねる。サブエージェントはメインエージェントのツールであり、メインエージェントが誰をいつ呼び出すかを決定する。

ユーザーリクエスト → メインエージェント（コーディネーター）→ サブエージェントA/B/Cに分散 → 結果を集約 → ユーザーに返却

いつ使うべきか？ タスクが複数の独立したドメインに関わる場合。例えばカスタマーサービスシステム：あるサブエージェントが注文照会を処理し、別のエージェントが返金を処理し、また別のエージェントがクレームを処理する。各ドメインには独自のナレッジベースとツールがあり、メインエージェントは「振り分け」だけを担当する。

コード例（LangGraph）：

from langgraph.prebuilt import create_react_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="あなたは返金の専門家です。返金関連の問題のみを処理してください。"
)

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

利点：コンテキストがきれいに分離され、各サブエージェントは自分が見るべきものだけを見る。並列実行の効率が高い。

欠点：各サブエージェントが独立したLLM呼び出しになるため、トークン消費が大きい。サブエージェント間で状態を共有する場合、遠回りが必要になる。

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

一つのエージェントで複数の「ペルソナ」を持つ。Skillsは本質的に動的に読み込まれるプロンプトテンプレートだ。エージェントはタスクに応じて「アイデンティティ」を切り替えるが、常に同じエージェントである。

ユーザーリクエスト → 単一エージェント → 「コードレビュー」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")  # 実行時に切り替え
)

利点：軽量で、エージェント間の調整オーバーヘッドが不要。トークン消費はSubagentsより低い。

欠点：コンテキストが蓄積される。10回Skillを切り替えると、前の9回のSkillの内容がコンテキストに残り、どんどん乱雑になる。

Handoffs（ハンドオフ）- 状態駆動パターン

エージェント間でバトンを渡すようにタスクを受け渡す。エージェントAが作業を終え、状態をエージェントBに「投げ」、エージェントBが継続する。リレー競走のようなものだ。

ユーザーリクエスト → エージェントA（情報収集）→ 移行 → エージェントB（問題分析）→ 移行 → エージェント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}、ソリューションを作成中..."

# エージェントチェーン
triage_agent = create_react_agent(
    tools=[handoff_to_diagnosis],
    prompt="あなたは問題分類担当です。ユーザーの問題を収集し、診断専門家に移管してください。"
)

diagnosis_agent = create_react_agent(
    tools=[handoff_to_solution],
    prompt="あなたは診断専門家です。問題の根本原因を分析し、ソリューション専門家に移管してください。"
)

利点：会話フローが自然で、人間の協調の直感に合う。各エージェントは現在の段階のみに集中できる。

欠点：状態管理が複雑。エージェントAからエージェントBに渡すデータの形式が正しいことを保証する必要がある。そうでないとチェーンが切れる。

Router（ルーター）- 並列分散パターン

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

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

いつ使うべきか？ 一つのリクエストが複数のデータソースを照会する必要がある場合。例えば企業のナレッジベース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")

利点：並列実行で最速。ステートレスで各クエリが独立している。

欠点：マルチターン会話には適さない。各リクエストが新しく、エージェントは前回の会話を覚えていない。

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

ここまで説明してきたが、結局どれを選ぶべきか？簡単な決定フローを描いた。

あなたの要件は何か？
         │
         ├─→ 複数の独立したドメインを並列処理する必要がある？
         │         │
         │         └─→ Subagents（中央オーケストレーション）
         │
         ├─→ 単一エージェントで複数段階のスキル切り替え？
         │         │
         │         └─→ Skills（オンデマンド読み込み）
         │
         ├─→ 順序付きワークフロー、バトンタッチ？
         │         │
         │         └─→ Handoffs（状態駆動）
         │
         └─→ 複数データソースの照会と統合が必要？
                   │
                   └─→ Router（並列分散）

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

パターン	分散開発	並列化	マルチホップ会話	直接ユーザー対話	トークン消費
Subagents	高	高	高	低	高
Skills	高	中	高	高	低
Handoffs	なし	なし	高	高	中
Router	中	高	なし	中	高

この表の読み方：

• 分散開発：チームが別々のモジュールを開発しているか？そうであれば、SubagentsとSkillsが適している。各メンバーがサブエージェントまたはSkillを担当できる。
• 並列化：速度を追求しているか？RouterとSubagentsは複数のエージェントを並列で実行でき、効率が最も高い。
• マルチホップ会話：ユーザーが複数回の対話を必要とするか？HandoffsとSkillsは会話フローを自然にサポートする。
• 直接ユーザー対話：ユーザーが直接サブエージェントと話すか？SkillsとHandoffsはサポートするが、Routerはサポートしない。
• トークン消費：コストに敏感であれば、Skillsが最も節約でき、RouterとSubagentsが最も消費する。

私の経験則：シンプルに始める。まずSkillsまたはHandoffsでMVPを検証し、ボトルネックが見つかったらSubagentsまたはRouterにアップグレードする。最初から分散アーキテクチャを構築しようとしないこと。過剰設計の痛みはよく知っている。

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

デモから本番まで、太平洋ほどの距離がある。以下の落とし穴は全て経験した。

状態管理

マルチエージェントが状態を共有する場合、最も問題になりやすいのが「競合条件」だ。二つのエージェントが同時に同じ変数に書き込み、最終的にどちらがどちらを上書きするのか？

LangGraphの解決策はoutput_keyだ。各エージェントは専用のキーにのみ書き込める。

from langgraph.graph import StateGraph, MessagesState

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

# セキュリティエージェントは security_result のみに書き込む
async def security_agent(state: GraphState):
    result = await analyze_security(state["messages"])
    return {"security_result": result}  # このキーのみに書き込む

# スタイルエージェントは style_result のみに書き込む
async def style_agent(state: GraphState):
    result = await analyze_style(state["messages"])
    return {"style_result": result}

これにより、並列でも順次でも、各エージェントは自分の領域のみを操作し、互いに干渉しない。

もう一つの一般的な問題は「コンテキスト汚染」だ。エージェントAの出力がエージェントBに読まれるが、Bはその情報を全く必要としていない。私の解決策：状態にrelevant_keysフィールドを追加し、各エージェントは必要なキーのみを読み込む。

パフォーマンス最適化

マルチエージェントシステムのトークン消費は底なし沼だ。いくつかのトークン節約テクニック：

1. SubagentsはSkillsより67%トークンを節約（マルチドメインシナリオ）

LangChainのテストデータによると、あるタスクが3つの独立したドメインに関わる場合、Subagentsのトークン消費はSkillsの3分の1だ。なぜか？Subagentsはコンテキストが分離され、各サブエージェントは自分のドメインのコンテンツのみを見る。Skillsの場合、全てのSkillのコンテキストが蓄積され、どんどん大きくなる。

2. ステートフルモードで40-50%の重複呼び出しを節約

タスクに大量の重複クエリがある場合（例えば同じ質問を10回する）、ステートフルなHandoffsモードを使うと、エージェントは以前の回答を覚えていられる。LangChainのデータによると、ステートフルはステートレスより約半分のLLM呼び出しを節約できる。

3. リフレクションモードの反復回数を制限

多くの人はエージェントに「リフレクション」能力を加えるのが好きだ。出力を自己チェックし、問題を発見し、再生成する。これは良いが、無限ループに陥りやすい。私は通常max_iterations=2または3に制限し、それを超えたら強制終了する。

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回リフレクション
)

よくある落とし穴

無限ループ：エージェントが自分自身を呼び出し、自分が自分を呼び出し…終わらない。解決策：max_iterationsと明確な終了条件を設定する。

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

コンテキスト膨張：エージェントがだんだん「愚か」になり、出力が短くなる。これはコンテキストに太多のものが詰め込まれているためだ。解決策：Blackboardパターン（共有ボード）を使用し、必要なコンテキストのみを保持し、定期的にクリーンアップする。

調整税：エージェント数が増えると、通信オーバーヘッドが指数関数的に増加する。あるシステムをテストしたところ、3つのエージェントから10個に増やすと、レスポンス時間が2秒から15秒になった。解決策：責任が近いエージェントを統合し、エージェント数を5つ以内に制御する。

完全な実装例

理論は十分だ。実践的な例を見ていこう。Router + ParallelAgentパターンを使用したコードレビューマルチエージェントシステムを構築した。

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

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}

# セキュリティ監査エージェント
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": []}

# スタイルチェックエージェント
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": []}

# パフォーマンス分析エージェント
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つのエージェントが並列で実行され、約3-5秒で結果が出る。順次実行なら少なくとも10秒はかかるだろう。

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

マルチエージェント協調システムの構築

ゼロからコードレビューマルチエージェントシステムを構築する

1. 1

   ステップ1: アーキテクチャパターンを選択

   タスクの特性に応じて適切なアーキテクチャパターンを選択
2. 2

   ステップ2: 状態構造を定義

   TypedDictを使用してマルチエージェント共有状態を定義
3. 3

   ステップ3: エージェントノードを作成

   各エージェントに独立したノード関数を作成
4. 4

   ステップ4: 実行グラフを構築

   LangGraph StateGraphを使用して実行フローを構築
5. 5

   ステップ5: 状態管理を追加

   output_keyを使用して競合条件を回避

結論

ここまで説明してきたが、実は3つの文に要約できる。

基盤ロジック：パターンの選択がフレームワークの選択よりも重要だ。LangGraph、AutoGen、CrewAIは全て優れたツールだが、RouterパターンでHandoffsが必要な問題を解決しようとすれば、どんなに良いフレームワークでも救えない。

中層戦略：シンプルに始め、徐々にアップグレードする。まずSkillsまたはHandoffsでMVPを検証し、ボトルネックが見つかったらSubagentsまたはRouterを検討する。過剰設計は最大の落とし穴だ。私は踏んだ、あなたも踏まないで。

トップレベル実装：本番環境では状態管理、パフォーマンス、コストに注目する。トークン消費、無限ループ、コンテキスト汚染、この3つの問題を解決できれば、マルチエージェントシステムは安定して稼働できる。

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

6 min read · 公開日: 2026年3月25日 · 更新日: 2026年3月25日