◈ Agents & Manager/2026-04-11上級

AIエージェント・オーケストレーション：マルチエージェントシステムの設計と実装

マルチエージェントの並列実行は、1体の失敗で成功分まで失われます。4つの基本パターンを整理したうえで、部分失敗・LLM応答のJSON抽出・状態共有という本番で壊れる境界を、動くコードで修正しながら追いかけます。

エージェント⁷² オーケストレーション⁷ マルチエージェント⁴¹LLM自動化⁴⁰AI設計

✦ プレミアム記事

3体のエージェントを並列で走らせ、そのうち1体がレート制限で落ちる。それだけで、すでに完了していた残り2体の出力まで丸ごと失われる。マルチエージェントを本番に載せようとしたとき、私が最初に踏み抜いた床がこれでした。

原因は壮大な設計思想の側にはありませんでした。future.result() を素直にループで回していただけです。最初の例外がその場で送出され、他のエージェントが返し終えていた結果を受け取る前に、関数ごと終わってしまう。

オーケストレーションの難所は、パターンの選定よりもこうした「境界の実装」に宿ります。4つの基本パターンを整理したうえで、オーケストレーターのどこが本番で壊れるのかを、動くコードで追っていきます。

なお本稿は公開後に一度手を入れています。当初掲載していた実装には、部分失敗の扱い・LLM応答からのJSON抽出・タイムスタンプの3点に穴がありました。該当箇所は修正済みのコードへ差し替え、なぜ壊れるのかを併記しています。

なぜマルチエージェントが必要か

単一エージェントは、シンプルなタスクに対して非常に効率的です。しかし、現実のビジネスプロセスはそう単純ではありません。

単一エージェントの限界

コンテキストウィンドウの制約: 最新のLLMでも、処理できる情報量には上限があります。長大なドキュメントの分析や、多段階にわたる複雑なタスクでは、コンテキストウィンドウがボトルネックになります。

専門性の欠如: 1つのエージェントがあらゆるタスクに対応しようとすると、プロンプトが複雑になり、出力の品質が低下します。会計処理のプロフェッショナルと法務レビューの専門家を同一人物に求めるようなものです。

並列処理の不可能性: 単一エージェントは本質的にシーケンシャルです。独立したタスクAとタスクBがある場合でも、一方が完了するまで他方を開始できません。

エラー伝播のリスク: 単一エージェントがエラーを起こすと、ワークフロー全体が停止します。分離された複数エージェントなら、一部の失敗が全体に波及しにくくなります。

マルチエージェントが解決すること

マルチエージェントシステムでは、これらの課題を次のように解決します。各エージェントは明確に定義された役割と、その役割に必要なツールのみを持ちます。エージェント間の通信は構造化されたプロトコルで行われ、並列実行が可能です。そして、一部のエージェントの失敗が全体を止めない「耐障害性」が生まれます。

オーケストレーションの4つの基本パターン

パターン1：中央集権型オーケストレーター

最も一般的なパターンです。中央のオーケストレーターがすべての判断を行い、サブエージェントに指示を送ります。

ユーザー
  ↓
オーケストレーター（中央指揮）
  ├── 指示 → サブエージェントA
  ├── 指示 → サブエージェントB
  └── 指示 → サブエージェントC
        ↑
     結果を集約してユーザーへ返す

利点: 全体の状態を1か所で管理できるため、デバッグが容易。タスクの依存関係を明示的に制御できます。

欠点: オーケストレーター自体がSPOF（Single Point of Failure）になるリスク。オーケストレーターのコンテキストが肥大化しやすい。

適した場面: タスク間に複雑な依存関係がある場合、厳密な実行順序の制御が必要な場合。

パターン2：分散型ピアツーピア

エージェントが相互に通信し、中央指揮なしに協調します。

エージェントA ←→ エージェントB
     ↕                ↕
エージェントC ←→ エージェントD

利点: 単一障害点がありません。各エージェントが独立してスケールできます。

欠点: 全体の状態把握が難しいです。デッドロックや無限ループのリスクがあります。

適した場面: 各エージェントの役割が明確で独立性が高い場合。ピアレビューや相互検証が必要な場合。

パターン3：階層型マルチレベル

オーケストレーターが複数の中間マネージャーを通じてリーフエージェントを管理します。

トップオーケストレーター
  ├── マネージャーA
  │     ├── ワーカーA1
  │     └── ワーカーA2
  └── マネージャーB
        ├── ワーカーB1
        └── ワーカーB2

