マルチエージェントAIの「会話」を壊れにくくする構造化通信の実装ガイド

「エージェントAがエージェントBに話しかける」——マルチエージェントのチュートリアルって、だいたいここで終わっていますよね。😅

でも実際に本番環境で動かすとなると、「その会話が壊れたときどうするの？」という問題が必ずやってきます。今回は、エージェント間通信を文字列ベースから構造化メッセージへ切り替えることで、スケールしても壊れにくい設計にする方法を解説します。

Contents

🔴 文字列ベース通信のどこが問題なのか？
✅ 構造化メッセージで解決する
🛠 なぜこれがスケールに効くのか
まとめ

🔴 文字列ベース通信のどこが問題なのか？

AI agents network / Photo by Tara Winstead via Pexels

多くのフレームワークで見かける実装はこんな感じです。

# よくある「とりあえず動く」エージェント間通信
result = agent_a.run("データを分析してください")
agent_b.run(result)  # resultが何の形式かわからない…

これ、小規模ならなんとかなります。でもエージェントが増えてくると一気に崩壊します。理由はシンプルで、

📭 受け取った文字列の意味・構造が保証されていない
🔍 エラーが起きてもどこで壊れたか追跡できない
🔄 バージョンアップで出力フォーマットが変わると連鎖的にバグる

イメージとしては、「口頭だけで伝言ゲームをしている状態」です。人数が増えるほど内容がズレていく、あの感じですね。

✅ 構造化メッセージで解決する

解決策は、エージェント間でやりとりするデータに型と契約（スキーマ）を持たせることです。Pythonでは pydantic を使うのが定番です。

from pydantic import BaseModel
from typing import Literal
import uuid
from datetime import datetime

# エージェント間で受け渡すメッセージの「型定義」
class AgentMessage(BaseModel):
    message_id: str = str(uuid.uuid4())  # 追跡用のユニークID
    sender: str                           # 送信元エージェント名
    receiver: str                         # 送信先エージェント名
    message_type: Literal["request", "response", "error"]  # 種別を明示
    payload: dict                         # 実際のデータ
    timestamp: datetime = datetime.now()  # いつ送られたか

# 送信側エージェントがメッセージを作る
def agent_a_send() -> AgentMessage:
    return AgentMessage(
        sender="agent_a",
        receiver="agent_b",
        message_type="request",
        payload={"task": "analyze", "data": [1, 2, 3]}
    )

# 受信側エージェントはバリデーション済みの型で受け取る
def agent_b_receive(msg: AgentMessage):
    if msg.message_type == "error":
        print(f"❌ エラーを受信: {msg.payload}")
        return
    print(f"✅ [{msg.message_id}] {msg.sender} からタスク受信: {msg.payload}")

# 実際に動かしてみる
message = agent_a_send()
agent_b_receive(message)

ポイントをまとめるとこんな感じです👇