如果你做过几个Agent demo大概率会经历这个过程一开始用LangChain这类重框架写起来确实快但等你想改一点底层的消息流、想精确定制工具调用的时机或者排查一个问题时往往会一头扎进层层抽象里半天找不着北。这正是我决定自研并开源 hermes-agent 的原因——一个以消息驱动为核心的轻量级Agent运行时。Hermes是希腊神话里的信使之神负责传递信息、引导灵魂穿越边界。Agent要做的事情本质上也一样接收任务、调度工具、在LLM和外部世界之间传递消息。这个名字不是拍脑袋起的它直接点出了框架的核心设计理念——一切皆为消息。hermes-agent 不打算成为另一个包罗万象的大平台它只解决一个问题如何用最少的抽象让开发者清楚看到每一轮“LLM思考 - 工具调用 - 结果回传”的完整链路。这篇文章我会从框架的核心设计动机讲起然后带你走一遍最小可运行的Agent从注册到工具调用的完整链路再深入到任务编排、状态管理和可观测性的实现思路最后分享几个我在真实业务场景里踩过的坑。适合两类人看一类是好奇Agent底层原理、想自己掌控每个环节的开发者另一类是被重框架折磨过、想找一个高可控性轻量方案的工程师。1. 为什么我不再用大而全的Agent框架先说说我为什么会对现有框架不满。这不是说大框架不好它们解决了很多人的问题但当你需要精细控制的时候框架反而会成为阻力。1.1 重框架的抽象层级太多了用过一段时间LangChain的人应该都有体会Chain、Agent、Tool、Memory、Callback、OutputParser……每个概念都有若干种变体类与类之间互相引用。你写一个简单的“调用工具再让LLM总结结果”的流程背后要经过至少六七个抽象层。多数情况下我们并不需要这么多层Agent的核心逻辑拆到最底其实就三个环节LLM根据当前对话状态生成下一步动作。如果这个动作是调用工具就执行工具并返回结果。把结果喂回给LLM继续循环直到LLM认为任务完成。这个循环在5分钟内就能写出来。问题是光写出来还不够——真实场景里你还要考虑任务编排、并发控制、上下文裁剪、成本控制、可观测性。大框架把前面那5分钟的活简化了但后面工程化的部分反而因为抽象过多变得更难做。1.2 我需要的是“可观测”而不仅仅是“能跑”我接手过一个用某个重框架写的客服机器人出问题时你根本不知道Agent在哪个环节开始跑偏。Log里全是框架自己的事件自定义信息的传递要翻源码。而像hermes-agent这样消息驱动的设计天然就有优势每一次LLM输出、每一个工具调用、每一条错误信息都是消息流里的一环你可以把它完整地落盘、回放、按链路ID检索。1.3 hermes-agent 的定位所以我把hermes-agent定位成一个“看得见内部运转”的轻量Agent运行时而不是一个完整平台。它的核心依赖只有一个——兼容OpenAI接口的大模型服务。工具注册、消息协议、任务编排、Token统计全部由框架本身提供代码量控制在几千行以内任何一个模块你都可以打开源码直接修改。提示如果你需要企业级的图形化编排、多人协作、大模型管理平台这类功能hermes-agent确实不适合你。它更适合那些愿意自己掌控流程、希望Agent逻辑透明可审计的团队和个人项目。2. 核心设计一切皆为消息的三层协议hermes-agent里没有 Chain、没有 Node、没有 Edge只有消息。整个框架就是一台处理消息的机器LLM生成的消息进来控制层决定下一步该做什么然后继续生成新消息。2.1 消息类型命令、事件、查询与很多框架用一个统一的Message结构不同我把消息分成了三种语义类型消息类型语义典型场景Command命令要求某个模块执行动作调用工具、发HTTP请求、写入数据库Event事件描述已经发生的事实工具执行完成、LLM返回结果、任务超时Query查询请求获取信息读取上下文、查询会话历史、获取工具列表这个设计一开始被朋友说“过度设计”但在实际跑业务的时候我发现它是必要的。Agent在处理复杂任务时如果不区分“我想干什么”和“发生了什么”调试时根本无法定位是消息发送方的问题还是执行方的问题。举个具体例子工具调用超时如果只存一条“Timeout”消息你没法知道是工具执行很久之后才超时还是消息根本没送达。区分Command和Event之后超时发生时我们能同时看到“Command已发送”和“Event长时间未返回”两个事实排查范围瞬间缩小。2.2 消息头部里的可追溯信息每条消息除了业务payload头部还携带了四个关键字段trace_id链路追踪ID、sender发送方、receiver接收方、sequence全局递增序号。from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional from datetime import datetime class MessageType(Enum): COMMAND command EVENT event QUERY query dataclass class AgentMessage: type: MessageType sender: str receiver: str payload: Any trace_id: str field(default_factorylambda: uuid4().hex) sequence: int 0 timestamp: datetime field(default_factorydatetime.now) parent_id: Optional[str] None # 用于关联父子消息sequence字段是消息队列里传进来的全局自增序号靠它才能实现发送顺序的严格还原。有一次我排查一个工具调用结果覆盖的bug——两个工具并行返回结果旧消息反而后到把新消息覆盖了。正是靠着sequence字段才确认是并发下消息写入时序问题而不是逻辑问题。2.3 消息总线与Agent主循环每个Agent实例内部维护一个消息总线主循环不断从总线里取消息、分发、处理。伪代码大致如下async def run(self, task: str): self.context.set(user_task, task) await self.bus.publish(AgentMessage( typeMessageType.COMMAND, senderorchestrator, receiverllm, payload{instruction: task} )) while not self.is_finished(): msg await self.bus.receive() processed await self.dispatch(msg) if processed.is_tool_result(): await self.feed_back_to_llm(processed)这里面的核心思路是Agent的运行过程不是一段顺序执行的代码而是一个有限状态的事件循环。这带来的直接好处是你可以在任意节点插入暂停、记录、校验逻辑而不需要侵入业务代码。3. 跑通最小可用Agent从工具注册到工具调用前面聊了不少原理现在进入实操环节。这部分我会带你从零搭一个最小可用的Agent它能完成“查天气并总结”这种典型的复合任务。3.1 环境准备与安装要求Python 3.10操作系统不限。安装方式推荐直接Clone仓库用源码运行因为hermes-agent本身就不大源码方式方便你随时进去看实现。git clone https://github.com/yourname/hermes-agent.git cd hermes-agent pip install -e .安装好后需要配置环境变量指向你的大模型API。hermes-agent只要求接口兼容OpenAI规范所以不管是官方服务还是其他兼容网关都能直接对接export LLM_API_KEYyour_api_key export LLM_BASE_URLhttps://api.example.com/v1 export LLM_MODELyour-model-name3.2 定义Agent实例和工具工具注册用装饰器的方式框架会自动读取函数的签名和docstring生成给LLM看的工具Schema。这里有个小坑函数的docstring一定要写清楚参数含义和边界情况因为LLM真正依赖的是这个描述来生成结构化调用参数。import asyncio from hermes_agent import Agent, tool agent Agent( nameweather_assistant, system_prompt你是一个天气助手当用户询问天气时调用工具获取信息。, max_iterations10, ) tool async def get_weather(city: str, date: str today) - dict: 获取指定城市的天气信息。 参数: city: 城市名中文如北京 date: 日期默认今天格式YYYY-MM-DD # 实际项目中这里替换为真实天气API调用 mock_data { 北京: {today: 晴22°C微风}, 上海: {today: 多云25°C东南风3级}, } return {city: city, date: date, weather: mock_data.get(city, 暂无数据)} agent.register_tool(get_weather)注意tool装饰器建议直接用在async函数上。Agent主循环是异步的如果工具是同步阻塞的比如某SDK只提供同步方法要自己用asyncio.to_thread包一层否则一个慢工具会卡住整个事件循环。3.3 运行任务与内部消息流转创建好了Agent和工具后运行一个任务async def main(): result await agent.run(北京今天天气怎么样可以出门吗) print(result) asyncio.run(main())运行过程中hermes-agent默认会在日志里打印完整的消息流转输出大致长这样[01] COMMAND orchestrator - llm: 北京今天天气怎么样 [02] EVENT llm - orchestrator: tool_calls[get_weather(city北京)] [03] COMMAND orchestrator - tool.get_weather: {city: 北京} [04] EVENT tool.get_weather - orchestrator: {weather: 晴22°C微风} [05] COMMAND orchestrator - llm: 工具结果为... [06] EVENT llm - orchestrator: 北京今天晴22°C适合出门。 [07] EVENT orchestrator - final: 北京今天晴22°C适合出门。这七条消息完整记录了一次Agent交互。我第一次把自己的业务接到这个框架时看着这份日志直接把之前的调试工具全扔了——链路太清楚了。3.4 关键参数说明max_iterations一定要设置。LLM在遇到无法解决的问题时会陷入循环调用工具不设上限等于给Token烧钱开绿灯。我一般默认给8复杂任务给15超过阈值直接中断并返回当前收集到的信息。temperature建议固定为0或接近0Agent场景里我们要的是稳定的工具调用不需要创造性输出。4. 任务编排与状态管理串行、并行和条件分支单个Agent能做的事情有限实际业务里通常需要编排多个步骤。hermes-agent没有引入独立的“编排器”概念而是直接用消息模式来实现编排逻辑。4.1 串行链路前一步的输出就是下一步的输入串行是最常见的编排模式。比如“先查询订单状态再根据状态生成回复”这两个步骤必须按顺序执行。在hermes-agent里串行不是通过配置一个Chain对象实现的而是靠Command消息的parent_id挂钩from hermes_agent import Orchestrator orch Orchestrator(agent) # 第一个任务 query_result await orch.run_step(查询订单ORD-2024-001的状态) # 第二个任务直接使用前一个任务的结果 final_reply await orch.run_step(f根据以下订单状态生成回复{query_result})这里的run_step内部会创建一条新的Command消息并把上一条Event设为parent_id。这样整条链路的因果关系在消息流里一目了然任何一个步骤出问题都可以根据parent_id回溯到它的输入是什么。4.2 并行工具调用LLM一次返回多个tool_calls很多场景下工具之间没有依赖比如“同时查北京和上海的天气再对比”。hermes-agent支持LLM一次返回多个structured tool_calls框架会并发执行这些调用然后合并结果喂回给LLM。# 框架内部对多个tool_calls的执行逻辑 async def _execute_tool_calls(self, tool_calls: list[dict]): tasks [] for call in tool_calls: func self.tools[call[name]] tasks.append(func(**call[arguments])) return await asyncio.gather(*tasks, return_exceptionsTrue)这里有几个并发控制的关键点我在实际项目中踩过必须给每个工具调用设置独立超时。用asyncio.wait_for包裹超时部分返回一个明确的错误Event而不是把异常冒泡。一个工具超时不应该拖垮整个并行组。并发上限要控制。默认并发数是4如果不控制一次返回10个tool_calls就同时打10个外部API很容易触发上游限流。return_exceptionsTrue一定要加。否则一个工具挂了所有结果都拿不到。4.3 条件分支用Router消息实现Agent场景里我不推荐用复杂的图编排来做条件分支——LLM本身就是最好的Router。你只需要在系统提示词里约定好判断规则并给Agent注册一个“路由工具”tool async def route_to_department(issue_type: str) - str: 根据问题类型路由到对应处理流程。 参数: issue_type: 取值billing/technical/general之一 route_map { billing: billing_handler, technical: technical_handler, } return route_map.get(issue_type, general_handler)这样“条件分支”本质上仍然是一次工具调用没有引入新的抽象。好处是规则可以随时用自然语言修改不用改代码坏处是依赖LLM的判断准确性——所以在关键业务上我通常会在分支前加一道规则校验防止LLM把billing路由到technical去。4.4 状态隔离每个任务会话独立的上下文状态管理是Agent框架最容易出问题的部分。我在设计时把状态分成两层任务级状态当前这轮任务Task的输入输出存储在context对象里任务结束即释放。会话级状态跨多轮任务共享的用户偏好、历史摘要存储在可插拔的memory_backend里默认基于内存也可以接Redis或向量库。agent Agent( namesupport_bot, memory_backendRedisMemory(hostlocalhost, port6379), context_ttl600, )任务级状态和会话级状态必须严格隔离。我见过很多Agent“串味”的例子——上一轮任务里的临时数据跑到下一轮的系统提示词里去了。hermes-agent里每创建一条任务都会new一个干净的context只有显式调用context.remember()的内容才会进入会话级状态。5. 可观测性与调试验证让Agent的思考过程可回放我觉得一个Agent框架值不值得用很大程度要看它能不能“兜底”——出问题时你能不能快速搞清楚发生了什么。这一章讲hermes-agent在可观测性和调试验证上的具体设计。5.1 消息日志落盘JSONL格式全量记录hermes-agent默认把每次运行的所有消息追加写入hermes_trace.log每行一条JSON。前端展示时你可以把同一条trace_id的所有消息聚合出来按序号排序即可还原整个会话。{trace_id: a3f9..., seq: 1, type: command, sender: orchestrator, receiver: llm, payload: 北京今天天气怎么样, ts: 2024-06-01T10:00:00Z} {trace_id: a3f9..., seq: 2, type: event, sender: llm, receiver: orchestrator, payload: {tool_calls: [{name: get_weather, arguments: {city: 北京}}]}, ts: 2024-06-01T10:00:01Z}这个JSONL格式是我从事件溯源Event Sourcing模式里借鉴的。每条业务事实都以追加写入的方式持久化不修改历史记录。好处有两个一是写入性能高顺序IO远快于随机IO二是天然可回放你可以把任意一次线上请求的消息流提取出来喂给一个测试Agent做复现。5.2 回放调试模式某个请求出了问题如果你想精确复现当时的场景可以进入回放模式from hermes_agent.debug import ReplaySession session ReplaySession(hermes_trace.log, trace_ida3f9...) await session.replay(step_by_stepTrue) # 可以逐条消息跟进回放模式下LLM接口会走mock层不实际消耗Token工具调用也默认不真实执行而是返回日志里记录的响应。这样就算线上出了诡异问题你也可以离线反复试改逻辑不需要依赖当时的输入数据。5.3 成本与步数统计我在这类框架里最想看到的一个指标是每完成一个任务到底烧了多少Token、调了几次工具。hermes-agent在每次运行结束时输出统计包括LLM调用次数、总输入/输出Token数按消息里的usage字段汇总工具调用次数、成功/失败占比总耗时、各步骤耗时Top5输出示例如下Task completed in 6 iterations. LLM calls: 4 | Tokens in: 12533 | Tokens out: 462 Tool calls: 3 | succeeded: 3 | failed: 0 Latency: 3.82s total | slowest: get_weather(1.9s)这个统计在优化prompt时特别有用。我做过一个实验只是调整了工具的描述措辞工具调用失败率就从13%降到了5%单任务平均Token消费下降约30%。没有数据你根本不知道改动效果如何。6. 实战中踩过的三个大坑现象、排查与修复这一章全部来自真实项目的血泪经验我按“现象 - 排查思路 - 修复方案”的结构来写希望能帮你绕开那些共同的大坑。6.1 工具循环风暴Agent重复调用同一个工具且参数不变现象Agent在一个问题上反复调用同一个工具每次参数完全一样返回结果也一样但LLM就是不结束。日志显示second round之后的tool_calls和第一轮完全相同一直循环到max_iterations才被强制打断。排查我先看了消息流确认工具每次返回的结果都正确写入了上下文。然后用回放模式把LLM的完整上下文dump出来逐字检查发现一个细节——工具结果是注入了但我们没有在消息流中保存“这个工具已经在同一轮任务里被调用过”这个事实Agent判断“那次调用”和“这次调用”是完全独立的。修复两条策略并用。第一是幂等拦截框架内置一个调用指纹缓存同一任务内出现完全相同的工具名参数组合时直接返回缓存里的历史结果并标记为重复调用。第二是在系统提示词里追加规则“如果某个工具返回的结果与最近一次完全相同请基于现有信息作答不要再次调用。”指纹缓存的代码位于orchestrator的_execute_tool_calls里# 框架内置的重复调用拦截 async def _with_duplicate_guard(self, name: str, args: dict): fingerprint f{name}:{json.dumps(args, sort_keysTrue)} if fingerprint in self._call_cache: return self._call_cache[fingerprint], True # 返回缓存并标记重复 result await self._invoke_tool(name, args) self._call_cache[fingerprint] result return result, False这招把线上任务的平均迭代次数从5.6降到2.1Token成本直接腰斩。6.2 上下文泄漏工具返回的原始数据撑爆上下文窗口现象一个查询数据库的工具返回了500行记录Agent后续的每次交互都要带着这500行records一起计算几轮之后Token用量飙升甚至直接触发上下文超限。排查我dump了一次完整上下文发现LLM收到的工具结果是“原样注入”的没有任何裁剪。原始数据里有很多字段Agent根本用不上比如数据库记录的created_at、updated_at、内部ID。但LLM不管这些它统统作为“当前状态”读进注意力窗口。修复给工具结果增加一个可选的summary回调。tool(summary_funcsummarize_db_result) async def query_orders(user_id: str) - list[dict]: # 返回大量原始数据 ... def summarize_db_result(raw: list[dict]) - str: # 只提取LLM真正需要的信息 summary { total_orders: len(raw), latest_order: raw[0] if raw else None, status_distribution: Counter(o[status] for o in raw), } return json.dumps(summary, ensure_asciiFalse, indent2)这样LLM真正看到的是提炼后的摘要而不是原始数据。调用详情存在另一张表里需要时可查。这个设计直接让长任务的平均Token量降了一个数量级。6.3 超时风暴与并发告警外部API变慢殃及整个任务现象某次线上事故里上游的一个支付接口异常变慢单次请求从200ms涨到30s。由于Agent是串行等待工具返回的这个慢请求直接把整个任务卡住——用户侧的反馈是“机器人没反应了”实际是事件循环被中的同步工具阻塞。排查通过日志很快定位到是工具超时无兜底。翻了一圈代码发现框架初版编写时没有给工具调用包asyncio.wait_for所以一个工具能跑多久Agent就得等多久。修复第二版加入两层防护。第一层是每个工具独立的超时时间通过tool(timeout5.0)声明第二层是并发信号量默认限制同时进行的工具调用数。tool(timeout8.0) async def call_payment_api(order_id: str) - dict: 调用支付网关查询订单支付状态。 ... # 并发信号量初始化 self._semaphore asyncio.Semaphore(4) async def _invoke_tool(self, name: str, args: dict): async with self._semaphore: tool_def self.tools[name] timeout tool_def.timeout return await asyncio.wait_for( tool_def.func(**args), timeouttimeout )连外部依赖一起做好兜底之后我再没遇到过“一个工具拖垮整个Agent”的情况。这里也想提醒一句工具调用的健壮性永远是Agent系统里优先级最高的事你可以在prompt上省功夫但绝不能在工具超时和并发控制上省。写在最后从开始写hermes-agent到今天我最大的体会是Agent框架的复杂度从来不在“让LLM调用工具”这一步而在工程化——可控、可观测、可评估。很多框架通过增加抽象层来掩盖这些难点但hermes-agent坚持用最少的概念来直面它们。如果你正在自研Agent或者对LangChain的抽象感到疲惫建议先试着写一个最小的事件循环把“消息”作为一等公民来设计再把工具、记忆、编排都挂到消息总线上。这套思路摆脱重框架的束缚同时给你一个足够清晰的骨架来承载复杂的业务逻辑运行几个月下来你会发现维护成本比预想的低很多。