一、モノリスから大規模モデルへ：なぜマルチエージェント協調が必須の課題なのか

モノリシックなLLMアプリケーションでは、1回のリクエストは1回のモデル呼び出しに依存し、コンテキストウィンドウがその「思考の境界」となります。しかし、実際のビジネスでは、複雑なタスクは検索、推論、コード実行、外部ツール呼び出しなど複数のサブ能力を必要とし、これらの能力には依存関係とフィードバックがあります。それらを無理に1つのプロンプトに詰め込むと、コンテキストウィンドウを超えやすくなるだけでなく、モデルが膨大な情報の中で迷い、推論品質が著しく低下します。マルチエージェントアーキテクチャの核心理念は、大きなタスクを複数のサブタスクに分解し、各サブタスクを専門のエージェントが担当し、明示的な通信プロトコルを通じて協調して完了することです。これは本質的に「分割して統治する」という工学的パラダイムです。

オーケストレーションから自律へ、業界は2つの段階を経てきました：初期は中央集権的なオーケストレーションが主流で、1つのオーケストレーターがすべてのエージェントを統一的にスケジュールし、プロセスは固定され、意思決定は集中していました。現在の最先端の実践は自律モードへと移行しつつあり、エージェント同士がメッセージングを通じて自主的に交渉し、動的に役割分担を行います。明確にしておくべきは、自律は無秩序ではなく、安定した「メタプロトコル」の上に構築されているということです。本章の後半では、これら2つのモードの工学的実装を詳しく掘り下げ、実行可能なコードを提供します。

まず直感的な指標から始めましょう：同じハードウェアとモデルの下で、モノリシックな呼び出しで「調査+分析+レポート」タスクを完了する失敗率は約23％（私の300回のテストに基づく）ですが、3エージェントの協調を使用すると失敗率は7％に低下しますが、遅延は1.8倍に増加します。これは、協調モードが銀の弾丸ではなく、複雑なタスクには適していますが、単純なタスクには余計なものになることを示しています。工学的には、マルチエージェントを有効にするかどうかを決定する「複雑度判定器」が必要です。これが第3節で説明するオーケストレーション戦略です。

二、基盤アーキテクチャ：協調におけるDeepSeek APIの基本的な役割

すべてのエージェントは最終的にDeepSeekのAPIを介してLLM呼び出しを行います。ここでの重要な点は、各エージェントにAPIを直接呼び出させるのではなく、履歴、温度制御、エラーリトライなどの機能を提供する統一された「モデルクライアント」をカプセル化することです。マルチエージェント協調の通信基盤はメッセージであり、生の文字列ではありません。したがって、メッセージのデータ構造を定義する必要があります。これには、role（system/user/assistant/tool）、content、agent_id、message_id、timestampなどのフィールドが含まれます。

以下は、requestsライブラリを使用してDeepSeekのチャットインターフェースに直接アクセスし、基本的なタイムアウトとリトライメカニズムを追加した最小限のモデル呼び出しラッパーです。公式SDKを意図的に使用していないことに注意してください。これにより、ネットワーク層の問題がより明確に露呈し、エンジニアリング上のトラブルシューティングが容易になります。

import requests
import time
import json

class DeepSeekClient:
    BASE_URL = 'https://api.deepseek.com'
    def __init__(self, api_key: str, model: str = 'deepseek-chat'):
        self.api_key = api_key
        self.model = model
        self.session = requests.Session()
    def chat(self, messages: list, temperature: float = 0.3, max_tokens: int = 2048) -> str:
        for attempt in range(3):
            try:
                resp = self.session.post(
                    f'{self.BASE_URL}/v1/chat/completions',
                    headers={'Authorization': f'Bearer {self.api_key}'},
                    json={'model': self.model, 'messages': messages, 'temperature': temperature, 'max_tokens': max_tokens},
                    timeout=30
                )
                if resp.status_code == 200:
                    data = resp.json()
                    return data['choices'][0]['message']['content']
                else:
                    print(f'HTTP {resp.status_code}: {resp.text}')
            except Exception as e:
                print(f'Attempt {attempt+1} failed: {e}')
                time.sleep(2**attempt)
        raise RuntimeError('DeepSeek API call failed after retries')

このクライアントには重要な設計があります：エージェント間のロジックを認識せず、モデル呼び出しのみを担当します。マルチエージェントフレームワークでは、これを「サブシステム」として各エージェントに注入し、単一責任を保証します。並行環境では、このクライアントはスレッドセーフである必要があります。最も簡単な方法は、スレッドごとにインスタンスを持つか、グローバルロックを追加することです。実際のプロジェクトでは、複数のスレッドが単一のrequests.Sessionを共有したためにコネクションプールが枯渇し、大量のタイムアウトが発生したことがあります。

