把 LangChain、AutoGen 这些框架的抽象一层层剥掉剩下的东西其实不多任务进来做一次决策调一个工具再把结果传给下一步。顺着这个思路我写了 hermes-agent一个核心代码只有几百行、但能稳定支撑单 Agent 数据整理和多 Agent 分派协作的轻量编排框架。它没有内置 vector store没有花哨的角色模板也没打算做成通用平台。但如果你想折腾明白Agent 到底是怎么把一句话变成一次工具调用这个框架会是一个特别合适的参考样本。1. 从大框架迁到轻量框架我拆掉了哪些抽象1.1 真正碍事的不是大模型是框架的全我早期用成熟框架搭过一个跑批项目功能确实全链式调用、回调系统、记忆持久化、Prompt 模板引擎样样都有。但项目越往后问题越明显。每次版本升级抽象就漏一层以前能直接传参的地方新版本要套一个新的Runnable包装想给某个工具动态注入上下文框架的固定抽象绕来绕去最后写出来的适配代码比业务代码还多。这不是说大框架不好。它适合做 Demo、做验证、做团队规范统一的场景。但对一个想折腾清楚底层机制的人来说它给的太多了。尤其当你的需求只是读取一个 CSV让 Agent 决定怎么汇总再写一个 Markdown 报告这些重量级框架本身的启动成本已经接近任务复杂度。1.2 hermes-agent 的取舍Agent 就是一个带工具的消息处理器所以在做 hermes-agent 的时候我给自己定了三个核心概念其余全都砍掉Message消息、Agent决策者、Tool执行器。你不需要理解回调链、不需要配置模型缓存、不需要学一套专用的图编排 DSL。消息进来Agent 决定怎么办工具去执行结果再作为新消息传回。这套设计最大的好处是所有交互都能被追踪。只要消息格式稳定框架内部怎么改都不会伤到业务。我在迭代中碰到的大部分问题最后都归纳为消息没传对或消息格式不对而不是某个内部机制失效。这也是为什么我一直强调Agent 框架的骨架是消息流转不是模型本身。1.3 和主流方案的核心差异我把自己实际用过的几个框架和 hermes-agent 做了一个直观对照按我的场景来衡量差异很清楚方案核心抽象协作方式上手成本适合场景LangChain / LangGraphChain / Graph图编排较高复杂流程、团队统一技术栈CrewAICrew / Agent / Task角色协作中等快速搭建多角色流程AutoGenConversableAgent多智能体对话中等研究、对话式推理hermes-agentMessage / Agent / Tool消息总线低想搞懂原理、定制自己的编排这不是踩一捧一。既然目标只是把任务流转跑顺一个能看穿全部机制的小框架比一个黑盒大框架更符合我的需求。2. Hermes 不是随便起的名字——消息流转才是 Agent 骨架2.1 信使的隐喻Hermes 在神话里是信使之神负责信息传递、解释和沟通。我起这个名字有两层意思一是 Agent 本质上就是信使从用户消息里接收任务从工具返回值里接收事实再从模型生成里把答案送出去二是我本地部署的开源模型恰好也是 Hermes 系列两个来源一碰名字就这么定了。想清楚Agent 是信使这一点框架设计就简单了。不需要给 Agent 赋予太多戏份它不是一个会思考的实体它是一段反复循环的逻辑接收消息、调用模型做决策、决定调哪个工具、把工具结果回传给模型、生成最终回应。这个循环里消息是唯一的血液。2.2 Message 数据结构一张能追踪任务的票据我在 hermes-agent 里定义了这样一个消息结构# hermes_agent/core/types.py from dataclasses import dataclass, field from enum import Enum from typing import Any class MessageKind(str, Enum): USER user AGENT agent TOOL tool SYSTEM system dataclass class Message: task_id: str # 任务唯一标识跨 Agent 串联日志 sender: str # 谁发的 receiver: str # 发给谁Agent 名或工具名 kind: MessageKind # 消息类型 content: Any # 文本、工具参数、工具结果 metadata: dict field(default_factorydict) turn: int 0 # 第几轮对话用于限制循环这里面最关键的是task_id。多 Agent 协作时同一个任务会在 supervisor、worker、tool 之间产生几十条消息没有task_id你根本没法把散落各处的日志串起来。我在框架里把task_id设成必填字段就是为了避免只看消息内容、不知道属于哪个任务的混乱。turn则是防止 Agent 在工具结果和模型推断之间无限循环的哨兵值后面踩坑部分会细说。2.3 消息总线不直接用函数调用而是用队列Agent 之间不直接互相调用而是通过一个消息总线投递消息。这个设计参考了邮箱模型每个人有自己的邮箱往邮箱里投信邮递员按收件人分拣收件人异步取信。我最初的实现也试图让 Agent 直接调用彼此后来发现只要 A 和 B 互相依赖直接调用就会变成死锁现场。改成消息总线后A 不需要认识 BA 只需要把消息扔给总线由总线负责投递。# hermes_agent/core/bus.py import asyncio class MessageBus: def __init__(self): self._queues: dict[str, asyncio.Queue] {} def mailbox(self, name: str) - asyncio.Queue: 返回某个 Agent 或消费者专属的邮箱队列 return self._queues.setdefault(name, asyncio.Queue()) async def send(self, msg: Message): await self.mailbox(msg.receiver).put(msg)用 asyncio.Queue 而不是普通列表是因为异步队列天然支持并发。多个 Agent 可以并行的从自己的队列里取任务而总线本身只是一个轻量转发器不持有业务状态。这个设计让整个系统可以在不修改其他模块的前提下随时加一个新 Agent——只要给它在总线里挂一个邮箱。2.4 一条消息的完整旅程以请写一份销售简报为例我会把消息流转拆成八个明确步骤用户在主入口创建一条USER类型消息receiver 填assistant投递给总线。assistantAgent 从自己的邮箱队列里取出消息。Agent 把系统提示词、历史消息、工具 JSON Schema 组装成模型请求。模型返回两种可能要么直接给最终回答要么要求调用某个工具。若是工具调用请求Agent 解析参数到工具注册表里找到对应函数执行。工具执行结果封装为TOOL类型消息重新投递回 Agent 自己的邮箱。Agent 再次调模型把工具结果喂回去。模型认为任务完成生成最终文本Agent 把结果作为消息发给调用方。整套流程看起来平平无奇但它定义了一条极稳定的闭环。后面加多 Agent 协作、加人工介入、加超时处理都是在这个闭环的外面加触角而不是改闭环本身。3. 搭建最小框架目录、基类和工具注册3.1 目录布局项目很小所以我不搞复杂工程结构一个一眼能看完的目录足够hermes_agent/ ├── core/ │ ├── __init__.py │ ├── types.py # Message、MessageKind │ ├── bus.py # MessageBus │ ├── agent.py # Agent 基类 │ └── router.py # 消息分发调度器 ├── tools/ │ ├── __init__.py │ ├── registry.py # 工具注册表 │ └── builtin.py # 内置常用工具 ├── providers/ │ ├── __init__.py │ ├── base.py # Provider 抽象 │ └── openai_compatible.py ├── config.py └── cli.py # 命令行入口router.py实际上是整个调度的枢纽。它启动时把所有 Agent 的邮箱订阅起来然后在一个 asyncio 循环里等待消息根据msg.receiver找到对应 Agent 并触发其处理。任何时候想加日志、加指标、加断点都只在 router 这一层动手不需要侵入业务代码。3.2 工具注册用装饰器把函数签名变成 JSON SchemaAgent 要使用工具第一步是让模型知道有哪些函数可用这一步靠 OpenAI 兼容接口里的tools参数完成。但手写 JSON Schema 太痛苦所以我在 hermes-agent 里做了一个装饰器直接从函数签名生成 Schema# hermes_agent/tools/registry.py import inspect from functools import wraps _REGISTRY: dict[str, dict] {} def tool(name: str , description: str ): def decorator(func): nonlocal name name name or func.__name__ sig inspect.signature(func) properties {} required [] for pname, param in sig.parameters.items(): # 简化示例生产环境需要按类型做更完整映射 properties[pname] {type: string} if param.default is inspect.Parameter.empty: required.append(pname) schema { name: name, description: description, parameters: { type: object, properties: properties, required: required, }, } wraps(func) async def wrapper(*args, **kwargs): result func(*args, **kwargs) if inspect.isawaitable(result): result await result return result _REGISTRY[name] {func: wrapper, schema: schema} return wrapper return decorator为什么选用装饰器因为声明式最直观。工具的调用逻辑、schema 生成、副作用处理都收敛在函数定义处业务代码里不需要出现任何注册动作。真实环境里我会进一步把参数类型映射到number、boolean、array并通过typing.get_origin处理Optional、List这类复合类型但核心思想始终不变函数签名即描述源码即文档。3.3 Provider 抽象所有模型调用收敛成一个方法对接模型这一步我只保留一个接口chat(messages, tools)。不管底层是本地部署的开源模型服务还是商业 API框架层面只认这个统一签名# hermes_agent/providers/openai_compatible.py import httpx class OpenAICompatibleProvider: def __init__(self, base_url: str, api_key: str, model: str): self.base_url base_url.rstrip(/) self.api_key api_key self.model model async def chat(self, messages: list[dict], tools: list[dict] | None None) - dict: payload { model: self.model, messages: messages, tools: tools or [], } async with httpx.AsyncClient(timeout120) as client: resp await client.post( f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, jsonpayload, ) resp.raise_for_status() data resp.json() return data[choices][0][message]这里有两处细节值得展开。第一timeout120不是随手写的。工具调用场景里模型经常要先思考再生成尤其当工具结果特别长时第二次请求的生成时间会明显上升。默认超时往往只有 30 秒实际跑批时很容易因为超时白白丢任务。第二为什么只用 OpenAI 兼容协议因为几乎所有的开源模型服务和主流商业模型服务都支持这个协议统一它成本最低雇人写多个 provider 的适配属于过度设计。3.4 配置管理YAML 写结构密钥走环境变量配置这块我不搞花活一个 YAML 文件加环境变量占位就够了# config.yaml provider: base_url: http://localhost:11434/v1 api_key: ${API_KEY} model: hermes-3-llama3.1-8b agent: default_system_prompt: 你是一个只回答事实、不编造数据的助手。 tool_call_mode: auto max_turns: 8api_key从环境变量读取YAML 里只放占位符这个习惯不是为了本地模型没风险就不做。无论对接什么服务密钥不进代码仓库是底线。我的做法是在config.py里加载 YAML 后再用os.getenv去替换${API_KEY}这类占位符逻辑简单但能避免密钥被误提交。4. 跑通第一个任务报表整理能验证什么4.1 任务定义框架搭到能跑模型和工具第一个要验证的任务我选了一个特别务实的场景读取当前目录下的sales_log.csv按日期汇总销售额把结果写成report.md。这个任务同时考验三个能力工具参数能否正确解析、Agent 能否读懂工具返回的表格、多轮调用时上下文是否稳定。4.2 定义两个工具from hermes_agent.tools.registry import tool import csv tool(description读取CSV文件返回列名和前N行数据) def read_csv(path: str, max_rows: int 20): with open(path, newline, encodingutf-8) as f: reader csv.DictReader(f) rows list(reader)[:max_rows] return {columns: reader.fieldnames, rows: rows} tool(description把最终报告写入指定 Markdown 文件) def write_report(path: str, content: str): with open(path, w, encodingutf-8) as f: f.write(content) return {path: path, chars: len(content)}注意read_csv我故意控制了max_rows。这个设计是为了防止一次性读入超大文件导致上下文爆炸。真实项目里这个参数通常由 Agent 根据任务自主决定先读前几行感知结构再决定要不要用其他方式全量处理。4.3 System Prompt 的设计同样的工具不同的 System Prompt 效果差异很大。我第一版 Prompt 写得太空泛Agent 经常自作主张编数据后来改成明确约束版你是一个数据整理助手。 规则 1. 必须通过工具获取数据禁止编造数字。 2. 工具返回不足时明确说不知道不要猜测。 3. 最终报告用 Markdown 表格输出。 4. 每一步只调用一个工具得到结果后再决定下一步。第 4 条很关键。如果允许模型一次调多个工具返回的解析逻辑会复杂度翻倍而且中途失败时很难定位。让模型一步一步来虽然多几次请求但稳定性和可排查性会好很多。4.4 一次真实运行的日志跑起来之后我在日志里看到这样的流转过程[user ] tasktask-0001 请汇总 sales_log.csv 的销售额并生成 report.md [agent ] tasktask-0001 先读取 CSV 数据结构 [tool ] tasktask-0001 read_csv(pathsales_log.csv, max_rows20) [tool ] tasktask-0001 - columns[date,product,amount], rows[...] [agent ] tasktask-0001 需要读取全部数据我将使用全量读取路径 [tool ] tasktask-0001 read_csv(pathsales_log.csv, max_rows1000) [tool ] tasktask-0001 - total rows365 [agent ] tasktask-0001 开始写报告 [tool ] tasktask-0001 write_report(pathreport.md, content...) [agent ] tasktask-0001 任务完成报告已生成这套日志让我特别满意的地方在于每一步决策都清晰可见没有黑盒。我可以很明确地告诉别人Agent 在这一步决定读取全量数据也可以迅速定位是哪一次工具调用出了问题。4.5 这个闭环为什么是地基完成这个任务意味着整个框架的最小可行链路已经打通。之后要做的所有事——多 Agent 协作、任务队列、超时重试、人工审核——都是在这个闭环外面加挂件。如果这个最小闭环都跑不顺后面一切高级功能都不成立。所以我建议任何想自己写 Agent 框架的人先不要想复杂协作先把一句话到一次工具调用再到一句话这条链路调通再往上盖楼。5. 多 Agent 协作把任务像信使一样转交出去5.1 三种协作模式单 Agent 能解决的问题有边界。真实任务往往需要多个角色分工一个负责拆解任务一个负责查数据一个负责写报告。我在设计多 Agent 协作时梳理出三种基础模式Supervisor / Worker主管 Agent 负责拆分任务worker Agent 分别执行结果回收后由主管汇总。适合任务边界清晰的场景。Pipeline 流水线任务按固定顺序流转A 处理完交给 BB 再交给 C。适合流程固定的场景。Publish / Subscribe某事件触发后广播给所有订阅者各干各的。适合事件驱动的异步场景。hermes-agent 的 MessageBus 天然支持第三种模式但实际项目里用得最多的还是第一种。说到底多数业务问题都是先拆再干的问题。5.2 Handoff 消息不是转大 JSON而是转摘要多 Agent 转交任务时最容易犯的错是把原始对话全量塞过去。我之前这么干过结果 worker 每次都要重读一遍用户原始指令既费 token又容易抓错重点。后来我专门定义了一个 Handoff 结构# hermes_agent/core/handoff.py from dataclasses import dataclass dataclass class Handoff: task_id: str # 任务 ID from_agent: str # 来自哪个 Agent to_agent: str # 转交给谁 question: str # 当前这个 Agent 需要解决的问题 context_summary: str # 上游已得出的关键结论摘要 max_turns: int 8 # 限制内部迭代次数context_summary是这套设计里最重要的字段。主管 Agent 拆任务时不是把用户原话一字不差地传给 worker而是把已经确定的前提和期望 worker 解决的具体问题提炼出来。这样 worker 的输入短、聚焦、不易跑偏。5.3 Supervisor 的调度骨架主管 Agent 的职责拆成三步规划、分发、回收。核心逻辑可以抽象成这样# hermes_agent/core/agent.py 片段 class Supervisor(Agent): async def run(self, task: str): plan await self.planner.plan(task) # 1. 把任务拆成子步骤 results {} for step in plan.steps: msg Message( task_idstep.task_id, senderself.name, receiverstep.worker, kindMessageKind.USER, contentHandoff( task_idstep.task_id, from_agentself.name, to_agentstep.worker, questionstep.description, context_summaryplan.summary, max_turns8, ), ) await self.bus.send(msg) # 2. 发给对应 worker for _ in plan.steps: result await self.receive_results() # 3. 等待回收结果 results[result.task_id] result return self.synthesize(results)这里我用receive_results()统一收结果而不是为每个 worker 写单独回调。它的实现本质上是往总线里订阅senderworker_name、receiversupervisor的消息然后放进一个结果队列。只要 worker 完成时把最终结论作为消息发给主管主管就一定能收到顺序没关系反正按task_id对号入座。5.4 一次失败协作的复盘我第一次跑多 Agent 协作时任务很简单合并两个 CSV 并生成总结。主管把任务拆成了读取 A 文件读取 B 文件合并计算写报告每个 worker 都收到了几乎同样的用户原始指令。结果让人头疼读文件的两个 worker 都认为自己是汇总方各自生成了半份报告主管等结果时又超时了。复盘后发现根因不是并发问题而是任务描述不聚焦。worker 只需要知道你要读哪个文件、把列名和行数报告回来不需要知道整个任务的背景。所以我改成每个 worker 收到的question尽量一带一背景信息压成 2 到 3 行的context_summary。这一步改动之后协作完成率提升非常明显。多 Agent 不是堆模型而是把任务边界切清楚切得越清楚模型犯错的概率越低。6. 实测踩坑上下文膨胀、工具参数解析与协作死锁6.1 上下文膨胀消息越长模型越笨这是我最早踩到的坑。Agent 每次工具调用后我都会把结果追加进 memory跑几个任务后上下文中堆满了 CSV 表格、日志片段、中间输出。模型开始发呆要么重复总结要么把早先的噪音当成重要信息。原因不难理解模型注意力是有限的超长上下文里关键信息被稀释了。而且 token 费也水涨船高。我的解决思路是分层记忆。短期记忆只保留最近 20 条消息超过上限后触发折叠把最早的 20 条消息用模型压缩成一段摘要替代原文塞进历史里。这个折叠动作我在邮件里归档旧邮件时会做在 Agent 里也一样旧信息不是删除而是变成摘要。async def fold_memory(history: list[dict], provider, max_items20): if len(history) max_items: return history early history[:-max_items] summary_prompt 用3句话概括这些信息的核心保留数字和结论 summary await provider.chat( messages[{role: user, content: f{summary_prompt}\n\n{early}}], tools[], ) return [{role: system, content: f早期消息摘要{summary.content}}] history[-max_items:]这个方案也能防止 agent 在遇到隐藏的循环时无限膨胀。需要说明的是折叠的调用时机很重要我放在每次工具调用完成之后判断一次不放在每次消息到达时判断这样可以减少额外请求。6.2 工具参数解析模型给的 JSON 不一定合法工具调用另一个高频坑模型返回的参数内容经常不是严格 JSON。我见过最多的情况是参数里多了一个尾逗号、JSON 里夹杂了 markdown 代码块标记、或者所有字符串值被写成了单引号。如果直接json.loads立刻抛异常整个任务中断。后来我写了一个宽松的解析函数按顺序尝试四种解析方式import json import re import ast import yaml def parse_tool_args(raw: str) - dict: text raw.strip() # 1. 标准 JSON try: return json.loads(text) except Exception: pass # 2. 抽取最外层大括号后再解析 m re.search(r\{.*\}, text, re.S) if m: try: return json.loads(m.group()) except Exception: pass # 3. Python 字面量处理单引号 try: return ast.literal_eval(text) except Exception: pass # 4. YAML 的宽松解析 try: data yaml.safe_load(text) return data if isinstance(data, dict) else {} except Exception: raise ValueError(f无法解析工具参数: {raw[:200]})这套兜底逻辑不是万能药最有效的还是让模型别乱来。我会在 system prompt 里写死严格输出 JSON键名与提供的 schema 一致不要加任何解释也会在工具 schema 里把每个字段的类型、描述、是否必填写清楚。两个手段一起用解析失败率能降到非常低。6.3 协作死锁信使也会迷路多 Agent 场景里最隐蔽的坑是协作死锁。我第一次写 supervisor 调度时主管等 worker 结果worker 又在等另一个 worker 的产出两边都以为对方会主动发消息结果消息总线里安静了一分钟整个任务卡死。解决这个问题我用了两个手段第一个是给每条 Handoff 带max_turnsworker 内部做决策时每转一圈turn加一超过上限就强制返回我无法完成。这防止单一 Agent 内部出现无限循环。第二个是加了一个看门狗协程周期扫描总线里的未完成任务任何消息如果超过超时阈值仍然没有任何回应就向一个dead_letter邮箱投递一条超时消息async def watchdog(bus, timeout: float 60.0): while True: await asyncio.sleep(5) for msg in bus.snapshot_pending(): if msg.age() timeout and msg.metadata.get(handled) is False: await bus.send(Message( task_idmsg.task_id, senderwatchdog, receiversupervisor, kindMessageKind.SYSTEM, contentf任务 {msg.task_id} 超时来源 {msg.sender} 未在 {timeout}s 内完成, )) msg.metadata[handled] True加上看门狗之后协作任务的失控概率大幅下降。它不直接解决问题但至少能让你快速发现哪个环节死等了而不是等任务超时才翻日志。对我这种经常改框架的人来说快速感知异常比避免异常更重要。6.4 调试方法把消息流水变成事故现场还原工具最后分享一个调试习惯。hermes-agent 里我把所有横切逻辑收敛到MessageBus.send这一个入口这意味着日志、监控、断点都只需要在一个地方做。具体来说我加了一个简单的dump(task_id)命令把某个任务相关的所有消息按时间顺序打出来。排错的时候我完全不看模型内部在想什么只看消息流水用户说了什么Agent 调了哪个工具工具返回了什么下一步 Agent 又做了什么。大部分问题只需要看流水就能定位根本不需要断点调试。hermes-agent dump --task task-0001这比到处埋 print 高效得多也比 Python 调试器更适合异步场景。消息流水是 Agent 系统里最接近真相的东西只要它完整、有序、可回放你就永远有办法还原现场。如果你也要写类似的 Agent 骨架我最后的建议是先把 Message 和 Bus 这两个概念扣死再考虑 Agent 内部怎么实现。我在迭代中发现凡是让人难受的改动几乎都是因为消息格式不够稳定或者消息传递路径不够清晰。框架可以轻消息不能散模型可以换总线不能乱。把这个根基打稳后面加什么功能都不会心虚。