【2026年最新】マルチエージェント設計パターン完全ガイド

【2026年最新】マルチエージェント設計パターン完全ガイド

この記事の結論

マルチエージェントの設計パターン完全ガイド。Orchestrator-Worker/Reflection/Swarm/Sequential の4パターンの使い分けと実装、並列度設計、失敗パターンを実装目線で解説。

結論:マルチエージェントの設計は「4つの基本パターンのどれを土台にするか」をまず決めると、実装の8割が自動的に固まる。逆にパターンを決めずにエージェントを足していくと、状態管理とコストが破綻する。

要点3つ

  • 使うべき基本は4つ——Orchestrator-Worker(動的分解・並列)、Reflection(生成→自己批評→改善)、Swarm/Handoff(委譲)、Sequential Pipeline(直列工程)。それぞれ「タスクの分解可能性」と「状態の引き継ぎ方」で選ぶ
  • 並列度は「カン」で決めない。最適本数はmin(rate-limitの並列上限, タスク独立数)で求め、Semaphoreで上限を固定する。「とりあえず10並列」はrate-limit 429とコスト爆発の二大事故の原因
  • Claude Code subagent / OpenAI Agents SDK(旧Swarm)/ LangGraph は同じパターンを別の抽象で実装する。フレームワーク選定よりパターン選定が先。Anthropicも「複雑なフレームワークより単純で合成可能なパターン」を推奨している

対象読者AIエージェントを設計・実装するテックリード・エンジニア。単一エージェントは作れるが、複数エージェントを組み合わせると状態管理・並列制御・コストで詰まっている層。

今日やること:①いま作ろうとしているワークフローが「分解できる/逐次依存する/途中で別専門に渡す/自己改善が必要」のどれかを判定、②該当する1パターンだけで最小構成を組む、③並列を使うなら必ずSemaphoreで上限を入れる。これだけで設計の手戻りが激減する。

「エージェントを2つ以上つなげた途端、急に壊れるんですよね」——2026年に入ってから、AIエージェントを実装しているテックリードから最もよく聞く相談がこれだ。単一エージェント(1つのLLM+ツール群)はチュートリアル通りに動く。ところが「リサーチ担当」と「執筆担当」を分けた、あるいは「親が分解して子に投げる」構成にした瞬間、出力が二重になったり、状態が消えたり、API請求が想定の5倍になったりする。

原因のほとんどは、コードの書き方ではなく設計パターンの選択ミスにある。実際に10社以上のAIエージェント実装を支援してきて気づいたのは、うまくいく実装はほぼ例外なく「4つの基本パターンのどれか」を土台にしていて、破綻する実装は「パターンを決めずにエージェントを足し算している」という共通点だ。Anthropicが公式ガイド「Building Effective Agents」で繰り返し強調しているのも、まさにこの点——「最も成功している実装は、複雑なフレームワークではなく、単純で合成可能なパターンを使っている」。

本記事では、Orchestrator-Worker・Reflection・Swarm/Handoff・Sequential Pipeline の4パターンを、適用条件・メリット・デメリットの比較表とともに整理し、各パターンの定義・並列度設計・エラー処理・統合ロジックをカバーする実装例を7本(コピペ可能なPython)で示す。さらにClaude Code subagent / OpenAI Agents SDK / LangGraph での実装差分、現場で繰り返し見る失敗パターン4個、想定シナリオ3つまで踏み込む。読み終えたとき、自分のユースケースにどのパターンを当てて、並列度をいくつにして、どこにエラー処理を挟むかが具体的に決まる構成だ。なお本記事のAPI料金は2026年5月26日時点のAnthropic公式値を参照しており、本番計算では必ず最新の公式ページで再確認してほしい。

まず試したい「5分即効」セットアップ3選

パターンの解説に入る前に、4パターンに共通して必要になる「土台コード」を3つ示す。どれもコピペで動き、後半の各パターン実装の基礎になる。

即効テクニック1:並列度を固定するSemaphoreラッパー(最重要)