利点: 大規模システムへのスケーラビリティ。各階層での責任分離。

欠点: レイテンシの増加。階層間のコミュニケーションオーバーヘッド。

適した場面: 複数の独立したサブシステムを統合する大規模ワークフロー。

パターン4：動的エージェント生成

オーケストレーターがタスクの内容に応じて、必要なエージェントを動的に生成・破棄します。

def dynamic_orchestrator(task: str) -> str:
    """タスク分析に基づいて必要なエージェントを動的生成"""
    
    # タスク分析フェーズ
    task_analysis = analyze_task(task)
    required_agents = task_analysis["agents_needed"]
    
    # エージェントを動的に初期化
    active_agents = {}
    for agent_spec in required_agents:
        active_agents[agent_spec["id"]] = create_agent(
            role=agent_spec["role"],
            tools=agent_spec["tools"],
            system_prompt=agent_spec["prompt"]
        )
    
    # タスク実行
    results = execute_with_agents(task, active_agents)
    
    # リソース解放
    for agent in active_agents.values():
        agent.cleanup()
    
    return results

利点: リソースの効率的な使用。タスクに最適化されたエージェント構成。

適した場面: タスクの種類が多様で、事前に必要なエージェントを特定できない場合。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦マルチエージェントシステムの主要アーキテクチャパターンとその使い分けの判断基準

✦オーケストレーターとサブエージェントの役割分担・通信プロトコル・状態管理の実装手法

✦本番運用で直面するスケーリング・信頼性・コスト課題と実践的な解決アプローチ

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

オーケストレーターの実装：詳細解説

実際にオーケストレーターを実装する際の重要なポイントを順を追って説明します。

タスク分解エンジン

オーケストレーターの核心は、複雑なタスクを実行可能な小タスクに分解する能力です。

まず共通のユーティリティを用意します。LLM は「はい、承知しました」といった前置きや、コードフェンスで囲んだ形で返すことがあり、json.loads() をそのまま当てるとここで落ちます。実際に前置き付きの応答を json.loads() に渡すと JSONDecodeError: Expecting value が返ります。

import json
import re
 
def extract_json(text: str) -> dict:
    """LLM 応答から最初の JSON オブジェクト/配列を取り出す"""
    fence = chr(96) * 3  # コードフェンス（バッククォート3つ）
    body = text.strip()
 
    fenced = re.search(fence + r"(?:json)?\s*(.+?)\s*" + fence, body, re.S)
    if fenced:
        body = fenced.group(1).strip()
 
    starts = [i for i in (body.find("{"), body.find("[")) if i != -1]
    if not starts:
        raise ValueError(f"JSON not found in model output: {text[:200]}")
 
    # raw_decode なら JSON の後ろに地の文が続いていても読み取れる
    obj, _ = json.JSONDecoder().raw_decode(body[min(starts):])
    return obj

フェンス付き・素の JSON・末尾に地の文が続くケースのいずれも、この1関数で吸収できます。構造化出力（tool use）を使える場面ならそちらが確実ですが、モデルを差し替えながら試作する段階では、この受け皿があるだけで事故が目に見えて減ります。

from anthropic import Anthropic
import json
 
client = Anthropic()
 
def decompose_task(task: str, available_agents: list[dict]) -> list[dict]:
    """
    LLMを使ってタスクを分解し、各サブタスクをエージェントに割り当てる
    """
    decompose_prompt = f"""
    あなたはタスク分解の専門家です。以下のタスクを、利用可能なエージェントで実行可能な
    サブタスクに分解してください。
    
    メインタスク: {task}
    
    利用可能なエージェント:
    {json.dumps(available_agents, ensure_ascii=False, indent=2)}
    
    以下のJSON形式で回答してください:
    {{
        "subtasks": [
            {{
                "id": "task_1",
                "description": "サブタスクの説明",
                "assigned_agent": "エージェントID",
                "depends_on": [],
                "can_parallel": true
            }}
        ],
        "execution_order": [["task_1", "task_2"], ["task_3"]]
    }}
    
    execution_orderは並列実行可能なグループを表す二次元配列です。
    """
    
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=4096,
        messages=[{"role": "user", "content": decompose_prompt}]
    )
    
    return extract_json(response.content[0].text)
 
import concurrent.futures
 
