マルチエージェントシステム

動機：なぜ複数のエージェントなのか

人工知能の歴史は、多くの意味でスケールの歴史です。初期のAIシステムはモノリシックでした。単一のプログラム、単一の知識ベース、単一の推論エンジンです。問題が複雑になるにつれ、研究者は、どれほど能力が高くても、単一のエージェントでは豊かでオープンエンドなタスクのすべてを効率よく扱えないことを発見しました。分散AIとマルチエージェントシステム（MAS）研究で長く確立されてきたこの洞察（Weiss 1999、Wooldridge 2009）は、大規模言語モデルの時代に再び切迫した課題となっています。

Tip

中核となる直観

どれほど大きなLLMでも、単一のLLMは汎用家です。狭いサブ問題に集中し、結果を伝え合う専門LLMのチームは、複雑で多面的なタスクにおいて汎用家を上回れます。複雑なエンジニアリングプロジェクトで人間の専門家チームが単一の汎用家を上回るのと同じです。

モノリシックなエージェントからエージェント社会への移行を促す、4つの根本的な動機があります。

専門化

異なるサブタスクは、異なる能力、プロンプト戦略、さらには異なるベースモデルから恩恵を受けます。コード生成エージェントはプログラミングコーパスでファインチューニングでき、ファクトチェックエージェントは検索ツールでグラウンディングでき、創作エージェントは文体の多様性を促すプロンプトを使えます。単一のエージェントにこれらすべてを同時に得意にさせるのは非効率で、多くの場合不可能です。

並列化

現実世界の多くのタスクは、同時実行できる独立したサブタスクへ分解できます。文献レビュー、データ分析、レポート執筆を必要とする研究パイプラインは、3つすべてを並列に実行して実時間を大幅に短縮できます。単一エージェントによる逐次処理はボトルネックであり、マルチエージェント並列化が解消します。

堅牢性

単一エージェントは単一障害点です。幻覚を起こしたり、ループに陥ったり、微妙に誤った答えを出したりしても、チェックがありません。マルチエージェントシステムは冗長性を導入します。第2のエージェントが結果を検証、批評、独立再導出できます。敵対的エージェントは出力を信頼する前に弱点を探れます。

創発的能力

最も興味深いのは、エージェントの集合が、個々のエージェントにはない能力を示し得ることです。討論、ネゴシエーション、反復的な改善を通じて、マルチエージェントシステムは単一エージェントだけでは生み出せない解に到達できます。これは社会的な生物における創発知能の計算論的類推です。

Important

歴史的背景

マルチエージェントシステム研究は1980年代にさかのぼります。分散問題解決（Durfee et al. 1989）、Contract Net Protocol（Smith 1980）、FIPAエージェント通信標準（Intelligent Physical Agents 2002）に関する基礎研究がありました。LLMベースのエージェントへの移行は、これらの古典的な考え方を新しい基盤で再生させます。記号推論を行う手書きのエージェントに代わり、現在は学習されたニューラル表現から「認知」が創発するエージェントを使います。中核的なアーキテクチャパターン、すなわち階層、マーケット、ブラックボード、メッセージパッシングは、今も驚くほど関連性があります。

モノリシックなエージェントからエージェント社会への移行は、複雑系におけるより広いパターンを反映します。問題空間が大きくなるほど、分散・モジュール型アーキテクチャは中央集権的でモノリシックなアーキテクチャを一貫して上回ります。問題はもはや複数エージェントを使うかどうかではなく、どのように組織するかです。

マルチエージェントアーキテクチャ

マルチエージェントシステムのトポロジー、すなわちエージェントの接続方法と権限の流れは、最も影響の大きいアーキテクチャ上の意思決定です。それぞれ異なるトレードオフを持つ、4つの典型パターンが登場しています。

中央集権型（スーパーバイザー／マネージャー）アーキテクチャ

中央集権型アーキテクチャでは、単一のオーケストレータエージェント（スーパーバイザー、マネージャー、プランナーなどと呼ばれる）がグローバル状態を保持し、タスクを分解し、サブタスクをワーカーエージェントへ委譲し、結果を集約します。トポロジーは ハブ・アンド・スポーク で、すべての通信が中央ノードを通ります。

マネージャーの責務は次のとおりです。

• タスクルーティング: 各サブタスクに最適なワーカーを決定します。

• コンテキスト管理: 各ワーカーにグローバルコンテキストの関連部分を提供します。

• 結果の集約: ワーカーの出力を一貫した全体へ統合します。

• エラー処理: ワーカーの失敗を検出し、再ルーティングまたは再試行します。

Note

LangGraphにおけるスーパーバイザーパターン

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal

class TeamState(TypedDict):
    task: str
    plan: str
    code: str
    tests: str
    review: str
    next_agent: str
    final_output: str

def supervisor_node(state: TeamState) -> TeamState:
    """Central orchestrator: decides which agent to invoke next."""
    messages = [
        {"role": "system", "content": SUPERVISOR_PROMPT},
        {"role": "user",   "content": f"Task: {state['task']}\n"
                                      f"Plan: {state.get('plan','')}\n"
                                      f"Code: {state.get('code','')}\n"
                                      f"Tests: {state.get('tests','')}\n"
                                      "Which agent should act next? "
                                      "Options: planner, coder, tester, reviewer, FINISH"}
    ]
    response = llm.invoke(messages)
    return {**state, "next_agent": response.content.strip()}