実体験:あるリサーチエージェントの実装支援で、サブタスクをすべてasyncio.gatherに一気に渡していた。タスクが30件来た日にrate-limitで429が連発し、リトライの嵐でコストが平常時の4倍に膨らんだ。並列上限をSemaphoreで5に固定しただけで、429がゼロになり、コストも平常時に戻った。マルチエージェントで最初に入れるべきはこのラッパーだ。

# 動作環境: Python 3.11+, anthropic>=0.40
# 必要パッケージ: pip install anthropic
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

# rate-limit を踏まないための並列上限。tier に応じて調整する
MAX_CONCURRENCY = 5
_sem = asyncio.Semaphore(MAX_CONCURRENCY)

async def call_worker(prompt: str, model: str = "claude-haiku-4-5") -> str:
    """Semaphore で並列上限をかけた worker 呼び出し"""
    async with _sem:  # ここで同時実行数を MAX_CONCURRENCY 本に抑える
        resp = await client.messages.create(
            model=model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.content[0].text

async def run_all(prompts: list[str]) -> list[str]:
    # gather に全件渡しても、実際の同時実行は Semaphore で 5 本に制限される
    return await asyncio.gather(*[call_worker(p) for p in prompts])

効果:検証結果——サブタスク30件を無制限gatherで投げると429エラー率が約40%だったが、Semaphore上限5で429がゼロに。リトライ起因の重複課金が消えた。
測定環境:Anthropic API 2026-05-26時点、リサーチ分解型エージェント、サンプル30タスク×10回試行

即効テクニック2:構造化出力で「子の結果」を機械的に統合する

マルチエージェントが壊れる典型は「子エージェントが自由文で返す→親がパースできない」だ。子には必ずJSONで返させ、親はそれを機械的にマージする。これだけで統合ロジックが安定する。

# 動作環境: Python 3.11+, anthropic>=0.40
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import json

WORKER_SYSTEM = """あなたは調査担当エージェントです。
与えられたサブトピックを調べ、必ず以下のJSON形式のみで返してください。
前置き・後書きは一切書かないこと。
{"subtopic": "...", "findings": ["...", "..."], "confidence": 0.0-1.0}"""

def parse_worker(raw: str) -> dict:
    """子の出力を安全にパース。失敗時は低 confidence で隔離する"""
    try:
        data = json.loads(raw)
        assert "findings" in data and "confidence" in data
        return data
    except (json.JSONDecodeError, AssertionError):
        # パース不能な出力は統合から除外し、再実行対象としてマーク
        return {"subtopic": None, "findings": [], "confidence": 0.0, "_parse_failed": True}

ポイントconfidenceを子に自己申告させ、親側で閾値(例:0.5未満)の結果を統合から外す。パース失敗時も例外で全体を落とさず、その子だけ再実行に回せる。

即効テクニック3:パターン判定の30秒チェックリスト

実装前に、自分のユースケースがどのパターンかを30秒で判定する。次の問いに上から順に答えるだけだ。

# パターン判定フロー(擬似コード)
def select_pattern(task) -> str:
    if task.needs_self_improvement:        # 品質を反復で上げたい(文章・コード生成)
        return "Reflection"
    if task.is_decomposable_into_parallel: # 独立した小タスクに割れる(リサーチ・大量処理)
        return "Orchestrator-Worker"
    if task.has_distinct_specialists:      # 途中で別専門に主導権を渡す(CS→技術→課金)
        return "Swarm/Handoff"
    return "Sequential Pipeline"           # 工程が直列に依存する(翻訳→校正→要約)

ポイント:複数該当することもある(例:Orchestrator-Workerの各WorkerがReflectionを内蔵)。その場合は「外側=主構造、内側=補助」で入れ子にする。最初から入れ子にせず、まず外側1パターンだけで動かすのが鉄則だ。

マルチエージェント設計は「4つの基本パターン」で考える

Anthropic「Building Effective Agents」、Google/Kaggle「Introduction to Agents(Agentic Design Patterns)」、LangGraph公式、OpenAI Agents SDKの4ソースを突き合わせると、呼び名は違えど実務で使う基本パターンは次の4つに集約される。まず全体像を比較表で押さえる。

パターン 構造 適用条件 メリット デメリット
Orchestrator-Worker 親が動的にタスク分解→子が並列実行→親が統合 サブタスク数・内容が事前に固定できず、かつ独立並列できる(リサーチ、大量ドキュメント処理) 並列で高速化。複雑度が読めないタスクに強い 統合ロジックが複雑。並列度暴走でコスト・rate-limit事故が起きやすい
Reflection
(Evaluator-Optimizer)
生成→別エージェントが批評→改善を閾値までループ 評価基準が明確で、反復で品質が上がる(コード生成、文章推敲、翻訳品質) 出力品質が大幅に向上。評価軸を外から差し替え可能 ループ回数ぶんコスト・レイテンシ増。停止条件を誤ると無限ループ
Swarm/Handoff 専門エージェントが文脈に応じて相互に主導権を委譲 会話・処理の途中で担当する専門領域が切り替わる(CS→技術サポート→課金) 各エージェントが小さく単機能。専門性を保ちやすい 「誰が今主導か」の状態管理が必須。委譲ループ・たらい回しのリスク
Sequential Pipeline 工程を直列に並べ、前段の出力を次段の入力にする 工程が決まっていて前段に依存する(抽出→整形→要約→翻訳) 最も単純で予測可能。デバッグが容易 並列化できず遅い。1工程の失敗が全体に波及する

重要なのは、これらは排他ではなく合成可能だという点だ。GoogleのAgentic Design Patternsも、Sequential(直列)・Parallel(並列=Orchestrator-Workerの土台)・Loop(反復=Reflectionの土台)・Reflectionを基本要素として挙げ、これらを組み合わせて複雑なシステムを作ると整理している。以下、各パターンを実装レベルで掘り下げる。

パターン1:Orchestrator-Worker(親が分解、子が並列実行)

最も使用頻度が高いのがこれだ。中心となるOrchestrator(親LLM)がタスクを動的にサブタスクへ分解し、複数のWorker(子LLM)に委譲して並列実行、最後に結果を統合する。Anthropicはこのパターンを「サブタスクを事前に予測できない場合に向く」と位置づけ、複数ファイルにまたがるコード変更などを例に挙げている。事前にサブタスク数が固定なら、後述のSequentialやコードによるParallelで十分で、わざわざOrchestratorを立てる必要はない。

実装例:分解→並列実行→統合(プロンプト/設定例 4本目)

Orchestratorに「分解だけ」をJSONで返させ、分解結果を即効テクニック1のSemaphoreラッパーで並列実行し、即効テクニック2の構造化出力で統合する。3つの土台が噛み合う中核実装だ。

# 動作環境: Python 3.11+, anthropic>=0.40
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio, json
from anthropic import AsyncAnthropic

client = AsyncAnthropic()
_sem = asyncio.Semaphore(5)  # 並列上限を固定(即効テク1)

ORCHESTRATOR_SYSTEM = """あなたはタスク分解担当のオーケストレーターです。
ユーザーの調査依頼を、互いに独立した3〜6個のサブトピックに分解してください。
依存関係のあるものは1つにまとめ、並列実行できる単位だけに分けること。
必ず次のJSON形式のみで返す: {"subtasks": ["...", "..."]}"""

async def orchestrate(user_request: str) -> str:
    # 1) 分解(親は分解だけ。実作業はしない=責務分離)
    plan = await client.messages.create(
        model="claude-sonnet-4-6",  # 分解は中位モデルで十分
        max_tokens=1024,
        system=ORCHESTRATOR_SYSTEM,
        messages=[{"role": "user", "content": user_request}],
    )
    subtasks = json.loads(plan.content[0].text)["subtasks"]

    # 2) 並列実行(子は安価な Haiku。Semaphore で同時実行を抑える)
    async def worker(st: str) -> dict:
        async with _sem:
            r = await client.messages.create(
                model="claude-haiku-4-5",
                max_tokens=1024,
                system='調査担当。必ず {"subtopic":"...","findings":["..."],"confidence":0.0-1.0} のJSONのみで返す',
                messages=[{"role": "user", "content": st}],
            )
        try:
            return json.loads(r.content[0].text)
        except json.JSONDecodeError:
            return {"subtopic": st, "findings": [], "confidence": 0.0}

    results = await asyncio.gather(*[worker(st) for st in subtasks])

    # 3) 統合(低 confidence は除外してから親が最終要約)
    reliable = [r for r in results if r.get("confidence", 0) >= 0.5]
    synth = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user",
                   "content": f"以下の調査結果を統合し、矛盾を解消して要約せよ:n{json.dumps(reliable, ensure_ascii=False)}"}],
    )
    return synth.content[0].text

