hermes-agent实战:消息路由驱动多Agent协作编排
2026/9/8 18:20:34 网站建设 项目流程

做 Agent 开发的朋友,最近应该都被同一个问题折腾过:单 Agent 跑通很容易,多 Agent 一上,消息满天飞,工具调用一会儿串了,一会儿超时,上下文里还混着别家 Agent 的历史记录,最后翻日志得靠关键词搜索碰运气。我手里这个 hermes-agent 项目,从一开始就是奔着解决这类“编排混乱”去的。

hermes-agent 是一个轻量级的 Agent 运行时框架,核心思路很简单:把每个 Agent 当成独立的消息处理节点,Agent 之间不直接互调,而是通过统一的路由层收发消息。你只需要告诉它“什么样的消息交给哪个 Agent 处理”,剩下的消息分发、工具调用、上下文隔离、链路追踪,都由它接管。对正在自己拼 Agent 框架,或者项目里 Agent 数量一多就开始失控的团队来说,这是一个非常值得参考的编排方案。下面我把这个项目从设计思路到落地细节完整拆开讲,包括路由规则怎么写、工具怎么注册、多 Agent 协作怎么配置、常见坑怎么填。

1. 为什么叫 Hermes:先搞清楚这个 Agent 框架到底想干什么

1.1 项目定位:它不是一个“大模型一揽子套件”

首先得说清楚,hermes-agent 不是什么。它不训练模型,不搞 RAG,也不是拿来跑一个 ChatBot 的“一键安装包”。它做的事情非常聚焦:负责 Agent 与 Agent 之间、Agent 与工具之间的消息调度和路由。希腊神话里 Hermes 是神的信使,专职传递消息,这个名字起得很贴切——这个框架在 Agent 生态里扮演的就是“信使”的角色。

我用一个生活化类比来解释。你想象一家外卖平台,商家是各种 Agent,骑手是消息,点餐用户是调用方。如果每家商户都直接派自己人去送餐,马路上全是各家的骑手,订单容易送错,路线也没法统一协调。hermes-agent 的做法是当那个集中调度平台:商家把餐(消息)交给平台,平台根据地址(路由规则)分配给最合适的骑手(Agent),你不需要管骑手怎么走,只需要告诉他送到哪里。

这也解释了为什么这类框架更适合多 Agent 场景,而不是单 Agent 单场景。如果你的项目只有一个 LLM 节点、一个工具调用,根本不需要中间层;但一旦你有多个 Agent——比如一个负责意图识别,一个负责检索知识库,一个负责总结输出——它们之间的消息流就需要一个明确的管理者,否则代码很快会变成一团粘在 main 函数里的意大利面条。可能有人会觉得“多一层中间件太重”,但等你真的在三个 Agent 之间手动维护消息传递时,就会明白中间层省下的不只是一点点心智负担。

1.2 核心设计理念:消息即事实,路由即逻辑

hermes-agent 最核心的设计理念,是把“逻辑判断”从代码里抽出来,放进“消息路由规则”里。传统写法里,你通常是这样做的:

# 传统写法:判断逻辑散落在业务代码里 if intent == "search": results = retriever.run(query) if results: answer = summarizer.run(results) else: answer = fallback_agent.run(query)

这种写法的问题在于,每加一个 Agent、每多一种消息类型,这段 if-else 就会膨胀一轮。而且当多个 Agent 可以互相转发消息时,嵌套判断基本没法维护。你只能通过不断加 flag 和分支来应对新需求,改一处代码就得担心破坏另一条链路。

换成消息路由的思路后,每个 Agent 只负责处理自己关心的消息,处理完把结果再次封装成消息发出去,由路由层根据规则把消息送到下一个节点:

# hermes-agent 风格:Agent 之间通过消息驱动协作 class RetrieverAgent(Agent): def handle(self, message: Message): if message.type == "retrieve_request": results = search_tool(message.payload["query"]) self.emit(Message(type="retrieved", payload=results))