def route(state: TeamState) -> Literal["planner","coder","tester","reviewer","__end__"]:
    return state["next_agent"] if state["next_agent"] != "FINISH" else END

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("planner",    planner_node)
builder.add_node("coder",      coder_node)
builder.add_node("tester",     tester_node)
builder.add_node("reviewer",   reviewer_node)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route)
for agent in ["planner", "coder", "tester", "reviewer"]:
    builder.add_edge(agent, "supervisor")   # always return to supervisor

graph = builder.compile()

Warning

中央集権型アーキテクチャのトレードオフ

長所: 制御フローが単純で、説明責任が明確、デバッグが容易（すべての意思決定が1か所にある）、実装が容易です。
短所: 単一障害点です。マネージャーが幻覚を起こしたり混乱したりするとシステム全体が失敗し、高負荷ではマネージャーがボトルネックになります。マネージャーのコンテキストウィンドウがグローバル状態を保持しなければならず、スケーラビリティが制限されます。

分散型（ピアツーピア）アーキテクチャ

分散型アーキテクチャでは、中央コーディネータなしにエージェント同士が直接相互作用します。トポロジーは メッシュ で、任意のエージェントが他の任意のエージェントと通信できます。調整はグローバルな計画ではなく、局所的な相互作用から創発します。

ピアツーピアシステムにおける創発的な調整は、次のような仕組みから生じます。

• ネゴシエーション: エージェントがタスクやリソースに入札します。

• スティグマジー: エージェントが共有状態を変更し、他のエージェントがそれを観測します（節10.3.6を参照）。

• ゴシッププロトコル: エージェントがネットワーク全体に情報を伝播します。

• 局所合意: 小さなエージェント集団がグローバルな調整なしに合意へ到達します。

Warning

分散型アーキテクチャのトレードオフ

長所: 個々のエージェントの失敗に強く、エージェントを追加すると自然にスケールし、ボトルネックがありません。
短所: デバッグが難しく、創発的な挙動を追跡しにくい、エージェントが状態について一貫しない見解を持つと競合が起こり得る、単純なメッセージパッシングでは調整オーバーヘッドが \(O(n^2)\) で増大する、グローバルな一貫性の保証が難しい、という問題があります。

階層型アーキテクチャ

階層型アーキテクチャは、中央集権型パターンを複数の管理レベルを持つ 木構造 へ一般化します。トップレベルのオーケストレータが分野固有のサブマネージャーへ委譲し、サブマネージャーが専門ワーカーへ委譲します。これは大企業の組織構造を反映します。

階層型システムの主な特徴は次のとおりです。

• 委譲チェーン: 権限とコンテキストが木を下へ流れ、結果が上へ流れます。

• エスカレーション経路: ワーカーは解決できない問題をマネージャーへエスカレーションできます。

• 分野分離: サブマネージャーが分野固有のコンテキストを維持し、トップレベルオーケストレータの認知負荷を減らします。

• 範囲の限定: 各エージェントは直属の上位者と下位者だけを知ればよい構造です。

企業との類推が適切です。CEO（トップオーケストレータ）が戦略を定め、VP（サブマネージャー）が戦略を分野計画へ変換し、個々の貢献者（ワーカー）が実行します。階層は説明責任を保ちながらスケールを可能にします。

スウォームアーキテクチャ

スウォームアーキテクチャは、生物システム（アリのコロニー、鳥の群れ）に着想を得たもので、多数の 疎結合エージェント が単純な局所ルールに従い、中央コーディネータやグローバル状態なしに複雑なグローバル挙動を生み出します。

OpenAIの Swarm フレームワーク（OpenAI 2024b）は、2つのプリミティブでこれを実装します（現在はOpenAI Agents SDKに置き換えられましたが、その概念的プリミティブは今も影響力を持ちます）。

• ルーチン: エージェントがサブタスクを完了するために従う指示の列

• ハンドオフ: エージェントが制御（および関連コンテキスト）を別のエージェントへ移すこと

Note

OpenAI Swarm：ルーチンとハンドオフ

from swarm import Swarm, Agent

client = Swarm()

def transfer_to_billing():
    """Handoff: transfer control to the billing specialist."""
    return billing_agent

def transfer_to_technical():
    """Handoff: transfer control to the technical support agent."""
    return technical_agent

triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are a customer service triage agent.
    Determine the nature of the customer's issue:
    - For billing questions, transfer to billing.
    - For technical issues, transfer to technical support.
    - For general questions, answer directly.""",
    functions=[transfer_to_billing, transfer_to_technical],
)

billing_agent = Agent(
    name="Billing Specialist",
    instructions="You handle billing inquiries. "
                 "Access account data and resolve payment issues.",
    functions=[lookup_account, process_refund],
)

technical_agent = Agent(
    name="Technical Support",
    instructions="You resolve technical issues. "
                 "Diagnose problems and provide step-by-step solutions.",
    functions=[run_diagnostics, escalate_to_engineering],
)

# No global state --- each agent operates on its local context
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "My invoice is wrong"}]
)

Important

スウォームの特性

• グローバル状態なし: 各エージェントはローカルのコンテキストウィンドウだけを維持します。

• 局所的な意思決定: ルーティング判断は中央プランナーではなく、現在のエージェントが行います。

• 集合的な挙動によるタスク完了: 複雑なタスクはハンドオフの連鎖で完了し、各エージェントが専門性を提供します。

• 軽量: オーケストレーションのオーバーヘッドがなく、エージェントはハンドオフ間でステートレスです。

調整メカニズム

エージェントがどのように調整するか、すなわち情報共有、作業分担、競合解決の方法は、トポロジーと同じくらい重要です。LLMベースのマルチエージェントシステムには、6つの典型的な調整メカニズムがあります。

共有状態（グローバルブラックボード）

ブラックボードアーキテクチャ （Hayes-Roth 1985）は、すべてのエージェントが読み書きできる共有データ構造を提供します。LLMシステムでは通常、共有辞書、データベース、構造化文書として実装します。

import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List

@dataclass
class BlackboardEntry:
    value: Any
    author: str
    timestamp: float
    confidence: float = 1.0

class Blackboard:
    """Thread-safe shared state for multi-agent coordination."""

    def __init__(self):
        self._data: Dict[str, BlackboardEntry] = {}
        self._lock = threading.RLock()
        self._subscribers: Dict[str, List[Callable]] = {}

    def write(self, key: str, value: Any, author: str,
              confidence: float = 1.0) -> bool:
        """Write to blackboard; higher-confidence entries win conflicts."""
        with self._lock:
            existing = self._data.get(key)
            if existing and existing.confidence > confidence:
                return False  # Conflict: existing entry wins
            import time
            self._data[key] = BlackboardEntry(
                value=value, author=author,
                timestamp=time.time(), confidence=confidence
            )
            self._notify(key, value)
            return True

    def read(self, key: str) -> Any:
        with self._lock:
            entry = self._data.get(key)
            return entry.value if entry else None

    def subscribe(self, key: str, callback: Callable):
        """Agents subscribe to changes on specific keys."""
        self._subscribers.setdefault(key, []).append(callback)

    def _notify(self, key: str, value: Any):
        for cb in self._subscribers.get(key, []):
            cb(key, value)

メッセージパッシング

メッセージパッシングはLLMエージェントにとって最も自然な調整メカニズムです。エージェントが構造化テキストメッセージを互いに送り、通信します。主な設計判断は次のとおりです。

• メッセージ形式: 構造化（JSONスキーマ）、自然言語、ハイブリッドのどれにするか。

• ルーティング: 直接（エージェント間）、ブロードキャスト、トピックベースのPub/Subのどれにするか。

• 会話スレッド: マルチターン交換をまたいでコンテキストを維持すること。

• 確認応答: 送信者が受信・処理の確認を必要とするかどうか。

計画と分解

マネージャーエージェントが高レベルタスクをサブタスクの 有向非巡回グラフ（DAG） へ分解し、それぞれを適切なワーカーに割り当て、依存関係を追跡します。これは古典的な階層型タスクネットワーク（HTN）計画のマルチエージェント版です。

from dataclasses import dataclass, field
from typing import List, Optional
import asyncio

@dataclass
class Task:
    id: str
    description: str
    assigned_to: str
    dependencies: List[str] = field(default_factory=list)
    status: str = "pending"   # pending | running | done | failed
    result: Optional[str] = None

class TaskDAG:
    def __init__(self):
        self.tasks: dict[str, Task] = {}

    def add_task(self, task: Task):
        self.tasks[task.id] = task

    def ready_tasks(self) -> List[Task]:
        """Return tasks whose dependencies are all completed."""
        return [
            t for t in self.tasks.values()
            if t.status == "pending"
            and all(self.tasks[d].status == "done"
                    for d in t.dependencies)
        ]

    async def execute(self, agent_pool: dict):
        while any(t.status != "done" for t in self.tasks.values()):
            ready = self.ready_tasks()
            if not ready:
                await asyncio.sleep(0.1)
                continue
            # Execute ready tasks in parallel
            await asyncio.gather(*[
                self._run_task(t, agent_pool[t.assigned_to])
                for t in ready
            ])

    async def _run_task(self, task: Task, agent):
        task.status = "running"
        try:
            task.result = await agent.execute(task.description)
            task.status = "done"
        except Exception as e:
            task.status = "failed"
            raise

投票と合意形成

複数エージェントが競合する出力を生成した場合、投票メカニズムが応答を単一の意思決定へ集約します。一般的な方式は次のとおりです。

• 多数決: 最も多い回答が勝ち、事実に関する質問に有効です。

• 重み付き投票: 実績や信頼度スコアの高いエージェントに大きな重みを与えます。

• 討論ベースの解決: エージェントが自分の立場を主張し、審判エージェントが決定します。

• デルファイ法: 他者の推論を見た後に回答を修正するラウンドを反復します。

形式的には、\(n\) 個のエージェントが出力 \(\{o_1, \ldots, o_n\}\) を生成し、重み \(\{w_1, \ldots, w_n\}\) を持つとき、重み付き合意は次のようになります。

\[ o^* = \arg\max_{o} \sum_{i=1}^{n} w_i \cdot \mathbf{1}[o_i = o] \]

連続出力（例: 確率推定）には、重み付き平均を適用します。

\[ \hat{p} = \frac{\sum_{i=1}^{n} w_i \cdot p_i}{\sum_{i=1}^{n} w_i} \]

市場ベースの調整

市場メカニズムは、 オークションと入札 を通じてタスクとリソースを割り当てます。最古のマルチエージェント調整メカニズムの1つであるContract Net Protocol（Smith 1980）は、タスクオークションです。

1. マネージャーが要件付きのタスク告知をブロードキャストします。

2. 請負エージェントが入札（能力宣言＋コスト見積もり）を提出します。

3. マネージャーが最良の入札者へ契約を割り当てます。

4. 落札した請負エージェントが実行し、結果を報告します。

LLMシステムでは、入札を自然言語（「高い信頼度で3ステップで完了できます」）または構造化形式で表現できます。市場メカニズムは、APIコストを最小化する必要があるリソース制約環境で特に有効です。

スティグマジー：環境を介した間接通信

スティグマジー （Grassé 1959）は、明示的なエージェント間メッセージをより単純な仕組みに置き換えます。各エージェントが作業の副作用として共有環境を変更し、他のエージェントが直接の信号ではなく変更へ反応します。典型例は、採餌するアリが帰路にフェロモンを残し、アリ同士が「会話」しなくても後続のアリが成功経路を強化することです。

LLMマルチエージェントシステムでは、スティグマジーは次の形で現れます。

• 共有文書: エージェントが共有文書へ書き込み、他のエージェントが読み取って発展させます。

• コードリポジトリ: 1つのエージェントがコードをコミットし、別のエージェントが読み取って拡張します。

• アノテーション層: エージェントが共有成果物に注釈を付けます（エラーの強調、コメントの追加）。

• タスクキュー: エージェントが共有キューへタスクを追加し、そこから取り出します。

スティグマジーは明示的な通信オーバーヘッドなしに調整を可能にします。エージェントは共有環境の状態を観測し、それに応じて行動するだけです。

通信プロトコル

効果的なマルチエージェントシステムには、明確に定義された通信プロトコルが必要です。エージェント間メッセージの形式、意味論、パターンについて合意します（標準化されたエージェント間プロトコルについては、章9を参照）。

構造化メッセージ形式

LLMエージェント間のメッセージは、確実なパースとルーティングを可能にするため構造化すべきです。最小限のメッセージスキーマを示します。

from pydantic import BaseModel, Field
from typing import Literal, Optional, Dict, Any
from datetime import datetime, timezone
import uuid

PerformativeType = Literal[
    "inform",    # Share information
    "request",   # Request an action
    "propose",   # Propose a course of action
    "accept",    # Accept a proposal
    "reject",    # Reject a proposal
    "query",     # Ask a question
    "confirm",   # Confirm receipt/completion
    "failure",   # Report a failure
]

class AgentMessage(BaseModel):
    message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    conversation_id: str          # Groups related messages
    sender: str                   # Agent identifier
    receiver: str                 # Target agent (or "broadcast")
    performative: PerformativeType
    content: str                  # Natural language content
    metadata: Dict[str, Any] = {} # Structured payload
    reply_to: Optional[str] = None  # message_id being replied to
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    def to_llm_prompt(self) -> str:
        """Render message as a prompt fragment for the receiving agent."""
        return (
            f"[MESSAGE from {self.sender}]\n"
            f"Type: {self.performative}\n"
            f"Content: {self.content}\n"
            + (f"Metadata: {self.metadata}\n" if self.metadata else "")
        )

発話行為タイプ（FIPA-ACLに着想）

FIPA Agent Communication Language（Intelligent Physical Agents 2002）を基に、LLMエージェント向けに現代化したものです。

LLMエージェントメッセージ向けのFIPA-ACL着想の発話行為タイプ。

コンテキスト共有戦略

マルチエージェント通信の重要な課題は コンテキスト管理 です。各エージェントにどれだけの履歴が必要でしょうか。戦略は3つあります。

• 全履歴: 会話履歴全体を各エージェントへ渡します。情報量は最大ですが高コストで、コンテキストウィンドウがすぐに埋まります。

• 要約: 要約エージェントが過去の交換を簡潔な要約へ圧縮します。効率的ですが情報損失があり、重要な詳細が落ちる可能性があります。

• 関連抜粋: 意味検索を使って、過去の最も関連するメッセージだけを取得します。コストと情報量のバランスが取れますが、検索メカニズムが必要です。

Important

コンテキスト共有の経験則

短い会話（\(<\)10ターン）には 全履歴 、中程度の会話には 要約 、長時間実行されるエージェントセッションには 検索拡張抜粋 を使います。直近のコンテキストを保つため、最新の \(k\) メッセージは常に逐語的に含めます。

役割設計と専門化

エージェントの役割、すなわち能力、ペルソナ、責務の設計は、科学であると同時に芸術でもあります。適切に設計された役割は専門化を可能にし、設計の悪い役割は混乱と冗長性を生みます。

エージェントの役割を定義する

LLMマルチエージェントシステムにおける一般的な役割は次のとおりです。

LLMマルチエージェントシステムにおける一般的なエージェントの役割。

能力ベース対役割ベースの割り当て

タスク割り当てには2つの考え方があります。

• 役割ベース: 事前定義された役割ラベルに基づいてタスクを割り当てます。単純で予測可能ですが、タスクが複数の役割にまたがる場合は最適でない可能性があります。

• 能力ベース: タスク要件に対する各エージェントの能力を動的に評価してタスクを割り当てます。より柔軟ですが、能力レジストリとマッチングメカニズムが必要です。

動的な役割再割り当て

長時間稼働するシステムでは、静的な役割割り当てが最適でなくなります。動的な再割り当てにより、次に基づいてエージェントが新しい役割を担えるようになります。

• 現在のワークロード（負荷分散）

• 直近タスクで実証された性能

• 変化するタスク要件

• カバーが必要なエージェントの失敗

思考の多様性のためのペルソナ設計

微妙ですが強力な手法として、多様な視点を促す 異なるペルソナ をエージェントに与えます。5つの同一な「アシスタント」エージェントではなく、次のように設計します。

• 機会を重視する楽観主義者

• 前提に疑問を投げかける懐疑主義者

• 実装に焦点を当てる実用主義者

• 長期的に考える先見者

• 反対の立場を主張する悪魔の代弁者

この思考の多様性は、Six Thinking Hats（6色ハット法）(Bono 1985)のような手法に着想を得たもので、集団浅慮を減らし、より堅牢な集団推論を生み出します。

Warning

役割の衝突の解決

エージェントの責任範囲が重複すると、衝突が生じます。明示的な 優先順位ルール で解決します。つまり、タスク種別ごとにどの役割を優先するかを定義します。あるいは、衝突の調停だけを担当する メタエージェント を使う方法もあります。役割の衝突を暗黙のままにしてはいけません。矛盾した出力や無限ループとして現れることになります。

LLMのマルチエージェントパターン

アーキテクチャのトポロジーに加えて、いくつかの 相互作用パターン がLLMベースのマルチエージェントシステムで特に有効であることが分かっています。（これらは、第5章で扱った単一エージェントの設計パターンを補完します。）

ディベートパターン

複数のエージェントが異なる立場を主張し、判定エージェントが議論を評価して決定します。ディベートによって事実の正確性が向上し、ハルシネーションが減少することが示されています（Du et al. 2023）。

async def debate_round(question: str, agents: list, judge: Agent,
                       rounds: int = 2) -> str:
    """Run a multi-agent debate and return the judge's verdict."""
    positions = {a.name: await a.generate_position(question)
                 for a in agents}

    for round_num in range(rounds):
        # Each agent sees others' positions and can rebut
        rebuttals = {}
        for agent in agents:
            others = {k: v for k, v in positions.items()
                      if k != agent.name}
            rebuttals[agent.name] = await agent.rebut(
                question, positions[agent.name], others
            )
        positions = rebuttals

    # Judge evaluates all final positions
    verdict = await judge.evaluate(question, positions)
    return verdict