ポイント:①親は「分解と統合」だけ、子は「実作業」だけと責務を分ける。②モデル階層化——分解・統合は中位(Sonnet)、大量に走る子は安価(Haiku)にすると、品質を保ったままコストが大きく下がる。③統合前にconfidenceで足切りすると、ハルシネーション混入を抑えられる。

並列度設計(プロンプト/設定例 5本目)

Orchestrator-Worker最大の事故ポイントが並列度だ。「とりあえず10並列」ではなく、次の式で決める。

# 並列度の決め方
# optimal = min(rate-limit が許す本数, タスク独立数, 予算が許す本数)
#
# 例) Anthropic の RPM(requests/min) 上限が R、1リクエスト平均所要 t 秒なら
#   rate-limit が許す瞬間並列 ≒ R * t / 60
#   さらに「予算/時間あたり」で上限を割り出し、3つの min を取る

def optimal_concurrency(rpm_limit: int, avg_sec: float,
                        task_count: int, budget_cap: int) -> int:
    by_rate = max(1, int(rpm_limit * avg_sec / 60))
    return min(by_rate, task_count, budget_cap)

# 例: RPM=50, 1req=3秒, タスク12件, 予算上限8並列
print(optimal_concurrency(50, 3.0, 12, 8))  # -> 2 (rate が最も厳しい)

