生产级AI Agent工程化:七要素与七个决策点
AI Agent这词在技术圈火了大半年真正上手做过生产级 Agent 的人其实没有想象中那么多。我接触过不少团队、也评测过不少开源项目大家的状态基本一样demo 跑得飞快一聊到工程实现就开始卡壳。所谓工程实现不是把 LangChain 调通就算完而是要考虑模型选型、工具设计、记忆策略、并发承载、失败恢复、可观测性这一整条链路。这篇就把我在多个 Agent 落地项目里沉淀下来的一套方法写透先说清楚 Agent 内部的七个要素再讲落地时必须拍板的七个决策点中间穿插代码示例和踩坑经验。适合正在搭 Agent 原型、或者准备把 Agent 推上生产的开发者和架构师参考。1. 重新理解 AI Agent从调模型接口到构建自主系统1.1 一条命令和一套系统的差别在哪里普通程序是确定性的你写了一段逻辑给定输入必然得到预期输出。Agent 不一样它内部有一个大模型在理解任务、拆解步骤、决定下一步动作这一环从根上决定了它不是一个普通函数而是一个由模型驱动、围绕目标循环运转的自主系统。用一个最通俗的类比传统程序像一条流水线每个工位的动作都是预先定死的Agent 更像你刚招进来的实习生——你告诉他目标他根据自己的理解查资料、写草稿、做校验中途遇到问题还会自己调整方案你只需要在关键节点验收结果。这个自主性既是价值所在也是所有工程难题的来源。因为模型每一步的输出都有概率性同一个任务跑十次执行路径可能完全不同。这意味着我们不能拿单元测试思维去定义一个 Agent 的成败而是要拿系统设计思维去设计它的结构、约束和反馈机制。1.2 工程实现为什么会卡壳很多人初次跑通 Agent 之后下一步就懵了本地跑得好好的放到生产环境里各种问题全冒出来。我总结下来卡壳的根源集中在四件事第一是不确定性无法消除模型理解偏差、工具选错、输出格式漂移这些都是常态第二是成本不可预估Agent 每一步都会调用模型Token 消耗可能远超预算第三是执行路径不可控你不知道它会绕多远到达终点甚至不知道自己其实已经偏了第四是排错极度困难传统程序有清晰的调用栈Agent 的行为却是一连串模型判断的结果出了幻觉你连在哪一步开始错的都找不到。所以我把Agent 工程化定义成一句话在不牺牲太多能力的前提下给模型的自主行为套上约束、校验、回退和观测框架让它从能跑变成稳定跑。接下来的七要素就是搭建这套框架需要的最小组件七个决策点则是这套框架在不同项目里必须回答的取舍问题。2. 七要素生产级 AI Agent 的工程骨架2.1 快速看懂七要素清单我拆过不少 Agent 项目最后发现不管背后用的是 LangChain、LangGraph、Spring AI 还是自研编排框架只要是能稳定跑起来的 Agent内部几乎都有七个组成部分。把它们列出来就是一张非常好用的工程自查表要素核心职责工程选型要点任务定义明确目标、输入、成功标准成功标准必须可判断否则 Agent 会自我感动LLM 基座推理、理解、生成关注函数调用能力、上下文有效长度、稳定性提示词与指令策略限制行为方式、约定输出格式系统提示词 few-shot贴近真实输入工具集Agent 接触真实世界的唯一通道工具描述质量 工具数量记忆系统保存对话上下文和长期知识短期靠窗口管理长期靠向量检索规划器决定下一步做什么常见模式有 ReAct、Plan-and-Execute执行与反馈环调用工具、回填结果、终止判断必须防死循环、防重复执行2.2 逐个拆解职责、要点、常见坑任务定义是最容易被忽视的要素。很多人把 Agent 跑不好归咎于模型太笨但真正的问题往往是任务描述太含糊。比如帮我处理一下这份文档Agent 不知道处理是要总结、翻译、校对还是提取字段。工程上合格的任务定义至少要包含目标是什么、输入是什么、成功标准是什么、哪些是绝对不能做的边界。尤其是成功标准模型要靠它来判断自己有没有完成任务没有明确标准它就会在差不多的时候擅自宣布完成。LLM 基座是算力的核心。做工程选型时只盯着榜单分数是不够的更值得看的是三个硬指标函数调用Function Calling稳不稳、上下文窗口在长文本下会不会掉效果、推理价格和延迟能不能撑住业务量。同一个模型处理短任务和长任务的表现可能差很多实际使用时要压测候选任务集而不是跑一两个高光样例。提示词与指令策略决定了模型的行为边界。系统提示词要写明角色、约束、输出格式、权限范围必要的时候提供 few-shot 示例尤其当工具很多、容易选错时几个真实案例比一大段规则更管用。这里有一个常见的误区把提示词写得越来越长、越来越像法律条文结果模型反而被冗长指令干扰表现更差。提示词的核心不是详尽而是辨别度高、可执行。工具集是 Agent 的手和脚。一个 Agent 能做什么取决于挂载了哪些工具工具能不能被正确调用取决于描述和参数定义是否清晰。我做过实验同一个工具用查询商品价格返回数字这种朴素描述时模型偶尔会选错改成查询指定商品的最新价格入参为商品名称返回价格文本适合在商品比价场景中使用之后调用准确率立刻上了一大截。写工具描述要记住一句心得让模型在十米外就能看出这个工具是干什么用的。记忆系统负责跨轮保存信息。短期记忆就是上下文窗口但窗口是有限的长期记忆则靠向量库和摘要把历史信息压缩保存。工程上最怕的是把记忆当成越多越好实际上取回不相关的记忆反而会误导模型。好的记忆系统应该像人脑一样只取用当前任务相关的信息无关的全都不要带进上下文。规划器决定了 Agent 的工作路径。最经典的是 ReAct 模式思考Reason→ 行动Act→ 观察Observation→ 再思考循环直到完成目标。复杂任务还会用到 Plan-and-Execute先拆计划再逐步执行或者结合 Reflexion 做失败反思。选择哪种规划模式取决于任务的开放程度任务越开放越需要弹性规划任务路径明确就越应该把它固化成流程减少模型自由发挥的空间。执行与反馈环是把模型思考落地为真实动作的最后一公里。模型决定调用某个工具后工具执行结果要回填给模型让模型基于新信息做下一步判断。这个环节要处理三类问题工具调用失败时怎么反馈、怎么避免同一个失败操作被反复重试、以及任务完成到什么程度就该终止。后面我要说的可观测性和安全护栏本质上就是在为这个环节兜底。2.3 一个例子看懂七要素怎么协同拿查询光伏组件实时价格并生成对比表这个需求举例。任务定义要素负责把用户意图拆成查价格、选可比字段、生成表格三个子目标LLM 基座理解用户需求并产出中间推理工具集里挂了搜索价格和生成表格两个工具记忆系统保存了用户之前提到的厂商偏好规划器决定先搜价格、再做筛选和排序执行与反馈环把搜索结果回填给模型最后输出一张表格文件给用户。整个过程中任何一个要素缺失都会出问题没有任务定义Agent 不知道表格要不要含运费没有反馈环搜索结果出来了模型却不知道下一步该干嘛。七要素不是并列的模块而是首尾相接的闭环。3. 七个决策点原型到生产级的每一道岔路3.1 决策顺序为什么比决策本身更重要如果说七要素是 Agent 的积木那七个决策点就是搭积木时的选择。我在实际项目里最大的体会是决策的顺序往往比决策本身更致命。很多人一上来就选模型、选框架结果业务场景一变前面所有选择都要推翻重来。我自己在做项目时会按照下面这个顺序逐个踩点决策点要回答的问题核心取舍一、场景边界这个任务真的需要 Agent 吗自主性 vs 可控性二、模型选型用哪个模型、哪种部署方式效果 vs 成本 vs 合规三、架构模式单 Agent 还是多 Agent如何编排能力 vs 复杂度四、工具设计暴露哪些工具、参数怎么定义覆盖面 vs 准确性五、记忆策略保留什么、遗忘什么、从哪里取回丰富度 vs 干扰六、并发与性能怎么扛住线上流量吞吐 vs 成本 vs 延迟七、可观测性与安全出问题怎么查、风险动作怎么拦自由 vs 护栏3.2 决策一与决策二场景边界、模型选型先泼一盆冷水不是所有场景都适合上 Agent。适合 Agent 的任务一般有三个特征目标可以被清楚描述、行动空间相对受限、允许在过程中犯错并重试。反过来如果业务流程是固定的、对实时性要求极高、或者某些动作一旦做错就不可挽回那就要谨慎。比如自动发布营销消息、自动执行交易等场景Agent 只是大脑真正的执行环节必须加人工审批或风控闸门。这个判断会在最开始替你省下大量返工成本。决策二才轮到模型选型。这个环节我不太建议只看跑分更务实的维度是效果是否满足任务下限、单次任务综合成本有没有超预算、数据能不能合法出境或上云。APIs 与模型服务的函数调用一般比开源模型更稳但自托管开源模型在成本和数据控制上有优势。还有一个容易忽略的坑厂商标注的上下文长度不等于有效长度。很多模型在长上下文下的推理精度明显下降工程上应该按有效工作区间来设计而不是硬顶到窗口上限。3.3 决策三与决策四架构模式、工具设计架构模式是单 Agent 还是多 Agent这个选择直接决定了项目的复杂度。单 Agent 适合目标单一的中等复杂任务开发成本低、调试容易多 Agent 适合流水线很长、或需要明确角色分工的复杂场景比如一个负责拆解任务、一个负责写代码、一个负责审查。但多 Agent 的 Token 消耗、上下文管理、结果串联复杂度都会成倍上升。团队如果刚起步我强烈建议先跑通单 Agent确实压不住了再拆分。编排框架方面LangGraph 这类偏状态机的方案会比纯链式调用更稳Java 生态的团队用 Spring AI 做底层集成性能敏感或需要近实时计算的模块也可以用 Rust 编写核心服务再通过中间层对接编排框架。工具设计里面最值得花时间的是给每个工具写清楚 JSON Schema。模型能不能正确传参很大程度上取决于你在 Schema 里有没有把每个参数的类型、含义、取值范围写明白。比如一个发送消息工具只写 chat_id 和 content 两个字符串参数是远远不够的要说明 chat_id 从哪查出来、content 有没有长度限制、调用前需不需要二次确认。工具越少越好但每个工具都要像一个被仔细打磨过的 API 一样描述清晰、参数有界、错误可解释。3.4 决策五与决策六记忆策略、并发与性能记忆策略是周期遗忘的艺术。短期记忆靠窗口管理比如只保留最近 N 轮对话超过的部分用摘要压缩长期记忆靠向量库但检索质量直接决定记忆有没有用。这里有一个独家经验优先做按需取回而不是全量塞入。每个任务开始前先用用户当前的问题去向量库检索 Top K 条相关历史再把它们合入上下文。这样记忆的投入产出比最高也不容易干扰模型判断。第六个决策点也就是很多人搜的AI Agent 怎么扛并发。拆开看主要有四件事模型调用是 IO 密集型的天然适合异步化同一类请求可以靠语义缓存减少模型调用次数对模型 API 的 rate limit 要做限流和指数退避重试最后耗时的多步 Agent 任务不要同步阻塞 HTTP 请求应该起一个后台任务或消息队列前端轮询状态。技术栈上FastAPI 的异步接口 Redis 队列 后台 Worker 是性价比很高的组合Java 团队则可以用成熟的 MQ 体系。这部分在下一章会有更具体的代码演示。3.5 决策七可观测性与安全护栏最后一个决策点往往决定项目能走多远。Agent 的执行路径不确定传统日志打印根本看不出来问题出在哪一步。我给项目搭建可观测性时坚持为每次 Agent 运行记录完整执行路径、每一步的模型输入输出、工具调用参数和返回结果、Token 消耗和耗时。有了这些 trace出问题才可以在几分钟内定位到到底是模型理解错了、工具选错了还是反馈环节丢信息了。安全护栏的核心是默认信任关键动作加闸。凡是对外部产生真实影响的动作——发消息、下单、删除数据、转账——都必须有白名单约束和人工确认机制。宁可让 Agent多问一句而稍微损失效率也不要让它拿到太宽的授权之后闯出大祸。我在多个项目里反复验证过一句话Agent 的能力越强护栏的优先级越高。4. 实操落地基于 FastAPI LangChain LangGraph 的最小生产骨架4.1 技术栈怎么选、为什么是这三件套市面上 Agent 框架不少但我最常用也最推荐团队起步的组合是 FastAPI LangChain LangGraph。为什么不直接用轻量封装因为 Agent 的难点恰恰不在调用模型而在管理多步状态和分支逻辑。LangGraph 天然用状态图的方式来描述 Agent每一步是一个节点节点之间用边连接条件边决定下一步走向——这正好和我们说的执行与反馈环对应。LangChain 负责把模型、工具、提示词统一抽象起来FastAPI 则负责把整个 Agent 暴露成 HTTP 服务并承接并发请求。这套组合最大的好处是结构清晰、每一步都可观测、适合渐进式扩展。4.2 Agent 状态机代码骨架先跑通一条最短路径下面这个例子是一个最小可用的查询价格 Agent。它演示了三件事状态定义、模型节点、工具节点、条件终止。核心逻辑是模型判断需要调工具时就进入工具节点工具执行完把结果回传给模型直到模型判断任务完成。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import ToolMessage class AgentState(TypedDict): messages: Annotated[list, operator.add] steps: int tool def query_price(product: str) - str: 查询指定商品的最新价格。入参为商品名称返回价格与更新时间。 # 真实场景里这里可以接内部价格服务、数据库或第三方 API return f{product} 的最新价格为 100 元2025-06-01 10:00 更新 def call_model(state): llm ChatOpenAI(modelgpt-4o, temperature0).bind_tools([query_price]) response llm.invoke(state[messages]) return {messages: [response], steps: state.get(steps, 0) 1} def call_tools(state): last_message state[messages][-1] new_messages [] for tool_call in last_message.tool_calls: result query_price.invoke(tool_call[args]) new_messages.append( ToolMessage(contentresult, tool_call_idtool_call[id]) ) return {messages: new_messages} def should_continue(state): last state[messages][-1] if last.tool_calls and state.get(steps, 0) 5: return call_tools return END graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.add_node(call_tools, call_tools) graph.add_edge(call_tools, call_model) graph.add_conditional_edges( call_model, should_continue, {call_tools: call_tools, END: END} ) app graph.compile()几个值得注意的细节steps字段是死循环熔断没有这个字段模型反复要求调用工具时会无限循环直接把 token 耗尽ToolMessage的tool_call_id必须和原始模型输出的id对齐否则模型无法把工具结果和之前的调用对应起来条件边里的分支名必须和图里的节点名严格一致写错时 LangGraph 会在运行时抛异常。4.3 并发不拖垮服务的三个关键动作第一个关键动作是异步化。Agent 内部要调模型、要调外部工具整个流程耗时常常在几秒到几十秒。如果在同步接口里跑 Agent服务并发一上来必然被打爆。我的做法是接口层只负责接收请求、生成任务 ID、把任务丢给后台执行然后立刻返回。前端或调用方通过任务 ID 轮询状态接口拿到最终结果。from fastapi import FastAPI, BackgroundTasks import uuid app FastAPI() task_store {} # 单机演示用生产环境必须换 Redis app.post(/agent/run) async def run_agent(request: RunRequest, background_tasks: BackgroundTasks): task_id str(uuid.uuid4()) task_store[task_id] {status: running, result: None} background_tasks.add_task(run_agent_job, task_id, request) return {task_id: task_id} app.get(/agent/status/{task_id}) async def get_status(task_id: str): return task_store.get(task_id, {status: not_found}) async def run_agent_job(task_id: str, request: RunRequest): try: result await agent_execute(request) task_store[task_id] {status: done, result: result} except Exception as exc: task_store[task_id] {status: failed, error: str(exc)}第二个关键动作是信号量限流。即使有后台任务也不能无限放行 Agent 请求。每个模型 API 都有并发和每分钟调用数的限制超出就会触发 429。我用一个asyncio.Semaphore控制同时在跑的 Agent 数量超出的请求在队列里等待。semaphore asyncio.Semaphore(10) # 同时最多 10 个 Agent 任务 async def run_agent_job(task_id: str, request: RunRequest): async with semaphore: # 只有拿到信号量才会真正执行 Agent 主流程 result await agent_execute(request)第三个关键动作是语义缓存。同样的或者非常相似的用户问题没必要每次都重新走一遍 Agent。先把用户输入做 embedding再去缓存里做相似度匹配命中直接返回历史结果。这个策略在高频同类问题场景下效果极其显著实测能减少 30% 到 50% 的模型调用量。注意只有对结果一致性问题才能开语义缓存如果每个问题都要求最新数据或者涉及个性化结果缓存就必须谨慎使用。4.4 上线前自检清单代码能跑只是第一步。我每次把 Agent 推上线前都会过一遍下面的清单这些是用真金白银换来的经验死循环保护steps 上限、超时时间、最大 Token 预算三个最少有其一。失败重试策略工具调用失败时是否有明确反馈给模型重试次数有没有上限。成本估算按单次任务的 token 消耗乘预估调用量算出来的数字能不能被业务接受。权限白名单Agent 能调用的工具和 API 是否是最小集高风险动作是否有审批环节。trace 日志每一步的模型输入输出、工具参数、耗时是否都落盘。并发压测模拟真实调用量跑一轮压测确认限流和队列策略是有效的。5. 常见问题与排障实录5.1 高频故障速查表问题现象可能原因排查思路解决办法Agent 陷入死循环缺少终止条件或无步数上限查看 trace 中反复出现的动作加 steps 上限、加最大 token 预算模型输出格式错乱上下文被无关内容干扰检查最后几轮信息的来源清理历史记忆、加格式校验与重试工具选择错误工具描述缺乏辨别度对比同一输入在不同描述下的表现重写工具描述提供 few-shot 示例上下文溢出历史消息过多看请求前 tokens 统计裁剪窗口、摘要压缩、向量检索取回API 频繁 429并发超过模型限流检查日志里 429 出现的时间分布加信号量限流、指数退避重试输出明显幻觉任务无明确成功标准让模型复述自己的执行结论增加结果校验、关键数据人工确认并发时用户数据串扰全局状态被多个会话共享通过 trace 对比不同会话的上下文按会话隔离状态不要用全局变量5.2 我踩过的四个坑第一个坑是全局状态串扰。最早用 LangGraph 做多用户服务时我把状态缓存放到了模块级全局变量里结果用户 A 的对话历史出现在用户 B 的上下文中。排查了半天才发现是共享引用的问题。从那以后我所有 Agent 的状态管理都强制要求会话级隔离每个请求进来都创建一个独立的状态副本绝不复用。第二个坑是工具描述写得太文艺。早期给工具写描述时讲究简洁优雅比如获取天气数据结果模型经常在多个天气工具之间选错。后来我把描述改成结构化模板动词 宾语 适用场景 数据来源 返回内容准确率立刻改善。现在给团队培训时我一定会强调工具描述是给模型看的接口文档不是给人看的诗。第三个坑是上线初期没有 trace。有段时间 Agent 在线上偶尔给出错误答案但我连它执行了哪几步都不知道只能在日志里大海捞针。补上完整 trace 之后问题定位时间从半天降到了十分钟。现在我甚至连每次模型调用的 token 数都记录这个数据对成本优化太重要了。第四个坑是流式输出与多步执行冲突。最初想给用户打字机效果直接在 Agent 执行时做流式输出但多步 Agent 会在工具调用之后出现输出中断、节奏奇怪。后来改成两段式Agent 后台执行时通过 SSE 推送当前正在执行哪一步的过程状态最终结果一次性输出。用户感知比硬做 token 流式好很多实现也简单。6. 几句实在话把这些项目做下来我最大的感触是Agent 工程不太像传统后端更像是在训练和约束一个有判断力的执行者。你交给它的不是一整套写死逻辑而是边界、工具、反馈和护栏。七要素是它的器官七个决策点是它的成长路径。如果只能给一条建议我会说先别急着搭复杂的多 Agent 架构老老实实把一个单 Agent 的七要素配齐、七个决策点踩完把最短路径跑稳再去想扩展。Demo 做得好是模型的本事生产跑得稳才是你的本事。这条路上没有银弹但有了这套框架至少每一步都知道自己走到了哪里。