def execute_task_plan(plan: dict, agents: dict) -> dict:
    """実行計画に従ってタスクを順次・並列実行"""
    results: dict = {}
    failures: dict = {}
    
    for parallel_group in plan["execution_order"]:
        if len(parallel_group) == 1:
            # シーケンシャル実行
            task_id = parallel_group[0]
            subtask = next(t for t in plan["subtasks"] if t["id"] == task_id)
            agent = agents[subtask["assigned_agent"]]
            
            # 依存タスクの結果を注入
            context = {dep: results[dep] for dep in subtask["depends_on"]}
            results[task_id] = agent.execute(subtask["description"], context)
        else:
            # 並列実行（1体の失敗で他の結果を失わない）
            with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
                futures = {}
                for task_id in parallel_group:
                    subtask = next(t for t in plan["subtasks"] if t["id"] == task_id)
                    agent = agents[subtask["assigned_agent"]]
                    context = {dep: results[dep] for dep in subtask["depends_on"]}
                    futures[executor.submit(
                        agent.execute, subtask["description"], context
                    )] = task_id
                
                for future in concurrent.futures.as_completed(futures):
                    task_id = futures[future]
                    try:
                        results[task_id] = future.result()
                    except Exception as exc:
                        failures[task_id] = f"{type(exc).__name__}: {exc}"
    
    return {"results": results, "failures": failures}

ここが冒頭で触れた床の抜けていた箇所です。修正前は for task_id, future in futures.items(): results[task_id] = future.result() と書いていました。この形だと、辞書の並び順で最初に触れた future が例外を持っていた瞬間に execute_task_plan 全体が送出側に抜けます。3体中2体が成功していても、その2体分のトークン代だけが消えて手元には何も残りません。

as_completed で終わった順に取り出し、try/except で失敗を failures に隔離すれば、成功分は必ず手元に残ります。3体のうち1体だけをレート制限で失敗させて手元で動かすと、戻り値は {'results': {'c': 'c:done', 'a': 'a:done'}, 'failures': {'b': 'RuntimeError: rate limit'}} となり、成功2件と失敗1件がきれいに分離されます。

戻り値の形が変わる点には注意してください。呼び出し側は failures が空かどうかを見て、再実行するのか、欠けたまま先へ進めるのかを判断します。部分失敗を「握りつぶす」のではなく「見えるところに置く」のが要点です。

エージェント間の状態共有

マルチエージェントシステムでは、エージェント間で状態を共有するための仕組みが必要です。

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict
import threading
 
@dataclass
class SharedWorkspace:
    """エージェント間で共有される作業スペース"""
    artifacts: Dict[str, Any] = field(default_factory=dict)
    messages: list = field(default_factory=list)
    _lock: threading.RLock = field(default_factory=threading.RLock)
    
    def write_artifact(self, key: str, value: Any, agent_id: str):
        """アーティファクトの書き込み（スレッドセーフ）"""
        with self._lock:
            self.artifacts[key] = {
                "value": value,
                "written_by": agent_id,
                "timestamp": datetime.now(timezone.utc).isoformat()
            }
    
    def read_artifact(self, key: str) -> Any:
        """アーティファクトの読み取り"""
        with self._lock:
            item = self.artifacts.get(key)
            return item["value"] if item else None
    
    def post_message(self, from_agent: str, to_agent: str, content: str):
        """エージェント間メッセージング"""
        with self._lock:
            self.messages.append({
                "from": from_agent,
                "to": to_agent,
                "content": content,
                "timestamp": datetime.now(timezone.utc).isoformat()
            })
    
    def get_messages_for(self, agent_id: str) -> list:
        """特定エージェント宛のメッセージを取得"""
        with self._lock:
            return [m for m in self.messages if m["to"] == agent_id]

フィードバックループとクオリティゲート

高品質な出力を保証するために、エージェントの出力を評価してフィードバックを返すループが重要です。

def quality_gate(output: str, criteria: dict) -> dict:
    """
    エージェントの出力品質を評価するクオリティゲート
    """
    evaluation_prompt = f"""
    以下の出力を評価してください。
    
    出力:
    {output}
    
    評価基準:
    {json.dumps(criteria, ensure_ascii=False)}
    
    JSON形式で回答:
    {{
        "passed": true/false,
        "score": 0-100,
        "issues": ["問題点1", "問題点2"],
        "improvement_suggestions": ["改善案1", "改善案2"]
    }}
    """
    
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": evaluation_prompt}]
    )
    
    return extract_json(response.content[0].text)
 