以下に、実際のAPIリクエスト形式を示すJSON例を示します。ネットワーク問題のデバッグ時に参照してください。

{
  "model": "deepseek-chat",
  "messages": [
    {"role": "system", "content": "You are a helpful coding agent."},
    {"role": "user", "content": "Implement a binary search in Python."}
  ],
  "temperature": 0.2,
  "max_tokens": 1024
}

三、オーケストレーションパターン：中央集権スケジューラの動作メカニズム

中央集権オーケストレーションは最も直感的な形態です。Coordinator Agentを定義し、ユーザータスクを解析し、どのサブエージェントを呼び出すかを決定し、順序または依存関係グラフに従って実行します。サブエージェント間は直接通信せず、すべてのメッセージはCoordinatorを経由します。このモードの利点は、プロセスが制御可能でデバッグが容易なことです。欠点は、Coordinatorがボトルネックになりやすく、タスクが複雑になると意思決定のプレッシャーが指数関数的に増加することです。

私の実務経験では、タスクの依存関係トポロジーが明確で、ステップが固定されている場合、中央集権が最適です。例えば、「技術記事を書く」タスクは、アウトライン生成器→段落ライター→コードチェッカー→最終ポリッシャーに分解できます。各ステップの出力が次のステップの入力となり、依存関係は単純です。しかし、「市場データに基づいてプロモーション戦略を策定する」のようなオープンな意思決定タスクを扱う場合、中央集権は硬直的になります。サブタスク間で相互にトレードオフを検討する必要があるからです。

中央集権オーケストレーションを実装する際、プロセスを管理するために単純なステートマシンを使用することがよくあります。以下は簡略化されたPythonコードで、

辞書を使用して依存関係を構成し、エージェントを順番に実行する方法を示します。

def run_pipeline(coordinator, agents: dict, flow: list):
    context = {'task': None}
    for step in flow:
        agent = agents[step]
        # Coordinator が現在のコンテキストを付けてプロンプトを構築
        prompt = coordinator.build_step_prompt(step, context)
        result = agent.execute(prompt)
        context[step] = result
    return context

このコードは一見単純ですが、エンジニアリング上いくつかの落とし穴があります。第一に、ステップ間のデータ受け渡しには明確なスキーマが必要です。各 Agent の戻り値を強制的に JSON 形式にし、ステップ間でフィールドを検証することをお勧めします。そうしないと、下流で KeyError が発生しやすくなります。第二に、あるステップが失敗した場合、パイプライン全体が「そのステップの再試行」または「前のステップへのロールバック」をサポートする必要があり、単に abort するのではありません。そのため、実際のプロジェクトでは、各 Agent の戻り値をステータスコード付きでラップし、裸の文字列にしないようにしています。

さらに、オーケストレーターの決定ロジック（つまり、中間結果に基づいて次のステップを決定する方法）は、Coordinator が毎回「考える」のではなく、明示的なルールを使用するのが最善です。Coordinator の動的決定が柔軟すぎると、再現性が失われ、テストが困難になります。頻繁に発生する固定フローはテンプレート化し、例外時のみ LLM 決定を有効にすることを強くお勧めします。

四、自律モード：Agent に「会話」能力を持たせる

自律モードの核心は、複数の Agent がメッセージキューや共有ブラックボードを介して非同期通信し、各 Agent が処理するメッセージや生成する新しいメッセージを自律的に決定できるようにすることです。中央スケジューリングはありません。このモードは Actor モデルに触発されています。実装では通常、ブラックボードアーキテクチャとメッセージバスアーキテクチャの2つの形態があります。ブラックボードアーキテクチャでは、すべての Agent が共有の「ワークスペース」を読み書きし、浅いコラボレーションに適しています。メッセージバスアーキテクチャでは、Agent は特定のタイプのイベントを発行/購読することで通信し、より疎結合です。

DeepSeek API のシナリオでは、自律モードは追加の課題をもたらします。モデル呼び出しは同期であり、非同期通信には並列性が必要です。そのため、複数の Agent のループを管理するためにスレッドまたは asyncio を使用する必要があります。例えば、「リサーチチーム」は Researcher、Analyst、Writer の3つの Agent で構成され、非同期キューを介してメッセージを渡します。Agent がメッセージ処理を完了すると、結果を新しいメッセージとしてバスに送り返し、他の Agent が関連トピックを購読できます。

