コア痛点:モノリシックAgentがスケールで失敗する理由
- コンテキスト上限。 複雑タスクが中間状態でコンテキストを埋め尽くし、推論品質が急降下。ハンドオフエラーも静かに累積。
- 何でも屋の希薄化。 検索・コード生成・監査を1体で担うとどれも中途半端。ロール単位のアップグレードも全体書き換えが必要。
- 並行なしの直列レイテンシ。 全ステップの合計が総レイテンシ。独立サブタスクも明示的オーケストレーションなしでは並列化不可。
- 単一障害点と不可視エラー。 1回の不良モデル呼び出しでワークフロー停止。幻覚がハンドオフを連鎖しながら HTTP 200 でダッシュボードは緑のまま。
1. 単一Agentでは足りない理由
「モノリシックAgent」— 推論・ルーティング・実行を1つのLLMが担う構造はプロトタイプは容易だが、本番スケールでは脆い。問題はモデル固有ではなく構造的。
- コンテキスト上限 — 中間状態がウィンドウを埋め、推論品質が急降下。
- 何でも屋問題 — 検索・コード生成・監査を同時に担い、どれも十分にできない。
- 並行なし — 直列実行では総レイテンシが各ステップの合計。
- 単一障害点 — 1回の不良呼び出しでワークフロー全体が停止。
Google 内部 Agent Bake-Off(MLflow 2026 本番ガイド)では、分解型マルチAgentアーキテクチャが処理時間を1時間から10分へ短縮(6倍)、サブAgentを個別にアップグレード可能だった。
AdaptOrch(2026)は、オーケストレーション・トポロジ—Agent の構成と協調方法—が基盤モデル選択よりシステム性能への影響が大きいことを形式証明し、適切なトポロジでコーディング・推論・RAG ベンチで 12–23% 改善。
結論:本番向けならマルチAgentはほぼ正解。問いはどのパターンか。
2. マルチAgentシステムとは
マルチAgentシステム(MAS)は、定義された通信プロトコルとオーケストレーション機構を通じて協調し、単一Agentでは効率的に処理できないタスクを達成する独立AI Agent の集合。
| 特性 | 意味 |
|---|
| 単一責任 | 検索・推論・生成・検証など明確にスコープされた1ジョブ |
| ツール装備 | ロールに必要な特定ツールへのアクセス |
| 状態分離 | 他Agentを汚染しない独自コンテキストとメモリ |
| 交換可能 | より良いモデルが出れば独立アップグレード可能 |
3つの制御トポロジ
Centralized Decentralized Hierarchical
[Orchestrator] A ←→ B ←→ C [Top Orchestrator]
/ | \ ↕ ↕ / \
[A] [B] [C] D ←→ E ←→ F [Team Lead-1] [Team Lead-2]
/ \ / \
Pros: auditable, controllable Pros: resilient, fast [a] [b] [c] [d]
Cons: bottleneck at center Cons: hard to debug
Pros: balances both
3. 六大オーケストレーション設計パターン
本番の大多数をカバーする6パターン。いつ使うかがエージェント型AIエンジニアリングで最も重要なスキル。
パターン1:順次パイプライン(Sequential Pipeline)
考え方: Agent A の出力が Agent B の入力。厳密な線形実行。
[User Input] → [Retrieval Agent] → [Analysis Agent] → [Writer Agent] → [Review Agent] → [Output]
使うとき: ステップ間に厳密な依存、固定予測可能ワークフロー。用途:コンテンツ制作、コンプライアンス審査、文書処理。
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class PipelineState(TypedDict):
query: str
retrieved_docs: str
analysis: str
final_report: str
def retrieval_agent(state: PipelineState):
docs = search_knowledge_base(state["query"])
return {"retrieved_docs": docs}
def analysis_agent(state: PipelineState):
result = llm.invoke(f"Analyze the following: {state['retrieved_docs']}")
return {"analysis": result.content}
def writer_agent(state: PipelineState):
report = llm.invoke(f"Write a report based on: {state['analysis']}")
return {"final_report": report.content}
builder = StateGraph(PipelineState)
builder.add_node("retriever", retrieval_agent)
builder.add_node("analyzer", analysis_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "retriever")
builder.add_edge("retriever", "analyzer")
builder.add_edge("analyzer", "writer")
builder.add_edge("writer", END)
pipeline = builder.compile()
| 長所 | 短所 |
|---|
| 実装・デバッグが容易 | 総レイテンシ = 全ステップの合計 |
| 挙動が予測可能 | 1ステップ失敗で下流全体ブロック |
| 監査しやすい | 動的分岐に不向き |
パターン2:並列扇出/扇入(Parallel Fan-Out / Fan-In)
考え方: 独立サブAgentが並行実行。コレクターが集約。総レイテンシは max(T1,T2,...,Tn)。
┌──→ [Research Agent A] ──┐
[Supervisor] ───────├──→ [Research Agent B] ──┼──→ [Synthesizer] → [Output]
└──→ [Research Agent C] ──┘
使うとき: サブタスクが真に独立、レイテンシ削減が重要。多源リサーチ、並列リスク評価、競合分析。
from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator
class ResearchState(TypedDict):
query: str
research_results: Annotated[list, operator.add]
final_synthesis: str
def supervisor(state: ResearchState):
subtasks = [
{"query": state["query"], "source": "academic"},
{"query": state["query"], "source": "industry"},
{"query": state["query"], "source": "news"},
]
return [Send("research_worker", task) for task in subtasks]
def research_worker(state: dict):
result = search_by_source(state["query"], state["source"])
return {"research_results": [result]}
def synthesizer(state: ResearchState):
combined = "\n".join(state["research_results"])
synthesis = llm.invoke(f"Synthesize the following research: {combined}")
return {"final_synthesis": synthesis.content}
builder = StateGraph(ResearchState)
builder.add_node("research_worker", research_worker)
builder.add_node("synthesizer", synthesizer)
builder.add_conditional_edges(START, supervisor, ["research_worker"])
builder.add_edge("research_worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()
要点: LangGraph の Send API が真の並行サブグラフをディスパッチ。Annotated[list, operator.add] Reducer が結果を自動マージ。
パターン3:階層 Supervisor-Worker
考え方: Supervisor が意図認識・分解・ルーティング。専門 Worker が実行。Synthesizer が集約。
[User Request]
↓
[Supervisor Agent] ← Plans tasks and routes
/ | \
[Code Agent] [Search Agent] [Data Agent]
\ | /
[Synthesizer Agent]
↓
[Final Output]
2層ルーティング(キーワード高速経路 + LLM フォールバック):
KEYWORD_ROUTING = {
"code": "code_agent", "debug": "code_agent",
"search": "search_agent", "find": "search_agent",
"data": "data_agent", "analyze": "data_agent",
}
def supervisor_with_fast_path(state):
query = state["query"].lower()
for keyword, agent_name in KEYWORD_ROUTING.items():
if keyword in query:
return {"next": agent_name} # <1ms, no LLM call
decision = llm.invoke(f"Route this request: {state['query']}")
return {"next": decision.content.strip()}
パターン4:スワーム(Peer-to-Peer)
考え方: 中央コーディネータなしで Agent 間直接タスク受渡し。終了ルール(ラウンド数・合意・タイムアウト)で停止。多ラウンド交渉向きだが非決定論が高く、本番では階層型に収束しがち。本番では慎重に。
import autogen
reviewer_1 = autogen.AssistantAgent(
name="SecurityReviewer",
system_message="You are a security expert focused on code vulnerabilities."
)
reviewer_2 = autogen.AssistantAgent(
name="PerformanceReviewer",
system_message="You are a performance expert focused on efficiency."
)
human_proxy = autogen.UserProxyAgent(
name="CodeAuthor",
human_input_mode="NEVER",
max_consecutive_auto_reply=2,
is_termination_msg=lambda x: "APPROVED" in x.get("content", "")
)
groupchat = autogen.GroupChat(
agents=[human_proxy, reviewer_1, reviewer_2],
messages=[],
max_round=6 # Hard termination cap — always required
)
manager = autogen.GroupChatManager(groupchat=groupchat)
パターン5:ブラックボードアーキテクチャ
考え方: 構造化共有ワークスペース。前提条件を満たした Agent が自律的に読み書き。長時間非同期(時間〜日)、異チーム異種サービス、事前ルーティング不可の条件分岐ワークフロー向け。
┌─────────────────────────────────────┐
│ Blackboard (Shared State) │
│ task_status: "research_done" │
│ research_data: { ... } │
│ analysis_result: null │
└──────┬─────────────────────┬────────┘
↑ writes ↓ reads (when precondition met)
[Research Agent] [Analysis Agent]
パターン6:ハイブリッド
考え方: 複数パターンを1システムに組合せ。最も一般的なのは Supervisor + パイプライン—上位で階層ルーティング、各ブランチ内で順次実行。
[User Request] → [Intent Router]
├──→ [Simple query] → Direct answer
└──→ [Complex report] → [Supervisor]
/ \
[Parallel Research Fan-Out] [Quality Pipeline: Review → Human Approval → Publish]
↙ ↓ ↘
[A] [B] [C] → [Synthesizer]
4. フレームワーク対決:LangGraph vs CrewAI vs AutoGen
| 次元 | LangGraph | CrewAI | AutoGen(Microsoft) |
|---|
| アーキテクチャ | 状態機械グラフ | ロール制クルー | 会話型グループ |
| 言語 | Python / JS/TS | Python | Python / .NET |
| 学習曲線 | 急 | 緩やか | 中程度 |
| ネイティブ状態管理 | あり | 限定的 | 限定的 |
| Human-in-the-Loop | ネイティブ interrupt() | カスタム実装 | サポート |
| 可観測性 | LangSmith(商用) | 限定的 | Azure Monitor |
| 本番就绪度 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| プロトタイプ速度 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| Azure/Microsoft | ⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐⭐⭐ |
| 最適用途 | 複雑ステートフルWF | ロール制コンテンツ | 会話型マルチAgent |
- LangGraph: 規制本番、複雑状態管理・永続化、HITL チェックポイント、条件分岐と動的ルーティング。
- CrewAI: 1–2日プロトタイプ、「役職付きAgent」思考、低い状態管理複雑度。
- AutoGen: Microsoft/Azure スタック、会話による多ラウンド議論と反復改善。
信頼性・可観測性・Human oversight が必要なワークフローでは LangGraph が最も本番向き。決定論的グラフ実行、ネイティブ永続化、LangSmith トレースが規制業界のデフォルト。
5. 双协议層:MCP + A2A
2026年、マルチAgent通信は Linux Foundation Agentic AI Foundation 管轄の2つの補完プロトコルに標準化。
┌─────────────────────────────────────────────────────────┐
│ Multi-Agent System │
│ Agent-1 ←────── A2A Protocol ──────→ Agent-2 │
│ │ │ │
│ MCP Protocol MCP Protocol │
│ ↓ ↓ │
│ [Tools / DBs / APIs] [Tools / DBs / APIs] │
└─────────────────────────────────────────────────────────┘
MCP (vertical layer): Agent ↔ external tools and data
A2A (horizontal layer): Agent ↔ Agent
TCP と HTTP のような異なる層。MCP が「手」、A2A が「同僚間の会話」。
MCP(Model Context Protocol)
Anthropic 発、Linux Foundation 管轄。Agent が外部ツール・DB・API にアクセスする方法を標準化— 一度書けば MCP 互換 Agent が利用可能。
from mcp.server import Server
from mcp.types import Tool, TextContent
app = Server("customer-data-mcp")
@app.list_tools()
async def list_tools():
return [Tool(name="query_customer_db", description="Query by id, name, or email", ...)]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "query_customer_db":
result = db.query(arguments["field"], arguments["value"])
return [TextContent(type="text", text=str(result))]
A2A(Agent-to-Agent Protocol)
Google が2025年4月ローンチ、2026年初 v1.0、Atlassian・Salesforce・SAP 等50+パートナー。JSON-RPC 2.0 over HTTP でタスク委譲と能力発見を標準化。各 Agent は /.well-known/agent.json に Agent Card を公開。
// /.well-known/agent.json
{
"name": "ResearchAgent",
"version": "1.0",
"description": "Professional web research and summarization agent",
"url": "https://research-agent.internal/a2a",
"capabilities": { "streaming": true, "async": true },
"skills": [
{
"id": "web_research",
"name": "Web Research",
"description": "Search and summarize latest information from the web",
"tags": ["research", "summarization", "web"]
}
]
}
import httpx
async def discover_and_delegate(agent_url: str, task: str):
card = (await httpx.get(f"{agent_url}/.well-known/agent.json")).json()
skills = [s["id"] for s in card["skills"]]
if "web_research" not in skills:
raise ValueError(f"Agent {card['name']} does not support web_research")
payload = {"jsonrpc": "2.0", "method": "message/send", "id": "task-001",
"params": {"message": {"role": "user", "parts": [{"type": "text", "text": task}]}}}
return (await httpx.post(card["url"], json=payload)).json()
6. 本番エンジニアリングの要点
6.1 状態永続化とリカバリ
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/agentdb") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-12345"}}
result = graph.invoke({"query": "Analyze Q2 financial report"}, config)
6.2 Human-in-the-Loop チェックポイント
from langgraph.types import interrupt
def high_risk_action_agent(state):
proposed_action = plan_action(state)
human_decision = interrupt({
"proposed_action": proposed_action,
"risk_level": "HIGH",
"message": "This action will modify the production database. Confirm to proceed."
})
if human_decision["approved"]:
return execute_action(proposed_action)
return {"status": "cancelled", "reason": human_decision.get("reason")}
6.3 サーキットブレーカーパターン
@CircuitBreaker(failure_threshold=3, recovery_timeout=30)
async def call_external_agent(task):
return await agent_client.send(task)
6.4 Token 予算管理
暴走 Token 支出は本番で最も多い驚きの一つ。Agent 別予算・ハード上限・TokenBudgetManager による BudgetExceededException を初日から計装。
class TokenBudgetManager:
def __init__(self, total_budget: int = 100_000):
self.total_budget = total_budget
self.used_tokens = 0
self.agent_usage = {}
def check_budget(self, agent_name: str, estimated_tokens: int) -> bool:
remaining = self.total_budget - self.used_tokens
if estimated_tokens > remaining:
raise BudgetExceededException(
f"Agent {agent_name} requested {estimated_tokens} tokens, "
f"but only {remaining} tokens remain"
)
return True
def record_usage(self, agent_name: str, actual_tokens: int):
self.used_tokens += actual_tokens
self.agent_usage[agent_name] = self.agent_usage.get(agent_name, 0) + actual_tokens
7. 可観測性:ブラックボックスを開く
MAST 研究チームの1,642トレース分析:57% の組織が本番 Agent を稼働、可観測性実装完了はわずか8%。幻覚連鎖、リトライループ、HTTP 200 の緑ダッシュボード。
| カテゴリ | 割合 | 何が起きるか |
|---|
| システム設計失敗 | 41.77% | ステップ反復、誤ツール選択、コンテキスト溢れ、終了条件欠如 |
| Agent間不整合 | 36.94% | ハンドオフでコンテキスト喪失;幻覚が次Agentの「事実」に |
| タスク検証失敗 | 21.30% | 早期終了、不完全検証、完了に見える未完了 |
OpenTelemetry 相関 ID で Agent 境界を横断トレース。
from opentelemetry import trace
import uuid
tracer = trace.get_tracer("multi-agent-system")
def traced_agent_call(agent_name: str, task: dict, correlation_id: str = None):
if not correlation_id:
correlation_id = str(uuid.uuid4())
with tracer.start_as_current_span(f"agent.{agent_name}") as span:
span.set_attribute("agent.name", agent_name)
span.set_attribute("correlation.id", correlation_id)
try:
result = agent_registry[agent_name].run(task)
span.set_attribute("tokens_used", result.get("tokens", 0))
span.set_attribute("agent.status", "success")
return result
except Exception as e:
span.set_attribute("agent.status", "error")
span.set_attribute("error.message", str(e))
raise
MONITORING_METRICS = {
"task_success_rate": "End-to-end task completion (target: >85%)",
"e2e_latency_p95": "P95 end-to-end latency (target: <30s)",
"total_cost_per_task": "Average token cost per task",
"agent_error_rate": "Per-agent error rate (target: <5%)",
"agent_retry_count": "Retry count (high = investigate)",
"output_quality_score": "Output quality score",
"hallucination_rate": "Hallucination detection rate",
}
追跡すべきコア指標: task_success_rate(>85%)、e2e_latency_p95(<30s)、cost_per_task、Agent別 error_rate(>5%でアラーム)、retry_count、LLM-as-Judge 品質スコア。
8. よくある落とし穴と回避策
落とし穴1:コンテキスト汚染(幻覚連鎖)
Agent A が幻覚「事実」を生成→B/C に伝播→最終出力が誤前提の上に構築、HTTP は200。対策: 各ハンドオフで JSON Schema、信頼度閾値(<0.7 拒否)、必須フィールド検証。
def validate_agent_output(output: dict, schema: dict) -> bool:
jsonschema.validate(output, schema)
if output.get("confidence_score", 1.0) < 0.7:
raise LowConfidenceError(f"Agent output confidence too low: {output['confidence_score']}")
required_fields = schema.get("required", [])
missing = [f for f in required_fields if not output.get(f)]
if missing:
raise MissingFieldsError(f"Missing required fields: {missing}")
return True
落とし穴2:暴走ループとコスト爆発
リトライループやツール呼び出しスパイラルで1タスクが $0.02 から $47 へ。対策: MAX_ITERATIONS、MAX_TOOL_CALLS_PER_AGENT、MAX_TOTAL_TOKENS_PER_REQUEST、LangGraph interrupt_before。
MAX_ITERATIONS = 10
MAX_TOOL_CALLS_PER_AGENT = 20
MAX_TOTAL_TOKENS_PER_REQUEST = 50_000
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["high_cost_tool"]
)
落とし穴3:過剰エンジニアリング
単純2ステップLLMチェーンを8 Agent に分解。ルール: 順次パイプラインから開始。測定可能な根拠があるときだけ Agent 追加。本番の最適数は3–8 Agent。
落とし穴4:デモ→本番ギャップ
内部デモは成功、ローンチ2週間後にエッジ入力で連鎖失敗。対策: 初日から ProductionGuardrails—入力長制限、プロンプト注入検知、PII マスキング、有害コンテンツ分類。
class ProductionGuardrails:
def __init__(self):
self.input_validators = []
self.output_validators = []
def validate_input(self, user_input: str) -> str:
if len(user_input) > 10000:
raise InputTooLongError("Input exceeds 10000 character limit")
injection_patterns = ["ignore previous instructions", "forget everything"]
for pattern in injection_patterns:
if pattern.lower() in user_input.lower():
raise PromptInjectionError("Potential prompt injection detected")
return user_input.strip()
def validate_output(self, output: str) -> str:
output = self.pii_filter.redact(output)
if self.content_classifier.is_harmful(output):
raise HarmfulContentError("Output contains harmful content")
return output
落とし穴5:並列ブランチ同期問題(LangGraph)
現象: Send API で並列ブランチをディスパッチ。実行時間差があり Supervisor が遅いブランチ完了前に再実行→重複実行と不完全結果。
対策 — 遅延実行:
# The defer=True parameter creates an explicit synchronization barrier.
# The supervisor node won't execute until ALL parallel branches have completed.
builder.add_node("supervisor", supervisor_node, defer=True)
9. 意思決定フレームワーク
Does your task have strict sequential dependencies between steps?
├─ YES → Can any of those steps run in parallel?
│ ├─ NO → [Sequential Pipeline]
│ └─ YES → [Hybrid: Sequential Pipeline + Parallel Fan-Out]
│
└─ NO → Does one agent have clear decision-making authority?
├─ YES → Does scale require sub-teams?
│ ├─ NO → [Supervisor-Worker Hierarchical]
│ └─ YES → [Hierarchical (Supervisors of Supervisors)]
│
└─ NO → Is the task long-running and async (hours to days)?
├─ YES → [Blackboard Architecture]
└─ NO → Agent count ≤ 5 and termination is well-defined?
├─ YES → [Swarm — with hard round/time limits]
└─ NO → [Refactor into Hierarchical instead]
10. まとめと今後の展望
要点
- トポロジ > モデル選択。 AdaptOrch:Agent の構成方法が基盤モデルより重要。
- シンプルに始め、必要時に Agent 追加。 初回は順次パイプライン。本番最適は3–8 Agent。
- MCP + A2A は新興標準。 Linux Foundation 管轄、広範な業界支持。
- 可観測性は必須。 本番57% vs 可観測性8% の49ptギャップが $47K 請求の温床。
- 各ハンドオフをバージョン付きAPIとして。 スキーマ検証と信頼度閾値で連鎖失敗を防止。
2026年注目トレンド
- 連邦オーケストレーション: 各チームが独立サブオーケストレータを維持し学習ルーティングを共有
- マルチモーダルマルチAgent: 視覚・音声 Agent とテキスト Agent の協調が成熟
- 適応トポロジ選択: タスク特性に応じ最適パターンを自動選択(AdaptOrch 方向)
- EU AI Act コンプライアンス: 完全な決定監査トレイル—Agent レベル追跡可能性が必須
本番五步Runbook
ステップ1 — トポロジとフレームワーク選定
セクション9の意思決定ツリーに従う。順次パイプラインから開始。レイテンシ・コンテキスト溢れ・ロール別アップグレード需要の測定根拠があるときだけ扇出や Supervisor-Worker を追加。規制本番は LangGraph、1–2日プロトタイプは CrewAI。
ステップ2 — MCP ツールと A2A 委譲
各 Agent のツールを MCP Server で公開。/.well-known/agent.json に Agent Card。オーケストレータは JSON-RPC 2.0 message/send で委譲。
ステップ3 — 永続化とガードレール
PostgresSaver チェックポイント、TokenBudgetManager 上限、外部 Agent 呼び出しのサーキットブレーカー、高リスクDB書き込み前の interrupt()。
ステップ4 — 可観測性計装
OpenTelemetry と Agent 境界を横断する相関 ID。task_success_rate、e2e_latency_p95、Agent別エラー率。LLM-as-Judge で品質と幻覚検知。
ステップ5 — launchd で Mac クラウドにホスト
Cursor / Claude Desktop STDIO 向けに VPSMAC Mac クラウドで launchd KeepAlive、リソース制限、PostgreSQL チェックポイントで 7×24 稼働。
引用可能なハードファクト(2026)
- トポロジ > モデル: AdaptOrch(arXiv 2602.16873)— オーケストレーション・トポロジが SWE-bench・RAG で 12–23% 改善、モデル差換え単独より大。
- 6倍スループット: Google Agent Bake-Off(MLflow 2026)— 分解型マルチAgentで1時間→10分。
- 可観測性ギャップ: MAST 1,642トレース— 本番57%、可観測性完了8%;41.77% が設計問題。
- プロトコル標準: MCP と A2A は Linux Foundation AAIF 管轄;A2A v1.0 は50+エンタープライズパートナー。
まとめ
マルチAgentアーキテクチャはもはや実験段階ではなく、2026年本番エージェント型システムのデフォルトパターン。六大パターン、MCP+A2A、可観測性実践がプロトタイプから本番への完全ブループリント。
参考リソース:
LangGraph オーケストレータをノートPCや汎用 Linux VPS で動かす検証は可能だが、スリープ切断、macOS STDIO Host 非互換、Docker 抽象層で 7×24 Agent WF は脆い。PostgreSQL チェックポイントと OpenTelemetry もプロセス再起動に耐える永続基盤が必要。Cursor、Claude Desktop、MCP Server をオーケストレーションと24時間同居させるチームには、VPSMAC Mac クラウドノードのレンタルがより安定した Apple ツールチェーン向けの選択—ネイティブ macOS、launchd KeepAlive、ベアメタル性能、デモ→本番ギャップなし。