搞AI Agent的人十有八九都会在某个深夜陷入同一个困境单个智能体跑demo没什么问题一旦想让它同时协调几个工具、对接几个数据源、跟另一个Agent协作干活代码就开始失控。回调套回调、状态散落一地、排查问题时不知道消息到底走到哪一环。我前段时间做的hermes-agent就是想解决这个“多智能体协作混乱”的问题。听名字就知道Hermes是希腊神话里给众神传信的信使用这个名字做项目名核心思想就一句话一切协作皆消息。整个项目不是又造了一个大模型应用框架而是一个以消息驱动为核心的多智能体协作中间层。它管的是Agent之间怎么说话、怎么找到对方、怎么把一次复杂的任务拆成多段异步流程并且保证整个过程可观测、可追踪、可恢复。这篇文章把整个项目的设计思路、核心模块、落地过程和踩坑经历完整写出来适合正在做Agent项目但觉得架构乱成一团的人也适合想了解Agent协作中间层设计细节的开发者。1. 整体设计思路为什么是消息驱动而不是函数直调1.1 函数直调的天花板早期我做Agent原型时最顺手的方案就是直接函数调用。Agent A需要调用Agent B的能力就在代码里agent_b.do_something()再不行就封装一个HTTP接口互相请求。这个模式在小规模、链路固定的场景下完全够用直到我遇到三个问题。第一个是耦合失控。当Agent数量超过五六个、每个Agent又要对接多个工具时调用关系会变成一张蜘蛛网。A调BB调CC又要回调A这时候你想要改动任意一个环节都得先理清楚到底谁在依赖谁。第二个是状态难以追踪。直调模式下一次任务的执行轨迹散落在各个服务的日志里你需要靠时间戳去拼凑完整调用链。任务一旦失败从哪里断的、消息有没有丢、是哪一步吞掉了异常排查效率极其低下。第三个是恢复能力几乎为零。直调模式是同步阻塞的一旦某个环节超时整个链路就卡住。你不可能把一条执行到一半的任务“存起来”等依赖的Agent恢复后再继续。1.2 消息驱动解决什么问题针对上面的痛点hermes-agent选择了消息驱动架构核心转变是把“Agent之间互相调用”变成“Agent之间互相发消息”。每个Agent不关心消息由谁产生、最终由谁处理它只做三件事接收消息、处理消息、发布新消息。这个转变带来几个显著优势。解耦是显而易见的。Agent之间不再有直接依赖只要消息格式不变你随时可以替换、升级、下线任何一个Agent其他Agent完全无感。异步能力自然具备。消息发出后发出方立刻返回不用干等接收方处理完。处理慢、处理快、甚至暂时不可用都不会阻塞上游流程。可追踪性大幅提升。每一条消息都带有唯一的message_id和完整链路标识trace_id全流程的流转路径、耗时、处理结果都可以被完整记录。排查问题不再靠猜直接按trace_id拉出整条链路即可。还有一个容易被忽视的好处是削峰填谷。消息进入队列后即使某个Agent的负载短时间内飙高消息也可以排队等待系统不会因为瞬时流量被打爆。1.3 整体架构分成哪几层整个系统分四层每层职责单一边界清晰。接入层负责所有对外接口的通信协议适配比如HTTP、WebSocket等。外部系统的请求先进这一层转换成内部消息后进入消息层。消息层是整个系统的骨架包含消息队列、消息路由以及消息的持久化存储。这层解决的是消息怎么流转、怎么不丢、怎么重试的问题。智能体层是业务逻辑的承载者每个Agent通过订阅自己感兴趣的Topic来接收消息处理后把结果发布到新的Topic。Agent本身是无状态的所有状态都通过消息携带或者存储在独立的存储模块中。观测层贯穿所有层级负责链路追踪、日志聚合、指标采集。没有这一层前面的解耦和异步都会变成灾难因为你根本不知道系统内部正在发生什么。这四层加起来其实做的事情很简单把一群AI Agent变成一套像快递系统一样的协作网络。消息就是包裹Topic就是地址Agent就是处理站链路追踪就是快递单号。2. 消息协议与通信层设计2.1 消息信封的数据结构既然是消息驱动消息长什么样就是最核心的约定。在实际设计消息结构时我参考了消息队列和事件溯源里常见的Envelope模式给每条消息包了一层标准信封避免业务字段侵入传输层。消息信封的核心字段如下字段类型说明message_idstring全局唯一消息IDUUID格式用于幂等和追踪trace_idstring链路追踪ID一次业务流程共享同一个trace_idtopicstring消息发布的主题决定消息被谁消费message_typestring消息类型如request、response、event、errorsource_agentstring发送方Agent标识target_agentstring可选指定接收方Agent标识payloadobject消息体具体业务数据timestamplong消息创建时间Unix时间戳毫秒ttlint消息存活时间超过则丢弃防止死信堆积reply_tostring可选回执Topic用于请求响应模式这个信封可以做到所有Agent共用一套语义清晰且扩展性强。实际过程中我看到不少项目在消息设计上偷懒直接在消息体里塞业务JSON一旦协议调整就要所有Agent联动修改那种痛我深有体会。2.2 消息类型如何划分消息类型看起来简单但它是Agent协作语义的地基。如果语义划分不清Consumer端的逻辑会变得特别混乱。hermes-agent把消息划分为四类。request表示需要接收方处理并返回结果的请求。对应同步调用场景但底层是异步流转。response是对某条request消息的响应必须携带reply_to字段告诉路由层回哪里。event是单向通知发出去就不管了没有响应适合状态变更通知这类场景。error是处理失败的信息用于反馈异常的Agent或编排器通常带有error_code和error_message。这四类语义覆盖了Agent协作中绝大多数情况。实际编程中只要遵循一个原则——发送方明确这条消息是否需要回复、接收方明确自己处理完要不要发回执整个系统的语义就不会混乱。2.3 序列化方案与消息队列选型消息最终要通过网络传输序列化方案决定了传输效率和跨语言能力。这里我没有直接用大模型的JSON输出那玩意儿结构性太差。系统内部的消息序列化选择了JSON原因是Agent之间的消息体结构多变JSON可读性好、调试方便配合JSON Schema可以做消息格式校验。对于追求更高性能的场景系统预留了MessagePack和Protobuf的适配接口。实际生产环境如果是纯内部通信、对延迟极度敏感Protobuf的效率和类型安全优势会更明显。消息队列的选型是这类架构里最关键的决策之一。不同体量和场景选型逻辑完全不同。场景推荐方案原因本地开发/单机部署Redis Stream部署简单天然支持消息持久化和消费者组中小规模集群RabbitMQ路由灵活支持多Topic和多消费组生态成熟大规模/高吞吐Kafka吞吐量极高适合海量事件流但运维成本高云原生环境云厂商托管MQ免运维自带监控和告警我做hermes-agent时默认适配的是RabbitMQ因为它的Topic交换器完美匹配“Agent订阅Topic”的逻辑模型。但架构层面已经把消息层抽象成了接口切换不同MQ只是配置差异业务代码不受影响。3. Agent注册、发现与任务编排3.1 注册中心的角色定位消息驱动给了Agent灵活性但也带来一个新问题系统怎么知道哪些Agent存在、各自能处理什么类型的消息。这就需要一个注册中心。hermes-agent里注册中心承担三个功能。第一个是能力登记。每个Agent启动后要向注册中心上报自己的标识、能力描述、可处理的Topic列表、健康检查地址相当于对外广播“我在这里我能处理这些事情”。第二个是动态发现。Agent扩容、缩容、故障下线注册中心都能感知。其他Agent不需要硬编码对方地址只需要按Topic发送消息路由层会自动找到当前可用的接收方。第三个是健康管理。注册中心定时探测Agent心跳如果连续N次无响应就把该Agent标记为不健康不再往它路由新消息。3.2 Agent之间的寻址与路由匹配路由匹配是消息能否正确到达的关键。hermes-agent里每条消息发布到Topic后路由器会执行三层匹配逻辑。第一层按精确Topic匹配。如果消息里指定了target_agent优先精确投递给这个Agent适合一对一请求。第二层按Topic通配符匹配。比如某条消息发布到task.image.generate那么订阅了task.image.*的Agent都能收到适合一对多广播场景。第三层按能力抽象匹配。如果消息只声明了需要的处理能力比如一个capability: image_generation字段路由器会在注册中心查找具备该能力的Agent并选择最合适的接收方。第三层匹配一般用于任务编排器调度阶段。它让Agent的物理标识和逻辑能力彻底分离——你不需要关心“到底是哪个Agent在生成图片”只需要声明“我需要一张图片生成能力”系统找到谁就是谁。3.3 编排器如何驱动多Agent协作单个Agent只能完成单一领域任务真实业务几乎都是多步协作。hermes-agent的编排器采用“执行计划驱动”的模式。我用一个例子来说明假设业务是“根据用户需求生成一张配图并配一段文案”。编排器收到请求后会做以下事情。先解析用户需求把一个大任务拆成子任务生成图片、生成文案、审核结果。然后通过路由层把“图片生成能力”和“文案生成能力”的消息分别发给对应Agent两个任务之间没有严格的先后依赖可以并行执行。两个Agent返回结果后编排器做聚合与校验确认最终产出是否符合要求。校验失败则直接把消息返回错误分支由对应Agent重新处理。整个编排过程不是写死的代码分支而是由一个可配置的DAG有向无环图描述。每个节点是一个执行步骤边是消息流转路径。新增一个协作流程时只需要新增一个DAG定义完全不改编排器主逻辑。这套设计在实际使用中最大的感受是加需求不再是噩梦。以前加一个Agent、加一个协作关系要连带改一大片业务代码现在只是加一个节点和几条边的配置。4. 实操过程与核心模块实现4.1 开发环境与依赖准备如果现在你有兴趣复现这个项目我先列出基本的环境依赖Python 3.10RabbitMQ 3.9Redis 6.x用于状态存储Docker和Docker Compose本地快速起依赖。项目目录结构按模块拆分核心层级如下hermes-agent/ ├── hermes_core/ # 核心消息协议、路由匹配、配置管理 ├── hermes_registry/ # Agent注册中心、健康检查 ├── hermes_orchestrator/ # 任务编排器、DAG执行引擎 ├── hermes_transport/ # 消息队列适配层支持RabbitMQ/Redis Stream ├── hermes_agents/ # 内置示例Agent ├── hermes_observability/ # 链路追踪、日志、指标 ├── hermes_web/ # 管理控制台API ├── docker-compose.yml └── config.yaml使用Docker Compose可以快速起一套依赖环境docker-compose up -d rabbitmq redis4.2 核心代码消息路由与Agent基类路由模块是消息层的心脏。这里给出简化后的核心思路核心逻辑是根据Topic找到对应消费者并转发消息。# hermes_core/router.py import asyncio from collections import defaultdict class Router: def __init__(self): self._subscribers defaultdict(list) # topic pattern - [handler] def subscribe(self, topic_pattern, handler): self._subscribers[topic_pattern].append(handler) async def dispatch(self, message): matched False for pattern, handlers in self._subscribers.items(): if self._match_topic(pattern, message.topic): matched True for handler in handlers: # 每个handler在本Agent独立任务队列中执行互不影响 asyncio.create_task(handler(message)) if not matched: self._handle_unroutable(message) staticmethod def _match_topic(pattern, topic): # 支持 * 匹配一级、# 匹配多级类似MQTT主题规则 if pattern #: return True pattern_parts pattern.split(.) topic_parts topic.split(.) if len(pattern_parts) len(topic_parts) and pattern_parts[-1] ! #: return False for p, t in zip(pattern_parts, topic_parts): if p #: return True if p ! * and p ! t: return False return len(pattern_parts) len(topic_parts)Agent基类封装了消息处理的公共逻辑业务Agent只需要继承并实现handle_message方法。# hermes_core/agent.py import json import uuid class BaseAgent: def __init__(self, agent_id, transport, registry): self.agent_id agent_id self.transport transport self.registry registry async def start(self): await self.registry.register(self.agent_id, self.get_capabilities()) for topic in self.get_subscribed_topics(): self.transport.subscribe(topic, self._on_message) self.transport.register_consumer(self) async def stop(self): await self.registry.deregister(self.agent_id) async def _on_message(self, envelope): try: result await self.handle_message(envelope) if envelope.message_type request and envelope.reply_to: await self.reply(envelope, result) except Exception as exc: await self.publish_error(envelope, exc) async def publish(self, topic, payload, message_typeevent, **kwargs): envelope { message_id: str(uuid.uuid4()), trace_id: kwargs.get(trace_id), topic: topic, message_type: message_type, source_agent: self.agent_id, payload: payload, timestamp: int(time.time() * 1000), ttl: kwargs.get(ttl, 60000), } await self.transport.publish(envelope) async def reply(self, request_envelope, result): await self.publish( request_envelope.reply_to, result, message_typeresponse, trace_idrequest_envelope.trace_id, ) async def handle_message(self, envelope): raise NotImplementedError def get_capabilities(self): return [] def get_subscribed_topics(self): return []这段代码虽然结构简单但有一个设计细节值得单独说_on_message里用异常捕获把异常转换成error消息而不是吞掉。这样编排器可以收到失败信号并做补偿处理而不是等超时。生产环境中这个设计救了我好几次因为它让失败变得可见、可追踪。4.3 编排器的DAG执行引擎编排器接收请求后会加载对应的DAG定义按依赖关系执行各个节点。DAG定义使用YAML配置一个典型的双Agent协作流程长这样# workflows/image_with_caption.yaml name: image_caption_workflow version: 1.0 nodes: - id: generate_image capability: image_generation timeout_ms: 30000 - id: generate_caption capability: text_generation timeout_ms: 20000 - id: validate_result capability: content_validation timeout_ms: 10000 edges: - from: generate_image to: validate_result - from: generate_caption to: validate_result执行引擎的核心是拓扑排序和消息派发。每个节点完成后将结果存入状态存储然后检查下游节点是否满足执行条件。这个实现借鉴了流式数据处理里的Batch同步思想只不过处理单元从数据块变成了Agent消息。4.4 链路追踪与日志关联可观测性是这个项目里我最坚持的部分。每条消息从进入系统的第一刻起就携带trace_id所有日志、指标、链路信息都以trace_id为关联键。{ timestamp: 2025-01-15T10:23:45.123Z, level: INFO, trace_id: d8f3a2e1-9c4b-4f6e-9a2b-1c2d3e4f5a6b, message_id: b2e4f6a8-1234-4c5d-9e0f-abcdef123456, agent_id: agent_image_gen, event: agent.message.received, topic: task.image.generate, latency_ms: 23 }实际排查问题时只需要在日志系统里搜索trace_id就能看到这条消息经历了哪些Agent、每一步耗时多少、最终是成功还是失败。这个能力在生产环境里至关重要尤其是当一条任务链跨了四五个Agent时没有链路追踪你根本不可能定位到瓶颈。有次一个用户反馈说“生成文案偶尔超时”我们就是通过链路追踪发现超时不是文案Agent本身慢而是图片Agent并发太高导致消息队列积压文案Agent等前置依赖等了几秒。链路一拉出来问题一目了然。4.5 实际部署与运行效果我用一套双Agent协作Demo验证了系统运行效果。一个图片生成Agent一个文案生成Agent编排器串联两条链路模拟100个并发请求。实际运行结果如下指标数值总请求数100成功完成数98平均端到端耗时3.2秒P95端到端耗时5.8秒消息吞吐量约180 msg/sAgent故障恢复时间约5秒失败的2个请求是图片生成Agent所在服务内存溢出导致但关键点在于——这两个请求没有丢失消息在队列中保留服务重启后消息被重新消费并处理成功。这就是消息驱动架构容错能力的直接体现。5. 常见问题与排查技巧实录5.1 消息消费重复怎么处理消息队列的at-least-once投递语义决定了消费重复是常态。同一个Agent崩溃重启后队列里未被确认的消息会被重新投递如果不做幂等就会出现重复处理。解决思路有两层。第一层是消息级幂等每条消息带唯一message_idAgent处理前先查询是否已处理过该ID处理过则直接返回上次的结果。第二层是业务级幂等比如“生成图片”这个动作设计成可覆盖式写入重复生成只是覆盖旧文件不会产生脏数据。我强烈建议所有Agent在处理消息前都加一层幂等检查。这个习惯能帮你避开海量重复消息引发的连锁Bug。5.2 消息堆积导致延迟飙升有次我压测时发现Agent消费速度跟不上生产速度队列积压越来越多端到端耗时从3秒飙到30秒。排查流程如下。先看消费者速度查看每个Agent的latency_ms和消费速率指标确认是哪个环节成为瓶颈。再看队列深度用RabbitMQ管理页面查看对应Queue的堆积数量确认积压规模。然后看下游依赖如果Agent本身不慢检查它调用的外部API或模型服务是否超时。那次的问题根源是某Agent调用的外部模型服务在高峰期响应变慢单个请求耗时从2秒涨到10秒消费者实际吞吐大幅下降。解决方案是给外部调用加熔断和快速失败。一旦外部服务响应超过阈值立刻返回错误而不是无限等待让编排器走重试或降级分支。5.3 编排任务卡死不动怎么办DAG执行中偶尔出现“任务卡住”的情况表现为编排器收到请求后长时间无后续消息。排查思路先确认是否为死信查消息是否因TTL过期被丢弃如果是调整TTL配置。然后确认是否为节点悬挂查对应Agent的日志看是否异常退出但没有发送错误消息。最后确认是否为拓扑配置错误检查DAG定义是否存在环或孤立节点拓扑排序失败会导致整个流程无法启动。一个我印象很深的坑是某次新加Agent时它的消费Topic写错了层级导致消息被路由到通配符匹配的其他Agent原Agent完全没收到。排查时在注册中心看到的订阅关系是正常的但实际运行的Router里订阅关系没刷新。重启Router后问题消失。这个Bug的教训是改配置后一定要验证运行中的路由表别只改文件不重载。5.4 消息序列化兼容性Agent迭代过程中消息结构经常变化新增字段没问题但删除或改名老字段会导致老版本Agent反序列化报错。我的经验是两个原则。一是只增不改新需求通过新增字段实现不修改旧字段语义。二是消费端容错反序列化时对缺失字段给默认值不强制要求全字段存在。这样能保证新旧版本Agent在混跑阶段不会互相踩坏消息。5.5 问题排查速查表问题现象可能原因排查/解决办法消息发出但无Agent处理Topic订阅不匹配检查Router的订阅关系确认Topic层级一致消息被重复处理消费后未及时确认开启手动ACK处理成功后再确认Agent状态偶发不一致注册中心缓存过期缩短心跳间隔强制刷新注册表端到端延迟越来越高消费者处理慢或队列堆积查看Agent耗时指标增加消费者实例或优化外部依赖编排流程无响应DAG拓扑错误或节点悬挂核对DAG定义检查Agent异常日志Agent重启后消息丢失队列未持久化配置持久化队列消息发布时标记持久化6. 项目扩展方向与个人心得6.1 从单机到多活的演进思路目前hermes-agent的默认部署是单集群模式注册中心、路由层都在一个逻辑域内。如果业务规模继续扩大一个值得探索的扩展方向是多活部署。思路是给每个部署单元加一个region标签注册中心按region做路由亲和。消息发布时优先路由到同region的AgentRegion内没有合适消费者再跨区转发。这样既能降低跨机房延迟又能保障单Region故障时整体可用。但需要注意跨区转发会引入网络抖动和消息延迟需要给TTL留足余量。另一个方向是Agent能力升级时做到平滑迁移。利用注册中心的能力抽象匹配新版本Agent注册成功后旧版本Agent定向消息全部切换到新版本业务方无感知。这个我已经在代码里预留了接口实现时只需要在注册中心增加一个preferred_version配置即可。6.2 补充一点我对Agent框架选型的思考做这类中间件时很多人会纠结“为什么不用现成的Agent框架”。我自己的体会是通用Agent框架核心解决的是“单Agent怎么思考、怎么调工具”而hermes-agent解决的是“多Agent怎么组织、怎么协作”。两者关注点不同可以组合使用。实际项目中我通常会用LangChain或自研Reasoning Loop处理单Agent内部的思考决策然后把Agent的输入输出封装成标准消息接入hermes-agent的消息网络中。这样既保留了单Agent的智能性又获得了协作架构的稳定性。这是一个被低估但非常有价值的组合方式。6.3 如果你要复刻这个项目我的建议不要一上来就追求大而全。先在一台机器上把RabbitMQ和Redis跑起来定义好消息信封结构写好一个Router和两个测试Agent跑通“Agent A发消息给Agent B”的最小闭环。这个最小闭环建立后再逐步加入编排器、注册中心、链路追踪。每一步都留出可观测的日志和指标因为消息驱动架构的调试难度远高于传统的直调模式可观测性不是加分项而是必需品。消息协议是系统最关键的部分宁可多花一周时间讨论设计也不要在上线后频繁改协议。协议一旦定稿所有Agent和编排器都会依赖它改动成本非常高。我刚做这个项目时在协议上吃过亏前后改了两次消息信封结构所有Agent都跟着动那种挫败感至今记忆犹新。最后一点心得多智能体协作的本质不是单个Agent有多聪明而是消息流转得有多顺畅、系统容错有多强。把消息设计和可观测性做好了哪怕每个Agent的逻辑都很朴素整个系统的稳定性和可维护性也远超那些单点智能很强但协作混乱的设计。这大概也是Hermes作为“信使”的真正意义——真正重要的不是谁在处理而是消息可靠地抵达了需要它的地方。
消息驱动多智能体协作：Hermes Agent架构设计与实践
搞AI Agent的人十有八九都会在某个深夜陷入同一个困境单个智能体跑demo没什么问题一旦想让它同时协调几个工具、对接几个数据源、跟另一个Agent协作干活代码就开始失控。回调套回调、状态散落一地、排查问题时不知道消息到底走到哪一环。我前段时间做的hermes-agent就是想解决这个“多智能体协作混乱”的问题。听名字就知道Hermes是希腊神话里给众神传信的信使用这个名字做项目名核心思想就一句话一切协作皆消息。整个项目不是又造了一个大模型应用框架而是一个以消息驱动为核心的多智能体协作中间层。它管的是Agent之间怎么说话、怎么找到对方、怎么把一次复杂的任务拆成多段异步流程并且保证整个过程可观测、可追踪、可恢复。这篇文章把整个项目的设计思路、核心模块、落地过程和踩坑经历完整写出来适合正在做Agent项目但觉得架构乱成一团的人也适合想了解Agent协作中间层设计细节的开发者。1. 整体设计思路为什么是消息驱动而不是函数直调1.1 函数直调的天花板早期我做Agent原型时最顺手的方案就是直接函数调用。Agent A需要调用Agent B的能力就在代码里agent_b.do_something()再不行就封装一个HTTP接口互相请求。这个模式在小规模、链路固定的场景下完全够用直到我遇到三个问题。第一个是耦合失控。当Agent数量超过五六个、每个Agent又要对接多个工具时调用关系会变成一张蜘蛛网。A调BB调CC又要回调A这时候你想要改动任意一个环节都得先理清楚到底谁在依赖谁。第二个是状态难以追踪。直调模式下一次任务的执行轨迹散落在各个服务的日志里你需要靠时间戳去拼凑完整调用链。任务一旦失败从哪里断的、消息有没有丢、是哪一步吞掉了异常排查效率极其低下。第三个是恢复能力几乎为零。直调模式是同步阻塞的一旦某个环节超时整个链路就卡住。你不可能把一条执行到一半的任务“存起来”等依赖的Agent恢复后再继续。1.2 消息驱动解决什么问题针对上面的痛点hermes-agent选择了消息驱动架构核心转变是把“Agent之间互相调用”变成“Agent之间互相发消息”。每个Agent不关心消息由谁产生、最终由谁处理它只做三件事接收消息、处理消息、发布新消息。这个转变带来几个显著优势。解耦是显而易见的。Agent之间不再有直接依赖只要消息格式不变你随时可以替换、升级、下线任何一个Agent其他Agent完全无感。异步能力自然具备。消息发出后发出方立刻返回不用干等接收方处理完。处理慢、处理快、甚至暂时不可用都不会阻塞上游流程。可追踪性大幅提升。每一条消息都带有唯一的message_id和完整链路标识trace_id全流程的流转路径、耗时、处理结果都可以被完整记录。排查问题不再靠猜直接按trace_id拉出整条链路即可。还有一个容易被忽视的好处是削峰填谷。消息进入队列后即使某个Agent的负载短时间内飙高消息也可以排队等待系统不会因为瞬时流量被打爆。1.3 整体架构分成哪几层整个系统分四层每层职责单一边界清晰。接入层负责所有对外接口的通信协议适配比如HTTP、WebSocket等。外部系统的请求先进这一层转换成内部消息后进入消息层。消息层是整个系统的骨架包含消息队列、消息路由以及消息的持久化存储。这层解决的是消息怎么流转、怎么不丢、怎么重试的问题。智能体层是业务逻辑的承载者每个Agent通过订阅自己感兴趣的Topic来接收消息处理后把结果发布到新的Topic。Agent本身是无状态的所有状态都通过消息携带或者存储在独立的存储模块中。观测层贯穿所有层级负责链路追踪、日志聚合、指标采集。没有这一层前面的解耦和异步都会变成灾难因为你根本不知道系统内部正在发生什么。这四层加起来其实做的事情很简单把一群AI Agent变成一套像快递系统一样的协作网络。消息就是包裹Topic就是地址Agent就是处理站链路追踪就是快递单号。2. 消息协议与通信层设计2.1 消息信封的数据结构既然是消息驱动消息长什么样就是最核心的约定。在实际设计消息结构时我参考了消息队列和事件溯源里常见的Envelope模式给每条消息包了一层标准信封避免业务字段侵入传输层。消息信封的核心字段如下字段类型说明message_idstring全局唯一消息IDUUID格式用于幂等和追踪trace_idstring链路追踪ID一次业务流程共享同一个trace_idtopicstring消息发布的主题决定消息被谁消费message_typestring消息类型如request、response、event、errorsource_agentstring发送方Agent标识target_agentstring可选指定接收方Agent标识payloadobject消息体具体业务数据timestamplong消息创建时间Unix时间戳毫秒ttlint消息存活时间超过则丢弃防止死信堆积reply_tostring可选回执Topic用于请求响应模式这个信封可以做到所有Agent共用一套语义清晰且扩展性强。实际过程中我看到不少项目在消息设计上偷懒直接在消息体里塞业务JSON一旦协议调整就要所有Agent联动修改那种痛我深有体会。2.2 消息类型如何划分消息类型看起来简单但它是Agent协作语义的地基。如果语义划分不清Consumer端的逻辑会变得特别混乱。hermes-agent把消息划分为四类。request表示需要接收方处理并返回结果的请求。对应同步调用场景但底层是异步流转。response是对某条request消息的响应必须携带reply_to字段告诉路由层回哪里。event是单向通知发出去就不管了没有响应适合状态变更通知这类场景。error是处理失败的信息用于反馈异常的Agent或编排器通常带有error_code和error_message。这四类语义覆盖了Agent协作中绝大多数情况。实际编程中只要遵循一个原则——发送方明确这条消息是否需要回复、接收方明确自己处理完要不要发回执整个系统的语义就不会混乱。2.3 序列化方案与消息队列选型消息最终要通过网络传输序列化方案决定了传输效率和跨语言能力。这里我没有直接用大模型的JSON输出那玩意儿结构性太差。系统内部的消息序列化选择了JSON原因是Agent之间的消息体结构多变JSON可读性好、调试方便配合JSON Schema可以做消息格式校验。对于追求更高性能的场景系统预留了MessagePack和Protobuf的适配接口。实际生产环境如果是纯内部通信、对延迟极度敏感Protobuf的效率和类型安全优势会更明显。消息队列的选型是这类架构里最关键的决策之一。不同体量和场景选型逻辑完全不同。场景推荐方案原因本地开发/单机部署Redis Stream部署简单天然支持消息持久化和消费者组中小规模集群RabbitMQ路由灵活支持多Topic和多消费组生态成熟大规模/高吞吐Kafka吞吐量极高适合海量事件流但运维成本高云原生环境云厂商托管MQ免运维自带监控和告警我做hermes-agent时默认适配的是RabbitMQ因为它的Topic交换器完美匹配“Agent订阅Topic”的逻辑模型。但架构层面已经把消息层抽象成了接口切换不同MQ只是配置差异业务代码不受影响。3. Agent注册、发现与任务编排3.1 注册中心的角色定位消息驱动给了Agent灵活性但也带来一个新问题系统怎么知道哪些Agent存在、各自能处理什么类型的消息。这就需要一个注册中心。hermes-agent里注册中心承担三个功能。第一个是能力登记。每个Agent启动后要向注册中心上报自己的标识、能力描述、可处理的Topic列表、健康检查地址相当于对外广播“我在这里我能处理这些事情”。第二个是动态发现。Agent扩容、缩容、故障下线注册中心都能感知。其他Agent不需要硬编码对方地址只需要按Topic发送消息路由层会自动找到当前可用的接收方。第三个是健康管理。注册中心定时探测Agent心跳如果连续N次无响应就把该Agent标记为不健康不再往它路由新消息。3.2 Agent之间的寻址与路由匹配路由匹配是消息能否正确到达的关键。hermes-agent里每条消息发布到Topic后路由器会执行三层匹配逻辑。第一层按精确Topic匹配。如果消息里指定了target_agent优先精确投递给这个Agent适合一对一请求。第二层按Topic通配符匹配。比如某条消息发布到task.image.generate那么订阅了task.image.*的Agent都能收到适合一对多广播场景。第三层按能力抽象匹配。如果消息只声明了需要的处理能力比如一个capability: image_generation字段路由器会在注册中心查找具备该能力的Agent并选择最合适的接收方。第三层匹配一般用于任务编排器调度阶段。它让Agent的物理标识和逻辑能力彻底分离——你不需要关心“到底是哪个Agent在生成图片”只需要声明“我需要一张图片生成能力”系统找到谁就是谁。3.3 编排器如何驱动多Agent协作单个Agent只能完成单一领域任务真实业务几乎都是多步协作。hermes-agent的编排器采用“执行计划驱动”的模式。我用一个例子来说明假设业务是“根据用户需求生成一张配图并配一段文案”。编排器收到请求后会做以下事情。先解析用户需求把一个大任务拆成子任务生成图片、生成文案、审核结果。然后通过路由层把“图片生成能力”和“文案生成能力”的消息分别发给对应Agent两个任务之间没有严格的先后依赖可以并行执行。两个Agent返回结果后编排器做聚合与校验确认最终产出是否符合要求。校验失败则直接把消息返回错误分支由对应Agent重新处理。整个编排过程不是写死的代码分支而是由一个可配置的DAG有向无环图描述。每个节点是一个执行步骤边是消息流转路径。新增一个协作流程时只需要新增一个DAG定义完全不改编排器主逻辑。这套设计在实际使用中最大的感受是加需求不再是噩梦。以前加一个Agent、加一个协作关系要连带改一大片业务代码现在只是加一个节点和几条边的配置。4. 实操过程与核心模块实现4.1 开发环境与依赖准备如果现在你有兴趣复现这个项目我先列出基本的环境依赖Python 3.10RabbitMQ 3.9Redis 6.x用于状态存储Docker和Docker Compose本地快速起依赖。项目目录结构按模块拆分核心层级如下hermes-agent/ ├── hermes_core/ # 核心消息协议、路由匹配、配置管理 ├── hermes_registry/ # Agent注册中心、健康检查 ├── hermes_orchestrator/ # 任务编排器、DAG执行引擎 ├── hermes_transport/ # 消息队列适配层支持RabbitMQ/Redis Stream ├── hermes_agents/ # 内置示例Agent ├── hermes_observability/ # 链路追踪、日志、指标 ├── hermes_web/ # 管理控制台API ├── docker-compose.yml └── config.yaml使用Docker Compose可以快速起一套依赖环境docker-compose up -d rabbitmq redis4.2 核心代码消息路由与Agent基类路由模块是消息层的心脏。这里给出简化后的核心思路核心逻辑是根据Topic找到对应消费者并转发消息。# hermes_core/router.py import asyncio from collections import defaultdict class Router: def __init__(self): self._subscribers defaultdict(list) # topic pattern - [handler] def subscribe(self, topic_pattern, handler): self._subscribers[topic_pattern].append(handler) async def dispatch(self, message): matched False for pattern, handlers in self._subscribers.items(): if self._match_topic(pattern, message.topic): matched True for handler in handlers: # 每个handler在本Agent独立任务队列中执行互不影响 asyncio.create_task(handler(message)) if not matched: self._handle_unroutable(message) staticmethod def _match_topic(pattern, topic): # 支持 * 匹配一级、# 匹配多级类似MQTT主题规则 if pattern #: return True pattern_parts pattern.split(.) topic_parts topic.split(.) if len(pattern_parts) len(topic_parts) and pattern_parts[-1] ! #: return False for p, t in zip(pattern_parts, topic_parts): if p #: return True if p ! * and p ! t: return False return len(pattern_parts) len(topic_parts)Agent基类封装了消息处理的公共逻辑业务Agent只需要继承并实现handle_message方法。# hermes_core/agent.py import json import uuid class BaseAgent: def __init__(self, agent_id, transport, registry): self.agent_id agent_id self.transport transport self.registry registry async def start(self): await self.registry.register(self.agent_id, self.get_capabilities()) for topic in self.get_subscribed_topics(): self.transport.subscribe(topic, self._on_message) self.transport.register_consumer(self) async def stop(self): await self.registry.deregister(self.agent_id) async def _on_message(self, envelope): try: result await self.handle_message(envelope) if envelope.message_type request and envelope.reply_to: await self.reply(envelope, result) except Exception as exc: await self.publish_error(envelope, exc) async def publish(self, topic, payload, message_typeevent, **kwargs): envelope { message_id: str(uuid.uuid4()), trace_id: kwargs.get(trace_id), topic: topic, message_type: message_type, source_agent: self.agent_id, payload: payload, timestamp: int(time.time() * 1000), ttl: kwargs.get(ttl, 60000), } await self.transport.publish(envelope) async def reply(self, request_envelope, result): await self.publish( request_envelope.reply_to, result, message_typeresponse, trace_idrequest_envelope.trace_id, ) async def handle_message(self, envelope): raise NotImplementedError def get_capabilities(self): return [] def get_subscribed_topics(self): return []这段代码虽然结构简单但有一个设计细节值得单独说_on_message里用异常捕获把异常转换成error消息而不是吞掉。这样编排器可以收到失败信号并做补偿处理而不是等超时。生产环境中这个设计救了我好几次因为它让失败变得可见、可追踪。4.3 编排器的DAG执行引擎编排器接收请求后会加载对应的DAG定义按依赖关系执行各个节点。DAG定义使用YAML配置一个典型的双Agent协作流程长这样# workflows/image_with_caption.yaml name: image_caption_workflow version: 1.0 nodes: - id: generate_image capability: image_generation timeout_ms: 30000 - id: generate_caption capability: text_generation timeout_ms: 20000 - id: validate_result capability: content_validation timeout_ms: 10000 edges: - from: generate_image to: validate_result - from: generate_caption to: validate_result执行引擎的核心是拓扑排序和消息派发。每个节点完成后将结果存入状态存储然后检查下游节点是否满足执行条件。这个实现借鉴了流式数据处理里的Batch同步思想只不过处理单元从数据块变成了Agent消息。4.4 链路追踪与日志关联可观测性是这个项目里我最坚持的部分。每条消息从进入系统的第一刻起就携带trace_id所有日志、指标、链路信息都以trace_id为关联键。{ timestamp: 2025-01-15T10:23:45.123Z, level: INFO, trace_id: d8f3a2e1-9c4b-4f6e-9a2b-1c2d3e4f5a6b, message_id: b2e4f6a8-1234-4c5d-9e0f-abcdef123456, agent_id: agent_image_gen, event: agent.message.received, topic: task.image.generate, latency_ms: 23 }实际排查问题时只需要在日志系统里搜索trace_id就能看到这条消息经历了哪些Agent、每一步耗时多少、最终是成功还是失败。这个能力在生产环境里至关重要尤其是当一条任务链跨了四五个Agent时没有链路追踪你根本不可能定位到瓶颈。有次一个用户反馈说“生成文案偶尔超时”我们就是通过链路追踪发现超时不是文案Agent本身慢而是图片Agent并发太高导致消息队列积压文案Agent等前置依赖等了几秒。链路一拉出来问题一目了然。4.5 实际部署与运行效果我用一套双Agent协作Demo验证了系统运行效果。一个图片生成Agent一个文案生成Agent编排器串联两条链路模拟100个并发请求。实际运行结果如下指标数值总请求数100成功完成数98平均端到端耗时3.2秒P95端到端耗时5.8秒消息吞吐量约180 msg/sAgent故障恢复时间约5秒失败的2个请求是图片生成Agent所在服务内存溢出导致但关键点在于——这两个请求没有丢失消息在队列中保留服务重启后消息被重新消费并处理成功。这就是消息驱动架构容错能力的直接体现。5. 常见问题与排查技巧实录5.1 消息消费重复怎么处理消息队列的at-least-once投递语义决定了消费重复是常态。同一个Agent崩溃重启后队列里未被确认的消息会被重新投递如果不做幂等就会出现重复处理。解决思路有两层。第一层是消息级幂等每条消息带唯一message_idAgent处理前先查询是否已处理过该ID处理过则直接返回上次的结果。第二层是业务级幂等比如“生成图片”这个动作设计成可覆盖式写入重复生成只是覆盖旧文件不会产生脏数据。我强烈建议所有Agent在处理消息前都加一层幂等检查。这个习惯能帮你避开海量重复消息引发的连锁Bug。5.2 消息堆积导致延迟飙升有次我压测时发现Agent消费速度跟不上生产速度队列积压越来越多端到端耗时从3秒飙到30秒。排查流程如下。先看消费者速度查看每个Agent的latency_ms和消费速率指标确认是哪个环节成为瓶颈。再看队列深度用RabbitMQ管理页面查看对应Queue的堆积数量确认积压规模。然后看下游依赖如果Agent本身不慢检查它调用的外部API或模型服务是否超时。那次的问题根源是某Agent调用的外部模型服务在高峰期响应变慢单个请求耗时从2秒涨到10秒消费者实际吞吐大幅下降。解决方案是给外部调用加熔断和快速失败。一旦外部服务响应超过阈值立刻返回错误而不是无限等待让编排器走重试或降级分支。5.3 编排任务卡死不动怎么办DAG执行中偶尔出现“任务卡住”的情况表现为编排器收到请求后长时间无后续消息。排查思路先确认是否为死信查消息是否因TTL过期被丢弃如果是调整TTL配置。然后确认是否为节点悬挂查对应Agent的日志看是否异常退出但没有发送错误消息。最后确认是否为拓扑配置错误检查DAG定义是否存在环或孤立节点拓扑排序失败会导致整个流程无法启动。一个我印象很深的坑是某次新加Agent时它的消费Topic写错了层级导致消息被路由到通配符匹配的其他Agent原Agent完全没收到。排查时在注册中心看到的订阅关系是正常的但实际运行的Router里订阅关系没刷新。重启Router后问题消失。这个Bug的教训是改配置后一定要验证运行中的路由表别只改文件不重载。5.4 消息序列化兼容性Agent迭代过程中消息结构经常变化新增字段没问题但删除或改名老字段会导致老版本Agent反序列化报错。我的经验是两个原则。一是只增不改新需求通过新增字段实现不修改旧字段语义。二是消费端容错反序列化时对缺失字段给默认值不强制要求全字段存在。这样能保证新旧版本Agent在混跑阶段不会互相踩坏消息。5.5 问题排查速查表问题现象可能原因排查/解决办法消息发出但无Agent处理Topic订阅不匹配检查Router的订阅关系确认Topic层级一致消息被重复处理消费后未及时确认开启手动ACK处理成功后再确认Agent状态偶发不一致注册中心缓存过期缩短心跳间隔强制刷新注册表端到端延迟越来越高消费者处理慢或队列堆积查看Agent耗时指标增加消费者实例或优化外部依赖编排流程无响应DAG拓扑错误或节点悬挂核对DAG定义检查Agent异常日志Agent重启后消息丢失队列未持久化配置持久化队列消息发布时标记持久化6. 项目扩展方向与个人心得6.1 从单机到多活的演进思路目前hermes-agent的默认部署是单集群模式注册中心、路由层都在一个逻辑域内。如果业务规模继续扩大一个值得探索的扩展方向是多活部署。思路是给每个部署单元加一个region标签注册中心按region做路由亲和。消息发布时优先路由到同region的AgentRegion内没有合适消费者再跨区转发。这样既能降低跨机房延迟又能保障单Region故障时整体可用。但需要注意跨区转发会引入网络抖动和消息延迟需要给TTL留足余量。另一个方向是Agent能力升级时做到平滑迁移。利用注册中心的能力抽象匹配新版本Agent注册成功后旧版本Agent定向消息全部切换到新版本业务方无感知。这个我已经在代码里预留了接口实现时只需要在注册中心增加一个preferred_version配置即可。6.2 补充一点我对Agent框架选型的思考做这类中间件时很多人会纠结“为什么不用现成的Agent框架”。我自己的体会是通用Agent框架核心解决的是“单Agent怎么思考、怎么调工具”而hermes-agent解决的是“多Agent怎么组织、怎么协作”。两者关注点不同可以组合使用。实际项目中我通常会用LangChain或自研Reasoning Loop处理单Agent内部的思考决策然后把Agent的输入输出封装成标准消息接入hermes-agent的消息网络中。这样既保留了单Agent的智能性又获得了协作架构的稳定性。这是一个被低估但非常有价值的组合方式。6.3 如果你要复刻这个项目我的建议不要一上来就追求大而全。先在一台机器上把RabbitMQ和Redis跑起来定义好消息信封结构写好一个Router和两个测试Agent跑通“Agent A发消息给Agent B”的最小闭环。这个最小闭环建立后再逐步加入编排器、注册中心、链路追踪。每一步都留出可观测的日志和指标因为消息驱动架构的调试难度远高于传统的直调模式可观测性不是加分项而是必需品。消息协议是系统最关键的部分宁可多花一周时间讨论设计也不要在上线后频繁改协议。协议一旦定稿所有Agent和编排器都会依赖它改动成本非常高。我刚做这个项目时在协议上吃过亏前后改了两次消息信封结构所有Agent都跟着动那种挫败感至今记忆犹新。最后一点心得多智能体协作的本质不是单个Agent有多聪明而是消息流转得有多顺畅、系统容错有多强。把消息设计和可观测性做好了哪怕每个Agent的逻辑都很朴素整个系统的稳定性和可维护性也远超那些单点智能很强但协作混乱的设计。这大概也是Hermes作为“信使”的真正意义——真正重要的不是谁在处理而是消息可靠地抵达了需要它的地方。