以下に、キューに基づく自律協調のスケルトンコードを示します。ここでは、スレッドプールを使用して各 Agent の run_loop メソッドを駆動し、各 Agent は自分の inbox からメッセージを取得し、処理後に他の Agent の inbox に送信します。

import queue
import threading

class BaseAgent(threading.Thread):
    def __init__(self, name, model_client):
        super().__init__()
        self.name = name
        self.inbox = queue.Queue()
        self.model_client = model_client
        self.running = True
    def send(self, to_agent, message):
        to_agent.inbox.put(message)
    def process_message(self, message):
        raise NotImplementedError
    def run(self):
        while self.running:
            try:
                msg = self.inbox.get(timeout=1)
                self.process_message(msg)
            except queue.Empty:
                continue

このスケルトンの問題は、各 Agent の process_message 内でモデルを呼び出すことです。モデル呼び出しは IO 集中型なので、スレッドは有効ですが、CPU 集中型操作に対するグローバルインタプリタロック（GIL）の影響に注意する必要があります。Agent 内ではネットワーク応答を待つことが多いため、GIL の影響はほとんどありません。しかし、より厄介なのはメッセージの無制限な増加です。ある Agent の処理速度が追いつかないと、inbox が蓄積され、メモリ爆発を引き起こします。私のプロジェクトでは、Writer が長文を生成するために複数回のモデル呼び出しを必要とし、Researcher が毎秒5件のメッセージを公開したため、すぐにキューに数万件が蓄積され、最終的に OOM になりました。

解決策は背圧メカニズムを導入することです。一般的な方法はキューサイズを制限し、キューが満杯のときは送信者をブロックするか、メッセージを破棄してログを記録することです。よりエレガントな方法は「プルモード」を採用することです。イベント通知を受信した後に共有データベースからメッセージをプルする方式で、メモリ内キューにプッシュするのではありません。ただし、これにより複雑さが増すため、トレードオフを考慮する必要があります。

もう一つのエンジニアリング上の要点は、メッセージの冪等性です。自律システムでは、ネットワークタイムアウトや再試行により、同じメッセージが2回処理される可能性があります。そのため、各メッセージに一意の ID を割り当て、Agent 内で処理済み ID のセットを維持し、最初に遭遇したときのみ処理します。これにはメモリオーバーヘッドが伴いますが、Redis の Set を使用すれば簡単に解決できます。リストを自分で使うのは避けてください。

五、ハイブリッド実装：動的役割割り当てとタスクルーティング

純粋なオーケストレーションも純粋な自律も銀の弾丸ではありません。実践では、ハイブリッドアーキテクチャを推奨します。中央集権的なスケジューリングでタスクを開始し、サブ Agent 間の動的な交渉を許可し、スケジューラが交渉結果に基づいて後続のプロセスを調整します。例えば、「競合分析レポート生成」タスクでは、最初に Coordinator が3つの Agent（リサーチャー、技術アナリスト、マーケットアナリスト）を割り当てます。しかし、実行中に技術アナリストがリサーチャーと直接通信し、より詳細な API ドキュメントを要求することがあります。毎回 Coordinator を経由する必要はありません。

これを実現するために、メッセージプロトコルに hop_limit フィールドを追加します。これは、このメッセージが最大で何個の Agent を経由して転送されるかを示します。Coordinator は最初の役割分担で厳格なホップ制限を設定し、超過した場合は強制的に中央に回収します。これにより柔軟性を保ちながら、メッセージが Agent 間で無限ループするのを防ぎます。さらに、「タスクマネージャー」という Agent を導入し、定期的にすべての Agent の進捗をチェックします。ある Agent が長時間アイドル状態またはビジーでキューが膨張している場合、タスク分割や負荷転送などの調整アクションをトリガーします。

具体的なコード実装はここでは示しませんが、擬似コードのフローを示します：

  1. Coordinator はユーザーリクエストを受け取り、「初期タスク」としてマークします。
  2. Coordinator はタスクを複数のサブタスクに分解し、各サブタスクに dependency と output_schema を含む JSON メタデータを付けます。
  3. コントローラーは3つの Agent を起動し、共有の Redis Stream を購読させます。
  4. 各 Agent はサブタスクを完了した後、結果を「完了イベント」として Stream に公開し、他の Agent が公開した関連イベントを読み取ることもできます。
  5. すべてのサブタスクが完了すると、Coordinator は出力から最終レポートを抽出します。
  6. サブタスクが失敗した場合、Coordinator は失敗のタイプに基づいて再試行または再割り当てを決定できます。

このハイブリッドアプローチは、オーケストレーションの制御可能性と自律性の柔軟性を組み合わせ、複雑なマルチエージェントコラボレーションシナリオに適しています。

