轻量级AI Agent运行时设计：消息循环与工具调用实战-平芜编程栈

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(f"unexpected 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(name="search_files", description="在指定目录下递归搜索文件名中包含关键词的文件") def search_files(keyword: str, root_dir: str = ".") -> str: # 实际搜索逻辑 ... @agent.tool(name="get_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-agent

3.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_url=config["model"]["endpoint"], api_key=config["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, name=None, description=""): """装饰器：注册工具并自动生成Schema""" def decorator(func): schema = self.registry.register(func, name=name, description=description) 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( model=self.config["model"]["model_name"], messages=self.messages, tools=self.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_none=True)) 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_none=True)这一步。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, name=None, 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 f"Error: unknown tool {name}" func = self._tools[name]["func"] try: result = func(**args) if inspect.isawaitable(result): result = await result return json.dumps(result, ensure_ascii=False) except Exception: return f"Error: {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(name="list_directory", description="列出指定目录下的所有文件与子目录名") def list_directory(path: str = ".") -> list: entries = os.listdir(path) return sorted(entries) @agent.tool(name="read_file", description="读取文本文件内容，返回前max_chars个字符") def read_file(path: str, max_chars: int = 1000) -> str: with open(path, "r", encoding="utf-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项目时，把日志做成结构化数据，而不是简单的打印，这个投入产出比极高。