def iterative_refinement(
    task: str,
    agent,
    criteria: dict,
    max_iterations: int = 3
) -> str:
    """品質基準を満たすまで反復的に改善"""
    
    result = agent.execute(task)
    
    for i in range(max_iterations):
        quality = quality_gate(result, criteria)
        
        if quality["passed"]:
            return result
        
        # 改善フィードバックを含めて再実行
        improvement_task = f"""
        元のタスク: {task}
        
        前回の出力:
        {result}
        
        改善すべき点:
        {json.dumps(quality["improvement_suggestions"], ensure_ascii=False)}
        
        上記の改善点を踏まえて、より良い出力を生成してください。
        """
        result = agent.execute(improvement_task)
    
    return result  # 最大試行回数後の最良の結果を返す

実践ユースケース：マルチエージェントで構築するコンテンツ制作パイプライン

理論を実際のシナリオで確認しましょう。ブログ記事を自動生成するマルチエージェントパイプラインを例にとります。

システムアーキテクチャ

ユーザー（トピック指定）
  ↓
オーケストレーター
  ├── [並列] リサーチエージェント群
  │     ├── ウェブ検索エージェント
  │     ├── 学術論文検索エージェント
  │     └── 競合コンテンツ分析エージェント
  ├── [シーケンシャル] コンテンツ制作エージェント群
  │     ├── アウトライン作成エージェント
  │     ├── ドラフト執筆エージェント
  │     └── 編集・校正エージェント
  └── [最終] 品質検証エージェント
        ├── 事実確認エージェント
        ├── SEO最適化エージェント
        └── 最終承認エージェント

各エージェントの役割と実装

class ResearchAgent:
    def __init__(self, search_tools: list):
        self.tools = search_tools
        self.system_prompt = """あなたは優れたリサーチャーです。
        指定されたトピックについて、信頼性の高い情報源から情報を収集し、
        構造化された形式でまとめてください。"""
    
    def execute(self, topic: str, context: dict = None) -> dict:
        research_prompt = f"以下のトピックについて詳しく調査してください: {topic}"
        
        response = client.messages.create(
            model="claude-opus-5",
            max_tokens=8192,
            system=self.system_prompt,
            tools=self.tools,
            messages=[{"role": "user", "content": research_prompt}]
        )
        
        return {
            "findings": response.content[0].text,
            "sources": self._extract_sources(response)
        }
 
class WritingAgent:
    def __init__(self):
        self.system_prompt = """あなたは経験豊富なテクニカルライターです。
        リサーチ結果を基に、読みやすく情報量の豊富な記事を執筆してください。
        見出しには具体的で魅力的なタイトルを使い、コード例は実際に動作するものを提供してください。"""
    
    def execute(self, outline: str, research_data: dict) -> str:
        writing_prompt = f"""
        以下のアウトラインと調査データを基に、記事本文を執筆してください。
        
        アウトライン:
        {outline}
        
        調査データ:
        {json.dumps(research_data, ensure_ascii=False)}
        
        要件:
        - 各セクションを十分な深さで解説する
        - 実践的なコード例を含める
        - 読者が実際に活用できる知識を提供する
        """
        
        response = client.messages.create(
            model="claude-opus-5",
            max_tokens=16384,
            system=self.system_prompt,
            messages=[{"role": "user", "content": writing_prompt}]
        )
        
        return response.content[0].text

スケーリングと本番運用の課題

マルチエージェントシステムを本番環境で運用する際の現実的な課題と解決策を紹介します。

コスト管理

マルチエージェントシステムはAPI呼び出し数が多くなるため、コスト管理が特に重要です。

トークン予算制: 各エージェントに割り当てるトークン上限を設定し、予算内に収める。

class BudgetAwareAgent:
    def __init__(self, max_tokens: int = 4096, model: str = "claude-sonnet-5"):
        self.max_tokens = max_tokens
        self.model = model
        self.tokens_used = 0
    
    def execute(self, task: str) -> str:
        if self.tokens_used > self.max_tokens * 0.9:
            # 予算残量が10%を切ったら軽量モデルにフォールバック
            self.model = "claude-haiku-4-5"
        
        response = client.messages.create(
            model=self.model,
            max_tokens=min(2048, self.max_tokens - self.tokens_used),
            messages=[{"role": "user", "content": task}]
        )
        
        self.tokens_used += response.usage.input_tokens + response.usage.output_tokens
        return response.content[0].text