这个 Agent 根本不知道下一个节点是谁。它只关心“我收到了什么、我处理了什么、我产出了什么”。谁需要这个结果,由路由层决定。这就是“消息即事实,路由即逻辑”的含义:数据流是清晰的,决策是可配置的。

这个设计带来几个直接好处。第一,新增 Agent 不需要改动已有 Agent 的代码,只要加一条路由规则;第二,任意 Agent 可以被复用,A 项目里跑过的检索 Agent 拿过来就能用;第三,链路清晰,每条消息从哪来到哪去都能追踪;第四,出问题时可以单独替换或重试某个节点,不影响整条链路。当然,代价也有:多了一层抽象,对极简场景来说确实显得“重”。另外路由规则如果设计得不好,排查消息流向反而比看代码更头疼。这些后面我会展开讲。

2. 核心细节解析:路由规则、工具注册与上下文管理

2.1 路由规则怎么写:把“该谁处理”从代码里拆出来

hermes-agent 里的路由规则,核心就两块:消息匹配和目标节点。先看消息本身,一条标准消息一般包含 type(类型)、source(来源)、target(可选的目标)、payload(载荷)、trace_id(链路追踪 ID)这几个字段。路由层最核心的匹配依据就是 type 和 target。

项目的初期,我没有用 YAML 配置路由,因为那会儿消息类型变化太快,用代码配置更灵活。实现方式是在运行时里注册规则:

