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

多 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_factorylambda: 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_factorydict) # 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_factorytime.time) dataclass class TaskState: 任务状态机约束Agent行为不超出当前状态允许的动作 task_id: str current_state: str initialized # initialized|running|completed|failed assigned_agents: list[str] field(default_factorylist) # 设计原因记录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版本而非任意dictREQUEST必须有超时约束RESPONSE必须匹配request_idERROR必须携带原因和严重程度。协议校验器在传输层之上做语义检查错误重试需要上限和降级策略。