ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

轻量级智能体编排框架实战:基于消息协议与主循环状态机的Agent设计

2026/9/9 8:31:38 拓冰建站 浏览量
轻量级智能体编排框架实战:基于消息协议与主循环状态机的Agent设计 上个月我把攒了小半年的 hermes-agent 推到了 GitHub 上本来只是想整理一下自己的代码没想到陆陆续续有十几个朋友来问架构思路。趁着热乎劲我把项目里那些踩过的坑、想明白的设计、还有没来得及写进 README 的细节系统地整理成一篇长文。先说明白这是干什么的hermes-agent 是一个轻量级的智能体编排框架核心思路是用一套消息协议把大模型、工具函数和多个 Agent 串在一起让模型不只是生成文本而是能真正“收发消息、调度工具、交付结果”。给不想看长文的朋友一句话总结如果你也在做类似“让大模型调用工具完成多步任务”的事情并且受够了那些又重又黑盒的编排框架这篇文章可以帮你节省至少一周的试错时间。1. Hermes 这个名字背后的信使逻辑1.1 Agent 的本质工作是“端到端的消息转译”项目起名 Hermes 不是拍脑袋。希腊神话里的 Hermes 是信使之神负责在神与神、神与人之间传递消息。而 Agent 做的事情本质上也是一样的接收用户的请求把它翻译成模型能理解的指令把模型产生的决策翻译成工具能执行的调用再把工具返回的结果翻译回模型需要的上下文最后把结论翻译给用户。这个“翻译”链条里最容易被忽视的就是消息本身的格式和流转路径。很多人写 Agent 的第一版代码都是直接在主函数里写死几个 if else用户说查天气就调天气接口用户说算数学就调计算器。这种写法在小 demo 里跑得通一旦场景变成“先查天气再规划路线再生成行程单”代码就开始失控了。我在 hermes-agent 里做的第一个设计决定就是把所有 Agent 之间的交互都抽象成消息。无论是一条用户指令、一次工具调用、还是另一个 Agent 发来的协作请求在系统内部都统一成结构化的消息对象。这样做的好处很明显每一个环节都可以被记录、被重放、被测试而不是散落在函数调用栈里的一堆局部变量。1.2 我为什么不直接用现成的编排框架这里要先坦白我并不是一开始就想造轮子。最早我也老老实实试过 LangChain 和 AutoGen 那一套但实际用下来有几个问题特别难受一是版本变动太频繁。一个 Chain 的写法三个月前还是这么调的三个月后就变成了完全另一套 API。我花在追文档上的时间比写业务逻辑的时间还多。二是调试链路太长。框架帮你封装了太多东西prompt 是怎么拼的、工具结果是怎么塞回上下文的、模型的一次异常输出是在哪一层被吞掉的全靠翻源码。出了问题我根本不知道从哪看起。三是消息流转不透明。多 Agent 协作时框架内部的消息队列虽然在跑但你想把某一步的消息打印出来看得先弄明白它的内部数据结构。所以我决定自己写一个极简版本。不追求大而全的生态只要求三点链路透明、消息可观测、扩展成本低。hermes-agent 整个核心代码大概只有两千行但每一个环节我都能说清楚它是怎么工作的。1.3 Hermes-Agent 适合谁、不适合谁我把它定位成“轻量信使”而不是“重量级平台”所以它的适用边界非常明确。适合的场景包括需要让大模型稳定调用本地或远程工具的中小型项目、需要把任务拆给多个专职模型协作完成的流程、以及对消息内容有审计和回放需求的对内系统。不适合的场景也很清楚如果你要做的是一个大流量的在线推理服务或者需要一个自带模型微调、评测、部署全链路的企业级平台那请直接去用商业产品和成熟框架。Hermes 的目标是让中小团队和个人开发者能把 Agent 用起来、用明白而不是替代那些庞大体系。2. 核心架构消息总线与主循环是怎么咬合的2.1 三层消息模型Request / Task / Resulthermes-agent 的消息体系分三层这是我反复调整后才定下来的。Request 是外部入口层也就是用户或上游系统发来的请求。它只关心“用户想要什么”不关心内部实现。比如“帮我查一下明天北京天气并建议穿什么”这就是一个 Request。Task 是内部任务层是 Agent 真正在循环里处理的对象。一个 Request 会被拆解成一个或多个 Task。Task 里带着目标描述、当前状态、需要的工具列表以及给模型的指令上下文。Result 是结果反馈层是工具函数或子 Agent 执行完毕后的输出。Result 会被封装成标准结构包含状态、数据、错误信息然后作为新消息喂回给模型的上下文。这三层看起来简单但它解决了一个非常关键的工程问题内部消息和外部请求解耦。外部接口可以永远保持稳定内部哪怕把单 Agent 改成多 Agent把同步改成异步上层调用方也感知不到变化。这里我举个实际例子。早期版本里我直接把 HTTP 请求的 body 塞给模型当 prompt工具返回值也原样拼在 prompt 后面。结果模型偶尔会模仿 HTTP 报文的格式输出而不是正常说话。后来改成三层消息模型内部统一做序列化和转义这类“格式污染”的问题基本消失了。2.2 Agent 主循环一个朴素但足够用的状态机每个 Agent 内部都有一个主循环我把它实现为一个显式的状态机。状态包括 IDLE、PLANNING、EXECUTING、OBSERVING、FINISHED、ERROR。from enum import Enum class AgentState(str, Enum): IDLE IDLE PLANNING PLANNING EXECUTING EXECUTING OBSERVING OBSERVING FINISHED FINISHED ERROR ERROR主循环的逻辑很简单Agent 从消息总线里收到一个 Task进入 PLANNING 状态调用模型生成行动计划如果计划里有工具调用就进入 EXECUTING 状态执行工具工具结果回来后进入 OBSERVING 状态把结果转成消息喂回给模型模型判断任务已完成则进入 FINISHED 状态产出最终回复任何环节出现不可恢复的错误进入 ERROR 状态。这个状态机没有用什么复杂的编排引擎就是一层 while 循环加状态切换配合最大迭代次数限制。但它的价值在于每一轮循环都会产生一条带状态标签的事件记录。我在回顾一次对话时可以清晰看到这条链条PLANNING - EXECUTING(调用 get_weather) - OBSERVING(拿到 26 度) - FINISHED。哪一步出了问题一目了然。如果你觉得自己在写的 Agent 逻辑经常乱成一团我强烈建议先做这件事把主流程画成状态机然后老老实实写 while 循环。别急着上框架先让状态流转在代码里可见。2.3 工具注册协议让模型“看见”可调用的能力工具定义是 Agent 项目里绕不开的一环。hermes-agent 的做法是提供一个装饰器配合类型标注自动生成 JSON Schema注册后自然进入该 Agent 的工具列表。from hermes_agent import HermesAgent, tool agent HermesAgent(modelgpt-4o-mini, nameassistant) tool(description根据城市名查询实时天气返回温度与风力) def get_weather(city: str) - dict: # 这里通常是真实的 API 调用 return {city: city, temperature: 26, wind: 3}模型看到的其实是一段结构化的工具描述包含工具名称、参数名、参数类型、必填项和说明。关键点在于工具说明写得越细模型调用的成功率越高。比如参数名从 city 改成 city_name描述里补一句“中国的城市请带‘市’后缀例如北京市”这类看似不起眼的细节能让参数幻觉率明显下降。2.4 轻量消息总线的选型思考多个 Agent 之间要通信最简单的方式是直接函数调用Agent A 做完调用 Agent B 的函数。但这种写法会导致耦合Agent A 必须知道 B 的存在还必须在 B 挂掉时处理异常。Hermes 选择引入一个轻量消息总线基于 asyncio.Queue 实现。每个 Agent 有独立的 inbox 和 outbox通过总线地址寻址。一个 Agent 产出的消息只需要投递到总线上由总线根据消息头里的接收方字段分发出去。我故意没有用 Redis Stream 或者 Kafka 这类重量级消息中间件因为初期版本更重要的是降低心智负担。如果你只是在自己的服务器上跑几个 Agent 协作一个 asyncio 队列足够。等量大了再把总线实现替换成 Redis Stream 版本接口不变。3. 快速上手跑通第一个带工具调用的 Agent3.1 安装与最小配置安装没什么好说的pip install hermes-agent装完之后创建一个配置文件 config.yamlmodel: provider: openai name: gpt-4o-mini temperature: 0.2 tools: - service.weather.get_weather agent: max_iterations: 6 request_timeout: 30 memory: max_messages: 20 summarizer: true这里几个配置的作用先说清楚。temperature 我建议默认就设 0.2 左右Agent 执行链路里宁可让它“无趣”也不要让它“发挥”。max_iterations 是主循环最大轮数防止模型反复横跳把 token 烧完。max_messages 控制喂给模型的上下文条目上限超出后触发摘要压缩。启动代码写在 main.pyfrom hermes_agent import HermesAgent from service.weather import get_weather agent HermesAgent.from_yaml(config.yaml) agent.register(get_weather) result agent.run(明天杭州天气怎么样我应该穿短袖还是长袖) print(result.final_answer)跑起来的体感是你会先看到一条结构化日志显示模型决定调用 get_weather参数是 city杭州市然后日志显示工具返回了温度最后模型基于工具结果生成穿衣建议。整个链路 3 秒内完成每一行日志都能对上号。3.2 第一个完整 Demo天气查询与穿衣建议我把这个 demo 完整展开一下因为它几乎涵盖了 Agent 项目的标准范式意图解析、工具调用、结果回填、最终生成。当用户说出那句“明天杭州天气怎么样”时Agent 内部发生的动作拆开看是这样的Request 进入总线被分发给 assistant 这个 Agent。Agent 主循环启动把 Request 拼进系统 prompt调用模型。模型返回一个工具调用指令结构化输出里包含 tool_name 和 parameters。Agent 校验工具存在、参数 schema 合法执行 get_weather(city杭州市)。工具返回 {temperature: 26, wind: 3}Agent 把结果转成 Result 消息喂回上下文。模型看到天气信息后生成最终穿衣建议。主循环检测到最终回复状态置为 FINISHED返回给用户。这个过程看着不难但每一步都可能出错。参数里城市名多了个“明天”或者模型把明天理解成昨天都会导致结果偏差。所以我在工具描述里会特意注一句“入参 city 请解析为具体城市名不要包含日期词汇”。这是纯工程经验的积累。3.3 配置项里最容易忽略的几个字段顺着配置文件多说几个容易被忽略但很要命的字段。request_timeout 是给工具执行设置的超时。很多人不设结果某个第三方 API 卡住整个 Agent 跟着卡死。我建议所有工具调用都强制走超时20 秒足够判断一个工具是否无响应。summarizer 这个字段容易被默认值坑到。开启后当上下文消息数超过 max_messages 时系统会调用模型对历史消息做摘要。问题在于如果这个功能被触发得太频繁摘要过程本身也会烧掉不少 token而且摘要会丢掉关键细节。我的做法是能关就关能用“只保留最近 N 条 重要字段”的规则截断就别让模型做摘要。只有在长对话强交互场景下我才开 summarizer。还有一个很多人不关心的字段max_parallel_tools。它控制单轮里最多允许几个工具并行执行。默认是 1也就是串行调用。如果你有多个互不依赖的工具可以调到 3 或 5省一轮推理时间。但要注意这个值越大模型生成的工具调用列表就越可能出错谨慎调整。4. 多智能体协作如何把大任务拆给多个 Specialist4.1 为什么多 Agent 之间要“信使”而不是直接函数调用把任务拆给多个 Agent 协作是自然的需求。我最早写的版本是让 orchestrator 直接调用 specialist 的类方法。很快发现两个问题一是 orchestrator 里塞满了对不同 Agent 内部 API 的调用改一个 Agent 的接口所有调用方的代码都要跟着改二是没有统一的消息记录调度关系只能在代码里推演。后来我把所有 Agent 之间的调用都改成通过总线发消息。orchestrator 发出一个 Task 消息指定接收方和期望返回类型然后异步等待结果。这样 orchestrator 根本不关心接收方是这个文件里的 Agent还是另一个微服务里的 Agent。只要消息格式不变实现随便换。这个模式还有个隐藏好处你可以很容易地给消息加版本号和后处理逻辑。比如对 specialist 返回的 Result 做一次规范性检查不合格就退回让它重做。这类“质检节点”在函数调用直连的模式里很难优雅地插进去。4.2 一个“编排者 执行者”的协作实例我项目里内置了一个示例一个 orchestrator 分派任务一个 researcher 负责检索资料一个 writer 负责写段落。完整逻辑大概是这样from hermes_agent import HermesAgent from hermes_agent.message import TaskMessage orchestrator HermesAgent(modelgpt-4o, nameorchestrator) researcher HermesAgent(modelgpt-4o-mini, nameresearcher) writer HermesAgent(modelgpt-4o-mini, namewriter) orchestrator.send_message( TaskMessage( recipientresearcher, task收集 2024 年大模型推理成本下降的数据案例, require_fields[fact_list] ) ) research_result orchestrator.receive_result(researcher) orchestrator.send_message( TaskMessage( recipientwriter, task基于以下事实写一段 200 字的行业分析, contextresearch_result.data ) ) final_text orchestrator.receive_result(writer).data这里的关键设计是 require_fields。orchestrator 要求 researcher 返回的事实列表必须是一个数组字段 fact_list。如果 researcher 没有按要求返回orchestrator 可以直接判定消息不合格要求重做而不是把一堆乱七八糟的文本塞给 writer。有了这个字段约束Agent 之间的消息质量就有了契约感。4.3 协作场景下的死锁与超时设计多 Agent 协作最让人头疼的是死锁和超时。一个 Agent 等着另一个 Agent 的消息另一个 Agent 又在等模型接口返回模型接口偏偏卡了 50 秒整个流程就干瞪眼。我踩过的坑是在同步等待结果时用了简单的 queue.get()没有设置超时。结果某个 specialist 因为 API 限流慢了几秒orchestrator 那里直接等成了僵尸任务。后来我规定所有跨 Agent 的消息等待必须有超时时间方案是给 TaskMessage 加一个 deadline 字段并且在 receive_result 里做超时处理result orchestrator.receive_result(researcher, timeout45) if result is None: orchestrator.reasoning(researcher 超时我直接基于已有信息完成工作。)这个设计不止解决了死等的问题还给了模型一个兜底的机会——它可以在超时后用已有信息继续生成而不是整体失败。对于演示和中等可靠性要求的内部系统这个兜底已经够用。5. 模型不可控是常态四个真实踩坑与修复链路5.1 工具参数沦为“一坨 JSON 字符串”之后第一个让我印象深刻的坑出在工具参数解析上。早期版本我偷懒让模型直接输出一段 JSON 文本作为工具参数然后我用正则把 JSON 抽出来再解析。demo 里能跑通但真实场景一多就炸了。典型故障是模型输出把参数写成{“city”: “杭州”}用了中文引号或者干脆写成自然语言“城市是杭州市”又或者参数里多了换行和中文注释。正则抽 JSON 在这种输入下就像是走钢丝。修复方案是三重校验第一从模型接口侧要求严格的结构化输出能走 function calling API 就走走不了就用 response_format 约束第二解析后做 Pydantic schema 校验字段名、类型、必填项一项一项对第三校验失败后不直接报错而是把错误信息拼进上下文要求模型重新生成一次参数。from pydantic import BaseModel class WeatherParams(BaseModel): city: str unit: str celsius try: params WeatherParams.parse_raw(raw_params) except ValidationError as e: correction_prompt f参数格式错误请按 schema 重新生成: {e}这套流程跑起来之后工具调用成功率从大概 85% 提升到了 97% 以上。剩下的 3% 基本是模型确实理解错了用户意图这是语义问题不是格式问题了。5.2 上下文窗口被中间结果塞爆第二个坑是上下文爆炸。Agent 每调用一次工具工具返回结果就会塞进上下文。如果任务是“遍历 50 个城市查天气并汇总”每轮都往上下文里加一条城市天气数据不到十轮窗口就满了。我的处理策略分两步。第一步给 Agent 的记忆模块增加一个白名单机制让工具返回的 Result 进入上下文之前先经过一个精简步骤。比如天气工具返回的 dict 很大就只保留降序排列后的 top 3 字段其余丢弃。第二步对已经完成并确认无用的中间结果允许 Agent 模型把它标记为 memory_cleanup从上下文里移除。这里的核心思想是不是所有信息都值得放在上下文里。中间结果的价值在决策那一刻最高一旦决策完成它就是噪音。5.3 模型在失败工具上无限重试第三个坑最让人崩溃。一个工具因为网络问题临时不可用模型第一次调用失败后主循环把错误信息喂回去模型居然不死心地换个参数继续调用同一个工具连续重试了我没有设上限最后把 API 配额烧掉不少。修复方案分两层。第一层是硬限制在工具注册表里为每个工具维护一个“连续失败计数”超过阈值就把该工具临时置为不可用并告知模型“请放弃使用该工具”。第二层是给模型一个更优雅的备选路径当它收到某个工具连续失败的信息时应该在上下文中明确“我切换策略”而不是原样重试。我在主循环里加了对重复调用的检测如果同一工具在同一轮任务里被调用超过三次且三次参数完全相同直接跳过执行输出提示“检测到重复无效调用已终止该工具后续调用”。这几行判断代码救了不少 token。5.4 模型幻觉调用不存在的工具第四个坑是模型幻觉导致调用了一个根本没有注册的工具。正常来说模型只会看到 tools 列表里已有的工具但用一些开源模型或精简 prompt 时模型会自己编工具名。我处理这个问题的方式是在工具注册表里加一个 fallback 处理器如果解析出的 tool_name 不存在系统不会直接报错而是生成一条系统消息“工具 x 不存在可选工具为……请重新选择。”然后把这条消息塞回上下文让模型修正。实测大多数情况下一轮修正就能回到正轨。这几个坑串起来看核心结论就是别指望模型一次做对要让系统有“察觉错误、纠正错误、避免重复错误”的能力。这个能力不是靠一个魔法参数实现的是靠一层一层校验逻辑堆出来的。6. 从 Demo 到能上生产可靠性工程化的三个抓手6.1 重试、指数退避与熔断不能只靠装饰器很多教程会教你给工具调用加一个 retry 装饰器fail 了就重试三次。但实际生产中重试策略要复杂得多。我在 hermes-agent 里内置了一套重试策略配置场景策略说明工具接口瞬时超时重试 2 次指数退避 1s / 2s应对网络抖动模型 API 限流(429)重试 1 次退避 3s给服务端恢复时间工具逻辑型错误不重试代码 bug 重试没用单工具连续失败超过 5 次熔断 60 秒防止拖垮依赖服务很多人忽略的是重试的目标不只是“让这次成功”还包括“别把下游服务压垮”。所以指数退避的底数和上限很重要我通常用base * 2^retry_count但封顶 8 秒重试三次后就直接走熔断。6.2 结构化输出校验层消息进入总线之前先做“安检”Agent 之间传的消息如果格式不规范问题往往藏得很深。比如 writer 回复说“写好了”但正文内容是一段 Markdown 代码块下一环直接把它塞进表格排版就乱了。我在总线入口处加了一个 schema 校验层每条消息发出前必须通过注册过的合约校验。合约就是一份 JSON Schema规定消息必须包含哪些字段、字段类型是什么、要不要非空。校验不过的消息会被弹回发送方附带校验错误说明。这个设计有点像是给每个 Agent 之间立了合同。合同越清晰协作越稳定。你甚至可以在校验层里加业务规则比如“结果为 success 的 Result 必须携带 data 字段”这样上游写漏字段时不是等到下游炸了才发现而是在消息落地之前就被拦截。6.3 观测性request_id 串起完整调用链Agent 项目最怕的是什么是用户问“为什么这次回答是错的”而你连这次回答经历了哪些步骤都不知道。我在 hermes-agent 里给每个 Request 生成一个 request_id这个 ID 会跟随整条链路里的所有消息所有日志都会带上它。工具调用的入参和返回值、模型每轮的输入输出 token 数、状态机的每次状态切换、消息总线的每次投递全部记录在结构化日志里。排查问题时我只需要拿着 request_id 过滤一次日志就能看到完整时间线。哪个工具慢了、哪一轮模型回复跑偏了、哪条消息被校验层拦了清清楚楚。这个能力建议你在任何 Agent 项目里都优先做它比写一百行注释都管用。6.4 测试策略用 Mock 模型把主循环状态机打满Agent 项目测试难难在模型输出不可控。如果每次测试都真实调用模型既慢又贵结果还不稳定。我的做法是给 HermesAgent 传入一个 MockModel它不访问任何模型 API而是根据预设脚本返回固定输出。比如我要测“工具调用失败后模型是否切换策略”的流程就让 MockModel 第一次返回调用 get_weather 的指令第二次返回“工具失败我改用 fallback 策略”的文本。这样主循环的每个状态分支都能在 CI 里被稳定覆盖不花真实的 token。配合状态机事件日志我甚至在测试里直接断言某次运行出现了特定的状态序列比如[PLANNING, EXECUTING, OBSERVING, FINISHED]。这套做法让 Agent 的工程化程度提升了一个档次。7. 我下一步会优先补的三块能力项目到今天这个阶段基本能满足我自己的日常需求了但它还有很多明显短板。按照我目前实际使用时的痛点下一步会优先补三块。第一是让消息总线支持持久化。现在基于 asyncio.Queue 的实现重启即丢消息。虽然对纯内存调试没问题但一旦要接外部系统就必须把消息换到一个可持久化的中间件上。我正在设计一个 plugin 接口分别实现内存版和 Redis Stream 版目前接口已经留好只差搬砖。第二是把评测集做起来。我越来越觉得Agent 项目的质量不是靠调 prompt 调出来的而是靠一套固定的评测任务集跑出来的。我现在手头收集了大概 30 个真实场景包括工具调用、多轮对话、多 Agent 协作、异常恢复这几类下一步想把这些固化成一个命令行工具每次改完代码先跑一遍评测集用分数说话。第三是并发控制。当前每个 Agent 是单线程消费消息如果同一个 Agent 同时收到 10 个请求只能排队处理。我打算为这个 Agent 增加一个 concurrency 参数让它可以并发消费不同 Request 的消息同时用 per-request 的状态隔离来避免上下文串扰。这个项目从第一行代码到现在最大的收获不是框架本身而是让我彻底搞明白了 Agent 的消息流转机制。如果你也想自己做类似的框架我的建议很简单先从能看清每一步状态流转的小框架开始写别一上来就接一堆依赖。等你把消息、状态、工具这三件事理顺了后面加什么功能都顺手。