リフレクションパターン

1つのエージェントが出力を生成し、2つ目のエージェントがそれを批評し、最初のエージェントが批評に基づいて修正します。これは、品質を反復的に向上させる生成・批評・修正のループを実装します。

async def reflection_loop(task: str, generator: Agent,
                          critic: Agent, max_rounds: int = 3) -> str:
    draft = await generator.generate(task)

    for _ in range(max_rounds):
        critique = await critic.critique(task, draft)
        if critique.is_satisfactory:
            break
        draft = await generator.revise(task, draft, critique.feedback)

    return draft

分業パターン

タスクを独立したサブタスクに分解し、並列に実行します。結果は統合エージェントが集約します。このパターンは、容易に並列化できるタスクのスループットを最大化します。

パイプラインパターン

エージェントが逐次処理チェーンを形成し、各エージェントが前のエージェントの出力を変換します。Unixのパイプに似ています。明確な逐次依存関係を持つタスク（例: 調査 \(\to\) アウトライン \(\to\) 下書き \(\to\) 編集 \(\to\) 書式設定）に有効です。

アンサンブルパターン

複数のエージェントが同じ問題を独立に解き、選択メカニズムが最良の回答（best-of-\(N\)）を選ぶか、回答を集約します（mixture-of-experts方式）。計算量を犠牲にして信頼性を向上させます。

