★ FEATURED ARTICLE 1. 为什么我要自己动手写一个 Agent市面上关于 Agent 的文章十篇里有八篇在讲概念——什么自主规划、工具调用、记忆机制翻来覆去就那几个词。但真正落到代码层面很多人就卡住了Agent 的循环到底怎么写工具调用的结果怎么塞回上下文多轮对话的状态存在哪里这些问题光看架构图是看不出来的。我最初也是被这些问题困扰后来看到一个叫 pi-agent 的轻量实现代码量不大但把 Agent 的核心骨架搭得很清楚。于是我决定参照它的思路从零手搓一个小 Agent不依赖任何重型框架纯靠基础库把整个链路跑通。这篇文章就是整个过程的完整记录——包括设计思路、关键代码、踩过的坑以及我对每一步“为什么这么做”的理解。这篇文章适合两类人一是已经用过某些 Agent 框架但想搞清楚底层原理的开发者二是想自己写一个 Agent 但不知道从哪下手的新手。我会尽量把每个环节讲透代码可以直接抄作业也可以根据自己需求改。2. 整体设计一个 Agent 到底需要什么2.1 核心循环Agent 的心跳Agent 和普通的聊天机器人最大的区别在于它不是“你问一句我答一句”就结束了而是会根据你的目标自己决定下一步做什么调用工具拿到结果再决定下一步直到任务完成或者达到终止条件。这个“决定-执行-观察-再决定”的过程就是 Agent 的核心循环。用伪代码表示大概是这样while not done: response llm.chat(messages) if response.has_tool_call: result execute_tool(response.tool_call) messages.append(result) else: done True看起来很简单对吧但魔鬼在细节里。比如怎么判断任务完成了工具调用失败了怎么办循环次数太多怎么兜底这些后面会逐一展开。2.2 模块拆解四个核心组件参照 pi-agent 的设计我把整个 Agent 拆成了四个模块LLM 接口层负责和模型通信发送消息、接收回复。这一层要做的是把不同模型的 API 差异抹平上层不需要关心用的是哪家模型。工具注册与调度定义工具的描述格式解析模型的工具调用请求执行对应的函数把结果格式化后返回。对话状态管理维护消息历史包括用户输入、模型回复、工具调用记录、工具返回结果。这是 Agent 的“记忆”。主循环控制器串联以上所有模块控制循环的启动、终止和异常处理。为什么这么拆因为这样每个模块都可以独立替换。比如你想换个模型只改 LLM 接口层就行想加新工具只改工具注册表就行。这种解耦在实际开发中非常关键否则代码会很快变成一团乱麻。2.3 技术选型为什么不用现成框架你可能会问LangChain、AutoGPT 这些框架不香吗为什么要自己写我的理由有三个。第一学习目的——用框架你只能学到怎么调 API自己写才能理解每一步在干什么。第二可控性——框架的抽象层有时候会挡住关键细节出了问题很难排查。第三轻量需求——很多场景下你只需要一个简单的 Agent引入重型框架反而增加了复杂度和不确定性。当然如果你是要做生产级的产品该用框架还是得用。但作为学习和理解原理的手段自己动手写一遍是最有效的。3. 核心细节每个模块怎么实现3.1 LLM 接口层统一通信协议这一层的核心任务是给定一组消息返回模型的回复。回复可能是纯文本也可能包含工具调用请求。我选择用 OpenAI 兼容的接口格式作为标准因为目前大部分模型服务都支持这个格式。核心代码大概长这样import httpx import json class LLMClient: def __init__(self, base_url, api_key, model): self.base_url base_url self.api_key api_key self.model model def chat(self, messages, toolsNone): payload { model: self.model, messages: messages, } if tools: payload[tools] tools response httpx.post( f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, jsonpayload, timeout60, ) return response.json()[choices][0][message]这里有几个关键点需要注意。超时设置。Agent 场景下模型可能需要较长时间来推理尤其是带工具调用的复杂任务。我一开始设了 30 秒结果经常超时。后来改成 60 秒基本够用。如果你的场景更复杂可以设到 120 秒。错误处理。网络请求失败、API 返回错误码、返回格式不符合预期——这些都要处理。我的做法是封装一个重试机制最多重试 3 次每次间隔递增。如果连续失败就抛出一个明确的异常让主循环决定怎么处理。消息格式。OpenAI 的消息格式里每条消息有role和content两个核心字段。role可以是system、user、assistant、tool。工具调用的结果需要用tool角色返回并且带上tool_call_id来对应是哪次调用。这个细节很容易搞错搞错了模型就不知道工具返回的结果对应哪个请求。3.2 工具注册让模型知道它能做什么工具是 Agent 能力的延伸。没有工具Agent 就只是一个聊天机器人有了工具它才能查数据、算数学、读写文件、调用外部服务。工具的定义遵循 OpenAI 的 function calling 格式tools [ { type: function, function: { name: calculator, description: 执行数学计算支持加减乘除和幂运算, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式如 2 3 * 4 } }, required: [expression] } } } ]然后在本地维护一个名称到函数的映射TOOL_REGISTRY {} def register_tool(name, func): TOOL_REGISTRY[name] func def execute_tool(name, arguments): if name not in TOOL_REGISTRY: return f错误未找到工具 {name} try: result TOOL_REGISTRY[name](**arguments) return str(result) except Exception as e: return f工具执行失败{e}这里有个经验工具的 description 写得越清楚模型调用得越准确。我一开始写得很简略结果模型经常传错参数或者在不该调用的时候调用。后来我把每个工具的描述都写成了“什么场景下用、参数是什么意思、返回什么格式”准确率明显提升。另外参数校验不能省。模型有时候会传一些奇怪的参数比如该传数字的传了字符串该传数组的传了单个值。在工具函数内部做好类型检查和转换能避免很多莫名其妙的报错。3.3 对话状态管理Agent 的记忆消息历史是 Agent 的短期记忆。每次循环我们都要把完整的消息历史发给模型模型才能知道之前发生了什么。class Conversation: def __init__(self, system_prompt): self.messages [ {role: system, content: system_prompt} ] def add_user_message(self, content): self.messages.append({role: user, content: content}) def add_assistant_message(self, message): self.messages.append(message) def add_tool_result(self, tool_call_id, content): self.messages.append({ role: tool, tool_call_id: tool_call_id, content: content })看起来简单但有两个坑我踩过。第一个坑消息太长。Agent 循环几轮之后消息历史会变得很长可能超出模型的上下文窗口。解决办法有两种一是截断最早的消息但要保留 system prompt二是对历史消息做摘要压缩。我目前用的是简单截断保留最近 N 条消息。如果任务复杂建议上摘要方案。第二个坑工具调用消息的完整性。当模型返回一个工具调用请求时assistant 消息里会包含tool_calls字段。这条消息必须原样保存不能只保存文本内容。否则下一轮发给模型时模型会发现工具调用的结果没有对应的请求直接报错。3.4 主循环把所有东西串起来主循环是整个 Agent 的引擎。它的逻辑是把当前消息历史发给模型检查模型回复中是否有工具调用如果有执行工具把结果加入消息历史回到步骤 1如果没有输出模型的文本回复结束def run_agent(conversation, llm_client, tools, max_iterations10): for i in range(max_iterations): response llm_client.chat(conversation.messages, toolstools) conversation.add_assistant_message(response) if not response.get(tool_calls): return response.get(content, ) for tool_call in response[tool_calls]: name tool_call[function][name] args json.loads(tool_call[function][arguments]) result execute_tool(name, args) conversation.add_tool_result(tool_call[id], result) return 达到最大循环次数任务未完成max_iterations是一个安全阀。没有它模型可能会陷入无限循环——比如反复调用同一个工具但每次都得不到想要的结果。我一般设 10 次大部分任务 3-5 次循环就能完成。4. 实操过程从零跑通一个完整 Agent4.1 环境准备与依赖安装整个项目只需要三个依赖pip install httpx json5httpx用来发 HTTP 请求json5用来解析模型返回的 JSON有时候模型会返回带注释或尾逗号的 JSON标准库解析不了。如果你用的模型服务需要额外的 SDK按需安装即可。但我建议尽量用 HTTP 接口直接调用少一层封装少一层坑。4.2 定义工具函数我先定义两个最基础的工具计算器和当前时间查询。import math from datetime import datetime def calculator(expression: str) - str: allowed set(0123456789-*/.() ) if not all(c in allowed for c in expression): return 错误表达式包含不允许的字符 try: result eval(expression, {__builtins__: {}}, {math: math}) return str(result) except Exception as e: return f计算错误{e} def get_current_time(timezone: str local) - str: now datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S)calculator里我做了字符白名单校验只允许数字和基本运算符。这是为了防止模型生成恶意表达式。虽然在自己本地跑风险不大但养成好习惯没坏处。4.3 组装并运行把前面所有模块拼起来SYSTEM_PROMPT 你是一个智能助手可以使用工具来帮助用户解决问题。 当需要计算时使用 calculator 工具。 当需要知道当前时间时使用 get_current_time 工具。 请一步一步思考先理解用户需求再决定是否需要调用工具。 def main(): llm LLMClient( base_url你的模型服务地址, api_key你的密钥, model你的模型名称 ) register_tool(calculator, calculator) register_tool(get_current_time, get_current_time) conversation Conversation(SYSTEM_PROMPT) conversation.add_user_message(帮我算一下 (15 27) * 3 等于多少然后告诉我现在几点) result run_agent(conversation, llm, tools) print(result) if __name__ __main__: main()运行之后你会看到 Agent 先调用 calculator 算出 126再调用 get_current_time 获取时间最后把两个结果整合成一段完整的回复。整个过程不需要你手动干预Agent 自己决定调用哪些工具、按什么顺序调用。4.4 参数选择与调优记录在实际跑的过程中我调整了几个参数记录一下参数初始值调整后原因请求超时30s60s复杂任务推理时间长30s 经常超时最大循环次数510多工具任务需要更多轮次消息历史保留条数全部最近 20 条防止上下文溢出重试次数03网络抖动时自动恢复这些值不是固定的你需要根据自己的场景调整。比如如果你的任务很简单最大循环次数设 5 就够了如果任务特别复杂可能要设到 20。5. 常见问题与排查技巧5.1 模型不调用工具怎么办这是最常见的问题。模型要么直接用自己的知识回答要么说“我无法完成这个任务”。排查思路检查工具描述是否清晰。模型是根据 description 来判断什么时候该用工具的。如果描述太模糊模型就不知道什么时候该调用。检查 system prompt。在 system prompt 里明确告诉模型“你有以下工具可用遇到 X 场景请使用 Y 工具”。检查模型是否支持 function calling。不是所有模型都支持工具调用。如果不支持你需要在 prompt 里让模型输出特定格式的文本然后自己解析。5.2 工具调用参数错误模型传的参数类型不对、缺少必填参数、参数名拼错——这些都会导致工具执行失败。我的做法是在工具函数入口做严格的参数校验返回明确的错误信息。这样模型在下一轮看到错误信息后有机会自我修正。比如模型传了{expr: 11}但参数名应该是expression工具返回“缺少必填参数 expression”模型下一轮就会传对。5.3 循环停不下来有时候模型会反复调用同一个工具每次都得到相同的结果但就是不结束。解决办法有两个一是设max_iterations硬性截断二是在工具返回结果里加入提示比如“如果这个结果已经满足需求请直接回复用户不要再调用工具”。5.4 常见问题速查表问题现象可能原因解决方法模型不调用工具工具描述不清 / 模型不支持优化描述 / 换模型参数传递错误模型理解偏差加强参数校验和错误提示无限循环模型陷入死循环设最大循环次数 结果提示上下文溢出消息历史太长截断或摘要压缩请求超时任务太复杂增加超时时间 重试工具结果格式错误返回值不是字符串统一转成字符串再返回5.5 独家避坑技巧技巧一给工具返回值加前缀。比如计算器返回计算结果126而不是直接返回126。这样模型更容易理解这个结果的含义减少误用。技巧二在 system prompt 里加 few-shot 示例。给模型展示一两个完整的“用户提问-工具调用-结果整合”的示例能显著提升工具调用的准确率。技巧三日志要打全。每次请求和响应都完整记录下来包括消息历史、工具调用参数、工具返回结果。出问题的时候翻日志比猜原因快得多。技巧四工具粒度要适中。太粗的工具比如一个工具能做十件事模型不知道怎么传参太细的工具比如每个数学运算一个工具模型要调用很多次。我一般按“一个工具做一类事”的原则来划分。6. 后续可以怎么扩展这个最小 Agent 跑通之后你可以根据需求往几个方向扩展。加记忆。目前的对话历史只存在内存里程序退出就没了。你可以加一个持久化层把消息存到文件或数据库下次启动时加载回来。加更多工具。比如网页搜索、文件读写、代码执行、API 调用。每加一个工具Agent 的能力边界就扩大一圈。加多 Agent 协作。一个 Agent 负责规划一个负责执行一个负责审核。这种模式在复杂任务上效果更好但实现复杂度也更高。加流式输出。目前是等模型完整回复后才处理用户体验不够好。改成流式输出模型边生成边显示体验会好很多。加人工确认。对于危险操作比如删除文件、发送请求在工具执行前弹一个确认框让用户决定是否继续。这在生产环境里是必须的。我自己在实际使用中发现最小 Agent 最大的价值不是它能做多少事而是它让你真正理解了 Agent 的工作原理。一旦你理解了循环、工具调用、状态管理这三个核心概念再去用任何框架都会觉得理所当然。反过来如果你直接上框架遇到问题可能连从哪里排查都不知道。最后分享一个小技巧调试 Agent 的时候把每一轮的消息历史打印出来你会直观地看到模型是怎么“思考”的。有时候模型的决策逻辑会让你意外但正是这些意外让你更懂它。
阅读完成 · 觉得有帮助？
