"hermes-agent"这个项目名,最早是我在折腾多智能体协作时随手起的。Hermes 是希腊神话里的信使神,专门在众神之间传递消息、执行任务、引渡灵魂,我给一个负责调度其他 Agent 的编排型智能体起这个名字,倒不是想蹭神话的热度,而是它干的事确实像信使——接收请求、拆解意图、分派给合适的执行单元、再把结果汇总送回去。
这个项目从最初的实验脚本,慢慢变成了一个相对完整的 Agent 框架。核心解决的不是"怎么让单个 Agent 更聪明",而是"多个 Agent 之间怎么高效协作、怎么调用工具、怎么不把上下文撑爆"。这篇文章把她的设计思路、核心模块、最小实现和实测踩坑完整梳理一遍,想自己搭 Agent 框架、或者正在被多 Agent 协作搞得头疼的朋友,应该能直接用上。
1. 为什么叫 Hermes:Agent 的核心价值不是"会聊天",而是"会调度"
1.1 从信使神到编排者,一个被低估的定位
现在聊 Agent,大家普遍关注的还是单点能力:模型推理强不强、工具调用准不准、RAG 检索好不好。但我在实际项目里发现,真正让 Agent 从"玩具"变成"生产力"的,往往是它怎么组织自己的行为链条,以及多个 Agent 一起工作时怎么不打起来。
Hermes 这个名字其实代表了我的设计主张:一个合格的 Agent,应该像信使一样,知道自己什么时候该说话、什么时候该闭嘴、什么时候该把消息转给谁。它不是所有任务的最终执行者,而是任务的协调者。单 Agent 解决不了的复杂任务,拆成多个子任务分给专精的 Agent,最后由编排者汇总——这套逻辑用 Hermes 来命名,再贴切不过。
1.2 hermes-agent 要解决的三类真实痛点
我在做这个项目之前,先梳理了当时几个主力 Agent 方案的通病,hermes-agent 就是冲着这三个痛点去的:
| 痛点 | 表现 | hermes-agent 的解法 |
|---|---|---|
| 上下文爆炸 | 一个 Agent 对话轮次一多,历史记录疯狂膨胀,模型输入成本飙升 | 工作记忆只保留关键摘要,完整历史落到长期记忆存储 |
| 消息协议混乱 | 多个 Agent 各说各话,A 输出的格式 B 解析不了 | 统一消息信封结构,所有通信走同一个消息总线 |
| 工具权限失控 | Agent 什么工具都能调,一次误触发导致连锁副作用 | 工具注册时声明权限等级,编排层做白名单校验 |
这三个问题,如果你只是玩单 Agent 的 Demo,大概率感受不到;一旦上了生产环境、接了真实业务系统,每一个都会变成拦路虎。hermes-agent 的架构,就是从这三个问题倒推出来的。
2. 核心架构拆解:消息总线、任务队列与三层记忆的设计逻辑
2.1 消息总线:所有通信的"标准信封"
多 Agent 系统里最容易被忽视的就是通信协议。很多初学者直接让两个 Agent 用自然语言互相对话,看起来灵活,实际上一轮下来格式就乱了——一个输出 JSON,另一个输出 Markdown,第三个直接丢一段代码。hermes-agent 的做法是定义统一的消息信封,所有 Agent 之间的通信都走这个结构:
from dataclasses import dataclass, field from typing import Any, Optional from enum import Enum import uuid import time class MsgType(Enum): REQUEST = "request" # 请求某个 Agent 执行任务 RESPONSE = "response" # 返回执行结果 CALLBACK = "callback" # 异步回调通知 HEARTBEAT = "heartbeat" # 心跳检查 ERROR = "error" # 错误信息 @dataclass class Message: msg_id: str = field(default_factory=lambda: uuid.uuid4().hex) msg_type: MsgType = MsgType.REQUEST sender: str = "" # 发送方 Agent 名字 receiver: str = "" # 接收方 Agent 名字,广播用 "all" task_id: str = "" # 所属任务 ID,用于追踪 payload: Any = None # 消息体,JSON 序列化 priority: int = 5 # 1-10,数值越大越优先 created_at: float = field(default_factory=time.time) trace_id: str = "" # 链路追踪 ID为什么强调"信封"而不是"内容"?因为只要你规定了信封格式,里面的 payload 是 JSON 还是 YAML 其实可以灵活处理,而路由、追踪、优先级、超时这些横切逻辑,全部可以基于信封字段完成。我在实际使用中体会很深的一点是:trace_id字段一定要从一开始就加上。Agent 一多,你排查一个问题可能是 A → B → C → D 的调用链,没有 trace_id 你根本不知道哪条消息对应哪次任务。
2.2 任务队列:优先级决定 Agent 先处理什么
hermes-agent 里的每个 Agent 都有一个自己的任务队列,但队列不是简单的先进先出。我们用的是优先级加权 + 超时抢占的策略:每条消息带 priority 字段,高优先级的请求可以插队;但为了防止低优先级任务被无限饿死,队列会做老化处理——等待超过一定时间的任务,优先级自动上调。
这里有一个容易踩的坑:优先级不是越高越好。我把所有消息都设成 9,结果高优任务互相抢占,低优任务全部饿死,系统吞吐直接崩了。后来定的规则是:用户直接请求的默认 8,Agent 内部子任务默认 5,日志类通知默认 2,这个弹性区间让整个系统的调度健康了很多。
队列消费端的实现也值得说一下。我早期用的是同步消费,Agent 处理完一条消息才取下一条,后来发现只要有一个子任务卡住,整个 Agent 就瘫了。改成异步消费 + 可中断后,每个 Agent 持有一个asyncio.TaskGroup,主循环负责接收消息,子任务并发执行,配合超时控制,健壮性提升了一个量级。
2.3 三层记忆:工作记忆、长期记忆与共享黑板
Agent 的记忆设计是另一个容易翻车的地方。hermes-agent 把记忆拆成三层,各干各的:
- 工作记忆(Working Memory):当前任务进行中的临时上下文,比如正在拆解的子任务列表、已获取的中间结果。容量限制很严格,超过一定 token 数就强制做摘要压缩。
- 长期记忆(Long-term Memory):任务结束后的重要信息沉淀。每次任务完成,编排层会生成一份结构化摘要存进向量库,供后续相似任务检索。
- 共享黑板(Shared Blackboard):多个 Agent 共同读写的一块公共区域,用来存"全局事实"。比如用户身份信息、项目全局约束、已经确认的决策,避免每个 Agent 各存一份导致信息不一致。
共享黑板是我特别想强调的设计。最开始我让每个 Agent 自己带着全量上下文跑,结果多个 Agent 对同一事实的理解出现偏差——A 认为用户要的是 A 方案,B 认为用户要的是 B 方案,最后汇总的时候才发现打架了。后来引入共享黑板,所有 Agent 在动手前先同步黑板上的约束条件,矛盾率大幅降低。三层记忆之间的关系,我习惯用一个通俗类比:工作记忆是桌面,长期记忆是档案柜,共享黑板是办公室公告栏——桌面只放手头正在干的活,干完归档进柜子,全局信息贴公告栏大家都能看到。
3. 最小闭环:从零实现一个能自己拆任务的 hermes-agent
3.1 环境与选型:为什么用 Python + asyncio
这一节我直接说说复现项目时的技术选型。语言层面我选了 Python,原因很现实:生态里模型 SDK、向量库、工具库最全,写 Agent 的胶水代码效率最高。异步框架用 asyncio 而不是多线程,是因为 Agent 的典型负载是大量 IO 等待——等模型响应、等工具返回、等外部 API 调用,这种场景协程的并发效率远超线程,而且没有 GIL 争抢和线程切换的额外开销。
模型层我做了抽象接口,方便切换不同的模型服务商。实际测试中,编排型 Agent 对模型的要求和对话型不太一样——它更需要强的指令遵循能力而不是创意生成能力。拆任务、调工具、判断是否完成,这些动作对模型的"听话程度"要求极高。
3.2 核心循环:规划 → 执行 → 观察 → 再规划
一个 Agent 的最小闭环是标准的 ReAct 模式,但 hermes-agent 在实现上做了几个关键增强。核心循环代码如下:
import asyncio import json class HermesAgent: def __init__(self, name="hermes", model=None, tools=None): self.name = name self.model = model self.tools = {t.name: t for t in (tools or [])} self.working_memory = [] self.blackboard = {} # 共享黑板引用由编排层注入 async def run(self, task: str, max_steps: int = 15) -> dict: self.working_memory.append({"role": "user", "content": task}) steps = [] for step in range(max_steps): # 1. 规划:让模型基于当前工作记忆决定下一步动作 action = await self._plan() # 2. 执行:动作可以是调工具、发子任务或直接回答 observation = await self._act(action) # 3. 观察结果并入工作记忆 observation = self._compress(observation) # 关键:压缩观察结果 self.working_memory.append({"role": "assistant", "content": json.dumps(action)}) self.working_memory.append({"role": "user", "content": f"观察: {observation}"}) steps.append({"step": step + 1, "action": action, "observation": observation}) # 4. 判断是否结束 if action.get("type") == "finish": return {"task_id": task, "steps": steps, "result": action["result"]} raise TimeoutError(f"超过 {max_steps} 步仍未完成,强制终止") async def _plan(self): # 拼接系统提示词 + 工具描述 + 工作记忆,让模型输出结构化动作 prompt = self._build_plan_prompt(self.working_memory, self.tools) resp = await self.model.chat(prompt) return self._parse_action(resp) async def _act(self, action): if action["type"] == "call_tool": tool = self.tools.get(action["tool"]) if not tool: return f"错误: 工具 {action['tool']} 不存在" return await tool.execute(**action["args"]) if action["type"] == "delegate": # 委派给其他 Agent router = self.blackboard.get("router") return await router.dispatch(action["target"], action["task"]) if action["type"] == "finish": return "任务完成" return "未知动作类型"这段代码是框架的核心骨架,实际使用时有三个细节必须注意:
第一,动作类型必须严格限定。我最初允许模型自由输出动作,结果模型经常编造出不存在的工具名和参数格式,错误率很高。后来在系统提示词里给出严格的 JSON Schema 约束,并加了 schema 校验,解析失败就重试一次,再失败就走兜底策略,工具调用准确率从 76% 提到了 94% 左右。
第二,观察结果必须压缩。_compress这个方法是我反复踩坑后加的。一次工具调用可能返回几千字的原始数据,如果不压缩直接塞回上下文,两三轮之后工作记忆就爆了。我的做法是让模型对观察结果做摘要提取,只保留与当前任务目标相关的关键信息。这一步对控制输入成本效果立竿见影。
第三,最大步数限制是硬性要求。没有上限的循环是最危险的,一个任务可能让 Agent 无限转圈,On 掉你全部的预算配额。设为 15 步后,配合观察压缩,单个任务的平均模型 token 消耗降了 40% 左右,而且因为强制终止的存在,再也没有出现过预算失控的情况。
3.3 最小闭环的验证结果
我用一个"整理项目周报"的任务做了验证。给 hermes-agent 挂了三个工具:读取 Git 提交记录、查询在线文档、汇总成 Markdown 报告。它的执行路径是:先调 Git 工具拿到提交列表,再调文档工具补全信息,最后生成报告并结束。完整走完 5 步,使用的模型 token 比之前直接塞全量上下文的方案少了约 35%,输出质量反而更稳定——因为每一步模型只需要关注当前那一段压缩后的信息,注意力更集中。
这个最小闭环告诉我一个道理:Agent 不是模型能力不够,而是提示词和工具的组织方式不对。同样的模型,用结构化的动作协议和压缩反馈循环包一层,效果天差地别。
4. 多 Agent 协作:消息路由与任务委派的实战细节
4.1 路由规则:谁适合处理这条消息
单 Agent 跑通之后,紧接着就是多 Agent 的编排。hermes-agent 里的多 Agent 通信不是让 Agent 之间互相喊话,而是统一走一个路由层。路由层的核心是一张路由表,逻辑上长这样:
ROUTING_RULES = { "planner": {"skills": ["任务拆解", "优先级规划"], "keywords": ["计划", "安排", "拆解"]}, "coder": {"skills": ["代码生成", "代码审查"], "keywords": ["代码", "函数", "bug", "实现"]}, "researcher": {"skills": ["信息检索", "资料汇总"], "keywords": ["查", "搜索", "资料", "调研"]}, "reviewer": {"skills": ["质量检查", "风险识别"], "keywords": ["检查", "审核", "评估", "风险"]}, }路由匹配采用两级策略:先是关键词精确命中,命中率不高;再把请求向量化后和各个 Agent 的技能描述做语义相似度计算,取最高分且超过阈值的 Agent 作为目标。关键词匹配是为了快,语义匹配是为了准,两者结合,路由准确率在测试集上做到了 89%。
这里有一个实际教训:不要把路由完全交给模型自由发挥。我早期试过让一个"超级调度 Agent"用自然语言决定把任务发给谁,效果极差——它经常因为上下文里出现过某个 Agent 的名字就乱派,而且调度逻辑不透明,出错完全没法追。后来改成规则 + 向量检索的混合路由,可解释性强多了,排查问题也能直接定位到路由表的哪条规则出了问题。
4.2 委派与结果回收:同步等待还是异步回调
多 Agent 协作的第二大问题是任务的委派和结果回收。hermes-agent 支持两种模式:
- 同步委派:编排者把子任务发给执行 Agent,然后挂起等待结果返回。适合子任务之间有依赖关系的场景。
- 异步委派:编排者发完子任务就不管了,执行 Agent 完成后通过回调消息通知。适合可以并行的子任务。
实际项目中两者是混合使用的。比如一个"开发新功能"的任务,planner 先拆解出设计、编码、测试三个子任务,设计和编码有依赖关系,编码和测试也有依赖关系,但多个不同的功能模块之间可以并行。实现上我用了一个简单的依赖图管理器,节点完成时触发下游节点入队,整个流程就像一个有向无环图的拓扑遍历。
结果回收还有一个容易忽略的点:结果的验证责任在谁。我最初的设计是执行方完成就算完事,但后来发现 coder 提交的代码质量参差不齐,reviewer 才发现一堆问题,白白浪费了往返轮次。现在改成:委派任务时带上验收标准,执行方完成后先自检再上报,编排层抽查验证。多了一道关卡,整体返工率明显下降。
4.3 失败重试、超时与幂等性
多 Agent 系统的容错设计,是和生产环境接轨时无法回避的问题。我在 hermes-agent 里做了三件事,每一件都是踩坑换来的:
超时控制:每个委派任务必须声明超时时间。我用的是指数退避重试策略——第一次超时等 2 秒重试,第二次 4 秒,第三次 8 秒,最多五次。重试前要检查任务是否具有幂等性。
幂等保障:这个是最隐蔽的坑。有一次执行 Agent 在处理"发送通知"的子任务时,因为网络抖动重试了三次,结果用户收到三条一模一样的通知。后来所有消息信封强制要求带task_id,每个执行 Agent 在本地维护一个"已处理任务 ID"集合,收到重复任务直接返回上次的结果。记住了:带有副作用的工具调用,必须在调用前做幂等检查,不然重试机制越完善,事故越严重。
失败归因:多 Agent 链路里一个任务失败,根因可能埋在上游好几跳。我在每个子任务的结果里强制附加执行摘要(用了哪些工具、耗时多少、关键中间结果),失败时能顺着 trace_id 回溯完整链路。有一次 researcher 一直超时,排查后才发现是它内部调用的检索接口经常变慢,而不是消息路由的问题——没有链路日志,这种问题根本无从查起。
5. 实测踩坑实录:上下文膨胀、循环调用与工具权限失控的完整排查
5.1 上下文膨胀:工作记忆是怎么一步一步失控的
前面提过观察压缩,这一节我把当时的完整排查过程写出来,因为这个问题非常典型,而且隐蔽。
现象很诡异:某个 Agent 在处理第 12 个任务时,响应速度突然从 3 秒涨到 15 秒,输入 token 从 4K 涨到 28K。一开始我以为是模型服务端波动,后来发现是稳定复现的——只要任务数超过 10 个,性能就断崖式下降。
排查过程:我先打印了每次模型调用的 token 统计,发现工作记忆里的历史消息在逐渐累积。原来我的"压缩"逻辑只压缩了观察结果(工具返回值),没有压缩对话历史本身。Agent 每完成一个子任务,之前的 planning 动作和对应判断都还留在working_memory里,几千字几千字地堆着。第 12 个任务时,历史对话已经占了 20K+ token。
根因找到了,修复方案分三层:
- 对话历史滚动窗口:work_memory 只保留最近 N 轮完整对话,更早的对话压缩成一段摘要,摘要里只写"已完成的步骤和关键决策"。
- 子任务完成即归档:一个子任务完成后,把它的完整记录移出工作记忆,转换为长期记忆中的一条向量记录。
- 全局约束锁定:用户在项目开始时的核心要求,从共享黑板每轮注入,不依赖对话历史保留。
修复之后,长任务场景下输入 token 稳定在 5K 以内,响应时间回落到正常水平。这个坑给我的教训是:记忆管理不是靠一个压缩函数就完事的,它是一个贯穿 Agent 生命周期、需要分层设计的体系。
5.2 Agent 互相调用的死循环:一次线上事故的完整回溯
这个坑让我印象最深,因为它是真实发生在一次演示时——两个 Agent 开始互相发消息,像两个人在群里无限"在吗?在。",一直循环了三分多钟,直到我手动 kill 进程。
回溯链路是这样的:planner 给 coder 派了一个"实现用户登录接口"的子任务,coder 完成回复后,planner 收到结果,但它的完成判断逻辑有个 bug——它要求"所有子任务的结果里必须包含 test 报告",而 coder 返回的结果里没有 test 字段,所以 planner 判定为"未完成",于是重新派发同一个任务。coder 这边已经处理过这个 task_id,但又没有做幂等检查,所以又重新跑了一遍,结果还是没带 test 字段。双方就这样你一句我一句,陷入循环。
修复做了三件事:
- 幂等检查前置:执行 Agent 收到消息时先查"已处理任务集合",重复任务直接返回上次结果,不重新执行。
- 完成判定标准明确化:planner 在委派时通过消息信封的
expects字段声明期望的结果结构,coder 按 Schema 返回,planner 只做 Schema 校验,不再做模糊判断。 - 全局循环检测:消息总线维护一个"近 10 分钟内同一 task_id 的派发次数"计数器,超过 5 次直接告警并挂起该任务。
这个坑的本质是:两个 Agent 各自以为自己在做正确的事,但它们的判断标准不一致。所以多 Agent 系统的协作规则不能靠默契,必须写成明确的协议,每个环节的"完成"标准必须机器可验证。
5.3 工具权限失控:一个误操作引发的连锁副作用
最后一个坑是关于工具权限的。我在给 hermes-agent 接内部系统时,给一个测试 Agent 挂了一个"执行数据库变更"的工具,本来想着测试环境无所谓,结果它在处理一个语义模糊的任务时,自行决定"顺手"执行了一条UPDATE语句,改了测试库里的几百条数据。
这事让我彻底意识到:Agent 的工具调用权限必须分级,而且默认是最小权限。hermes-agent 现在的工具权限设计是四级:
| 权限级别 | 说明 | 典型工具 | 使用条件 |
|---|---|---|---|
| L0 只读 | 无副作用,可直接调用 | 搜索、读取文档、查询接口 | 任意 Agent 默认可用 |
| L1 写临时 | 产生临时文件或可回滚的操作 | 写临时文件、缓存写入 | 任务上下文明确需要 |
| L2 写持久 | 对持久化数据有影响 | 写正式文档、发通知消息 | 需人工二次确认 |
| L3 高风险 | 不可逆或影响面大 | 删数据、执行 DDL、调支付 | 仅特定 Agent 持有,且需双人审批 |
L2 和 L3 的工具在调用前,编排层会生成一个"操作确认请求"发给用户,用户确认后才真正执行。这个机制上线后,再也没出现过 Agent 擅自改数据的状况。
另外一个细节:工具的描述信息要写得非常具体。模型判断要不要调用一个工具,完全依赖工具描述。我之前有个工具叫"get_user_info",描述写的是"获取用户信息",结果 Agent 什么任务都想调它。后来把描述改成"根据用户 ID 获取用户的基础注册信息,包括昵称、头像、注册时间。注意:本工具不返回用户的订单记录。",误调用率立刻降下来了。工具描述不是给开发者看的,是给模型看的,值得多花时间抠字眼。
6. 沉淀下来的几点经验与后续扩展方向
跑完这些坑,再回头看 hermes-agent,我最大的体会是:Agent 系统的问题,80% 不是出在模型上,而是出在工程上。消息协议、记忆管理、权限控制、幂等保障、超时熔断——这些传统分布式系统里早就有成熟方案的东西,一旦套上 Agent 的外壳,反而容易被忽略,因为大家都被"智能"二字吸引了注意力。
如果你也要搭类似的 Agent 框架,我建议按这个顺序推进:先把单 Agent 的任务循环和工具协议做扎实,再引入共享黑板解决多 Agent 的信息一致性问题,最后才是消息路由和权限体系。每一步都验证稳定了再走下一步,不要一上来就追求"大而全"。
后续我打算给 hermes-agent 加两个能力:一个是让 Agent 具备从失败案例中自动总结规则的学习机制,把踩过的坑固化进提示词模板;另一个是更细粒度的成本监控,按任务维度统计每个 Agent 的 token 消耗和耗时,方便优化调度策略。这些方向都是实践中真实遇到的瓶颈,等有阶段性成果了再单独写一篇。
最后分享一个实用小技巧:给 Agent 起名字的时候,把它的职责写进系统提示词的第一行。比如 coder 的系统提示词就这么写:"你是 coder,一个专注代码实现与代码审查的 Agent。你擅长 Python 与 TypeScript,你会严格遵循任务 Schema 返回结果。你不负责任务拆解,那是 planner 的工作。" 明确的身份边界,能少掉一大半毫无意义的跨 Agent 废话消息。