多 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必须携带原因和严重程度。协议校验器在传输层之上做语义检查，错误重试需要上限和降级策略。