このハイブリッドモードの利点は、タスクレベルでの制御が依然として可能であり、サブタスク間の連携がより自然になることです。欠点は、デバッグが複雑になることです。なぜなら、2つのエージェント間でメッセージが行き交うのを見ても、その理由がわからないからです。そのため、私は構造化ログに強く依存しています。各ログには、task_id、from_agent、to_agent、message_type、timestampが含まれ、ログ集約ツール（ELKなど）で検索します。そうしないと、本当に午前3時にログを見ながら悪態をつくことになります。

六、コンテキスト管理とトークン予算

マルチエージェント連携における最大の隠れたオーバーヘッドは、トークン消費です。各エージェントは履歴コンテキストを保持する必要があり、毎回全履歴を使用すると、64Kのコンテキストウィンドウをすぐに使い果たしてしまいます。実際に、単純な3エージェントの会話タスクで、毎ステップ全履歴を渡すと、10ステップ後には約20Kトークンを消費し、有効な情報はわずか5%かもしれません。したがって、「コンテキスト圧縮」メカニズムを実装する必要があります。

実用的な戦略は「階層型メモリ」です。各エージェントは短期メモリ（現在のタスクコンテキスト）と長期メモリ（重要な結論の要約）を維持します。短期メモリは現在のステップの推論に使用され、長期メモリは要約としてシステムプロンプトに注入されます。私は通常、独立した「要約エージェント」を使用して履歴を圧縮します。短期メモリがしきい値を超えるたびに、DeepSeekを呼び出して既存の情報を500文字以内の要約に凝縮し、古い履歴を置き換えます。注意点として、要約エージェントをメインエージェントと混在させないでください。そうしないと、「独り言」の混乱が発生します。

もう1つのトークン節約のテクニックは、各メッセージでシステムプロンプトを繰り返さないことです。セッションでは、システムプロンプトを最初のメッセージにのみ配置し、後続のメッセージにはユーザーロールのみを含めます。ただし、DeepSeek APIは現在自動的にコンテキストを維持しないため、自分でmessagesリストを組み立てる必要があります。そのため、DeepSeekClientにfreeze_systemパラメータを追加し、システムメッセージを自動的に最初に配置し、後続の呼び出しでは履歴リストを直近のNターンに切り詰め、さらに以前に生成された要約を組み込みます。

以下の表は、さまざまなトークン予算戦略の実際の効果を示しています（5エージェントの協調タスク、20のインタラクションステップに基づく）：

戦略総トークン消費タスク成功率備考
全履歴~120K82%上限超過しやすい
直近5ターン~45K74%情報損失
ターン+要約~50K90%推奨

この表は私の環境での実測です。要約戦略はトークンの無駄を減らすだけでなく、成功率も向上させます。なぜなら、モデルが処理中に関係のない履歴に邪魔されないからです。

七、フォールトトレランスとリカバリ：協調システムの高可用性設計

マルチエージェントシステムでは、サブエージェントのAPI呼び出しの失敗は異常ではなく通常です。タスク全体を直接失敗させると、エクスペリエンスは最悪です。そのため、私は3段階のフォールトトレランスを設計しました：ローカルリトライ、タスクデグラデーション、グローバルリスタートです。ローカルリトライとは、単一のAPI呼び出しが失敗した場合、指数バックオフで3回再試行することです。それでも失敗する場合は、そのメッセージを「ソフト失敗」としてマークし、特定のエラーコードを返します。タスクデグラデーションとは、重要なエージェントが機能しなくなった場合、Coordinatorが一時的に通常のLLM呼び出しでそのエージェントの機能を代替するか、そのサブタスクを「未完了」としてマークし、他の部分を続行することです。グローバルリスタートとは、協調ネットワーク全体がデッドロックした場合（例えば、すべてのエージェントが互いを待っている場合）、モニターがタスクフロー全体を再起動しますが、取得済みの中間結果は保持します。

デッドロック検出は自律システムの大きな課題です。私の実践では、各エージェントの「最終アクティブタイムスタンプ」を維持し、モニタリングスレッドが30秒ごとにチェックします。すべてのエージェントが5分以上アクティブでない場合、デッドロックと判断します。その後、リカバリをトリガーし、通常はすべてのタスクをキャンセルし、Coordinatorを再起動して、以前のコンテキストスナップショットを注入します。スナップショットメカニズムは重要です。すべてのエージェントのメモリとメッセージキューの状態を保存できるシリアライズ可能なオブジェクトを設計する必要があります。