runtime.route( topic="retrieve_request", # 订阅哪个类型的消息 to="retriever", # 转给哪个 Agent description="检索请求统一交给检索Agent", )

路由匹配的优先级严格按照注册顺序,先匹配先命中。这个顺序非常重要,后面排查路由不生效时会再次提到。

这里有几个实际经验值得记下来。第一,消息 type 的命名一定要克制,推荐统一风格:{verb}_{domain},比如retrieve_requestretrieve_resultsummarize_request。我见过项目里出现searchdo_search_pleasego_retrieve这种随心所欲的命名,后来光统一命名就花了半天。第二,建议尽量少用 target 字段,多用 type 来触发路由,因为 type 表达的是“这个消息是什么”,target 表达的是“这个消息给谁”,前者更符合解耦思路,也更容易复用。第三,路由规则可以带条件,比如按 payload 里的租户 ID 前缀做分流,但初期不建议搞太复杂,靠 type 就能覆盖绝大多数需求。

2.2 工具注册:让 Agent 学会用你的函数,而不是重新发明轮子

Agent 的 LLM 能力只能做“理解与生成”,要真正落地到业务,必须让它调用真实工具——查库存、发通知、读文件、调用第三方 API,都算。hermes-agent 对工具调用的处理方式很朴素:工具就是一个带输入输出定义的函数,注册到运行时后,Agent 统一通过路由消息去调用它。

我拿一个天气查询工具举例:

from hermes import tool @tool( name="get_weather", description="按城市名查询当前天气,返回温度、天气状况、湿度", input_schema={ "city": {"type": "string", "required": True, "desc": "城市名,如:上海"}, } ) def get_weather(city: str): # 这里是你的真实业务逻辑 return weather_api.fetch(city)

注册工具后,Agent 发起的工具调用请求会被路由到 ToolExecutor,由它解析工具名和参数、执行并返回结果。这个过程中,我强烈建议把工具调用也当作消息看待,而不是“函数直接调用”。原因只有一个:消息化之后,工具调用记录才能统一进入链路追踪,你才能回看某个 Agent 当时到底调了什么工具、传了什么参数、返回了什么结果。这对排查多 Agent 协作问题价值极大。

工具注册有几个细节必须注意。一是 input_schema 必须写清楚必填字段和字段描述,因为 LLM 生成参数时极度依赖描述文字,描述写得模糊,它就会脑补出奇怪的值。二是工具函数要尽量保证幂等,尤其是“发邮件”“创建订单”这类有副作用的操作,重试时不能造成重复动作;我通常会给这类工具加幂等键参数。三是返回结果尽量结构化,不要返回一长串人类可读的文本,给 LLM 的结构化 JSON 会更稳定,token 消耗也更小。

2.3 会话上下文:隔离还是共享,这是策略问题

多 Agent 协作里,上下文管理是最容易翻车的地方。A Agent 检索了半天的中间结果,B Agent 那边也能看到,通常不是你想要的效果;但某些全局信息,比如用户 ID、会话 ID,你又希望所有 Agent 都能拿到。hermes-agent 把上下文分成了两层:全局上下文和工作流上下文。

全局上下文保存跨 Agent 共享的信息,比如当前用户身份、租户配置、公共参数,它在整个运行时生命周期内有效,但一般不会直接填入模型输入,更多是作为执行环境信息存储。工作流上下文则是一次请求链路内的局部上下文,默认只有产生这条消息的 Agent 能写入,但它可以显式声明给下游 Agent 读取。这个设计很实用:上游做完检索,把精简后的摘要放进去,下游总结 Agent 只管消费,不必看到上游几十条原始检索结果。

实际使用中我的建议很明确:默认情况下,工作流上下文里不要放原始大文本,尽量放“处理后的结果”和“业务元数据”。原因很简单,LLM 的上下文窗口有限,如果你把检索出来的 50 条全文都塞进去,下一个 Agent 的上下文基本就废了。在项目里我经常看到有人把中间结果原封不动往下传,明明上游已经有提取、压缩能力,不用就是浪费。

3. 实操过程与核心环节实现:搭一个多 Agent 协作流水线

3.1 环境准备与项目引入

前面说的都是设计思路,接下来我们动手。我这里用的是 Python 运行时版本,安装非常简单:

pip install hermes-agent

项目依赖主要有两样:一是 pydantic,用于消息体校验;二是任意调用 LLM 的客户端 SDK,OpenAI、Ollama、通义等都可以,看你自己用哪家。它本身不强绑定某个模型厂商,你只需在自己写的 Agent 内部调用模型即可。因为 hermes-agent 只解决“消息怎么流转”的问题,不替代你选择模型。

初始化运行时也很简单,下面就是最基础的启动入口:

from hermes import AgentRuntime runtime = AgentRuntime() if __name__ == "__main__": runtime.start()

运行时启动后会初始化内部的消息队列、路由表和工具注册中心。此刻这个项目还是空壳,我们要往里面注册 Agent 和路由,跑通一个“检索-总结”的协作链路。

3.2 配置一个多 Agent 协作流水线

我们的场景是这样的:用户输入一个问题,RouterAgent(入口 Agent)先判断这个问题是否需要外部资料。如果需要,就发一条 retrieve_request 给 RetrieverAgent;RetrieverAgent 查询检索工具拿到结果,发一条 retrieved 消息;最后由 SummarizerAgent 汇总生成答案。先定义入口 Agent:

class RouterAgent(Agent): name = "router" def handle(self, message: Message): query = message.payload["query"] # 用 LLM 判断是否需要检索 decision = self.llm.judge(query) if decision == "need_search": self.emit(Message( type="retrieve_request", source="router", payload={"query": query, "trace_id": message.trace_id}, )) else: self.emit(Message( type="summarize_request", source="router", payload={"query": query, "context": []}, ))

注意,这里 emit 出去的 payload 除了业务字段,一定带上 trace_id。trace_id 是整个链路追踪的钥匙,没有它,日志根本串不起来。后面排查问题时会反复用到这个字段。

然后是 RetrieverAgent 和 SummarizerAgent。Retriever 的核心是调用检索工具拿到资料,把它封装成消息发出去:

class RetrieverAgent(Agent): name = "retriever" def handle(self, message: Message): if message.type != "retrieve_request": return results = self.tools["search_knowledge_base"].run(message.payload["query"]) self.emit(Message( type="retrieved", source="retriever", payload={ "query": message.payload["query"], "results": results[:5], # 只保留前5条结果,控制上下文长度 "trace_id": message.trace_id, }, ))

SummarizerAgent 接收 retrieved 消息,把检索结果交给 LLM 总结:

class SummarizerAgent(Agent): name = "summarizer" def handle(self, message: Message): if message.type == "summarize_request": query = message.payload["query"] context = message.payload.get("results", []) elif message.type == "retrieved": query = message.payload["query"] context = message.payload["results"] else: return answer = self.llm.summarize(query, context) self.emit(Message( type="answer", source="summarizer", payload={"answer": answer, "trace_id": message.trace_id}, ))

最后是路由注册和运行时提交:

runtime.register_tool(search_knowledge_base) runtime.register_agent(RouterAgent(runtime)) runtime.register_agent(RetrieverAgent(runtime)) runtime.register_agent(SummarizerAgent(runtime)) runtime.route(topic="retrieve_request", to="retriever") runtime.route(topic="retrieved", to="summarizer") runtime.route(topic="summarize_request", to="summarizer") runtime.submit(Message( type="user_query", payload={"query": "帮我查一下今年Q3的销售数据"}, ))

这样一个最简单的多 Agent 协作链就搭好了。流程是:入口判断需要检索 → 发检索请求给 Retriever → Retriever 调用工具并发出结果 → Summarizer 总结输出 answer 消息。跑通这个链路之后,你就可以继续加 Agent、加路由规则,扩展成更复杂的拓扑。

3.3 运行轨迹与链路追踪怎么看

多 Agent 系统跑通只是第一步,真正难的是“出了问题能不能快速定位”。hermes-agent 在运行时会给每条消息、每个 Agent 处理过程记录结构化日志,关键字段只有三个:trace_id、source、type。只要把日志集中收集,按 trace_id 一筛,就能看到一条请求从入口到出口经过哪些节点。

我自己用的时候习惯把 trace_id 打到业务日志的第一行,同时把工具调用参数、LLM 返回内容单独打一行。后续分析时就能确认“是 Agent 判断错了,还是工具返回错了,还是 LLM 生成错了”。看日志分三层去看:第一层看消息流,trace 经过了哪些 Agent,哪个节点丢消息了;第二层看决策,RouterAgent 当时为什么做了这个判断,看它传给 LLM 的 prompt;第三层看工具结果,工具返回的数据是否符合预期。这三层能直接定位 90% 的多 Agent 问题,剩下的才是模型本身的问题。

4. 常见问题与排查技巧实录

4.1 Agent 互相转发导致死循环

多 Agent 协作最常见的故障就是循环。A 发消息给 B,B 处理完又发回给 A,A 再发回给 B……本来这种情况可以通过条件控制,但某个条件在某种消息上永远不成立,消息就在节点之间弹来弹去,直到内存爆掉或者重试次数耗尽。

我在项目里真实踩过这个坑。当时两个 Agent 会互相发确认消息,正常逻辑是收到确认就停。结果其中一个 Agent 在确认消息里带了一个新的任务 ID,下游 Agent 看到任务 ID 就去执行任务,又把确认消息发回去,导致消息形成闭环,失败后无限重试。排查方法其实很简单:通过 trace_id 拉出整条消息链,看它是否在固定节点间反复横跳。如果是,优先检查两个方向的条件是否互斥。后来我加了两层保险:消息头里加 max_hops 字段,每经过一个节点减一,归零后直接丢弃并记录 warning;每个 Agent 处理消息前,先检查自己是否处理过来自同一 source 且 payload 相同的消息,如果处理过,直接返回空结果。

4.2 工具调用超时与重试风暴

Agent 调工具,工具调外部接口,任何一个环节超时,都可能引发重试风暴。默认重试策略是 3 次,指数退避。但如果你有多个 Agent 在并行调用同一个下游接口,接口一慢,可能同时涌出几十个重试请求,把本来就慢的服务彻底打挂。

处理这个问题我有两条经验。第一,在工具层加熔断:连续失败超过 5 次,就打开熔断开关,后续请求直接返回降级结果,不再打到下游。第二,超时时间要分层设置:外部 HTTP 请求超时建议 3 秒,但工具整体超时建议 10 秒,给 LLM 层的解析留出时间。你不能把工具超时和网络超时配成同一个值,否则一次网络抖动连重试机会都没有。还有一个容易忽略的点:工具调用超时后,LLM 会自动生成一个“看起来合理但实际瞎编”的降级回答。所以工具结果里必须带一个 success 标志,提示词里也要明确要求:只有 success=true 才信任工具结果,否则必须告诉用户“暂时无法获取”。

4.3 上下文窗口溢出:消息越攒越多

长时间会话和多跳协作有个绕不开的问题:上下文越来越多,最终超过 LLM 的窗口限制。我在项目里试过三种处理方案,按推荐程度排序如下:摘要替换、滑动窗口、关键信息提取。

摘要替换是当上下文超过阈值时,把最早的 N 条消息交给一个 SummarizerAgent 压缩成一段摘要,用摘要替换旧消息。成本可控,效果最好。滑动窗口只保留最近 K 条消息,实现最简单,但会丢失早期关键信息,对话逻辑容易断。关键信息提取则强制保留包含 tool_result 和 answer 的消息,丢弃中间判断类消息,适合多跳推理链,但依赖业务规则。特别提醒一句:摘要替换这个动作本身也要有 trace_id 关联,否则被压缩掉的旧消息和摘要之间没有对应关系,后续排查问题时会非常痛苦。

4.4 路由不生效:最隐蔽的开发期陷阱

配置了路由规则,但消息发出去了没有 Agent 接收,这是开发期最让人抓狂的问题。根因通常是三个。第一,消息 type 和路由 topic 对不上。比如你在 Agent 里 emit 的是 retrieve_request,路由注册的却是 retrieve。这种低级错误最容易出现在复制粘贴代码时,我建议在启动阶段加一条校验:所有 Agent 里 emit 过的 type,必须能在路由表中找到对应规则,否则启动时直接报警。

第二,Agent 注册顺序问题。路由规则是在注册 Agent 时绑定的,如果你先 route 了一个尚未注册的 Agent 名,运行时通常不会报错,只是把消息挂在队列里等一个永远不会出现的消费者。排查时优先确认:route 的目标 Agent 是否已经 register。第三,Agent 内部对 message.type 做了过滤。很多 Agent 的 handle 里有一段 if message.type != "xxx": return,如果 emit 的 type 和 Agent 里期望的 type 有一字之差,消息会被静默丢弃。这个最坑,因为完全不报错。我后来在所有 Agent 的基类里加了一个 ignored_message 的 debug 日志,专门输出被过滤的消息,省下大量排查时间。

4.5 消息体解析失败与字段遗漏

payload 字段遗漏也是高频问题。上游 Agent 只传了 query 没传 trace_id,下游 Agent 在处理时拿不到链路追踪信息,整条链路日志就对不上了。我的经验是,消息体必须用 pydantic 定义强类型模型,必填字段缺失直接抛异常,而不是返回空值。宁可让流程快速失败,也不要让错误数据悄悄往下传。特别是 trace_id 这个字段,我直接把它设计成 Message 的必填属性,所有构造消息的地方都必须显式传入,不给偷懒的机会。项目跑了一段时间后,这个约束帮我避免了很多次“日志对不上”的尴尬。

5. 实际操作中的一点个人体会

整个项目做下来,我最大的感受是:Agent 编排框架最大的价值,不是让你把代码写得少,而是让你在 Agent 数量增长时仍然能维持“可预测”。早期只有两三个 Agent,用什么方案都无所谓;等 Agent 超过五个、工具超过十个,消息路由和链路追踪就成了刚需。这跟我之前维护微服务架构的感受非常像:分布式系统的难点从来不是写业务逻辑,而是让所有节点之间的交互有秩序、可追踪。

所以如果你现在的 Agent 项目已经开始变乱,我的建议是先别急着加 Agent,先把“消息”这个中间层做好。哪怕不用 hermes-agent,你也可以在自己的项目里引入统一消息格式、trace_id、路由表这几个概念,收益立竿见影。把消息流理清楚,Agent 的开发才能真正快起来。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询