1. 为什么我要从零手写一个AI Agent1.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和parametersJSON 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, encodingutf-8) as f: return f.read() def write_file(path, content): with open(path, w, encodingutf-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( modelMODEL_NAME, messagesmessages, toolstools_schema, tool_choiceauto ) 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_outputTrue, textTrue, timeout30 ) 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_retries3): for attempt in range(max_retries): try: return client.chat.completions.create( modelMODEL_NAME, messagesmessages, toolstools_schema, tool_choiceauto ) 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列表每个实例独立维护messagestoken消耗过快对话历史过长定期压缩历史保留最近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不是魔法它就是一个能调用工具的循环你给它清晰的边界它才能好好干活。