轻量级AI Agent框架实践：从零搭建大模型工具调用与任务规划系统-开发者社区

Hermes 这名字，懂点希腊神话的朋友应该不陌生，就是那位脚上长翅膀、整天忙着传信的使者。把 agent 接在它后面，想表达的意思很直接：我想做一个负责“传递意图、调度工具”的智能体，让大模型不只会聊天，还能真正上手干点实事。hermes-agent 这个项目，本质上就是一个轻量级的 AI Agent 框架，解决的是“怎么让大模型稳定地调用外部工具、完成多步任务”这个核心问题。

过去一年我试过不少 Agent 框架，LangChain 用起来总觉得抽象层太厚，AutoGPT 自动化过头了很难控，最后还是决定自己撸一个最小可用的实现。这个项目不是要跟那些大框架掰手腕，而是想给同样好奇 Agent 内部原理、或者需要在业务系统里快速接工具的开发者一个可复现的参考。读完这篇，你能照着搭出一个能跑通“用户提问-拆解任务-调用工具-返回结果”完整链路的 Agent，还能学会怎么处理上下文溢出、工具调用失败这些实战里躲不掉的坑。

1. 这个项目解决什么问题：源起与核心设计思路

1.1 为什么叫 Hermes：信使隐喻与定位

Hermes 在神话里的职责是穿梭于神界和人界之间传递信息，特点是快、准、不迷路。Agent 系统里的“信使”也是个很贴切的比喻——用户把请求交给 Agent，Agent 要做的不是自己去算，而是判断该找谁帮忙、把请求翻译成对方听得懂的格式、拿到结果再带回来。

拿我自己常遇到的一个场景举例：运营同事发来一句话“帮我查下这批订单的支付金额，再算一下总利润”。如果只靠大模型，它既读不到数据库，也算不了精确的浮点；如果全写死在代码里，又没法消化人类这种模糊表达。hermes-agent 做的工作就是把这句话拆成两个子任务——先查数，再计算，中间靠“信使”把数据从查询工具传到计算工具。这个拆解和传递的过程，就是 Agent 最核心的价值。

所以这个项目的定位很明确：不是聊天机器人，不是训练框架，而是一个执行中枢。它把“大模型怎么想”和“工具怎么做”这层关系理清楚，让两边各自专注自己擅长的部分。

1.2 整体架构：路由、执行、记忆三层分离

我在设计时把整个 Agent 拆成三层，各管各的，互不掺和：

路由层（Planner）负责理解用户意图，决定下一步调用哪个工具。这一层只输出决定，不负责具体实现。就好比快递分拣中心，只看包裹面单，决定送哪条线，不碰货物本身。

执行层（Executor）负责真正调用工具、处理错误、拿结果。工具返回成功还是失败、结果合不合预期，都在这一层兜底。

记忆层（Memory）负责保存对话历史、用户偏好、中间观察结果。多轮对话能不能听懂“它”指的是谁，全靠这一层。

这个分层看起来简单，但很多 Agent 项目恰恰栽在这里——把路由和执行逻辑搅在一起，一个函数里既让模型生成决策又直接调工具，调试起来满头包。分开之后，每层都能单独测试，出问题定位也快。

1.3 与其他主流 Agent 框架的取舍对比

既然市面上已经有 LangChain、AutoGPT、CrewAI 这些成熟方案，为什么还要自己写？我用过一段时间，感受挺深的，列个表格看得更清楚：

框架	优势	我实际遇到的痛点
LangChain	生态全，工具链多	抽象层太厚，一个简单的 agent 要理解 Chain、Runnable、Tool 一大堆概念，排查问题时栈很深
AutoGPT	自主规划能力强	经常在子任务里绕圈，token 消耗大，出错后自动恢复能力不稳
CrewAI	多角色协作清晰	针对团队协作场景，单 Agent 场景反而显得重
hermes-agent	代码量小，逻辑透明	功能边界清楚，自己可以完全掌控每一步的行为

我并不是说这些框架不好，而是它们解决的是“复杂编排”和“规模化”问题。如果你今天想快速理解 Agent 原理、或者接入三五个工具做内部自动化，一个几百行的自研骨架往往比引入全套框架来得清爽。hermes-agent 的价值不是“比 LangChain 强大”，而是比 LangChain 清晰。

2. 环境准备与工程骨架搭建

2.1 技术选型：为什么是 Python 3.11 + openai SDK

