资讯动态

轻量级AI Agent运行时设计:消息循环与工具调用实战

发布时间:2026/9/10 7:42:39 来源:尧图企业网站定制
1. 项目定位与整体设计思路1.1 从“Hermes”这个名字聊起Agent的本质是替人跑腿“hermes-agent”这名字起得有点意思。Hermes是希腊神话里的信使职责是在众神之间传递消息、搬运指令。如果你把现代AI Agent拆开看真正干活的角色其实也就是这个——模型本身不直接操作外部世界它负责理解任务、生成意图然后通过工具去查数据、调接口、改文件最后把结果再带回来。所以“hermes-agent”这个命名从一开始就点出了项目的核心它不是在做模型而是在做模型与外部世界之间的信使通道。我自己对这类项目的定义是一个以“消息流转”为骨架的AI代理运行环境。它接收用户的自然语言输入把输入转成结构化的工具调用执行完再回填结果循环往复直到任务完成。市面上类似方向的项目很多但大多数都把自己做成了“全家桶”——对话管理、记忆、编排、插件市场、云端部署全塞进去。hermes-agent的思路不太一样它更倾向于做一个轻量的本地优先运行时让开发者自己掌控工具、上下文和权限边界。这个项目最适合谁我觉得有三类人值得关注一是刚接触Agent开发、想从零理解工具调用全流程的进阶学习者二是已经在用LangChain这类框架、但觉得黑盒太多、想自己掌控消息循环的技术控三是有明确后处理诉求的人——比如你在写一个本地知识库查询助手或者想给团队内部做一个基于大模型的运维工单处理机器人其实你需要的不是一个大而全的Agent平台而是一层薄薄的、可插拔的“信使”运行时。1.2 精简内核的取舍逻辑我在设计hermes-agent的时候最先定下来的不是功能清单而是三个原则。第一消息流转必须透明。模型生成的每条工具调用、每次结果回填都要有完整的日志链路出问题能直接复盘而不是抛出一个难懂的堆栈。第二工具扩展必须极简。一个新工具只需要写一个普通函数加上装饰器和类型注解剩下的参数校验、超时控制、失败重试都由运行时接管。第三运行形态必须本地优先。所有数据默认留在本机模型调用走本地或自选的云端接口不强制依赖任何专有平台。很多项目一上来就堆功能最后反而没人能跑通。你要是让一个新人在没有GPT-4 Key、没有向量数据库、没有Docker环境的情况下部署一套“完整Agent平台”他大概率会卡在环境问题上一整天。hermes-agent的最小运行依赖只有三样一个Python 3.10环境一个兼容OpenAI接口的模型端点以及一个能执行Python函数的宿主进程。其它全是选配。设计目标可以归纳成一张表设计维度选择运行形态本地进程优先可选API调用工具协议装饰器注册函数JSON Schema描述参数消息存储内存队列 可选持久化日志上下文策略显式管理按token预算裁剪权限控制工具级白名单 执行前确认钩子走轻量路线有一个直接的好处问题定位变得容易。工具调用循环卡住了直接看消息日志就知道是哪一步没回来上下文超限了看token计数就知道该裁剪哪一段。一个能让开发者“看穿”的Agent运行时才是生产可用的运行时而不是一个通上电能聊天、一上生产就歇菜的玩具。2. 核心细节解析与关键机制2.1 消息循环Agent的中枢神经如果你问一个Agent项目最核心的架构是什么我给的答案不是模型、不是工具而是消息循环。Agent的运行本质上是一个不断“推理-行动-观察”的循环过程模型收到用户请求在当前上下文中推理决定调用某个工具运行时拿到这个调用请求执行实际函数执行结果以消息形式追加回上下文模型看到结果后继续推理要么再调工具要么给出最终回答。在hermes-agent里我简化成了这样一个主循环:async def run_agent(task: str): messages [{role: user, content: task}] for _ in range(core.max_steps): response await llm.chat(messages) choice response.choices[0] if choice.finish_reason tool_calls: messages.append(choice.message) for tool_call in choice.message.tool_calls: result await execute_tool(tool_call) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) elif choice.finish_reason stop: return choice.message.content else: raise RuntimeError(funexpected finish_reason: {choice.finish_reason}) raise RuntimeError(max_steps exceeded)这个循环最关键的一点在于每一步都要把模型的原始输出、工具执行结果完整追加进对话历史而不是只把最终的答案传下去。原因是模型的推理是自上而下依赖上下文的如果中间某一步的结果被省略或改写模型后续的推理就可能“失忆”。这一点我在早期实现里踩过坑后续会专门说明。另一个值得注意的细节是消息角色必须严格区分。用户消息是user模型推理消息是assistant工具结果消息是tool并且tool消息必须携带对应的tool_call_id。很多兼容OpenAI协议的本地模型在这一点上比较脆弱一旦角色混杂后续推理质量就会明显下降。所以我在运行时里做了强制校验任何消息在被追加进上下文之前都要检查role的合法性非法消息直接拒绝并记入日志。2.2 工具注册协议把一切变成函数调用消息循环是骨架工具注册协议就是肌肉。hermes-agent的工具协议遵循一个简单原则任何可被调用的函数只要描述清楚自己的名字、用途和参数就能成为Agent的工具。这个描述格式我直接复用了OpenAI Function Calling的JSON Schema规范因为这是目前兼容性最好的方案之一几乎所有主流模型接口都支持。一个标准的工具描述长这样:{ type: function, function: { name: search_files, description: 在指定目录下递归搜索文件名中包含关键词的文件, parameters: { type: object, properties: { keyword: {type: string, description: 搜索关键词必填}, root_dir: {type: string, description: 起始目录默认当前工作目录} }, required: [keyword] } } }有了这个描述Agent在做函数调用时就有了依据模型读到用户的自然语言请求会尝试把它映射到某个工具的参数上。比如用户说“帮我找到home目录下所有包含日志的文件”模型就会输出一个类似tool_call的结构其中包含函数名search_files和参数{keyword: 日志, root_dir: /home}。运行时拿到这个结构后再从注册表找到实际函数执行。在代码层面用户注册工具时不需要手动写JSON Schema。我在hermes-agent里实现了一个基于类型注解的自动推导器agent.tool(namesearch_files, description在指定目录下递归搜索文件名中包含关键词的文件) def search_files(keyword: str, root_dir: str .) - str: # 实际搜索逻辑 ... agent.tool(nameget_weather, description获取指定城市的实时天气) def get_weather(city: str) - str: ...装饰器会读取函数的类型注解、默认值、docstring自动生成Schema。复杂类型比如列表、嵌套对象也能支持。这里要特别说明工具描述写得好不好直接决定了模型能不能正确触发调用。描述应该写“这个工具能干什么、适合什么场景”而不是堆砌技术细节。比如get_weather的描述如果写成“获取天气”就太模糊模型可能在用户问“明天需不需要带伞”的时候不触发这个工具但如果描述成“获取指定城市的实时天气用于回答出行建议、穿衣搭配、是否带伞等问题”触发率会显著提升。2.3 上下文管理与编排保护Agent跑崩的原因里上下文超限可能是最常见的一个。模型上下文窗口是有限的每轮循环的消息都会累积一旦超出上限请求就会直接报错或者在更糟的情况下——模型开始“忘事”回答牛头不对马嘴。hermes-agent的解决思路是显式管理而不是让用户被动承受。我在运行时里设置了上下文预算机制。你在配置文件里声明context_budget_tokens比如8000运行时会在每轮循环开始前统计当前消息列表的总token数如果接近预算就触发裁剪策略。裁剪策略有三种丢弃最早的用户消息、压缩工具返回的长结果、丢弃中间轮次的tool记录。这里需要注意工具调用是强依赖历史的不能随意删否则后续消息里的tool_call_id就找不到对应记录接口会报错。def trim_messages(messages, budget_tokens): total_tokens sum(estimate_tokens(m) for m in messages) while total_tokens budget_tokens: removed messages.pop(0) total_tokens - estimate_tokens(removed)这段逻辑虽然简单但直接决定了Agent在长对话中的稳定性。我建议把预算设置在模型窗口上限的60%到70%之间留出足够余量给模型的输出和工具调用结果。另外max_steps参数也要设上限我默认设为10防止Agent在循环里跑出死循环——比如某个工具一直返回“数据仍在生成中”模型就一直反复调用如果没有步数限制费用和耗时都会失控。3. 实操从零跑通一个hermes-agent3.1 环境准备与项目骨架纸上谈兵没意思下面直接落地。我先建一个干净的Python虚拟环境然后安装依赖。这里我选了FastAPI做API层、openai作为模型调用SDK因为这两个库足够通用换成其它兼容接口也方便。mkdir hermes-agent cd hermes-agent python3 -m venv .venv source .venv/bin/activate pip install fastapi uvicorn openai pydantic项目目录结构我习惯这样组织hermes-agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # 消息循环与调度器 │ ├── tools.py # 工具注册与执行 │ ├── context.py # 上下文管理 │ └── config.py # 配置加载 ├── tools/ │ ├── __init__.py │ ├── files.py # 文件检索工具 │ └── web.py # 网页摘要工具 ├── config.yaml # 运行配置 └── main.py # 入口脚本config.yaml里我放了最关键的几个参数:model: endpoint: http://localhost:11434/v1 model_name: qwen2.5:7b api_key: sk-local context_budget_tokens: 8000 max_steps: 10 request_timeout_seconds: 60这里我特意用了一个本地模型端点兼容OpenAI协议因为这样整个流程可以不依赖任何云端API跑起来没门槛。你要是想用别的模型服务只要把endpoint和model_name换掉即可协议是兼容的。我第一次搭这个项目的时候为了测试快速迭代也是先用本地小模型把整个链路跑通确认无问题后再切换到更大的模型。3.2 核心调度器实现接下来是骨干代码。我在core.py里实现了两件事一个是Agent类的初始化与工具注册另一个是前面说过的消息循环。这里我给出一个完整可运行的精简版你拿到手改改模型地址就能跑。# agent/core.py import asyncio import json from openai import AsyncOpenAI from .tools import ToolRegistry class Agent: def __init__(self, config): self.config config self.client AsyncOpenAI( base_urlconfig[model][endpoint], api_keyconfig[model][api_key], ) self.registry ToolRegistry() self.tool_defs [] self.messages [] self.max_steps config[max_steps] self.timeout config[request_timeout_seconds] def tool(self, nameNone, description): 装饰器注册工具并自动生成Schema def decorator(func): schema self.registry.register(func, namename, descriptiondescription) self.tool_defs.append({ type: function, function: { name: schema[name], description: schema[description], parameters: schema[parameters], }, }) return func return decorator async def run(self, task: str): self.messages [{role: user, content: task}] for step in range(self.max_steps): response await self.client.chat.completions.create( modelself.config[model][model_name], messagesself.messages, toolsself.tool_defs if self.tool_defs else None, ) choice response.choices[0] if choice.finish_reason tool_calls: self.messages.append(choice.message.model_dump(exclude_noneTrue)) for tc in choice.message.tool_calls: tool_name tc.function.name args json.loads(tc.function.arguments) print(f[step {step}] call {tool_name}({args})) output await self.registry.call(tool_name, args) self.messages.append({ role: tool, tool_call_id: tc.id, content: output, }) elif choice.finish_reason stop: return choice.message.content return Reached max_steps, early stop.这个精简版的主循环基本够用但你直接跑到生产环境会发现几个痛点一是没有超时保护模型接口挂了就会一直挂二是没有重试机制偶发网络错误直接中断三是没有日志复盘时无从下手。所以我在真实项目里给这些薄弱点都加了处理后面单独说。这里有一个重要的工程细节choice.message.model_dump(exclude_noneTrue)这一步。OpenAI的SDK返回的message对象里包含tool_calls字段如果直接拿对象本身追加后续再次发送请求时可能会出现序列化问题。先用model_dump转成字典再存进消息列表能保证后续请求体是干净的JSON。3.3 工具注册与执行引擎工具注册表在tools.py里实现。它要做的事很简单把装饰器传入的函数解析成Schema并存入字典执行时根据名字找到函数并传入参数。但实际写起来有几个容易忽略的细节我在注释里标出来。# agent/tools.py import inspect import json import traceback class ToolRegistry: def __init__(self): self._tools {} def register(self, func, nameNone, description): tool_name name or func.__name__ schema self._build_schema(func, tool_name, description) self._tools[tool_name] {func: func, schema: schema} return schema def _build_schema(self, func, name, description): sig inspect.signature(func) properties {} required [] for param_name, param in sig.parameters.items(): param_type self._map_type(param.annotation) properties[param_name] { type: param_type, description: f参数 {param_name}, } if param.default is inspect.Parameter.empty: required.append(param_name) else: default param.default properties[param_name][default] default return { name: name, description: description, parameters: { type: object, properties: properties, required: required, }, } def _map_type(self, annotation): if annotation in (str,): return string if annotation in (int, float): return number if annotation is bool: return boolean return string # 兜底 async def call(self, name, args): if name not in self._tools: return fError: unknown tool {name} func self._tools[name][func] try: result func(**args) if inspect.isawaitable(result): result await result return json.dumps(result, ensure_asciiFalse) except Exception: return fError: {traceback.format_exc()}实际执行函数时我统一把返回结果json.dumps成字符串再塞回消息列表。不能直接返回Python对象因为后续要作为消息内容发送给模型必须是字符串。这一步看着不起眼但不少Agent项目翻车就在这里——返回了非字符串对象SDK序列化报错或者说返回了一个超大的对象直接把上下文撑爆。所以我还建议在call方法里对结果长度做一次截断比如超过2000字符就只保留前2000字符并追加提示“结果过长已截断”。3.4 跑起来本地文件检索Agent实战为了验证整套机制我写了一个简单但完整的实战场景让Agent帮我们在本地项目目录里搜索文件、读取文件内容然后给出总结。这需要两个工具search_files按关键词搜文件名和read_file读取文件内容并返回指定行范围。# main.py import os from agent.core import Agent from agent.config import load_config def main(): config load_config(config.yaml) agent Agent(config) agent.tool(namelist_directory, description列出指定目录下的所有文件与子目录名) def list_directory(path: str .) - list: entries os.listdir(path) return sorted(entries) agent.tool(nameread_file, description读取文本文件内容返回前max_chars个字符) def read_file(path: str, max_chars: int 1000) - str: with open(path, r, encodingutf-8) as f: content f.read() return content[:max_chars] task 先看看当前目录里有什么文件然后读取main.py的内容告诉我这个文件大概做了什么。 result asyncio.run(agent.run(task)) print(result) if __name__ __main__: main()实测下来的效果是这样的模型先触发list_directory拿到文件列表然后看到main.py后触发read_file读取内容最后结合读取到的代码输出总结。整个过程在消息日志里一目了然[step 0] call list_directory({path: .}) [step 1] call read_file({path: main.py, max_chars: 1000})这个简单的链路背后其实已经把Agent的核心机制串起来了模型根据工具描述自主决策、运行时执行函数、结果回填、模型综合输出。你要学会观察的是模型在什么情况下会误判工具参数为什么某一步没有触发工具而是直接回答了这些调试经验只能靠多试、多看日志总结出来。4. 常见问题与排查技巧实录4.1 工具调用循环卡死Agent停不下来有一段时间我遇到一个很典型的症状Agent会不停调用同一个工具每次都返回同样的错误信息然后再次调用。比如搜索工具返回空结果模型不理解“空”意味着查无此物反而以为搜索参数不对换个角度再搜一次如此往复直到触达max_steps。排查思路分三步。第一步看工具返回内容。空结果或者报错信息必须明确是“正常空”还是“异常”。我自己规定所有工具返回都带一个固定的前缀比如“OK:”表示正常“ERROR:”表示异常这样模型能快速区分。第二步看工具描述是否足够清晰。如果描述里没写“当搜索结果为空时请如实告诉用户没有找到”模型就可能产生错误推理。第三步看max_steps设置。如果你确实想限制烧钱和耗时把它保持在10以内是我的建议。这个问题的根因不在消息循环本身而在工具的“可理解性”。你写的工具是给人和模型一起用的描述和返回格式必须尽量排除歧义。不少开发者在写工具时报错信息很随意比如返回一个Traceback模型根本读不懂自然没法做下一步决策。4.2 模型返回的JSON参数总解析失败工具调用的参数是模型生成的JSON字符串理论上没问题但现实是本地小模型经常返回带注释、带尾逗号、甚至带多余换行的JSON。json.loads直接用就会抛异常。我在生产里加入了容错解析函数import json import re def parse_args(raw: str): text raw.strip() # 去掉可能的注释 text re.sub(r//.*, , text) # 去掉尾逗号 text re.sub(r,\s*([}\]]), r\1, text) try: return json.loads(text) except json.JSONDecodeError: # 如果解析失败尝试提取第一对大括号内的内容 start, end text.find({), text.rfind(}) if start ! -1 and end start: return json.loads(text[start:end 1]) raise这个函数不是万能的但实测下来能把奇奇怪怪的JSON容错率提高不少。更重要的一招是从提示词层面给模型“打预防针”在系统提示词里加上一句“输出工具参数时请严格使用合法JSON不要包含注释或尾逗号”。模型对这句提示非常敏感能有效减少畸形输出。4.3 上下文膨胀与token超限长对话场景下最让人头疼的问题就是上下文爆掉。尤其当你让Agent做多轮工具调用每一轮都要把工具执行结果塞回消息列表体积增长很快。比如一个读文件工具返回了5000字符的代码再来几轮上下文预算瞬间就满了。我推荐三个策略组合使用。第一工具返回结果设置硬性上限。read_file这类工具默认只返回前1000到2000字符如果要读取长文可以分页读取或者用模型做摘要后只返回摘要。第二历史消息裁剪按预算走。前文已经给出trim_messages的实现裁剪时优先丢弃早期的低价值消息保留靠近当前轮次的关键上下文。第三工具执行完成后把中间轮次的tool记录合并成一条summary消息。这个策略实现稍复杂但对长任务的稳定性提升非常显著。为了看清楚到底消耗了多少token我在每个Agent实例里加了一个token计数器在每次请求前后分别统计一次消息总数并打印每一轮的增量。这个日志习惯帮我省了不少排查时间。4.4 安全边界与权限控制最后说一个容易被人忽略但生产环境绕不开的问题工具执行的权限边界。Agent能调用工具意味着它能在你的机器上执行实际功能如果工具写得随意模型一旦被诱导调用危险操作后果会很严重。比如前面示例里的read_file如果不对路径做限制模型可以读取任意文件这就比较危险。我的建议是至少做到三层收敛第一层工具设计层面的白名单。函数内部限定可访问的根目录对传入路径做规范化校验阻止“..”等路径穿越。第二层执行前确认钩子。对于写操作、删除操作、网络请求这类有副作用的工具在调用前打印提示并请求用户确认。第三层沙箱执行可选。把整个Agent进程放进容器或虚拟环境即使工具出问题影响范围也被限制住。def safe_path(root: str, user_path: str) - str: full os.path.abspath(os.path.join(root, user_path)) if not full.startswith(os.path.abspath(root)): raise PermissionError(path escapes root directory) return full工具设计得多“聪明”不重要重要的是“可控”。稳定运行了几个月之后我的体会是Agent项目最怕的不是模型不够聪明而是工具在异常情况下做了模型没预料到的事。多做一层校验后续夜里被叫醒的概率就小一分。5. 进阶扩展与个人体会5.1 多Agent协作与子任务分发如果单Agent跑通了下一步很自然的想法就是让多个Agent协作。在hermes-agent的架构里做多Agent本质上就是把每个Agent也注册成一个“工具”。主Agent调度时可以“调用”另一个Agent实例把子任务描述作为参数传给它等它返回结果后继续主流程。这个模式非常适合“规划-执行”分离的场景主Agent负责拆解任务子Agent专注执行某个具体环节。这一层我实际用下来的感受是副作用的隔离非常关键。每个子Agent应该有自己独立的上下文和工具列表不能直接共享主Agent的内存状态。否则两个Agent互相污染上下文调试起来就是一个灾难。另外要给子Agent单独设置max_steps和超时防止个别子任务卡死拖垮整个编排。5.2 可观测性与会话回放真正让一个Agent项目走向生产拼的不是跑通Demo而是出问题后能不能快速定位。我实现了一个简单的会话回放系统每轮对话的结构化日志用户输入、工具调用、返回结果、token消耗都写入JSONL文件出问题时用脚本重放就能精确看到模型在每一步做了什么决策。这个回放能力帮我解决了不少客户现场的问题。举个例子用户反馈“Agent回答错了”通过重放发现是工具返回了一个格式错误的字符串模型基于这个错误结果给出了看似合理但实际错误的回答。没有回放机制这种问题根本无从查起。所以我建议你在写Agent项目时把日志做成结构化数据而不是简单的打印这个投入产出比极高。5.3 踩坑心得与项目维护建议维护了hermes-agent一段时间后我印象最深的一条经验是工具数量不是越多越好而是要控制在一个模型能“理解”的范围内。每增加一个工具模型的决策空间就大一圈误调用的概率也随之上升。我一般建议单Agent的工具数量控制在5到10个之间超出就考虑拆分子Agent或者合并工具。最后再分享一个小技巧新工具上线先别直接让Agent全量调用先把新工具的注册打开给模型发几条强制触发该工具的任务观察返回质量和参数正确率。等确认稳定后再投入日常使用。这个习惯能帮你过滤掉大量“工具能用但模型根本不知道怎么触发”的尴尬情况。hermes-agent这个名字恰好说明了Agent项目的本质——不是把AI包装成一个万能机器人而是搭建一条可靠的消息通道让模型在合适的时机把指令传到合适的地方。把这条通道管好了Agent才能真正成为你手里得力的信使。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价