做了几年大模型应用开发,有一件事始终让我心里不踏实:别人问我"你项目里的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返回结果追加进messages,token数就会不断增长。这个点看着基础,却和成本、上下文溢出、性能都直接挂钩,后面第4章会专门展开。
2.2 工具层:Agent与普通Chatbot的分界线
Chatbot只会输出文字,Agent能动手做事。这个"动手"靠的就是工具层。
一个工具,本质上就是一个普通函数,加上一份"使用说明书"。说明书包含四部分:工具名、功能描述、参数Schema、执行函数。描述写得越清楚,模型越不容易用错。举个例子,一个发消息工具的描述如果只写"发消息",模型可能不知道该传哪个接收方;如果写清楚"给指定用户发送一条即时消息,user_id是接收方ID,content是要发送的文本",模型一读就懂。
现在主流大模型大多支持原生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_key=os.getenv("OPENAI_API_KEY"), base_url=os.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( name="get_weather", description="查询指定城市的当前天气,参数city为城市名", parameters={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名,例如:北京、上海"} }, "required": ["city"], }, func=get_weather, ), Tool( name="send_message", description="给指定用户发送一条即时消息", parameters={ "type": "object", "properties": { "user_id": {"type": "string", "description": "接收方用户ID"}, "content": {"type": "string", "description": "消息正文文本"}, }, "required": ["user_id", "content"], }, func=send_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_ascii=False, default=str) 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(llm=LLMClient(), tools=tools) 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 token,10个工具的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的认知就完全不一样了。
你写的每一行代码,都会让你离"调包侠"远一点。等轮子原理在你心里不再是一团迷雾,选择用别人的轮子还是自己造轮子,都会变成清醒的决策。