AI Agent实战:从零到生产,构建可控可观测的智能体系统

📅 发布时间:2026/10/18 23:20:39
AI Agent实战:从零到生产,构建可控可观测的智能体系统
先说个反直觉的结论一个真正能上线干活的AI Agent代码量往往不超过500行而这里面最值钱的部分不是调用大模型那两行而是你围绕循环、工具、记忆、兜底这四个环节做的设计。我见过太多人把Agent想复杂了一上来就堆重型框架结果Demo跑了两天连一个最简单的查天气再决定带不带伞都没闭环。这篇文章想给你一条可以照抄的路径从零开始构建一个实用的AI Agent系统并把从本地能跑到生产可用的每一步拆开讲清楚。不管你是什么基础只要跟着把这几层逻辑搭起来就能得到一个结构完整、能接真实数据、能部署上线的Agent脚手架。这篇文章没有绑定任何特定的大模型厂商接口代码里统一用chat_model这样的伪SDK来演示你换成自己项目里实际使用的模型客户端接口保持一致即可。这样做的目的是让思路不被某一家产品绑定也不让品牌名称干扰你对本质的理解。1. AI Agent相比普通程序到底多了什么——先看清边界再动手1.1 传统程序和Agent的分界线其实只有一条很多新手对AI Agent的第一反应是能自动聊天的机器人这个理解不算错但会把方向带偏。我习惯用一个更朴素的对比普通程序是做选择题的Agent是做填空题的。普通程序把所有的执行路径提前设定好输入A走分支A输入B走分支B行为完全可预期Agent拿到一个目标之后需要自己判断下一步该干什么中途任何一步的结果都会改变它接下来的行动。举一个生活化的例子。假设你要做一个出行助手传统写法用户选城市、选日期程序调用天气接口把结果格式化输出结束。Agent写法用户说明天去青川市出差要不要带伞Agent先判断这需要青川市明天的天气然后调用天气工具拿到降雨概率发现上午有阵雨自动补一句建议带伞另外下午转晴行程不用改中间每一步都是模型根据工具返回的实时结果现场决策的。这里的关键差别在于传统程序的决策逻辑写在代码里Agent的决策逻辑写在模型的推理里。代码世界没有歧义和临场发挥而Agent面对的是一句随口说的话里隐含的任务拆解。理解了这条分界线你就能明白为什么Agent系统不是把API换成大模型接口就完事了你需要给模型搭一个决策-行动-观察的舞台这个舞台的骨架就是控制循环。1.2 拆开一个最小的Agent它只有四个部件我习惯跟团队新人说不管外面宣传的Agent概念多花哨一个能落地的Agent系统绕不开四样东西模型、工具、记忆、控制循环。模型是推理引擎负责理解输入、拆解任务、决定下一步该调用什么工具。它不需要多聪明但必须愿意输出结构化的工具调用指令否则整个循环推不动。工具是Agent的手臂让它能查数据库、调外部接口、读写文件。没有工具的Agent只是个高级问答盒子有了工具才真正能干活。记忆分两层短期记忆就是当前对话的上下文长期记忆是跨会话的摘要、用户偏好、历史结果。记忆的质量直接决定Agent会不会失忆。控制循环是骨架它是一段循环逻辑把用户问题交给模型模型说要调用工具程序执行工具把结果返回给模型模型看到结果后继续判断直到它认为可以给出最终答案为止。四个部件的关系可以这样理解模型是大脑工具是手脚记忆是记事本控制循环是那个每走一步停下来看一眼方向的机制。后面所有篇幅都是在把这四个部件一个个做扎实。如果你是零基础第一遍读可以先不纠结代码只用这四个词对照着看思路会清晰很多。之后再去看任何Agent框架、Agent产品你都能一眼看穿它到底在解决哪个部件的问题。2. 动手前的脚手架技术选型逻辑和最小目录结构2.1 我为什么没有一上来就选重型编排框架现在的Agent生态里编排框架和可视化平台非常多有些确实做得漂亮拖拽就能搭流程。但我的建议是第一版一定不要用重型框架先用最朴素的方式把控制循环手写出来。原因很简单Agent的调试难度和抽象层级成正比。你用了框架等于给模型套了一层壳一旦出现模型乱调用工具或者上下文丢消息这类问题你很难分清是框架的问题、你自己的提示词写得有问题还是模型的判断出了偏差。手写循环的情况下每一行消息如何组装、每一步调用如何发生你全部看得见、改得动。而且对小白来说手写一遍循环本身就是最好的学习过程。等你把循环逻辑真正跑通、把踩坑经验沉淀下来再引入框架提升工程效率完全来得及。从零到生产这条路第一版不设黑盒比什么都重要。所以这篇文章里的代码不绑定任何特定框架。你会看到我用chat_model(messages, tools)这样的伪SDK来约定模型调用用Tool数据结构约定工具形态。你只需要把你实际使用的模型客户端塞进chat_model的位置其他逻辑原样保留就能跑起来。2.2 先把接口约定死后面才不会被改到怀疑人生写任何功能之前先定义两个核心数据结构这两个是你的地基。第一个是模型返回结果。我约定一个ModelResponse对象包含三个字段dataclass class ModelResponse: content: str # 模型输出的文字内容 tool_calls: list[ToolCall] # 模型要求调用的工具列表可能为空 usage: dict # token消耗信息生产里一定要看 dataclass class ToolCall: id: str # 工具调用ID返还结果时要对齐 name: str # 工具名 arguments: dict # 传给工具的参数JSON字典第二个是工具注册表。每个工具都是一个Tool对象包含四样信息dataclass class Tool: name: str # 工具名建议小写加下划线 description: str # 给模型看的描述写清楚用途和适用场景 parameters: dict # JSON Schema格式的参数说明 func: Callable # 真正的执行函数这两个接口看起来简单但它们是后面所有逻辑站稳的前提。tool_call_id必须保留因为模型可能在同一轮里并发发起多次工具调用返还结果时必须准确匹配到对应的那一次description是写给模型看的不是写给程序员看的很多人会在这里偷懒后面必吃苦头。2.3 一个能一路走到生产的项目目录长这样目录不需要豪华但分层的习惯从第一天就要开始养。我常用的最小目录是这样agent_app/ ├── main.py # HTTP服务入口 / 命令行入口 ├── config.py # 模型参数、限额、超时、环境配置 ├── agent/ │ ├── core.py # 控制循环思考-行动-观察 │ ├── prompts.py # 系统提示词集中管理 │ ├── memory.py # 会话记忆短期缓冲与长期摘要 │ └── tools/ │ ├── registry.py # 工具注册与分发 │ ├── realtime.py # 实时查询类工具 │ └── business.py # 业务类工具 ├── observer.py # 日志、追踪、费用统计 ├── tests/ # 循环逻辑和工具的单元测试 └── requirements.txt我见过太多人把工具函数直接写在main.py里第一周很爽第二周就开始痛苦。工具一旦超过五个注册、路由、测试、回滚都需要清晰的目录边界。这里顺带说一句observer.py不是可有可无的后面第五章会专门讲一个生产级Agent的命脉其实是观测。运行的时候看不到每一轮推理等出了问题就只能靠猜。3. 第一版可运行的Agent最简单的思考-行动-观察循环3.1 定义第一个工具从嘴炮到能动手理论讲再多不如来一个能跑的。我们做一个最简单的信息查询Agent用户问天气Agent能调用天气接口把结果拿回来。先定义工具import json def get_weather(city: str, date: str 今天) - str: 查询指定城市、指定日期的天气情况。 返回JSON包含: city, date, high_temperature, low_temperature, precipitation_probability, weather_desc。 # 真实项目里请替换为你的天气API调用这里用固定返回便于演示 data { city: city, date: date, high_temperature: 27, low_temperature: 19, precipitation_probability: 80, weather_desc: 中雨转阵雨, } return json.dumps(data, ensure_asciiFalse) tool_weather Tool( nameget_weather, description( 查询某城市某日天气。适用于用户询问天气、出行准备、穿衣建议等场景。 date参数支持今天、明天、YYYY-MM-DD。 查询失败时返回包含successfalse和error字段的JSON。 ), parameters{ type: object, properties: { city: {type: string, description: 城市名如青川市}, date: {type: string, description: 日期默认今天}, }, required: [city], }, funcget_weather, )这里有两个细节建议你死记。第一description是写给模型看的。你写获取天气数据模型也能懂但如果把适用场景、触发条件、参数约束都写清楚模型在模糊询问时就知道该不该调用它。第二工具返回一定是稳定结构最好直接返回JSON字符串不要返回Python对象。因为后面要把结果塞回对话上下文喂给模型字符串永远是最不容易出错的载体。3.2 核心循环代码把思考-行动-观察变成可运行的结构接下来是Agent的心脏控制循环。这里我刻意不用任何框架def run_agent(user_input: str, system_prompt: str, tools: list[Tool], max_steps: int 8): tool_map {tool.name: tool for tool in tools} messages [{role: system, content: system_prompt}] messages.append({role: user, content: user_input}) for step in range(max_steps): resp chat_model( messagesmessages, tools[ {name: t.name, description: t.description, parameters: t.parameters} for t in tools ], tool_choiceauto, ) if not resp.tool_calls: return resp.content messages.append({ role: assistant, content: resp.content, tool_calls: [ {id: c.id, type: function, function: {name: c.name, arguments: c.arguments}} for c in resp.tool_calls ], }) for call in resp.tool_calls: tool tool_map.get(call.name) if tool is None: result json.dumps({success: False, error: f未知工具: {call.name}}) else: try: result tool.func(**call.arguments) except Exception as exc: result json.dumps({success: False, error: str(exc)}) messages.append({ role: tool, tool_call_id: call.id, content: result, }) # 观察结果后循环继续让模型决定下一步 log_step(stepstep, messagesmessages, usageresp.usage) return 很抱歉这个问题比较复杂我连续处理了多步还没能收敛。建议换个说法或者把需求拆小一些。这段代码的骨架就是经典的Agent模式模型说做什么程序做什么把结果还回去模型再看一眼。你可能会问为什么要有max_steps因为模型是完全有可能陷入死循环的比如天气接口一直返回异常它就可能一直重试。没有步数上限的Agent轻则烧钱重则把下游系统打到限流。8步是我比较常用的初始值业务复杂的可以放到15但尽量不要更高。3.3 上下文组装细节为什么消息顺序这么重要上面循环里最容易被忽略的是messages这个列表的组装逻辑。它必须严格按时间顺序追加system提示词在最前接着是用户输入然后是模型要求调用工具的那条assistant消息工具的执行结果以tool角色紧跟其后下一轮模型根据工具结果做出的新判断再作为新assistant消息出现。顺序一旦错了模型就会失忆。工具结果放到用户消息之前模型可能根本不知道这个结果对应哪次查询漏掉上一轮assistant消息模型连自己刚才提出要查天气这件事都不记得只会重复调用同一个工具。这类问题属于看着简单、踩起来很痛。提示第一版跑通之后建议专门打印一次完整消息列表亲眼看看多轮循环之后它长什么样。Debug定位上下文问题时这条经验能帮你省下至少半天的排查时间。4. 让Agent真正实用的三次改造记忆、工具编排、边界兜底第一版能跑但肯定不好用。你试试连续问三轮问题第二轮开始它就可能忘了你第一轮说过的关键信息工具加到十几个它又开始乱选工具。这章讲实用化的三次改造都是真实项目里验证过有效的手段。4.1 记忆怎么落从进程内字典到短期缓冲长期压缩本地开发时很多人把记忆简单实现成一个Python字典session_id - messages。我得说这没问题但你要清楚它的两个致命伤进程一重启就丢多实例部署时互相看不见。生产环境的底线是把它放到独立的内存数据存储里保证无状态Worker都能读到同一个会话。比存储更棘手的是上下文长度。一个Agent多跑一步消息数量就会翻着倍涨用户问题、模型调用、工具结果、模型下一轮判断四段才算一轮。所以要用两层记忆短期缓冲最近N轮消息原样保留比如6轮以内。长期压缩超过阈值时调用模型把旧消息总结成摘要放在用户消息之前。简化的压缩逻辑def compress_messages(messages, max_turns6): if len(messages) max_turns * 4 2: return messages old_part messages[:-max_turns * 4] text \n.join( f[{m[role]}] {m[content]} for m in old_part if m.get(content) ) summary chat_model([ {role: system, content: 请将以下对话历史压缩为300字以内的关键事实摘要保留用户意图、已查到的结果和未完成事项。}, {role: user, content: text}, ]).content return [{role: system, content: f[历史摘要] {summary}}] messages[-max_turns * 4:]这个方案不复杂但非常有效。注意压缩时一定不要漏掉未完成事项比如用户上一轮让Agent查了两家酒店的报价还没决定压缩时漏了这条下一轮Agent就会假装这事没发生过用户体验会非常差。4.2 工具编排工具一多模型就选择困难第一版只有两三个工具时模型选得很准等工具加到十五个模型开始频繁误调用。用户问数据库的问题它去调了新闻搜索。这几乎是每个Agent项目必经的阶段不用怀疑是自己能力不行。解决办法是给工具分域并在系统提示词里声明路由规则。我习惯分三组实时查询组天气、新闻、股票、汇率这类外部信息内部业务组查订单、改库存、读报表这类系统内部操作系统能力组计算器、单位换算、日期计算这类通用小工具然后在提示词里写明如果用户问题跨多个域先拆解再行动如果单一域内工具不明确先输出判断依据再调用不要猜测。更彻底的方案是在进入循环之前加一次意图路由先用一次轻量模型调用判断问题属于哪个域再把该域的工具列表传给Agent循环。工具少了准确率自然会上去。代价是多一次模型调用但换来的是稳定性这笔账算得过来。一个反直觉的经验是给模型的选项越少它的选择越可靠。不要把所有工具都塞进上下文工具不是越多越好。4.3 防跑飞超时、幂等、费用限额一个都不能少生产环境最怕的不是Agent能力不够而是它不受控制地跑飞疯狂烧token、反复调用工具、把下游系统打崩。我的防跑飞三板斧第一单次调用和整个循环都要有超时。单次模型调用可以容忍慢但绝不能无限慢整个循环更要有外墙。我常用的配置是单次模型调用60秒、整个Agent循环150秒超时就返回预设的兜底话术。第二写操作工具必须幂等。Agent循环一旦重试写操作可能被执行两遍用户就多了两个重复订单这种事故一次就够你记住一整年。最简单的做法是给写操作加一个request_id参数每次发起时自动生成下游用它做去重。第三按会话设置费用上限。每一步循环都累加token消耗超过阈值就主动停止告诉用户这个问题超出了我可承担的处理成本请简化后重试。不要觉得设一块钱太少失控循环烧掉的钱会远超你的想象第六章我会算一笔实际的账。5. 从本地Demo到生产环境部署链路和可观测性改造本地跑通和生产可用之间有一条很宽的大河绝大多数项目是在这里没的。不是说把代码扔到服务器上就叫生产而是系统能扛住真实用户、真实流量、真实故障。这一章是我做过的最重要的三层改造。5.1 生产环境跑Agent和本地跑最不一样的三点本地跑Agent你是一个人用一个会话内存里一个dict就够。生产环境面对的是几十上百个并发用户每个人的会话必须隔离不能串。这是第一点会话隔离。本地你只有一个消息列表生产里每个用户有自己独立的上下文你不能把所有用户的对话塞进同一个全局变量。第二点是长任务不阻塞HTTP。一个Agent循环动辄几秒甚至几十秒如果直接用HTTP同步请求处理用户那头的请求早超时了Web服务线程也会一堆堆占满。生产方案是把它放到后台任务队列里用户提交请求后先拿到一个任务ID前端轮询或者通过消息通道等结果。第三点是可观测。本地跑错了可以打印日志看一眼生产环境出问题你连用户当时的上下文都看不到没有完整追踪就只能靠猜。这三点是Agent项目上线的高速公路护栏不做的话系统上线满一周就会被迫回头补课。5.2 无状态化改造把会话和记忆从进程里搬出去我建议的第一刀就是砍掉Agent进程里的所有本地状态。改造前Worker进程里挂着一个大字典存着所有用户的消息列表。 改造后消息列表存到独立的内存数据存储里每个会话一个key任务队列收到新任务再把它读出来跑完再写回去。这样Worker可以自由扩缩容、滚动重启任何一个实例挂掉都不会丢会话。核心流程的伪代码如下def handle_task(task): session_id task[session_id] messages kv_store.get(fsession:{session_id}) or [] messages.append({role: user, content: task[user_input]}) final_reply run_agent(messagesmessages, toolsload_tools(), ...) messages.append({role: assistant, content: final_reply}) kv_store.set(fsession:{session_id}, messages, ttl60 * 60 * 24) return final_reply注意Agent循环跑到一半时的中间消息也要及时写回存储。为什么用户刷新页面、Worker崩溃都可能让跑了一半的任务彻底丢失用户还得从头解释需求体验非常差。你只需要在每轮循环结束时调用一次写入成本几乎可以忽略收益却很大。5.3 记录每一次推理日志、追踪和费用控制生产环境的Agent调试基本靠日志活命。我给每个会话都生成一个trace_id让用户的请求、后台任务、每一次模型调用、每一次工具执行都带上同一个ID。这样用户报问题时你只要拿到trace_id就能把他在那个会话里的所有推理过程拉出来回放。一条生产日志至少要有这些字段字段说明trace_id / session_id链路追踪与会话隔离step当前是第几轮循环rolesystem / user / assistant / toolmessage摘要消息过长时截断保存tool_name与参数调了哪个工具、传了什么参数tool_result摘要工具返回结果失败原因必须记录token用量prompt_tokens / completion_tokens耗时这一步花了多少毫秒最终答复/兜底原因是正常收尾还是被强制终止这套日志最大的价值不只是排查bug更是费用与效果归因。定期跑一次统计平均每个会话消耗多少token、哪些工具调用占比最高、哪类问题经常走到max_steps兜底。有了这些数据你才有依据去优化提示词、裁剪工具、调整步数上限而不是凭感觉乱改。5.4 上线前发布检查清单每次Agent项目上线前我照着这张单子过一遍会话存储是否独立于进程重启会不会丢状态Agent任务是否已放入后台队列HTTP请求会不会被长任务拖死每个会话的并发是否做了控制同一个用户连续发两条消息上下文会不会互相覆盖模型调用有没有超时、重试、步数上限工具有没有失败兜底返回下游异常会不会把脏数据写进会话是否记录了完整trace日志能按会话回放全过程是否按会话或用户设置了费用上限和限流系统提示词和工具列表是否做了版本管理改动可回溯前面几项偏工程后面几项偏治理。新项目上线前哪怕只过了前四项也已经比大多数能跑的Demo强了一个层级。6. 我在真实项目里踩过的坑和对应的止损办法最后这部分是我最想分享的。前面讲的是应该怎么做这里讲讲我当时做错了什么。6.1 工具返回的脏数据污染了模型的后续判断我们早期有个查库存的工具在查不到时返回{result: null}这样一条看似无害的数据。问题来了模型看到null经常把它当成一个叫null的仓库名继续推理然后一本正经告诉用户null仓库显示无库存。这个错误非常隐蔽提示词、循环、上下文看起来都没问题纯粹是工具返回值格式设计得不够严谨。我现在的做法是两条硬规则工具返回必须带明确的success字段失败时用error字段说明原因系统提示词里加一句工具返回失败时直接告诉用户暂时无法获取该信息不要推测。记住模型相信工具返回的任何字符串脏数据就是脏逻辑。6.2 多轮会话里上下文越攒越脏Agent开始自己编历史项目早期没有做记忆压缩把所有消息一直堆在上下文里。跑到第十几轮模型开始出幻觉把前面某次工具调用失败记成了成功把用户已经放弃的方案当成正在执行。原因也很好理解旧对话里的失败信息、模棱两可的表述一直在上下文里反复搅动。止损办法就是4.1里的压缩和剪枝。但我要额外提一个点压缩摘要时一定要带上已放弃方案否则模型容易在后续轮次重新提出用户已经否掉的方案显得特别没记性。这个细节我在第一次压摘要时漏掉了结果线上用户跟Agent在一个问题上兜了好几圈因为它第三次推荐了同一个已经被否定的选项。6.3 用户一句话触发了十分钟的Agent马拉松有一次用户问了一个跨多个工具的大问题我们没有设费用上限Agent在循环里连续调用了将近二十次工具前后跑了快十分钟token烧了一大把。用户早就在页面上关掉了留下一张离谱的费用账单。这次的教训很直接Agent不是越能跑越好能及时停下是更稀缺的能力。现在的兜底策略是三步以内搞不定的复杂需求Agent会主动向用户确认优先级和范围而不是默默一路狂奔同时max_steps和总耗时限制写死在配置中心任何改动都要走审批。6.4 成本失控要算总账不能只看单次很多人只关注一次调用多少钱忽略了一个Agent循环里会有多少步。我统计过我们早期某个Agent平均一个会话要调模型8到10次加上工具调用token消耗是单纯问答场景的十倍以上。单次价格再低乘以十再乘以日活用户账单就会非常难看。我的止损方案是在观测层把费用折算成每会话成本每天出报表超过阈值当天告警。同时把一些高频、结果稳定的工具做缓存比如天气查询一小时内的结果直接复用避免模型重复调用。省下来的钱是一方面更重要的是让团队重新把注意力放回这个Agent到底值不值这个问题上。说实话这些坑没有一个是技术太难造成的全是工程意识不到位。Agent的代码本身不难难的是你愿意在它失控之前设好边界在出问题时能快速看到过程。这也是我一直强调手写循环、强调可观测的原因——你看到的每一步才是你最终能控制的每一步。按这条路径走下来你大概率会经历几个阶段头三天觉得不过如此一周后开始撞上上下文污染半个月后突然理解为什么要有上限和兜底。我个人的体会是Agent系统真正的门槛不在让它会而在让它收敛、可追溯、敢上线。所以脚手架搭好之后先别急着加工具去接一个真实的数据源跑一周把日志和坑记录下来。这条闭环跑完你再看各种框架和平台就再也不会被术语绕晕了。祝你的第一个Agent早点收口上线。