1. 智能体通信协议到底是什么
1.1 单机智能体为什么也要谈通信
很多人一开始接触智能体(Agent)时,会觉得通信协议是个“分布式系统”才会关心的事:不就是让几个大模型工具互相调用吗?搞那么复杂做什么?但实际做过智能体应用的人都知道,一旦你的系统里出现第二个智能体,立刻就会遇到一个灵魂拷问——它们俩说话,到底按什么规矩来?
我见过不少团队,一开始就是让两个智能体直接用自然语言互相“对话”。表面上看起来它们确实在聊,但真正跑起来全是问题:A智能体以为B智能体已经执行了某个动作,实际上B只是回复了“好的,我知道了”;A问了一个问题,B回答了一长篇,A从里面抽取关键信息时发现字段对不上;最要命的是,当某个环节出错,A根本不知道是自己理解错了,还是B执行错了,还是消息在传输过程中就丢了。这种“听上去在沟通,实际上在鸡同鸭讲”的状态,恰恰就是缺少通信协议约束的典型表现。
智能体通信协议,简单说就是给智能体之间的交互定下一套双方都认的“规矩”。它要解决的不只是“消息怎么传”,还包括“这个消息是什么意思”“对方收到后应该怎么回”“出错了找谁负责”“流程走到哪一步了”这一连串问题。你可以把它理解成两个陌生人合作搬家:如果只是互相喊“你搬这个,我搬那个”,大概率会撞在一起或者漏掉东西;但如果提前约定好“谁负责客厅、谁负责厨房,搬完一件在清单上打个勾,遇到搬不动的就喊停”,整个协作才有底。
1.2 协议在智能体系统里的角色定位
放到整个智能体系统的角度来看,通信协议其实处在很底层但又很关键的位置。底层是因为它直接决定了消息怎么组织、怎么解析、怎么路由;关键是因为它影响了整个系统的可靠性、可扩展性和可调试性。
举个例子。一套典型的智能体系统里,可能有规划智能体负责拆解任务,有执行智能体负责调工具,有记忆智能体负责存取上下文,还有评估智能体负责检查结果。这些角色之间天然就要传递大量信息:任务目标、执行状态、工具返回、错误信息、上下文片段、最终结果。如果这些信息没有一个统一的结构化格式,全靠每个智能体临场发挥,那么每加一个新智能体,都要给它单独写一套“翻译逻辑”,改一个老智能体的输出格式,可能所有关联方都要跟着改。做过一次类似事情的人,绝对不想再体验第二次。
通信协议在这里起的作用,就是把这些“临场发挥”变成“有章可循”。它规定了消息长什么样(格式层),规定了双方怎么对话(交互层),规定了单词和短语在上下文里的准确含义(语义层),还规定了消息该往哪儿送(路由层)。有了这四层约束,智能体之间的协作才能从“两个人七嘴八舌地商量”变成“一套标准流程下各司其职”。
下面我会依次把这几层展开讲清楚,然后给出一份可以直接落地的最小协议设计,最后聊聊我在实操中踩过的坑。整个内容偏基础,但基础的东西一旦理解透彻,后面做复杂系统会顺手很多。
2. 通信协议设计的四个关键层面
2.1 消息格式层:结构化是一切的前提
消息格式层解决的是“消息长什么样”的问题。这个层面最核心的决定,就是选择纯文本还是结构化数据。
纯文本的问题,前面已经说了——机器解析起来太吃力。哪怕用的是今天已经很强的大模型,让它从一长段对话里准确抽取“发送者、接收者、意图、参数、时间戳”,依然存在不可忽略的出错概率。一旦解析出错,后面的流程全乱套。
所以我的建议是,即使是两个智能体之间的直接对话,也不要让它真的变成“大白话聊天”。底层传输的消息必须是结构化的,常见的选择是 JSON 或者类似 JSON 的格式。原因很直接:
- JSON 有明确的键值对结构,发送方和接收方都按照同一个 Schema 来读写;
- 绝大多数编程语言都有成熟的 JSON 解析库,接入成本低;
- JSON 可读性比二进制协议好得多,调试的时候肉眼能看懂。
一个最小可用的消息格式,至少要包含这些字段:
| 字段 | 含义 | 举例 |
|---|---|---|
msg_id | 消息唯一标识,用于追踪和重试 | msg_20250412_001 |
sender | 发送方标识 | planner_agent |
receiver | 接收方标识(或目标角色) | executor_agent |
msg_type | 消息类型(请求/响应/事件/错误) | request |
intent | 消息意图或动作名 | execute_task |
payload | 具体业务数据 | {"task_id": "t_01", "action": "fetch_data"} |
timestamp | 发送时间 | 2025-04-12T10:00:00Z |
correlation_id | 关联 ID,用于把请求和响应配对 | corr_88 |
这里特别要提一下correlation_id和msg_id的区别。msg_id是这条消息自己的身份证,每条消息都不同;correlation_id是“这场对话”的编号,一个请求和它对应的响应、后续的事件,都应该携带同一个correlation_id。没有这个字段,当系统里流转着几千条消息时,你根本没法判断某条响应到底是在回答哪个请求。
2.2 语义层:光有结构还不够,得让双方理解一致
格式层解决的是“机器能不能解析”的问题,语义层解决的是“双方理解的是一回事吗”的问题。
这是一个非常隐蔽但杀伤力极强的坑。举个我实际遇到的例子:A 智能体给 B 智能体发消息,说"status": "done",A 的意思是“这个子任务已经成功执行完毕,结果放在result字段里了”;但 B 的理解是“收到,了解了”,于是 B 直接返回"status": "ok"表示自己知道了。两边都觉得对方懂了,实际上 A 在等 B 去读result,B 在等 A 发下一个指令。整个流程就挂在“done”这个词的两个不同解释上。
出现这个问题的根源在于,大模型对自然语言的边界本身就模糊,同一个词在不同语境下完全可能被理解成不同含义。通信协议里的“语义”,不能靠“这个词看起来挺清楚”来保证。
要让语义层可靠,目前比较实用的做法是引入“意图-槽位”模型。就像语音助手里的“帮我订一张明天上午从北京到上海的高铁票”,意图是book_ticket,槽位包括时间、起点、终点。智能体之间的消息也可以对齐这套思路:intent字段明确说要做什么,payload里的字段明确带做什么所需的参数。
为了进一步减少歧义,我建议在项目初期就维护一份字段定义清单。比如task_status字段只允许这几个值:
pending:任务已创建,等待处理;running:任务正在执行中;succeeded:任务成功执行,结果在result字段;failed:任务执行失败,原因在error字段;cancelled:任务被取消。
规定好每个值的确切含义,然后让所有智能体在收发消息时严格按这份清单来。有人可能会觉得这太死板,但在一个有多方协作的系统里,死板恰恰是可靠的来源。
2.3 路由层:消息该往哪里送
路由层解决的是“消息给谁”的问题。小系统里可能只有两三个智能体,直接点名道姓就行。但系统一大,智能体数量增加,甚至同一类角色有多个实例(比如有 5 个执行智能体在并行处理不同任务),路由就没那么简单了。
常见的做法有三种:
第一种是直接寻址。发送方在receiver字段里写清楚接收方的 ID,接收方订阅所有发给自己的消息。这种最简单,适合角色固定、数量少的系统。
第二种是基于角色的路由。发送方不指定具体给谁,而是指定“我需要一个执行类智能体来处理”,由中间的消息分发层根据当前各智能体的负载、能力、可用状态来选择合适的接收者。这种好处是扩展性好,加一个同类型智能体不需要改动发送方的逻辑。
第三种是基于内容的路由。发送方把消息发到一个公共通道,由路由层根据消息的intent或payload里的关键词决定转发给哪个智能体。这种方式最灵活,但也要承担路由层本身成为瓶颈或误判的风险。
从稳定性角度考虑,我建议初期从“直接寻址”或“角色路由”起步,不要急着上“内容路由”。内容路由听起来智能,实际调试时你会在路由层多出一堆“为什么这条消息被送到那个智能体去了”的疑惑。
2.4 交互层:请求-响应之外,还要考虑事件与状态
交互层解决的是“双方按照什么流程对话”的问题。
最基础的模式是请求-响应。A 发请求给 B,B 处理完返回响应。这个模式跟 HTTP 很像,容易理解,适合大部分“你问我答”的场景。
但智能体系统里还有一种很常见的场景,请求-响应模式处理不了,就是异步任务。A 给 B 发了一个“跑一个耗时很长的数据分析”的任务,B 需要十分钟才能做完。如果 A 一直阻塞等待响应,整个流程就卡死了。
这种情况应该用任务-事件模式:A 发送任务请求,B 立即返回一个“已接收,任务编号为 X”的确认,等 B 真正做完了,再通过事件消息把结果推给 A。A 收到确认后可以继续做别的事,等后续事件到达后再把状态更新掉。
这里就要引入消息类型的设计了。在我常用的协议里,msg_type会区分这几类:
request:请求对方做某件事;response:对请求的即时应答,表示“收到,处理状态如何”;event:主动发出的事件通知,例如任务完成、失败、状态变更;error:错误信息,通常携带错误码和描述。
区分这些类型的价值在于,接收方看到response就知道这是对之前某个请求的确认,看到event就知道这是一个新的状态推送,需要用correlation_id把它关联回原来的任务上下文。这个区分看似简单,但能避免大量“递归式”的误解,否则双方都分不清对方是在回答自己,还是在主动发起新问题,会话就会无限循环下去。
3. 从零设计一个可落地的最小通信协议
3.1 第一步:定义消息信封和基础字段
有了前面的框架,下面我们来设计一份具体的协议。设计目标的定位是:结构尽量简单,不需要引入独立的消息中间件,适用于单体应用内多个智能体的通信,也适合通过 HTTP 转发到远程智能体。
消息统一用 JSON 封装,外层叫“信封”(Envelope),内层是业务数据。我习惯的最小信封长这样:
{ "protocol_version": "1.0", "msg_id": "msg_20250412_001", "sender": "planner_agent", "receiver": "executor_agent", "msg_type": "request", "intent": "execute_task", "correlation_id": "corr_88", "timestamp": "2025-04-12T10:00:00Z", "payload": { "task_id": "t_01", "action": "fetch_data", "params": { "url": "https://example.com/data" } } }有些字段看起来是冗余的,比如既有msg_id又有correlation_id,但它们在调试的时候作用完全不同。msg_id帮你定位“某一条具体消息”,correlation_id帮你拉出“一整条会话链路”。排查线上问题的时候,这两个 ID 都是救命稻草。
有个实操中的细节值得注意:时间戳一律用 UTC,不要用本地时间。因为智能体可能部署在不同的服务器上,如果各自用本地时间,等到排查消息顺序问题时,你会发现根本没法判断谁先谁后。统一 UTC 之后,至少能在时间轴上对齐事件。
3.2 第二步:设计任务处理流程的状态机
光有消息格式还不够,我们还得定义一套流程状态,否则双方对“现在进行到哪一步了”的理解仍然可能不一致。
我把智能体处理一个子任务的典型流程定义成下面这个状态机:
pending:任务创建,还没发出去;sent:任务消息已发出,等待对方确认;accepted:收到接收方的response,确认对方已接收;running:收到event,对方报告任务开始执行;succeeded:收到event,对方报告任务执行成功;failed:收到event,对方报告任务执行失败;cancelled:任务被取消。
发送方在发出任务请求之后,需要维护一张任务状态表。每一次收到新的event消息,就根据correlation_id找到对应的任务记录,更新状态。这样整个流程走到哪一步,永远有据可查。
我见过不少团队省掉accepted这个状态,觉得“都把任务发出去了,对方一定会收到”。但网络请求可能丢、消息队列可能挂、对方智能体可能宕机重启,有了accepted这个确认点,发送方才能及时发现“消息根本没送到”,而不是傻傻等到超时。
接收方这边也一样,接了任务就要在自己的任务表里建一条记录,状态从pending开始。收到请求时返回response,开始执行时发event,执行结束再发一个带结果的event。
3.3 第三步:定义错误处理和重试机制
没有任何系统能保证永远不出错,通信协议里一定要把“出错之后怎么办”提前想好。
错误信息建议至少携带两个字段:
error_code:机器可读的错误码,比如TASK_NOT_FOUND、INVALID_PARAMS、EXECUTION_FAILED;error_message:人类可读的错误描述,方便调试。
举个例子:
{ "protocol_version": "1.0", "msg_id": "msg_20250412_002", "sender": "executor_agent", "receiver": "planner_agent", "msg_type": "error", "intent": "execute_task", "correlation_id": "corr_88", "timestamp": "2025-04-12T10:01:00Z", "payload": { "error_code": "INVALID_PARAMS", "error_message": "Missing required param: url", "task_id": "t_01" } }重试机制上,我踩过几次坑之后总结出来的教训是:不要无脑重试。上来就重试 5 次,每次间隔 1 秒,一旦对端真有压力,这样的重试只会把对方压垮。
合理的做法是采用指数退避,而且给每次重试设置上限。比如第一次失败后等 1 秒再试,第二次等 2 秒,第三次等 4 秒,最多试 5 次,超过上限就放弃并把任务标记为failed。另外,重试时要带上原始的correlation_id,这样接收方才能识别出“这是之前那个失败请求的重试”,而不是当成一个全新的任务来处理。
3.4 实操示例:双智能体协作跑通一次完整交互
这一节,把上面定义好的协议放进一个具体的场景里跑一遍。假设有两个智能体:规划智能体(planner_agent)和执行智能体(executor_agent)。规划智能体负责拆解任务,执行智能体负责实际处理。
第一步,规划智能体向执行智能体发出任务请求:
{ "protocol_version": "1.0", "msg_id": "msg_1001", "sender": "planner_agent", "receiver": "executor_agent", "msg_type": "request", "intent": "execute_task", "correlation_id": "corr_1001", "timestamp": "2025-04-12T10:00:00Z", "payload": { "task_id": "t_01", "action": "fetch_data", "params": { "url": "https://example.com/data" } } }第二步,执行智能体收到请求后,先校验参数是否完整。没问题就立刻返回一个确认响应:
{ "protocol_version": "1.0", "msg_id": "msg_1002", "sender": "executor_agent", "receiver": "planner_agent", "msg_type": "response", "intent": "execute_task", "correlation_id": "corr_1001", "timestamp": "2025-04-12T10:00:01Z", "payload": { "status": "accepted", "task_id": "t_01" } }第三步,执行智能体真正开始干活了,发一个事件通知:
{ "protocol_version": "1.0", "msg_id": "msg_1003", "sender": "executor_agent", "receiver": "planner_agent", "msg_type": "event", "intent": "task_progress", "correlation_id": "corr_1001", "timestamp": "2025-04-12T10:00:05Z", "payload": { "task_id": "t_01", "status": "running" } }第四步,执行智能体干完了,发最终结果。注意这里用的还是同一个correlation_id,而且intent变成了task_completed,这样规划智能体就能准确地把这条事件关联回t_01这个任务,并把任务状态更新成succeeded:
{ "protocol_version": "1.0", "msg_id": "msg_1004", "sender": "executor_agent", "receiver": "planner_agent", "msg_type": "event", "intent": "task_completed", "correlation_id": "corr_1001", "timestamp": "2025-04-12T10:02:00Z", "payload": { "task_id": "t_01", "status": "succeeded", "result": { "data_points": 128, "summary": "示例数据拉取完成" } } }整套流程走完,双方都清楚知道任务从发起到完成的全链路状态。不需要任何一方去“猜”对方的意思,因为每一步都有明确的消息类型、意图和 ID 关联。
4. 常见问题与排查技巧实录
4.1 消息格式没问题,但接收方“理解错了”
这是一个高频问题。发出去的 JSON 字段都在,Schema 校验也通过了,但接收智能体就是理解错了。跟踪下来,最终的根源多数不是格式,而是语义。
比如你定义了"status": "succeeded"表示任务成功,但接收方的大模型在读到这条消息时,把succeeded理解成“对方已经确认完毕”,于是没有去读payload.result。这个问题的根源在于:你让大模型直接去解析并理解原始 JSON 消息。
解决思路是,在大模型和通信层之间加一道“翻译层”。大模型不直接读原始 JSON,而是由代码先把 JSON 解析成结构化的对象,再把关键信息拼接成一段语义明确的话喂给大模型。比如:
收到来自 planner_agent 的消息,类型为事件。 任务 t_01 已执行成功,结果数据点数 128,摘要为:示例数据拉取完成。这种经过翻译后的文本,比直接把 JSON 丢给大模型要稳得多。我这个习惯的养成,就是因为早期项目里大模型直接解析 JSON 出现过太多“看走眼”的情况。
4.2 上下文丢失,导致关联不上的问题
还有一类很典型的问题:接收方收到了消息,但不知道这条消息是接着哪一段对话的。常见表现是,任务已经执行到一半,执行智能体突然问“这个任务是要做什么来着”,说明上下文在某个环节丢了。
排查的时候,先看correlation_id有没有传对。很多上下文丢失的根本原因,就是某个智能体在处理消息时没有把correlation_id透传下去,导致下一次消息关联失败,整个会话链断了。
所以我给每条消息都要求携带correlation_id,并且这一条会被写进项目的代码评审规范里。透传这个 ID 本身不费事,真正费事的是,你忘了传之后要在日志里把它找回来。
4.3 消息风暴和并发更新
智能体数量一多,消息量会快速上涨。如果多个智能体同时往同一个目标发消息,接收方处理不过来;或者同一个任务的多个事件几乎同时到达,发送方并发更新同一个任务记录,就可能出现覆盖问题。
我的建议是,在发送方维护任务状态表这一步,把状态更新放到一个带锁的结构里,或者采用“按事件顺序逐个应用”的方式,避免并发写覆盖。
接收方这边的处理能力也要做流量控制。我见过比较稳妥的做法是:接收方维护一个限流队列,超过处理能力的事件先排队,不直接丢弃。如果队列也满了,返回一个error消息说BUFFER_OVERFLOW,让发送方过一会儿再重试。这里用到的仍然是前面定义的那套错误码机制。
4.4 排查问题时的日志规范
通信层面的问题,光靠“看代码”很难定位。真正高效的排查方式,是把日志规范好。我现在的做法是,每个智能体在收发消息时都打一条结构化的日志,包含消息方向(收/发)、msg_id、correlation_id、msg_type、intent、status。
当线上出了问题,优先通过correlation_id把整条链路的日志拉出来,按时间顺序排列,基本一眼就能看出是哪一步断了、哪一步消息格式异常、哪一步响应超时。这套排查方式比挨个去问“你刚收到什么了”高效得多。
5. 关于协议演进的一些建议
很多人在设计通信协议时会纠结一个问题:协议到底是越规范越好,还是越灵活越好?我的看法是,对于智能体通信这个场景,初期规范一点,后期收益更大。因为智能体协作最怕的就是“不确定”,协议里的每一个字段、每一个状态值,都是在消除不确定。
协议本身也应该允许版本演进。比如消息里的protocol_version字段,不是摆着好看的。当协议发生不兼容变更时,通过版本号可以在新旧智能体并存的过渡期做到兼容。升级的时候,先让一部分智能体切到新协议版本,剩下的继续跑旧版,通过协议适配层转换,比“一步到位”地全量升级要稳得多。
还有一个很实用的建议:协议文档一定要及时更新。我自己见过不少项目,代码改了一版,字段名都换了,文档还停留在最初版本,结果新同事接手的头几个星期,全在跟文档和代码的对不上号作斗争。协议是多人协作的约定,约定不落到文档里,就谈不上约定。
提醒一句:通信协议不是越复杂越好。给你的系统加上“够用”的约束,让每个字段都有明确含义,让每条消息都能被追踪,让每个错误都有处理路径,这就已经比绝大多数智能体系统做得可靠了。等到实际遇到了协议表达不了的新场景,再去扩展也不迟。