1. 先聊清楚为什么放着现成框架不用非要手写一个Agent1.1 只会调用API时我经历的几个尴尬现场我先讲几个真实的尴尬场景都是我自己踩过的。第一个是框架升级事故。去年我有个项目用了一个很流行的Agent框架当时图的是封装好、上手快。结果框架发了一个大版本把工具调用的内部协议改了我的Agent突然不会调用工具了。我打开框架源码一看整整几千行根本不知道从哪里查起。最后只能回滚版本把框架锁死在旧版。项目是跑起来了但我心里清楚我对这个Agent的控制力其实约等于零。所有关键的决策逻辑、上下文组装、工具调度全被框架的黑盒吞掉了。我表面上在用Agent实际上只是在填配置。第二个是线上故障。Agent在某个场景下反复死循环同一句话来回调用同一个工具把三方API的配额打爆了。我排查的时候发现框架自带的循环控制在那个版本里有一个众所周知但没人提的坑它对重复动作没有做频率限制。我当时很想改一个参数搜了半天文档也没找到暴露的配置项最后只能自己在外围做了一个拦截。那种想改改不动的憋屈感我相信很多用过封装框架的人都懂。第三个场景更触发我反思。有一次做技术分享讲到Agent的原理我发现自己只能讲出框架里配置了这些参数、这些Prompt这类操作层面的东西一旦被问到模型到底是怎么决定调用哪个工具的上下文是怎么拼接的我就开始含糊了。你能配置它但你说不清它这本身就是一种能力的缺失。1.2 手写Agent到底在锻炼什么能力后来我下定决心把之前用框架实现的Agent用纯代码重新写一遍。那个项目我起了个名叫大都督因为我当时跟朋友开玩笑说真正把一个Agent从0写出来就像大都督都督兵马调的每一路兵、走的每一条路都得自己心里有数。这个项目后来成了我验证零基础手写AI Agent这条路径的素材。先说结论手写Agent跟重复造轮子是两回事。它的核心目的不是让你生产一个比LangChain更牛的框架而是让你亲手把一个Agent从收到用户消息到完成任务这条链路走通一遍。走通以后你会发现很多东西跟外面文章里写的不一样。比如很多人以为Agent的核心是那个大模型。但手写一遍你才会意识到大模型只占整条链路的三分之一。剩下三分之二是工程问题工具怎么注册、参数怎么解析、上下文怎么管理、循环怎么终止、失败怎么重试。这些环节恰恰是框架帮你藏起来的部分——也是你出问题时找不到原因的部分。这个道理跟手写MNIST数字识别、手写Transformer一脉相承。你用封装好的深度学习框架几行代码就能训练一个识别模型但只有手写一遍反向传播你才知道梯度到底是怎么传的。用封装好的大模型SDK你可以轻松搞一个能聊天的程序但只有手写一遍Agent循环你才知道智能发生在哪里。1.3 手写与直接上框架的边界判断我也踩过另一个极端什么都想手写。实际上有些东西是不需要手写的。我给自己的边界是这样的大模型的调用肯定用官方SDK或HTTP接口没必要自己实现一个HTTP客户端去接SSE流那是重复劳动。Agent的编排逻辑第一遍必须自己手写。包括消息组装、工具分发、循环终止、上下文裁剪这些是理解Agent的关键。基础中间件比如Web服务框架、数据库连接池直接选成熟方案不折腾。Agent内部的状态管理前期自己写后期如果变得复杂再考虑接入成熟的状态机框架。也就是说手写的重点是Agent属于自己的那一层逻辑而不是什么都从零造。这样既能把原理吃透又不至于陷在基础设施的泥潭里出不来。这个项目的完整代码我放在了自己的一个工具仓库里博文里我会把最核心的骨架、工具调用、上下文管理、并发改造四个部分逐一拆开讲。每个部分都会给关键代码并且说明为什么这么写。如果你也是那种用了很久框架但心里发虚的人这篇应该能帮你把缺失的那块拼图补上。2. 最简Agent骨架从一次Chat Completion开始的闭环2.1 先忘掉Agent从一次普通对话接口调用说起我们先把Agent这个概念放到一边。你现在手头有一个大模型的API接口能让你发一段消息然后得到一个回答。不管这个接口是官方的OpenAI、DeepSeek还是本地部署的Ollama、LMStudio本质上都是同一个模式你发一个消息列表过去对方返回一个消息回来。一个最简单的聊天机器人就这样完成了。但如果你想让它变成Agent需要多出几个东西它能理解你给它的任务而不仅仅是回答问题。它能自主决定下一步做什么。它能调用外部工具比如查天气、查数据库、发请求。它能根据工具返回的结果决定是继续行动还是收尾回答。这就是Agent和Chat Completion之间的本质区别从一问一答变成了多轮自主执行。2.2 手写一个最小Agent循环我写的第一个骨架非常简单核心就是一个while循环。我先定义消息列表然后反复让模型返回结果直到它认为任务完成。import json from openai import OpenAI client OpenAI( api_keyyour-api-key, base_urlhttps://your-endpoint/v1 # 兼容OpenAI协议的均可 ) SYSTEM_PROMPT 你是一个智能助手。你要根据用户的目标决定是否需要调用工具。 如果不需要工具直接给最终答案。 如果需要工具请返回包含 tool_calls 的消息。 def run_agent(user_input: str, max_steps: int 10): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] for step in range(max_steps): response client.chat.completions.create( modelyour-model, messagesmessages, toolsTOOLS, # 工具定义稍后讲 tool_choiceauto, ) message response.choices[0].message messages.append(message) # 没有工具调用说明模型觉得可以收尾了 if not message.tool_calls: return message.content # 有工具调用就逐个执行 for tool_call in message.tool_calls: result execute_tool(tool_call) # 稍后讲 messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) return 超出最大步数任务终止。这个循环最关键的一点是每一步都把模型的输出追加到messages里然后把工具执行结果以tool角色的消息追加回去。模型看到工具结果之后会继续推理——要么再调用下一个工具要么给出最终答案。我第一遍写这个循环时犯过一个错误把工具结果放到user消息里返回给模型。虽然有的接口也能凑合用但格式不规范在多轮循环中容易出各种幺蛾子尤其是当模型同时调用多个工具时tool_call_id对应关系会乱。所以一定要走标准的tool消息格式这个不能省。2.3 骨架跑通后先把终止条件想清楚骨架写出来跑通之后很多人会急着加各种花哨功能。但我劝你先琢磨一个问题这个循环凭什么停下来我在上面写了两个终止条件模型回答里没有tool_calls意味着它认为任务完成了。循环次数超过max_steps强制终止。第二个条件是我后加的。原因是一次线上演示时模型在一个需要查数据库的任务里连续调用了将近二十次工具每一次都查出一堆中间结果但就是不给出最终答案。我知道这是模型过度思考了但当时我的循环没有上限只能卡在那儿。后来我在System Prompt里补了一句如果工具结果已经足够回答用户的诉求请立即给出最终答案不要继续调用工具。同时把max_steps的默认值从10调低到了6。为什么是6因为我观察过实际任务大多数合法任务在3到5步之内就能完成。超过6步多半是模型在绕圈子。这个值你可以根据自己业务的复杂度调但一定要有而且要充分测试。还有一个小细节终止条件不只是模型说完成。有些时候模型假装没看见工具报错强行给出一个错误答案。所以我在循环里加了异常捕获和日志一旦工具执行抛异常我会把错误信息作为工具结果回传给模型让模型根据错误信息重新规划。这个策略非常有效——模型不是不能纠错而是你得给它纠错的机会。3. Function Calling的实现大模型如何真正操作外部工具3.1 模型到底是怎么学会调用工具的你要理解Function Calling工具调用的原理得先搞清楚一件事模型本身并不会执行任何代码。它只是一个文本生成器。所谓调用工具其实是模型在一段结构化的文本里按照你预定义的格式输出一个我想调用某个函数参数是这些的JSON。以OpenAI兼容协议为例你发起请求时在tools参数里传入一个JSON Schema数组描述每个工具的名字、描述、参数结构。模型读到这些描述之后会根据自己的理解在合适的时机输出一个tool_calls结构{ tool_calls: [ { id: call_abc123, type: function, function: { name: query_weather, arguments: {\city\: \杭州\} } } ] }看到没有模型只负责决定调用谁、传什么参数真正执行的是你的代码。这就是为什么我强调手写一遍工具分发逻辑特别重要——你会在这一层看到很多模型自作聪明的情况比如参数名对不上、缺参数、甚至幻觉出一个不存在的工具名。这些都需要你的代码去兜底。我第一次手写这个分发逻辑时天真地以为模型一定会严格按我给的JSON Schema输出。结果它一次给了五个参数有三个是我不认识的。后来我学乖了不要信模型的自觉要做参数校验和容错。3.2 手写工具注册表与参数解析工具注册表是Agent里非常核心的一个组件。它的作用就类似于路由器模型说要调用某个工具你的代码拿着工具名找到对应的Python函数把参数传进去执行。我用的方案是一个装饰器简洁且扩展性很好TOOL_REGISTRY {} def register_tool(nameNone, description, parametersNone): def decorator(func): tool_name name or func.__name__ TOOL_REGISTRY[tool_name] { function: func, description: description, parameters: parameters, } return func return decorator def get_tool_schemas(): schemas [] for name, info in TOOL_REGISTRY.items(): schemas.append({ type: function, function: { name: name, description: info[description], parameters: info[parameters], } }) return schemas def execute_tool(tool_call): try: fn_name tool_call.function.name arguments json.loads(tool_call.function.arguments) func_info TOOL_REGISTRY[fn_name] result func_info[function](**arguments) return json.dumps(result, ensure_asciiFalse) except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)这样一个工具就只需要三行定义就能接入Agentregister_tool( description查询指定城市的当前天气, parameters{ type: object, properties: { city: {type: string, description: 城市名如北京、杭州} }, required: [city], } ) def query_weather(city: str): # 这里调真实天气API或者直接返回mock数据 result get_weather_from_api(city) return {city: city, weather: result}有几个点我想特别提醒你参数的description一定要写得非常具体。同一个city参数你写城市名和城市名中文如杭州、北京不要带市字的效果天差地别。模型是靠描述来理解参数的描述越精确它给错参数的概率越低。函数的返回信息也最好用JSON结构化。模型在下一轮推理时需要从你的返回文本里提取关键信息。如果返回的是一堆半结构化文本它很容易读岔。execute_tool一定要写异常捕获。我在开发中遇到过模型传了负数当订单号、传了不存在的用户ID等各种情况。如果你不捕获异常整个Agent循环会在这一步崩掉连带整个Web服务都挂掉。捕获异常并把错误信息回传给模型反而能让模型知错就改。3.3 让Agent去调用外部系统的几种常用方式工具注册表搞定之后Agent就已经具备了操作外部世界的能力。但具体怎么操作取决于你的业务形态。我实际项目里用过几种方式分享下各自的使用场景直接调用三方HTTP API最简单的方式。订单查询、天气查询、汇率换算这类都可以直接用requests或httpx封装成工具函数。调用本地脚本或命令行适合Agent需要操作文件、跑批处理任务的场景。用subprocess在工具函数里调python脚本或shell命令注意设置超时时间。访问数据库把SQL查询封装成工具函数。注意永远不要直接暴露一个执行任意SQL的工具给模型否则模型会给你整出各种花活。我用的方案是把查询按业务场景封装成几个固定函数比如查最近订单查用户余额参数也只有ID、时间范围这些安全值。调用本地模型接口比如你本地用Ollama或LMStudio跑了一个专用小模型Agent可以把某些子任务委派给它。我实际试过用Ollama运行一个本地模型做文本分类任务Agent负责调度效果很稳。关键是给这个小模型的输入输出格式定义清楚并且设置好超时。还有一点我想特别强调工具函数的执行是同步阻塞的这意味着它会在Agent循环里卡住整个线程。如果你的工具里有比较耗时的操作一定要考虑超时控制和异步化。后面讲并发的时候我们会再展开。4. 上下文与记忆设计决定Agent智商上限的隐藏工程4.1 Token膨胀Agent的记忆力其实是个上下文窗口很多人刚开始写Agent时会觉得多轮对话嘛把所有历史都塞给模型不就行了。这个想法在小步数内没问题但一旦任务复杂你就会碰到一个很现实的问题Token会爆。我做过一次压力测试。一个需要调用工具查询多次的任务跑完5步之后messages列表里的内容已经有将近8000个Token。其中模型自己的输出尤其是带思考过程的输出占了七成。如果业务是连续多轮比如用户和Agent聊了一整个下午还反复修改需求那上下文窗口迟早被撑爆。Token爆了会怎样两种后果一是API报错上下文超过模型上限二是虽然没报错但模型被海量历史信息干扰把早期的错误决定当成依据或者忽略了你最新的指令。后者比前者更隐蔽也更危险。我自己的处理策略是分三层滑动窗口只保留最近N轮对话更早的直接丢弃或压缩。N我常用20轮这个值取决于你的业务复杂度。优点是实现简单缺点是会丢失一些早期的关键约束。摘要压缩当历史超过阈值时触发一次摘要。让模型把早期对话压缩成一段摘要文本替换掉原始的冗长消息。这个摘要我一般放在System Prompt的顶部作为长期记忆的一部分。实现不复杂但效果很好。关键信息提取对于明确的结构化信息比如用户的ID、订单号、偏好设置在每一轮对话后单独提取出来存到结构化的状态里。下次组装上下文时直接把这些关键状态拼进去而不是依赖模型从历史里回忆。4.2 短期记忆与长期记忆的分工这里我想引入两个概念工作记忆和长期记忆。工作记忆就是当前任务的多轮对话历史存在messages里长期记忆则是用户画像、业务偏好、历史订单这类跨会话的信息存在数据库或向量库里。我做的第一个版本只用了工作记忆结果发现了一个很真实的问题同一个Agent实例服务于多个用户时上下文串了。那个场景是这样的——我的Agent被封装成一个Web服务收到请求就创建一个Agent实例然后把用户的问题丢进去。一开始我图省事把Agent实例放在全局变量里结果用户A的消息还没处理完用户B的消息又进来了两个任务的上下文混在一起Agent开始胡言乱语甚至把A用户的信息漏给了B用户。这个坑给我留下的教训就是Agent实例必须与会话绑定一次会话一个独立的上下文。也就是常说的session隔离。具体实现上我当时用了一个简单的字典来管理会话sessions {} def get_session(session_id: str): if session_id not in sessions: sessions[session_id] { messages: [], state: {}, # 结构化关键信息 } return sessions[session_id]这样一个用户一个session_id消息互不干扰。当然这个方案在重启后会丢内存数据生产环境我后来换成了Redis但思路是一样的。长期记忆的做法稍微复杂一点但核心也不难。我当时的做法是每当对话结束调用一个记忆提取工具让模型从这段对话里提取出结构化的用户偏好信息存进数据库。下次用户再进来时先查数据库加载这些偏好放进System Prompt里。当模型不需要从浓雾般的聊天记录里猜你的偏好时它的每一步推理都会更稳。4.3 上下文管理里最容易踩的两个坑第一个坑是System Prompt被挤到中间位置。有些框架在组装上下文时会把System Prompt放在最前面这没问题。但我见过一些实现把最新的工具结果、摘要等信息插入到了System Prompt和对话历史之间。这会导致模型对系统级指令和对话内容的边界产生混淆。我踩过之后就把所有系统级的内容统一放到最前并且用极清晰的标记区分比如# 系统指令 这里放角色设定、行为规则、关键偏好摘要 # 对话记录 这里只放历史信息第二个坑是重复的摘要叠加。有一版实现里每压缩一次摘要就把新的摘要追加到上一段摘要后面几十轮之后摘要本身变成了几万Token。模型每次都要同时读两段又长又矛盾的陈年旧事反而忽略了当前的对话内容。后来我改成每次压缩时把旧的摘要合并进新的摘要只保留一段最终摘要。摘要是一个不断更新的结论而不是一本不断增厚的历史书。5. 从单机脚本到高并发服务Agent项目怎么扛住真实流量5.1 一个Agent实例和一个Agent服务的差别早期的Agent是命令行脚本跑一次就完事。但你要把它产品化就得面对一个绕不开的问题并发。网上关于AI Agent怎么扛并发的帖子很多但我看下来很多都太虚了。Agent服务和普通Web服务的高并发本质区别在于普通接口是快进快出毫秒级返回Agent接口是长跑可能要好几十秒甚至更久。一个Agent请求在运行期间要多次调用大模型API、多次执行外部工具这中间的时间大部分是IO等待。我用FastAPI封装Agent服务时最开始的实现非常天真每个请求进来直接在请求处理函数里跑Agent循环。结果第一个并发测试就挂了——FastAPI的同步路由会占用工作线程而我的Agent循环里有大量阻塞式API调用线程池被瞬间打满其他所有请求全部排队等待。5.2 用任务队列和信号量给并发上锁后来我调整了架构核心思路是不要让请求直接驱动Agent循环而是把任务扔进队列由固定的Worker池来消费。我用的是Python的asyncio.Queue配合一个简单的Worker循环import asyncio task_queue asyncio.Queue() MAX_CONCURRENCY 5 semaphore asyncio.Semaphore(MAX_CONCURRENCY) async def worker(): while True: session_id, user_input await task_queue.get() try: async with semaphore: result await run_agent(session_id, user_input) # 把结果写回某个存储或者通过回调通知前端 finally: task_queue.task_done() # 启动固定数量的worker for _ in range(MAX_CONCURRENCY): asyncio.create_task(worker())这里的关键是asyncio.Semaphore。它限制同一时刻最多只有5个Agent任务在真正执行。为什么是5不是拍脑袋定的我的大模型API供应商限制每分钟请求数RPM换算下来平均每秒最多5个请求。一个Agent任务在运行周期内大约会调用模型2到5次所以5个并发任务已经能把API配额吃得很紧。超出并发的请求进入队列排队而不是直接打穿上游。这个方案的好处是流量再大也不会把大模型API打爆。排队等待虽然会增加延迟但总比请求直接失败要好。很多做Agent服务的团队初期都不重视这一层结果上线第一天就被爬虫或活动流量打崩。我自己真实见过一次第三方API直接给回了429限流错误一大片Agent任务全部失败。5.3 超时、重试和优雅降级不能拍脑袋并发只是第一步。Agent服务真正难的是异常场景处理。我给你列几个我处理过的真实问题问题一模型API偶发超时。这种情况很常见。我用的是指数退避重试策略第一次失败等1秒重试第二次等2秒第三次等4秒最多重试3次。超过重试次数就把任务标记为失败。import time def call_model_with_retry(messages, tools, max_retries3): for attempt in range(max_retries): try: return client.chat.completions.create( modelMODEL_NAME, messagesmessages, toolstools, tool_choiceauto, timeout30, ) except Exception as e: if attempt max_retries - 1: raise wait_time 2 ** attempt print(f模型调用失败{wait_time}秒后重试: {e}) time.sleep(wait_time)这里的timeout30和wait_time2 ** attempt都是根据实际压测调过的。短了治不了临时抖动长了拖累整体任务时间。问题二工具执行卡死。有的工具函数调用外部API对方服务挂了请求就悬在那儿把整个Worker卡死。解决办法就是给工具执行加超时。Python里可以用asyncio.wait_for来包一层超时后强制放弃这个工具结果并回传给模型一条工具执行超时的信息让模型决定是重试还是换一种方案。问题三任务整体超时。Agent任务本身也要有总超时。我用的是90秒硬上限。超过90秒不管模型在干什么直接返回任务超时请简化需求或稍后重试。这个值你也可以自己调但一定要有。还有一个很实用的兜底当Agent连续失败两次时自动降级为普通问答模式。也就是不再加载任何工具直接让模型用自身知识回答。虽然可能不那么精准但至少用户不是干等着。这种锦上添花的处理在产品体验上非常加分。6. 回看手写之路我保留的三份代码与零基础进阶建议6.1 手写过程中最值得保留的三份代码从零手写Agent到现在我复盘了一下有三份代码是无论如何都要留着的。第一份是工具注册表。它是整个Agent的神经中枢。我后来把这个工具注册表扩展成了一个通用的插件系统新工具只需要写一个函数加一个装饰器就能接进来完全不用改Agent的主循环。团队里其他同学接手时学这部分只花了一个下午。第二份是上下文管理器。它负责维护每个会话的messages列表、触发摘要压缩、管理长期记忆。没有它会话一长就崩。有了它你甚至可以调整各种压缩策略来对比模型效果。我强烈建议你把摘要触发条件、窗口大小都做成可配置的方便后续做优化实验。第三份是重试与超时装饰器。这个通用工具不只在Agent里能用任何Python服务里都能用。我封装成一个装饰器支持自定义重试次数、退避策略、超时时间。有了它Agent调模型、调工具、调三方API全部统一走同一套兜底逻辑。6.2 给零基础学习者的路径建议如果你也想从零开始手写一个Agent我给你一个经过验证的路径不需要你有很深的AI基础但要有Python基础第一阶段1到2天搭一个最简聊天循环。用你自己的API Key调用一次Chat Completion接口完成用户输入-模型返回的最基本闭环。可以不接任何工具纯体验。第二阶段2到3天加入工具调用。先写一个最简单的工具比如获取当前时间或者计算器按我上面讲的注册表方式接入。跑通之后再逐步加第二个、第三个工具观察模型怎么在多个工具之间做选择。第三阶段1到2天加入上下文管理。实现会话隔离、滑动窗口。做一个简单的Web界面用FastAPI 基本前端即可让多个用户同时使用你的Agent验证会话不串。第四阶段2到3天做并发与稳定性改造。加信号量限流、重试机制、超时控制。用压测工具跑几个并发请求观察队列排队情况。这样走下来两周左右的时间你已经不是一个会调用API的调包侠了而是一个能从第一性原理解释Agent为什么这样设计的开发者。你在学习过程中遇到的大多数问题Google能帮你解决。但要记住一个原则凡是报错先把报错信息完整看一遍再看你的消息结构再看模型返回值最后再上网查。这个排查顺序能帮你解决80%的问题。6.3 一个调试小技巧和一个工具选型建议最后分享两个实操层面的小贴士。第一个是调试技巧。在Agent的每个关键环节打印事件日志包括模型返回了什么、模型决定调用什么工具、工具执行耗时多久、工具返回了什么、当前消息有多少Token。我早期调试Agent全靠这些日志。别看它们丑定位死循环、定位参数错误、定位上下文异常全靠它们。我见过很多新人遇到问题就喜欢去调提示词其实很多问题都不是提示词的锅是消息结构错了、是工具参数解析错了。日志会告诉你答案。第二个是工具选型建议。手写第一遍Agent时模型API建议用兼容OpenAI协议的国内服务或本地模型成本低、调试方便。比如DeepSeek的API、阿里的DashScope、或者本地Ollama跑一个开源模型。它们都支持tools参数。本地模型调试的时候响应快、还不花钱非常适合学原理。等你把整条链路跑熟了再换更强的大模型测试你会发现换模型这件事只影响智能上限不影响你的工程架构。我在这个手写项目中最大的收获并不是写出了一个多能跑的Agent而是建立了一种安全感当Agent出问题时我知道问题出在哪一环而不是对着黑盒发呆。这种安全感是任何框架都给不了你的。如果你正在框架的包围里感到很忙但心里发虚建议你抽两周时间亲手拆一次这个黑盒。这条路不好走但走完之后你会对AI Agent这四个字有一个完全不同的理解。