环境上我选了 Python 3.11，主要看重它的asyncio库更成熟、类型标注体验也更好。Agent 天然是 IO 密集型的活儿——等模型返回、等工具执行，全是网络等待，用异步能在一个进程里并发出多个任务。

依赖我只保留了最必需的几个：

openai>=1.30.0 pydantic>=2.7.0 pydantic-settings>=2.2.0 fire>=0.6.0

很多人会问：为什么直接依赖openaiSDK，而不是自己裸写 HTTP 请求？原因有两个。第一，这个 SDK 的 Function Calling 数据模型已经封装得很完善，工具定义、返回解析都有现成的数据类；第二，它其实是事实上的“行业标准协议”，Ollama、vLLM、各种网关服务都兼容 OpenAI 格式，一套代码可以无缝切换到本地模型或者云端模型。这点后面配置里会再提到。

2.2 项目目录结构与核心模块

我习惯按模块职责分目录，而不是按类型分（models/ utils/ 那种），这样每个业务能力都有清晰的归属：

hermes-agent/ ├── hermes/ │ ├── __init__.py │ ├── config.py # 配置管理，负责读环境变量 │ ├── core/ │ │ ├── agent.py # Agent 主循环，调度 Planner 和 Executor │ │ ├── registry.py # 工具注册表，所有工具的“通讯录” │ │ ├── memory.py # 对话记忆管理，含摘要压缩 │ │ └── planner.py # 任务规划模块，与大模型交互 │ ├── tools/ │ │ ├── __init__.py │ │ ├── calculator.py # 示例工具：精确计算 │ │ ├── http.py # 示例工具：HTTP 请求 │ │ └── file_ops.py # 示例工具：文件读写 │ └── __main__.py # CLI 入口 ├── requirements.txt ├── .env.example └── README.md

目录结构看着多，其实核心代码量很小，agent.py 和 planner.py 加起来不到两百行。多出来的 tools 目录是给扩展工具的，我实际项目里接了十几个业务工具，全部独立放在这个目录下，互不影响。

2.3 最小配置与启动入口

配置这块我用了pydantic-settings，好处是配置项能自动从环境变量读取，有类型校验，还支持嵌套配置。.env.example长这样：

MODEL_NAME=gpt-4o-mini BASE_URL=https://api.openai.com/v1 API_KEY=sk-xxx MAX_ITERATIONS=6 CONTEXT_MAX_MESSAGES=12

对应config.py：

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") model_name: str = "gpt-4o-mini" base_url: str = "https://api.openai.com/v1" api_key: str = "" max_iterations: int = 6 context_max_messages: int = 12

BASE_URL这个配置至关重要。如果本地装了 Ollama，把BASE_URL改成http://127.0.0.1:11434/v1、MODEL_NAME改成qwen2.5:7b，同一套代码立刻变成跑本地模型的 Agent。这也是我一直推荐先用 openai SDK 的原因——模型服务本身已经变成了一项插拔资源，Agent 框架不该跟某个厂商绑死。

3. 核心环节实现：从零写一个可运行的 Agent

3.1 工具注册机制：让模型“看见”你的能力

大模型本身不会知道你有什么工具，你需要把工具“介绍”给它。OpenAI 的 Function Calling 规范里，每个工具都有一个名字和一段描述，模型根据描述来决定要不要调用、怎么填充参数。所以工具注册机制的设计目标就一个：让开发者用最小的代码量定义一个工具，让工具描述足够清晰。

我用装饰器实现，原因很朴素——装饰器能把工具函数变成“自带说明书”的对象，定义和注册一步到位：

import inspect import json from typing import Callable, Any class Tool: def __init__( self, name: str, description: str, func: Callable, parameters: dict, ): self.name = name self.description = description self.func = func self.parameters = parameters def to_openai_schema(self) -> dict: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters, }, } async def run(self, **kwargs) -> str: return await self.func(**kwargs) def tool(name: str, description: str, parameters: dict): def decorator(func: Callable): return Tool(name=name, description=description, parameters=parameters, func=func) return decorator

用的时候写个普通函数加一行装饰器就行：

@tool( name="calculator_add", description="精确计算两个数字相加的结果，输入必须是数字。", parameters={ "type": "object", "properties": { "a": {"type": "number", "description": "第一个加数"}, "b": {"type": "number", "description": "第二个加数"}, }, "required": ["a", "b"], }, ) async def add(a: float, b: float) -> str: return str(a + b)

