マルチエージェント通信プロトコルの設計 — AI開発チュートリアル

マルチエージェント通信の核心的課題

システム内で3つ、10個、さらには100個のAIエージェントが同時に動作するとき、それらの間で効率的かつ正確に情報を交換するにはどうすればよいでしょうか？この一見単純な問題には、分散システムにおける最も古典的な課題が含まれています：メッセージ形式の不一致、ルーティングの混乱、時間的依存関係、障害の伝播、一貫性の保証です。エージェント間通信がマイクロサービスのAPI呼び出しと異なる点は、意味的曖昧性（同じ意図でも複数の表現）、文脈依存性（メッセージの意味が履歴に依存）、動的ルーティング（受信者が事前に決まっていない可能性）、ストリーミング転送（中間結果をストリーミングで送受信する必要がある）です。

メッセージ形式の設計：構造化と柔軟性のバランス

推奨されるメッセージ構造は3つの部分で構成されます：ヘッダー（message_id、session_id、送信者/受信者、優先度などの純粋なメタデータで、ルーティングや重複排除に使用）、ボディ（迅速なルーティングのためのインテントラベル、自然言語と構造化パラメータを含む実際のタスク内容のペイロード）、メタデータ（task_chain_id、retry_count、ttlなどのオーケストレーション用メタ情報）。この設計により、メッセージミドルウェアとエージェントがそれぞれ関連する部分を処理できます。

ルーティング戦略：静的から知的へ

静的ルーティング：オーケストレーション段階で送信者と受信者を決定。実装は簡単だが柔軟性に欠ける。
能力ベースの動的ルーティング：エージェントがレジストリに能力タグを宣言し、ルーティング層がインテントに基づいてマッチング。現在最も一般的な方法。
意味ベースの知的ルーティング：埋め込みを使用してメッセージと能力記述を同じベクトル空間にマッピングしてマッチング。最も柔軟性が高い。

まず能力ベースの動的ルーティングから始め、経験を積んだ後にエッジケースを処理するために意味ルーティングを導入することをお勧めします。

実践：エージェント通信バス

import json, uuid, time
from collections import defaultdict
from openai import OpenAI

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

class AgentBus:
    def __init__(self):
        self.agents = {}
        self.queue = []
        self.history = {}

    def register(self, aid, caps, handler):
        self.agents[aid] = {"caps": set(caps), "handler": handler}

    def send(self, sid, rid, intent, payload):
        msg = {"header":{"msg_id":str(uuid.uuid4()),"sender":sid,"recipient":rid,
                         "ts":time.time(),"type":intent},"body":{"intent":intent,"payload":payload}}
        self.history[msg["header"]["msg_id"]] = msg
        self.queue.append(msg)
        return msg["header"]["msg_id"]

    def route(self, sid, intent, payload):
        matched = [aid for aid,info in self.agents.items() if aid!=sid and intent in info["caps"]]
        return [self.send(sid, aid, intent, payload) for aid in matched]

    def process(self):
        results = {}
        for msg in self.queue:
            rid = msg["header"]["recipient"]
            if rid in self.agents:
                results[msg["header"]["msg_id"]] = self.agents[rid]["handler"](msg)
        self.queue.clear()
        return results

bus = AgentBus()
bus.register("exec1", ["code_gen"], lambda m: f"executed {m['body']['intent']}")
bus.register("exec2", ["code_gen","test"], lambda m: f"tested {m['body']['intent']}")
bus.route("planner", "code_gen", {"task":"login module"})
print(bus.process())

通信パターンと一貫性

ポイントツーポイント：メッセージは送信者から指定された受信者に直接送信され、タスク割り当てシナリオに適しています。パブリッシュ/サブスクライブ：メッセージはトピックに公開され、すべての購読者が受信します。ブロードキャストシナリオに適しています。実際のシステムでは通常、これらを組み合わせて使用します。一貫性保証の戦略：冪等性設計（message_idによる重複排除）、トランザクションセッション（同じsession_idの下で原子的に処理）、ハートビートとタイムアウト（オンライン状態を監視し自動的に再割り当て）、デッドレターキュー（失敗したメッセージを破棄せず、障害処理エージェントが分析）。

