多 Agent 通信协议：不是发一段 JSON 就叫通信

一、Agent 之间的消息传递需要结构、语义和错误处理

AI Agent系统的设计者常把多Agent协作想象成一群人开会：每个Agent有自己的职责，需要时互相发消息就行。但"发一段JSON"和"通信"是两件事。JSON只是传输格式，通信协议还需要定义消息类型、语义约定、错误处理和状态同步。没有协议的Agent系统就像一群人各说各话，消息格式混乱、语义歧义、失败后没人知道该重试还是放弃。设计一个可用的多Agent通信协议，见证奇迹的时刻是五个Agent协同完成一个复杂任务时零冲突零冗余。

多Agent通信与人类协作的根本区别在于：Agent没有共享常识。人类说"帮我查一下"，对方知道查什么、返回什么格式、出错怎么办。Agent之间必须把这些全部显式定义，否则每次交互都可能产生误解和死循环。

二、协议架构：消息类型、状态机和错误传播的三层设计

flowchart TD
    A[发送Agent] --> B[消息封装层]
    B --> C[消息类型定义]
    B --> D[语义约束标注]
    B --> E[错误处理声明]
    C --> F[传输层]
    D --> F
    E --> F
    F --> G[接收Agent]
    G --> H[消息解析与校验]
    H --> I[状态机推进]
    I --> J[结果返回或错误上报]

通信协议的第一层是消息类型定义。至少需要四种类型：请求（Request）、响应（Response）、通知（Notification）和错误（Error）。每种类型有不同的语义：请求必须有回复，响应必须匹配请求ID，通知不需要回复，错误必须携带原因和建议动作。只有JSON字段没有类型标记的消息，接收方无法决定该回复还是静默处理。

第二层是状态机。每个协作任务应该有明确的状态流转：初始化、进行中、已完成、失败。Agent之间的消息必须推进状态机，而不是随意触发新动作。没有状态机的系统容易出现重复执行、步骤跳跃和无法终止的死循环。

第三层是错误传播。Agent A调用Agent B失败后，A不能只是自己重试。错误信息必须沿调用链向上传播，让任务调度器决定是换一个Agent重试、降级处理还是终止整个任务。盲目重试只会让系统在同一个错误点反复打转。

三、消息协议定义：从传输格式到语义约束的完整封装

下面是多Agent通信协议的核心结构。代码注释解释了设计约束的原因。

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import time
import uuid

class MessageType(Enum):
    """消息类型：不同类型决定接收方的处理行为"""
    REQUEST = "request"       # 必须回复，有超时约束
    RESPONSE = "response"     # 必须匹配request_id
    NOTIFICATION = "notify"   # 不需要回复，但建议确认接收
    ERROR = "error"           # 必须携带原因和可恢复性标记

class ErrorSeverity(Enum):
    """错误严重程度：决定调度器的干预策略"""
    RETRYABLE = "retryable"    # 可重试：换Agent或换参数再试
    DEGRADABLE = "degradable"  # 可降级：返回部分结果
    FATAL = "fatal"            # 不可恢复：终止整个任务

@dataclass
class AgentMessage:
    """Agent间通信消息：结构化封装替代裸JSON"""
    msg_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    msg_type: MessageType = MessageType.REQUEST
    sender: str = ""           # 发送Agent标识
    receiver: str = ""         # 接收Agent标识
    # 设计原因：sender和receiver必须显式标注，
    # 否则广播场景下无法追踪消息来源和责任归属
    request_id: Optional[str] = None  # RESPONSE必须回填REQUEST的msg_id
    payload: dict[str, Any] = field(default_factory=dict)
    # payload必须包含schema声明，不能是任意dict
    payload_schema: str = ""   # payload结构的版本标识
    timeout_ms: float = 5000  # REQUEST的超时时间
    # 设计原因：超时是通信协议的基本约束，
    # 没有超时的请求在对方挂死时会永远等待
    error_info: Optional[dict] = None
    # error_info格式：{reason, severity, suggestion}
    timestamp: float = field(default_factory=time.time)

@dataclass
class TaskState:
    """任务状态机：约束Agent行为不超出当前状态允许的动作"""
    task_id: str
    current_state: str = "initialized"  # initialized|running|completed|failed
    assigned_agents: list[str] = field(default_factory=list)
    # 设计原因：记录assigned_agents防止同一任务被多个调度器重复分配
    retry_count: int = 0
    max_retry: int = 3
    # 见证奇迹的时刻：状态机让五Agent协同零冲突

class ProtocolValidator:
    """协议校验器：在传输层之上做语义校验"""
    def validate(self, msg: AgentMessage) -> tuple[bool, str]:
        """校验消息是否符合协议语义约束"""
        if msg.msg_type == MessageType.RESPONSE and not msg.request_id:
            return False, "RESPONSE必须携带request_id"
        if msg.msg_type == MessageType.REQUEST and msg.timeout_ms <= 0:
            return False, "REQUEST必须有有效超时"
        if msg.msg_type == MessageType.ERROR:
            if not msg.error_info:
                return False, "ERROR必须携带error_info"
            severity = msg.error_info.get("severity", "")
            if severity not in [s.value for s in ErrorSeverity]:
                return False, f"未知的错误严重程度: {severity}"
        if not msg.payload_schema:
            return False, "payload必须声明schema版本"
        return True, "valid"

四、协议权衡：灵活性、一致性和调试成本的三角矛盾

通信协议设计面临三个维度的权衡。第一是格式灵活性vs一致性：payload用任意dict最灵活，但接收方无法校验，调试时看到一堆不明字段只能猜测含义；payload强制schema声明最严格，但每次新增字段需要版本协商。工程折中是schema版本化：新版本可以新增字段但必须保持旧字段语义不变，接收方按版本号做兼容解析。

第二是同步vs异步：同步通信简单，请求-响应一问一答，但发送方必须等待，Agent数增加后延迟叠加；异步通信用事件驱动，发送方不等待，但需要状态机追踪每个任务的进度，调试时很难复现时序问题。一般建议：短任务用同步，长任务用异步，混合场景用同步发请求、异步收结果。

第三是广播vs定向：广播消息（如任务通知）简单但容易产生重复处理，定向消息精确但需要维护Agent间的拓扑关系。工程上通常用一个中心调度器管理拓扑，Agent只与调度器通信，调度器负责路由。

错误处理的权衡也很关键。retryable错误最多重试N次，但N设太大会让系统在同一个失败点反复消耗资源；N设太小可能在偶发故障时过早放弃。degradable错误返回部分结果，但调用方需要知道返回的是完整结果还是部分结果，否则下游逻辑可能出错。

五、总结

多Agent通信协议需要三层设计：消息类型定义决定处理行为，状态机约束任务流转，错误传播机制沿调用链上报失败。payload必须声明schema版本而非任意dict，REQUEST必须有超时约束，RESPONSE必须匹配request_id，ERROR必须携带原因和严重程度。协议校验器在传输层之上做语义检查，错误重试需要上限和降级策略。