\[ o^* = \arg\max_{o \in \{o_1,\ldots,o_N\}} \text{score}(o, \text{task}) \]

ここで \(\text{score}\) は、報酬モデル、判定用LLM、検証器のいずれかです。

教師・生徒パターン

より能力の高いエージェント（教師）が、ヒント、修正、説明を提供しながら、能力の低いエージェント（生徒）をタスクに導きます。このパターンにより推論時の知識蒸留が可能になり、生徒エージェントのファインチューニングにも利用できます。

レッドチームパターン

敵対的エージェント（レッドチーム）が、他のエージェントの出力にある弱点、誤り、安全性違反を積極的に探します。レッドチームエージェントには、攻撃において最大限に批判的かつ創造的になるようプロンプトを与えます。このパターンは、安全性が重要なアプリケーションに不可欠です。

Note

レッドチームエージェントのプロンプト

RED_TEAM_PROMPT = """You are a red team agent. Your job is to find
flaws, errors, biases, safety violations, and failure modes in the
following output. Be adversarial, creative, and thorough.

Consider:
1. Factual errors or hallucinations
2. Logical inconsistencies
3. Safety and ethical concerns
4. Edge cases the solution doesn't handle
5. Ways a malicious user could exploit this output
6. Unintended consequences

Output: {agent_output}

Provide a detailed critique with specific examples of each flaw found."""

強化学習によるマルチエージェントシステムの訓練

RLでマルチエージェントシステムを訓練すると、単一エージェントのRLを超える課題が生じます。根本的な難しさは、各エージェントの環境に他の学習エージェントが含まれるため、個々のエージェントから見た環境が 非定常 になることです。

数学的定式化

マルチエージェントシステムは マルコフゲーム （確率ゲームとも呼ばれます）として定式化されます（Shapley 1953）。

\[ \mathcal{G} = \langle \mathcal{N}, \mathcal{S}, \{\mathcal{A}^i\}_{i \in \mathcal{N}}, \mathcal{T}, \{R^i\}_{i \in \mathcal{N}}, \gamma \rangle \]

ここで \(\mathcal{N} = \{1, \ldots, n\}\) はエージェント集合、\(\mathcal{S}\) は共有状態空間、\(\mathcal{A}^i\) はエージェント \(i\) の行動空間、\(\mathcal{T}: \mathcal{S} \times \mathcal{A}^1 \times \cdots \times \mathcal{A}^n \to \Delta(\mathcal{S})\) は遷移関数、\(R^i: \mathcal{S} \times \mathcal{A}^1 \times \cdots \times \mathcal{A}^n \to \mathbb{R}\) はエージェント \(i\) の報酬関数、\(\gamma\) は割引係数です。

