轻量级多智能体编排框架hermes-agent：消息路由与任务编排实践

1. 项目定位为什么我非要自己写一个Agent框架先说结论hermes-agent不是那种大而全的AI应用平台而是一个为了解决多智能体协作时消息到底该怎么传、任务到底该交给谁这种具体问题而生的轻量级编排框架。最早萌生这个念头是我在用现成Agent框架做项目时踩了一连串坑。当时我需要在系统里同时跑好几个独立Agent分别负责意图识别、信息检索、内容生成和结果校验。理想很丰满但一落地就发现框架本身太重了。要么是把所有Agent写死在一条链路上想调整顺序就得改代码重新部署要么是通信机制黑盒出问题根本不知道是哪一步丢的消息最难受的是Agent之间想做个简单的你把结果给我我接着处理都要绕一堆抽象接口。当时我就想与其在这些框架的边界上反复试探不如自己写一套核心机制足够简单、但路由和编排能力足够灵活的方案。hermes-agent就这么来了。这个名字起得比较直白。Hermes在神话里是众神的信使负责传递消息、引导旅人。我的Agent框架干的事情也差不多把任务从发起方手里接过来按照路由规则送到该干活的那个Agent手上再把结果带回去。所以项目定位非常聚焦——不是去做模型调用层面的封装而是专注在消息路由、任务编排、Agent生命周期管理这三个方向上。适合谁来用它如果你正在做的项目处于下面这几个场景之一那这套思路和代码可能正好解你的渴手头有好几个各自独立的Agent服务互相之间没有固定的调用顺序每次要根据用户请求临时决定由谁处理。需要一个轻量级的中间层让前端、后端服务或者别的系统能通过统一入口调用多个Agent能力而不是每个服务自己接一套SDK。被各种重量级Agent框架的复杂配置折磨过想找一种核心机制简单、出了问题容易排查的编排方式。我不打算在这儿放PPT性质的能力清单后面章节会直接拆源码结构和核心流程。你把它当成一篇偏设计的实战笔记看就行有需要的话照着核心思路改一版属于自己的Agent调度器也完全可行。2. 整体设计思路把消息传递和业务逻辑彻底拆开2.1 核心抽象Agent变成消息处理单元在设计hermes-agent之前我先把市面上常用的Agent框架做了一次横向对比重点看了它们对Agent这个概念的定义方式。有一类框架把Agent视为带记忆的LLM调用链每个Agent内部包含prompt模板、模型实例、记忆模块Agent之间通过链式调用串联。优点是开箱即用缺点是灵活性差Agent内部结构绑得太死一旦想换模型或调整prompt就要动整个对象。另一类框架把Agent当成可独立部署的服务Agent之间通过网络协议通信。优点是天然分布式的缺点是对小项目来说太重了光处理网络连接、服务发现、消息序列化就够写一堆代码。hermes-agent选择了一条中间路线。我把Agent抽象成一个只认消息的处理器每个Agent对外暴露的接口极其简单输入一条标准格式的消息输出一条标准格式的消息。至于这个Agent内部是调LLM、跑Python脚本、查数据库还是调第三方API框架完全不关心。这个设计的直接好处是Agent的接入成本被压到了最低。现有代码不需要为了适配框架进行大规模重构只要写一个适配层把输入消息翻译成内部逻辑需要的数据格式再把执行结果包装成标准输出消息就行。2.2 消息协议让一切通信都建立在同一个格式上既然Agent之间是通过消息通信的那消息格式就必须先定好。我在设计消息协议时参考了HTTP和消息队列的做法保留必要的信息字段但把业务数据完全放开。核心消息结构如下dataclass class AgentMessage: msg_id: str # 消息唯一ID用于全链路追踪 trace_id: str # 链路追踪ID一次完整任务的所有消息共享 msg_type: str # 消息类型如 query / result / error / heartbeat source: str # 来源Agent名称 target: str # 目标Agent名称广播时可填 * priority: str # 优先级urgent / normal / deferred payload: dict # 业务数据Agent自行解析 created_at: float # 消息创建时间戳这个协议看起来简单但实际使用中几个字段的作用比想象中大得多。trace_id解决了分布式排查问题的老大难一次完整任务从用户请求到最终响应中间经过多少个Agent、每条消息的流转路径是什么拉着trace_id一查便知。priority字段则给调度器提供了按优先级处理消息的依据紧急请求可以先于普通任务被执行。2.3 注册机制Agent先注册消息后路由解决了消息格式之后第二个关键设计是Agent管理。我采用的是注册-发现-路由三段式机制。所有Agent启动后会向hermes-agent核心注册自己的身份信息、能力描述和当前负载状态核心维护一张动态路由表。每个Agent在注册时需要提供一个manifest描述自己能处理什么类型的消息。这个设计借鉴了微服务架构中的服务注册表思想好处是新增或下线一个Agent不需要修改其他任何代码。注册示例{ name: intent_agent, capabilities: [intent_recognition, query_rewrite], endpoint: grpc://localhost:50051, max_concurrency: 5, weight: 3, tags: {team: nlp, model: proprietary-v2} }注册表里存的不只是服务地址更重要的是capabilities字段。Router做消息分发时会根据消息内容里声明的需求能力去匹配注册表找到能处理该能力的Agent。这也为后面实现复杂的意图路由打下了基础。2.4 选型取舍为什么用Redis做消息队列而不是上Kafka在通信层选型时我一开始在Redis Stream和Kafka之间纠结过。Kafka在吞吐量、持久化和消费者组机制上确实强悍但引入Kafka对一个小体量Agent编排框架来说运维成本直接拉满——要额外维护ZooKeeper或者KRaft节点、磁盘分区配置、消费位点管理。而hermes-agent的目标场景是单机或少量几台服务器就能跑起来的多Agent协作Kafka属于用牛刀杀鸡。Redis Stream在这个场景下提供了一个不错的平衡点。它本身就是Redis的内置数据结构不需要额外部署其他组件读写性能足够支撑Agent间毫秒级的消息往返而且消费者组机制可以很好地支持多个Agent实例对同一队列进行负载均衡消费。唯一的顾虑是消息持久化问题Redis内存保存的话重启会丢数据但我处理缓存和可恢复任务时都不依赖消息持久化来保证一致性所以这个短板在实际使用中基本无感。我还做了另一层保险发送消息时如果Redis写入失败框架会自动降级到内存队列同时打印告警日志。这个降级机制保证核心流程不会因为缓存件故障就全部瘫痪。3. 核心模块拆解消息处理管线的每个环节3.1 Message BusAgent之间通信的环形跑道整个框架的心脏是一个基于消息总线的分发系统。我把它设计成一个环形结构消息从任意一个Agent发出后统一进入总线由总线根据路由表决定下一步去哪里而不是Agent之间直接点对点通信。这里有一个非常反直觉的设计Agent之间不直接通信。所有消息都发送到总线总线来做转发。一开始团队成员不理解觉得这不是白白增加了一层中转吗我解释了很久才达成一致。Agent直接通信看着效率高实际上会让系统的依赖关系变得错综复杂。A调BB调CC又调A一旦某一个环节出问题排查起来就是一场灾难。而总线模式天然地把所有通信路径集中管理以一种收敛的方式来组织。哪怕系统里跑着几十个Agent所有的消息流转都在这一个地方可以追踪到再加上消息里携带的trace_id每个Agent处理了多长时间、在哪一步延误了全都一目了然。3.2 Router怎么知道这条消息该给谁Router是决定消息去向的策略器。在实现上我提供了几个不同层级的匹配机制第一层是显式路由。消息里写明了target字段Router直接根据Agent名称匹配把消息投递过去。这个最简单也最可靠。第二层是能力路由。消息没有指定具体Agent而是声明自己需要的能力。Router通过加载Agent注册表构建的一个能力倒排索引来匹配先查这个能力有多少Agent注册了再根据这些Agent当前的负载权重来分配。举个例子系统配了模型服务A和模型服务B能力都是text_generationRouter会把新请求按权重比例分发避免某一台机器被打满。第三层是意图路由。这一层就涉及一点语义匹配了。有些Agent注册时声明的不只是capabilities里的几个能力词而是一段对自身处理范围的语义描述。Router在收到无法用前两层匹配的消息时会对消息内容做轻量的分类再拿去和Agent的语义描述算相关性取分最高且超过阈值的那个作为目标的候选。我自己实际项目里跑得最多的还是前两层。意图路由对语义分类模型有依赖响应时间会比纯规则匹配慢几十毫秒如果对性能敏感的团队建议慎开第三层。3.3 Scheduler任务的拆解、分配与状态跟踪Router解决的是单个消息往哪走的问题Scheduler解决的是一个完整的任务怎么拆解成多个消息并按顺序派发的问题。Scheduler里预置了几种编排模式。第一种叫流水线模式把任务拆成step1到stepN严格按顺序执行上个Agent的输出作为下个Agent的输入。适合内容生成后接审核这样的固定链路。第二种叫扇出模式一个任务拆成多个并行子任务同时派发给多个Agent执行。比如一份合同需要同时做条款审查、金额核对和合规检查三个Agent互不依赖全跑完后由下游收集器合并结果。第三种叫动态依赖模式任务的子步骤在运行时才确定。这种模式最复杂也最灵活适合一个任务里某个步骤的结果会影响后续步骤到底执行不执行这样的场景。每个任务在Scheduler里都有一个对应的任务状态机状态流转为pending - dispatching - running - completed / failed / timeout。所有状态变更同时写入Redis支撑外部系统通过API实时查询任务进度。3.4 Memory StoreAgent的临时记忆与状态保存还有一个容易被忽视但实际很重要的模块——Memory Store。多Agent协作时经常出现这种情况一个Agent的处理结果后面好几个环节都要用。如果没有共享存储就得把数据塞在每一跳的消息里一路传下去消息体越来越肥还容易超出传输上限。我把Memory Store设计成一个带命名空间的临时缓存Agent在处理消息的过程中可以把中间结果写入Memory Store后续其他Agent按key取用。命名空间对一次完整任务同一个trace_id隔离。这个机制还能支持一种简单的跨Agent上下文保留Agent在收到一条新消息处理时先去Memory Store检查有没有历史上下文有的话加载进来用完了再更新回去。这个模块在单机场景下我直接用进程内LRU缓存实现但部署到多实例时Memory Store底层需要替换成Redis这样才能保证不同机器上的Agent访问的是同一份记忆。4. 部署与实操从拉代码到跑通一个多Agent协作任务4.1 环境准备与快速启动项目整体的运行依赖很克制。建议环境配置如下组件版本要求用途Python3.10框架主语言Redis6.2消息总线与记忆存储FastAPI0.100对外提供HTTP APIpydantic2.0消息模型校验启动方式很常规# 安装依赖 pip install -e . # 启动核心服务默认监听8000端口 python -m hermes_agent.server start --config configs/default.yaml # 注册一个Agent以本地文件注册方式为例 python -m hermes_agent.cli register --manifest examples/intent_agent.json注意第一次跑通的话建议先按默认配置来Redis的地址和密码改成本地环境就行。不要一上来就动消息协议、消费线程数这些参数等整个链路通了再逐个调优不然出了问题根本分不清是配置错误还是代码逻辑错误。4.2 动手写一个自己的Agent接入一个自定义Agent需要在代码里实现底下的包装类from hermes_agent.agent import BaseAgent class MySummarizeAgent(BaseAgent): async def handle_message(self, message): # 取消息里的业务字段 content message.payload.get(content, ) # 调用自己的模型服务或业务逻辑 result your_summarize_function(content) # 返回标准输出消息 return self.reply( message, msg_typeresult, payload{summary: result} ) if __name__ __main__: agent MySummarizeAgent( namesummarize_agent, capabilities[text_summarization] ) agent.run()这个类唯一必须实现的方法是handle_message。返回的reply消息会自动填好msg_id、trace_id和source、target字段不需要手动维护消息之间的关联关系。4.3 模拟一个完整的多Agent协作流程为了让你直观感受这套机制怎么运转我模拟一个具体任务用户提交一篇技术文章要求做三件事——提炼摘要、提取关键词、判断内容质量是否合格。对应的三个Agent分别是summarize_agent、keyword_agent和quality_agent注册好能力后外部系统只需要发起一个任务curl -X POST http://localhost:8000/api/tasks \ -H Content-Type: application/json \ -d { task_type: content_analysis, payload: { article_content: ……待分析的技术文章正文…… } }Scheduler接收到这个任务后按照content_analysis这个任务模板里配置的编排逻辑先发一条摘要任务给summarize_agent同时扇出两条并行任务给keyword_agent和quality_agent。三个Agent处理完的结果由Scheduler合并成最终响应返回给调用方{ task_id: task_8f3a12, status: completed, result: { summary: ……, keywords: [hermes-agent, 消息路由, Agent编排], quality_score: 0.86 } }整个过程中每条消息都带着同一个trace_id通过查询接口可以看到时序图式的流转记录。由于Scheduler的总控模式最终响应不管中间有多复杂的并行关系对外暴露的都是一个同步接口调用方不需要自己去聚合多个异步回调。4.4 任务模板的配置方式上面的示例里提到了任务模板这个概念。编排逻辑在hermes-agent里不是写死在代码里的而是通过声明式配置。拿content_analysis举例它的YAML配置长这样task_type: content_analysis steps: - id: summarize capability: text_summarization mode: pipeline - id: extract_keywords capability: keyword_extraction mode: fan_out depends_on: [] - id: check_quality capability: quality_check mode: fan_out depends_on: [] - id: merge_result mode: merge depends_on: [summarize, extract_keywords, check_quality]这种声明式配置的好处是改流程不需要改代码。今天想让summarize和check_quality并行执行把depends_on改一改重新加载就行Agent本体一行不用动。我自己的经验是把任务编排逻辑从代码里剥离出来之后维护成本和犯错的概率都显著下降。4.5 性能与并发配置上线前有一项必须做的测试——压测。我用Locust简单跑过一组数字供参考。单机部署8核16G机器Redis同机默认配置下10个Agent实例同时消费消息吞吐大概在每秒2800条上下单个消息从入队到被消费的平均延迟约12毫秒。这个量级对绝大多数中后台Agent编排场景是够用的。如果要追求更高的吞吐优先调整下面几个参数scheduler: queue_capacity: 5000 # 队列容量超过后触发背压 max_inflight_tasks: 200 # 同时处理中的任务数上限 consumer: thread_pool_size: 16 # 消费线程池大小 poll_batch_size: 32 # 每次批量拉取消息数 poll_timeout_ms: 200 # 拉取超时这几个参数是一组互相牵制的组合线程池越大单位时间能处理的消息越多但也会对Redis和下游Agent造成更大的压力max_inflight_tasks限流靠的是排队设太低了会拖高延迟设太高了又可能瞬时打爆下游服务。建议按下游Agent的最大承受并发数来倒推设置而不是盲调。5. 上线三个月落地中遇到的坑与排查实录这一节整理的每个问题都是我自己在项目里真实踩过并修复过的。放在这里当排查速查表用能省不少时间。现象可能原因处理方法消息发出后Agent一直没反应目标Agent未成功注册检查agent list命令输出确认Agent连接状态为online消息被Router丢弃无任何日志能力路由匹配失败开启Router的debug级日志确认消息的能力声明与Agent的manifest是否一致Agent回调成功但任务状态一直pending回调消息缺少trace_id检查Agent回复消息时是否使用了self.reply()手写消息容易漏字段高并发时出现消息重复消费消费者处理超时后Redis Stream重新投递业务接口设计为幂等或在消费前检查任务状态任务整体超时但单个Agent耗时正常扇出模式下存在慢节点对并行分支设置独立超时时间慢分支失败不影响整体任务Redis重启后消息丢失Stream未启用持久化启用AOF持久化或将关键消息同时落地MySQL备份5.1 死循环问题Agent链路中的环路风暴上线后遇到最惊险的一次事故是线上出现Agent互相循环调用。有个链路设计是A处理完发给BB处理完再发给A做二次校验。逻辑上这个闭环是合理的但代码里A做二次校验时漏了一个关键判断——没有识别出这条消息其实是我自己上一轮发的。结果就是A发给BB处理完回传给AA又当成新任务发给BB又回传……形成了一个无限循环直接把Redis的消息队列打爆了。后来我做了三层防护才彻底解决这个问题跳数限制TTL每条消息带一个max_hops字段每经过一个Agent减一归零后消息直接进入死信队列。这是最后一道防线。去重机制在处理入口处对msg_id做幂等判断已经处理过的消息直接丢弃。链路守卫cycle detection消息的流转路径会持续记录如果检测到source和target序列出现重复循环立刻终止任务并告警。这三层目前在实际运行中还没被触发过属于平时用不上、但绝对不能没有的保险。5.2 慢消费导致的连锁雪崩还有一次很典型的性能事故。某个Agent依赖的模型服务响应时间突然从300毫秒涨到3秒一开始只是这个Agent变慢但因为它占据着消费线程池里的线程不释放整个系统的可用消费线程数越来越少消息积压越来越重其他Agent的合作任务也被拖住最终表现为全链路大规模超时。这个坑教会我一件事任何下游依赖都可能变慢而消费框架必须对这种情况有专门的设计。修复方案是在消费端增加一个基于信号量的隔离机制每个Agent独立分配允许的最大并发数某一路径用的线程数超过配额时新消息直接进入等待队列而不是抢占其他Agent的线程。这样上游服务再慢也只会积压它自己那一路的队列别的链路不受影响。5.3 消息体膨胀问题消息协议里payload是自由的dict用起来方便但也很容易失控。有个同事在业务迭代中不断往payload里塞数据从最初的几个字段膨胀到几十个字段单个消息直接从2KB涨到了30KB。在QPS不高的时候没什么感觉但压测一上量Redis的内存占用急剧上升序列化和反序列化耗时也成倍增加。在这个问题上我定了一条编码规范payload里只放必要的数据任何能从共享存储或外部系统拉取的内容一律放引用。比如一个大文件的内容不放进payload而是放一个object_idAgent拿到ID后去对象存储读取。规范定了之后“消息体膨胀”这个类问题基本绝迹。5.4 数据一致性Agent失败后怎么补偿Agent执行失败时Scheduler会按照配置的策略做处理。默认策略是标记失败并停止整个任务但业务上经常需要某个步骤失败后执行补偿逻辑。比如步骤A生成了一段营销文案步骤B做合规审核时发现文案里有违规词希望系统能回到步骤A把违规词屏蔽后再重新提交审核而不是直接判整个任务失败。这类回退重试的需求我在Scheduler里加了compensation_handler配置给每个步骤声明一个补偿回调地址。如果下一步骤检测到异常会触发回退把状态重置到指定步骤重跑后续流程。这个机制还不能说做到完全自动因为重跑的前提是下游步骤的资源都是可重入的比如限制在有效期内幂等创建资源但至少把需要人工介入的边界缩小了很多。6. 可扩展性探索从多人协作到跨团队复用6.1 插件机制Agent的即插即用做到了这一个版本之后我开始思考一个问题能不能让不同团队各自开发Agent不经过我这个核心维护者的代码审查就能直接接进来用答案是要做成插件化的。具体方案是定义一套标准的Agent打包格式每个Agent以独立进程方式运行通过gRPC与核心通信。核心只负责注册管理和消息转发不加载Agent的任何业务代码。Agent打包格式 - manifest.yaml # Agent元信息与能力声明 - proto/ # 自定义消息类型的定义 - entrypoint.sh # 启动脚本 - health_check.sh # 健康检查脚本这个方案好在哪里不同团队可以用完全不同的技术栈开发自己的Agent只要保证消息格式合法、注册信息正确就能纳入统一调度体系。核心与Agent之间通过gRPC的双向流式通信既可以做传统的请求-响应也能支持Agent主动向总线推送异步事件。这为后续跨团队、跨微服务共享Agent能力留了很好的空间。6.2 与外部系统的集成开发过程中我发现很多用户并不是直接使用API而是希望把Agent框架嵌入到自己已有的技术栈里。所以我对外提供的API里专门做了一套事件订阅机制外部系统可以订阅所有Agent消息流。订阅的方式也很简单在配置中声明一个Webhook地址每当有指定类型或指定Agent的消息产生核心会向这个Webhook推送消息内容。webhooks: - name: monitor url: http://your-monitor-service/hook events: - task_completed - task_failed - agent_dead这套机制本身不复杂但带来的效果是业务系统可以方便地拿到Agent执行过程中的实时状态做监控、做审计、做数据回流都变得非常容易。6.3 未来的规划我自己的规划是下面三条线并行推进。编排能力可视化现在任务编排全靠写YAML不够直观。计划做一个Web UI用拖拽的方式画出任务流程图自动生成配置文件。更智能的路由策略基于请求特征的动态路由不只依赖静态权重而是根据Agent的历史表现、当前负载、响应耗时这些指标实时调整分发权重。更细粒度的多租户隔离现在命名空间层面做了整体隔离但不同业务线共用一套Agent资源时需要在配额、限流、审计上进一步做精细化管理。7. 写在最后的个人体会hermes-agent这几个月做下来我最深的一个感受是框架的设计目标不是把功能做得多复杂而是要让消息流转这件事足够透明。Agent本身可以随便换、随便升级但只要消息协议和路由机制稳定整个系统就不会乱。还有一个体会是做这种偏底层的编排组件一定要宁可少做一些花哨功能也要把排查问题的体验做好。很多时候用户反馈不好用本质上不是功能缺失而是出了问题不知道从哪里查。有了trace_id全链路追踪和清晰的消息流转日志很多框架不好用的抱怨其实都能化解掉。最后再分享一个实际使用的小技巧消息总线上所有Agent的输入输出日志默认都是全量打印的这在开发和测试阶段很舒服但上线后日志量会大到失控。建议生产环境关掉消息体打印只保留元信息消息ID、Agent名称、耗时、状态把消息体内容通过采样机制按比例记录。这样既能保证链路可追踪又不会让日志系统被淹没——我一开始没注意结果第一版上线半天日志就吃掉了20GB磁盘这个教训记忆犹新。