デッドロック防止

エージェント間の循環依存はデッドロックを引き起こします。タイムアウトとサーキットブレーカーを組み合わせることで防止できます。

import asyncio
 
async def execute_with_timeout(agent, task: str, timeout: float = 30.0) -> str:
    """タイムアウト付きエージェント実行"""
    try:
        result = await asyncio.wait_for(
            asyncio.to_thread(agent.execute, task),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        return f"ERROR: Agent timed out after {timeout}s"

観測可能性の確保

分散システムでは、全体の動作を把握するための観測可能性が不可欠です。

各エージェントの呼び出しをトレースし、OpenTelemetry などのツールで可視化することで、ボトルネックや異常を迅速に発見できます。エージェントIDとワークフローIDを関連付けた構造化ログを出力することで、問題発生時のトレースが容易になります。

セキュリティとガバナンス

マルチエージェントシステムは、セキュリティリスクも単一エージェントより複雑になります。

エージェントID管理と認証

各エージェントに一意のIDを割り当て、他のエージェントからのリクエストを検証します。信頼できるエージェントからのみの指示を受け付けることで、なりすましを防ぎます。

権限スコープの分離

各エージェントは自分の役割に必要なツールのみにアクセスできるようにします。書き込み権限を持つエージェントと読み取り専用エージェントを明確に分ける点が肝心です。

監査ログ

すべてのエージェント間通信と外部ツール呼び出しをログに記録します。これにより、問題発生時の原因追跡と、規制要件への準拠が可能になります。

個人開発でオーケストレーションを一度やめた話

壁紙アプリのメタデータ整備で、私は一度この構成を解体しています。1枚の画像に対して「カテゴリ分類」「説明文の生成」「タグ付け」の3工程があり、素直に3体のエージェントへ割り当てました。役割が分かれているのだから分けるべきだ、という判断です。

結果として増えたのは API 呼び出し回数と、失敗の組み合わせパターンでした。分類は落ちたのに説明文だけ生成されている、タグだけが前回の内容のまま残っている——そうした中途半端な状態を後から解きほぐす処理のほうが、本体より複雑になっていきました。

いま残しているのは分類の1工程だけです。説明文とタグは分類結果から決定的に組み立てる処理に戻しました。判断が必要なのは「この画像が何であるか」だけで、そこから先は分岐表で足りる、と気づいたからです。

エージェントに向くのは、入力が曖昧で判断を要する工程です。逆に、入力さえ決まれば出力が一意に定まる工程をエージェント化すると、非決定性と失敗モードだけが増えます。パターン選定の前に「この工程は本当に判断を要するか」を1つずつ潰していくと、必要なエージェント数はたいてい想定より少なくなります。

個人開発では、落ちたときに駆けつける当番が自分しかいません。エージェントを1体増やす判断は、監視対象と失敗パターンを1つ増やす判断でもある、と考えるようにしています。

設計原則として手元に残ったもの

ここまでの実装と失敗から、繰り返し効いた判断基準を挙げます。

シンプルから始める: 最初から複雑なマルチエージェント構成を選ばないでください。単一エージェントで解決できることを確認してから、必要に応じて複雑さを追加します。

役割の明確化: 各エージェントの責任範囲を明確に定義し、重複を最小化します。「1エージェント1責任」の原則が保守性を高めます。

観測可能性を最初から設計する: ログ、メトリクス、トレーシングは後付けではなく、設計段階から組み込みます。

失敗を前提に設計する: エージェントは失敗します。リトライ、フォールバック、サーキットブレーカーを最初から実装し、一部の失敗がシステム全体を止めない設計にします。

コスト意識を持つ: 各エージェント、各API呼び出しのコストを追跡し、不要な呼び出しを削減します。モデルの使い分けとプロンプトキャッシングを積極的に活用します。

次の一手として提案したいのは、いま動かしているワークフローの並列実行部分を開いて、future.result() がどう扱われているかを1行だけ確認することです。例外がその場で送出される書き方になっていれば、成功分を失っている可能性があります。as_completed と try/except への置き換えは十数行で終わり、効果はすぐに測れます。

私自身もこの記事のコードで同じ穴を残していました。設計論より先に、境界の数行を疑うほうが早いことがあります。お読みいただきありがとうございました。