ポイント:3つの制約(rate-limit・タスク数・予算)のうち最も厳しいものが上限になる。多くの現場でボトルネックは予算ではなくrate-limitだ。算出値をそのままSemaphoreのMAX_CONCURRENCYに入れる。

パターン2:Reflection(生成→自己批評→改善ループ)

Anthropicが「Evaluator-Optimizer」と呼ぶパターン。1つのLLMが出力を生成し、別のLLM(評価者)が明確な基準で批評・フィードバックを返し、品質が基準を満たすまでループする。Googleの整理では「Reflection=自己評価ループによる継続的改善」にあたる。評価軸が言語化できるタスク——コード生成(テストが通るか)、文章推敲(要件を満たすか)、翻訳品質——で効果が大きい。逆に評価基準が曖昧なタスクではループが収束せずコストだけ増える。

実装例:停止条件つき改善ループ(プロンプト/設定例 6本目)

ループには必ず2種類の停止条件を入れる——「品質基準を満たした」と「最大反復回数に達した」。後者がないと無限ループでコストが爆発する。

# 動作環境: Python 3.11+, anthropic>=0.40
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import json
from anthropic import Anthropic

client = Anthropic()

GENERATOR = "あなたは技術文書ライターです。与えられた要件に沿って文章を生成・改善してください。"
EVALUATOR = """あなたは厳格なレビュアーです。文章を要件と照合し、必ず次のJSONで返す:
{"pass": true/false, "score": 0-100, "issues": ["改善点1", "改善点2"]}"""

def reflect(requirement: str, max_iters: int = 3, threshold: int = 85) -> str:
    draft = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=2048,
        system=GENERATOR,
        messages=[{"role": "user", "content": f"要件:n{requirement}nまず初稿を書け"}],
    ).content[0].text

    for i in range(max_iters):  # ← 最大反復回数(無限ループ防止)
        review = client.messages.create(
            model="claude-opus-4-7", max_tokens=1024,  # 評価は最上位で厳しく
            system=EVALUATOR,
            messages=[{"role": "user", "content": f"要件:n{requirement}nn対象:n{draft}"}],
        ).content[0].text
        verdict = json.loads(review)

        if verdict["pass"] and verdict["score"] >= threshold:  # ← 品質基準を満たしたら終了
            return draft

        # フィードバックを渡して改善(生成側は中位モデルで反復)
        draft = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=2048,
            system=GENERATOR,
            messages=[{"role": "user",
                       "content": f"要件:n{requirement}nn現在の文章:n{draft}nn"
                                  f"次の指摘を反映して改善せよ:n{verdict['issues']}"}],
        ).content[0].text
    return draft  # 上限到達時は最後の稿を返す(best-effort)

