1. 为什么我要从零手写一个AI Agent
1.1 一个让我下定决心的事件
去年年底,我在一个技术群里看到有人问:“AI Agent怎么扛并发?”底下有人回了一句“加机器”,然后就没人继续了。我当时正好在做一个内部工具,需要让模型自动完成一些多步骤的任务,比如根据一段需求描述生成代码、跑测试、根据报错再改代码。我一开始用的是现成的Agent框架,拖拖拽拽,配几个工具,跑起来确实快。但上线之后问题就来了:并发一上来,任务队列堵死;某个工具调用超时,整个链路卡住;更离谱的是,有时候模型明明该调用搜索工具,却自己编了一段答案。
我去翻框架的源码,发现它内部的状态管理、工具路由、错误重试逻辑都封装得很深,想改一个超时参数,得顺着继承链找三四层。那一刻我意识到,如果我不懂Agent内部到底怎么运转的,我就永远只能“会调用”,没法“懂原理”,更没法在出问题的时候快速定位。
所以我决定,从零手写一个最小可用的AI Agent。不依赖任何Agent框架,只用最基础的模型API和Python标准库。目标很简单:能理解Agent的核心循环,能自己控制工具调用、状态管理和错误处理,能在这个基础上做并发优化。
1.2 这个Agent到底能做什么
我手写的这个Agent,核心能力是:接收一个自然语言任务描述,自动拆解成多个步骤,每一步决定是直接回答还是调用工具,调用工具后根据结果继续推理,直到任务完成或达到最大步数。它支持的工具包括:本地文件读写、执行Python代码片段、调用外部HTTP接口、查询SQLite数据库。整个Agent的核心循环大约200行代码,加上工具定义和配置,总共不到500行。
它解决的问题是:让你彻底看清Agent的“思考-行动-观察”循环是怎么运转的。适合谁呢?适合已经用过LangChain、Spring AI或者Coze这类平台,但总觉得“黑盒”太多、想自己掌控每一个环节的开发者。也适合零基础但有一定Python基础的人,因为我会把每一步的原理和代码都拆开讲。
1.3 为什么选择“手写”而不是用框架
这里我得说清楚,我不是反对用框架。在生产环境里,LangGraph、Spring AI这些框架能帮你省掉大量重复劳动。但学习阶段,手写一遍的价值在于:你会被迫面对所有框架帮你隐藏的问题。比如,模型返回的tool_calls字段格式是什么?如果模型同时返回多个工具调用怎么办?工具执行抛异常了,怎么把错误信息喂回给模型让它重新决策?这些细节,你不手写一遍,永远不知道。
另一个原因是,手写之后你再去看框架源码,会有一种“原来如此”的感觉。你知道它在哪里做了重试,在哪里做了状态压缩,在哪里做了并发控制。这种理解,是单纯看文档得不到的。
2. 手写Agent前必须搞清楚的几个核心概念
2.1 Agent和普通LLM调用的本质区别
普通LLM调用是“一问一答”:你给一个prompt,模型返回一个文本。Agent调用是“多轮决策”:模型不仅返回文本,还可能返回一个“我要调用某个工具”的指令,你执行完工具后,把结果再喂给模型,模型继续决策,直到它认为任务完成。
这个区别听起来简单,但实现起来涉及几个关键问题:第一,你怎么让模型知道有哪些工具可用?第二,模型返回的工具调用请求是什么格式?第三,工具执行结果怎么拼回对话历史?第四,怎么防止模型陷入死循环?这四个问题,就是手写Agent要解决的核心。
2.2 工具调用协议:模型怎么“告诉”你要用什么工具
目前主流模型(如OpenAI系列、通义千问系列)都支持function calling或tool calling。你需要在请求里传入一个tools数组,每个工具包含name、description和parameters(JSON Schema格式)。模型如果决定调用工具,会在返回的message里带上tool_calls字段,里面包含工具名和参数。
这里有个坑:不同模型对tool_calls的返回格式略有差异。有的返回的是JSON字符串,有的直接返回对象。你需要根据你用的模型API文档做适配。我一开始用的是一个国产模型,它的tool_calls字段里arguments是字符串,需要自己json.loads;后来换了一个模型,arguments直接是dict。这个适配层必须自己写。
2.3 状态管理:对话历史怎么存
Agent的对话历史不是简单的user/assistant交替。它包含:system prompt、user的初始任务、assistant的tool_calls消息、tool的执行结果消息、assistant的最终回答。这些消息按顺序拼成一个列表,每次调用模型时整个传进去。
这里的关键是:tool消息必须紧跟在对应的assistant tool_calls消息之后,否则模型会报错。我踩过的坑是,有一次工具执行超时,我直接跳过了tool消息,结果下一次调用模型时格式校验失败。后来我改成:即使工具执行失败,也要生成一个tool消息,内容里写明错误信息,这样模型能根据错误信息决定下一步。
2.4 循环控制:怎么防止Agent“鬼打墙”
模型有时候会反复调用同一个工具,或者在一个步骤上卡住。你需要设置最大循环次数(比如10步),超过就强制终止并返回当前结果。另外,你可以在system prompt里明确告诉模型:“如果某个工具连续两次返回相同结果,请尝试其他方法或直接给出答案。”这个提示词技巧很管用。
还有一个技巧是:给每个工具调用加一个唯一ID,在对话历史里记录哪些工具已经调用过、结果是什么。如果模型再次请求调用相同工具且参数相同,你可以直接返回缓存结果,而不是重新执行。这能显著减少不必要的工具调用。
3. 从零搭建Agent核心循环的完整实操
3.1 环境准备与依赖选择
我用的Python 3.10,依赖只有三个:openai(用于调用模型API)、requests(用于HTTP工具)、sqlite3(标准库)。不需要LangChain,不需要任何Agent框架。如果你用的是其他模型,把openai替换成对应的SDK即可。
安装命令很简单:
pip install openai requests这里我特意选了openai这个包,因为它兼容很多国产模型的API格式。你只需要改base_url和api_key就行。我实测下来,通义千问、DeepSeek、Moonshot的API都能用这个包调通。
3.2 定义工具:从最简单的文件读写开始
工具的定义分两部分:一是给模型看的JSON Schema描述,二是实际执行的Python函数。我先定义两个最基础的工具:read_file和write_file。
tools_schema = [ { "type": "function", "function": { "name": "read_file", "description": "读取指定路径的文件内容", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"} }, "required": ["path"] } } }, { "type": "function", "function": { "name": "write_file", "description": "将内容写入指定路径的文件", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"}, "content": {"type": "string", "description": "要写入的内容"} }, "required": ["path", "content"] } } } ]对应的执行函数:
def read_file(path): with open(path, 'r', encoding='utf-8') as f: return f.read() def write_file(path, content): with open(path, 'w', encoding='utf-8') as f: f.write(content) return f"已写入{len(content)}字符到{path}"注意:description字段非常重要。模型就是靠这个描述来判断什么时候该用哪个工具。我试过把description写得很简略,结果模型经常该读文件的时候去写文件。后来我把描述写详细,比如“读取指定路径的文件内容,如果文件不存在会抛出异常”,准确率明显提升。
3.3 核心循环:思考-行动-观察
这是整个Agent的心脏。我用一个while循环实现,伪代码如下:
messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": task} ] for step in range(MAX_STEPS): response = client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=tools_schema, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if not msg.tool_calls: # 模型没有调用工具,直接返回答案 return msg.content for tool_call in msg.tool_calls: func_name = tool_call.function.name args = json.loads(tool_call.function.arguments) try: result = TOOL_MAP[func_name](**args) except Exception as e: result = f"工具执行失败:{str(e)}" messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) return "达到最大步数,任务未完成"这段代码有几个关键点。第一,msg对象直接append到messages里,不要手动构造,因为不同SDK返回的msg对象结构不一样,直接append最安全。第二,tool消息必须带tool_call_id,且必须紧跟在assistant消息之后。第三,工具执行失败不要抛异常中断循环,而是把错误信息作为tool结果返回,让模型自己决定怎么办。
3.4 系统提示词的设计技巧
SYSTEM_PROMPT我改了十几版,最后稳定下来的版本包含这几层意思:第一,明确Agent的身份和任务范围;第二,列出所有可用工具及其使用场景;第三,规定输出格式,比如“如果任务完成,直接给出最终答案,不要调用工具”;第四,设置行为边界,比如“不要编造工具执行结果”“如果工具返回错误,尝试其他方法或如实告知用户”。
我实测发现,提示词里加一句“你最多只能执行10步,请高效利用每一步”能显著减少无效循环。另外,把工具列表在提示词里再列一遍(即使tools_schema里已经有了),模型的选择准确率会更高。这可能是因为模型在生成时同时参考了system prompt和tools字段。
4. 进阶:让Agent真正“能干活”的几个关键改造
4.1 加入代码执行工具:让Agent能跑Python
文件读写只是基础,真正让Agent强大的是能执行代码。我加了一个run_python工具,接收一段Python代码字符串,在子进程里执行,返回stdout和stderr。
import subprocess def run_python(code): result = subprocess.run( ['python', '-c', code], capture_output=True, text=True, timeout=30 ) output = result.stdout if result.stderr: output += "\n[stderr]\n" + result.stderr return output or "执行成功,无输出"这里必须加timeout,否则模型写了一个死循环,你的Agent就卡死了。我设的是30秒,你可以根据任务复杂度调整。另外,子进程执行比exec()安全,因为exec()会污染当前进程的命名空间,而且没法限制执行时间。
4.2 工具路由与并发控制
当工具数量变多时,你需要一个工具注册表。我用一个字典做映射:
TOOL_MAP = { "read_file": read_file, "write_file": write_file, "run_python": run_python, "query_db": query_db, "http_get": http_get }并发控制方面,如果你的Agent要同时处理多个任务,最简单的方案是用线程池。每个任务独立维护自己的messages列表,互不干扰。但要注意,模型API通常有速率限制,你需要加一个信号量或令牌桶来控制并发数。我实测下来,对于大多数个人项目,并发数设3-5就够了,再高反而容易触发限流。
4.3 错误重试与降级策略
模型API调用可能失败(网络超时、限流、服务端错误)。我封装了一个带指数退避的重试函数:
import time def call_model_with_retry(messages, max_retries=3): for attempt in range(max_retries): try: return client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=tools_schema, tool_choice="auto" ) except Exception as e: if attempt == max_retries - 1: raise wait = 2 ** attempt time.sleep(wait)降级策略是:如果模型连续三次调用失败,Agent返回“服务暂时不可用,请稍后重试”,而不是一直卡在那里。这个逻辑在循环里加一个计数器就行。
4.4 对话历史压缩:防止token爆炸
多轮工具调用后,messages列表会变得很长,token消耗急剧上升。我的做法是:当messages长度超过阈值(比如20条)时,把最早的几条工具调用和结果合并成一条摘要消息。摘要由模型生成,内容大概是“之前执行了A、B、C操作,结果分别是...”。
这个压缩策略要小心:不能把system prompt和最近的几条消息压掉,否则模型会丢失上下文。我一般保留system prompt、最近5条消息,中间的做摘要。实测下来,这样能把token消耗降低40%左右,同时不影响任务完成率。
5. 常见问题与排查技巧实录
5.1 模型不调用工具,直接编答案
这是最常见的问题。原因通常是:工具描述不够清晰,或者system prompt没有强调“必须使用工具获取信息”。我的解决方法是:在system prompt里加一句“对于需要实时数据或文件内容的任务,你必须调用相应工具,禁止凭记忆回答”。另外,把tool_choice设为"auto"而不是"none",给模型选择权。
如果还是不行,可以试试在用户消息里加一句“请使用工具完成这个任务”。这个技巧在任务描述模糊时特别管用。
5.2 工具调用参数格式错误
模型返回的arguments有时候不是合法JSON,比如多了逗号、少了引号。我加了一个容错解析:
def safe_parse_args(args_str): try: return json.loads(args_str) except json.JSONDecodeError: # 尝试修复常见错误 fixed = args_str.replace("'", '"').rstrip(',') return json.loads(fixed)如果修复失败,就把原始字符串作为参数传给工具,让工具自己处理。或者返回一个错误消息给模型,让它重新生成参数。
5.3 Agent陷入无限循环
表现是模型反复调用同一个工具,或者在不同工具之间来回跳转。我的排查步骤是:第一,看messages里是否有重复的tool_calls;第二,检查工具返回结果是否为空或异常;第三,在system prompt里加“如果某个操作连续两次没有进展,请停止并给出当前最佳答案”。
还有一个隐藏原因是:工具返回的结果太长,模型每次都要重新处理,导致决策变慢。这时候可以在工具函数里做截断,比如只返回前2000字符。
5.4 并发场景下的状态污染
如果你用多线程跑多个Agent实例,一定要确保每个实例的messages列表是独立的。我踩过的坑是:把messages定义成了类变量,结果多个线程共享同一个列表,对话历史串得一塌糊涂。改成实例变量或者用threading.local()就解决了。
另外,如果你用的模型API支持并发请求,注意控制QPS。我一般用信号量限制同时进行的请求数,避免触发限流。
5.5 常见问题速查表
| 问题现象 | 可能原因 | 解决方法 |
|---|---|---|
| 模型不调用工具 | 工具描述不清、提示词未强调 | 完善description,加“必须使用工具”提示 |
| 参数解析失败 | 模型返回非标准JSON | 加容错解析,失败时返回错误让模型重试 |
| 无限循环 | 工具结果无进展、提示词缺边界 | 加最大步数,提示词加“连续两次无进展则停止” |
| 并发状态污染 | 共享了messages列表 | 每个实例独立维护messages |
| token消耗过快 | 对话历史过长 | 定期压缩历史,保留最近5条 |
| 工具执行超时 | 代码死循环、网络慢 | 加timeout,超时返回错误信息 |
6. 从“会调用”到“懂原理”的进阶路径
6.1 第一阶段:跑通最小闭环
先不要追求功能完整,就实现一个能读文件、能写文件的Agent。任务设定为“读取a.txt的内容,转成大写后写入b.txt”。这个任务足够简单,能让你快速验证核心循环是否跑通。我当初在这个阶段卡了两天,原因是tool消息的格式不对,模型一直报错。后来对照API文档逐字段检查,才发现tool_call_id必须和assistant消息里的id完全一致。
6.2 第二阶段:加入代码执行和错误处理
跑通闭环后,加入run_python工具,然后故意让模型执行一段会报错的代码,观察它怎么处理错误。这个阶段你会深刻理解“观察”环节的重要性:工具返回的错误信息,就是模型下一步决策的依据。如果错误信息写得太模糊(比如只返回“执行失败”),模型就不知道该怎么改。所以工具函数里要尽量返回详细的错误堆栈。
6.3 第三阶段:优化提示词和工具描述
这个阶段是提升Agent“智商”的关键。你需要反复调整system prompt,观察模型在不同提示词下的行为差异。我建议你建一个测试集,包含10-20个典型任务,每次改完提示词就跑一遍,统计任务完成率和平均步数。我自己的测试集里有“读取CSV文件并计算某列平均值”“调用HTTP接口获取天气并写入文件”“执行一段排序代码并验证结果”等任务。
6.4 第四阶段:并发与生产化改造
当你对单任务Agent足够熟悉后,就可以考虑并发化了。最简单的方案是用FastAPI包一层HTTP接口,每个请求启动一个Agent实例。然后用uvicorn多worker模式跑起来。但要注意,模型API的速率限制是瓶颈,你需要加请求队列和限流。我实测下来,单机跑3-5个并发任务比较稳,再高就需要考虑用多个API key轮询或者换更高配额的模型。
6.5 我个人的学习路线建议
如果你是完全零基础,我建议先花两天学Python基础(变量、函数、字典、异常处理),然后直接上手这个手写Agent项目。不要先去学LangChain,因为框架的抽象层会掩盖太多细节。等你手写跑通了,再去看LangGraph的源码,你会发现它的StateGraph本质上就是你写的那个while循环加了一个状态机。这时候你再决定是用框架还是继续手写,心里就有底了。
最后分享一个我踩过的最大的坑:不要试图让Agent一次完成太复杂的任务。我一开始让Agent“分析一个Git仓库的代码质量并生成报告”,结果它调用了十几次工具都没完成,token烧了一大半。后来我把任务拆成“先列出所有Python文件”“再逐个分析”“最后汇总”,每个子任务单独跑Agent,成功率高得多。Agent不是魔法,它就是一个能调用工具的循环,你给它清晰的边界,它才能好好干活。