这里有个我踩过多次的坑：工具的 description 千万别写得太泛。你说“一个计算工具”，模型真的可能在需要求和时调用减法工具；你说清楚了“计算两个数字相加的结果，输入必须是数字”，准确率立刻能上来一截。工具的 parameters 里每个字段也尽量写 description，模型填参数的时候参考价值很大。这套模式运行了半年，我认为它对结果准确率的影响权重至少占三成。

3.2 任务解析与规划循环：核心中的核心

Agent 的主循环核心逻辑可以用一句话概括：把对话历史和工具列表发给模型，看模型想调哪个工具，调完把结果放回去，再问模型下一步怎么办，直到模型说不需要工具了或者达到最大轮数。

这个循环在论文里叫 ReAct 或者 Plan-Execute-Replan，实现起来就是while循环加函数调用：

import asyncio from openai import AsyncOpenAI from hermes.core.registry import ToolRegistry class Agent: def __init__(self, config, registry: ToolRegistry): self.config = config self.client = AsyncOpenAI(api_key=config.api_key, base_url=config.base_url) self.registry = registry self.messages = [ { "role": "system", "content": "你是一个严谨的助手。请分析用户请求，必要时调用工具获取信息后再回答。", } ] async def run(self, user_input: str) -> str: self.messages.append({"role": "user", "content": user_input}) for step in range(self.config.max_iterations): response = await self.client.chat.completions.create( model=self.config.model_name, messages=self.messages, tools=self.registry.all_schemas(), tool_choice="auto", ) message = response.choices[0].message if not message.tool_calls: self.messages.append(message) return message.content or "（模型未返回内容）" self.messages.append(message) for tool_call in message.tool_calls: tool = self.registry.get(tool_call.function.name) if tool is None: continue try: args = json.loads(tool_call.function.arguments) result = await tool.run(**args) except Exception as exc: result = f"工具执行失败：{exc}" self.messages.append( { "role": "tool", "tool_call_id": tool_call.id, "content": str(result), } ) print(f"[step {step + 1}] 调用了 {len(message.tool_calls)} 个工具") return "已达最大迭代次数，任务未完成"

一个明显的特点是：工具执行结果不是直接返回给用户，而是以tool角色的消息送回模型。这样模型就能看到“你的请求执行得怎么样”，然后决定继续调下一个工具、换个参数重试，还是整理结果回答。这个细节是整个 Agent 能自主纠错的关键。

3.3 上下文管理与记忆持久化：别让 Agent“失忆”

多轮对话里最常见的翻车就是你问上一轮的事情，它完全不记得了。原因很简单——大模型的上下文窗口是有限的，你在循环里不断往messages里塞工具执行结果，很快就把窗口塞满了。我在memory.py里做了两层记忆管理：

第一层，滑动窗口，只保留最近 N 条消息，超过的旧消息压缩成一条摘要。这个 N 不是拍脑袋定的，跟你用的模型上下文窗口强相关。拿 8K 上下文举例：工具定义占约 1200 token，系统提示占 300 token，留 1000 token 给模型输出，剩下 5500 token 给对话历史，按每条消息平均 400 token 算，大约能放 13 条。所以我默认把CONTEXT_MAX_MESSAGES设成 12，正好留一点余量。

第二层，摘要压缩，当消息条数超过上限时，把最旧的一部分丢给模型生成摘要，把摘要作为一条系统消息放到对话开头：

async def compress(self, messages: list[dict]) -> list[dict]: if len(messages) <= self.max_messages: return messages # 旧消息压缩成摘要 to_compress = messages[1:-self.max_messages // 2] # 保留最近的各一半消息 recent = messages[-self.max_messages // 2:] summary_text = await self._summarize(to_compress) return [ {"role": "system", "content": f"以下是对更早对话的摘要：{summary_text}"}, *recent, ]

这两个机制合起来，基本能应对日常使用。如果将来要支持“长期记忆”，可以加向量数据库做语义检索，但那是另一个复杂度，初期不建议急着上。

3.4 对话接口与多轮交互：CLI 和 HTTP 二选一

有了核心 Agent，接下来就是怎么跟它说话。CLI 是最快的验证方式，我用 fire 库写了个入口：

import fire from hermes.core.agent import Agent from hermes.core.registry import ToolRegistry def main(model_name: str = "gpt-4o-mini"): config = Settings() registry = ToolRegistry() registry.load_builtin_tools() agent = Agent(config, registry) while True: user_input = input("你 > ") if user_input.strip() in ("exit", "quit"): break result = asyncio.run(agent.run(user_input)) print(f"助手 > {result}") if __name__ == "__main__": fire.Fire(main)

