☰
从零搭建可复用的AI智能体编排框架:Hermes-Agent实战笔记
2026/10/30 15:36:30 网站建设 项目流程

Hermes-Agent 实战笔记:从零搭建一个可复用的 AI 智能体编排框架

这几年做大模型应用落地,最让我头疼的从来不是模型本身,而是“怎么把模型接进真实的业务流”。LLM 单独调用只能做单轮对话,一旦涉及多步骤任务、工具调用、状态维护,代码就开始失控。后来我花了一个周末把一个内部项目重构成了名为 hermes-agent 的智能体编排框架,思路简单——用一套轻量调度内核把大模型、工具函数、记忆模块串起来,让 Agent 自己拆解任务、按步骤执行、遇到问题再调整。这篇文章就是这套框架的完整复盘,包括设计思路、核心源码结构、避坑记录和参数调优经验,适合正在做 AI 应用集成、自动化流程编排、或者单纯想理解 Agent 内部机制的开发者阅读。

1. 整体设计思路:为什么我不直接写业务代码,而是先做框架

1.1 从“硬编码调用”到“声明式编排”的转变

在 hermes-agent 之前,我处理多步任务的方式基本是“写死流程”。举个例子,做一个“查天气然后提醒带伞”的接口,代码就是先调天气 API,拿到结果拼 prompt,再调 LLM 生成提醒文案,每一步都写在一个大函数里。问题很明显:任务一多,if-else 满天飞,模型换一个就要重写逻辑,还很难复用。

hermes-agent 的核心转变,是把任务的“执行逻辑”和“业务代码”分离。我不再写“先干什么再干什么”,而是给 Agent 一套工具列表和一个目标,让它自己决定调用顺序。这个思路借鉴了 ReAct 范式的做法:模型在每轮输出中先思考(Thought),然后决定动作(Action),观察工具返回结果(Observation),再进入下一轮思考。hermes-agent 把这个循环做成了框架级的调度器,业务方只需要注册工具、设置系统提示词,剩下的循环交给框架。

1.2 框架的核心模块划分

我在设计时把整个系统分成了五个核心模块,每一个都对应一个明确职责:

  • 调度内核(Core Scheduler):维护 Agent 的循环状态,控制“思考-行动-观察”的迭代次数和退出条件
  • 工具注册中心(Tool Registry):管理所有可被 Agent 调用的函数或 API,提供统一的参数校验和调用接口
  • 记忆管理器(Memory Manager):负责短期上下文和长期记忆的读写,避免上下文爆炸
  • 模型适配层(Model Adapter):屏蔽不同 LLM 的 API 差异,让框架可以切换 GPT、Claude 或本地模型
  • 执行日志(Execution Logger):记录每一轮思考、工具参数和返回结果,方便排查问题

这样的分层带来一个很直接的好处——任何一个模块都能单独替换。比如我的项目前期用 OpenAI 接口,后期切到国产模型,只需要写一个新的 Model Adapter,其他代码完全不动。

1.3 为什么用 Python 而不是 TypeScript

选型的时候我其实纠结过。团队里有同事更熟悉 Node.js 生态,但最终我还是选了 Python,原因有三:

一是 AI 生态的 Python 库最全,像 Pydantic 做参数校验、tenacity 做重试、pytest 做测试都非常顺手,这些库能直接嵌入框架;二是项目需要跑一些数据分析和向量检索的代码,Python 这边生态无缝衔接;三是部署方便——我的目标环境是内网服务器和边缘设备,Python 的单文件分发比 Node 的 node_modules 目录友好得多。

当然 Python 也有缺点,最明显的是性能不如编译型语言。但 Agent 的瓶颈几乎都在网络 I/O(调模型 API、调工具接口),本地 CPU 计算占比很小,Python 完全够用。

2. 核心细节解析与实操要点:调度循环、工具协议和记忆管理

2.1 调度循环的内核实现

