★ FEATURED ARTICLE 1. 从零理解 Agent 开发的核心骨架1.1 为什么突然想自己写一个 Agent市面上关于 Agent 的文章铺天盖地但大多数停留在概念科普层面——什么是 Agent、Agent 和传统程序有什么区别、未来会怎样改变世界。这些内容看多了会产生一种错觉好像理解了但真让你动手写一个又不知道从哪行代码开始。我最初也是这个状态。直到某天看到一个开源项目 pi-agent 的设计思路才意识到 Agent 的本质远比想象中朴素。它不是什么玄学核心就是一个循环接收输入 → 交给模型推理 → 模型决定调用工具 → 执行工具 → 把结果喂回模型 → 继续推理直到得出最终答案。这个循环就是 Agent 的心脏剩下的都是围绕它做的工程优化。这篇文章的目标很明确带你从零实现一个能跑起来的小 Agent不依赖任何重型框架用最少的代码把核心机制讲透。适合有一定编程基础、想真正搞懂 Agent 内部运转逻辑的开发者。读完你不仅能自己写一个还能理解那些主流框架到底在帮你做什么。1.2 一个最小 Agent 需要哪些零件在动手之前先把零件清单列清楚。一个能用的 Agent 至少需要四个部分大模型接口负责推理和决策是整个系统的大脑。你需要一个能调用模型 API 的封装层处理请求发送、响应解析、错误重试。工具注册与调度Agent 之所以是 Agent就是因为它能调用外部工具。你需要一个机制来定义工具、描述工具、让模型知道有哪些工具可用以及实际执行工具调用。对话循环控制器管理整个推理-行动-观察的循环判断什么时候继续、什么时候结束。上下文管理维护对话历史控制 token 消耗确保模型在每一轮都能拿到足够但不冗余的信息。这四个部分缺一不可。很多人写 Agent 失败往往不是模型不够聪明而是循环控制写错了——要么死循环出不来要么工具调用结果没正确回传要么上下文爆炸导致模型开始胡言乱语。提示不要一上来就追求功能完整。先把“模型能调用一个工具并返回结果”这条链路跑通再逐步加功能。这是最稳妥的路径。1.3 为什么参照 pi-agent 而不是从零发明pi-agent 的设计哲学很值得借鉴极简但不简陋。它没有堆砌大量抽象层而是用清晰的模块划分把核心逻辑暴露出来。你读它的代码能直接看到 Agent 循环长什么样而不是在一堆设计模式里迷路。参照它来实现好处有三个。第一你能学到经过验证的架构思路避免自己拍脑袋设计出难以维护的结构。第二遇到问题时有一个对照物可以快速定位是设计问题还是实现问题。第三它的代码量适中不会让你陷入“为了理解一个功能要读十个文件”的困境。当然参照不等于照抄。我会在关键位置解释为什么这样设计以及如果你要扩展功能应该从哪里入手。这样你得到的不只是一个能跑的 Demo而是一套可以持续迭代的骨架。2. 核心模块拆解与实现要点2.1 模型接口层把 API 调用封装成可靠组件模型接口层看起来简单——不就是发个 HTTP 请求吗但实际写起来有几个坑必须提前处理。首先是超时和重试。模型 API 偶尔会抽风响应慢或者直接报错。如果不做重试Agent 跑到一半就挂了用户体验极差。我的做法是设置三层保护单次请求超时 30 秒失败后重试最多 3 次每次重试间隔递增1秒、2秒、4秒。这样既能应对偶发故障又不会在服务彻底不可用时无限等待。其次是响应格式的统一。不同模型提供商的返回结构不一样有的把内容放在choices[0].message.content有的放在output.text。你需要在接口层做一层归一化让上层代码不用关心底层用的是哪家模型。这个归一化层还能顺便处理流式和非流式响应的差异。class ModelClient: def __init__(self, api_key, base_url, model_name): self.api_key api_key self.base_url base_url self.model_name model_name self.max_retries 3 def chat(self, messages, toolsNone, temperature0.7): for attempt in range(self.max_retries): try: response self._send_request(messages, tools, temperature) return self._normalize(response) except TimeoutError: if attempt self.max_retries - 1: raise time.sleep(2 ** attempt) except RateLimitError: time.sleep(5 * (attempt 1))这里有个细节值得注意温度参数的选择。Agent 场景下我建议用较低的温度0.1 到 0.3因为你需要模型稳定地做出工具调用决策而不是发挥创意。温度太高会导致模型偶尔“忘记”工具的存在或者生成格式错误的调用请求。2.2 工具系统设计让模型知道它能做什么工具系统的核心问题是如何让模型理解一个工具的功能、参数和返回值。目前主流做法是用 JSON Schema 描述工具然后把这个描述塞进系统提示词或者专门的 tools 参数里。一个工具的定义包含四个要素要素说明示例名称唯一标识符模型调用时使用get_weather描述自然语言说明工具用途“查询指定城市的当前天气”参数 SchemaJSON Schema 格式的参数定义{city: {type: string}}执行函数实际干活的代码def get_weather(city): ...描述的质量直接决定模型能否正确使用工具。我踩过的坑是描述写得太简略模型不知道什么时候该调用描述写得太模糊模型会滥用工具。好的描述应该包含功能说明、适用场景、参数含义、返回内容四个部分。tools [ { name: search_notes, description: 在本地笔记库中搜索包含指定关键词的笔记。适用于用户询问历史记录、已保存信息等场景。返回匹配的笔记标题和摘要。, parameters: { type: object, properties: { keyword: { type: string, description: 搜索关键词支持中文和英文 }, limit: { type: integer, description: 返回结果数量上限默认5条, default: 5 } }, required: [keyword] } } ]工具执行环节需要处理参数校验和异常捕获。模型生成的参数不一定完全符合 Schema比如该传整数的地方传了字符串。你需要在执行前做一次校验和类型转换执行时用 try-except 包住把错误信息作为工具结果返回给模型让它自己决定下一步怎么办。2.3 对话循环控制器Agent 的心脏这是整个 Agent 最核心的部分也是最容易写错的地方。循环的基本逻辑是把用户输入和对话历史发给模型检查模型返回中是否有工具调用请求如果有执行工具把结果追加到对话历史回到步骤 1如果没有说明模型给出了最终答案循环结束听起来简单但实际实现时要考虑几个边界情况。最大循环次数限制。必须设置一个上限比如 10 轮。否则模型可能陷入“调用工具 → 结果不满意 → 再调用 → 还不满意”的死循环烧钱又耗时。达到上限时强制让模型基于现有信息给出回答。工具调用失败的处理。如果工具执行抛异常不要把异常直接抛给用户而是把错误信息格式化后返回给模型。模型看到“搜索失败关键词为空”这样的反馈可能会修正参数重新调用或者换一种方式回答。并行工具调用。有些模型支持一次返回多个工具调用请求。这时候可以并行执行这些工具然后把所有结果一起返回。但要注意并行执行的前提是工具之间没有依赖关系。如果工具 B 需要工具 A 的输出就必须串行。def run_agent_loop(user_input, max_turns10): messages [{role: user, content: user_input}] for turn in range(max_turns): response model_client.chat(messages, toolstools) if not response.tool_calls: return response.content messages.append(response.message) for tool_call in response.tool_calls: try: result execute_tool(tool_call.name, tool_call.arguments) messages.append({ role: tool, tool_call_id: tool_call.id, content: str(result) }) except Exception as e: messages.append({ role: tool, tool_call_id: tool_call.id, content: f工具执行失败{str(e)} }) return 已达到最大循环次数基于现有信息回答如下 model_client.chat(messages).content注意每轮循环都要检查 token 消耗。如果对话历史太长需要做截断或摘要。我的经验是当历史消息超过模型上下文窗口的 70% 时就开始把早期的工具调用结果压缩成简短摘要。2.4 上下文管理别让 token 悄悄吃掉你的预算上下文管理是很多个人开发者容易忽视的环节。Agent 跑起来之后每一轮循环都会往对话历史里追加内容token 消耗是线性增长的。一个复杂的任务跑十几轮token 用量可能轻松突破几万。我的策略是分层管理系统提示词始终保留不参与截断。这是 Agent 的“人格”和“行为准则”。最近 N 轮对话完整保留确保模型能理解当前上下文。更早的工具调用结果压缩成一句话摘要比如“已搜索到 3 条相关笔记标题分别是 A、B、C”。更早的用户输入如果和当前任务无关可以丢弃。具体实现时我会维护一个context_window列表每次追加新消息后检查总 token 数。超过阈值就从最老的非系统消息开始压缩。压缩不是简单删除而是让模型自己总结——把一段工具调用历史发给模型让它用一句话概括关键信息。这个做法有个好处模型在后续推理中仍然知道“之前做过什么”但不需要为每个细节都付出 token 成本。实测下来一个原本需要 8000 token 的对话历史压缩后能控制在 3000 token 以内效果几乎不受影响。3. 完整实操从零跑通你的第一个 Agent3.1 环境准备与项目结构先把项目骨架搭起来。我习惯用最简单的结构不搞复杂的包管理mini-agent/ ├── main.py # 入口处理用户输入 ├── agent.py # 核心循环控制器 ├── model_client.py # 模型接口封装 ├── tools.py # 工具定义与注册 ├── context.py # 上下文管理 └── config.py # 配置项依赖只需要两个requests用于 HTTP 调用json标准库处理数据。如果你用的模型 SDK 有官方库也可以直接用它替代 requests。配置项集中放在config.py# config.py MODEL_API_KEY your-api-key MODEL_BASE_URL https://api.example.com/v1 MODEL_NAME your-model-name MAX_TURNS 10 MAX_CONTEXT_TOKENS 6000 TEMPERATURE 0.2提示API Key 不要硬编码在代码里用环境变量读取。这是基本的安全习惯也方便在不同环境切换。3.2 实现一个能查天气和搜笔记的工具集为了演示工具调用我准备两个工具一个查天气模拟外部 API一个搜笔记模拟本地数据访问。# tools.py import json def get_weather(city: str) - str: 模拟天气查询 weather_data { 北京: 晴15-25°C北风3级, 上海: 多云18-26°C东南风2级, 广州: 小雨22-28°C南风2级 } return weather_data.get(city, f未找到{city}的天气数据) def search_notes(keyword: str, limit: int 5) - str: 模拟笔记搜索 notes [ {title: Agent开发笔记, summary: 记录了Agent循环控制的关键点}, {title: 模型接口封装, summary: 超时重试和响应归一化的实现}, {title: 工具系统设计, summary: JSON Schema描述工具的方法}, {title: 上下文管理策略, summary: 分层压缩控制token消耗}, {title: 常见错误排查, summary: 死循环和参数校验问题} ] matched [n for n in notes if keyword in n[title] or keyword in n[summary]] return json.dumps(matched[:limit], ensure_asciiFalse) TOOL_REGISTRY { get_weather: { function: get_weather, schema: { name: get_weather, description: 查询指定城市的当前天气情况。适用于用户询问天气、温度、是否下雨等场景。, parameters: { type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city] } } }, search_notes: { function: search_notes, schema: { name: search_notes, description: 在本地笔记库中搜索包含关键词的笔记。适用于用户询问已保存的信息、历史记录等。, parameters: { type: object, properties: { keyword: {type: string, description: 搜索关键词}, limit: {type: integer, description: 返回数量上限, default: 5} }, required: [keyword] } } } } def execute_tool(name: str, arguments: dict) - str: if name not in TOOL_REGISTRY: return f错误未知工具 {name} func TOOL_REGISTRY[name][function] try: return func(**arguments) except TypeError as e: return f参数错误{str(e)} except Exception as e: return f执行失败{str(e)}这里有个设计决策值得说明为什么用字典注册而不是装饰器。装饰器写法更优雅但字典方式更直观调试时能一眼看到所有工具。对于小项目可读性比优雅更重要。等你工具数量超过 20 个再考虑用装饰器自动注册。3.3 把循环控制器和模型接口串起来现在把前面几个模块组装起来。核心是agent.py里的循环逻辑# agent.py import json from model_client import ModelClient from tools import TOOL_REGISTRY, execute_tool from context import ContextManager import config class MiniAgent: def __init__(self): self.model ModelClient( api_keyconfig.MODEL_API_KEY, base_urlconfig.MODEL_BASE_URL, model_nameconfig.MODEL_NAME ) self.context ContextManager(max_tokensconfig.MAX_CONTEXT_TOKENS) self.tools [t[schema] for t in TOOL_REGISTRY.values()] self.max_turns config.MAX_TURNS def run(self, user_input: str) - str: self.context.add_user_message(user_input) for turn in range(self.max_turns): messages self.context.get_messages() response self.model.chat( messagesmessages, toolsself.tools, temperatureconfig.TEMPERATURE ) if not response.get(tool_calls): final_answer response[content] self.context.add_assistant_message(final_answer) return final_answer self.context.add_assistant_message( contentresponse.get(content, ), tool_callsresponse[tool_calls] ) for tool_call in response[tool_calls]: tool_name tool_call[function][name] try: arguments json.loads(tool_call[function][arguments]) except json.JSONDecodeError: arguments {} result execute_tool(tool_name, arguments) self.context.add_tool_result(tool_call[id], result) fallback self.model.chat(self.context.get_messages()) return fallback[content]ContextManager负责维护消息列表和 token 计数# context.py class ContextManager: def __init__(self, max_tokens6000): self.messages [] self.max_tokens max_tokens self.system_prompt 你是一个乐于助人的助手可以调用工具来帮助用户解决问题。 def add_user_message(self, content): self.messages.append({role: user, content: content}) self._maybe_compress() def add_assistant_message(self, content, tool_callsNone): msg {role: assistant, content: content} if tool_calls: msg[tool_calls] tool_calls self.messages.append(msg) def add_tool_result(self, tool_call_id, content): self.messages.append({ role: tool, tool_call_id: tool_call_id, content: content }) self._maybe_compress() def get_messages(self): return [{role: system, content: self.system_prompt}] self.messages def _maybe_compress(self): estimated sum(len(str(m)) // 2 for m in self.messages) if estimated self.max_tokens: self._compress_oldest() def _compress_oldest(self): # 保留最近6条消息压缩更早的工具结果 if len(self.messages) 6: return for i in range(len(self.messages) - 6): msg self.messages[i] if msg[role] tool and len(msg[content]) 100: msg[content] msg[content][:100] ...(已压缩)3.4 跑起来看看效果入口文件很简单# main.py from agent import MiniAgent def main(): agent MiniAgent() print(Mini Agent 已启动输入 quit 退出) while True: user_input input(\n你) if user_input.lower() quit: break answer agent.run(user_input) print(f\nAgent{answer}) if __name__ __main__: main()测试几个场景场景一单工具调用你北京今天天气怎么样 Agent调用 get_weather参数 city北京北京今天晴气温 15-25°C北风3级。场景二多轮工具调用你帮我搜一下关于 Agent 的笔记然后告诉我北京天气 Agent先调用 search_notes再调用 get_weather找到 3 条相关笔记Agent开发笔记、模型接口封装、工具系统设计。北京今天晴15-25°C。场景三无需工具你你好 Agent你好有什么可以帮你的吗这三个场景覆盖了 Agent 的基本行为模式。如果都能正常跑通说明你的核心链路已经打通了。4. 踩坑记录与排查手册4.1 模型不调用工具怎么办这是最常见的问题。你明明定义了工具模型却直接回答完全不调用。原因通常有三个工具描述不够清晰。模型不知道这个工具能干什么自然不敢用。解决办法是把描述写得更具体包含“适用于什么场景”“返回什么内容”。比如把“查询天气”改成“查询指定城市的当前天气情况包括温度、天气状况和风力。适用于用户询问天气相关问题时使用”。系统提示词没有引导。在系统提示词里明确告诉模型“你可以使用工具来获取信息当用户问题涉及天气或笔记时优先调用工具”。这相当于给模型一个行为暗示。模型本身能力不足。有些小模型对工具调用的支持不好经常忽略 tools 参数。这种情况只能换模型或者用 few-shot 示例在提示词里演示工具调用格式。实操心得我习惯在系统提示词里加一句“如果不确定是否需要调用工具倾向于先调用工具获取信息”。这能显著提高工具调用率代价是偶尔会多调用一次不必要的工具但总体利大于弊。4.2 工具调用参数格式错误模型生成的参数经常有小毛病该传字符串的传了数字该传数组的传了字符串或者干脆漏了必填参数。处理策略分两层第一层是执行前校验。用 JSON Schema 校验参数不合法就返回错误信息给模型。第二层是容错处理。对于类型不匹配但可以转换的情况自动转换。比如模型传了{limit: 5}你可以在执行函数里做int(limit)。def safe_execute(func, arguments): sig inspect.signature(func) bound {} for name, param in sig.parameters.items(): if name in arguments: value arguments[name] if param.annotation int and isinstance(value, str): value int(value) bound[name] value elif param.default is not inspect.Parameter.empty: bound[name] param.default else: return f缺少必填参数{name} return func(**bound)4.3 循环停不下来Agent 陷入死循环是最危险的情况因为它在持续消耗 token。除了设置最大轮数还要加一个重复检测机制如果连续两轮调用了同一个工具且参数相同就强制中断让模型基于现有信息回答。def _is_repeated(self, tool_name, arguments): key f{tool_name}:{json.dumps(arguments, sort_keysTrue)} if key in self.recent_calls: return True self.recent_calls.append(key) if len(self.recent_calls) 5: self.recent_calls.pop(0) return False另一个隐蔽的死循环原因是工具返回结果为空。模型调用工具拿到空结果觉得信息不够又调用同一个工具。解决办法是在工具返回空时明确告诉模型“未找到相关结果请基于现有信息回答或尝试其他关键词”。4.4 常见问题速查表现象可能原因排查方向解决方案模型不调用工具描述不清/提示词缺失检查工具描述和系统提示补充场景说明和调用引导参数格式错误模型生成不稳定打印模型原始返回加校验层和类型转换循环不停止无最大轮数限制检查循环退出条件设置 max_turns 和重复检测响应越来越慢上下文过长统计每轮 token 数启用压缩或截断工具执行报错参数缺失或类型错查看异常堆栈用 try-except 包裹并返回错误信息最终答案不完整提前退出循环检查退出逻辑确保模型给出非工具调用响应才退出4.5 几个让 Agent 更稳的工程技巧日志要打全。每一轮循环的输入消息、模型原始返回、工具调用参数、工具执行结果全部打到日志里。出问题时日志是你唯一的线索。我习惯用 JSON 格式打日志方便后续用脚本分析。给工具加超时。外部 API 调用可能卡住给每个工具执行设置超时比如 10 秒超时后返回“工具执行超时”让模型决定下一步。这能防止整个 Agent 被一个慢工具拖死。温度调低。前面提过Agent 场景下温度建议 0.1-0.3。我实测过温度 0.7 的情况模型经常“忘记”工具的存在或者生成格式错误的调用请求。温度低一点行为稳定很多。系统提示词要短而精。系统提示词太长会占用大量 token而且模型对长提示词的遵循度会下降。把最关键的行为准则放在前面细节说明放在后面。我的一般控制在 200 字以内。定期清理对话历史。如果 Agent 要长时间运行每隔一段时间把对话历史做一次完整摘要用摘要替换原始消息。这能显著降低 token 消耗同时保留关键上下文。5. 从能跑到好用下一步可以怎么扩展5.1 加一个简单的记忆系统现在的 Agent 每次对话都是独立的关掉再打开就忘了之前聊过什么。加一个简单的记忆系统不难把每次对话的摘要存到本地文件或轻量数据库下次启动时加载最近几条摘要作为上下文的一部分。具体做法是在ContextManager里加一个load_memory()方法启动时读取历史摘要拼接到系统提示词后面。每次对话结束后让模型生成一句摘要追加到记忆文件。这样 Agent 就有了跨会话的“长期记忆”。5.2 支持流式输出现在的实现是等模型完整返回后才显示结果用户等待感很强。改成流式输出能大幅提升体验模型每生成一个 token 就实时显示用户能看到 Agent 在“思考”。流式输出的难点在于工具调用。流式响应中工具调用的参数是分片到达的你需要累积所有分片后再解析。处理逻辑是遇到tool_calls的 delta 就追加到缓冲区遇到finish_reason为tool_calls时才执行工具。5.3 多 Agent 协作的雏形单个 Agent 能力有限但你可以让多个 Agent 分工协作。比如一个“规划 Agent”负责拆解任务一个“执行 Agent”负责调用工具一个“审核 Agent”负责检查结果。它们之间通过消息传递来协调。实现上每个 Agent 就是一个独立的MiniAgent实例只是系统提示词不同。规划 Agent 的输出作为执行 Agent 的输入执行 Agent 的结果交给审核 Agent 判断是否合格。这个模式适合处理复杂任务但要注意控制总轮数避免 Agent 之间互相等待导致超时。5.4 工具生态的扩展思路工具是 Agent 能力的边界。你可以从几个方向扩展工具集信息获取类网页搜索、文档读取、数据库查询操作执行类文件读写、命令执行、API 调用数据处理类格式转换、计算、统计分析交互类发送通知、创建任务、更新状态每加一个工具都要问自己三个问题模型能理解它的用途吗参数设计合理吗返回结果对模型友好吗这三个问题答好了工具才能真正被用起来。我个人在实际操作中的体会是Agent 开发最难的不是写代码而是设计好模型和工具之间的“接口”。工具描述写得好模型就能用得顺写得含糊模型就各种误用。这个平衡需要反复调试没有捷径。另外别追求一次做到完美先把核心循环跑通再逐步加功能每一步都验证到位比一口气写完再调试要高效得多。
阅读完成 · 觉得有帮助？
