1. 为什么我会做 hermes-agent从“提示词蘸一切”到消息驱动hermes-agent 这个词懂点技术的人都看得出来这是一个叫 Hermes 的 Agent 项目。Hermes 是希腊神话里替众神跑腿的信使脚底生风负责把消息从一个神传到另一个神那里。我给自己写的 Agent 框架起这个名字就是看中它“传递任务、协调调度”的定位。跟传统的单体 Prompt 应用不一样的是hermes-agent 的核心是一个消息协议——所有任务进来之后都被标准化成一条消息经过解析、路由、执行再把结果回传。整套流程下来你面对的不再是一个“只会聊天的模型接口”而是一个能同时对接多个工具、多个子任务、甚至多个 Agent 协作的调度枢纽。我最早做这个项目是因为被 AI Agent 的工程化问题反复折磨。单轮调用大模型做点小功能其实很简单难的是让它在一个真实业务里稳定干活。比如你让它“帮我查一下明天到上海的高铁然后定一个提醒再把结果整理成邮件发出来”这种事如果只靠一段 Prompt 塞给模型它大概率会胡编乱造工具参数或者在某个步骤上卡住就整体崩掉。hermes-agent 就是奔着解决这个问题去的把任务拆成“规划—执行—反馈”循环每一步都走消息队列每一步都有超时和重试工具调用也全部标准化注册。谁适合用它我觉得三类人最合适第一自己在做个人助理或者自动化工作流的开发者第二团队里需要给多个业务方提供统一 Agent 能力的后端工程师第三想搞懂 Agent 内部消息流转机制的学习者。这套设计的远期收益其实比功能本身更重要。hermes-agent 的各类子模块彼此解耦意味着你换一个大模型 API、加一个新的工具、甚至把单 Agent 扩展成多 Agent 协作都不用推倒重来。很多项目做着做着就变成“屎山”问题就出在最开始没有把消息链路定清楚。我在这篇文章里会把你从设计思路到代码落地完整带一遍重点讲我踩过的坑和那些文档里不会写的取舍。1.1 Hermes 这个名字背后的设计隐喻很多人起名字只是为了好听但 Hermes 这个名字恰好映射了系统的核心隐喻信使。在 hermes-agent 里所有组件之间不直接互相调用而是通过消息互相通信。Agent 核心是一个信使它接收外界发来的任务消息把消息转译成模型能理解的指令模型给出决策后信使再把决策转译成工具调用拿到工具结果后再一次把结果转译回模型可读的观察。一个完整的 Agent 回合本质上就是消息的编码、路由、解码循环。这种设计的直接好处是你可以随时在消息链路上插入新环节。想记录日志在消息进队列的时候加一个拦截器就行。想给任务限流消息队列天然支持。想让两个 Agent 协作让一个 Agent 把产出消息发给另一个 Agent 的队列就好。我在最初版直接把工具函数的返回值拼接进 Prompt 让模型继续思考结果发现上下文越来越乱出错的概率呈指数上升。后来我把每个步骤的输入输出都规范成消息结构模型只看当前需要的部分整个系统的稳定性立刻上了一个台阶。所以“Hermes”这个名字不是装饰它就是整个架构的工作方式。1.2 整体架构一条消息在主链路上怎么流转hermes-agent 的整体架构一句话概括就是一个核心循环两套基础设施三类标准化接口。核心循环是 Agent 主循环负责驱动模型反复走“思考—行动—观察”的过程两套基础设施分别是工具注册中心和记忆模块三类标准化接口是任务消息、工具描述、模型反馈。先看任务消息。我定义了一个统一的消息结构包含 task_id、task_type、payload、created_at 这些字段不管任务来自 HTTP 接口、定时触发还是另一个 Agent都必须先包装成这种结构再进入队列。task_type 尤其关键它是路由的关键字段比如 weather、reminder、summarize消费者会根据 task_type 决定把任务交给哪个处理器。payload 里存放业务参数比如“查询上海明天天气”就把城市和日期放进去。这种标准化契约带来的收益是你后续加新任务类型完全不用改主框架只需要新增一个 task_type 的处理器即可。主链路的流程大概是这样的外部请求进来统一打包成 TaskMessage投递到 Redis Stream 任务队列Agent Worker 从队列里消费消息调用规划器让大模型生成下一步动作如果动作是调用工具就把参数交给 ToolRegistry 执行工具结果重新包装成观察消息再次交给模型直到模型输出 finish 动作Agent 把最终结果回写队列或者直接返回调用方。链路虽然长每一跳都有迹可循排查问题的时候打开日志就能看清楚消息在哪一环出了岔子。2. 核心模块设计三个决定成败的关键选择我在写 hermes-agent 之前也参考过一些开源 Agent 框架比如 LangChain 的 AgentExecutorAutoGPT 的整套任务栈。结果发现一个问题这些框架封装层级太多中间夹了不少黑盒逻辑出了问题很难查。所以我在自己的项目里坚持“简单的轮子自己造复杂的对接才引入依赖”。核心模块就三个Agent 主循环、工具注册中心、多 Agent 通信协议。这三个模块从第一天起就是定制设计的后面所有新功能都是围绕它们长出来的。2.1 Agent 主循环用 ReAct 模式还是纯粹的 Function Calling关于 Agent 主循环用什么模式我在 ReAct 和 Function Calling 之间纠结了很久。ReAct 模式比较经典思路是让模型交替输出 Thought、Action、Observation自由度最高但缺点是需要写大量解析逻辑而且模型的输出稍微不规范就容易解析失败。Function Calling 是 OpenAI 等厂商提供的原生能力模型直接输出结构化的函数调用参数解析稳很多但前提是你用的模型必须支持这个能力。hermes-agent 最终选择了“以 Function Calling 为主保留 ReAct 兜底”的双轨设计。主路径上我们给模型传入工具描述列表模型返回的标准 JSON 里包含 action 和 args 两个字段Agent 直接拿这个字段去工具注册中心执行调用。碰到某些本地模型不支持 Function Calling 的情况就退回到 ReAct 的纯文本解析从输出里截取 Action: xxx 和 Action Input: xxx 字段。双轨设计确实多写了一点代码但换来的是对不同模型厂商的兼容性这个投入很值得。主循环的核心实现可以简化成下面这段伪代码async def agent_loop(task_message): context init_context(task_message) for step in range(MAX_STEPS): response await llm.chat(context.messages, toolsschema) if response.action finish: return response.output tool_result await tool_registry.call(response.action, response.args) context.messages.append(tool_result) return MAX_STEPS_REACHED这里有两个参数需要特别解释。MAX_STEPS 我一般设置成 8太小了复杂任务跑不完太大了模型会陷入无效循环白白消耗 token。另外每次循环结束后我会把历史 messages 做一次裁剪只保留最近三轮的交互记录同时把早期关键信息抽出来放进一个 summary 字段。实战下来这个策略能显著缓解长任务下的上下文漂移问题。2.2 工具注册中心把“会用的技能”标准化工具注册中心是 hermes-agent 里最不起眼但最重要的模块。它解决的核心问题是如何让大模型“知道”你可以调用哪些工具以及如何让工具的结果足够规整方便模型理解。每一类工具在注册的时候要提供 name、description、parameters 三个字段description 要写清楚工具能做什么、适合什么场景parameters 是 JSON Schema 格式的参数定义。很多初学者懒得多写描述结果模型老是选错工具或者填错参数这真的不能怪模型是你描述不够。我用装饰器来做工具注册写起来非常直观tool_registry ToolRegistry() tool_registry.register( nameget_weather, description查询指定城市未来几天的天气情况适合需要出行建议或穿衣建议的场景, parameters{ type: object, properties: { city: {type: string, description: 城市中文名比如 上海}, days: {type: integer, description: 查询天数默认3} }, required: [city] } ) async def get_weather(city: str, days: int 3): # 对接天气服务商API这里做数据清洗 raw await weather_api.query(city, days) return { city: city, forecast: [parse_day(d) for d in raw[days]] }工具执行有个细节容易踩坑函数的返回值不能是任意结构最好固定成 JSON 可序列化的 dict关键的绝对数值和状态要放在顶层字段里。比如天气工具返回时city、condition、temperature 直接暴露给模型避免模型从一大段嵌套 JSON 里去猜。模型猜测成本越低整体稳定性越高。这是我迭代了很多版本后总结出来的经验。2.3 多 Agent 协作的消息协议task_type 是约定也是边界单 Agent 能处理的任务有限一旦任务量上来或者任务本身需要不同领域的知识就得考虑多 Agent 协作。hermes-agent 的多 Agent 设计没有搞得很复杂核心原则是“每个 Agent 都是一个独立的消息消费者”它们之间唯一的通信方式是消息队列。这样设计的好处是你可以把一个 Agent 部署成多个实例做水平扩展也可以用完全不同的技术栈来写另一个 Agent只要它遵守消息协议就能接入。每个 Agent 创建时会声明自己感兴趣的 task_type 列表。举个例子Planner Agent 收到用户复杂请求后会把任务拆成多个子任务通过 bus.publish(task_typeexecute_code, payload{...}) 发给代码执行 Agent代码执行 Agent 出结果后再发给 Report Agent 做汇总。任务类型定了协作边界就清晰了不会出现两个 Agent 抢同一个任务的混乱情况。我用 Redis Stream 的 consumer group 来做消息分配之后还顺带获得了重试和故障转移能力消费失败的消息可以重新进入 pending 队列由另一个 Worker 继续处理这一点在单机内存队列里很难实现。3. 从零手写一个可运行的 hermes-agent前面讲了设计现在说落地。我接下来用一个最小可运行版本演示你照着敲一遍就能跑起来。核心依赖只有三个FastAPI 提供 HTTP 入口、Redis 作为任务队列、OpenAI 兼容接口的大模型。我自己平时用 Ollama 跑本地模型测试因为调 token 方便开发迭代速度比直接调云端 API 快很多。3.1 项目结构与依赖模块拆得清楚一点后面加功能会省很多力气。我的参考结构是这样的hermes-agent/ ├── agent/ │ ├── core.py # Agent 主循环 │ ├── registry.py # 工具注册中心 │ ├── memory.py # 上下文与长期记忆 │ └── message.py # 消息协议定义 ├── bus/ │ └── redis_stream.py # Redis Stream 封装 ├── tools/ │ ├── weather.py # 天气工具 │ └── reminder.py # 提醒工具 ├── main.py # HTTP 入口 Worker 启动 └── requirements.txtrequirements.txt 里主要就这些fastapi、uvicorn、redis、openai、pydantic。其中 openai 库即使对接 Ollama 也能用只要把 base_url 改成本地地址。这算是 OpenAI SDK 的一个隐形福利生态里很多东西都能复用。3.2 Agent 主循环代码实现主循环是 Agent 的“大脑”它做三件事组织上下文、调用模型、执行工具。代码结构如下# agent/core.py import asyncio from dataclasses import dataclass, field dataclass class AgentContext: task_message: dict messages: list field(default_factorylist) step: int 0 class HermesAgent: def __init__(self, name, llm_client, tool_registry, max_steps8): self.name name self.llm_client llm_client self.tool_registry tool_registry self.max_steps max_steps async def run(self, task_message: dict) - dict: ctx AgentContext(task_messagetask_message) ctx.messages.append({ role: system, content: 你是一个任务规划助手请根据可用的工具逐步完成任务。 }) ctx.messages.append({ role: user, content: task_message[payload].get(prompt, ) }) while ctx.step self.max_steps: ctx.step 1 response await self.llm_client.chat( messagesctx.messages, toolsself.tool_registry.schema() ) if response.action finish: return { task_id: task_message[task_id], status: done, output: response.output, } if response.action not in self.tool_registry: return { task_id: task_message[task_id], status: failed, error: funknown action: {response.action} } result await self.tool_registry.call(response.action, response.args) ctx.messages.append({ role: assistant, content: , tool_calls: [{ id: fcall_{ctx.step}, name: response.action, arguments: response.args }] }) ctx.messages.append({ role: tool, tool_call_id: fcall_{ctx.step}, content: json.dumps(result, ensure_asciiFalse) }) return { task_id: task_message[task_id], status: timeout, error: fexceeded max_steps: {self.max_steps} }你可能已经发现了我把模型的响应结构假设成包含 action、args、output 三种字段。如果你用的是 OpenAI 官方 SDK它返回的 tool_calls 是标准结构需要做一个转换如果你走 Ollama返回结构会稍有差异。所以这里的 llm_client 其实是一个薄封装我自己写了一个 OpenAIAdapter 和 OllamaAdapter统一输出成上面这个简化结构。这是整个框架里最值得复用的设计之一后面换模型服务商只需要新增一个 Adapter。3.3 用 Redis Stream 支撑任务队列任务队列我选了 Redis Stream而不是 Celery 或者 RabbitMQ理由很简单Redis 大多数项目里已经有了不需要额外养一个消息中间件而且 Stream 支持消费组、消息确认、pending 列表足够撑住 Agent 场景的任务调度。关键代码如下# bus/redis_stream.py import json import redis class RedisStreamBus: def __init__(self, redis_url, stream_namehermes:task:stream): self.redis redis.from_url(redis_url) self.stream stream_name self.group hermes:workers def ensure_group(self): try: self.redis.xgroup_create(self.stream, self.group, id0, mkstreamTrue) except redis.ResponseError: # 消费组已存在时会报错忽略 pass def publish(self, message: dict): self.redis.xadd(self.stream, {data: json.dumps(message)}) def consume(self, count1, block5000): entries self.redis.xreadgroup( self.group, self.consumer_name, {self.stream: }, countcount, blockblock, ) result [] if entries: for _, messages in entries: for msg_id, fields in messages: result.append((msg_id, json.loads(fields[data]))) return result def ack(self, msg_id): self.redis.xack(self.stream, self.group, msg_id)消费端脚本是一个无限循环不断从 Stream 里取消息调用 Agent 的 run 方法成功之后 ack。这里有个细节如果 Agent 抛出异常我不会马上 ack而是让消息留在 pending 列表里等下一次消费重试。配合 Redis 的 xpending 命令可以监控堆积情况。生产环境下我建议再加一层守护用一个定时任务检查 pending 里超过 5 分钟的旧消息执行 xclaim 把它们转移给其他消费者避免个别 Worker 挂掉导致消息卡死。3.4 跑通一个完整任务天气查询与定时提醒我直接拿一个实战例子演示用户通过 HTTP 接口提交了一个任务“明天上海如果下雨就提醒我带伞顺便查一下后天杭州的天气”。这个任务实际上要被 Planner Agent 拆成两个子任务一个查天气一个设置定时提醒。Planner Agent 拆解后发布两条 task_type 不同的消息一条是 weather_query一条是 reminder_create分别被对应的 Worker 消费。这个场景里最核心的一点是子任务的执行结果最终要汇总回主任务。我实现了一种简单的关联机制每个子任务消息里带上 parent_task_id子任务完成后把结果写入 Redis 的 Hashkey 是 parent_task_idfield 是 task_type。Planner 等到所有子任务的 field 都齐了就把结果组装成最终回复。这套做法不优雅但非常实用复杂度可控调试也直观。# main.py from fastapi import FastAPI from bus.redis_stream import RedisStreamBus app FastAPI() bus RedisStreamBus() app.post(/task) async def create_task(prompt: str): task_message { task_id: uuid4().hex, task_type: planner, payload: {prompt: prompt}, created_at: time.time(), } bus.publish(task_message) return {task_id: task_message[task_id]}演示重点其实不是代码量而是整个链路能跑通。我第一次跑通的时候超级兴奋虽然当时的输出还很粗糙但已经能从一张白纸式的任务描述走到“拆任务—查天气—设置提醒—汇总结果”的完整流程。那一刻你就会理解Agent 真正值钱的地方不是模型对话而是它把模型能力嵌进业务流程的能力。4. 实操中的坑这些问题我踩了不止一次任何框架只有跑到生产环境坑才会真正暴露出来。hermes-agent 从玩具到能稳定跑任务我反复踩过几个问题每一个都伴随过线上事故级别的排查。这里整理出来希望你能跳过这些坑。4.1 上下文爆炸该裁剪的缓存一定要裁剪Agent 每轮循环都会往上下文里塞工具结果工具结果一多messages 的体积迅速膨胀。模型输出质量会随着上下文变长而下降尤其是中段位置的信息容易被遗忘这个问题业界叫“lost in the middle”。我第一次跑一个 20 步的长任务时上下文直接炒到了 3 万多 token响应延迟翻了五倍而且后面的轮次模型开始漏掉工具参数。后来我在每次循环结束之后做了上下文压缩。策略很简单只保留最初的 system 指令、最近两轮的用户输入和助手输出、最近的工具结果中间的历史对话每隔五轮做一次摘要把摘要作为一条 system 消息放在最前面。这个方案实现成本低但对效果的提升立竿见影。有条件的读者可以更进一步用摘要模型专门做历史压缩不过对于大部分场景截断加摘要已经够用了。4.2 工具调用超时没有超时控制的 Agent 会卡死Agent 调用外部工具本质上是发一个 HTTP 请求。如果第三方 API 迟迟不返回Agent 就会一直干等任务队列里的消息越积越多。我最早没有给工具调用设超时有一次对接的天气服务商接口挂了整个任务队列直接卡了半个多小时。修复方式是在工具注册中心统一加超时与重试机制。我在 ToolRegistry.call 里包了一层 asyncio.wait_for默认超时时间 10 秒超时后抛错交给上层走重试逻辑。重试间隔用指数退避第一次 1 秒第二次 2 秒第三次 4 秒最多三次。接下来是熔断一个工具连续失败 5 次后我直接把它标记为不可用后续请求不再调用它避免把资源耗死在已挂掉的服务上。这套机制上线后整个 Agent 的可用性稳定了很多。4.3 消息重复消费与 Agent 死锁Redis Stream 的消费机制是 at-least-once也就是说在网络抖动或者 Worker 宕机的时候消息可能被重复投递。我的第一次多 Agent 协作 demo 就遇到过一个问题同一个定时提醒被创建了两次因为 Worker 在创建提醒之后、发送 ack 之前挂了重启后消息被重新消费。针对这种问题我引入了幂等机制。每条消息带一个 task_id消费者在处理前先查 Redis 里的 processed_tasks 集合如果 task_id 已经存在就直接跳过。对于“创建提醒”这类对外部世界有副作用的操作幂等尤其重要不做好就会重复扣费、重复发消息。另一种容易踩的是 Agent 死锁比如 Planner 等两个子任务的结果而其中一个子任务的消费者挂了。我的对策是给所有汇总操作加一个整体超时时间并且定期检测 pending 状态超过 10 分钟仍然没有完成的父任务直接标记失败并通知用户重试。分布式下“等待”从来都不是免费的必须设计超时和兜底。4.4 排查问题的日志规范Agent 系统调试起来比普通 API 服务麻烦很多因为一次任务会跨越多个模块、多个消息、多次循环。没有好的日志规范出问题的时候只能靠猜。我从第二版开始强制所有关键节点打结构化日志每条日志必须包含 task_id、step、event 三个字段。我给关键环节定了事件名task_received、plan_created、tool_called、tool_result_received、task_succeeded、task_failed。排错的时候只需要 grep task_id就能把一次任务的完整生命周期串起来。另外我会把工具调用的入参和出参脱敏后都打到日志里这样模型传错了参数也能立刻看出来。这个习惯帮我省下了无数排查时间。4.5 常用配置速查表下面的表格是我自己项目里的常用配置项参数设置可以根据实际模型和任务复杂度调整。配置项推荐值说明MAX_STEPS8单任务最大循环步数过长会浪费 tokenTOOL_TIMEOUT10s单次工具调用超时MAX_RETRIES3工具失败最大重试次数RETRY_BACKOFF2^n * 0.5s指数退避基数CIRCUIT_BREAK_THRESHOLD5连续失败 N 次触发熔断CONTEXT_RECENT_TURNS2上下文裁剪时保留的最近交互轮数SUMMARY_INTERVAL5每 N 轮历史做一次摘要压缩PARENT_TASK_TIMEOUT600s父任务汇总子任务的最大等待时间5. 写在最后hermes-agent 的下一步扩展我做 hermes-agent 的时间不算长但它已经成为了我日常折腾自动化工作流的基础底座。最初只是为了解决“多步骤任务容易崩”的小问题做着做着就意识到消息驱动的架构给这个项目带来了很强的扩展性后来加提醒、加定时任务、加多 Agent 协作都没有改动主循环的骨架。所以我很建议你也从这种架构切入而不是一上来就追求炫酷的 Agent 能力。根据我自己的经验下一步值得尝试的方向大概有三个。第一个是加一个评估系统把任务的输入输出收集成样本用一套打分逻辑判断模型每一步决策是否合理这样能快速定位是哪一环出了问题。第二个是长期记忆层目前多数 Agent 都是无状态的任务跑完上下文就丢了如果接一个向量数据库存历史交互摘要同一个用户再次提出相关任务时就能直接复用之前的信息体验会好很多。第三个是任务依赖编排现在的 parent_task_id 其实只支持简单的扇出汇聚复杂场景可以引入 DAG 图结构让任务之间支持串行、并行、条件分支。最后分享一个很实用的小技巧给每个 Agent 起一个容易辨认的名字在日志和消息里带上 agent_name 字段。多 Agent 协作的时候日志里不同 Agent 的痕迹全混在一起有名字一过滤就清爽了。我自己的规划 Agent 叫 planner执行 Agent 叫 executor提醒 Agent 叫 messenger光看名字就能判断哪一环出现问题排查效率高到离谱。这个小习惯我强烈推荐你也用上。
消息驱动AI Agent框架hermes-agent的设计与落地实践
1. 为什么我会做 hermes-agent从“提示词蘸一切”到消息驱动hermes-agent 这个词懂点技术的人都看得出来这是一个叫 Hermes 的 Agent 项目。Hermes 是希腊神话里替众神跑腿的信使脚底生风负责把消息从一个神传到另一个神那里。我给自己写的 Agent 框架起这个名字就是看中它“传递任务、协调调度”的定位。跟传统的单体 Prompt 应用不一样的是hermes-agent 的核心是一个消息协议——所有任务进来之后都被标准化成一条消息经过解析、路由、执行再把结果回传。整套流程下来你面对的不再是一个“只会聊天的模型接口”而是一个能同时对接多个工具、多个子任务、甚至多个 Agent 协作的调度枢纽。我最早做这个项目是因为被 AI Agent 的工程化问题反复折磨。单轮调用大模型做点小功能其实很简单难的是让它在一个真实业务里稳定干活。比如你让它“帮我查一下明天到上海的高铁然后定一个提醒再把结果整理成邮件发出来”这种事如果只靠一段 Prompt 塞给模型它大概率会胡编乱造工具参数或者在某个步骤上卡住就整体崩掉。hermes-agent 就是奔着解决这个问题去的把任务拆成“规划—执行—反馈”循环每一步都走消息队列每一步都有超时和重试工具调用也全部标准化注册。谁适合用它我觉得三类人最合适第一自己在做个人助理或者自动化工作流的开发者第二团队里需要给多个业务方提供统一 Agent 能力的后端工程师第三想搞懂 Agent 内部消息流转机制的学习者。这套设计的远期收益其实比功能本身更重要。hermes-agent 的各类子模块彼此解耦意味着你换一个大模型 API、加一个新的工具、甚至把单 Agent 扩展成多 Agent 协作都不用推倒重来。很多项目做着做着就变成“屎山”问题就出在最开始没有把消息链路定清楚。我在这篇文章里会把你从设计思路到代码落地完整带一遍重点讲我踩过的坑和那些文档里不会写的取舍。1.1 Hermes 这个名字背后的设计隐喻很多人起名字只是为了好听但 Hermes 这个名字恰好映射了系统的核心隐喻信使。在 hermes-agent 里所有组件之间不直接互相调用而是通过消息互相通信。Agent 核心是一个信使它接收外界发来的任务消息把消息转译成模型能理解的指令模型给出决策后信使再把决策转译成工具调用拿到工具结果后再一次把结果转译回模型可读的观察。一个完整的 Agent 回合本质上就是消息的编码、路由、解码循环。这种设计的直接好处是你可以随时在消息链路上插入新环节。想记录日志在消息进队列的时候加一个拦截器就行。想给任务限流消息队列天然支持。想让两个 Agent 协作让一个 Agent 把产出消息发给另一个 Agent 的队列就好。我在最初版直接把工具函数的返回值拼接进 Prompt 让模型继续思考结果发现上下文越来越乱出错的概率呈指数上升。后来我把每个步骤的输入输出都规范成消息结构模型只看当前需要的部分整个系统的稳定性立刻上了一个台阶。所以“Hermes”这个名字不是装饰它就是整个架构的工作方式。1.2 整体架构一条消息在主链路上怎么流转hermes-agent 的整体架构一句话概括就是一个核心循环两套基础设施三类标准化接口。核心循环是 Agent 主循环负责驱动模型反复走“思考—行动—观察”的过程两套基础设施分别是工具注册中心和记忆模块三类标准化接口是任务消息、工具描述、模型反馈。先看任务消息。我定义了一个统一的消息结构包含 task_id、task_type、payload、created_at 这些字段不管任务来自 HTTP 接口、定时触发还是另一个 Agent都必须先包装成这种结构再进入队列。task_type 尤其关键它是路由的关键字段比如 weather、reminder、summarize消费者会根据 task_type 决定把任务交给哪个处理器。payload 里存放业务参数比如“查询上海明天天气”就把城市和日期放进去。这种标准化契约带来的收益是你后续加新任务类型完全不用改主框架只需要新增一个 task_type 的处理器即可。主链路的流程大概是这样的外部请求进来统一打包成 TaskMessage投递到 Redis Stream 任务队列Agent Worker 从队列里消费消息调用规划器让大模型生成下一步动作如果动作是调用工具就把参数交给 ToolRegistry 执行工具结果重新包装成观察消息再次交给模型直到模型输出 finish 动作Agent 把最终结果回写队列或者直接返回调用方。链路虽然长每一跳都有迹可循排查问题的时候打开日志就能看清楚消息在哪一环出了岔子。2. 核心模块设计三个决定成败的关键选择我在写 hermes-agent 之前也参考过一些开源 Agent 框架比如 LangChain 的 AgentExecutorAutoGPT 的整套任务栈。结果发现一个问题这些框架封装层级太多中间夹了不少黑盒逻辑出了问题很难查。所以我在自己的项目里坚持“简单的轮子自己造复杂的对接才引入依赖”。核心模块就三个Agent 主循环、工具注册中心、多 Agent 通信协议。这三个模块从第一天起就是定制设计的后面所有新功能都是围绕它们长出来的。2.1 Agent 主循环用 ReAct 模式还是纯粹的 Function Calling关于 Agent 主循环用什么模式我在 ReAct 和 Function Calling 之间纠结了很久。ReAct 模式比较经典思路是让模型交替输出 Thought、Action、Observation自由度最高但缺点是需要写大量解析逻辑而且模型的输出稍微不规范就容易解析失败。Function Calling 是 OpenAI 等厂商提供的原生能力模型直接输出结构化的函数调用参数解析稳很多但前提是你用的模型必须支持这个能力。hermes-agent 最终选择了“以 Function Calling 为主保留 ReAct 兜底”的双轨设计。主路径上我们给模型传入工具描述列表模型返回的标准 JSON 里包含 action 和 args 两个字段Agent 直接拿这个字段去工具注册中心执行调用。碰到某些本地模型不支持 Function Calling 的情况就退回到 ReAct 的纯文本解析从输出里截取 Action: xxx 和 Action Input: xxx 字段。双轨设计确实多写了一点代码但换来的是对不同模型厂商的兼容性这个投入很值得。主循环的核心实现可以简化成下面这段伪代码async def agent_loop(task_message): context init_context(task_message) for step in range(MAX_STEPS): response await llm.chat(context.messages, toolsschema) if response.action finish: return response.output tool_result await tool_registry.call(response.action, response.args) context.messages.append(tool_result) return MAX_STEPS_REACHED这里有两个参数需要特别解释。MAX_STEPS 我一般设置成 8太小了复杂任务跑不完太大了模型会陷入无效循环白白消耗 token。另外每次循环结束后我会把历史 messages 做一次裁剪只保留最近三轮的交互记录同时把早期关键信息抽出来放进一个 summary 字段。实战下来这个策略能显著缓解长任务下的上下文漂移问题。2.2 工具注册中心把“会用的技能”标准化工具注册中心是 hermes-agent 里最不起眼但最重要的模块。它解决的核心问题是如何让大模型“知道”你可以调用哪些工具以及如何让工具的结果足够规整方便模型理解。每一类工具在注册的时候要提供 name、description、parameters 三个字段description 要写清楚工具能做什么、适合什么场景parameters 是 JSON Schema 格式的参数定义。很多初学者懒得多写描述结果模型老是选错工具或者填错参数这真的不能怪模型是你描述不够。我用装饰器来做工具注册写起来非常直观tool_registry ToolRegistry() tool_registry.register( nameget_weather, description查询指定城市未来几天的天气情况适合需要出行建议或穿衣建议的场景, parameters{ type: object, properties: { city: {type: string, description: 城市中文名比如 上海}, days: {type: integer, description: 查询天数默认3} }, required: [city] } ) async def get_weather(city: str, days: int 3): # 对接天气服务商API这里做数据清洗 raw await weather_api.query(city, days) return { city: city, forecast: [parse_day(d) for d in raw[days]] }工具执行有个细节容易踩坑函数的返回值不能是任意结构最好固定成 JSON 可序列化的 dict关键的绝对数值和状态要放在顶层字段里。比如天气工具返回时city、condition、temperature 直接暴露给模型避免模型从一大段嵌套 JSON 里去猜。模型猜测成本越低整体稳定性越高。这是我迭代了很多版本后总结出来的经验。2.3 多 Agent 协作的消息协议task_type 是约定也是边界单 Agent 能处理的任务有限一旦任务量上来或者任务本身需要不同领域的知识就得考虑多 Agent 协作。hermes-agent 的多 Agent 设计没有搞得很复杂核心原则是“每个 Agent 都是一个独立的消息消费者”它们之间唯一的通信方式是消息队列。这样设计的好处是你可以把一个 Agent 部署成多个实例做水平扩展也可以用完全不同的技术栈来写另一个 Agent只要它遵守消息协议就能接入。每个 Agent 创建时会声明自己感兴趣的 task_type 列表。举个例子Planner Agent 收到用户复杂请求后会把任务拆成多个子任务通过 bus.publish(task_typeexecute_code, payload{...}) 发给代码执行 Agent代码执行 Agent 出结果后再发给 Report Agent 做汇总。任务类型定了协作边界就清晰了不会出现两个 Agent 抢同一个任务的混乱情况。我用 Redis Stream 的 consumer group 来做消息分配之后还顺带获得了重试和故障转移能力消费失败的消息可以重新进入 pending 队列由另一个 Worker 继续处理这一点在单机内存队列里很难实现。3. 从零手写一个可运行的 hermes-agent前面讲了设计现在说落地。我接下来用一个最小可运行版本演示你照着敲一遍就能跑起来。核心依赖只有三个FastAPI 提供 HTTP 入口、Redis 作为任务队列、OpenAI 兼容接口的大模型。我自己平时用 Ollama 跑本地模型测试因为调 token 方便开发迭代速度比直接调云端 API 快很多。3.1 项目结构与依赖模块拆得清楚一点后面加功能会省很多力气。我的参考结构是这样的hermes-agent/ ├── agent/ │ ├── core.py # Agent 主循环 │ ├── registry.py # 工具注册中心 │ ├── memory.py # 上下文与长期记忆 │ └── message.py # 消息协议定义 ├── bus/ │ └── redis_stream.py # Redis Stream 封装 ├── tools/ │ ├── weather.py # 天气工具 │ └── reminder.py # 提醒工具 ├── main.py # HTTP 入口 Worker 启动 └── requirements.txtrequirements.txt 里主要就这些fastapi、uvicorn、redis、openai、pydantic。其中 openai 库即使对接 Ollama 也能用只要把 base_url 改成本地地址。这算是 OpenAI SDK 的一个隐形福利生态里很多东西都能复用。3.2 Agent 主循环代码实现主循环是 Agent 的“大脑”它做三件事组织上下文、调用模型、执行工具。代码结构如下# agent/core.py import asyncio from dataclasses import dataclass, field dataclass class AgentContext: task_message: dict messages: list field(default_factorylist) step: int 0 class HermesAgent: def __init__(self, name, llm_client, tool_registry, max_steps8): self.name name self.llm_client llm_client self.tool_registry tool_registry self.max_steps max_steps async def run(self, task_message: dict) - dict: ctx AgentContext(task_messagetask_message) ctx.messages.append({ role: system, content: 你是一个任务规划助手请根据可用的工具逐步完成任务。 }) ctx.messages.append({ role: user, content: task_message[payload].get(prompt, ) }) while ctx.step self.max_steps: ctx.step 1 response await self.llm_client.chat( messagesctx.messages, toolsself.tool_registry.schema() ) if response.action finish: return { task_id: task_message[task_id], status: done, output: response.output, } if response.action not in self.tool_registry: return { task_id: task_message[task_id], status: failed, error: funknown action: {response.action} } result await self.tool_registry.call(response.action, response.args) ctx.messages.append({ role: assistant, content: , tool_calls: [{ id: fcall_{ctx.step}, name: response.action, arguments: response.args }] }) ctx.messages.append({ role: tool, tool_call_id: fcall_{ctx.step}, content: json.dumps(result, ensure_asciiFalse) }) return { task_id: task_message[task_id], status: timeout, error: fexceeded max_steps: {self.max_steps} }你可能已经发现了我把模型的响应结构假设成包含 action、args、output 三种字段。如果你用的是 OpenAI 官方 SDK它返回的 tool_calls 是标准结构需要做一个转换如果你走 Ollama返回结构会稍有差异。所以这里的 llm_client 其实是一个薄封装我自己写了一个 OpenAIAdapter 和 OllamaAdapter统一输出成上面这个简化结构。这是整个框架里最值得复用的设计之一后面换模型服务商只需要新增一个 Adapter。3.3 用 Redis Stream 支撑任务队列任务队列我选了 Redis Stream而不是 Celery 或者 RabbitMQ理由很简单Redis 大多数项目里已经有了不需要额外养一个消息中间件而且 Stream 支持消费组、消息确认、pending 列表足够撑住 Agent 场景的任务调度。关键代码如下# bus/redis_stream.py import json import redis class RedisStreamBus: def __init__(self, redis_url, stream_namehermes:task:stream): self.redis redis.from_url(redis_url) self.stream stream_name self.group hermes:workers def ensure_group(self): try: self.redis.xgroup_create(self.stream, self.group, id0, mkstreamTrue) except redis.ResponseError: # 消费组已存在时会报错忽略 pass def publish(self, message: dict): self.redis.xadd(self.stream, {data: json.dumps(message)}) def consume(self, count1, block5000): entries self.redis.xreadgroup( self.group, self.consumer_name, {self.stream: }, countcount, blockblock, ) result [] if entries: for _, messages in entries: for msg_id, fields in messages: result.append((msg_id, json.loads(fields[data]))) return result def ack(self, msg_id): self.redis.xack(self.stream, self.group, msg_id)消费端脚本是一个无限循环不断从 Stream 里取消息调用 Agent 的 run 方法成功之后 ack。这里有个细节如果 Agent 抛出异常我不会马上 ack而是让消息留在 pending 列表里等下一次消费重试。配合 Redis 的 xpending 命令可以监控堆积情况。生产环境下我建议再加一层守护用一个定时任务检查 pending 里超过 5 分钟的旧消息执行 xclaim 把它们转移给其他消费者避免个别 Worker 挂掉导致消息卡死。3.4 跑通一个完整任务天气查询与定时提醒我直接拿一个实战例子演示用户通过 HTTP 接口提交了一个任务“明天上海如果下雨就提醒我带伞顺便查一下后天杭州的天气”。这个任务实际上要被 Planner Agent 拆成两个子任务一个查天气一个设置定时提醒。Planner Agent 拆解后发布两条 task_type 不同的消息一条是 weather_query一条是 reminder_create分别被对应的 Worker 消费。这个场景里最核心的一点是子任务的执行结果最终要汇总回主任务。我实现了一种简单的关联机制每个子任务消息里带上 parent_task_id子任务完成后把结果写入 Redis 的 Hashkey 是 parent_task_idfield 是 task_type。Planner 等到所有子任务的 field 都齐了就把结果组装成最终回复。这套做法不优雅但非常实用复杂度可控调试也直观。# main.py from fastapi import FastAPI from bus.redis_stream import RedisStreamBus app FastAPI() bus RedisStreamBus() app.post(/task) async def create_task(prompt: str): task_message { task_id: uuid4().hex, task_type: planner, payload: {prompt: prompt}, created_at: time.time(), } bus.publish(task_message) return {task_id: task_message[task_id]}演示重点其实不是代码量而是整个链路能跑通。我第一次跑通的时候超级兴奋虽然当时的输出还很粗糙但已经能从一张白纸式的任务描述走到“拆任务—查天气—设置提醒—汇总结果”的完整流程。那一刻你就会理解Agent 真正值钱的地方不是模型对话而是它把模型能力嵌进业务流程的能力。4. 实操中的坑这些问题我踩了不止一次任何框架只有跑到生产环境坑才会真正暴露出来。hermes-agent 从玩具到能稳定跑任务我反复踩过几个问题每一个都伴随过线上事故级别的排查。这里整理出来希望你能跳过这些坑。4.1 上下文爆炸该裁剪的缓存一定要裁剪Agent 每轮循环都会往上下文里塞工具结果工具结果一多messages 的体积迅速膨胀。模型输出质量会随着上下文变长而下降尤其是中段位置的信息容易被遗忘这个问题业界叫“lost in the middle”。我第一次跑一个 20 步的长任务时上下文直接炒到了 3 万多 token响应延迟翻了五倍而且后面的轮次模型开始漏掉工具参数。后来我在每次循环结束之后做了上下文压缩。策略很简单只保留最初的 system 指令、最近两轮的用户输入和助手输出、最近的工具结果中间的历史对话每隔五轮做一次摘要把摘要作为一条 system 消息放在最前面。这个方案实现成本低但对效果的提升立竿见影。有条件的读者可以更进一步用摘要模型专门做历史压缩不过对于大部分场景截断加摘要已经够用了。4.2 工具调用超时没有超时控制的 Agent 会卡死Agent 调用外部工具本质上是发一个 HTTP 请求。如果第三方 API 迟迟不返回Agent 就会一直干等任务队列里的消息越积越多。我最早没有给工具调用设超时有一次对接的天气服务商接口挂了整个任务队列直接卡了半个多小时。修复方式是在工具注册中心统一加超时与重试机制。我在 ToolRegistry.call 里包了一层 asyncio.wait_for默认超时时间 10 秒超时后抛错交给上层走重试逻辑。重试间隔用指数退避第一次 1 秒第二次 2 秒第三次 4 秒最多三次。接下来是熔断一个工具连续失败 5 次后我直接把它标记为不可用后续请求不再调用它避免把资源耗死在已挂掉的服务上。这套机制上线后整个 Agent 的可用性稳定了很多。4.3 消息重复消费与 Agent 死锁Redis Stream 的消费机制是 at-least-once也就是说在网络抖动或者 Worker 宕机的时候消息可能被重复投递。我的第一次多 Agent 协作 demo 就遇到过一个问题同一个定时提醒被创建了两次因为 Worker 在创建提醒之后、发送 ack 之前挂了重启后消息被重新消费。针对这种问题我引入了幂等机制。每条消息带一个 task_id消费者在处理前先查 Redis 里的 processed_tasks 集合如果 task_id 已经存在就直接跳过。对于“创建提醒”这类对外部世界有副作用的操作幂等尤其重要不做好就会重复扣费、重复发消息。另一种容易踩的是 Agent 死锁比如 Planner 等两个子任务的结果而其中一个子任务的消费者挂了。我的对策是给所有汇总操作加一个整体超时时间并且定期检测 pending 状态超过 10 分钟仍然没有完成的父任务直接标记失败并通知用户重试。分布式下“等待”从来都不是免费的必须设计超时和兜底。4.4 排查问题的日志规范Agent 系统调试起来比普通 API 服务麻烦很多因为一次任务会跨越多个模块、多个消息、多次循环。没有好的日志规范出问题的时候只能靠猜。我从第二版开始强制所有关键节点打结构化日志每条日志必须包含 task_id、step、event 三个字段。我给关键环节定了事件名task_received、plan_created、tool_called、tool_result_received、task_succeeded、task_failed。排错的时候只需要 grep task_id就能把一次任务的完整生命周期串起来。另外我会把工具调用的入参和出参脱敏后都打到日志里这样模型传错了参数也能立刻看出来。这个习惯帮我省下了无数排查时间。4.5 常用配置速查表下面的表格是我自己项目里的常用配置项参数设置可以根据实际模型和任务复杂度调整。配置项推荐值说明MAX_STEPS8单任务最大循环步数过长会浪费 tokenTOOL_TIMEOUT10s单次工具调用超时MAX_RETRIES3工具失败最大重试次数RETRY_BACKOFF2^n * 0.5s指数退避基数CIRCUIT_BREAK_THRESHOLD5连续失败 N 次触发熔断CONTEXT_RECENT_TURNS2上下文裁剪时保留的最近交互轮数SUMMARY_INTERVAL5每 N 轮历史做一次摘要压缩PARENT_TASK_TIMEOUT600s父任务汇总子任务的最大等待时间5. 写在最后hermes-agent 的下一步扩展我做 hermes-agent 的时间不算长但它已经成为了我日常折腾自动化工作流的基础底座。最初只是为了解决“多步骤任务容易崩”的小问题做着做着就意识到消息驱动的架构给这个项目带来了很强的扩展性后来加提醒、加定时任务、加多 Agent 协作都没有改动主循环的骨架。所以我很建议你也从这种架构切入而不是一上来就追求炫酷的 Agent 能力。根据我自己的经验下一步值得尝试的方向大概有三个。第一个是加一个评估系统把任务的输入输出收集成样本用一套打分逻辑判断模型每一步决策是否合理这样能快速定位是哪一环出了问题。第二个是长期记忆层目前多数 Agent 都是无状态的任务跑完上下文就丢了如果接一个向量数据库存历史交互摘要同一个用户再次提出相关任务时就能直接复用之前的信息体验会好很多。第三个是任务依赖编排现在的 parent_task_id 其实只支持简单的扇出汇聚复杂场景可以引入 DAG 图结构让任务之间支持串行、并行、条件分支。最后分享一个很实用的小技巧给每个 Agent 起一个容易辨认的名字在日志和消息里带上 agent_name 字段。多 Agent 协作的时候日志里不同 Agent 的痕迹全混在一起有名字一过滤就清爽了。我自己的规划 Agent 叫 planner执行 Agent 叫 executor提醒 Agent 叫 messenger光看名字就能判断哪一环出现问题排查效率高到离谱。这个小习惯我强烈推荐你也用上。
)
