ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

轻量级AI Agent框架实践:从零搭建大模型工具调用与任务规划系统

轻量级AI Agent框架实践:从零搭建大模型工具调用与任务规划系统 Hermes 这名字懂点希腊神话的朋友应该不陌生就是那位脚上长翅膀、整天忙着传信的使者。把 agent 接在它后面想表达的意思很直接我想做一个负责“传递意图、调度工具”的智能体让大模型不只会聊天还能真正上手干点实事。hermes-agent 这个项目本质上就是一个轻量级的 AI Agent 框架解决的是“怎么让大模型稳定地调用外部工具、完成多步任务”这个核心问题。过去一年我试过不少 Agent 框架LangChain 用起来总觉得抽象层太厚AutoGPT 自动化过头了很难控最后还是决定自己撸一个最小可用的实现。这个项目不是要跟那些大框架掰手腕而是想给同样好奇 Agent 内部原理、或者需要在业务系统里快速接工具的开发者一个可复现的参考。读完这篇你能照着搭出一个能跑通“用户提问-拆解任务-调用工具-返回结果”完整链路的 Agent还能学会怎么处理上下文溢出、工具调用失败这些实战里躲不掉的坑。1. 这个项目解决什么问题源起与核心设计思路1.1 为什么叫 Hermes信使隐喻与定位Hermes 在神话里的职责是穿梭于神界和人界之间传递信息特点是快、准、不迷路。Agent 系统里的“信使”也是个很贴切的比喻——用户把请求交给 AgentAgent 要做的不是自己去算而是判断该找谁帮忙、把请求翻译成对方听得懂的格式、拿到结果再带回来。拿我自己常遇到的一个场景举例运营同事发来一句话“帮我查下这批订单的支付金额再算一下总利润”。如果只靠大模型它既读不到数据库也算不了精确的浮点如果全写死在代码里又没法消化人类这种模糊表达。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 密集型的活儿——等模型返回、等工具执行全是网络等待用异步能在一个进程里并发出多个任务。依赖我只保留了最必需的几个openai1.30.0 pydantic2.7.0 pydantic-settings2.2.0 fire0.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_NAMEgpt-4o-mini BASE_URLhttps://api.openai.com/v1 API_KEYsk-xxx MAX_ITERATIONS6 CONTEXT_MAX_MESSAGES12对应config.pyfrom pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-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 12BASE_URL这个配置至关重要。如果本地装了 Ollama把BASE_URL改成http://127.0.0.1:11434/v1、MODEL_NAME改成qwen2.5:7b同一套代码立刻变成跑本地模型的 Agent。这也是我一直推荐先用 openai SDK 的原因——模型服务本身已经变成了一项插拔资源Agent 框架不该跟某个厂商绑死。3. 核心环节实现从零写一个可运行的 Agent3.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(namename, descriptiondescription, parametersparameters, funcfunc) return decorator用的时候写个普通函数加一行装饰器就行tool( namecalculator_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_keyconfig.api_key, base_urlconfig.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( modelself.config.model_name, messagesself.messages, toolsself.registry.all_schemas(), tool_choiceauto, ) 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 原理的同事也能“插”工具进来。5.3 使用体会这个项目值不值得复刻最后聊点实在的。这个项目前前后后跑了几个月我最大的感受是Agent 框架本身的技术门槛没有想象中那么高真正的门槛在对模型行为的理解和容错设计上。自研一个最小骨架值不值我认为特别值尤其适合这三类人一是刚接触 Agent 的开发者自己动手写一遍核心循环比读十遍 LangChain 文档都有用。二是被大框架抽象折腾到痛不欲生的人你会发现自己掌控一切的感觉有多爽。三是需要在内部快速集成工具的团队几百行代码带来的透明度和可控性在排查问题时收益巨大。如果你想在生产环境大规模跑复杂编排还是建议用 LangChain 那类成熟框架但如果目标是真正搞懂 Agent 是怎么工作的或者做内部轻自动化hermes-agent 这个思路值得你照着重写一遍。个人体会是从骨架出发去理解 Agent比从框架入门去理解 Agent顺畅太多了。
返回列表