本番環境への推奨事項

メッセージの永続化：インメモリキューではなくKafka/RabbitMQを使用。
監視とトレーシング：コールチェーン全体にTrace IDを伝播し、Jaeger/Zipkinで分散トレーシング。
メッセージサイズ制限：上限（例：1MB）を設定し、超過分は共有ストレージ参照を使用。
バージョン互換性：セマンティックバージョニングを使用し、メッセージヘッダーでプロトコルバージョンを宣言。

通信プロトコルの性能ベンチマークとストレステスト

マルチエージェント通信バスを本番環境にデプロイする前に、十分な性能テストが不可欠です。私たちはベンチマークテストスイートを設計しました：スループットテスト——10/50/100エージェントが同時にメッセージを送信するシミュレーションで、バスが1秒間に処理できるメッセージ数を測定（目標>1000msg/s）；レイテンシテスト——P50/P95/P99のメッセージ配信遅延を測定（送信から受信側が処理を開始するまでの時間、目標P99<100ms）；バックプレッシャーテスト——受信側の処理速度が

送信速度に追いつけない場合、バスがメッセージを破棄したりOOMを起こしたりせず、正しく背圧を適用できるかどうか；障害復旧テスト——エージェントのダウン、ネットワーク分断、メッセージブローカーの再起動をシミュレートし、メッセージの喪失がないこととセッションの一貫性を検証する。テスト結果は複数の最適化を導いた：メッセージバッチ処理（10件溜めるか5ms待ってから一括配信）、ゼロコピー転送（大きなメッセージはシリアライズコピーではなく共有メモリを使用）、および優先度付きキュー（高優先度メッセージは独立したチャネルを通り、低優先度にブロックされない）。

メッセージキューの選定比較

Agent通信バスの基盤となるメッセージミドルウェアの選択は、影響が大きい。Redis Streams — 導入が最も簡単（キャッシュとRedisを共有）、コンシューマグループとメッセージ確認をサポート、メッセージ量が<10,000件/秒の小規模システムに適している。RabbitMQ — 成熟して安定しており、複雑なルーティングルールとデッドレターキューをサポート、メッセージの信頼性が非常に高いシナリオに適している。Apache Kafka — 超高通量（百万件/秒）、メッセージの永続化と順序保証が最も強力で、大規模なAgentクラスタやイベントソーシングパターンに適している。NATS — 超低遅延（マイクロ秒レベル）、遅延に非常に敏感なリアルタイムAgentコラボレーションに適している。私たちの選択はKafka — Agentの各メッセージは貴重な監査データであり、Kafkaの長期保存とイベントリプレイ機能は問題解決時に非常に価値がある。

Agentの身元認証とメッセージ署名

マルチAgentシステムでは、メッセージの送信元の真正性を確保することがセキュリティの基盤です。私たちはJWTベースのAgent身元認証メカニズムを実装しました。各Agentは登録時に通信バスから発行された身元トークン（agent_id、公開鍵フィンガープリント、有効期限を含む）を取得します。Agentはメッセージ送信時に秘密鍵でメッセージ本文に署名し、受信者は通信バスを通じて署名とトークンの有効性を検証します。このメカニズムは2つの一般的な攻撃を防ぎます。Agentのなりすまし（悪意のあるプロセスがagent_idを偽装してメッセージを送信 — 有効なトークンと署名がないため直接拒否されます）とメッセージ改ざん（中間者がメッセージ内容を変更 — 署名検証が失敗します）。パフォーマンス面では、Ed25519署名アルゴリズムは通常のCPUでわずかマイクロ秒しかかからず、メッセージ遅延への影響は無視できます。

このスキルチェーンを自分でオーケストレーションしてみませんか？

スキルチェーンで開く →