各エージェント \(i\) は、期待割引収益の最大化を目指します。

\[ J^i(\pi^1, \ldots, \pi^n) = \mathbb{E}_{\pi^1,\ldots,\pi^n}\left[\sum_{t=0}^{\infty} \gamma^t R^i(s_t, a_t^1, \ldots, a_t^n)\right] \]

独立学習

最も単純な方法は、各エージェント \(i\) が他のエージェントを自分の環境の一部として扱い、標準的な単一エージェントRL（例: PPO、REINFORCE）を使って自身の方策 \(\pi^i\) を独立に最適化することです。

\[ \nabla_{\theta^i} J^i \approx \mathbb{E}\left[\nabla_{\theta^i} \log \pi^i(a^i_t | o^i_t) \cdot \hat{A}^i_t\right] \]

Warning

非定常性の問題

独立学習はマルコフ仮定に違反します。他のエージェントが方策を更新すると、エージェント \(i\) が見る遷移分布と報酬分布が変化するためです。これにより、訓練が不安定になったり、振動したり、収束に失敗したりする可能性があります。独立学習は単純な協調タスクでは実際に機能しますが、競争的な環境や複雑な協調環境では苦戦します。

集中訓練・分散実行（CTDE）

CTDE（Lowe et al. 2017; Rashid et al. 2018）は、協調型マルチエージェントRLの主要なパラダイムです。訓練中は、集中型クリティックがグローバル状態 \(s\) と全エージェントの行動 \(\mathbf{a} = (a^1, \ldots, a^n)\) にアクセスできます。実行時には、各エージェントは自身の局所観測 \(o^i\) だけを使って行動します。

エージェント \(i\) の集中型クリティック:

\[ Q^i_\phi(s, \mathbf{a}) = Q^i_\phi(s, a^1, \ldots, a^n) \]

エージェント \(i\) の分散型アクター:

\[ \pi^i_{\theta^i}(a^i | o^i) \]

集中型クリティックを使った方策勾配:

\[ \nabla_{\theta^i} J^i = \mathbb{E}\left[\nabla_{\theta^i} \log \pi^i(a^i | o^i) \cdot Q^i_\phi(s, \mathbf{a})\right] \]

CTDEは訓練中の非定常性を解消し（集中型クリティックが完全な共同状態を見る）、分散実行（推論時の通信が不要）を維持します。

通信学習

固定された通信プロトコルを使うのではなく、エージェントに 何を伝えるかを学習 させることができます。微分可能な通信フレームワーク（Sukhbaatar et al. 2016; Das et al. 2019）では、エージェントが連続通信ベクトル \(m^i_t\) を生成し、他のエージェントに渡します。

\[ a^i_t, m^i_t = \pi^i_{\theta^i}(o^i_t, \{m^j_{t-1}\}_{j \neq i}) \]

通信ベクトルは、共同報酬信号を通じて逆伝播するエンドツーエンド方式で最適化されます。LLMエージェントでは、タスク性能を最大化する構造化された自然言語メッセージを生成するよう訓練することで、これを近似します。

創発的通信

エージェントを報酬信号だけでゼロから訓練すると（事前定義された言語なしに）、 創発的通信プロトコル （Lazaridou and Baroni 2020）を発達させることがあります。これはタスクに関連する情報を符号化する共有記号体系です。科学的には興味深い一方で、LLMシステムにおける創発的通信は通常望ましくありません。エージェントには人間が解釈できる言語で通信してほしいからです。

自己対戦

競争的な環境や動機が混在する環境では、 自己対戦 （Silver et al. 2017）によって、エージェントを自分自身のコピーと競わせて訓練します。これにより自動カリキュラムが生成されます。エージェントが改善するにつれて、対戦相手（自分自身の以前のバージョン）に勝つことが難しくなります。

LLMエージェントでは、自己対戦は次の用途に使われます。

• レッドチーム対ブルーチームの訓練

• ディベート訓練（エージェント同士が互いに議論する）

• 交渉訓練（エージェント同士が交渉する）

集団ベース訓練

集団ベース訓練（PBT） （Jaderberg et al. 2019）は、異なる方策、ハイパーパラメータ、専門性を持つ多様なエージェント集団を維持します。エージェントは定期的に評価され、性能の低いエージェントは高性能エージェントの変異コピーに置き換えられます。

マルチエージェントLLMシステムでは、PBTによって次が可能になります。

• 効果的な役割の専門化の自動発見

• 個々のエージェントの失敗に対する頑健性（多様な集団）

• 集団の多様性による局所最適解の回避

社会的厚生とナッシュ均衡

マルチエージェント環境では、最適性の概念が単一エージェント環境より複雑になります。重要な解概念は2つあります。

ナッシュ均衡: どのエージェントも単独で逸脱して期待収益を改善できない共同方策 \((\pi^{1*}, \ldots, \pi^{n*})\):

\[ J^i(\pi^{i*}, \pi^{-i*}) \geq J^i(\pi^i, \pi^{-i*}) \quad \forall i, \forall \pi^i \]

ここで \(\pi^{-i}\) は、\(i\) 以外の全エージェントの共同方策を表します。

社会的厚生の最大化: 全エージェントの収益の合計を最適化します。

\[ \max_{\pi^1,\ldots,\pi^n} \sum_{i=1}^{n} J^i(\pi^1, \ldots, \pi^n) \]

