AI Agent工程落地：七要素定下限，七个决策点定上限

上半年我密集接触了十几个 AI Agent 项目有给知识库配助手的有给客服系统做自动应答的也有想用 Agent 做行情分析、自动化报表的。结果特别一致Demo 阶段人人兴奋一接真实业务数据就集体熄火。不是模型不够聪明是大多数人把 Agent 当成一个更长的提示词来做忘了它本质上是一套要长期运转的软件系统。这篇就聊聊我理解的 AI Agent 工程实现七要素决定下限七个决策点决定上限。看完你至少能回答三个问题——我的 Agent 该由哪些部分组成、每一步该怎么拍板、上了生产环境之后会遇到什么。1. 为什么这么多 Agent 项目跑不起来Demo 与生产的鸿沟先说我观察到的现象。很多团队搭建 Agent 的第一版只花了一周调提示词调到看起来会自己干活然后直接上生产。结果是什么要么在第五轮对话时突然失忆要么调用工具传错了参数要么一个循环任务跑了几十次没有终止条件账单先爆了。这些问题没有一个是大模型本身造成的全是你周边工程没跟上。1.1 把 Agent 当提示词工程做是最常见的误判我问过不少开发者你的 Agent 架构是什么得到的回答大多是就是调 OpenAI 或者 Claude 的接口写个 system prompt然后把用户问题丢进去。这种思路做聊天机器人没问题做 Agent 必翻车。原因很简单Agent 的核心特征是它会自主调用工具、多步推理、根据中间结果调整下一步行动。这意味着你的系统里必须有计划执行观察再计划的循环。提示词只能让模型想不能替你处理循环中的状态、异常、超时和追溯。真实的 Agent 项目里模型能力只占一半另一半是流程控制、数据存取、工具协议和观测手段。1.2 Agent 的本质一个有状态的异步任务系统我把 Agent 理解成一个更聪明的任务执行引擎。它接收一个目标帮我把这个月的销售数据汇总成报告然后自己拆解出查数据→清洗→分析→生成图表→输出报告这些步骤每一步调用不同工具最后交付结果。这套流程跟人工执行的流程没有本质区别区别在于执行者是模型加代码。既然是个任务系统就绕不开几个经典工程问题任务状态怎么存步骤失败了怎么重试多个请求并发进来怎么不互相踩踏输出怎么流式返回给用户这些问题的答案就是下文七要素和七个决策点要解决的内容。记住一句话Agent 不是一个模型而是一条流水线。2. 七要素拆解一套 Agent 要能干活最少需要哪些部件我把一个可运转的 Agent 系统拆成七个部件缺一个生产环境都会出问题。要素职责生产环境常见痛点大模型底座理解、推理、生成决策输出不稳定、JSON 解析失败记忆系统短期上下文与长期知识存取上下文爆炸、旧信息覆盖新信息规划引擎任务拆解、决定下一步动作规划循环不终止、拆解过细工具层连接外部系统和数据源参数格式错误、权限过大反馈闭环将执行结果喂回决策链路错误被吞掉、结果没校验状态管理会话、任务、进度的持久化多实例下状态丢失安全与边界防注入、限流、预算控制不加边界直接被薅羊毛下面逐个说。2.1 大模型底座所有决策的源头模型选型直接决定 Agent 的能力上限。工程上不只看榜单分数要看三点是否支持稳定的工具调用function calling、结构化输出是否可靠、上下文窗口多大。以我常用的几个模型为例Claude 系列和 GPT 系在工具调用上都很成熟DeepSeek 在中文场景性价比很高本地部署的 Qwen 系列适合私有化。实操里有个很容易踩的坑模型返回的工具调用参数是字符串必须做 schema 校验。我习惯在 Prompt 里直接要求模型输出严格 JSON再用 Pydantic 做模型校验不合格就让模型重新生成最多重试两次。千万别把模型的字符串输出当结构体直接用。2.2 记忆系统短期上下文与长期知识分开管很多 Agent 项目死就死在什么都在聊天历史里。短期记忆就是当前任务的上下文一般放在模型上下文窗口里。但窗口是有限的对话长了必须压缩。我的做法是分层最近的 10 轮对话原样保留更早的对话交给模型做摘要摘要再存进上下文。这个策略能有效防止上下文爆炸。长期记忆则放到外部存储里。用户偏好、历史订单、知识库条目都应该是可检索的。工程上用向量数据库Milvus、pgvector、Qdrant做语义检索或者干脆用普通数据库按 key 存取。甚至很多场景用 Redis 就够。关键不是工具多高级而是想清楚什么信息需要长期存、怎么在合适的时机被取回。注意长期记忆收回来的内容也要有优先级。我见过一个项目把用户一年前的偏好顶到最前面结果 Agent 每次回答都推荐过期的产品用户直接投诉。2.3 规划引擎思考和执行要分开规划引擎是 Agent 区别于普通问答系统的关键。它负责把大目标拆成小步骤。工程实现上主要有两种模式ReAct 式动态规划和预定义工作流。ReAct 的模式是思考→行动→观察→再思考适合开放性任务比如帮我调研一下这个行业。但它的缺点是可能陷入循环所以必须设置最大步数比如 10 步和终止条件。预定义工作流则是把固定业务路径固化成代码比如查订单→查物流→生成售后单适合规则明确的场景。生产系统里最稳的做法是两者结合能用流程解决的走流程流程覆盖不了的再让模型动态规划。2.4 工具层模型与外部世界的桥梁工具层是 Agent 的手脚。每个工具其实就是一段函数有输入输出定义、有错误处理。工程实现上最普遍的做法是把函数名和参数 Schema 一起发给模型模型决定调用哪个、参数填什么。这个机制在很多模型里原生支持叫 function calling 或者 tool use。工具层的坑通常在两方面。一是参数描述写得不清不楚模型传了错误参数二是工具输出没有统一返回格式后续解析全靠猜。我建议每个工具返回一个标准结构至少包含是否成功、业务数据、错误信息。这样无论工具内部多复杂Agent 的规划层只认这一种返回格式后续处理就简单了。2.5 反馈闭环与状态管理别让 Agent失忆模型本身是无状态的每次调用都是独立的。这意味着你必须把当前进行到第几步、已经拿到什么结果、下一步该做什么存在外面。这就是状态管理。LangGraph 这类编排工具解决的就是这个问题它把 Agent 流程画成一张图节点是规划执行判断图会持有状态每一步的结果会更新状态。如果你不用框架就得自己在数据库里维护任务状态表任务一多很容易乱。状态管理的核心设计原则是状态里只放必须跨步骤保留的数据临时变量别塞进去否则图会越来越复杂。2.6 安全与边界最后一环最容易被忽略Agent 比普通 API 更危险因为它手里有工具。你的工具可能包含发送短信删除订单转账这类高权限操作。生产环境里必须加权限边界哪个用户能用哪个工具、高权限操作是否需要人工审批、单任务的 token 预算上限是多少。还有一个经常被忽视的点提示词注入。外部数据可能携带恶意指令试图让 Agent 执行不该做的操作。工程上的缓解手段是把外部输入当数据处理不直接拼进 system prompt工具执行前做参数校验涉及资金和隐私的操作一律走人工确认。3. 七个决策点从要素到实现必须拍板的取舍七要素是必须有什么七个决策点是每一步怎么选。同一个 Agent 目标选型不同成本、稳定性、开发周期能差出十倍。下面这七个决策点是我每次搭 Agent 前都会过一遍的清单。3.1 决策一单体 Agent 还是多 Agent要不要用编排框架先回答最基础的问题这个任务用一个 Agent 干还是拆成多个专职 Agent 协作我的经验是能用单体的就别上多智能体。多智能体听起来高级但每一个 Agent 都是一份 token 开销和一个新的失败点Agent 之间的通信协议也需要设计。只有当一个 Agent 需要的能力差异过大比如一个做代码、一个做设计时才值得拆。框架选择上我的建议很实际小项目或验证阶段直接用裸代码加状态机就能跑需要复杂编排、分支、人机确认时上 LangGraph团队对 Java 栈有要求可以考虑 Spring AI对性能和低资源占用敏感Rust 也是一个方向生态虽小但跑得很快。框架的价值是帮你管理状态和流程代价是引入抽象。别为了用框架而用框架。3.2 决策二主模型与子模型怎么分工预算怎么分一个 Agent 内部其实可以混用多种模型。主规划模型要强负责理解目标和拆解任务用最强最贵的子执行模型可以弱一点、便宜一点比如信息提取、摘要、格式化这些事用小模型就够了。我常用的组合是主模型用旗舰款工具调用和简单分类用中端款大批量文本清洗甚至可以用本地小模型。这样平均单次任务成本能降 40% 以上。但注意子模型的输出质量必须严格校验越便宜的模型越需要外部兜底。3.3 决策三记忆策略要存什么、取什么、多久失效这个决策决定了 Agent 是聪明地懒还是勤奋地傻。工程上要回答三个问题会话历史怎么管理、知识库检索怎么召回、用户画像要不要存。会话历史管理我建议三级策略最近对话原样保留→中间层摘要压缩→更早的按主题归档。知识库检索要控制召回数量一般取 top 3 到 top 5太多会让模型注意力发散。用户画像这类敏感信息要先确认是不是非存不可很多场景其实不用存每次让用户提供或者从数据库实时查就够了。3.4 决策四动态规划还是固定工作流边界画在哪这是我跟很多开发者反复争论的点。我的观点能写死的流程就不要让模型自由发挥。固定流程的错误率可以压到非常低动态规划是不得不开放场景的兜底。实操里我一般这么定边界如果任务的步骤是确定性的比如查库存→下单→通知写成代码流程如果任务没有固定路径比如帮我对比这几家供应商的报价才交给模型做 ReAct。还要设计一个逃逸机制模型发现当前任务超出能力范围时必须主动说我不确定而不是硬编。这比任何技巧都管用。3.5 决策五工具协议怎么定Function Calling 还是 MCP 还是直接执行代码工具层有三种主流接入方式选择取决于你的上下游生态。一是模型原生 Function Calling最简单定义一个 schema 列表传进去模型返回结构化调用请求。适合工具数量少、调用方单一的场景。二是 MCPModel Context Protocol把工具封装成标准服务端Agent 通过客户端统一发现和调用适合工具多、要复用、多 Agent 共享的场景。三是直接让模型生成代码再执行灵活性最高但风险最大只建议在沙箱里做数据分析类任务。我的建议工具少于 10 个直接用 Function Calling超过 10 个或者要跨团队共享工具时引出 MCP 是值得的投入。无论哪种工具的输出都必须做超时和限流控制。3.6 决策六并发承载与 token 成本模型先算账再动手这是生产环境最关键、也最容易在 Demo 阶段被忽略的决策。Agent 请求不是一次模型调用而是一个多步循环一个用户请求可能触发 5 到 20 次模型调用。这意味着你的成本不是一个 prompt 的成本而是一个流程的成本。并发方面我建议先做一件简单的事给每次完整任务设置 token 上限比如单任务最多 5 万 token并在中间步骤做缓存。相同的工具调用结果、相同的检索结果命中缓存直接返回。此外能用流式输出就绝不等待完整生成用户体验和资源占用都能改善。关于怎么扛并发我后面专门用一章展开先记住一个公式系统并发能力约等于单实例连接数 ÷ 平均单任务耗时秒 × 3600 × 实例数。算完这个账再决定要不要上队列和异步架构。3.7 决策七可观测性与评估基线你拿什么证明 Agent 在干活没有观测就没有优化。Agent 链路比普通接口长得多一步错后面全错所以必须记录每个步骤的输入输出。我见过的生产事故里大量是三四个小时前出的错因为没留 trace根本查不到是哪一步开始编数据。工程上至少要落地三件事全链路 trace从用户请求到每一次模型调用和工具调用结构化日志每一步的 token 数、耗时、成本评估集定期用固定任务集跑回归防止模型升级或提示词改动后效果倒退。工具方面Langfuse、LangSmith、Arize 都可以做 trace自研的话至少把 prompt、响应、工具入参出参全部落库。4. 一次完整落地FastAPI LangGraph 搭一个能处理查订单催发货生成客服答复的 Agent概念讲完得有个能抄的案例。我选一个最常见的业务场景客服助手。用户可能说帮我查一下订单 JD2024001 到哪了这个订单能不能催一下把物流信息总结成一段话发给我。这三件事对应三个工具、两轮左右的规划刚好够展示七要素怎么落进代码。4.1 整体架构和状态设计我先想清楚状态里放什么。按前面的原则状态只放跨步骤数据用户 ID、当前意图、工具结果列表、中间答复。我用 LangGraph 的 State 来管理结构大概是from typing import List, Dict, Any, TypedDict class OrderAgentState(TypedDict): user_id: str messages: List[Dict[str, str]] # 对话历史 intent: str # 识别出的意图 tool_results: Dict[str, Any] # 各工具返回数据 final_answer: str # 最终生成的答复这个 State 会被图里的每个节点读写。LangGraph 的好处是它自动帮你做状态合并你不用手写 setter/getter。4.2 定义工具层接着定义两个工具查订单、催发货。每个工具都返回统一结构。def query_order(order_id: str) - dict: # 模拟调用订单系统 if order_id JD2024001: return {ok: True, data: {status: 运输中, location: 上海分拨中心, eta: 明天 18:00}} return {ok: False, error: 订单不存在} def urge_delivery(order_id: str) - dict: # 调用工单系统创建催单任务 return {ok: True, data: {ticket_id: TICKET_9001, status: 已创建}}这里我故意把返回包成是否成功 业务数据 错误信息的标准结构。后面规划层拿到这个结构无论查订单成功还是失败都能继续决策。4.3 规划节点和执行节点的图编排下面把识别意图→执行工具→生成答复串成一张图from langgraph.graph import StateGraph, END def planner(state: OrderAgentState) - OrderAgentState: # 调用模型判断用户意图决定调用哪个工具 # 实际场景中这里会把 intent order_id 写入 state intent detect_intent(state[messages][-1][content]) state[intent] intent return state def executor(state: OrderAgentState) - OrderAgentState: order_id JD2024001 if state[intent] 查订单: state[tool_results][order] query_order(order_id) elif state[intent] 催发货: state[tool_results][urge] urge_delivery(order_id) # 其他意图继续下钻 return state def responder(state: OrderAgentState) - OrderAgentState: state[final_answer] generate_answer(state[intent], state[tool_results]) return state graph StateGraph(OrderAgentState) graph.add_node(planner, planner) graph.add_node(executor, executor) graph.add_node(responder, responder) graph.set_entry_point(planner) graph.add_edge(planner, executor) graph.add_edge(executor, responder) graph.add_edge(responder, END)这个例子很短但结构就是生产级 Agent 的骨架一个节点负责思考一个节点负责执行一个节点负责汇总。真实项目里你会在 planner 前面加上要不要终结的条件边在 executor 后面加上工具失败重试的回边。这些加边逻辑LangGraph 里都是同一个套路。4.4 FastAPI 接入层把 Agent 包成一个接口最后包一层 FastAPI加超时、限流和 tracefrom fastapi import FastAPI, HTTPException app FastAPI() app.post(/agent/order) async def handle_order(request: dict): user_id request[user_id] message request[message] # 限制单次任务最多 30 秒、最多 8 个模型步 try: agent_result run_agent_with_limits(user_id, message, max_steps8, timeout30) return {answer: agent_result[final_answer], trace_id: agent_result[trace_id]} except TimeoutError: raise HTTPException(status_code504, detail处理超时请稍后重试)run_agent_with_limits 内部是把用户消息初始化进 state然后 graph.invoke(state)。超时和最大步数一定要在入口就卡死否则一个死循环任务能把你后端打爆。代码是简化版但流程是完整的。你要复刻的时候把 detect_intent 换成真实的模型调用、把 query_order 换成真实的 HTTP 请求就能跑。5. 并发、Token 与成本生产环境的三座大山好多人在热搜上搜ai agent 怎么扛并发我直接说结论Agent 的并发问题跟传统接口完全不同因为单次请求的耗时长、模型调用次数多、中间状态重不能按照短请求高吞吐的思路设计。5.1 先算清一次 Agent 任务的 Token 账Token 是所有 Agent 成本的基本单位。你调用一次模型账单按输入 token 加输出 token 算输入是上下文system prompt 历史 工具结果输出是模型回答。Agent 多步循环会不停累积上下文所以同样一个问题普通聊天可能花 2000 tokenAgent 可能花 2 万 token。我给自己定的基线做一个查订单→催发货→生成答复的三步流程每一步输入约 4000 token、输出约 500 token一次流程约消耗 (4000500) × 3 13500 token。如果主模型定价是输入 5 美元/百万 token、输出 15 美元/百万 token一次完整任务成本大概在 0.09 美元左右。日活一万用户、人均一天三次请求一个月成本你可以自己乘一下。这就是为什么我强烈建议给单任务设 token 上限并且能缓存的一定要缓存。比如同一条物流信息五个人各问一遍没有缓存就是五份开销加了缓存模型查一次引出五次和一次的成本几乎一样。5.2 为什么 Agent 天然比 Chat 难扛并发Chat 一次请求通常几十秒就能结束后端承载几百并发问题不大。Agent 一次请求可能长达一分钟期间要多次调用模型其中任何一步阻塞都会拉低整体吞吐。更麻烦的是Agent 是有状态的你不能像普通 REST 接口那样随便横向扩容——如果两个请求被路由到不同实例状态就散了。我的建议是把状态放进 Redis 或数据库让每个实例无状态化。这样扩缩容就变成一个纯粹的计算资源问题。请求进来后先把任务写入队列后端 worker 消费队列执行 Agent 流程前端通过轮询或 WebSocket 拿结果。这套架构扛并发能力远强于请求进来直接同步跑。5.3 扛并发的三条实用路线按投入从低到高排第一限流加超时。最少做这两个防止单用户请求拖垮全系统。按 QPS 限流按步数限制死循环按时间砍掉慢任务。第二缓存加异步化。对工具调用结果做 key-value 缓存把同步等待改为队列消费。这两件事做完相同业务量下成本能降一半左右。第三自动伸缩和多实例。基于消息队列的积压情况做实例扩缩容。这一步的上限最高但需要你先把第一、第二步做扎实。否则缩容只会暴露更多隐患。6. 让 AI 真下地干活调试、观测与评估写代码容易让 Agent 稳定工作难。这章节讲怎么证明你的 Agent 真的在干活——以及出了问题能不能快速找到病根。6.1 Agent 最常见的四种失败模式失败模式典型表现排查方向规划循环反复调用同一个工具、不终止检查最大步数、终止条件工具参数错把订单号传成用户 ID检查工具 schema 描述、模型重试逻辑上下文污染回答里出现旧任务的信息检查记忆压缩策略、状态隔离幻觉兜底查不到数据就瞎编检查工具结果校验、拒答指令这四种问题我在不同项目里反复踩过。一个共同教训它们几乎都能通过记录下每一步的输入输出来定位。你只要在某一步发现模型传给工具的订单号是空字符串根因基本就浮出水面了。6.2 从日志到 Trace排查链路的三层体系我在 Agent 项目里必做三层观测第一层应用日志。记录请求入口、意图识别结果、每个节点开始结束时间。这一层最快能看到卡在哪一步。第二层模型调用记录。把每一次调用的 prompt、response、token 数、耗时落库。这一层能看到模型在想什么。很多诡异行为一看完整 prompt 就明白了——八成是上下文里掺了不该有的东西。第三层全链路 Trace。把用户请求 ID 贯穿到每次模型调用、工具调用、数据库查询。工具有现成的我自己用 Langfuse 比较多自研也可以核心是把 Trace ID 透传到所有环节。6.3 建一套评估基线防止改一下坏一片Agent 没有集中的测试用例可跑所以你需要建自己的评估集。我建议准备三组测试功能测试每个工具单独调用是否正确、流程测试完整的端到端任务是否能跑通、鲁棒测试输入乱序、模糊、带干扰时是否还能工作。每次改 prompt、换模型、改工具 schema都拿这组测试跑一遍。你会发现很多这次改好了其实只是这次碰巧过了一两个 case。没有基线你就没法判断一个改动到底是改进还是回退。这个动作成本不高但收益是实打实的我靠它救回过两次看起来没问题、上线就出事的版本。回到最开始的问题——Agent 项目为什么容易跑不起来因为它根本不是一个提示词问题而是一个工程问题。七要素是地基七个决策点是承重墙地基和墙都搞清楚了上面住多少户都稳。我个人实操中的体会是每上一个 Agent 项目先别急着调模型花半天把要素盘点清楚、把决策点逐项过一遍比后面返工省出几倍时间。就这些希望对正在做 Agent 的你有点用。