做了几年大模型应用开发有一件事始终让我心里不踏实别人问我你项目里的Agent到底是怎么跑起来的我能给出的最深答案只有用了LangChain创建AgentExecutor然后run一下。功能确实能跑demo也确实精彩但一旦线上出问题——模型在死循环、工具反复报错、上下文直接撑爆——框架一层套一层的抽象就把我死死卡在中间。这种状态就是大家常说的调包侠。这个标题里最有意思的组合是AI Agent配手写。很多人觉得手写Agent是不自量力另一拨人觉得无非就是把LangChain的底层代码再抄一遍。我的亲身体验是都不对。手写一个最小可用的AI Agent系统核心循环大概就两百行代码但这两百行写下来你对Agent的认知会从用法升级到原理。这中间的价值怎么强调都不过分。下面这篇文章就按我从零手写的完整路径来展开包含模型层封装、工具注册、ReAct主循环、记忆管理以及大量翻车现场和扩展思路。适合谁读已经能用LangChain或Dify这类框架、但总觉得知其然不知其所以然的人想给团队做内部Agent、又不愿意背一堆框架依赖的人以及单纯想从调包侠迈向下一个阶段的开发者和学生。1. 为什么我劝你别一上来就造轮子但一定要拆一次轮子1.1 调包侠不是贬义词它是每个人的必经阶段先把这个称呼放平看。调包不是耻辱框架能流行本身就说明它对绝大多数场景是高效的。LangChain帮你封装了模型调用、Prompt模板、工具加载、记忆模块你几行代码就能搭出一个有模有样的Agent这个效率是真的香。我自己最早做Agent原型时也是从调包开始的。问题出在永久停留在调包状态。框架是黑盒黑盒在顺利的时候无所谓但一旦出问题你要跨过三层抽象去定位第一层是你的业务代码第二层是框架的state、node、callback机制第三层才是模型真正的输入输出。我第一次从LangChain的Agent架构迁移到一个更细的图编排方案时被一个node状态序列化的问题卡了整整两天最后不得不翻了框架源码才搞明白。那一刻我就意识到我不懂Agent本身我只懂Agent的API。这也是为什么社区里一直有手写传统。你看那些被反复讨论的热门话题手写Transformer、MNIST手写数字识别、手写Spring、甚至Tomoe这种开源的手写识别项目本质上都是在做同一件事——把别人封装好的黑盒重新拆开用最小的代码量验证自己是否真正理解了原理。AI Agent现在完全值得进这个列表。反过来说为什么很少人愿意手写JS因为手写一个V8级别的JS引擎是几千人年的工程没人会拿这个当学习路径。但手写一个能跑通的简易解释器是完全可行且必要的。Agent也一样别去复刻LangChain全家桶但一定要能写出一版两百行的ReAct循环。1.2 手写一遍具体能解锁哪些认知我把自己手写前后的变化列出来你会发现这不是多会一个技能而是认知底座变了真正理解Agent的定义Agent不是某个神秘的class它本质上就是一个带工具的while循环。这个认知会捅破大量玄学。看懂框架参数的来龙去脉以前LangChain里的max_iterations、return_intermediate_steps、early_stopping_method对我只是照着文档填的数字。手写一遍之后这些参数的语义全部自然浮现因为你已经在自己的循环里亲手实现过它们。出问题时知道该看哪一层模型没按格式输出、工具抛异常、上下文超长、循环不收敛这些问题发生在不同的环节处理方式完全不同。没有手写经验时你只会对着一条混合了三个环节的报错日志干瞪眼。能自己加框架没提供的能力比如针对内部业务定制的记忆策略、特殊的输出解析器、带人工确认的危险工具操作。框架是通用件通用就意味着很多公司内部场景它管不到。1.3 什么时候该自己写、什么时候该用框架按我的经验这不是二选一而是分场景决策。你可以直接对照这张表场景我的建议理由学习、教学、验证原理手写两百行代码认知收益最大内部小工具、简单自动化手写或轻量框架依赖少、可控、好维护生产级复杂编排并行、状态机、多Agent使用成熟框架或自研编排中间件需要工程健壮性、可观测性、生态组件高吞吐异步、多租户、大规模部署手写核心框架思路混合需要精确控制资源和成本另外提一句如果你对Rust有兴趣手写一个基于Rust的Agent是很棒的进阶项目但我强烈建议先用Python把整套逻辑跑通。语言的难度不该和学习概念的难度混在一起否则你会分不清自己是被Agent难住还是被所有权检查难住。2. 先拆清楚Agent的骨架工具、循环、记忆三位一体2.1 一个Agent的本质是一个带工具的while循环现在市面上讲Agent架构的术语很多ReAct、Plan-and-Execute、Reflexion、多Agent协作。不管哪家云厂商的白皮书核心模块也就那么几块规划Planning、记忆Memory、工具Tool、执行Execution。但落到代码层面最基础、也最值得手写的就是这个ReAct循环。ReAct是Reasoning和Acting的组合意思是先思考再行动观察结果继续思考。把它翻译成程序逻辑就是下面这个循环把系统提示、用户任务、历史对话、工具描述打包成一组messages。调用大模型拿到一个结构化决策要么是我想清楚了最终答案是X要么是我要调用工具T参数是Y。如果是后者就执行工具把执行结果作为observation追加进上下文。回到第1步直到模型给出最终答案或者达到最大迭代次数。你看整个过程没有魔法。它就像一个人查资料的过程看资料、做判断、翻工具书、记录结果、再判断。每个环节都对应代码里的一块逻辑。理解了这个循环你就知道为什么Agent会比普通Chatbot聪明——不是模型变聪明了而是它多了一个能动手查证的权利。顺带解释一个高频问题AI Agent token是什么意思Token是模型处理文本的最小单位可以是半个词、一个词也可以是一个汉字的一部分。每次循环都会把tool返回结果追加进messagestoken数就会不断增长。这个点看着基础却和成本、上下文溢出、性能都直接挂钩后面第4章会专门展开。2.2 工具层Agent与普通Chatbot的分界线Chatbot只会输出文字Agent能动手做事。这个动手靠的就是工具层。一个工具本质上就是一个普通函数加上一份使用说明书。说明书包含四部分工具名、功能描述、参数Schema、执行函数。描述写得越清楚模型越不容易用错。举个例子一个发消息工具的描述如果只写发消息模型可能不知道该传哪个接收方如果写清楚给指定用户发送一条即时消息user_id是接收方IDcontent是要发送的文本模型一读就懂。现在主流大模型大多支持原生function calling。你不再需要让模型输出Action: xxx, Action Input: xxx这种脆弱的文本格式而是直接把工具的JSON Schema放进请求的tools参数里模型就会返回一个结构化的工具调用对象里面包含工具名和参数。这是Agent系统的地基之一。很多老项目还在用文本解析方式遇到格式漂移就崩溃原因就是没走原生function calling这条更稳的路。工具能做的事情非常广查数据库、调日历、执行代码、调用OCR识别手写文字、发消息、操作浏览器。关键是这些工具不需要多高级它们就是你业务系统里已有的函数。Agent的价值就是让模型学会在正确的时候调用这些函数。2.3 记忆层上下文、状态与外部存储要分清楚提到记忆新手最容易混成一大锅。至少应该分三层第一层是上下文窗口内的工作记忆。就是当前这一轮messages包括系统提示、用户任务、模型输出、工具observation。它的容量受模型的上下文长度限制也是最容易出现成本爆炸的地方。第二层是会话记忆。比如用户和Agent聊了二十轮不可能每轮都原封不动堆进上下文。要按session持久化对话记录并在进入模型之前做裁剪或摘要。第三层是长期记忆。用户的偏好、历史事实、业务知识这些要存到向量库、KV存储或者数据库里在需要时检索出来注入上下文。没有长期记忆的Agent每次对话都是失忆的没有上下文管理的Agent聊不了几轮就会因为token超限而崩溃。这三层对应到系统设计上就是不同的存储方案和不同的代码模块。手写Agent时建议先只做第一层再把第二层加上第三层按需引入。一上来就上全套多半会把自己绕晕。3. 从零手写最小可用Agent核心代码拆解这一章是全文最硬核的部分。我会按四步走每一步都会给出能直接跑的代码。还是那句话手写不是造一个生产级框架而是造一个最小闭环能理解任务、能调用工具、能基于结果继续思考。3.1 第一步封装模型层让换模型变成改一个配置很多人以为手写Agent要从HTTP请求开始包其实没必要。OpenAI SDK本质只是一个网络客户端用它不算调包算使用传输协议。真正该亲手写的是模型之上的抽象层。import os from openai import OpenAI class LLMClient: def __init__(self, model: str None): self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), # 兼容任意OpenAI风格服务 ) self.model model or os.getenv(LLM_MODEL, gpt-4o-mini) def chat(self, messages: list[dict], tools: list[dict] None) - dict: kwargs {model: self.model, messages: messages} if tools: kwargs[tools] tools resp self.client.chat.completions.create(**kwargs) msg resp.choices[0].message # 把tool_calls统一转成dict方便后续解析和调试 tool_calls [] if getattr(msg, tool_calls, None): for tc in msg.tool_calls: tool_calls.append({ id: tc.id, name: tc.function.name, arguments: tc.function.arguments, }) return { content: msg.content, tool_calls: tool_calls, finish_reason: resp.choices[0].finish_reason, }这一层做了三件事统一了模型入口、支持了tools参数、把输出结构化。以后不管后面接的是哪家模型只要它兼容OpenAI风格接口换模型就只改环境变量Agent主体代码完全不动。这就是面向接口编程在AI应用里的最小实践。3.2 第二步工具注册表与参数Schema接下来定义工具类。不要小看这一步工具的Schema写不好后面模型会频繁幻想出根本不存在的参数。import json from typing import Any, Callable, Dict, List class Tool: def __init__( self, name: str, description: str, parameters: Dict[str, Any], func: Callable[..., Any], ): self.name name self.description description self.parameters parameters self.func func def to_openai_schema(self) - Dict[str, Any]: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, }然后定义两个示例工具。一个查询天气一个给用户发送消息后者就是Agent自动发消息类场景的最小雏形def get_weather(city: str) - str: # 真实项目里这里会调天气API演示直接返回模拟值 return f{city}今天晴气温18-26摄氏度。 def send_message(user_id: str, content: str) - str: # 真实项目里这里会调消息网关 return f已向 {user_id} 发送{content} tools [ Tool( nameget_weather, description查询指定城市的当前天气参数city为城市名, parameters{ type: object, properties: { city: {type: string, description: 城市名例如北京、上海} }, required: [city], }, funcget_weather, ), Tool( namesend_message, description给指定用户发送一条即时消息, parameters{ type: object, properties: { user_id: {type: string, description: 接收方用户ID}, content: {type: string, description: 消息正文文本}, }, required: [user_id, content], }, funcsend_message, ), ]注意参数的description这是模型理解工具的说明书。说明书越具体模型越不会传错。比如在send_message里如果不写清content是消息正文模型可能把一句废话塞进去或者干脆漏传。3.3 第三步ReAct主循环——Agent的心脏下面这段就是造轮子的核心时刻。整个Agent逻辑不过几十行class Agent: SYSTEM_PROMPT ( 你是一个通过调用工具来完成任务的助手。 如果工具返回了结果请结合结果继续推理 当任务完成时直接给出最终答案。 ) def __init__(self, llm: LLMClient, tools: List[Tool], max_steps: int 10): self.llm llm self.tools {t.name: t for t in tools} self.max_steps max_steps def run(self, task: str) - str: messages [{role: system, content: self.SYSTEM_PROMPT}] messages.append({role: user, content: task}) for _ in range(self.max_steps): resp self.llm.chat( messages, tools[t.to_openai_schema() for t in self.tools.values()], ) if resp[tool_calls]: # 1. 先把assistant的工具调用包进消息历史 assistant_msg { role: assistant, content: resp[content] or , tool_calls: [ { id: tc[id], type: function, function: { name: tc[name], arguments: tc[arguments], }, } for tc in resp[tool_calls] ], } messages.append(assistant_msg) # 2. 逐个执行工具结果作为tool角色消息追加上去 for tc in resp[tool_calls]: tool self.tools.get(tc[name]) if not tool: output f未找到工具{tc[name]} else: try: args json.loads(tc[arguments]) result tool.func(**args) output json.dumps(result, ensure_asciiFalse, defaultstr) except Exception as e: output f工具执行失败{type(e).__name__}: {e} messages.append({ role: tool, tool_call_id: tc[id], content: output[:4000], # 防止单条observation过长 }) continue # 3. 没有工具调用说明模型给出了最终答案 return resp[content] or return 达到最大步数未得到最终答案。这段代码里有几个容易忽略的关键点assistant消息必须先入栈在OpenAI的function calling协议里tool角色消息必须通过tool_call_id关联到对应的assistant工具调用顺序错了API直接报错。工具执行要在try里任何工具都可能抛异常不能让异常炸掉整个Agent进程。把错误转成字符串返回给模型模型才有机会自行修正。observation要截断工具返回结果可能很长先截到4000字符。省token也防止上下文被单个结果污染。跑一个简单的例子agent Agent(llmLLMClient(), toolstools) print(agent.run(上海今天天气怎么样顺便提醒一下用户u_10086明天记得带伞))模型大概率会先调用get_weather拿到结果后再调用send_message最后给出已完成之类的收尾。这就是一个完整的最小Agent闭环。3.4 第四步加上最朴素的记忆管理第三步跑通之后你会立刻遇到一个问题任务复杂一点多轮工具调用后messages会越来越长。这时就要引入最简单的记忆管理记录历史、统计token、按预算裁剪。import tiktoken _enc tiktoken.get_encoding(cl100k_base) def count_tokens(messages: list[dict]) - int: return sum(len(_enc.encode(str(m))) for m in messages) def trim_messages(messages: list[dict], budget: int 12000) - list[dict]: if count_tokens(messages) budget: return messages # 保留system、最近两轮对话丢弃中间最旧的部分 head messages[:1] tail messages[-4:] if len(messages) 4 else messages[1:] trimmed head tail trimmed.append({ role: system, content: 注意较早的对话已被截断请基于剩余信息继续。, }) return trimmed这个策略当然很粗糙但麻雀虽小五脏俱全它让你理解记忆管理的本质是预算管理。更高级的版本无非是超出预算时优先保留什么、用摘要压缩旧消息、把关键事实抽出来存长期存储。核心判断逻辑都是同一套。4. 最容易翻车的四个细节解析、上下文、死循环、工具异常我自己在从能跑到好用的过程中踩过的坑基本都集中在这四个地方。每个都值得单独开一节。4.1 模型返回的JSON别指望永远是干净的JSON原生function calling能解决大部分格式问题但遇到弱模型、或者某些不支持function calling的模型时你仍然需要从模型输出里解析JSON。这时会遇到各种奇幻输出前面加一句好的我来调用工具用代码块包裹甚至JSON里混注释。我当时踩坑之后写了一个通用的容错解析函数import json import re def extract_json(text: str) - dict: text text.strip() # 去掉markdown代码块标记 if text.startswith(): text re.sub(r^(?:json)?, , text).rstrip().strip() # 直接截出第一个{到最后一个} start, end text.find({), text.rfind(}) if start ! -1 and end ! -1: return json.loads(text[start:end 1]) raise ValueError(输出中找不到JSON对象)经验教训是解析越健壮模型越自由。如果你能用原生function calling优先用如果必须文本解析就上容错解析再不行就做二次解析让模型自己修正上一次的JSON。4.2 Token预算上下文不是无限大的也没你想的那么小很多人不看token直到账单出来才清醒。我来算一笔账假设系统提示500 token10个工具的Schema合计3000 token每轮模型输出500 token每次observation平均800 token。10轮跑下来仅新增部分就有13000 token加上system和工具Schema一次任务轻松到20000 token。如果做多Agent编排、长会话撑爆128k上下文是迟早的事。所以实际开发中我通常会在Agent里做三件事给工具Schema瘦身只保留模型真正需要知道的字段详细说明放到工具内部文档里。给observation限长上面代码里的[:4000]就是干这个的。定期压缩历史当上下文超过预算时触发trim_messages或者把旧对话总结成一句摘要。不要觉得上下文还够不用管上下文是钱也是延迟。省token既是省成本也是保性能。4.3 死循环模型不停地调用同一个工具这是Agent最经典的失控姿势。任务描述模糊、工具返回信息没帮助、模型反复尝试同一方案就很容易陷入循环。只靠max_steps兜底虽然不会死进程但用户体验很糟。我用的一个有效手段是重复调用检测维护一个最近N步的调用指纹发现同一工具、同一参数连续出现多次就往messages里注入一条系统提示明确让模型换策略。call_fingerprints {} # 在循环内、执行工具前加上这段 fingerprint f{tc[name]}:{tc[arguments]} call_fingerprints[fingerprint] call_fingerprints.get(fingerprint, 0) 1 if call_fingerprints[fingerprint] 3: messages.append({ role: system, content: 注意你已经在重复调用同一个工具且没有进展请改变策略或者直接给最终答案。, })这个提示会作为后续模型的输入模型看到后通常会改变方案。比硬切断更平滑也比傻等到max_steps更高效。4.4 工具异常不要中断循环把错误喂回模型工具执行必然会有失败网络超时、参数格式错、权限不足、上游接口变更。我第一次写Agent时工具一报错就让整个循环break结果就是用户只看到执行失败Agent完全没机会自我修复。后来改成把异常转成observation喂回给模型。比如刚才代码里的工具执行失败{type(e).__name__}: {e}。模型看到这个信息通常会自己调整参数重新调用或者换一个工具。这就像人做菜时发现酱油瓶空了不会直接放弃整桌菜而是会换个调味料或者下单补货。但这里有个安全红线对危险操作不要无条件让模型自己重试。比如删除数据、转账、对外发消息这类工具最好加一个requires_confirmation标记触发时回到人工确认流程。自动化不是目的可控才是。5. 从能跑到好用接业务、做编排、上日志把最小闭环跑通之后你就要开始考虑真实场景了。这一章讲三件我实际项目里最常用到的事多Agent协作、接入业务系统、可观测性。5.1 单Agent之外两种最实用的多Agent协作模式我见过很多团队一上来就追求复杂的多Agent框架结果连单Agent的稳定性都没解决。我的建议是先掌握两种最基础的模式。模式一监督者模式Supervisor。一个主 Agent 负责拆解任务把子任务分发给不同的子 Agent最后汇总结果。实现起来有个很巧妙的技巧把子Agent封装成一个Tool。也就是说子Agent对外表现为一个能完成某类任务的工具主Agent通过function calling调用它。这样一来你不需要新的编排框架复用前面的工具注册表就行。模式二流水线模式Pipeline。多个Agent按顺序处理一个任务的不同阶段比如第一个Agent负责提取信息第二个Agent负责生成方案第三个Agent负责执行。每个Agent只关注一件事职责单一调试也容易。这两种模式配合单Agent的ReAct循环已经能覆盖大部分业务场景。再往上加并行调度、状态机、动态拓扑那是框架该做的事别急于手写。5.2 把Agent接进业务系统以Django为例Agent本质上就是一个函数输入任务字符串输出结果字符串。所以接入业务系统非常直接。以Django为例import json def agent_api(request): task request.POST.get(q, ) answer agent.run(task) return JsonResponse({answer: answer})生产环境当然不能这么粗暴。耗时较长的任务建议用Celery放到后台执行前端轮询任务状态如果需要流式输出就改造模型层把token流通过SSE推给前端。而Agent的工具完全可以是Django里的ORM查询、缓存操作、消息网关、内部REST服务。这也是我一直强调的Agent的落地能力不在Agent本身而在你能让模型碰到多少真实的业务函数。5.3 可观测性没有日志的Agent就是玄学调参Agent是多轮决策系统每一轮的状态都影响下一轮。如果连每一步模型想了什么、调了什么工具、花了多少token都不知道那调试就完全是猜谜。我会在循环里记录如下字段字段说明step第几轮input_messages进入本轮的消息列表或摘要、长度model_output模型输出内容tool_calls本轮发起的工具调用tool_results工具执行结果或错误信息token_count当前累计token数latency_ms本轮耗时每一轮落一条日志按request_id和session_id归档。看起来很简单但等你在线上遇到Agent为什么突然开始乱调工具的问题时这批日志会帮你把原因锁定到具体某一轮而不是对着完整对话记录从头猜。另外强烈建议统计成本。把每轮的prompt_tokens和completion_tokens加总按模型单价折算成费用。很多Agent跑飞的成本黑洞都是靠这个字段抓住的。6. 一点最后的体会造轮子的终点是知道自己该怎么选写到这里我想把最实在的体会留在最后。手写完这个最小Agent之后我在团队内部留了一个叫micro-agent的极简核心不到三百行。现在做任何Agent需求我都会先用它跑一个最小验证任务能不能被拆解、工具够不够用、模型表现稳不稳定。验证通过再决定要不要上成熟框架。这让我避开了很多一上来就上重型框架结果连问题都描述不清的尴尬。关于调包侠到造轮子的人我的理解不是以后再也不调包而是调包之前知道包里是什么换包的时候知道差别在哪拆包的时候知道哪些零件必须自己造。如果你也想走这条路我的建议非常具体不要一上来就手写一个巨大的平台先让两百行核心跑通一个能搜索、能查数据库、能发消息的Agent然后把这篇文章提到的四个翻车点挨个踩一遍。踩完了你对Agent的认知就完全不一样了。你写的每一行代码都会让你离调包侠远一点。等轮子原理在你心里不再是一团迷雾选择用别人的轮子还是自己造轮子都会变成清醒的决策。 
长河君 · 2026/10/25 13:02:50 · 阅读 1.2k 两百行代码手写AI Agent：拆解ReAct循环与工具调用原理
做了几年大模型应用开发有一件事始终让我心里不踏实别人问我你项目里的Agent到底是怎么跑起来的我能给出的最深答案只有用了LangChain创建AgentExecutor然后run一下。功能确实能跑demo也确实精彩但一旦线上出问题——模型在死循环、工具反复报错、上下文直接撑爆——框架一层套一层的抽象就把我死死卡在中间。这种状态就是大家常说的调包侠。这个标题里最有意思的组合是AI Agent配手写。很多人觉得手写Agent是不自量力另一拨人觉得无非就是把LangChain的底层代码再抄一遍。我的亲身体验是都不对。手写一个最小可用的AI Agent系统核心循环大概就两百行代码但这两百行写下来你对Agent的认知会从用法升级到原理。这中间的价值怎么强调都不过分。下面这篇文章就按我从零手写的完整路径来展开包含模型层封装、工具注册、ReAct主循环、记忆管理以及大量翻车现场和扩展思路。适合谁读已经能用LangChain或Dify这类框架、但总觉得知其然不知其所以然的人想给团队做内部Agent、又不愿意背一堆框架依赖的人以及单纯想从调包侠迈向下一个阶段的开发者和学生。1. 为什么我劝你别一上来就造轮子但一定要拆一次轮子1.1 调包侠不是贬义词它是每个人的必经阶段先把这个称呼放平看。调包不是耻辱框架能流行本身就说明它对绝大多数场景是高效的。LangChain帮你封装了模型调用、Prompt模板、工具加载、记忆模块你几行代码就能搭出一个有模有样的Agent这个效率是真的香。我自己最早做Agent原型时也是从调包开始的。问题出在永久停留在调包状态。框架是黑盒黑盒在顺利的时候无所谓但一旦出问题你要跨过三层抽象去定位第一层是你的业务代码第二层是框架的state、node、callback机制第三层才是模型真正的输入输出。我第一次从LangChain的Agent架构迁移到一个更细的图编排方案时被一个node状态序列化的问题卡了整整两天最后不得不翻了框架源码才搞明白。那一刻我就意识到我不懂Agent本身我只懂Agent的API。这也是为什么社区里一直有手写传统。你看那些被反复讨论的热门话题手写Transformer、MNIST手写数字识别、手写Spring、甚至Tomoe这种开源的手写识别项目本质上都是在做同一件事——把别人封装好的黑盒重新拆开用最小的代码量验证自己是否真正理解了原理。AI Agent现在完全值得进这个列表。反过来说为什么很少人愿意手写JS因为手写一个V8级别的JS引擎是几千人年的工程没人会拿这个当学习路径。但手写一个能跑通的简易解释器是完全可行且必要的。Agent也一样别去复刻LangChain全家桶但一定要能写出一版两百行的ReAct循环。1.2 手写一遍具体能解锁哪些认知我把自己手写前后的变化列出来你会发现这不是多会一个技能而是认知底座变了真正理解Agent的定义Agent不是某个神秘的class它本质上就是一个带工具的while循环。这个认知会捅破大量玄学。看懂框架参数的来龙去脉以前LangChain里的max_iterations、return_intermediate_steps、early_stopping_method对我只是照着文档填的数字。手写一遍之后这些参数的语义全部自然浮现因为你已经在自己的循环里亲手实现过它们。出问题时知道该看哪一层模型没按格式输出、工具抛异常、上下文超长、循环不收敛这些问题发生在不同的环节处理方式完全不同。没有手写经验时你只会对着一条混合了三个环节的报错日志干瞪眼。能自己加框架没提供的能力比如针对内部业务定制的记忆策略、特殊的输出解析器、带人工确认的危险工具操作。框架是通用件通用就意味着很多公司内部场景它管不到。1.3 什么时候该自己写、什么时候该用框架按我的经验这不是二选一而是分场景决策。你可以直接对照这张表场景我的建议理由学习、教学、验证原理手写两百行代码认知收益最大内部小工具、简单自动化手写或轻量框架依赖少、可控、好维护生产级复杂编排并行、状态机、多Agent使用成熟框架或自研编排中间件需要工程健壮性、可观测性、生态组件高吞吐异步、多租户、大规模部署手写核心框架思路混合需要精确控制资源和成本另外提一句如果你对Rust有兴趣手写一个基于Rust的Agent是很棒的进阶项目但我强烈建议先用Python把整套逻辑跑通。语言的难度不该和学习概念的难度混在一起否则你会分不清自己是被Agent难住还是被所有权检查难住。2. 先拆清楚Agent的骨架工具、循环、记忆三位一体2.1 一个Agent的本质是一个带工具的while循环现在市面上讲Agent架构的术语很多ReAct、Plan-and-Execute、Reflexion、多Agent协作。不管哪家云厂商的白皮书核心模块也就那么几块规划Planning、记忆Memory、工具Tool、执行Execution。但落到代码层面最基础、也最值得手写的就是这个ReAct循环。ReAct是Reasoning和Acting的组合意思是先思考再行动观察结果继续思考。把它翻译成程序逻辑就是下面这个循环把系统提示、用户任务、历史对话、工具描述打包成一组messages。调用大模型拿到一个结构化决策要么是我想清楚了最终答案是X要么是我要调用工具T参数是Y。如果是后者就执行工具把执行结果作为observation追加进上下文。回到第1步直到模型给出最终答案或者达到最大迭代次数。你看整个过程没有魔法。它就像一个人查资料的过程看资料、做判断、翻工具书、记录结果、再判断。每个环节都对应代码里的一块逻辑。理解了这个循环你就知道为什么Agent会比普通Chatbot聪明——不是模型变聪明了而是它多了一个能动手查证的权利。顺带解释一个高频问题AI Agent token是什么意思Token是模型处理文本的最小单位可以是半个词、一个词也可以是一个汉字的一部分。每次循环都会把tool返回结果追加进messagestoken数就会不断增长。这个点看着基础却和成本、上下文溢出、性能都直接挂钩后面第4章会专门展开。2.2 工具层Agent与普通Chatbot的分界线Chatbot只会输出文字Agent能动手做事。这个动手靠的就是工具层。一个工具本质上就是一个普通函数加上一份使用说明书。说明书包含四部分工具名、功能描述、参数Schema、执行函数。描述写得越清楚模型越不容易用错。举个例子一个发消息工具的描述如果只写发消息模型可能不知道该传哪个接收方如果写清楚给指定用户发送一条即时消息user_id是接收方IDcontent是要发送的文本模型一读就懂。现在主流大模型大多支持原生function calling。你不再需要让模型输出Action: xxx, Action Input: xxx这种脆弱的文本格式而是直接把工具的JSON Schema放进请求的tools参数里模型就会返回一个结构化的工具调用对象里面包含工具名和参数。这是Agent系统的地基之一。很多老项目还在用文本解析方式遇到格式漂移就崩溃原因就是没走原生function calling这条更稳的路。工具能做的事情非常广查数据库、调日历、执行代码、调用OCR识别手写文字、发消息、操作浏览器。关键是这些工具不需要多高级它们就是你业务系统里已有的函数。Agent的价值就是让模型学会在正确的时候调用这些函数。2.3 记忆层上下文、状态与外部存储要分清楚提到记忆新手最容易混成一大锅。至少应该分三层第一层是上下文窗口内的工作记忆。就是当前这一轮messages包括系统提示、用户任务、模型输出、工具observation。它的容量受模型的上下文长度限制也是最容易出现成本爆炸的地方。第二层是会话记忆。比如用户和Agent聊了二十轮不可能每轮都原封不动堆进上下文。要按session持久化对话记录并在进入模型之前做裁剪或摘要。第三层是长期记忆。用户的偏好、历史事实、业务知识这些要存到向量库、KV存储或者数据库里在需要时检索出来注入上下文。没有长期记忆的Agent每次对话都是失忆的没有上下文管理的Agent聊不了几轮就会因为token超限而崩溃。这三层对应到系统设计上就是不同的存储方案和不同的代码模块。手写Agent时建议先只做第一层再把第二层加上第三层按需引入。一上来就上全套多半会把自己绕晕。3. 从零手写最小可用Agent核心代码拆解这一章是全文最硬核的部分。我会按四步走每一步都会给出能直接跑的代码。还是那句话手写不是造一个生产级框架而是造一个最小闭环能理解任务、能调用工具、能基于结果继续思考。3.1 第一步封装模型层让换模型变成改一个配置很多人以为手写Agent要从HTTP请求开始包其实没必要。OpenAI SDK本质只是一个网络客户端用它不算调包算使用传输协议。真正该亲手写的是模型之上的抽象层。import os from openai import OpenAI class LLMClient: def __init__(self, model: str None): self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), # 兼容任意OpenAI风格服务 ) self.model model or os.getenv(LLM_MODEL, gpt-4o-mini) def chat(self, messages: list[dict], tools: list[dict] None) - dict: kwargs {model: self.model, messages: messages} if tools: kwargs[tools] tools resp self.client.chat.completions.create(**kwargs) msg resp.choices[0].message # 把tool_calls统一转成dict方便后续解析和调试 tool_calls [] if getattr(msg, tool_calls, None): for tc in msg.tool_calls: tool_calls.append({ id: tc.id, name: tc.function.name, arguments: tc.function.arguments, }) return { content: msg.content, tool_calls: tool_calls, finish_reason: resp.choices[0].finish_reason, }这一层做了三件事统一了模型入口、支持了tools参数、把输出结构化。以后不管后面接的是哪家模型只要它兼容OpenAI风格接口换模型就只改环境变量Agent主体代码完全不动。这就是面向接口编程在AI应用里的最小实践。3.2 第二步工具注册表与参数Schema接下来定义工具类。不要小看这一步工具的Schema写不好后面模型会频繁幻想出根本不存在的参数。import json from typing import Any, Callable, Dict, List class Tool: def __init__( self, name: str, description: str, parameters: Dict[str, Any], func: Callable[..., Any], ): self.name name self.description description self.parameters parameters self.func func def to_openai_schema(self) - Dict[str, Any]: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, }然后定义两个示例工具。一个查询天气一个给用户发送消息后者就是Agent自动发消息类场景的最小雏形def get_weather(city: str) - str: # 真实项目里这里会调天气API演示直接返回模拟值 return f{city}今天晴气温18-26摄氏度。 def send_message(user_id: str, content: str) - str: # 真实项目里这里会调消息网关 return f已向 {user_id} 发送{content} tools [ Tool( nameget_weather, description查询指定城市的当前天气参数city为城市名, parameters{ type: object, properties: { city: {type: string, description: 城市名例如北京、上海} }, required: [city], }, funcget_weather, ), Tool( namesend_message, description给指定用户发送一条即时消息, parameters{ type: object, properties: { user_id: {type: string, description: 接收方用户ID}, content: {type: string, description: 消息正文文本}, }, required: [user_id, content], }, funcsend_message, ), ]注意参数的description这是模型理解工具的说明书。说明书越具体模型越不会传错。比如在send_message里如果不写清content是消息正文模型可能把一句废话塞进去或者干脆漏传。3.3 第三步ReAct主循环——Agent的心脏下面这段就是造轮子的核心时刻。整个Agent逻辑不过几十行class Agent: SYSTEM_PROMPT ( 你是一个通过调用工具来完成任务的助手。 如果工具返回了结果请结合结果继续推理 当任务完成时直接给出最终答案。 ) def __init__(self, llm: LLMClient, tools: List[Tool], max_steps: int 10): self.llm llm self.tools {t.name: t for t in tools} self.max_steps max_steps def run(self, task: str) - str: messages [{role: system, content: self.SYSTEM_PROMPT}] messages.append({role: user, content: task}) for _ in range(self.max_steps): resp self.llm.chat( messages, tools[t.to_openai_schema() for t in self.tools.values()], ) if resp[tool_calls]: # 1. 先把assistant的工具调用包进消息历史 assistant_msg { role: assistant, content: resp[content] or , tool_calls: [ { id: tc[id], type: function, function: { name: tc[name], arguments: tc[arguments], }, } for tc in resp[tool_calls] ], } messages.append(assistant_msg) # 2. 逐个执行工具结果作为tool角色消息追加上去 for tc in resp[tool_calls]: tool self.tools.get(tc[name]) if not tool: output f未找到工具{tc[name]} else: try: args json.loads(tc[arguments]) result tool.func(**args) output json.dumps(result, ensure_asciiFalse, defaultstr) except Exception as e: output f工具执行失败{type(e).__name__}: {e} messages.append({ role: tool, tool_call_id: tc[id], content: output[:4000], # 防止单条observation过长 }) continue # 3. 没有工具调用说明模型给出了最终答案 return resp[content] or return 达到最大步数未得到最终答案。这段代码里有几个容易忽略的关键点assistant消息必须先入栈在OpenAI的function calling协议里tool角色消息必须通过tool_call_id关联到对应的assistant工具调用顺序错了API直接报错。工具执行要在try里任何工具都可能抛异常不能让异常炸掉整个Agent进程。把错误转成字符串返回给模型模型才有机会自行修正。observation要截断工具返回结果可能很长先截到4000字符。省token也防止上下文被单个结果污染。跑一个简单的例子agent Agent(llmLLMClient(), toolstools) print(agent.run(上海今天天气怎么样顺便提醒一下用户u_10086明天记得带伞))模型大概率会先调用get_weather拿到结果后再调用send_message最后给出已完成之类的收尾。这就是一个完整的最小Agent闭环。3.4 第四步加上最朴素的记忆管理第三步跑通之后你会立刻遇到一个问题任务复杂一点多轮工具调用后messages会越来越长。这时就要引入最简单的记忆管理记录历史、统计token、按预算裁剪。import tiktoken _enc tiktoken.get_encoding(cl100k_base) def count_tokens(messages: list[dict]) - int: return sum(len(_enc.encode(str(m))) for m in messages) def trim_messages(messages: list[dict], budget: int 12000) - list[dict]: if count_tokens(messages) budget: return messages # 保留system、最近两轮对话丢弃中间最旧的部分 head messages[:1] tail messages[-4:] if len(messages) 4 else messages[1:] trimmed head tail trimmed.append({ role: system, content: 注意较早的对话已被截断请基于剩余信息继续。, }) return trimmed这个策略当然很粗糙但麻雀虽小五脏俱全它让你理解记忆管理的本质是预算管理。更高级的版本无非是超出预算时优先保留什么、用摘要压缩旧消息、把关键事实抽出来存长期存储。核心判断逻辑都是同一套。4. 最容易翻车的四个细节解析、上下文、死循环、工具异常我自己在从能跑到好用的过程中踩过的坑基本都集中在这四个地方。每个都值得单独开一节。4.1 模型返回的JSON别指望永远是干净的JSON原生function calling能解决大部分格式问题但遇到弱模型、或者某些不支持function calling的模型时你仍然需要从模型输出里解析JSON。这时会遇到各种奇幻输出前面加一句好的我来调用工具用代码块包裹甚至JSON里混注释。我当时踩坑之后写了一个通用的容错解析函数import json import re def extract_json(text: str) - dict: text text.strip() # 去掉markdown代码块标记 if text.startswith(): text re.sub(r^(?:json)?, , text).rstrip().strip() # 直接截出第一个{到最后一个} start, end text.find({), text.rfind(}) if start ! -1 and end ! -1: return json.loads(text[start:end 1]) raise ValueError(输出中找不到JSON对象)经验教训是解析越健壮模型越自由。如果你能用原生function calling优先用如果必须文本解析就上容错解析再不行就做二次解析让模型自己修正上一次的JSON。4.2 Token预算上下文不是无限大的也没你想的那么小很多人不看token直到账单出来才清醒。我来算一笔账假设系统提示500 token10个工具的Schema合计3000 token每轮模型输出500 token每次observation平均800 token。10轮跑下来仅新增部分就有13000 token加上system和工具Schema一次任务轻松到20000 token。如果做多Agent编排、长会话撑爆128k上下文是迟早的事。所以实际开发中我通常会在Agent里做三件事给工具Schema瘦身只保留模型真正需要知道的字段详细说明放到工具内部文档里。给observation限长上面代码里的[:4000]就是干这个的。定期压缩历史当上下文超过预算时触发trim_messages或者把旧对话总结成一句摘要。不要觉得上下文还够不用管上下文是钱也是延迟。省token既是省成本也是保性能。4.3 死循环模型不停地调用同一个工具这是Agent最经典的失控姿势。任务描述模糊、工具返回信息没帮助、模型反复尝试同一方案就很容易陷入循环。只靠max_steps兜底虽然不会死进程但用户体验很糟。我用的一个有效手段是重复调用检测维护一个最近N步的调用指纹发现同一工具、同一参数连续出现多次就往messages里注入一条系统提示明确让模型换策略。call_fingerprints {} # 在循环内、执行工具前加上这段 fingerprint f{tc[name]}:{tc[arguments]} call_fingerprints[fingerprint] call_fingerprints.get(fingerprint, 0) 1 if call_fingerprints[fingerprint] 3: messages.append({ role: system, content: 注意你已经在重复调用同一个工具且没有进展请改变策略或者直接给最终答案。, })这个提示会作为后续模型的输入模型看到后通常会改变方案。比硬切断更平滑也比傻等到max_steps更高效。4.4 工具异常不要中断循环把错误喂回模型工具执行必然会有失败网络超时、参数格式错、权限不足、上游接口变更。我第一次写Agent时工具一报错就让整个循环break结果就是用户只看到执行失败Agent完全没机会自我修复。后来改成把异常转成observation喂回给模型。比如刚才代码里的工具执行失败{type(e).__name__}: {e}。模型看到这个信息通常会自己调整参数重新调用或者换一个工具。这就像人做菜时发现酱油瓶空了不会直接放弃整桌菜而是会换个调味料或者下单补货。但这里有个安全红线对危险操作不要无条件让模型自己重试。比如删除数据、转账、对外发消息这类工具最好加一个requires_confirmation标记触发时回到人工确认流程。自动化不是目的可控才是。5. 从能跑到好用接业务、做编排、上日志把最小闭环跑通之后你就要开始考虑真实场景了。这一章讲三件我实际项目里最常用到的事多Agent协作、接入业务系统、可观测性。5.1 单Agent之外两种最实用的多Agent协作模式我见过很多团队一上来就追求复杂的多Agent框架结果连单Agent的稳定性都没解决。我的建议是先掌握两种最基础的模式。模式一监督者模式Supervisor。一个主 Agent 负责拆解任务把子任务分发给不同的子 Agent最后汇总结果。实现起来有个很巧妙的技巧把子Agent封装成一个Tool。也就是说子Agent对外表现为一个能完成某类任务的工具主Agent通过function calling调用它。这样一来你不需要新的编排框架复用前面的工具注册表就行。模式二流水线模式Pipeline。多个Agent按顺序处理一个任务的不同阶段比如第一个Agent负责提取信息第二个Agent负责生成方案第三个Agent负责执行。每个Agent只关注一件事职责单一调试也容易。这两种模式配合单Agent的ReAct循环已经能覆盖大部分业务场景。再往上加并行调度、状态机、动态拓扑那是框架该做的事别急于手写。5.2 把Agent接进业务系统以Django为例Agent本质上就是一个函数输入任务字符串输出结果字符串。所以接入业务系统非常直接。以Django为例import json def agent_api(request): task request.POST.get(q, ) answer agent.run(task) return JsonResponse({answer: answer})生产环境当然不能这么粗暴。耗时较长的任务建议用Celery放到后台执行前端轮询任务状态如果需要流式输出就改造模型层把token流通过SSE推给前端。而Agent的工具完全可以是Django里的ORM查询、缓存操作、消息网关、内部REST服务。这也是我一直强调的Agent的落地能力不在Agent本身而在你能让模型碰到多少真实的业务函数。5.3 可观测性没有日志的Agent就是玄学调参Agent是多轮决策系统每一轮的状态都影响下一轮。如果连每一步模型想了什么、调了什么工具、花了多少token都不知道那调试就完全是猜谜。我会在循环里记录如下字段字段说明step第几轮input_messages进入本轮的消息列表或摘要、长度model_output模型输出内容tool_calls本轮发起的工具调用tool_results工具执行结果或错误信息token_count当前累计token数latency_ms本轮耗时每一轮落一条日志按request_id和session_id归档。看起来很简单但等你在线上遇到Agent为什么突然开始乱调工具的问题时这批日志会帮你把原因锁定到具体某一轮而不是对着完整对话记录从头猜。另外强烈建议统计成本。把每轮的prompt_tokens和completion_tokens加总按模型单价折算成费用。很多Agent跑飞的成本黑洞都是靠这个字段抓住的。6. 一点最后的体会造轮子的终点是知道自己该怎么选写到这里我想把最实在的体会留在最后。手写完这个最小Agent之后我在团队内部留了一个叫micro-agent的极简核心不到三百行。现在做任何Agent需求我都会先用它跑一个最小验证任务能不能被拆解、工具够不够用、模型表现稳不稳定。验证通过再决定要不要上成熟框架。这让我避开了很多一上来就上重型框架结果连问题都描述不清的尴尬。关于调包侠到造轮子的人我的理解不是以后再也不调包而是调包之前知道包里是什么换包的时候知道差别在哪拆包的时候知道哪些零件必须自己造。如果你也想走这条路我的建议非常具体不要一上来就手写一个巨大的平台先让两百行核心跑通一个能搜索、能查数据库、能发消息的Agent然后把这篇文章提到的四个翻车点挨个踩一遍。踩完了你对Agent的认知就完全不一样了。你写的每一行代码都会让你离调包侠远一点。等轮子原理在你心里不再是一团迷雾选择用别人的轮子还是自己造轮子都会变成清醒的决策。