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_factorytime.time)这里有一个容易被忽略的设计细节context_id是必须的。Agent协作不是一次性的请求响应而是一个任务可能会经过A Agent处理后交给B AgentB处理完可能又回来找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(maxlen50) 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, keylambda x: x.get(ts, 0), reverseTrue)[:max_results]长期记忆不追求存得多而追求查得准。每次存入之前Agent会先判断这个信息是否值得长期保留这个判断逻辑可以是一个简单规则也可以是一个单独的判断Agent。2.3 ToolRegistry让工具接入像注册回调一样简单工具接入是所有Agent框架最看重体验的地方。很多框架的工具接入要写一堆样板代码定义输入输出类、写校验逻辑、声明权限……我自己的感受是过度的形式化约束会让开发者根本不想加新工具。hermes-agent里注册一个工具只需要两步写一个普通函数加一个描述用的装饰器。tool( namecalculator, 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(reverseTrue) 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.3coder的任务是精确编码温度给到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, encodingutf-8)) runtime AgentRuntime() runtime.load_agents_from_config(config[agents]) initial_task AgentMessage( msg_idtask-001, msg_typetask_request, senderuser, recipientplanner, context_idctx-001, payload{request: 写一个Python函数接收URL列表批量下载其中的HTML并提取标题。} ) result runtime.submit(initial_task) print(result) if __name__ __main__: main()整个调用链是这样走的用户消息发给plannerplanner拆解出详细步骤后把结果组装成一条新的task_request发给codercoder编写代码并调用tool运行自测最后把代码发给reviewer做审查。reviewer如果有修改意见会把消息打回给coder意见通过则整个任务结束。这三段式的结构你可以在无数真实项目中看到影子因为它确实符合人类团队协作的基本方式先想清楚做什么再动手做最后检查质量。4. 实测中的三个坑超时风暴、上下文漂移和工具参数幻觉任何框架只有跑过真实任务才知道哪里会疼。hermes-agent在开发过程中遇到过很多问题其中三个最值得分享因为它们不是“代码写错了”能解释的而是分布式协作场景下天然存在的隐患。4.1 坑一Agent间消息无超时导致级联阻塞第一版MessageBus投递消息时是没有超时概念的。这导致一个很隐蔽的问题当某个Agent因为模型API超时或工具执行卡住时它会一直占用任务线而后续依赖它的所有Agent全部排队等待。表现就是整个系统“卡死”在一个任务上其他高优先级任务全部无法调度。我花了一个下午排查最后在日志里看到一条任务在“等待coder响应”这个状态上停留了超过15分钟。问题本质不是网络挂了而是API返回了一个异常的慢响应没有触发错误处理逻辑。修复方案很直接给每一条Agent消息设置TTLTime 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( fMessage {self.msg_id} from {self.sender} to {self.recipient} timed out. ) # 构造一条显式的timeout错误消息返回给发送方 error_resp AgentMessage( msg_typetask_result, senderself.recipient, recipientself.sender, payload{error: agent timeout}, statusfailed ) 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_typesummary, 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对我来说就是这样一块可以自由裁切的“积木”希望你也能找到属于自己的那块。 
长河君 · 2026/9/9 12:27:46 · 阅读 1.2k 自研轻量级Agent运行时：Hermes-Agent的架构设计与实战踩坑
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_factorytime.time)这里有一个容易被忽略的设计细节context_id是必须的。Agent协作不是一次性的请求响应而是一个任务可能会经过A Agent处理后交给B AgentB处理完可能又回来找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(maxlen50) 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, keylambda x: x.get(ts, 0), reverseTrue)[:max_results]长期记忆不追求存得多而追求查得准。每次存入之前Agent会先判断这个信息是否值得长期保留这个判断逻辑可以是一个简单规则也可以是一个单独的判断Agent。2.3 ToolRegistry让工具接入像注册回调一样简单工具接入是所有Agent框架最看重体验的地方。很多框架的工具接入要写一堆样板代码定义输入输出类、写校验逻辑、声明权限……我自己的感受是过度的形式化约束会让开发者根本不想加新工具。hermes-agent里注册一个工具只需要两步写一个普通函数加一个描述用的装饰器。tool( namecalculator, 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(reverseTrue) 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.3coder的任务是精确编码温度给到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, encodingutf-8)) runtime AgentRuntime() runtime.load_agents_from_config(config[agents]) initial_task AgentMessage( msg_idtask-001, msg_typetask_request, senderuser, recipientplanner, context_idctx-001, payload{request: 写一个Python函数接收URL列表批量下载其中的HTML并提取标题。} ) result runtime.submit(initial_task) print(result) if __name__ __main__: main()整个调用链是这样走的用户消息发给plannerplanner拆解出详细步骤后把结果组装成一条新的task_request发给codercoder编写代码并调用tool运行自测最后把代码发给reviewer做审查。reviewer如果有修改意见会把消息打回给coder意见通过则整个任务结束。这三段式的结构你可以在无数真实项目中看到影子因为它确实符合人类团队协作的基本方式先想清楚做什么再动手做最后检查质量。4. 实测中的三个坑超时风暴、上下文漂移和工具参数幻觉任何框架只有跑过真实任务才知道哪里会疼。hermes-agent在开发过程中遇到过很多问题其中三个最值得分享因为它们不是“代码写错了”能解释的而是分布式协作场景下天然存在的隐患。4.1 坑一Agent间消息无超时导致级联阻塞第一版MessageBus投递消息时是没有超时概念的。这导致一个很隐蔽的问题当某个Agent因为模型API超时或工具执行卡住时它会一直占用任务线而后续依赖它的所有Agent全部排队等待。表现就是整个系统“卡死”在一个任务上其他高优先级任务全部无法调度。我花了一个下午排查最后在日志里看到一条任务在“等待coder响应”这个状态上停留了超过15分钟。问题本质不是网络挂了而是API返回了一个异常的慢响应没有触发错误处理逻辑。修复方案很直接给每一条Agent消息设置TTLTime 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( fMessage {self.msg_id} from {self.sender} to {self.recipient} timed out. ) # 构造一条显式的timeout错误消息返回给发送方 error_resp AgentMessage( msg_typetask_result, senderself.recipient, recipientself.sender, payload{error: agent timeout}, statusfailed ) 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_typesummary, 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对我来说就是这样一块可以自由裁切的“积木”希望你也能找到属于自己的那块。