1. 为什么我劝你手写一个 AI Agent,而不是直接调 API
先说一个我观察到的现象:现在市面上讲 AI Agent 的内容,九成都在教你怎么用现成框架——拖几个节点、连几条线、填个 API Key,一个“智能体”就跑起来了。这没错,能快速出活。但问题在于,一旦线上出了诡异问题,比如工具调用死循环、上下文突然爆炸、多轮对话后模型开始胡言乱语,你打开框架源码一看,层层封装,根本不知道从哪下手。
我自己就踩过这个坑。早些年用某个主流 Agent 框架做一个客服场景,测试环境一切正常,上线第二天开始间歇性抽风——同一个用户问题,有时候走工具查询,有时候直接编答案。排查了整整两天,最后发现是框架内部对工具描述做了自动截断,导致模型在某些上下文长度下“看不见”那个工具。这种事,你不手写一遍,永远不知道底层发生了什么。
所以这篇内容的核心思路很明确:用最朴素的方式,从零手写一个最小可用的 AI Agent,不依赖任何重型框架,把“会调用”变成“懂原理”。适合两类人:一是刚入门 AI 应用开发、被各种框架名词绕晕的新手;二是已经会用框架、但想搞清楚内部机制的进阶者。读完你至少能自己实现一个带工具调用、多轮记忆、错误重试的 Agent 骨架,并且知道每一行代码为什么这么写。
2. 手写之前,先把 AI Agent 的骨架拆干净
2.1 Agent 到底是什么:一个循环加三样东西
很多人把 Agent 想复杂了。剥掉所有包装,一个 Agent 本质上就是一个循环,循环里做三件事:感知输入、决策下一步、执行动作。用大白话说,就是“看情况、想一下、动手做”,然后重复,直到任务完成。
那“三样东西”是什么?第一是大模型,负责“想”;第二是工具集,负责“做”;第三是记忆,负责“记住之前发生了什么”。这三样缺一不可。没有模型,Agent 没有决策能力;没有工具,Agent 只能空谈;没有记忆,Agent 每轮都从零开始,多轮任务直接崩。
我见过不少新手一上来就研究 LangGraph 的状态图、AutoGPT 的任务分解,结果连最基本的“模型怎么决定调用哪个工具”都没搞明白。这就像学编程先背设计模式,却不会写 for 循环。手写的意义就在于,你被迫面对每一个细节:工具描述怎么给模型看、模型返回的调用请求怎么解析、执行结果怎么塞回上下文。
2.2 为什么选“手写”而不是“调框架”
框架当然有它的价值,生产环境该用还是得用。但学习阶段,框架是毒药。原因有三:
- 黑盒太多:框架帮你做了 prompt 拼接、工具路由、上下文裁剪,你不知道它怎么做的,出了问题只能猜。
- 抽象泄漏:所有框架都有抽象泄漏,用久了总会遇到“框架行为不符合预期”的时刻,这时候底层知识就是你的救命稻草。
- 调试困难:手写代码你可以随时 print 中间状态,框架里加个日志都得翻文档。
我的建议是:先手写一个 200 行左右的版本,跑通完整流程,再去看框架源码。你会发现框架做的事你全做过,只是它做得更工程化。这个顺序不能反。
2.3 最小可用 Agent 的四个模块
我手写的版本拆成四个模块,每个模块职责单一:
| 模块 | 职责 | 关键点 |
|---|---|---|
| 模型接口层 | 封装大模型调用 | 统一输入输出格式,方便换模型 |
| 工具注册层 | 管理可用工具 | 工具描述要清晰,参数要校验 |
| 记忆管理层 | 维护对话历史 | 控制上下文长度,避免爆炸 |
| 循环控制层 | 驱动 Agent 运转 | 处理终止条件、错误重试 |
这四个模块加起来不到 200 行 Python,但麻雀虽小五脏俱全。下面逐个拆。
3. 核心模块逐个手撕:从模型接口到循环控制
3.1 模型接口层:别让模型调用散落各处
模型接口层的作用是把“调用大模型”这件事封装成一个函数,输入是消息列表,输出是模型的回复。为什么要封装?因为你需要换模型。今天用这个,明天可能换那个,如果调用代码散落在十几个地方,换一次模型改到吐血。
我用的接口长这样:
def call_llm(messages, tools=None): # messages: [{"role": "user", "content": "..."}] # tools: 工具描述列表,可选 response = client.chat.completions.create( model="your-model", messages=messages, tools=tools, temperature=0 ) return response.choices[0].message这里有个关键决策:temperature 设成 0。Agent 场景下,你需要的是稳定、可复现的决策,不是创意。温度高了,同一个输入可能这次调工具 A,下次调工具 B,调试起来会疯。我实测下来,Agent 场景温度超过 0.3 就开始不稳定,0 是最稳的。
另一个细节是tools参数。不同模型厂商的工具描述格式不一样,有的用 JSON Schema,有的用特定字段。接口层要做的就是抹平这个差异,让上层代码不用关心底层用的是哪家模型。我一般会写一个适配函数,把统一的工具描述转成各家格式。
注意:模型接口层一定要加超时和重试。大模型调用偶尔抽风是常态,没有重试机制,你的 Agent 会在网络抖动时直接挂掉。我一般设 30 秒超时,重试 2 次,指数退避。
3.2 工具注册层:工具描述写得好,模型调用错不了
工具注册层是很多人忽略的重灾区。新手往往随便写个函数就注册进去,结果模型要么不调用,要么调用时参数传错。问题出在工具描述上。
模型决定调用哪个工具,完全依赖你给的描述。描述写得含糊,模型就瞎猜。我总结了一个工具描述的三要素模板:
- 功能说明:一句话说清这个工具干什么,用模型能理解的自然语言。
- 参数说明:每个参数的类型、含义、是否必填,越具体越好。
- 使用场景:什么情况下该用这个工具,什么情况下不该用。
举个例子,一个查询天气的工具,描述可以这样写:
{ "name": "get_weather", "description": "查询指定城市的当前天气。当用户询问天气、气温、是否下雨等问题时使用此工具。不要用于查询历史天气或未来预报。", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,如'北京'、'上海',不要带'市'字" } }, "required": ["city"] } }注意最后那句“不要用于查询历史天气或未来预报”——这就是使用场景的边界说明。没有这句话,模型可能在用户问“明天天气”时也调用这个工具,然后返回错误结果。我踩过这个坑,加了边界说明后,误调用率明显下降。
工具注册层还要做参数校验。模型返回的参数不一定符合你的预期,可能缺字段、类型不对、甚至多传字段。执行工具前必须校验,校验失败就返回错误信息给模型,让它重新决策。这比直接抛异常优雅得多。
3.3 记忆管理层:上下文不是越多越好
记忆管理是 Agent 最容易出问题的地方。新手常犯的错是把所有对话历史一股脑塞给模型,结果上下文越来越长,token 费用飙升,模型还开始“失忆”——因为关键信息被淹没在大量无关内容里。
我的做法是分层记忆:
- 短期记忆:最近 N 轮对话,完整保留。
- 长期记忆:更早的对话,压缩成摘要。
- 工具记忆:工具调用和结果,只保留最近几次,避免重复调用。
具体实现上,我设一个 token 阈值,比如 4000 token。超过阈值就把最早的几轮对话拿出来,让模型总结成一段话,替换掉原始对话。这样上下文长度可控,关键信息也不丢。
def manage_memory(messages, max_tokens=4000): if count_tokens(messages) <= max_tokens: return messages # 保留最近 5 轮,更早的做摘要 recent = messages[-10:] older = messages[:-10] summary = summarize(older) return [{"role": "system", "content": f"之前的对话摘要:{summary}"}] + recent这里有个经验:摘要要保留实体和决策。比如“用户想查北京天气,已调用 get_weather 返回晴”,而不是“用户问了天气相关的问题”。前者模型能接着用,后者等于没说。
提示:工具调用的结果不要全部保留。比如查询数据库返回了 100 行数据,塞进上下文纯属浪费。我一般只保留前几行加一个“共 N 条”的说明,模型需要更多会再调工具。
3.4 循环控制层:Agent 的心脏
循环控制层是整个 Agent 的心脏,负责驱动“想-做-看”的循环。核心逻辑是一个 while 循环,每轮做四件事:
- 把当前消息列表发给模型。
- 检查模型返回:是普通回复还是工具调用请求。
- 如果是工具调用,执行工具,把结果追加到消息列表,继续循环。
- 如果是普通回复,输出给用户,结束循环。
伪代码大概这样:
def run_agent(user_input, max_turns=10): messages = [{"role": "user", "content": user_input}] for turn in range(max_turns): response = call_llm(messages, tools=tool_schemas) if response.tool_calls: for call in response.tool_calls: result = execute_tool(call.name, call.arguments) messages.append({"role": "tool", "content": result}) else: return response.content return "达到最大轮次,任务未完成"这里有几个关键参数:
- max_turns:最大循环轮次,防止死循环。我一般设 10,复杂任务可以放宽到 20。
- 终止条件:模型返回普通文本就终止。但要注意,模型可能返回“我需要更多信息”这种,这时候要判断是继续问用户还是继续循环。
- 错误处理:工具执行失败时,把错误信息作为工具结果返回给模型,让它决定下一步。不要直接抛异常终止。
我踩过的一个坑是:模型有时候会连续调用同一个工具,参数还一样。这是典型的死循环前兆。我的处理方式是加一个重复调用检测,如果连续两次调用相同工具且参数相同,就强制终止并返回提示。
4. 完整实操:从零跑通一个能查天气和算数的 Agent
4.1 环境准备与依赖选择
先说环境。Python 3.10 以上,这是底线。依赖就三个:openai(或对应厂商的 SDK)、pydantic(参数校验)、tiktoken(token 计数)。不需要 LangChain,不需要任何 Agent 框架。
pip install openai pydantic tiktoken为什么不用 LangChain?因为我们要手写,用了框架就失去意义了。而且这三个依赖足够支撑一个完整的 Agent,没必要引入重型依赖。
API Key 通过环境变量传入,不要硬编码在代码里。这是基本的安全习惯,我见过太多人把 Key 写死在代码里然后传到公开仓库。
4.2 定义两个工具:天气查询和四则运算
工具不用复杂,两个就够演示。一个是get_weather,一个是calculate。
def get_weather(city: str) -> str: # 模拟数据,实际应调用天气 API mock_data = {"北京": "晴,25°C", "上海": "多云,28°C"} return mock_data.get(city, f"未找到{city}的天气数据") def calculate(expression: str) -> str: try: # 安全起见,只允许数字和运算符 allowed = set("0123456789+-*/.() ") if not all(c in allowed for c in expression): return "表达式包含非法字符" return str(eval(expression)) except Exception as e: return f"计算错误:{e}"注意calculate里的安全校验。直接用eval是危险的,必须限制字符集。这是手写 Agent 时必须考虑的安全问题,框架不会帮你做。
工具描述按前面说的三要素写,注册到一个字典里:
TOOLS = { "get_weather": { "function": get_weather, "schema": { "name": "get_weather", "description": "查询指定城市的当前天气。用户询问天气时使用。", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名"} }, "required": ["city"] } } }, "calculate": { "function": calculate, "schema": { "name": "calculate", "description": "计算数学表达式。用户需要做算术时使用。", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式,如'2+3*4'"} }, "required": ["expression"] } } } }4.3 拼装主循环:让 Agent 真正跑起来
把前面几个模块拼起来,主循环大概 50 行:
import json from openai import OpenAI client = OpenAI() def run_agent(user_input, max_turns=10): messages = [ {"role": "system", "content": "你是一个助手,可以查天气和做计算。需要时调用工具。"}, {"role": "user", "content": user_input} ] tool_schemas = [t["schema"] for t in TOOLS.values()] for turn in range(max_turns): response = client.chat.completions.create( model="your-model", messages=messages, tools=tool_schemas, temperature=0 ).choices[0].message if not response.tool_calls: return response.content messages.append(response) for call in response.tool_calls: name = call.function.name args = json.loads(call.function.arguments) if name in TOOLS: result = TOOLS[name]["function"](**args) else: result = f"未知工具:{name}" messages.append({ "role": "tool", "tool_call_id": call.id, "content": str(result) }) return "达到最大轮次,任务未完成"跑一下:
print(run_agent("北京天气怎么样?顺便帮我算一下 25*4+10"))正常的话,Agent 会先调get_weather查北京,再调calculate算数,最后综合两个结果给出回复。整个过程你能看到每一轮的消息列表变化,这就是手写的好处——完全透明。
4.4 参数计算:max_turns 和 token 阈值怎么定
这两个参数没有标准答案,但有参考逻辑。
max_turns取决于任务复杂度。单工具任务 3-5 轮足够,多工具协作 10 轮,复杂任务分解 20 轮。设太大浪费 token,设太小任务做不完。我的经验是:先设 10,观察实际任务的轮次分布,取 P95 值加 2。
token 阈值取决于模型上下文窗口。假设模型支持 8K 上下文,你要留出输出空间,输入控制在 4K 左右。其中系统提示占 500,工具描述占 500,剩下 3K 给对话历史。按每轮 200 token 算,大概保留 15 轮。超过就做摘要。
这两个参数都要根据实际运行数据调整,不要拍脑袋定死。
5. 踩坑实录:手写 Agent 最常见的五个问题
5.1 模型不调用工具,直接编答案
这是最高频的问题。用户问天气,模型不调工具,直接说“北京今天晴”。原因通常是工具描述不够清晰,或者系统提示没强调“必须调用工具获取实时信息”。
解决办法:在系统提示里明确写“涉及天气、计算等问题时,必须先调用工具,不要凭记忆回答”。同时检查工具描述是否说清了使用场景。我实测下来,加了这句提示后,工具调用率从 60% 提升到 95% 以上。
5.2 工具调用参数格式错误
模型返回的参数可能是字符串而不是对象,或者字段名拼错。这是模型能力问题,但你可以通过参数校验加错误反馈来兜底。校验失败时,把错误信息返回给模型,它下一轮通常会修正。
def validate_args(schema, args): required = schema["parameters"].get("required", []) for field in required: if field not in args: return f"缺少必填参数:{field}" return None5.3 上下文爆炸导致模型失忆
前面提过,所有历史一股脑塞进去,模型反而记不住关键信息。解决办法就是分层记忆加摘要。我还会在系统提示里加一句“优先关注最近的对话和工具结果”,引导模型注意力。
5.4 工具执行超时或异常
工具调用外部 API 时可能超时。必须给每个工具加超时控制,超时后返回“工具执行超时”给模型,让它决定重试还是换方案。不要让它无限等待。
5.5 死循环:同一个工具反复调用
前面提过重复调用检测。具体实现是记录最近几次调用的工具名和参数哈希,如果连续两次相同,就强制终止。这个检测救过我很多次,尤其是模型陷入“调用-失败-再调用”循环时。
| 问题 | 现象 | 解决 |
|---|---|---|
| 不调工具 | 模型直接编答案 | 系统提示强调必须调用 |
| 参数错误 | 工具执行报错 | 参数校验加错误反馈 |
| 上下文爆炸 | 模型失忆、费用高 | 分层记忆加摘要 |
| 工具超时 | Agent 卡住 | 加超时控制 |
| 死循环 | 反复调同一工具 | 重复调用检测 |
6. 从手写版到生产级:下一步该补什么
手写版跑通后,你已经理解了 Agent 的核心机制。但生产环境还需要补几块:
并发处理。手写版是单线程的,多个用户同时请求会排队。生产环境需要异步化,把模型调用和工具执行都改成 async。这块改动不小,但思路清晰:每个请求独立的消息列表,共享工具注册表。
可观测性。加日志、加追踪,记录每一轮的输入输出、token 消耗、工具调用耗时。出了问题能快速定位。我一般用结构化日志,每条记录带 request_id,方便串联。
工具生态。手写版的工具是硬编码的,生产环境需要动态注册、权限控制、限流。这块可以参考 MCP 协议的设计思路,但不用全盘照搬。
评测体系。怎么知道你的 Agent 好不好?需要一套评测集,覆盖常见场景和边界情况,每次改动跑一遍,看通过率变化。没有评测,优化就是盲人摸象。
我个人的体会是,手写一遍最大的收获不是代码本身,而是对 Agent 行为有了直觉。当模型不调工具时,你知道该去看工具描述;当上下文爆炸时,你知道该去查记忆管理。这种直觉,是调框架调不出来的。后续你再去看任何 Agent 框架的源码,都会有“哦,它这里是在做我手写过的那件事”的感觉。这个内容后续还可以这样扩展:把手写版改造成支持多 Agent 协作,或者接入真实的工具生态,都是水到渠成的下一步。