完全に協調的な環境（全エージェントが同じ報酬を共有する）では、社会的厚生の最大化が適切な目的です。競争的な環境では、ナッシュ均衡が適切な解概念です。現実のマルチエージェントLLMシステムの多くは 動機が混在 しています。エージェントの目的は一部が整合し、一部が対立します。

Important

さらに学ぶ: マルチエージェントRLのゲーム理論

マルチエージェントシステムのゲーム理論的基礎に関心のある読者には、次を勧めます。

• Shoham & Leyton-Brown （Shoham and Leyton-Brown 2008）— ナッシュ均衡、メカニズムデザイン、エージェントシステムの社会的選択理論を扱う包括的な教科書。

• Zhang et al. （Zhang et al. 2021）— 協調的、競争的、混合的な環境で収束保証を持つマルチエージェントRLアルゴリズムのサーベイ。

• Nisan et al. （Nisan et al. 2007）— オークション、均衡計算、アナーキーの価格を扱うアルゴリズムゲーム理論の決定版。

課題と解決策

調整オーバーヘッド

エージェント間のすべてのメッセージはトークンを消費し、その結果として時間と費用も消費します。素朴な実装では、不要な場合でもエージェントが絶えず通信します。

Important

通信しない場合

• 情報がすでに共有ブラックボードにある場合

• 受信エージェントが現在のタスクにその情報を必要としない場合

• メッセージがすでに送信した情報と重複する場合

• タスクが単一エージェントで十分に処理できるほど単純な場合

ルール: 情報の期待価値がメッセージのコストを上回る場合にのみ通信する。

通信コストの定量化: メッセージが \(c\) トークンを消費し、受信エージェントのタスクの価値が \(v\) である場合、タスク価値の期待改善 \(\Delta v > c \cdot \text{cost_per_token}\) のときにのみ通信します。

冗長性と効率性

複数のエージェントが同じサブ問題を独立に解き、計算資源を浪費することがあります。解決策は次のとおりです。

• 重複検出: タスクを開始する前に、ブラックボードに既存の結果がないか確認する

• 結果のキャッシュ: 完了したサブタスクの結果を意味的なキーとともに保存し、取得できるようにする

• タスクロック: 重複実行を防ぐため、タスクを「進行中」としてマークする

帰属

マルチエージェントシステムが成功または失敗したとき、どのエージェントに責任があるのでしょうか。帰属は次の点で重要です。

• RLの報酬割り当て（信用割当問題）

• デバッグと改善

• 信頼度の調整（どのエージェントを頼るか）

反実仮想信用割当 の手法では、「このエージェントが異なる行動を取っていたら、結果はどれほど変わっていただろうか」と問い、各エージェントの貢献度を推定します。

\[ \text{credit}^i = J(\pi^1, \ldots, \pi^n) - J(\pi^1, \ldots, \pi^{i}_{\text{default}}, \ldots, \pi^n) \label{eq:counterfactual-credit} \]

スケーラビリティ

素朴なメッセージパッシングは、エージェント数に対して \(O(n^2)\) でスケールします。解決策は次のとおりです。

• 階層型通信: エージェントが自分のサブツリー内でのみ通信する

• トピックベースのPub/Sub: エージェントが関連するメッセージトピックだけを購読する

• 疎な通信グラフ: 相互作用が必要なエージェントだけを接続する

• 非同期通信: エージェントが応答を待ってブロックしない

創発的な挙動と安全性

マルチエージェントシステムは、予期しない創発的な挙動を示すことがあります。これは、個々のエージェントが生み出すよう設計されていない結果を、エージェント間の相互作用が生み出すということです。これは特徴（創発的能力）であると同時に、リスク（創発的な失敗）でもあります。

Warning

マルチエージェントシステムにおける安全性の懸念

• プロンプトインジェクションの連鎖: 1つのエージェントへの悪意ある入力がシステム全体に伝播する

• 報酬ハッキング: エージェントが意図に反する予期しない方法で報酬を最大化する

• 共謀: 競争的な環境で、エージェントが暗黙の共謀戦略を発達させることがある

• 増幅: 1つのエージェントの誤りや偏りが、下流のエージェントによって増幅される

すべてのエージェント間通信を監視し、安全でない挙動が検出された場合にシステムを停止できる 安全性監視エージェント を必ず含めてください。

評価

マルチエージェントシステムの評価には、複数のレベルの指標が必要です。

マルチエージェントシステムの多レベル評価指標。

実世界のマルチエージェントアプリケーション

ソフトウェア開発チーム

マルチエージェントのソフトウェア開発チームは、実際のエンジニアリング組織を模倣します。

from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class SoftwareTeamState:
    requirements: str
    architecture: Optional[str] = None
    code: Optional[str] = None
    tests: Optional[str] = None
    review_feedback: Optional[str] = None
    final_code: Optional[str] = None
    approved: bool = False