もう1つの隠れた落とし穴は、APIの429レート制限です。マルチエージェントの並行処理では、DeepSeekのレート制限を簡単にトリガーします。そのため、グローバルなトークンバケットレートリミッターを実装し、すべてのエージェントの毎秒のAPI呼び出し回数を厳密に制御しています。例えば、毎分最大60回の呼び出しに設定し、上限に達すると後続のリクエストはブロックされキューに入ります。これにより並行性は犠牲になりますが、429で拒否されて再試行するよりははるかにましです。

八、テストとモニタリング：実験から本番へ

マルチエージェントシステムは、出力が生成的で非決定的であるため、アサートが困難です。しかし、エンジニアリングとしては、テスト戦略が必要です。私の推奨は3層のテストです：ユニットテスト（個々のエージェントのプロンプトテンプレートとユーティリティ関数）、統合テスト（小規模な協調フローで、固定されたAPIレスポンスの録音を使用）、エンドツーエンドテスト（実際のAPI呼び出しですが、低コストで低温度を使用し、主要な結果に特定のエンティティが含まれているかをアサート）。APIレスポンスを録音するために、DeepSeekClientにrecordモードとreplayモードを追加しました。これはデバッグ時に特に役立ちます。

モニタリングに関しては、標準的なログとメトリクス（トークン消費、レイテンシなど）に加えて、各エージェントの「意思決定トレース」も追跡します。つまり、各呼び出しの入力と出力を記録します。このトレースはJSON Linesファイルとして保存され、後でオフライン分析に使用してプロンプトを最適化できます。例えば、Researcherが長いクエリ用語を受け取ると、その応答の事実誤認率が高くなることがわかりました。そのため、プロンプトを最適化して、構造化された引用を強制的に出力させました。

最後に、バージョン管理の重要性を強調したいと思います。マルチエージェントシステムでは、各エージェントに独自のプロンプトバージョンがあり、協調プロトコルも変更される可能性があります。そのため、すべてのエージェントの定義とプロトコルをJSONで設定し、Gitで管理しています。新しいバージョンをリリースするたびに、シャドウモードで新旧システムを同時に実行し、結果を比較します。新しいバージョンが100のテストタスクで主要なメトリクス（成功率、レイテンシなど）が旧バージョンより優れている場合のみ、切り替えます。これにより、システムが制御不能になることなく継続的に進化します。

九、実際のケース：研究指向のマルチエージェントチームの構築

本番環境で稼働しているケースを説明します：「業界調査アシスタント」システムで、5つのエージェントを統合しています—Researcher（検索と事実抽出）、Analyst（トレンド分析）、Compiler（レポート統合）、Critic（品質レビュー）、Writer（出力の推敲）。これらはRedisストリームを介して通信し、ハイブリッドモードを採用しています。ユーザーがトピックを入力すると、Coordinatorが最初にタスクを作成し、Researcherが並行してウェブデータを取得し（ツール呼び出しを介して）、Analystが数値を分析し、Compilerが初期レポートを生成し、Criticが論理的な欠陥をチェックし、Writerが最終的に記事を作成します。
このプロセスでは、CriticがCompilerのレポートにデータの裏付けがないことを頻繁に発見し、Researcherに補足を要求します。このようなフィードバックループが自律性の本質です。プロセス全体は約10〜15サイクルで、API呼び出しは約30回、総トークンは約80K（要約を含む）です。負荷テストを通じて、当初の手動スクリプトから、現在は50のタスクを並行処理できるシステムへと進化させました。

数え切れない落とし穴が教えてくれました：マルチエージェントシステムの開発は、マイクロサービスアーキテクチャを構築するようなものですが、サービスが高知能エージェントである点が異なります。各モデル呼び出しを外部サービスを呼び出すかのように、リトライ、タイムアウト、サーキットブレーカーを設計する必要があります。さらに、エンドツーエンドの可観測性が必須です。そうでなければ、「エージェントの幻覚」がチェーン全体を汚染し、特定が困難になります。これは私たちの血と涙の教訓です—当初、システムが稼働して1週間後

... Researcher が時折引用元を捏造することが判明し、その誤った情報が Analyst によって分析に使用され、誤った結論に至りました。それ以来、Researcher の出力に取得した URL のリストを必須とし、独立した「ファクトチェッカー」を設置してフィルタリングしています。

要約すると、マルチエージェントの協調がオーケストレーションから自律へと進化することは、単なる技術選定ではなく、システム工学の思想です。それを真に使いこなすには、タスク分解、プロトコル設計、コンテキスト管理、フォールトトレランスとリカバリ、テストと監視をしっかり行う必要があります。この記事があなたの回り道を少しでも減らす助けになれば幸いです。ご質問があれば、コメント欄でお気軽にご議論ください。