前阵子在折腾多 Agent 协作的开发框架发现一个特别容易被忽略但又绕不开的底层组件Agent 之间到底怎么说话。写单 Agent 的时候根本不用考虑这个问题模型调用一下、工具跑一下、返回结果就完事了。可一旦进入多 Agent 场景通信直接就变成了第一个拦住你的坎。我的落地方案是参考了近期社区里很热的 Harness 工程思路在底座上加了一层专门面向 Agent 协作的消息总线Message Bus。这篇聊的就是这套总线怎么设计、怎么实现、踩了哪些坑以及为什么我把重点放在“面向 Agent”而不是“面向服务”上。1. 为什么 Agent 协作需要一套独立的通信基建先聊动机。如果没想清楚“为什么要做”后面所有设计都会跑偏。1.1 单体 Agent 的路越来越窄单一 Agent 通过“思考-行动循环”完成一个复杂任务是当前最常见的形态模型接收用户需求生成计划调用工具观察结果再继续下一步。这个模式处理“中等复杂度”的单目标任务很好用比如写一段文案、查一组资料、做一次图表分析。但当任务变成“多步骤、多实体、需要并行或分阶段推进”的时候单体 Agent 就开始露怯了模型上下文窗口再大也塞不下一个团队的全部状态所有工具调用串行推进整体耗时线性增长一个子任务的失败会导致整条链路回滚容错极差不同领域的专业知识比如调研、写作、代码审查、发布硬塞进同一个 Agent互相干扰。我自己的实际体会是当任务链条超过三个环节、参与角色超过两个时单体 Agent 的回复质量和稳定性会肉眼可见地下降。不是模型能力不够而是架构形态限制了它。1.2 Harness 在协作中的角色Harness 这个词在工程领域的原意是“线束”、“控制装置”。这里要说明一下Agent 开发语境下的 Harness 并不是指某个具体的开源框架而是一类“把模型能力与外部工具、上下文环境捆绑在一起运行”的控制层设计思路。如果把大语言模型比作发动机那 Harness 就是底盘和中控系统。模型只负责“输出下一个标记”而 Harness 负责回答这些关键问题模型目前拥有哪些工具每个工具的调用协议是什么模型的“思考-行动循环”怎么转终止条件是什么模型的上下文窗口如何管理历史记录怎么压缩、怎么裁剪模型在什么条件下允许触发外部事件如何把事件转换成动作模型输出被截断、解析失败、工具报错时怎么恢复处理。没有 Harness模型就只是一堆权重参数有了 Harness模型才变成一个真正可以执行任务的 Agent。在我的设计里每个 Agent 拥有一个独立的 Harness 实例由它负责该 Agent 的“主循环”运行同时充当 Agent 与外部世界的唯一通信出入口。这个“出入口”的角色定位非常重要它保证了总线只需要面向一个稳定接口不需要关心各个 Agent 内部模型到底是哪家、上下文策略是什么。1.3 消息总线解决的四个核心矛盾多 Agent 之间要协作本质上就是要有序地交换“任务”、“结果”、“事件”和“状态”。设计消息总线的动机是把团队协作中的通信逻辑从 Agent 内部拆出来集中解决四个核心矛盾第一解耦。让参与协作的 Agent 互不知道彼此的存在只知道自己通过总线收发消息。写一个新的发布 Agent 不需要改任何调研 Agent 的代码只要注册进来、声明能力、订阅相关主题即可。第二异步化。Agent 的工具调用往往耗时较长——调研、抓取、渲染、模型推理都可能是秒级甚至分钟级。同步 RPC 在这种场景下会让调用方阻塞很久、白白占用上下文资源。消息总线天然支持异步模式发完消息就继续干别的活等结果回调再处理。第三全局状态可观测。单体 Agent 运行时的状态变化你还能通过日志跟踪一旦变成多 Agent谁在什么时候给谁发了什么消息、当前卡在哪个环节如果没一个集中式的消息层排查问题会变成灾难。第四流量削峰和任务排队。多个 Agent 同时提交高并发任务是常态如果没有总线在中间做缓冲和流量控制下游 Agent 会被瞬时打爆导致大量重试和错误。总线本质上是一个异步的缓冲池天然承担了削峰填谷的作用。注意这里“总线”不是指进程内的伪消息队列而是指一个独立运行的基础设施服务。我的实现里它是一个独立进程Agent 和总线之间通过本地 HTTP 或者嵌入式消息协议通信。这也是我认为“消息总线”和“直接点对点调用”之间最本质的差异所在。2. 面向 Agent 的消息总线核心设计思路讲完动机进入设计阶段。这节是全文最核心的内容。我会分几个子主题展开包含消息模型、拓扑与路由、可靠性以及上下文关联性设计。这些不是理论空谈都是我在实际搭建过程中反复调整出来的结论。2.1 消息模型怎么定消息模型是总线的地基。面向 Agent 协作的消息和面向微服务的消息有一个显著区别Agent 之间的消息需要附着更多“意图”和“状态推导”语义。我最终定义的消息结构包含四个核心段Message Envelope信封负责路由和投递控制包括消息 ID、来源 Agent ID、目标 Agent ID 或目标能力标识、消息类型、优先级、时间戳、TTL。Message Payload负载实际内容部分包含任务描述、数据引用、附件 ID 列表、上下文快照引用。ContextRef上下文引用指向一次协作会话的上下文存储地址。这是 Agent 协作里非常关键的字段等下我会单独解释。TraceMeta追踪元数据包含全局唯一的 Trace ID 和当前消息序号用于全链路追踪和排查。一个简化的消息 JSON 长这样{ id: msg_9f8e7d6c5b4a3210, trace_id: trace_ab12cd34ef56, seq: 7, type: task_request, source: coordinator_agent, target: { type: capability, name: content_review }, priority: 5, ttl_seconds: 300, payload: { task: review_article, data_ref: store://artifacts/draft_v3.md, params: { strictness: medium, focus: [grammar, factual_consistency] } }, context_ref: { conversation_id: conv_001, snapshot_version: 12 }, created_at: 2025-06-18T10:24:31Z }我给出的建议消息类型不要只设计“请求”和“响应”两种。按照多年实际经验至少需要以下五类消息类型用途例子task_request有明确目标的任务指派“请校对这篇稿件的第2-5段”task_response任务执行后的结果返回“校对完成发现3处语法问题”event_notify状态变化广播“调研阶段已结束结果已落库”heartbeat存活检测与能力探活“content_review 在线当前负载3”control_signal暂停、恢复、取消、升级“暂停所有发布类任务”这五类消息共同支撑了 Agent 协作中最常见的行为模式指派任务、返回结果、同步进度、发现成员、干预流程。2.2 拓扑与路由策略在传统微服务架构里服务之间通过服务名与注册中心完成寻址。Agent 协作场景则不太一样你通常不需要知道“哪个 Agent”来处理而在乎“哪个 Agent 具备什么能力”以及“当前谁最空闲”。因此我在单总线模式下实现了三种路由策略。基于能力标签的语义路由pervasive use。发布消息的时候目标字段填写的是一个能力标识而不是具体的 Agent ID比如content_review、web_search、image_generation。总线维护一张“能力-在线 Agent”映射表收到消息后按策略挑选合适的接收者。这种设计使得 Agent 的替换和扩缩容变得异常简单——只要新 Agent 注册了同样的能力标签旧 Agent 下线不影响到调用方。基于负载的亲和路由。多个 Agent 拥有同一能力时总线优先把消息路由给当前“负载最低”的实例。我这里用的是一个简单的滑动窗口计数每个 Agent 在处理消息时上报当前在途任务数总线基于这个数据做加权轮询。虽然简陋但在我的场景下实测效果已经足够。基于会话的粘性路由。同一会话上下文中的消息尽量路由到上一次处理该会话的 Agent。这么做的原因是Agent 的 Harness 会维护进程内的局部记忆比如临时的思考状态、工具调用偏好强行换一个 Agent 会导致记忆重置影响协作连贯性。粘性路由用会话 ID 做哈希确保同一个会话的消息稳定落在同一个 Agent 实例上。2.3 可靠性保障Agent 协作中的消息可靠性和传统消息队列还不太一样不只要求“送达”还要求“收件人能正确理解并执行”。在总线上我从三个层面保证可靠性ACK 三层握手。收到消息不代表处理成功。我设计了“已投递-已接收-已处理完成”三层确认总线把消息投递到 Agent Harness 的接收端Harness 返回 ACK1表示“已收到消息实体”Harness 解析消息并交给主循环返回 ACK2表示“语义解析成功将启动处理”Agent 处理完业务逻辑并广播结果时回执 ACK3表示“任务落地完成”。这种三层确认避免了一个经典问题消息没丢但接收方 Agent 在处理一半时崩溃导致调用方永远等不到结果。如果没有 ACK2/ACK3 的区分你根本无法判断是传输丢了、解析失败了还是 Agent 执行过程中挂了。超时、重试和死信。每条消息带 TTL超过 TTL 未收到 ACK3 的消息进入重试队列最多重试 3 次。连续重试失败消息进入死信队列由人工或管理员 Agent 介入处理。这其实就是把后端系统的重试哲学迁移到了 Agent 协作层实测非常管用。幂等消费。为了配合可能出现的重试消费端必须做到幂等。我给每条消息分配全局唯一 IDAgent Harness 在处理前先检查这个 ID 是否已经处理过如果处理过就直接返回上次的结果。否则一次网络波动引发的重试就可能让下游 Agent 重复执行一次“发布文章”或“转账扣款”。3. 关键实现Harness 与总线的对接设计思路定下来就该动手实现了。这节我想重点讲三个实现层面的硬骨头Agent Harness 主循环怎么“感知”总线、能力注册与发现怎么做、上下文怎么跨 Agent 保持同步。3.1 Harness 主循环示例Agent 的 Harness 本质上就是一个事件驱动循环。在我这个系统里主循环做的事刚好比“接消息-干活-回消息”多一点点但它多出来的部分决定了协作是否顺畅。下面是我抽出的最精简主循环伪代码。我用 Python 描述是为了让更多读者能看懂实际生产实现用什么语言都行逻辑是一样的class AgentHarness: def __init__(self, bus_client, agent_id, capability_tags): self.bus bus_client self.agent_id agent_id self.capabilities capability_tags self.context_store ContextStore() def run(self): # 1. 注册能力和探活 self.bus.register_capability(self.agent_id, self.capabilities) self.bus.start_heartbeat(self.agent_id, interval_sec10) # 2. 主循环 while self.is_running: msg self.bus.recv(self.agent_id, timeout5.0) if msg is None: continue ack1 self.bus.ack_received(msg.id) # 3. 幂等检查 if self.context_store.is_processed(msg.id): self.bus.reply_with_stored_result(msg.id) continue # 4. 新上下文初始化 conv_id msg.context_ref.conversation_id context self.context_store.load_or_init(conv_id, msg) # 5. 交给业务处理函数支持异步 future self.dispatch_to_agent_loop(msg, context) # 6. 当 future 完成自动回执 ACK2 / ACK3 future.add_done_callback( lambda f: self.bus.ack_processed(msg.id, f.result()) )这段代码看起来很短但我当初真正跑通它踩了不少坑。几个关键点单独解释一下第2步里的recv超时是为了让主循环有机会响应总线发来的控制信号比如挂起与恢复如果 Agent 始终阻塞在等待消息上就没法处理控制类的短指令。第4步的load_or_init非常关键。Agent 不仅处理消息还要维护任务会话的上下文历史。如果一个消息携带了会话 ID但 Agent 本地找不到就要从总线或对象存储里拉取该会话的快照恢复上下文再处理。第6步的回调设计让 Agent 可以在处理长任务时不阻塞主循环接收其他低优先级消息。这本质上是一个两层的异步消费模型。我强烈建议不要把整个消息处理做成同步阻塞式原因很简单Agent 的模型推理是慢操作几分钟是常态。如果主循环在等模型输出时不能接受新的心跳或取消指令那么整个协作调度就全部堵死了。3.2 能力注册与发现机制能力注册是总线调度核心。在我的系统里每个 Agent 启动时向总线注册三要素Agent ID、能力标签列表、连接地址或通道标识。能力注册报文的简化格式如下{ action: register, agent_id: researcher_agent, capabilities: [ web_search, document_summarize, fact_check ], load_config: { max_concurrency: 4, runtime_pool: gpu_pool } }总线的能力发现模块会维护一个三层映射能力标识 - 候选 Agent 列表 - 当前负载最低者。每次路由时先按能力标识过滤再按节点负载排序然后选出最优投递对象。这里有一个实际经验值得分享能力名称必须先约定好且要用语义清晰的命名空间。如果所有 Agent 都自定义能力名总线根本没法做有效路由。一个 Agent 叫 “search”另一个叫web_search两者其实都在做类似的事总线却会当成两个不同的能力。我给自己的项目定了一套能力命名规范领域_动作的格式比如knowledge_fetch获取知识条目content_generate生成内容quality_evaluate质量评估workflow_execute执行子流程通过这种规范化的能力注册总线可以做“语义近义匹配”即使调用方声明的能力名称和注册方略有差异也能通过同义词表做归一化大大提升了系统的灵活性。3.3 上下文跨 Agent 同步多 Agent 协作里最容易被低估的就是上下文同步。很多项目一开始跑通“任务-返回”就觉得万事大吉结果做到第二阶段就会遇到一个尴尬Agent B 在处理 Agent A 发来的任务时完全没有 Agent A 之前调研得到的信息。原因是 Agent A 的 Harness 只在自己进程里维护了一份上下文快照没有把必要的信息随消息一起发给 B。我最后用的方案是“引用共享 按需拉取”的结合方式Agent A 在处理任务时把阶段性的产出物调研摘要、素材列表、草稿或数据表持久化到对象存储发送消息时信封里携带的数据引用是data_ref: store://artifacts/draft_v3.mdAgent B 的 Harness 收到消息后按需从对象存储中拉取该数据对象并加载到自己的上下文。这样做的好处是消息体本身保持精炼不传送大块内容Agent B 的上下文窗口只加载它真正需要的数据不会被无关历史淹没。上下文跨 Agent 同步的安全性也有保障——通过 ContextRef 指向的存储位置可以用访问控制列表限制谁能读取避免敏感信息在 Agent 之间漫游。3.4 隔离与安全设计Agent 协作场景还有一个被反复追问的点多个 Agent 共用一个总线怎么防止一个 Agent 发疯把总线搞挂怎么防止 Agent A 读取到 Agent B 的上下文我的设计里采用了两层隔离。命名空间隔离。总线为每个协作团队分配独立的命名空间Agent 在注册时声明自己的命名空间。消息路由只发生在同一命名空间内部跨命名空间的通信需要显式声明 Bridge 策略。这样可以防止不同团队之间的消息互相污染对排查问题也有帮助——从源头区分“哪个空间的消息在流动”。消息级权限控制。每条消息带一个access_scope字段Agent 消费消息前会检查该字段是否匹配自己的能力标签和角色。只有匹配的消息才被投递避免敏感信息被无权限的 Agent 拉取。控制信号消息则必须有管理员角色的签名否则总线直接丢弃。在实际项目中安全设计一定要从第一版就加入否则 Agent 数量一多权限问题就变成重构噩梦。我见过太多系统的总线一开始是全开放后面补权限硬是把所有消息都改了一遍。4. 工程化踩坑实录最后这节我想把从“原型跑通”到“稳定运行”之间踩过的坑整理出来。很多设计在第一眼看起来很合理真的跑上数据量之后问题才显山露水。4.1 常见问题速查表问题现象根因处理方案消息总超时但 Agent 正在正常运行消息在总线的等待队列里排太久Agent 并发上限未调优按 Agent 实际吞吐调整max_concurrency或提升消息优先级Agent 收到任务后没反应日志也无报错粘性路由把消息路由给了已经“僵死”的 Agent 实例增加心跳检测超过3个周期未回心跳的 Agent 自动摘除路由表同一个任务被重复执行多次消费端缺乏幂等检查重试机制触发了重复消费在 Harness 里按消息 ID 做幂等记录重复请求直接返回旧结果上下文轮询导致总线流量暴涨Agent 频繁调用总线查询上下文更新改成长连接推送模式减少总线上频繁的轮询请求消息体过于庞大模型处理变慢发送消息时直接内嵌了大段文本改为对象存储引用消息里只带数据指针死信队列堆积但不知道谁的问题死信消息缺乏足够的环境信息写入死信前自动附加当前 Agent 状态快照和最近调用链协作任务进入循环重试卡死失败消息重试策略不区分“可重试”和“不可重试”对“模型输出非法”类错误直接进死信对“服务超时”类错误才重试4.2 印象最深的一个故障调度风暴运行一段时间后我遇到一个非常典型的故障多个 Agent 同时向外发布任务总线上的消息瞬间几千条然后 Agent 自身处理变慢、心跳超时路由表把部分 Agent 临时摘掉摘掉后其他 Agent 又反复重试导致总线上的重试消息更多。最后所有 Agent 都处于“超时-重试-摘除”的循环里整个协作团队冻死。排查下来根因是总线缺少“全局流量视图”和“自我保护机制”。原来的实现里每条消息的管理是独立的没人关注整条总线当前在途总量。当某个 Agent 的处理能力下降时总线的重试调度器并没有感知到下游正在过载仍然疯狂重投形成恶性循环。修复方案包含三点一是给总线增加全局在途消息数量上限超过阈值时新任务直接返回“busy”由调用方决定是否延后而不是无限堆积二是重试策略改为带退避因子的指数退避并加上“下游负载感知”下游 Agent 负载高于阈值时暂停自动重投三是在 Agent 侧增加背压backpressure机制——Agent 本地消费队列超过一定长度时主动通知总线暂缓投递。这套组合拳打了上去系统才真正稳定下来。4.3 排查思路备忘日志里看到“消息超时”“任务卡住”这类现象时我的排查顺序一般如下先看指标不是看日志。优先检查总线在途数量、Agent 负载、消息队列深度、心跳丢失率四个指标快速定位是“链路断”还是“节点慢”。再看消息轨迹。通过 Trace ID 在总线上检索这条消息走过的所有节点确定卡在哪一跳是投递卡住还是处理卡住。最后看 Agent 内部日志。确认收到消息后 Harness 是否完成了幂等校验、上下文是否成功加载、模型推理有没有超时。整个过程最怕的就是一上来就翻 Agent 的模型日志信息量太大反而淹没了问题。总要有个从全局到局部、从指标到日志的排查顺序效率会高很多。4.4 给后来者的三条建议最后分享三条纯经验的东西算是我这个项目做下来比较重要的一些认知。第一第一版就把消息结构设计得宽一点别太抠字段。后面加消息类型、加扩展字段都是家常便饭字段太紧会让你每加一个特性就得做一次兼容。第二一定给总线本身留一个“管理 Agent”的入口。我后来加了一个 admin_agent它订阅总线上的全部控制信号和错误信号能直接往任意 Agent 下发暂停、恢复、单步调试指令。没有它生产环境出问题的时候你只能发控制消息还得保证这个控制消息不被业务流量挤掉。第三不要把 Agent 的“当前状态”只存在 Agent 内存里。Agent 是可以随时崩溃的一旦崩溃它内存里的会话状态就全没了。协作过程中涉及的关键状态该写到对象存储的写对象存储该写到数据库的写数据库。总线上的消息只是“状态的流转”不是“状态的存储”。一开始保持这个意识后期恢复和扩容的代价会小很多。这个系统我从原型到稳定运行下来最深的一个体会是面向 Agent 协作的基建不是帮 Agent “聪明”而是帮 Agent “不乱”。模型会越来越强但协作的秩序和可靠性始终是工程层面必须给出的回答。希望这篇对你有实际帮助。