跑起来之后，你可以看到 Agent 在每一步打印出调用工具的信息。这种“中间过程可见”的设计，对排查模型行为至关重要。生产环境需要暴露给其他系统时，把run方法包一层 FastAPI 接口就行，核心逻辑不用动。我在实际项目里就是这么干的——CLI 给自己调试，HTTP 接口给上下游系统对接。

4. 实操过程中的坑与排查技巧

4.1 Function Calling 返回的 JSON 不合法

这是我在整个开发过程中遇到频率最高的问题，没有之一。模型端返回的tool_call.function.arguments有时会带上多余的文字说明，或者因为输出被截断导致 JSON 不完整。解决办法分两步：先清洗，再兜底。

清洗逻辑是，从返回字符串里截取第一个{到最后一个}的子串，再用json.loads解析；如果还是失败，就把这条消息重新丢给模型，提示“你刚才返回的工具参数格式不对，请只输出合法 JSON，不要添加任何额外文字”。代码里的兜底是捕获异常后把错误信息塞回去，让模型自己修正：

def safe_load_json(text: str) -> dict: start = text.find("{") end = text.rfind("}") if start == -1 or end == -1: raise ValueError("JSON 解析失败，找不到对象边界") return json.loads(text[start : end + 1])

这个技巧让我在接非 GPT 系列模型时少掉了很多头发，强烈建议做成一个通用函数。

4.2 上下文窗口溢出的三种处理策略

除了前面提到的“滑动窗口 + 摘要压缩”，实际场景里还要区分情况。如果 Agent 在做数据分析或者代码生成，才跑两步对话历史就已经很长，摘要压缩会把中间计算过程丢掉，模型后面可能“接不上”。这时候我改用截断策略：只保留系统消息、最近一轮用户消息、以及最近一个 tool 执行结果，其他全部丢掉。虽然模型会“忘记”细节，但保住最关键的那条输入输出就够了。

如果任务本身不依赖长历史，我甚至会直接把历史清空，只保留当前这一步的输入。场景不同策略不同，不能一个方案走天下。开发 Agent 时要把这个问题当成一等公民考虑，不然上线后必炸。

4.3 并发请求与 API 限流

接本地模型时，并发太高会把机器打崩；接云端 API 时，并发太高会触发限流。我的做法是在执行层外面包一个Semaphore限流器：

class RateLimiter: def __init__(self, max_concurrency: int = 4): self.semaphore = asyncio.Semaphore(max_concurrency) async def acquire(self): await self.semaphore.acquire() def release(self): self.semaphore.release()

然后每次调用模型前await limiter.acquire()，调用完在finally里release()。如果还是遇到限流，就在请求异常时的except分支里做指数退避重试，等 1 秒、2 秒、4 秒这样的递增间隔后再试。日志里把每次重试都打出来，方便判断是不是限流问题了。

4.4 日志与追踪：Agent 排错的基本盘

Agent 是个多步骤系统，任何一个环节出错都可能“失之毫厘谬以千里”。我要求自己至少要在每次模型调用前后打三样信息：本轮耗时、本轮消耗 token 数、调用工具的参数。用标准库logging就够了：

logger.info( "model call | step=%s | tokens=%s | tool=%s | args=%s", step, response.usage.total_tokens if response.usage else "N/A", tool_call.function.name, tool_call.function.arguments, )

这套日志在我的排查中帮了大忙。有一次 Agent 连续重试三次同一个错误参数，我看日志才发现是模型反复生成错误 JSON，不是工具本身的问题。没有日志的话，这种问题你只能靠猜。

5. 扩展方向与实操体会

5.1 多 Agent 协作：让专业工具各司其职

做到这一步，单 Agent 已经够用了。但真实业务里的工具往往跨领域：有数据库工具、有私有 API、有文件系统操作，全塞进一个 Agent 的工具描述里，会让模型在选择时“选择困难症”。我的方案是 supervisor-worker 模式：一个主 Agent 负责理解用户意图并分派任务，几个子 Agent 各自只带一组高内聚工具。主 Agent 的输出结果是子 Agent 的问答接口，整体结构并没有变复杂，但每个 Agent 的工具选择准确率提升很明显。

5.2 插件化加载：按需注入工具

还有一个很实用的扩展方向是把工具做成插件。我现在的做法是在tools/目录下约定一个加载协议，扫描目录里所有实现了register(registry)函数的模块，自动注册。这样新工具就是扔一个文件进去的事，不用改核心代码。对于团队协作，这个机制让不懂 Agent 原理的同事也能“插”工具进来。