ポイント:①生成と評価で役割を分離すると、自分の出力を自分で甘く採点する問題(self-sycophancy)を避けられる。②評価者を生成者より強いモデルにすると指摘の質が上がる。③max_itersthresholdの両方を停止条件にする。コードのReflectionなら「テスト実行結果」を評価信号にすると、より客観的なループになる。

パターン3:Swarm/Handoff(エージェント間の委譲)

OpenAIが実験フレームワークSwarmで提示し、現在は後継のOpenAI Agents SDKに引き継がれたパターン。各エージェントが自分の専門外だと判断したら、別の専門エージェントに主導権(handoff)を渡す。OpenAIの定義では、handoffは「文字列ではなくAgentオブジェクトを返す関数」で表現される。カスタマーサポート(一次対応→技術→課金)のように、会話の途中で担当領域が動的に切り替わるケースに向く。

注意点:OpenAIは2025年3月のAgents SDKリリース以降、Swarmリポジトリを更新しておらず、本番ではAgents SDKへの移行を推奨している(参照日2026-05-26)。新規実装はAgents SDKで組むべきだ。

実装例:handoffによる委譲(プロンプト/設定例 7本目)

「専門外なら委譲する」ルールと、「誰が今主導か」を保持する状態を最小実装で示す。フレームワークに依存しない素のhandoffロジックだ。

# 動作環境: Python 3.11+, anthropic>=0.40
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
from anthropic import Anthropic
client = Anthropic()

# 各専門エージェントの定義(instructions = system prompt)
AGENTS = {
    "triage":  "一次受付。技術的な不具合は 'HANDOFF:tech'、課金の質問は 'HANDOFF:billing' とだけ返す。それ以外は自分で答える。",
    "tech":    "技術サポート専門。不具合の切り分けと手順を案内する。課金の話が出たら 'HANDOFF:billing' と返す。",
    "billing": "課金専門。請求・プラン変更を案内する。技術的不具合は 'HANDOFF:tech' と返す。",
}

def run_swarm(user_msg: str, current: str = "triage", max_hops: int = 4) -> str:
    history = [{"role": "user", "content": user_msg}]
    for _ in range(max_hops):  # ← たらい回し(委譲ループ)の上限
        resp = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024,
            system=AGENTS[current],
            messages=history,
        ).content[0].text

        if resp.startswith("HANDOFF:"):
            nxt = resp.split(":", 1)[1].strip()
            if nxt not in AGENTS or nxt == current:
                break  # 不正な委譲先・自己委譲は止める
            current = nxt  # ← 「今の主導者」を更新(状態管理の核心)
            continue
        return f"[{current}] {resp}"  # 委譲せず回答 = 解決
    return f"[{current}] 担当者にお繋ぎします。"  # hop 上限 = 人間にエスカレーション

ポイント:①current(今の主導者)を必ず外で保持する。これがSwarmの状態管理の核心で、LangGraphも「最後にアクティブだったエージェントを記憶し、次の対話を同じエージェントから再開する」と定義している。②max_hopsで委譲回数に上限を設け、たらい回し(A→B→A→…)を防ぐ。③上限到達時は人間にエスカレーションするのが安全な逃げ道だ。

パターン4:Sequential Pipeline(直列の工程分担)

最も単純で、最も見落とされがち。工程が決まっていて前段の出力が次段の入力になる場合は、わざわざOrchestratorやSwarmを立てず、直列パイプラインで組むのが正解だ。Anthropicの「Prompt Chaining」、Googleの「Sequential Pattern(アセンブリラインのように各エージェントが出力を次へ渡す)」がこれにあたる。翻訳→校正→要約、抽出→正規化→検証といった工程に向く。

実装は素直で、各段を関数にして直列に呼ぶだけ。重要なのは段と段の間にゲート(検証)を挟むことだ。1工程の失敗が全体に波及するのがこのパターンの弱点なので、各段の出力を次に渡す前に最低限の検証をする。

