Agent间通信从总线走向点对点：hermes peer协议设计与实践

做了几年Agent系统我最大的感受就是Agent越多通信越乱。早期项目里所有Agent都往一个消息总线上塞数据人一多、Agent一多总线就变成了瓶颈排查问题的时候日志像瀑布一样往下滚根本分不清谁在跟谁说话。后来我把通信方式改成了点对点核心就是一套叫hermes peer的轻量协议Agent之间直连、直接交换消息不经过任何中转。这篇文章把协议设计和落地经验完整写出来包括我踩过的坑以及一个真实的多Agent全栈协作案例适合已经在做Agent开发、或者正在设计Agent间通信方案的工程师参考。1. Agent间通信为什么一定要点对点1.1 中心化总线模式的瓶颈先说清楚我对中心化总线的态度不是全盘否定而是它有自己的适用边界。很多Agent框架默认把所有消息都送到一个中心队列Agent A要调用Agent BA把请求发到总线总线再路由给B。这套模型在小规模时非常舒服逻辑集中、权限好控制、调试也直观。但Agent数量一旦超过二十个麻烦事就接二连三地来。第一个问题是单点压力。每个Agent的心跳、任务请求、结果回传全都要经过总线总线的CPU、内存、带宽就成了硬瓶颈。我做过一次压测三十个Agent同时跑任务消息队列的积压量每秒增长几千条最后总线先挂了所有Agent一起失联。这种故障模式下你甚至没法区分是总线挂了还是Agent挂了排查成本极高。第二个问题是语义隔离困难。总线上混着太多业务消息订单消息、模型调用消息、日志消息挤在一起谁都可以订阅所有主题权限边界形同虚设。有一次排查问题我发现日志Agent居然在订阅支付回调的topic因为它的过滤规则写得太宽。这不是代码bug是架构上就没有隔离机制。第三个问题是协作链路被拉长。本来Agent A可以直接问Agent B要一个结果走总线的话要经过A发总线、总线存队列、B拉取、B回传总线、总线再通知A这一来一回的时延对实时协作场景是不可接受的。尤其是Agent之间需要频繁协商、你问我答的交互总线模式会让整个系统显得非常笨重。1.2 点对点模式给协作带来的变化hermes peer的核心思路是让每个Agent既是客户端又是服务端Agent之间直接建立会话通道消息不经第三方转发。这套设计和日常生活中的沟通场景很像小团队协作时你更愿意直接去找同事确认问题而不是把所有沟通都发到公司大群再等人回复。总线模式就是大群点对点模式就是直接私聊。采用点对点之后最明显的变化是时延降下来了。同一台机器上的两个Agent一次消息往返可以从几十毫秒压缩到个位数毫秒。即使在跨机器的场景下消息也只需要一跳省掉了中心节点的排队和转发时间。可靠性模型也变得更清晰了。在总线模式下“消息到底有没有送达”是个很难回答的问题因为中间环节太多可能是总线丢了可能是接收方没消费。在hermes peer里因为链路是直连的每个Agent自己对消息的送达负责有明确的ACK机制和重试策略失败时能立刻感知并触发补偿逻辑。还有一点容易被忽略点对点模式让Agent的自治性更强。每个Agent可以自己决定接受哪些连接、拒绝哪些请求通信策略完全是本地的。这对安全隔离和故障隔离都有好处一个Agent被攻破或崩溃不会自动把所有消息都泄露给其他Agent。1.3 什么时候不该用点对点我也要泼一盆冷水点对点不是银弹。如果一个场景是“一对多广播”比如状态变更通知、全局事件分发点对点的效率就很低因为你要维护一份接收者列表逐个发送还得处理每个链路的失败情况远不如topic广播来得干脆。还有一种场景是临时性的Agent组装。某个任务需要临时拉三五个Agent一起干但大家彼此之间不知道对方在哪这时候需要一个“介绍人”角色完成寻址甚至需要一个临时编排者来协调整个任务。hermes peer并不排斥这种模式寻址和发现环节仍然可以依赖一个轻量的注册中心但消息的传输尽量走点对点通道。简单来说我的设计原则是高频的一问一答用点对点低频的全局事件用广播复杂的多Agent编排用编排器但编排器只传控制指令不承载业务消息流。2. hermes peer 协议设计拆解2.1 消息信封与载荷设计hermes peer的消息结构参考了经典的“信封信纸”模型。信封负责传输控制信纸是真正的业务内容。一个完整的消息长这样{ hermes_peer: 1.0, version: 1.2.0, message_id: c8f9c4a1-6d22-4c1f-9f3a-2e9a1e7f9d20, type: request, method: inventory.check, ttl: 30, timestamp: 1712345678901, from: { agent_id: agent-order-001, instance_id: inst-x7f3, address: hermes://192.168.1.20:7891 }, to: { agent_id: agent-inventory-001, address: hermes://192.168.1.31:7891 }, trace_id: trace-98765, payload: { sku: SPU-8842, warehouse: SH-01 }, signature: base64signature... }协议版本直接放在最外层这样任何节点收到消息后不需要解析完整内容扫一眼version字段就能决定是否兼容避免老版本服务解析新消息时报错。message_id是全局唯一的接收方靠它做幂等去重这个字段特别重要因为hermes peer默认采用“至少一次”的投递语义网络超时后发方会重试接收方必须能识别重复消息。type字段我只用了四种基础类型request、response、event、error。没有设计更复杂的语义类型因为我发现业务层的东西一旦写进协议层协议就会变得越来越臃肿而且永远跟不上业务变化。协议层保持小而稳定业务语义交给method字段去扩展。payload是业务内容可以是任意JSON结构。这里有一个我一直坚持的原则payload里的字段必须是无状态的也就是说每个请求都要带齐上下文不能依赖接收方记住之前的状态。这样做的好处是Agent可以随时重启重启之后处理逻辑完全不用变。2.2 寻址与发现不依赖中心的“点名册”Agent要直接通信首先得知道对方在哪。hermes peer的寻址模型是三层结构agent_id负责标识“是谁”instance_id标识“哪一次实例”address标识“当前在哪”。agent_id是稳定的逻辑ID类似人的身份证号一个Agent无论部署在哪个机器上、重启多少次agent_id不变。但同一类Agent可能有多个实例比如多个库存Agent同时运行每个实例有独立的instance_id。调用方在发请求时需要指定agent_id如果目标Agent是多实例部署hermes peer会按照负载策略选择一个instance_id。address是一个URI格式的地址当前主要用hermes://host:port表示这是一个hermes peer的直连地址。未来如果需要走代理或者加密隧道可以扩展协议前缀调用方不需要感知。寻址信息的获取有两条路径。一是启动时从配置中心拉取全量Agent清单各个Agent把自身信息和能力标签注册上去。二是在运行期通过查询机制动态发现比如Agent A要找一个“能做库存校验”的Agent它向注册中心发起能力查询返回候选列表然后直接和候选者建立点对点连接。这里有一个很关键的细节注册中心只负责“介绍”不负责“传话”。一旦A和B建立了直连通道后续所有消息都走点对点通道注册中心即使宕机也不会影响已经建立的连接。这个设计把控制面和数据面分离了避免注册中心成为新的单点瓶颈。2.3 握手、心跳与超时预算hermes peer在建立连接时会做一次轻量握手目的是协商协议版本和交换会话密钥。握手过程我刻意做得比HTTP的TLS握手更简化因为它只解决“身份确认”和“版本协商”两件事不做很多扩展性的参数协商减少握手轮次。握手请求包含发起方的agent_id、instance_id、协议版本、能力标签列表和一个随机nonce。接收方收到后校验发起方身份是否在白名单中如果通过返回自己的agent_id、实例ID、版本号和签名。这个nonce是为了防止重放攻击每个nonce只有效一次握手完成后即失效。连接建立成功之后双方并不长期保持连接而是采用“空闲即断开”的策略。如果连续三十秒没有消息传输连接进入空闲状态但不会立刻销毁而是保持待命直到超过心跳超时阈值才彻底关闭。这样做既能减少空闲连接对资源的占用又不会因为频繁重建连接增加时延。心跳机制上我默认设置了15秒发送一次心跳连续3次心跳无响应就判定对端失联。这个参数不是拍脑袋定的我算过一笔账Agent间正常请求的时延在几十毫秒到几百毫秒之间心跳如果太频繁比如1秒一次会产生大量无效的网络包徒增CPU开销心跳如果太慢比如60秒一次故障检测时间会拖得很长补偿逻辑迟迟无法触发。15秒一次、3次失败判死的组合意味着故障最迟45秒内能被发现这个时间窗口对于大多数业务协作场景是够用的。重试策略也有一套明确的超时预算。每个请求带一个ttl字段单位是秒表示这个请求最多存活多长时间。ttl从请求发起时开始倒计时消息每经过一个处理环节都会检查剩余时间如果剩余时间不足以完成下一个环节就直接返回错误不再继续传递。这样做避免了“请求超时了还在重试队列里排队”的尴尬。重试次数我限制在3次以内重试间隔采用指数退避第一次失败后等1秒第二次等2秒第三次等4秒之后不再自动重试而是返回业务错误交给上层决定是降级处理还是终止任务。2.4 安全模型身份、签名与密级Agent通信的安全往往被忽视但恰恰是协议设计中最不能省的部分。我参考了很多智能硬件蓝牙协议的安全设计经验发现一个共性协议安全不是加一层加密就叫安全而是要清楚地定义信任边界和攻击面。hermes peer的信任模型默认是“零信任”也就是所有Agent默认互不信任每次请求都需要校验身份和权限。实际落地时我分了三层安全措施。第一层是身份层。每个Agent在首次启动时生成一对公私钥公钥注册到Agent管理平台私钥保存在本地安全存储中。握手时双方用私钥对握手消息签名对方用公钥验证签名确保“和我说话的人确实是它声称的agent_id”。这层机制相当于给Agent发了一张无法伪造的身份证。第二层是传输层。握手成功后双方会协商出一个会话密钥后续的消息payload用这个密钥做对称加密。这里我选了AES-GCM因为它同时提供加密和完整性校验能够防止消息在传输过程中被篡改。有些开发者觉得加解密影响性能但实际上在现代CPU上AES-GCM是有硬件加速的对整体时延的影响可以控制在5%以内完全值得付出。第三层是权限层。消息信封里的to字段指定了目标agent_id接收方在执行业务处理前会先检查调用方是否有调用该method的权限。我借鉴了RBAC的思路在Agent管理平台维护了一个权限矩阵比如order-agent只能调用inventory-agent的check方法但不能调用payment-agent的refund方法。这层校验看似笨重但它保证了即使某个Agent被攻破攻击者的影响范围也被限制在同权限级别的Agent之间无法横向移动。必须强调一点安全设计要在协议设计的最早期就考虑进去而不是等功能做完了再补。因为安全机制一旦后置就需要为所有历史消息格式设计兼容方案改造成本至少翻三倍。3. 全栈协作案例一套多Agent订单履约系统3.1 案例背景与角色划分为了把hermes peer的用法讲清楚我用一个真实的线上项目来做案例分析。这是一个订单履约系统核心诉求是用户下单后系统自动完成库存校验、支付处理、出库调度全流程并且尽可能减少人工介入。这个系统里有四个核心Agentorder-agent订单接收Agent负责接收用户订单校验订单合法性发起后续流程。inventory-agent库存校验Agent负责查询仓库库存、锁定库存、释放库存。payment-agent支付路由Agent负责对接第三方支付渠道发起扣款、处理回调。fulfillment-agent履约调度Agent负责对接仓储系统生成出库单、物流单。四个Agent分别部署在不同模块中彼此只通过hermes peer协议通信。各个服务的开发语言也不一样order-agent用Python的FastAPIinventory-agent是Go写的老服务payment-agent是Java。好在hermes peer只定义了消息格式和安全机制不依赖具体语言的SDK只要各自实现协议规范就能互通。3.2 一次订单的完整协作时序用户在小程序里提交了一个订单order-agent创建订单记录后开始协调其余三个Agent。这个过程用文字描述就是一条清晰的链路order-agent先向inventory-agent发起库存校验请求method为inventory.check带参为sku和warehouse期望响应里返回available和lock_id。inventory-agent查询库存后如果库存充足会锁定库存并返回lock_id给order-agent这个lock_id在后续取消订单时用来释放库存。锁库成功的响应到达order-agent后它马上向payment-agent发起支付请求method为payment.create带参为order_id、amount和回调地址。payment-agent会创建一个支付单返回支付链接和pay_order_id给order-agent同时异步等待支付渠道的回调。用户在支付页面完成支付后payment-agent收到回调验签通过后主动向order-agent推送一条event消息通知支付成功。order-agent收到支付成功的event后把订单状态改为待履约然后向fulfillment-agent发送履约请求method为fulfillment.create带参为order_id、sku、warehouse、address。fulfillment-agent调用仓储系统生成出库单并把物流单号和预计发货时间作为响应返回给order-agent。消息示例里最关键的两个设计点一是trace_id贯穿整个链路四个Agent的日志里都带上同一个trace_id排查问题的时候按trace_id一查就能看到完整链路二是每个请求都带ttl比如库存校验的ttl是5秒支付创建的ttl是30秒履约创建的ttl是10秒不同操作按业务容忍度设置不同的超时预算避免一个慢操作拖住整个链路。3.3 异常分支库存不足、支付超时、履约失败实际运行中不可能全程顺利异常分支才是考验协议设计的地方。库存不足时inventory-agent返回一个error响应error_code为INSUFFICIENT_STOCK并带上可选替代仓库的信息。order-agent收到error后先判断是否有替代仓库如果有就向替代仓库的inventory-agent重新发起校验请求如果没有直接回滚订单状态为失败通知用户库存不足。整个处理过程不需要额外的编排器就是一次简单的请求-响应加业务判断。支付超时的情况更复杂。用户打开支付页面后迟迟没有操作payment-agent在30秒内没有收到回调它会主动向order-agent发一条event消息通知PAYMENT_TIMEOUT。order-agent收到后调用inventory-agent的release方法释放之前锁定的库存把订单置为超时关闭。这里有个细节为了避免order-agent因为网络问题没收到这个eventpayment-agent会重试发送三次同时order-agent自身也有一个定时任务如果下单超过5分钟仍然处于待支付状态主动向payment-agent查询支付单状态做兜底。履约失败时fulfillment-agent返回errorerror_code为FULFILL_FAILED带失败原因字段。order-agent收到这个error后做两件事一是向payment-agent发起退款申请payment.refund触发资金原路退回二是调用inventory-agent的release方法释放库存。这套补偿逻辑是显式写在order-agent里的因为hermes peer本身不提供分布式事务能力只能靠业务侧编排补偿流程但协议提供的可靠消息投递和幂等机制让补偿流程变得足够可靠。3.4 全栈中的协议落地Agent SDK与网关这套系统落地时我给每个服务都接入了hermes peer的SDKSDK封装了连接管理、消息收发、心跳、重试、日志这些基础能力业务代码只需要关注method的注册和调用。以Python为例一个Agent的启动流程是这样的初始化Agent身份、加载私钥、注册能力标签、监听端口、连接配置中心获取其他Agent的地址、建立路由表。然后通过装饰器注册自己的处理方法from hermes_peer import HermesAgent, method agent HermesAgent( agent_idagent-order-001, secret_key_path./keys/order_agent_sk.pem, registry_urlhermes://registry.internal:8800 ) agent.method(inventory.check) async def check_inventory(request): sku request.payload[sku] warehouse request.payload[warehouse] result await inventory_service.check(sku, warehouse) return {available: result.available, lock_id: result.lock_id} agent.start()调用其他Agent的方法也封装得很简单SDK会自动处理寻址、握手、超时和重试resp await agent.call( target_agent_idagent-inventory-001, methodinventory.check, payload{sku: SPU-8842, warehouse: SH-01}, ttl5, trace_idcurrent_trace_id )为了保证跨语言服务能正常通信我还在每个服务前面加了一个轻量网关网关负责协议解析和路由业务服务只处理业务逻辑。但这个网关和传统API网关不一样它不承担业务消息的转发只做协议适配和身份认证数据面仍然是点对点的避免网关成为性能瓶颈。4. 最小实现从零构建一个hermes peer节点4.1 工程目录与依赖选择为了验证协议设计我写了一个最小的hermes peer参考实现放在GitHub上供团队内部使用。这个参考实现只使用Python标准库加aiohttp目的是让你能快速看懂协议的核心逻辑而不是被框架细节淹没。工程目录大致如下hermes-peer-demo/ ├── hermes/ │ ├── __init__.py │ ├── message.py # 消息结构与序列化 │ ├── crypto.py # 签名和加密工具 │ ├── handshake.py # 握手协议 │ ├── connection.py # 连接管理与心跳 │ ├── router.py # 路由与method分发 │ └── agent.py # 高层Agent封装 ├── example/ │ ├── inventory_agent.py │ └── order_agent.py └── tests/ └── test_hermes.py依赖方面除了aiohttp之外加了一个cryptography库来做签名和AES-GCM加密。异步框架选aiohttp而不选FastAPI是因为这里主要是Agent之间的长连接通信不需要RESTful接口那一套东西aiohttp的底层原语更贴近协议本身的逻辑。4.2 核心代码消息体、握手、发送循环消息体结构的设计在第二节里已经讲过了核心代码就是一个dataclass加序列化方法。为了节省篇幅这里重点看它的发送循环和接收处理。发送方的核心逻辑在一个send_with_retry函数里它会维护一个已发送消息的缓存收到对应响应后从缓存里移除如果超时未收到响应就按照指数退避策略重发。接收方维护一个message_id到已处理结果的映射如果收到重复的message_id直接返回缓存里的结果不再执行业务逻辑。class HermesConnection: async def send_and_wait(self, message, timeout): msg_id message.message_id self.pending[msg_id] asyncio.get_event_loop().create_future() await self.transport.send(message.to_json()) try: return await asyncio.wait_for(self.pending[msg_id], timeout) finally: self.pending.pop(msg_id, None) async def handle_message(self, raw_message): message Message.from_json(raw_message) if not self.verify_signature(message): return if message.type request: if message.message_id in self.processed_cache: return self.processed_cache[message.message_id] result await self.router.dispatch(message) self.processed_cache[message.message_id] result resp self.build_response(message, result) await self.transport.send(resp.to_json()) elif message.type response: future self.pending.get(message.message_id) if future and not future.done(): future.set_result(message)这段代码虽然简略但包含了hermes peer最核心的四件事签名校验、幂等去重、请求分发、pending future回填。你把这四个点理解透整个协议的主干就掌握了。4.3 关键参数调参与压测参考实现跑通之后我针对关键参数做过一轮压测结论如下连接数方面单机同时保持五百个长连接是常态CPU占用率在10%左右超过一千个连接时内存占用开始明显上升此时需要调整操作系统的文件描述符上限。消息大小方面正常业务消息在1KB到10KB之间时延稳定在1到3毫秒消息超过100KB时GC开销显著增加所以协议里对payload大小做了限制超过5MB的消息直接拒绝。如果确实要传大数据建议做法是先把数据传到对象存储消息里只带一个下载地址。心跳频率方面15秒心跳在五百个连接下产生的网络包大约是每秒33个几乎可以忽略不计如果把心跳改成1秒这个数字变成每秒五百个包CPU会上升几个百分点但对故障感知的提升微乎其微不划算。安全加解密对性能的影响大约是5%到8%在可接受范围内。如果你对性能极其敏感可以做一层开关只在跨网络传输时启用加密同机通信时关闭加密以换取更低的时延。5. 踩坑记录与排查手册5.1 三个真实生产事故这套系统上线以来最让我印象深刻的三个事故都值得拿出来复盘。第一个事故是心跳风暴。上线初期我把心跳间隔设成了3秒想着越频繁越能快速发现问题。结果有一次网络抖动几百个Agent同时检测到对端失联然后一起尝试重建连接和重发积压消息瞬间把网络打满造成了集群级别的抖动。后来我加了抖动因子每个Agent的心跳间隔在基础值上加一点随机偏移避免所有Agent在同一时间点发起心跳同时把间隔调回15秒这个问题就彻底消失了。第二个事故是广播风暴。我本来设计的是点对点通信但有同事把一个状态变更的消息用循环遍历的方式群发给了所有Agent当Agent数量增多时每次状态变更会产生N个消息再叠加重试机制整个系统直接雪崩。后来我在协议层面加了一条规则禁止在业务代码中遍历所有Agent发送同一消息如果确实需要广播走单独的事件总线组件不走hermes peer。有些限制必须在框架层面做掉不能依赖工程师的自觉。第三个事故是协议版本兼容性。第一次升级协议版本时我直接改了消息结构新增了一个必填字段结果旧版本的Agent反序列化时报错整个集群有一半的节点突然不可用。从那以后我做了两条强制规定新增字段必须是可选的且默认值要对旧逻辑无影响每个版本升级前必须有新旧版本共存运行的兼容性测试至少运行二十四小时。5.2 问题速查表整理一份我在实际运维中常用的排查表症状可能原因排查手段解决方案消息超时重试3次后失败目标Agent失联或过载检查对端心跳日志用hermes-peer-cli ping目标地址确认Agent存活后排查资源占用必要时扩容大量重复消息被处理接收方幂等去重失效检查message_id是否唯一查看processed_cache大小确认消息生成逻辑检查去重缓存是否被误清理握手失败协议版本不兼容或签名验证失败抓包查看握手响应error_code升级SDK版本检查公私钥是否匹配消息能收到但时延很高对端业务代码阻塞了事件循环查看对端Agent的线程池状态和GC日志优化业务耗时操作把耗时代码移出主线程内存缓慢增长发送消息缓存未清理检查pending字典大小和processed_cache大小增加缓存上限超限时按LRU策略淘汰5.3 设计建议根据这些实战经验我给想采用类似方案的人三条建议。第一条先约定失败模型再约定成功模型。很多协议设计文档大篇幅描述正常流程但真正的工程质量取决于异常分支有多少种、处理得是否干净。在设计hermes peer时我花时间最多的不是happy path而是失败重试、超时补偿、幂等去重、版本不兼容这些场景。把这些想清楚系统才有机会做到健壮。第二条协议做小业务做宽。协议层只保留传输必需的最小字段业务语义全部放到payload里由method来定义。如果你发现协议层在不断增加各种业务相关的字段说明设计出问题了要尽快把业务字段挪出去让协议层回归传输的本质。第三条日志结构化是救命的。hermes peer的所有消息都带trace_id和agent_id日志必须输出这两个字段否则排查问题时会非常痛苦。我在订单履约系统里专门写了一个日志收集服务按trace_id聚合四个Agent的运行日志一条订单的完整执行链路可以在一秒钟查出来。没有这套日志体系点对点通信的优势会被排查成本压过。6. 后续扩展与我的使用体会写到这里正文主体已经覆盖了hermes peer从设计、实现到落地的全流程。最后再分享一些我对后续扩展方向的想法。我在实际使用中发现Agent间点对点通信最大的价值不在于省那几毫秒时延而在于它让Agent的边界重新变得清晰。总线模式下Agent之间是“弱连接”任何Agent都可以通过topic感知到其他Agent的存在这在一个大型系统里其实是安全隐患。点对点模式下每个连接都是显式的、经过授权的、可审计的系统整体更容易理解也更容易保护。后续我想做两块扩展。一是把安全模型升级到支持动态密钥轮换现在是会话开始时协商密钥并长期使用未来可以定期轮换进一步降低密钥泄露的风险。二是把传输层从TCP扩展到QUIC协议QUIC的零RTT握手和更好的多路复用特性可以进一步降低握手时延同时它还天然支持连接迁移对Agent重启和网络切换更友好。如果你在设计自己的Agent通信协议我的建议是先在小规模场景里把协议跑通不要一上来就追求高性能和高扩展性。协议这种东西越简洁越容易发现问题越复杂越难定位问题。hermes peer目前也只解决了Agent通信中一部分问题分布式事务、全局状态同步这些仍然依赖业务层去做协议本身并不提供魔法。但把通信这一层做扎实整个系统的稳定性就有了最基础的地基。