调度循环是 hermes-agent 的心脏。我最初写的一个雏形非常简陋,就是用 while 循环包一层 LLM 调用,后来在不断调试中逐渐完善。最终版的循环伪代码大致如下:

def run_agent(task: str, max_steps: int = 10): state = initialize_state(task) while not state.is_finished() and state.step_count < max_steps: response = model_adapter.generate( messages=state.get_messages(), tools=tool_registry.get_schemas() ) action = parse_action(response) if action.is_final_answer(): state.finish(action.answer) else: observation = tool_registry.execute(action) state.add_observation(action, observation) return state.get_final_answer()

看似简单,但有几个细节很容易踩坑。第一,max_steps不设上限的话,遇到模型陷入死循环,单次请求的成本会被无限放大。我实测过一个任务最多循环 20 轮,费用十几块钱,而正常任务三四轮就结束了。第二,parse_action必须稳,模型偶尔会输出不规范的 JSON 或穿插多余文字,我后来用一个专门的解析函数做容错,提取 action 名称和参数,解析失败就默认重试一次。第三,状态对象里要保存完整的消息列表,每轮把工具返回追加进去,而不是只存最后一轮,否则模型会丢失前文信息。

2.2 工具注册协议:如何让 Agent 正确调用任意函数

Agent 能不能用好工具,关键看工具定义是否清晰。hermes-agent 的工具注册中心用 Pydantic 定义了一套轻量接口,每个工具就是一个普通的 Python 函数,配上一个 JSON Schema 描述:

from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str = Field(description="城市名称,例如 北京 或 Shanghai") days: int = Field(default=1, description="预报天数,默认1天") def get_weather(city: str, days: int = 1) -> str: # 真实场景在这里调用第三方天气接口 return f"{city}未来{days}天天气:晴,22°C" tool_registry.register( name="get_weather", description="查询指定城市的天气预报,适合回答出行、穿衣等问题", input_schema=WeatherInput, func=get_weather )

这里最容易被忽略的是 description 的写法。我做了一版对比实验,description 太短(“查天气”)时,模型经常在该用天气工具时选了其他工具,准确率只有 60% 多;把 description 改得更具体(“查询指定城市的天气预报,适合回答出行、穿衣等问题”)后,准确率直接提升到 90% 以上。原因是模型在规划阶段依赖工具描述做意图匹配,描述越贴近真实使用场景,选择越准。另外,参数名要保持语义明确,少用a、b这种缩写,否则模型填参数时容易给出错误的映射关系。

注册中心还要处理一个问题——工具的并发与超时。我在 execute 方法里给每个工具调用加上了超时控制和并发限制,避免某个慢工具阻塞整个 Agent 循环:

import asyncio async def execute(self, action: dict) -> str: tool = self.registry.get(action["name"]) if tool is None: return f"错误:未找到工具 {action['name']}" try: timeout = tool.timeout or 10.0 result = await asyncio.wait_for(tool.func(**action["args"]), timeout=timeout) return json.dumps(result, ensure_ascii=False) except asyncio.TimeoutError: return "错误:工具调用超时" except Exception as e: return f"错误:{type(e).__name__}: {str(e)}"

注意返回结果统一转成字符串,这一步很关键。模型接口要求的输入是文本,工具返回的字典、列表如果不序列化,下一轮生成就会报错。

2.3 记忆管理器:避免上下文无限膨胀

刚开始跑 hermes-agent 时,我发现处理长任务时上下文很快就超限了。一次常规的多步骤任务,每轮工具返回几百字,十几轮下来光历史消息就有上万 token。模型接口的上下文窗口有限,超了就报错,不超费用也高。

我的解决办法是把记忆分层。短期记忆是当前任务的消息列表,超过一定条数就做裁剪;长期记忆用向量数据库存关键信息,需要时按相似度召回。hermes-agent 的记忆管理器提供了两个接口——remember()和recall():

