「Supervisorパターンは分かった。でも、もっと大規模なシステムを設計しようとすると、どのパターンを選べばいいか途端に分からなくなる」
マルチエージェントの入門を終えた開発者から、実際によく聞く声です。シングルエージェントとシンプルなSupervisorは動いた。でも、本番レベルのシステムを設計するには、パターンの引き出しが圧倒的に少ない。
実際に10社以上のAIエージェント構築を支援してきた経験から言えることがあります。マルチエージェントの複雑さは、パターンを知らないことから生まれます。逆に8つのパターンを体系的に理解すれば、設計のほとんどは「どのパターンの組み合わせか」に帰着します。
この記事では、マルチエージェントオーケストレーションの8パターンを、LangGraph・CrewAI・Google ADK・Claude Agent SDKの4フレームワーク対応のコード付きで完全解説します。既存記事 マルチエージェントAI設計パターン4選 の深掘り完全版として、より実践的な内容をお届けします。
まず押さえるべき：クローズド型 vs オープン型オーケストレーション
全パターンを理解する前に、この根本的な分類を押さえてください。
| 区分 | 制御の主体 | 特徴 | 適した場面 | リスク |
|---|---|---|---|---|
| クローズド型 | コード（開発者） | 決定的、再現性が高い | バッチ処理、データパイプライン | 柔軟性が低い |
| オープン型 | LLM（AI） | 動的、状況適応的 | CS対応、複雑な問い合わせ | 予測不能な動作 |
クローズド型はコードで実行順序を固定します。SequentialAgent（ADK）やDAGグラフ（LangGraph）がこれです。対してオープン型はLLMが次の行動を判断します。ReAct型エージェントやSupervisorパターンがこれに当たります。
実用的なシステムでは、どちらか一方に絞るより「決定的な処理はクローズド、判断が必要な処理はオープン」と組み合わせるのが鉄則です。
8パターン早見表
| # | パターン名 | 型 | 主なフレームワーク | 難易度 |
|---|---|---|---|---|
| 1 | Sequential Pipeline | クローズド | 全4SDK | ★☆☆☆☆ |
| 2 | Parallel Fan-Out/Gather | クローズド | ADK, LangGraph | ★★☆☆☆ |
| 3 | Supervisor（LLM駆動） | オープン | 全4SDK | ★★★☆☆ |
| 4 | Swarm（対等協調） | オープン | CrewAI, OpenAI Swarm | ★★★☆☆ |
| 5 | DAG（有向非巡回グラフ） | クローズド | LangGraph, ADK 2.0 | ★★★★☆ |
| 6 | Review/Critique（評価ループ） | 混合 | ADK, LangGraph | ★★★☆☆ |
| 7 | Loop with Early Exit | クローズド | ADK, LangGraph | ★★☆☆☆ |
| 8 | Human-in-the-Loop | 混合 | LangGraph, ADK 2.0 | ★★★★☆ |
パターン1：Sequential Pipeline — すべての起点
エージェントAの出力をエージェントBに渡し、BからCへと直列に処理するパターン。最もシンプルで、マルチエージェントの出発点です。
LangGraphでの実装
# 動作環境: Python 3.11+, langgraph>=0.3.0, langchain-openai>=0.3.0
# pip install langgraph langchain-openai
import os
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
llm = ChatOpenAI(model="gpt-4o")
def step1_analyzer(state: MessagesState):
"""ステップ1: 入力を分析"""
response = llm.invoke([
{"role": "system", "content": "入力データを分析し、キーポイントを3つ抽出してください。"},
*state["messages"]
])
return {"messages": [response]}
def step2_processor(state: MessagesState):
"""ステップ2: 分析結果を処理"""
response = llm.invoke([
{"role": "system", "content": "前の分析結果をもとに、アクションプランを作成してください。"},
*state["messages"]
])
return {"messages": [response]}
def step3_formatter(state: MessagesState):
"""ステップ3: 出力を整形"""
response = llm.invoke([
{"role": "system", "content": "アクションプランをMarkdown形式で整形してください。"},
*state["messages"]
])
return {"messages": [response]}
# グラフ定義
graph = StateGraph(MessagesState)
graph.add_node("analyzer", step1_analyzer)
graph.add_node("processor", step2_processor)
graph.add_node("formatter", step3_formatter)
# 直列エッジ
graph.add_edge(START, "analyzer")
graph.add_edge("analyzer", "processor")
graph.add_edge("processor", "formatter")
graph.add_edge("formatter", END)
app = graph.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
Google ADKでの実装（1行で同等）
# 動作環境: Python 3.10+, google-adk>=1.0.0
from google.adk.agents import SequentialAgent, LlmAgent
pipeline = SequentialAgent(
name="Pipeline",
sub_agents=[
LlmAgent(name="Analyzer", model="gemini-3-flash-preview",
instruction="入力を分析しキーポイントを3つ抽出", output_key="analysis"),
LlmAgent(name="Processor", model="gemini-3-flash-preview",
instruction="{analysis} をもとにアクションプランを作成", output_key="plan"),
LlmAgent(name="Formatter", model="gemini-3-flash-preview",
instruction="{plan} をMarkdown形式で整形"),
]
)
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン2：Parallel Fan-Out/Gather — 並列処理の基本
独立した複数のタスクを同時実行し、最後に統合するパターン。複数のデータソース取得、複数の言語への同時翻訳、複数の視点からの分析などに使います。
# 動作環境: Python 3.10+, google-adk>=1.0.0
from google.adk.agents import SequentialAgent, ParallelAgent, LlmAgent
MODEL = "gemini-3-flash-preview"
# Fan-Out: 3つのリサーチャーを並列実行
parallel_research = ParallelAgent(
name="ParallelResearch",
sub_agents=[
LlmAgent(name="JapanAnalyst", model=MODEL,
instruction="AIエージェントの日本市場の動向を調査",
output_key="japan_data"),
LlmAgent(name="USAnalyst", model=MODEL,
instruction="AIエージェントの米国市場の動向を調査",
output_key="us_data"),
LlmAgent(name="EUAnalyst", model=MODEL,
instruction="AIエージェントの欧州市場の動向を調査",
output_key="eu_data"),
]
)
# Gather: 結果を統合
gatherer = LlmAgent(
name="Gatherer",
model=MODEL,
instruction="""
3つの地域調査を統合してグローバルレポートを作成:
日本: {japan_data}
米国: {us_data}
欧州: {eu_data}
"""
)
# Fan-Out → Gather のワークフロー
workflow = SequentialAgent(
name="GlobalResearch",
sub_agents=[parallel_research, gatherer]
)
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン3：Supervisor — 動的ルーティングの主役
スーパーバイザーエージェントがLLMの判断で適切な専門エージェントに委任するパターン。カスタマーサポート、マルチスキルアシスタントなど、入力の種類が予測しにくいシナリオに最適です。
LangGraphでの実装（langgraph-supervisorライブラリ）
# 動作環境: Python 3.11+, langgraph>=0.3.0, langgraph-supervisor>=0.1.0
# pip install langgraph langgraph-supervisor langchain-openai
from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
model = ChatOpenAI(model="gpt-4o")
# 専門エージェントを定義
def search_web(query: str) -> str:
"""ウェブを検索するツール（実際の実装では検索APIを使用）"""
return f"検索結果: {query} に関する情報..."
def calculate(expression: str) -> float:
"""数式を計算するツール"""
return eval(expression) # 実際の実装では安全な評価器を使用
research_agent = create_react_agent(
model=model,
tools=[search_web],
name="research_expert",
prompt="あなたはリサーチの専門家です。情報収集と分析を担当してください。"
)
math_agent = create_react_agent(
model=model,
tools=[calculate],
name="math_expert",
prompt="あなたは数学の専門家です。計算と数値分析を担当してください。"
)
# Supervisorの作成
workflow = create_supervisor(
[research_agent, math_agent],
model=model,
prompt="ユーザーの要求を分析し、適切な専門エージェントに委任してください。"
)
app = workflow.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
Claude Agent SDKでの実装
# 動作環境: Python 3.10+, claude-agent-sdk>=0.1.0
# pip install claude-agent-sdk
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def run_supervisor_agent(user_input: str):
"""
Claude Agent SDKでSupervisorパターンを実装。
サブエージェントを並列に起動してタスクを委任。
"""
# オーケストレーターとして動作
async for message in query(
prompt=f"""
以下のタスクを分析し、適切なサブタスクに分解して処理してください:
{user_input}
必要に応じて複数の観点から並列に調査してください。
""",
options=ClaudeAgentOptions(
system="あなたはマルチエージェントオーケストレーターです。タスクを分解し、並列実行してください。",
)
):
if hasattr(message, 'content') and message.content:
print(message.content)
asyncio.run(run_supervisor_agent("AI市場の現状と将来予測をまとめてください"))
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン4：Swarm — 対等なエージェント協調
エージェント同士が上下関係なく対等に協調するパターン。CrewAIが得意とする形式で、各エージェントが役割（Role）を持ち、タスクを分担します。創造的なタスク、ディベート、多角的な分析に向いています。
# 動作環境: Python 3.10+, crewai>=0.80.0
# pip install crewai crewai-tools
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
search_tool = SerperDevTool()
# 対等な専門エージェント群
researcher = Agent(
role="シニアリサーチャー",
goal="最新のAIエージェント技術を徹底調査する",
backstory="10年以上の技術調査経験を持つ専門家。批判的思考で情報を精査する。",
tools=[search_tool],
verbose=True
)
analyst = Agent(
role="ビジネスアナリスト",
goal="技術トレンドをビジネス価値に転換する分析を行う",
backstory="McKinsey出身のアナリスト。データドリブンな意思決定を支援する。",
verbose=True
)
writer = Agent(
role="テクニカルライター",
goal="複雑な技術概念を分かりやすくドキュメント化する",
backstory="エンジニアとライターの両方の経験を持つ。読者目線で文章を構成する。",
verbose=True
)
# 各エージェントのタスク定義
research_task = Task(
description="2026年のAIエージェント市場の最新動向を調査し、主要フレームワークの比較レポートを作成",
agent=researcher,
expected_output="主要フレームワーク5つの詳細比較（機能、料金、ユースケース）"
)
analysis_task = Task(
description="リサーチ結果をもとに、日本企業がAIエージェントを導入すべき3つの優先ユースケースを特定",
agent=analyst,
expected_output="ROI分析付きの優先ユースケース3選"
)
writing_task = Task(
description="リサーチと分析の結果を、開発者向けの実践ガイドとして執筆",
agent=writer,
expected_output="コード例付きの実践ガイド（2,000字以上）"
)
# Swarm型のCrew（hierarchical processで自律的に協調）
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.hierarchical, # または Process.sequential
verbose=True
)
result = crew.kickoff()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン5：DAG（有向非巡回グラフ）— 複雑な依存関係の管理
タスク間の依存関係をグラフで定義するパターン。「AとBが完了したらCを実行、AとCが完了したらDを実行」のような複雑な依存関係を管理できます。LangGraphが特に得意とする領域です。
# 動作環境: Python 3.11+, langgraph>=0.3.0
# pip install langgraph langchain-openai
from typing import TypedDict, Annotated
import operator
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
llm = ChatOpenAI(model="gpt-4o")
class DAGState(TypedDict):
input: str
data_a: str
data_b: str
data_c: str
final_result: str
def fetch_data_a(state: DAGState) -> dict:
"""タスクA: データソースAから取得（他と並列で実行可能）"""
result = llm.invoke(f"データソースAの観点で分析: {state['input']}")
return {"data_a": result.content}
def fetch_data_b(state: DAGState) -> dict:
"""タスクB: データソースBから取得（Aと並列で実行可能）"""
result = llm.invoke(f"データソースBの観点で分析: {state['input']}")
return {"data_b": result.content}
def process_c(state: DAGState) -> dict:
"""タスクC: AとBの結果に依存（A, B完了後に実行）"""
result = llm.invoke(f"AとBを統合: A={state['data_a']}, B={state['data_b']}")
return {"data_c": result.content}
def finalize(state: DAGState) -> dict:
"""タスクD: Cの結果に依存（C完了後に実行）"""
result = llm.invoke(f"最終レポート作成: {state['data_c']}")
return {"final_result": result.content}
# DAGグラフの定義
graph = StateGraph(DAGState)
graph.add_node("fetch_a", fetch_data_a)
graph.add_node("fetch_b", fetch_data_b)
graph.add_node("process_c", process_c)
graph.add_node("finalize", finalize)
# 依存関係の定義（AとBは並列、CはA+B後、DはC後）
graph.add_edge(START, "fetch_a")
graph.add_edge(START, "fetch_b") # AとBは並列スタート
graph.add_edge("fetch_a", "process_c")
graph.add_edge("fetch_b", "process_c") # C はA+B両方完了待ち
graph.add_edge("process_c", "finalize")
graph.add_edge("finalize", END)
app = graph.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン6：Review/Critique — 品質保証ループ
生成エージェントと評価エージェントをペアで使い、品質基準を満たすまで改善するパターン。コード生成→レビュー→修正や、ドキュメント→品質チェック→加筆に使います。
# 動作環境: Python 3.10+, google-adk>=1.0.0
from google.adk.agents import LoopAgent, SequentialAgent, LlmAgent
MODEL = "gemini-3-flash-preview"
# 生成エージェント
generator = LlmAgent(
name="Generator",
model=MODEL,
instruction="""
以下のフィードバックを踏まえて、AIエージェントの解説記事のドラフトを書いてください。
フィードバック: {feedback}
前のドラフト: {draft}
初回はフィードバックなしで書いてください。
""",
output_key="draft"
)
# 評価エージェント
critic = LlmAgent(
name="Critic",
model=MODEL,
instruction="""
以下のドラフトを厳密に評価してください:
{draft}
評価基準:
1. 技術的正確性（コード例が正しいか）
2. 読みやすさ（専門用語の説明があるか）
3. 完全性（重要なユースケースが網羅されているか）
全基準をクリアした場合のみ "APPROVED" と記載してください。
それ以外は改善点を具体的に指摘してください（1-3点）。
""",
output_key="feedback"
)
# 最大3回のReview/Critiqueループ
review_loop = LoopAgent(
name="ReviewLoop",
sub_agents=[generator, critic],
max_iterations=3
)
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン7：Loop with Early Exit — 効率的な反復
条件が満たされたら早期終了する反復パターン。最大回数に達する前に目標を達成できた場合にコストを節約できます。エラーが解消されるまで自動修正するデバッグループなどに使います。
# 動作環境: Python 3.11+, langgraph>=0.3.0
from typing import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
llm = ChatOpenAI(model="gpt-4o")
class LoopState(TypedDict):
code: str
errors: list
iteration: int
max_iterations: int
def generate_or_fix_code(state: LoopState) -> dict:
"""コード生成または修正"""
if state["iteration"] == 0:
result = llm.invoke("シンプルなFibonacci関数を書いてください")
return {"code": result.content, "iteration": 1}
else:
result = llm.invoke(
f"このコードのエラーを修正してください:nコード: {state['code']}nエラー: {state['errors']}"
)
return {"code": result.content, "iteration": state["iteration"] + 1}
def check_errors(state: LoopState) -> dict:
"""コードのエラーチェック（簡易版）"""
try:
exec(state["code"])
return {"errors": []} # エラーなし
except Exception as e:
return {"errors": [str(e)]}
def should_continue(state: LoopState) -> str:
"""継続判定: エラーがないか、最大回数に達したら終了"""
if not state["errors"]:
return "end" # 早期終了: エラーなし
if state["iteration"] >= state["max_iterations"]:
return "end" # 最大回数到達
return "fix" # 継続: エラーあり
graph = StateGraph(LoopState)
graph.add_node("generate", generate_or_fix_code)
graph.add_node("check", check_errors)
graph.add_edge(START, "generate")
graph.add_edge("generate", "check")
graph.add_conditional_edges("check", should_continue, {"fix": "generate", "end": END})
app = graph.compile()
result = app.invoke({"code": "", "errors": [], "iteration": 0, "max_iterations": 5})
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン8：Human-in-the-Loop — 人間承認ゲートウェイ
AIの処理の途中に人間の確認・承認ステップを組み込むパターン。金額が大きい処理、法的リスクのある判断、ブランドイメージに関わる出力など、AIだけに任せられない処理に使います。
# 動作環境: Python 3.11+, langgraph>=0.3.0
# pip install langgraph langchain-openai
from typing import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
llm = ChatOpenAI(model="gpt-4o")
class HITLState(TypedDict):
request: str
ai_proposal: str
human_approved: bool
final_action: str
def ai_propose(state: HITLState) -> dict:
"""AIが提案を生成"""
result = llm.invoke(f"以下のリクエストへの対応を提案してください: {state['request']}")
return {"ai_proposal": result.content}
def human_review(state: HITLState) -> dict:
"""人間によるレビュー（interrupt で処理を一時停止）"""
# langgraph の interrupt_before/after でここで処理を中断
# 外部から approve/reject を受け取る
print(f"n[人間レビュー必要]n提案: {state['ai_proposal']}n")
approved = input("承認しますか？ (y/n): ").strip().lower() == "y"
return {"human_approved": approved}
def execute_or_reject(state: HITLState) -> str:
"""承認/却下の分岐"""
return "execute" if state["human_approved"] else "reject"
def execute_action(state: HITLState) -> dict:
result = llm.invoke(f"承認された提案を実行: {state['ai_proposal']}")
return {"final_action": result.content}
def reject_action(state: HITLState) -> dict:
return {"final_action": "却下されました。別の対応を検討します。"}
# チェックポイント付きグラフ（中断・再開をサポート）
checkpointer = MemorySaver()
graph = StateGraph(HITLState)
graph.add_node("propose", ai_propose)
graph.add_node("review", human_review)
graph.add_node("execute", execute_action)
graph.add_node("reject", reject_action)
graph.add_edge(START, "propose")
graph.add_edge("propose", "review")
graph.add_conditional_edges("review", execute_or_reject,
{"execute": "execute", "reject": "reject"})
graph.add_edge("execute", END)
graph.add_edge("reject", END)
app = graph.compile(checkpointer=checkpointer)
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
パターン選定フローチャート
どのパターンを選べばよいか分からない場合は、以下の質問で絞り込んでください。
| 質問 | Yes → このパターン | No → 次の質問へ |
|---|---|---|
| 人間の承認が必要か？ | パターン8: Human-in-the-Loop | ↓ |
| エラーが解消されるまで繰り返すか？ | パターン7: Loop with Early Exit | ↓ |
| 品質改善のフィードバックループが必要か？ | パターン6: Review/Critique | ↓ |
| タスク間に複雑な依存関係があるか？ | パターン5: DAG | ↓ |
| エージェント同士が対等に協調するか？ | パターン4: Swarm | ↓ |
| 入力の種類に応じて動的にルーティングするか？ | パターン3: Supervisor | ↓ |
| 独立したタスクを並列実行したいか？ | パターン2: Parallel Fan-Out | ↓ |
| シンプルな直列処理か？ | パターン1: Sequential Pipeline | 要件を再確認 |
【要注意】マルチエージェント設計でよくある失敗パターン
失敗1：全てをSupervisorで解決しようとする
❌ 全てのワークフローをLLM駆動のSupervisorで実装し、ルーティングが頻繁に失敗する
⭕ 決定的な処理（Sequential/Parallel）とLLM駆動（Supervisor）を役割に応じて使い分ける
なぜ重要か: Supervisorはルーティングに毎回LLM推論を使うため、コストが高く予測が難しい。固定フローはクローズド型に任せる。
失敗2：エージェント間のステート共有を設計しない
❌ 各エージェントが独立して動き、前のエージェントの出力が次に届かない
⭕ output_key（ADK）または state dict（LangGraph）で明示的に受け渡す設計をする
なぜ重要か: エージェント間通信の設計が甘いと、パイプラインの途中でコンテキストが消える。
失敗3：ループに終了条件を設けない
❌ LoopAgentやLangGraphのループにmax_iterationsを設定せず、無限ループが発生
⭕ 必ず上限回数と早期終了条件の両方を設定する
なぜ重要か: LLMが期待する終了シグナルを返さないケースは想定より多い。コスト爆発を防ぐために上限は必須。
失敗4：フレームワークの乗り換えコストを過小評価する
❌ LangGraphで構築後にCrewAIに全面移行しようとして、3週間の作業が発生
⭕ 最初にフレームワークを1つ選んで、まずPoC。移行コストは数週間〜数ヶ月になる
なぜ重要か: 各フレームワークはAPI、ステート管理、デプロイが大きく異なる。最初の選択が重要。
フレームワーク別の強み（2026年3月時点）
| フレームワーク | 最大の強み | 向いているシナリオ | 料金 |
|---|---|---|---|
| LangGraph | ステートフルワークフロー、チェックポイント | 複雑なDAG、Human-in-the-Loop | OSS無料（LangSmith有料） |
| CrewAI | 役割ベースのSwarm、MCP/A2A対応 | 役割分担型マルチエージェント | OSS無料（Enterprise有料） |
| Google ADK | Geminiとの統合、Vertex AIデプロイ | Google Cloud環境、マルチモーダル | OSS無料（Vertex AI従量課金） |
| Claude Agent SDK | 並列サブエージェント、独立コンテキスト | Claude Code自動化、大規模並列タスク | v0.1.48（2026-03-27時点） |
料金情報の最終確認: 2026-03-27
参考・出典
- Multi-agent systems – Agent Development Kit (ADK) — Google（参照日: 2026-03-27）
- langgraph-supervisor — LangChain reference — LangChain（参照日: 2026-03-27）
- Introduction – CrewAI — CrewAI Docs（参照日: 2026-03-27）
- Agent SDK overview – Claude API Docs — Anthropic（参照日: 2026-03-27）
- AutoGen vs CrewAI vs LangGraph vs PydanticAI vs Google ADK — Victor Dibia（参照日: 2026-03-27）
- Developer’s guide to multi-agent patterns in ADK — Google Developers Blog（参照日: 2026-03-27）
まとめ：今日から始める3つのアクション
- 今日やること: Sequential PipelineとParallel Fan-Outの2パターンをコピペで実装し、自分のユースケースに当てはまるか確認する
- 今週中: パターン選定フローチャートを使って、自社の主要ユースケースに最適なパターンを1つ選び、プロトタイプを作成する
- 今月中: 選んだフレームワークで本番用のマルチエージェントシステムをPoC展開し、Human-in-the-Loopゲートウェイを追加して安全性を担保する
あわせて読みたい:
- Google ADKでAIエージェントを作る入門ガイド — ADKの基礎から始めたい方はこちら
- AIエージェント開発フレームワーク7選｜2026年版完全比較ガイド — フレームワーク選定の全体像
著者: 佐藤傑（さとう・すぐる）
株式会社Uravation代表取締役。X（@SuguruKun_ai）フォロワー10万人超。
100社以上の企業向けAI研修・導入支援。著書累計3万部突破。
SoftBank IT連載7回執筆（NewsPicks最大1,125ピックス）。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。