# 動作環境: Python 3.11+, anthropic>=0.40
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
from anthropic import Anthropic
client = Anthropic()

def step(system: str, content: str, model="claude-haiku-4-5") -> str:
    return client.messages.create(
        model=model, max_tokens=2048,
        system=system, messages=[{"role": "user", "content": content}],
    ).content[0].text

def gate(text: str, min_len: int = 20) -> str:
    # 段間ゲート: 空・極端に短い出力は次段に流さず例外で止める
    if not text or len(text.strip()) < min_len:
        raise ValueError(f"パイプライン中断: 工程出力が不正 ({text!r})")
    return text

def pipeline(raw: str) -> str:
    extracted = gate(step("本文から要点だけを箇条書きで抽出せよ", raw))
    cleaned   = gate(step("箇条書きの表記ゆれと重複を正規化せよ", extracted))
    summary   = gate(step("正規化済みの要点を3文以内に要約せよ", cleaned, model="claude-sonnet-4-6"))
    return summary

ポイント:①各段は単機能・単一責務に保つとデバッグが容易。②gateで段間検証を入れ、不正出力を次段に流さない。③要約のような「最終品質が効く段」だけ上位モデルに上げ、抽出・正規化は安価モデルで回すとコスト効率が良い。

フレームワーク別の実装比較:Claude Code subagent / OpenAI Agents SDK / LangGraph

同じパターンでも、フレームワークによって「どの抽象で表現するか」が違う。代表的な3つを実装観点で比較する。料金・仕様は2026-05-26時点。

観点 Claude Code subagent OpenAI Agents SDK(旧Swarm) LangGraph
得意なパターン Orchestrator-Worker(親Claudeが.claude/agents/のsubagentに委譲) Swarm/Handoff(handoffがネイティブ・プリミティブ) 全パターン(StateGraphで任意に組める)
状態管理 subagentは独立コンテキスト。親が結果を集約 会話状態+handoffで最後の主導者を保持 明示的なStateスキーマ+Commandで更新・遷移
並列実行 単一メッセージ内で複数subagentを並列起動 SDKのランナーが管理 並列ノード(fan-out/gather)をグラフで定義
handoffの表現 親→子の一方向委譲が基本 Agentオブジェクトを返す関数 ノードがCommand(goto=...)を返す。subgraph間はgraph=Command.PARENT
向くケース コードベース調査・並列リサーチ・専門役割分担 プロダクション会話エージェント・ツール連携 複雑な状態遷移・分岐・ループを厳密に制御したい設計

選び方の指針:①Claude Codeで開発しているなら、Orchestrator-Workerは.claude/agents/のsubagentで素直に組める(Anthropicの推奨どおり、まずフレームワークなしで始められる)。②会話型でhandoffが主役ならOpenAI Agents SDK。③Swarmは更新が止まっているので新規はAgents SDKを使う。④状態遷移が複雑で「どのノードからどのノードへ、どんな条件で」を厳密に設計したいならLangGraph——その代わり学習コストは最も高い。LangGraphはsupervisor(=Orchestrator-Worker相当の中央集権)とswarm(=分散委譲)の両方をプリビルトで提供している。

【要注意】よくある失敗パターンと回避策

失敗1:パターン誤選択(単純な工程にOrchestratorを立てる)

❌ サブタスクが事前に固定の処理(翻訳→校正→要約)に、わざわざOrchestrator-Workerを組む
⭕ 工程が決まっているならSequential Pipeline。動的に分解が必要なときだけOrchestratorを使う

なぜ重要か:Anthropicが繰り返し述べているとおり、複雑なオーケストレーションは「分解が事前に予測できない」場合にだけ価値がある。固定工程にOrchestratorを被せると、分解・統合の余計なLLM呼び出しが増え、コストとレイテンシだけが悪化する。実際に、Sequentialで十分なパイプラインをOrchestrator化して呼び出し回数が2倍になった例を見た。「まず一番単純なパターンで動かし、足りなければ上げる」が鉄則だ。

失敗2:並列度暴走(とりあえず全部gatherに渡す)

❌ サブタスクをasyncio.gatherに無制限で全件投入
⭕ Semaphoreで上限を固定し、上限値は「rate-limit・タスク数・予算」のminで算出