class SoftwareDevelopmentTeam:
    """
    Multi-agent software team:
      Architect -> Coder -> Tester -> Reviewer -> (iterate or ship)
    """

    def __init__(self, llm_factory):
        self.architect = llm_factory(
            system_prompt="""You are a software architect. Given requirements,
            produce a clear technical design: components, interfaces, data
            structures, and implementation plan."""
        )
        self.coder = llm_factory(
            system_prompt="""You are an expert software engineer. Given a
            technical design, write clean, well-documented, production-ready
            code. Follow best practices for the language."""
        )
        self.tester = llm_factory(
            system_prompt="""You are a QA engineer. Given code, write
            comprehensive tests: unit tests, edge cases, integration tests.
            Identify potential bugs and failure modes."""
        )
        self.reviewer = llm_factory(
            system_prompt="""You are a senior code reviewer. Evaluate code
            for correctness, security, performance, and maintainability.
            Provide specific, actionable feedback. Approve only if excellent."""
        )

    async def build(self, requirements: str,
                    max_iterations: int = 3) -> SoftwareTeamState:
        state = SoftwareTeamState(requirements=requirements)

        # Phase 1: Architecture
        state.architecture = await self.architect.invoke(
            f"Requirements:\n{requirements}\n\nProduce technical design."
        )

        for iteration in range(max_iterations):
            # Phase 2: Implementation
            prompt = (f"Design:\n{state.architecture}\n\n"
                      + (f"Previous feedback:\n{state.review_feedback}\n\n"
                         if state.review_feedback else "")
                      + "Write the implementation.")
            state.code = await self.coder.invoke(prompt)

            # Phase 3: Testing
            state.tests = await self.tester.invoke(
                f"Code:\n{state.code}\n\nWrite comprehensive tests."
            )

            # Phase 4: Review
            review = await self.reviewer.invoke(
                f"Code:\n{state.code}\n\nTests:\n{state.tests}\n\n"
                "Review this code. End with APPROVED or NEEDS_REVISION."
            )

            if "APPROVED" in review:
                state.final_code = state.code
                state.approved = True
                break
            else:
                state.review_feedback = review

        return state

    async def run(self, requirements: str) -> str:
        state = await self.build(requirements)
        if state.approved:
            return f"# Final Implementation\n\n{state.final_code}"
        else:
            return f"# Best Attempt (not approved)\n\n{state.code}"

研究チーム

研究チームのエージェント社会は、学術的な協働を模倣します。

• 文献レビュアー: 既存研究を検索し、統合する

• 仮説生成器: 新しい研究の方向性を提案する

• 実験担当者: 実験を設計し、実行する（コード実行による）

• 統計担当者: 結果を分析し、有意性を評価する

• ライター: 知見を一貫した報告書にまとめる

カスタマーサービスシステム

階層型のカスタマーサービスシステム:

• ルーター: 受信したリクエストを分類し、専門担当者に振り分ける

• 請求担当者: 支払いとアカウントの問題を処理する

• 技術担当者: 製品やサービスの問題を解決する

• エスカレーションエージェント: 人間の判断が必要な複雑なケースを処理する

クリエイティブチーム

クリエイティブ制作パイプライン:

• ブレインストーマー: 自己検閲せず多様なアイデアを生成する

• ドラフター: 最も有望なアイデアを完全な下書きに発展させる

• エディター: 明確さ、文体、一貫性のために下書きを洗練する

• 批評者: 成果物を強化するため敵対的なフィードバックを提供する

アーキテクチャの比較

Important

アーキテクチャの選択

• 独立したサブタスク \(\rightarrow\) 並列アーキテクチャ（分業、アンサンブル）。

• 明確な依存関係を持つ逐次処理 \(\rightarrow\) パイプライン。

• フォールトトレランスが必要 \(\rightarrow\) 集中型を避け、階層型または分散型を優先する。

• デバッグ可能性が重要 \(\rightarrow\) 集中型またはパイプライン（すべての意思決定を追跡可能）。

• \(<\)5エージェント \(\rightarrow\) 集中型が最も単純。 \(>\)20エージェント \(\rightarrow\) 階層型またはスウォーム。

実際には、ほとんどの本番システムが 階層型 アーキテクチャを使います。最上位のスーパーバイザーがドメイン固有のサブスーパーバイザーに委譲し、サブスーパーバイザーが専門ワーカーの小規模チームを管理します。

まとめ

マルチエージェントシステムは、LLMの配備方法における根本的な転換を示します。孤立したアシスタントから、専門エージェントが協働する社会へと移行するのです。この節の主な知見は次のとおりです。

Important

マルチエージェントシステム: 主な要点

1. アーキテクチャが重要: エージェント接続のトポロジーが、スケーラビリティ、デバッグ可能性、フォールトトレランスを決定します。タスク構造と運用要件に基づいて選択してください。

2. 調整にはコストがかかる: エージェント間のすべてのメッセージがトークンを消費します。必要な情報の流れを維持しながら、オーバーヘッドを最小化する通信プロトコルを設計してください。

3. 専門化が品質を可能にする: 焦点を絞った役割と調整されたプロンプトを持つエージェントは、複雑なタスクで汎用エージェントを一貫して上回ります。

4. RL訓練は難しい: マルチエージェントRLには、非定常性、信用割当の課題、創発的な挙動が伴います。CTDEは、協調的な環境における現在のベストプラクティスです。

5. 安全性には明示的な設計が必要: マルチエージェントシステムは誤りを増幅し、予期しない創発的な挙動を示すことがあります。安全性監視は、第一級のアーキテクチャ上の関心事でなければなりません。

6. 単純に始める: 集中型スーパーバイザーパターンから始め、その限界を測定し、必要な場合にのみより複雑なアーキテクチャへ発展させてください。

マルチエージェントLLMシステムの分野は急速に進化しています。ここで説明したパターンと技法は現在の最先端を表しますが、新しいアーキテクチャ、調整メカニズム、訓練アルゴリズムが絶えず登場しています。専門化、調整、創発的な挙動、効率性と頑健性の緊張関係という基礎原則は、具体的な実装がどのように進化しても、引き続き重要です。