1. 为什么我在一堆Agent框架里，还是决定自己写一个hermes-agent
大概是从去年下半年开始，我手头的项目逐渐从“单模型调用”转向“多Agent协作”。市面上能叫得上名字的Agent框架我基本都试过一圈，有的重度依赖云端服务，有的配置复杂到光是学会怎么声明角色就要花一个下午，还有的为了支持一个很酷的演示Demo，把内部抽象层做得极其黑盒，出了问题根本不知道去哪排查。
当时我面临的实际场景其实并不算复杂：我需要让几个专职Agent协作完成一套自动化任务。它们要能互相传递中间结果，要能按需调用外部工具（比如搜索、代码执行、数据库查询），还要能根据任务状态动态决定下一步让谁接手。这些需求听起来不新鲜，但在现成框架里折腾了一周之后，我发现大量时间都花在了“绕过框架自己的约定”上，而不是花在业务逻辑上。
后来我就想，干脆自己维护一个轻量级的Agent运行时，不求功能大而全，只把我真正用到的几个核心机制做扎实。这个项目就是hermes-agent。
取名Hermes是有意为之。Hermes在神话里是信使神，而Agent系统从本质上讲就是一套消息驱动系统——各个Agent之间、Agent和工具之间的交互，全部可以归约成消息的产生、路由、消费和响应。把名字定为hermes-agent，其实就是定下了这个项目的灵魂：一切围绕高效、可靠的消息传递来设计。
这个项目解决的核心问题可以概括成三句话：复杂的多Agent任务编排不该依赖繁琐的图形界面；Agent之间的通信必须显式、可控、可观测；工具的接入方式要足够简单，简单到只需要一个函数加一段描述就能注册进去。
如果你也在做类似的Agent编排项目，或者你被现有框架的重抽象搞得头疼，那这个项目的设计思路、代码结构和踩坑记录应该能给你一些参考。我不打算写一份面面俱到的使用文档，我更想分享的是“我为什么这么设计”以及“真正跑起来之后遇到了哪些文档里不会写的问题”。
2. 核心架构拆解：消息总线、两级记忆和工具注册表
在动手写第一行代码之前，我先列了一个清单，写下这个Agent运行时绝对要做到的几件事。清单只有四条，但每一条在后面都被证明是刚需。
第一，Agent之间不能直接互相调用函数，它们只能通过消息通信。第二，所有消息都要有明确的类型、来源、目标和上下文ID。第三，工具对Agent不是无限开放的，Agent能用什么工具，必须在运行时由注册表统一管理。第四，每个Agent要有自己的独立记忆，同时系统还要有全局记忆。
这四条最终演变成了hermes-agent的四个核心模块：MessageBus、MemoryStore、ToolRegistry和Scheduler。
2.1 MessageBus：所有通信的中枢神经
MessageBus的设计参考了消息队列的思路，但要比传统MQ更轻量。在我们的场景里，Agent之间传递的不是高吞吐的业务日志，而是带有明确意图和上下文依赖的任务消息，所以我把消息设计成了一种结构化的对象，而非纯文本。
每条消息的基本结构如下：
@dataclass class AgentMessage: msg_id: str msg_type: str # task_request / task_result / query / event sender: str # 来源Agent名称 recipient: str # 目标Agent名称，支持通配符 context_id: str # 用于串联同一个任务链的上下文ID payload: dict # 实际内容 priority: int = 5 # 0-10，数字越大越紧急 timestamp: float = field(default_factory=time.time)这里有一个容易被忽略的设计细节：context_id是必须的。Agent协作不是一次性的请求响应，而是一个任务可能会经过A Agent处理后交给B Agent，B处理完可能又回来找A确认。如果没有一个全局的上下文ID把这一串消息串起来，后续的日志追踪和结果归因会变成灾难。
MessageBus内部维护了一个按优先级排序的队列，同时支持了简单的主题订阅模式。Agent启动时可以声明自己处理哪些msg_type，总线负责把对应的消息投递给合适的Agent。
2.2 MemoryStore：短期上下文和长期事实分开存
在Agent系统里，记忆问题永远绕不开。我踩过的第一个大坑就是把所有对话历史一股脑塞给大模型，结果是token消耗巨大，而且模型经常被早期不相关的信息干扰。
hermes-agent的做法是把记忆分成两层。短期记忆对应当前任务链的上下文，每个context_id维护一份独立的临时消息列表，任务结束就清理。长期记忆则存储Agent从任务中提取出来的、未来可能复用的事实信息，比如用户偏好、项目约定、常见错误记录等。
class MemoryStore: def __init__(self): self.short_term = {} # context_id -> deque[AgentMessage] self.long_term = {} # namespace -> list[Fact] def add_short_term(self, context_id: str, message: AgentMessage): if context_id not in self.short_term: self.short_term[context_id] = deque(maxlen=50) self.short_term[context_id].append(message) def remember_fact(self, namespace: str, fact: dict): self.long_term.setdefault(namespace, []).append(fact) def recall_facts(self, namespace: str, max_results: int = 5): facts = self.long_term.get(namespace, []) # 按相关度排序，这里简化为按时间倒序返回 return sorted(facts, key=lambda x: x.get("ts", 0), reverse=True)[:max_results]长期记忆不追求存得多，而追求查得准。每次存入之前，Agent会先判断这个信息是否值得长期保留，这个判断逻辑可以是一个简单规则，也可以是一个单独的判断Agent。
2.3 ToolRegistry：让工具接入像注册回调一样简单
工具接入是所有Agent框架最看重体验的地方。很多框架的工具接入要写一堆样板代码，定义输入输出类、写校验逻辑、声明权限……我自己的感受是，过度的形式化约束会让开发者根本不想加新工具。
hermes-agent里注册一个工具只需要两步：写一个普通函数，加一个描述用的装饰器。
@tool( name="calculator", description="执行四则运算表达式，例如 '(12 + 34) * 56'", parameters={ "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式"} }, "required": ["expression"] } ) def calculator(expression: str) -> str: # 这里只做简单的四则运算，用 eval 是演示用途，生产环境请用更安全的解析器 result = eval(expression) return str(result)ToolRegistry会在启动时扫描所有带@tool装饰器的函数，生成一份工具清单，这份清单会被注入到Agent的系统提示词中，让模型知道有哪些工具可用、每个工具的入参格式是什么。
我后来还加了一个很关键的能力：工具调用前的参数校验。大模型生成的JSON参数经常不靠谱，比如该传数字的地方给你传带单位的字符串，该传枚举值的地方传了一个近义词。在注册工具时加一套JSON Schema校验，调用前先校验，校验不过就返回一条格式化的错误信息给Agent，让它自己纠正。
2.4 Scheduler：谁来做，什么时候做
Scheduler是整个架构里最不像“框架”的部分，因为它更多是一套策略。它的职责只有一个：收到一条task_request类型的消息时，决定把这个任务派给哪个Agent。
最简单的策略是按能力匹配，也就是根据ToolRegistry里登记的Agent能力元数据来判断。复杂一点的策略还能结合当前各Agent的任务负载、历史成功率等因素。
我默认实现的策略是一个加权轮询加能力过滤的组合，逻辑大致如下：
def schedule(self, task: AgentMessage, agents: dict) -> str: candidates = [] for name, agent in agents.items(): if task.msg_type in agent.capabilities: # 权重 = 能力匹配度 * (1 / 当前队列长度) score = agent.match_score(task) * (1.0 / max(1, agent.queue_size())) candidates.append((score, name)) if not candidates: return None # 按得分降序，取第一个 candidates.sort(reverse=True) return candidates[0][1]调度策略这个模块被刻意设计成了可插拔的接口。实际项目中，有的场景需要尽可能快的响应，有的场景需要按严格的角色顺序执行，这两种情况对应完全不同的调度策略。我不会把某种策略写死，也不建议你用别人框架里写死的策略。
3. 手把手搭一个hermes-agent：从创建项目到跑通第一个多Agent任务
这一节我们不看太多概念，直接动手。我会带你从零初始化一个hermes-agent项目，实现一个简单的“需求拆解Agent + 编码Agent + 审查Agent”协作流水线。这套流水线我会用在实际项目中做小范围自动化代码生成，虽然不完美，但足够说明整个流程是怎么转起来的。
3.1 项目目录结构
先说明一下项目的基本组织方式。我习惯用src目录放核心代码，examples目录放示例，再加上一个configs目录放Agent配置文件。
hermes-agent/ ├── src/ │ ├── __init__.py │ ├── core/ │ │ ├── message.py │ │ ├── bus.py │ │ ├── memory.py │ │ └── agent.py │ ├── tools/ │ │ ├── __init__.py │ │ ├── calculator.py │ │ └── shell.py │ └── runtime.py ├── configs/ │ └── agents.yaml └── examples/ └── pipeline_demo.py这个结构不复杂，但每个文件的职责很明确，新手拿到手不会迷路。
3.2 Agent基类的设计
Agent是hermes-agent里唯一一个需要开发者继承的类。它的职责包括初始化模型客户端、接收消息、调用工具、维护记忆、返回结果。我不想把它搞成一个大而全的“智能体”，它本质上就是一个“带工具访问能力的消息处理器”。
class BaseAgent: def __init__(self, name: str, system_prompt: str, capabilities: list[str], tools: list = None): self.name = name self.system_prompt = system_prompt self.capabilities = capabilities self.tool_manager = ToolRegistry() if tools: for tool in tools: self.tool_manager.register(tool) def handle(self, message: AgentMessage) -> AgentMessage: raise NotImplementedError实际跑任务的时候，Agent内部会执行这样一条流程链：
- 把收到的消息payload、短期记忆、长期记忆相关事实组装成Prompt。
- 调用大模型接口，拿到回复。
- 解析回复，判断是普通回复、还是需要调用工具。
- 如果需要调用工具，执行工具、把结果拼进去再次请求模型。
- 生成最终回复，构造新的AgentMessage返回，同时把整个交互过程写入记忆。
最耗时的步骤往往在3和4之间反复循环。一次任务里，Agent可能需要连续调用三到四次工具才能得到满意的结果，这在长任务场景中尤为常见。
3.3 配置文件：用YAML声明Agent角色
hermes-agent有个原则：能用配置解决的事情就不要写代码。每个Agent的角色定义、能力声明、模型参数都写在一个YAML文件里，运行时会自动加载并实例化。
agents: - name: planner capabilities: [task_request] system_prompt: > 你是一个需求分析专家。你会收到用户提出的原始需求， 请将需求拆解为具体的开发步骤，并以JSON数组格式输出。 model: gpt-4o-mini temperature: 0.3 - name: coder capabilities: [task_request] system_prompt: > 你是一个资深Python开发工程师。你负责根据开发步骤编写代码。 只能调用可用的工具来完成代码编写和语法检查。 model: gpt-4o temperature: 0.1 - name: reviewer capabilities: [task_request] system_prompt: > 你是一个代码审查专家。你负责审查代码的正确性、可读性和安全性。 对于发现的问题，直接输出修改建议。 model: gpt-4o-mini temperature: 0.5这里有一个参数值得单独拿出来说：temperature。我分别在planner和coder上设置不同的温度值是有讲究的。planner需要一定的发散能力来拆解需求，温度给到0.3；coder的任务是精确编码，温度给到0.1，让它尽量别“创造”语法。
实际测试下来的效果是，coder的温度设置成0.1之后，常见的语法错误和变量命名不统一的情况减少了一大半。这个经验在Agent编排场景里非常实用。
3.4 编排流水线：让三个Agent接力干活
配置好Agent之后，剩下的就是写一个编排脚本，把整个流水线串起来。runtime模块负责启动所有Agent，并把用户请求注入系统。
import yaml from src.runtime import AgentRuntime def main(): config = yaml.safe_load(open("configs/agents.yaml", encoding="utf-8")) runtime = AgentRuntime() runtime.load_agents_from_config(config["agents"]) initial_task = AgentMessage( msg_id="task-001", msg_type="task_request", sender="user", recipient="planner", context_id="ctx-001", payload={"request": "写一个Python函数，接收URL列表，批量下载其中的HTML并提取标题。"} ) result = runtime.submit(initial_task) print(result) if __name__ == "__main__": main()整个调用链是这样走的：用户消息发给planner，planner拆解出详细步骤后把结果组装成一条新的task_request发给coder，coder编写代码并调用tool运行自测，最后把代码发给reviewer做审查。reviewer如果有修改意见会把消息打回给coder，意见通过则整个任务结束。
这三段式的结构你可以在无数真实项目中看到影子，因为它确实符合人类团队协作的基本方式：先想清楚做什么，再动手做，最后检查质量。
4. 实测中的三个坑：超时风暴、上下文漂移和工具参数幻觉
任何框架只有跑过真实任务才知道哪里会疼。hermes-agent在开发过程中遇到过很多问题，其中三个最值得分享，因为它们不是“代码写错了”能解释的，而是分布式协作场景下天然存在的隐患。
4.1 坑一：Agent间消息无超时，导致级联阻塞
第一版MessageBus投递消息时是没有超时概念的。这导致一个很隐蔽的问题：当某个Agent因为模型API超时或工具执行卡住时，它会一直占用任务线，而后续依赖它的所有Agent全部排队等待。表现就是整个系统“卡死”在一个任务上，其他高优先级任务全部无法调度。
我花了一个下午排查，最后在日志里看到一条任务在“等待coder响应”这个状态上停留了超过15分钟。问题本质不是网络挂了，而是API返回了一个异常的慢响应，没有触发错误处理逻辑。
修复方案很直接：给每一条Agent消息设置TTL（Time To Life）和超时回调。
class AgentMessage: def __init__(self, ...): self.ttl = 120 # 秒，超过120秒未处理则消息过期 self.on_timeout = self.default_timeout_handler def default_timeout_handler(self): owner.logger.warning( f"Message {self.msg_id} from {self.sender} to {self.recipient} timed out." ) # 构造一条显式的timeout错误消息返回给发送方 error_resp = AgentMessage( msg_type="task_result", sender=self.recipient, recipient=self.sender, payload={"error": "agent timeout"}, status="failed" ) self.owner.reply(self, error_resp)这条规则看起来简单，但它带来一个根本性的改变：Agent之间的信任关系从“默认对方会正常返回”变成了“默认对方可能失败，我需要超时预案”。后来的所有Agent逻辑都建立在这个假设之上，系统的鲁棒性提升非常明显。
4.2 坑二：多轮交互中的上下文漂移
多Agent协作时，消息经过两三轮传递后，经常会出现“偏题”的现象。举个例子，planner最初拆解出一个任务：“实现一个计算两个日期之间天数的函数”。到coder手里的消息却变成了：“实现一个函数”，而原始需求被截断了。
这个问题的根源在于Agent之间传输消息时，默认只传递了上一跳的结果，没有传递完整上下文。中间的Agent认为自己“已经消费了”原始需求，只把summary透传给了下游。
后来我在消息结构里增加了一个context_chain字段，专门保存从初始消息到当前消息的完整链路。下游Agent拿到的不是单一的payload，而是一条带时间戳的上下文链。模型在生成回复时能参考原始需求，而不是只能参考上一跳的转述。
class AgentMessage: def __init__(self, ...): self.context_chain = [] # 保存完整的上下游消息摘要这个改动虽然增加了token消耗，但换来的是任务执行方向和初始目标的高度一致。对于多跳协作场景，这个trade-off我认为非常值得。
4.3 坑三：LLM生成工具参数时出现的类型幻觉
这是我认为最值得详细讲的一个坑。大模型在决定调用工具时，生成的JSON参数经常不遵守预定义的类型约束。比如工具声明了一个count: integer参数，模型可能给你传count: "3 times"；工具声明了format: "json"|"text"参数，模型可能给你传format: "JSON格式"。
早期的处理方式是工具直接报错返回，但这样Agent会在同样的错误上反复横跳，浪费大量token和时间。
我的解决思路是在ToolRegistry加了一层的参数规范化层：
import json import re def normalize_call_arguments(tool_schema, raw_arguments: str) -> dict: arguments = json.loads(raw_arguments) normalized = {} for prop_name, prop_def in tool_schema["properties"].items(): value = arguments.get(prop_name) if value is None: continue if prop_def.get("type") == "integer": # 把字符串里的数字提取出来 if isinstance(value, str): match = re.search(r"\d+", value) normalized[prop_name] = int(match.group()) if match else 0 else: normalized[prop_name] = int(value) elif "enum" in prop_def: # 枚举值做模糊匹配 candidates = prop_def["enum"] best = None for candidate in candidates: if candidate.lower() in str(value).lower(): best = candidate break normalized[prop_name] = best else: normalized[prop_name] = value return normalized这个函数做的事情说白了就是让工具对模型的“不守规矩”保持宽容。它不改变工具的接口规范，而是在进入工具之前做一次适配。加了这层之后，因为参数格式问题导致的任务失败率下降了大概七成。
5. 进一步调优：让hermes-agent在多任务并发下跑得更稳
如果只是跑通一个Demo，前面部分已经足够了。但如果你想把这个框架用到稍微有点并发的真实场景，还有几个调优方向值得投入时间。
5.1 优先级队列与Worker池配置
默认的任务队列是先进先出的，但在真实场景里，不是所有任务都同等重要。我把队列改成了按priority字段排序的优先队列，同时为每个Agent配置了一个小型的worker池，默认是3个并发。这样高优先级任务可以直接插队，低优先级任务即使数量再多也不会淹没系统。
import heapq class PriorityTaskQueue: def __init__(self): self._heap = [] def push(self, message: AgentMessage): heapq.heappush(self._heap, (-message.priority, message.timestamp, message)) def pop(self): if self._heap: return heapq.heappop(self._heap)[2] return None5.2 记忆压缩策略
短期记忆最多保留50条消息，这个数值是经验值。实际跑任务时，一条复杂任务链的消息数经常能超过200条，放着不管的话既浪费token又污染模型的注意力。
我实现了一个简单的记忆压缩器，当短期记忆超过阈值时，会调用一次压缩Agent把旧消息汇总成要点，保留最近几条原文。
def compress_memory(self, context_id: str): messages = self.memory.short_term.get(context_id, []) if len(messages) <= 50: return old_messages = list(messages)[:-20] recent_messages = list(messages)[-20:] summary = self.summarizer.summarize(old_messages) self.memory.short_term[context_id] = [ AgentMessage(msg_type="summary", payload={"summary": summary}) ] + recent_messages这个策略把模型每次请求的上下文窗口控制在合理范围内，任务完成质量没有下降，反而因为模型更关注最近的有效信息而变得更高了。
5.3 工具调用的全链路日志
Agent系统最怕的就是“黑盒运行”。在调试阶段，我给每一条消息、每一个工具调用都加了结构化日志，输出格式是JSON Lines，方便后续用jq或其它工具分析。
{"ts": 1735000000, "event": "tool_call", "agent": "coder", "tool": "shell", "args": "ls -la", "duration_ms": 12} {"ts": 1735000001, "event": "tool_result", "agent": "coder", "tool": "shell", "output_preview": "...", "status": "success"}有了这套日志之后，排查问题的效率提升了不止一个量级。我现在遇到任何异常，第一反应就是翻日志而不是看代码猜。
6. 我的一些后续打算与经验总结
hermes-agent目前在我自己的几个自动化场景里跑得比较稳定了，包括批量文档处理、测试数据生成、小型代码生成流水线。这个项目的核心价值不在于它有多强的Agent能力，而在于它提供了一个“可控的消息骨架”，在此之上你可以灵活地搭建自己的Agent协作逻辑。
如果你打算在自己的项目里参考hermes-agent的思路，我建议你从最薄的一个垂直场景做起，先让自己跑通一条“消息闭环”，再逐渐加入更多的Agent、更多的工具。不要一开始就追求十几个Agent的大舞台，那只会让你被协调成本拖垮。先把两个Agent之间的消息传递打磨到极致，再考虑扩展。
关于Agent框架的选型，我的个人体会是：不要迷信开源项目有多大名气和多少star，而是要看它的代码是否真正符合你的场景。一个2000行的自有框架，只要你真正理解它的每一行，它能发挥出的战斗力远超一个看不懂的五万行框架。hermes-agent对我来说就是这样一块可以自由裁切的“积木”，希望你也能找到属于自己的那块。