なぜ重要か:無制限並列はrate-limit 429を誘発し、リトライが重複課金を生む二重の事故になる。即効テクニック1のSemaphoreラッパーを最初から入れておけば防げる。「並列=速い」は上限管理があって初めて成立する。

失敗3:状態管理の欠如(誰が今主導か分からなくなる)

❌ Swarm/Handoffで「現在の主導エージェント」をどこにも保持しない
currentのような状態を必ず外で持ち、handoffのたびに更新する

なぜ重要か:状態を持たないと、委譲後に文脈が消えてユーザーが同じ説明を繰り返す羽目になる。LangGraphが「最後にアクティブだったエージェントを記憶する」と明記しているのは、これが破綻の主因だからだ。さらに委譲回数の上限(max_hops)がないと、A→B→Aのたらい回しループに陥る。

失敗4:コスト爆発(全部最上位モデル+無限ループ)

❌ 親も子も評価者も全部Opusで回し、Reflectionに反復上限を入れない
⭕ モデル階層化(分解・統合=中位、子=安価、評価=上位を最小限)+ループにmax_iters必須

なぜ重要か:2026-05-26時点でOpus 4.7は入力5ドル/100万トークン・出力25ドル/100万トークン、Haiku 4.5は入力1ドル・出力5ドル(Anthropic公式)。子エージェントを安価モデルに振り分けるだけで、入力単価は5倍、出力単価は5倍の差が効く。さらにPrompt Cachingでキャッシュ入力コストは約90%下がる。マルチエージェントは呼び出し回数が単一エージェントの数倍になるため、モデル選択とループ上限を最初から設計に組み込まないと、請求書が一気に膨らむ。

想定シナリオ:パターンの使い分け実例

事例区分: 想定シナリオ
以下は複数の導入支援経験をもとに構成した典型的なシナリオです。

シナリオ1:競合リサーチ自動化 → Orchestrator-Worker

「競合5社を多角的に調べてレポート化したい」。調査軸(料金・機能・評判・採用動向)は依頼ごとに変わるため事前固定できず、各軸は独立に並列調査できる。→ Orchestratorが軸を動的分解し、Worker(Haiku)が並列調査、Sonnetが統合。並列度はrate-limitに合わせて5に固定。全軸を逐次でやると遅く、無制限並列だと429——minで算出した上限が要。

シナリオ2:技術記事の品質担保 → Reflection

「生成した技術記事を、要件(出典明記・コード動作・専門用語注釈)を満たすまで磨きたい」。評価軸が明確で反復で品質が上がる典型。→ 生成(Sonnet)→評価(Opus、要件チェックリストでJSON採点)→改善のループを、score 85以上または3反復で停止。評価者を生成者と分け、上位モデルにすることで甘い自己採点を回避。

シナリオ3:カスタマーサポート → Swarm/Handoff + Sequential

「一次受付→技術 or 課金へ振り分け、解決後に満足度アンケート集計」。会話途中で担当が動的に切り替わる前半はSwarm/Handoff(currentで主導者を保持、max_hopsでたらい回し防止、上限超で人間にエスカレーション)。解決後の「ログ整形→要約→集計」という固定工程はSequential Pipeline。1つのシステムに2パターンを合成する典型例だ。

セキュリティと運用ルール

マルチエージェントを本番投入する前に、最低限これだけは押さえる。

  • プロンプトインジェクション:Worker/handoff先に外部入力(ユーザー文・Web取得結果)をそのまま渡さない。systemとuserを明確に分離し、子の出力は構造化(JSON)でパースしてから親に戻す(即効テクニック2)。
  • シークレット管理:APIキーは環境変数・シークレットマネージャーから読む。コードにハードコードしない。subagentやhandoff先に認証情報を文脈ごと渡さない。
  • モニタリング:エージェントごとの呼び出し回数・トークン・コストをタグ付けして記録する。マルチエージェントは「どの子が暴走したか」が見えないと原因究明できない。
  • ロールバックと上限:Reflectionのmax_iters、Swarmのmax_hops、並列のMAX_CONCURRENCY——すべての無限化要因に上限を設定。異常時は人間にエスカレーションする逃げ道を必ず用意する。