class MemoryManager: def __init__(self, max_short_term_length: int = 20): self.short_term = [] self.long_term = MemoryVectorStore() self.max_short_term_length = max_short_term_length def remember(self, message: dict): self.short_term.append(message) if len(self.short_term) > self.max_short_term_length: # 把最早的几条消息压缩后存到长期记忆 summary = self.summarize(self.short_term[:5]) self.long_term.add(summary) self.short_term = self.short_term[5:] def recall(self, query: str, top_k: int = 3) -> str: return self.long_term.search(query, top_k=top_k)

剪裁策略不是简单丢消息,而是把最早的历史做一条摘要再存长期记忆。比如用户最开始说“帮我规划一个三天的杭州行程”,执行了 5 轮工具调用,这 5 轮的完整内容会被压缩成一句——“用户要求杭州三日游规划,已查询景点A/B/C和天气信息”。这样既节省了 token,又不会完全丢失任务背景。

长期记忆的向量化我用的是text-embedding模型,平时召回还挺准。如果预算有限,也可以用 BM25 做关键词检索,效果略差但零成本,看场景选就行。

3. 实操过程与核心环节实现:从安装到多智能体协作

3.1 环境准备与最小安装

hermes-agent 的安装依赖不多,核心就三样:Python 3.10 以上、openai 的 API 包、以及 Pydantic。我在项目里用uv做依赖管理,速度比 pip 快不少:

uv init hermes-demo && cd hermes-demo uv add openai pydantic python-dotenv

接着定义环境变量,框架会自动读取OPENAI_API_KEY和OPENAI_BASE_URL。这里我特别提一下BASE_URL的配置——很多本地部署或中转服务会用到自定义地址,我把模型适配层做成了兼容 OpenAI 协议,所以只要填对 base_url,就能无缝接各种推理服务。这个设计省了我后面很多事。

3.2 写一个完整可运行的 Agent

为了让读者更直观地理解,这里给一个完整的案例——让 hermes-agent 帮用户查资料并整理成表格。整个 Agent 配置分为三步:定义工具、写系统提示词、调起调度循环。

第一步,定义两个工具:一个“搜索网页摘要”,一个“生成 Markdown 表格”。工具函数本身不需要任何框架依赖,就是普通函数配 Schema:

from hermes_agent import Tool, AgentConfig search_tool = Tool( name="web_search", description="搜索引擎网页摘要查询,输入关键词返回标题和链接列表,适合查最新资料", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"} }, "required": ["query"] } ) table_tool = Tool( name="make_table", description="把原始数据渲染成 Markdown 表格,输入表头和行数据", parameters={ "type": "object", "properties": { "headers": {"type": "array", "items": {"type": "string"}}, "rows": {"type": "array", "items": {"type": "array"}} } } )

第二步,构造 Agent 配置,关键是系统提示词里明确“可以分多步搜索,每次搜索后判断信息是否足够,够了就生成表格”:

system_prompt = """ 你是一个信息整理助手。当用户要求整理某主题资料时: 1. 先拆解主题,判断需要哪几个子话题 2. 对每个子话题调用 web_search 搜索 3. 收集足够的搜索结论后,用 make_table 生成对比表格 4. 最后用文字总结重点 注意:一次只能调用一个工具,等结果返回后再决定下一步。 """

这里“一次只能调用一个工具”是我实验出来的稳定策略。虽然很多模型支持并行工具调用,但诊断问题时并行会让日志非常难读,而且某些工具之间存在依赖关系,串行执行逻辑更清晰。速度慢一点,但可解释性和稳定性好很多。

第三步,启动 Agent 并传入任务:

from hermes_agent import HermesAgent agent = HermesAgent( config=AgentConfig(system_prompt=system_prompt), tools=[search_tool, table_tool] ) result = agent.run("帮我比较华为、小米、苹果三家的旗舰手机参数") print(result)

运行之后,日志会清晰显示每一步:模型先 Thought 说“需要分别搜索三个品牌”,然后 Action 调用 web_search 搜索“华为 旗舰 参数”,观察到搜索结果后,再搜索小米,最后调用 make_table 生成表格。整个流程不需要我在代码里写任何 if-else,完全由模型自主编排。

实测下来,这类信息整理任务的成功率大概在 85% 左右,失败的场景通常集中在搜索工具返回内容过长、模型截断,或者参数解析出错。

3.3 参数调节与成本控制实操

Agent 类应用和普通 API 调用最大的不同在于,一次任务可能要调很多次模型接口。我统计过,一个 5 步任务大概要消费 3000~8000 token 的输入和 1000~2000 token 的输出。控制成本的核心参数有三个:

  • max_steps:任务最大迭代轮数,限制死循环和无效搜索
  • temperature:生成温度,Agent 任务建议 0~0.3,温度太高模型容易输出不稳定的 JSON
  • max_tokens:单次输出的最大 token 数,不要设太大,工具调用场景单次输出 500~1000 token 基本够用

我实际跑过一组对比,同样一个“调研竞品”任务,temperature=0 时工具调用格式错误率约为 1%,temperature=0.7 时错误率升到 8%,而且多花了两轮重试。原因很好解释——温度越高,模型越“发散”,在需要精确输出 JSON 的场景里就是灾难。所以如果你跑 hermes-agent 发现经常解析失败,先检查是不是温度设太高了。

4. 常见问题与排查技巧实录:五个高频故障的定位思路

4.1 模型陷入无限“思考-调用-失败”循环

这是 Agent 框架最经典的问题。现象是日志里出现连续七八轮“模型调用工具→参数错误→模型看错误信息再调用→再报错”,循环始终不退出。要么是模型不会用工具,一直在尝试某个错误参数;要么是工具返回的错误信息太含糊,模型不知道该怎么改。

我的处理是三层防护:第一,在调度内核里加最大重试次数,单工具连续失败 3 次就直接终止,把错误返回给上层;第二,工具报错信息要具体,比如“缺少必填参数 city”就比“调用失败”有用得多;第三,系统提示词里加一句“如果工具连续两次报同样的错误,换一种工具或直接给出已有信息的总结”。

这里最实用的是第一层。它保证即使模型完全“疯了”,任务也会按时终止,最多损失几次工具调用的费用,不会死循环耗掉整个额度。

4.2 工具参数类型不匹配

另一个高频问题是模型生成 JSON 参数时类型不对。比如 Schema 要求{"city": "北京"}里的 days 是整数,模型有时会输出一个字符串"1"。这在严格模式下就会校验失败,但很多模型接口不会主动做类型转换。

我在工具执行入口加了一个宽松转换函数:

def coerce_args(func, args: dict) -> dict: sig = inspect.signature(func) for name, param in sig.parameters.items(): if name not in args: continue if param.annotation is int and not isinstance(args[name], int): try: args[name] = int(args[name]) except (TypeError, ValueError): pass elif param.annotation is bool and not isinstance(args[name], bool): args[name] = str(args[name]).lower() in ("true", "1", "yes") return args

用了这个函数后,类型错误导致的失败率下降了至少一半。当然,过度宽松也有风险——如果参数差异太大(比如把对象转成字符串),工具内部可能拿到不可用的值。所以我只在整数和布尔两个类型上做兼容,其他类型严格校验。

4.3 上下文被工具返回内容撑爆

长任务最容易踩的坑是上下文长度超限。这里给一个我验证有效的方案:在工具返回时做一个“截断与摘要”。

MAX_OBSERVATION_LENGTH = 2000 def truncate_observation(raw_result: str) -> str: if len(raw_result) <= MAX_OBSERVATION_LENGTH: return raw_result # 保留开头和结尾,中间用摘要代替 head = raw_result[:800] tail = raw_result[-800:] middle_summary = f"...中间{len(raw_result) - 1600}字被省略..." return head + middle_summary + tail

保留开头和结尾是因为模型通常关注最前面的标题信息和最后面的结论,中间的大段正文很少用得到。绝大多数 Agent 框架都会做类似处理,但自己做一遍之后,对上下文窗口的理解会深很多。

4.4 模型选择了错误的工具意图

有时候工具本身没问题、参数也正确,但模型就是选了不该选的工具。比如用户问“苹果有什么新品”,模型去调天气工具,因为两个工具描述里都有“苹果”“天气”这些关键词。

排查这个问题的诀窍是看日志里的 Planning 阶段——模型在思考时写了什么。我习惯打开 verbose 日志,把每个 Thought 都打印出来。大多数时候可以发现,模型是因为术语歧义或描述不够具体而选错工具。解决方案就是前面说的:把工具 description 改得更具体,并且加一些“什么时候用、什么时候不用”的说明。

4.5 慢工具拖垮整体响应时间

最后一个高频问题不是错误,而是性能——某个工具 API 响应非常慢(比如第三方接口要 8~9 秒),导致整个 Agent 任务要一分钟才出结果。解决思路有两层:给单次工具加超时(比如最长 6 秒),超时后返回一个兜底错误信息;另一个是我后来加的“缓存层”,对同一个参数的工具调用结果做内存缓存,避免重复请求。

加缓存的效果非常明显,尤其是搜索类工具。用户连续问两个相似的查询,缓存命中率能到 40% 左右,平均每个任务的耗时降低 30%。

5. 踩坑记录与优化心得:这些经验文档里不会写

5.1 稳定输出:模型对 JSON 的“花式破坏”

我最初设计工具调用协议时,天真地认为只要在 prompt 里写清楚“输出 JSON 格式”,模型就会乖乖输出 JSON。现实是——有些模型会输出 JSON Markdown 代码块(包含 ```json 标记)、会在 JSON 前后加解释性文字、甚至把单引号当双引号用。这类问题如果不做容错,Agent 循环会频繁中断。

我写了一个extract_json函数,核心逻辑是:

  1. 先尝试直接json.loads()
  2. 如果失败,用正则提取第一个{...}或[...]片段
  3. 如果再失败,把单引号替换成双引号,再json.loads()
  4. 还失败的话返回特殊标记,让调度层决定重试或终止

这个函数的代码只有二十几行,但把工具调用的整体成功率从 70% 拉到了 95% 以上,是投入产出比最高的一段代码。

5.2 可观测性:日志里必须有“因果链”

Agent 应用和传统程序很大的区别在于,执行路径是不确定的。同一个任务换一个模型版本,可能走完全不同的调用链。所以我的框架里把执行日志设计成结构化的流水账,每一行都包含“第几步、模型思考、调用的工具、传入参数、返回结果摘要”五个字段。

排障时我基本不看错误代码,而是直接看“因果链”。比如某次任务结果是“数据不全”,翻日志发现模型在第二步就只搜索了一个关键词,后面的搜索请求都被 web_search 工具返回了重复结果。为什么?因为搜索工具对类似关键词做了去重。知道这个因果链,解决方案就很简单——去掉搜索去重,或者让模型在 prompt 里明确要新信息。

5.3 多智能体协作:把一个大任务拆给多个 Agent

hermes-agent 做多了之后,我发现很多任务其实可以拆成子任务并行跑。经典场景是“调研三个城市”的需求——一个 Agent 串行查三个城市需要九轮工具调用,改成三个 Agent 各查一个城市,再让汇总 Agent 合并结果,时间能省一半以上。

实现这个功能不需要改动框架核心,我是在 Agent 层之上加了一个 Orchestrator:

from concurrent.futures import ThreadPoolExecutor def run_parallel(agent_configs: list, task: str): with ThreadPoolExecutor(max_workers=3) as pool: futures = [pool.submit(config.run, subtask) for config, subtask in zip(agent_configs, split_task(task))] results = [f.result() for f in futures] return merge_results(results)

并行虽然快,但要注意两点。一是成本更高——三个子任务并行,峰值并发会把 API 配额打满,需要限流;二是汇总 Agent 的提示词要想清楚,否则合并出来的内容可能自相矛盾。我在项目里调试过一版合并提示词,要求“以时间线方式整合多方结论,冲突时标注来源”,效果比“说说你的看法”好得多。

5.4 模型选型与成本权衡

最后聊一下模型选型。hermes-agent 的模型适配层让我可以随心切换底层模型,我也是通过大量对比测试才发现不同模型在 Agent 任务上的表现差异巨大。强模型(比如 GPT-4 级别的)几乎不需要额外调 prompt 就能稳定用工具;弱模型(比如一些小参数量的开源模型)要额外加很多“护栏”——比如在系统提示词里给出一两个“好的调用示例”、工具 schema 写得更详细、输出格式要求更死板。

成本上,我算过一笔账:弱模型在 API 价格上可能便宜 10 倍,但一次任务的调用次数会多 2~3 倍(因为更容易出错重试),实际上省不了太多。如果你的 Agent 是做高频、低价值的任务,用弱模型就行;但如果是关键业务链路,老老实实上强模型,稳定性带来的价值远超模型费用。

6. 安全与合规实践:Agent 执行链路上容易忽略的两个问题

6.1 工具权限边界:Agent 不能被用来做危险操作

Agent 一旦有了工具调用能力,安全性就变得至关重要。我在设计 hermes-agent 时专门加了一层权限控制——每个工具可以声明是否需要人工确认。比如删除文件、发邮件这类敏感操作,默认要求人工审批;查资料、算数字这类只读操作则放行。

这样做的原因是,模型理解不了业务上下文中的“红线”。即使系统提示词写一千遍“不要删除重要文件”,模型在复杂任务里也可能误解用户的意图。人工确认虽然会打断自动流程,但安全永远优先于效率。

6.2 敏感信息过滤:防止工具返回内容泄露

另一个易忽略的问题是输出安全。Agent 的工具往往能访问数据库、内部文档甚至用户个人信息,模型生成的总结可能无意间把这些敏感信息透出。我在框架里加了输出过滤层,对最终答案做一次敏感信息检测(手机号、身份证、企业内部关键词),命中就替换成[已过滤]字段。

这个功能不复杂,但被我很多朋友忽略了。直到有次测试,Agent 在回答“公司网络怎么样”时,直接把内网服务器 IP 和账号名输出出来了,大家才重视这个问题。做 Agent 应用的人,务必把输出过滤当成标配功能。

7. 后续规划:hermes-agent 还能往哪些方向扩展

目前 hermes-agent 在我这边的版本已经稳定跑了几个月,支持了三个内部应用。后续我计划做几件事:

  • 把记忆管理器升级成可插拔的存储后端,支持 Redis 和 SQLite,方便在多实例部署时共享记忆
  • 增加更完善的评测集,让每次模型升级或提示词修改都能自动回归测试,而不是凭感觉判断好坏
  • 做一个可视化调试面板,实时查看 Agent 的执行路径和 token 消耗,方便非技术人员使用

如果你也在做类似的 Agent 编排工具,我的建议是从最小可运行版本开始,先把“单 Agent + 三五个工具”跑通,再考虑多智能体协作、记忆向量化这些高级功能。框架本身不复杂,复杂的是如何让模型稳定、可控、可预测地完成真实任务——这需要大量的实际业务打磨,光看文档是学不来的。

根据我个人这段时间的实践体会,最值得投入精力的地方是日志设计和错误信息规范化。把这两件事做好的,Agent 的调试成本能降一半以上。再一个小技巧:给每个工具加一个“使用示例”字段,模型调用准确率会有质的提升。希望这些经验对你有帮助,也欢迎在实际使用中多踩坑、多分享。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询