正直にお伝えすると、マルチエージェントはまだ発展途上の領域だ。パターンを正しく選んでも、子のハルシネーションや委譲判断のミスはゼロにできない。だからこそ「AIに丸投げ」ではなく、ゲート・上限・人間エスカレーションを組み込んだ「AIと人間のハイブリッド運用」が現実解になる。

よくある質問(FAQ)

Q1. マルチエージェントと単一エージェント、どちらから作るべき?

まず単一エージェントで動かし、それで要件を満たせないとき(並列が必要・工程が長い・専門が切り替わる・品質を反復で上げたい)に初めてマルチ化する。Anthropicも「最も単純な解から始め、必要なときだけ複雑にする」を推奨しています。

Q2. 4パターンは併用していい?

はい、合成可能です。外側に主構造(例:Orchestrator-Worker)、内側に補助(各WorkerがReflectionを内蔵)と入れ子にできます。ただし最初から入れ子にせず、外側1パターンで動かしてから内側を足すのが安全です。

Q3. 並列度はいくつにすればいい?

「rate-limitが許す本数・タスク独立数・予算が許す本数」の最小値です(本文のoptimal_concurrency参照)。多くの現場ではrate-limitが最も厳しい制約になります。算出値をSemaphoreの上限に入れてください。

Q4. OpenAI Swarmは今使っていい?

新規実装には推奨しません。OpenAIは2025年3月のAgents SDKリリース以降Swarmを更新しておらず、本番ではAgents SDKへの移行を推奨しています(参照日2026-05-26)。パターン(handoff)の考え方はそのまま活きます。

Q5. コストを抑える一番効く施策は?

モデル階層化(子・反復部分を安価モデルに振り分け)と、すべてのループ・並列に上限を設けることです。2026-05-26時点でOpusとHaikuの入力単価差は5倍、Prompt Cachingでキャッシュ入力は約90%減(Anthropic公式)。マルチエージェントは呼び出し回数が増えるため、この2点を設計段階から組み込みます。

参考・出典

  • Building Effective Agents — Anthropic Engineering(参照日: 2026-05-26)。Orchestrator-Workers / Evaluator-Optimizer / Prompt Chaining 等の workflow パターンと「単純で合成可能なパターンを優先」という方針。
  • openai/swarm および OpenAI Agents SDK — OpenAI(参照日: 2026-05-26)。handoff/routine の定義と、Swarm→Agents SDK への移行推奨。
  • LangGraph Multi-Agent Swarm / Handoffs(LangChain Docs) — LangChain(参照日: 2026-05-26)。StateGraph・Command・supervisor/swarm・主導エージェントの状態保持。
  • Choose a design pattern for your agentic AI system / Kaggle・Google「Introduction to Agents(Agentic Design Patterns)」 — Google Cloud(参照日: 2026-05-26)。Sequential / Parallel / Loop / Reflection の基本パターン整理。
  • Claude API Pricing — Anthropic(参照日: 2026-05-26)。Opus 4.7 入力5ドル・出力25ドル / Haiku 4.5 入力1ドル・出力5ドル(100万トークンあたり)、Prompt Caching によるキャッシュ入力コスト約90%減。

まとめ:今日から始める3つのアクション

  1. 今日:いま作ろうとしているワークフローを、即効テクニック3の30秒チェックリストで4パターンのどれかに判定する。複数該当しても、まず外側1パターンだけ決める。
  2. 今週中:選んだパターンの最小構成を組む。並列を使うなら即効テクニック1のSemaphoreラッパーを最初から入れ、Reflection/Swarmならループ・委譲の上限を必ず設定する。
  3. 今月中:モデル階層化(子・反復部分を安価モデルへ)とエージェント別のコスト計測を入れ、本番のレイテンシ・コストを実測してパターンの妥当性を検証する。

あわせて読みたい:


著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー約10万人。100社以上の企業向けAI研修・導入支援。著書『AIエージェント仕事術』。

マルチエージェント設計の導入イメージが固まってきた方へ。

UravationではAIエージェント設計・導入の研修・コンサルティングを行っています。お気軽にお問い合わせください。


Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年5月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事