最近半年我身边至少有三四个团队都在做 AI Agent有一个已经上线了有两个还在反复改工作流。聊到最后大家几乎都卡在同一个问题上——Demo 跑得很顺一上生产就到处漏风超时、乱调用工具、上下文越搞越长、成本失控。今天这篇就把我做生产级 AI Agent 工作流的思路、架构、代码骨架和踩坑记录完整摊开给真正要把 Agent 推到线上的人一个可复用的参考。这篇文章适合两类人一类是已经用大模型 API 做过原型、正准备往工程化方向走的后端开发者另一类是团队里要负责 Agent 系统设计和落地的人。我不讲概念只讲做法。1. 生产级 Agent 工作流卡点到底在哪1.1 Demo 和生产的真实距离我在很多项目里看到的 Agent Demo 通常是这么写的一段提示词一个循环让模型自己决定调哪个函数跑通一个用例就算完事。这种形态做内部演示、做技术验证完全够用但一旦面对真实用户立刻会遇到几个问题。第一个是行为不可控。模型不是每次都能做出正确的那一步尤其在多轮对话里它会突然决定调用一个无关工具或者在信息不足时强行编一个答案。Demo 里你拿着预期用例反复测觉得没问题是因为你恰好避开了边界场景。生产环境里用户的问题五花八门模型走偏的概率被无限放大。第二个是上下文膨胀。Agent 每轮都要把历史、工具返回结果、用户当前问题拼进提示词时间一长输入长度暴涨。我见过最夸张的一次上下文从 2 万 token 涨到 8 万 token单次请求成本翻了近三倍响应延迟也跟着崩了。第三个是链路断裂。工作流里调用外部工具网络抖动、接口限流、数据格式异常都是常态。如果每个环节都要人工介入那这个系统没法上线如果全部无条件重试又会引发连锁的重复扣费和资源占用。所以说生产级的核心不是模型能力多强而是整个工作流能不能像一个严肃的后端服务一样具备可观测、可控制、可回滚、可评估的能力。这是和 Demo 之间最本质的分水岭。1.2 先想清楚你的业务适合路由式还是自主式我一般把 Agent 工作流分成两种大模式一种是路由式也叫工作流式从用户进来开始每个分支、每一步做什么都是预先规划好的另一种是自主式Agentic模型在每一步自己决策下一步调用什么。路由式的典型例子是客服机器人加 RAG 检索。意图识别先分一遍走问答、工单、转人工每个分支内部是固定流程。它的优点是稳定、成本可控、每一步都能定位缺点是灵活性低没法处理完全没规划过的场景。自主式则适合任务探索性强、步骤不固定的场景比如数据分析助手、竞品信息调研、编程辅助。模型需要根据当前结果动态决定下一步查哪个库、调哪个接口。这种模式上限高但随之而来的就是决策不可控、token 消耗巨大。我的实际建议是不要一上来就做纯自主式。生产级系统最佳实践是“路由式为主、局部自主”。也就是主流程用确定的 DAG 编排每个节点内部如果需要再放一个自主决策的小 Agent。这样既保留灵活性又不会让整个系统失去骨架。我在后面给的代码骨架就是基于这个思路。1.3 生产级必须满足的四个硬指标判断你的 Agent 工作流有没有资格进生产我习惯看四个指标缺一个我心里就没底。稳定性端到端成功率要能稳定在 95% 以上概念是“链路某一个环节失败时系统能通过重试、兜底或降级完成回复”而不是直接报错给用户。可观测性每一次用户请求从入口到出口经过哪些节点、调用过什么工具、模型输入输出什么、每一步花了多少毫秒都要能追踪到。可控性对 Agent 的行为要有强约束比如禁止调用某些工具、禁止在无权限时执行某些写操作、超时后必须走兜底话术。可评估性每一次线上请求都要有日志能沉淀成回归集。改一个提示词或者换一个模型不是拍脑袋上而是先跑评估集再决定。这四个指标后面每一章都在围绕它们展开。2. 架构设计把 Agent 当成一套系统来搭2.1 一套可复用的六层结构我做过几个生产级工作流之后总结出一套通用分层不管后端是 Python 还是 Java逻辑都套得上。第一层是接入层负责接收用户输入、做权限校验、处理渠道对接这一层不接触任何模型逻辑。第二层是任务路由层用分类模型或用轻量规则去做意图识别把请求分到具体的工作流模板上。第三层是上下文管理层它决定哪些历史要保留、哪些要压缩以及当前这轮请求拼装出多大窗口的上下文。第四层是工具调度层负责将 Agent 的决策解析成具体工具调用并处理参数校验、调用、返回结果归一化。第五层是生成层把工具执行结果、上下文、系统提示词一起送给模型生成最终回复。第六层是反馈层记录每次请求的结果质量、用户是否满意、错误类型回流给评估和优化流程。每一层之间用内部接口隔离尽量不越层。我在前面项目里看到最典型的反面案例是为了省事直接在生成层的提示词里塞了一堆工具描述和权限信息结果任务路由的信息、上下文压缩的信息混在一起整个系统改一处崩三处。分层的目的不是增加代码量而是让每个环节都可以独立修改、独立测试、独立回滚。2.2 用状态机管理执行流程别让 Agent 一路自由发挥纯自主式 Agent 的问题在于你没法预判流程也就没法设置超时、熔断和兜底。我在生产里通常会把每一次会话的运行过程定义成一个状态机。还是用客服工单助手举例状态的流转可能是这样的INIT - ROUTING - RETRIEVING - TOOL_CALLING - GENERATING - DONE另外还有 FAILED、TIMEOUT、ESCALATED 三个异常状态。每一步都有超时上限TOOL_CALLING 超过 10 秒就在状态机上标记为 FAILED然后走降级文案而不是让用户一直等。状态机还有一个好处是便于恢复。用户中途断线重连或者服务重启只要把会话状态持久化就能从最近一个非终态恢复执行。这一点对生产环境特别重要因为模型调用和工具调用的耗时都很长动不动几十秒系统不能把这种长会话一直放在内存里。具体实现上不一定非要引入工作流引擎我后面会给出一种轻量的状态机实现核心就是用一个枚举加一张流转表配合超时控制。复杂场景再考虑引入成熟的编排框架但一开始不要过度设计。2.3 工具协议与权限边界工具是 Agent 面对真实世界的抓手也是最容易出问题的地方。我在生产项目中对工具有三条强制要求。第一每一个工具必须有严格的参数 Schema。模型返回的调用参数是自然语言生成出来的极容易多一个字段、少一个字段、类型错误。我这边统一用 JSON Schema 做校验校验失败直接返回给模型重新生成而不是盲抓瞎猜。第二工具必须分权限域。比如普通用户触发的 Agent 只能调用只读工具管理员会话才能调用写操作工具。最简单的方法是在模型可见的工具列表里做过滤——没权限的工具根本不出现在提示词中而不是等模型调用了再拦截。这招能有效降低误调用风险。第三工具返回结果要标准化包装。不能让模型直接看到千奇百怪的原生返回而是统一包一层包含 status、data、error、cost 这些字段。标准化的好处是生成层可以用统一的规则去解析也方便后续做结果审计。我这里特别强调一下模型能看见哪些工具直接影响它的行为边界。工具列表写得越长模型决策负担越重跑偏概率越大。生产上每个路由分支看到的工具集合应该控制在 5 到 8 个以内宁可让用户多绕一步也别把几十个工具全塞进去。2.4 上下文管理的三种策略上下文管理是 Agent 工作流里最容易被忽视、又最能拉开成本差距的一块。我常用的策略有三种可以搭配使用。裁剪式每轮对话结束后对历史消息做评分把明显不重要的信息直接丢弃。比如用户中途闲聊了两句这两句就可以从长期上下文里移除。摘要式当历史消息数量超过阈值用一个小模型把之前的对话压缩成结构化摘要再连同最近这轮完整消息一起作为上下文。我用过最快的一种做法是用摘要模型输出的 500 字关键事实替换掉前几轮完整内容效果损失很小token 节省约 40%。检索式长期记忆不塞进上下文而是落到向量库或者关系表里每次请求时根据当前问题召回最相关的记忆片段只拼最需要的部分。这套思路在客服场景里特别好用用户历史工单、账户状态都可以按需取用。上下文管理还要考虑一个点工具返回结果要不要进上下文。很多时候工具返回一大段 JSON模型其实只需要里面三个字段。所以我通常在工具调度层就做结果精简只把关键字段传给生成层这一步对 token 节省也非常可观。3. 从零实现一个可落地的 Agent 工作流3.1 项目骨架与依赖下面给你一套我在真实项目里用过的简化版本子目标是实现一个“工单助手”Agent用户提问 - 意图路由 - 检索知识库或调用工单查询工具 - 生成回答。技术栈选的是 Python依赖尽量少。我用到的核心组件有FastAPI 做 HTTP 服务、pydantic 做数据校验、asyncio 做异步编排、一个轻量的 JSON Schema 校验函数处理工具参数模型调用抽象成统一的 client 接口。这里没有引什么重型的 Agent 框架原因是生产环境里自定义控制点非常重要框架会替你封装很多东西出问题的时候反而不好定位。我把项目结构大概列一下agent_workflow/ ├── main.py # FastAPI 入口 ├── workflow/ │ ├── state.py # 状态机和状态定义 │ ├── router.py # 意图路由 │ ├── context.py # 上下文管理 │ ├── executor.py # 主执行编排 │ └── tools.py # 工具注册与调用 ├── llm/ │ └── client.py # 统一的模型调用封装 └── config.py # 超时、模型、阈值等配置接下来的代码示例我按关键模块给你讲透。3.2 核心代码状态定义与主执行编排先看状态定义。生产环境里一定要把状态做成显式枚举不要用字符串满天飞。from enum import Enum from typing import Optional, Any, Dict, List from dataclasses import dataclass, field class WorkflowState(str, Enum): INIT INIT ROUTING ROUTING RETRIEVING RETRIEVING TOOL_CALLING TOOL_CALLING GENERATING GENERATING DONE DONE FAILED FAILED TIMEOUT TIMEOUT dataclass class WorkflowContext: session_id: str user_query: str state: WorkflowState WorkflowState.INIT route: Optional[str] None history: List[Dict[str, str]] field(default_factorylist) tool_results: List[Dict[str, Any]] field(default_factorylist) final_answer: Optional[str] None error_message: Optional[str] None retry_count: int 0 metadata: Dict[str, Any] field(default_factorydict)这里的 WorkflowContext 就是整个工作流的“运行时数据总线”每个阶段往里写相应的字段。生产项目中我会额外加一个 trace_id 和 parent_span_id方便对接链路追踪这里先不展开。主执行编排我选择用一个 async run 方法配合超时控制。核心思路是每个状态节点都实现成独立的 async 函数主循环只负责按状态机推进一步超时或失败就跳到异常分支。import asyncio from typing import Callable, Dict, Any async def run_workflow(ctx: WorkflowContext, config: Dict[str, Any]) - WorkflowContext: handlers: Dict[str, Callable[[WorkflowContext], Awaitable[WorkflowContext]]] { WorkflowState.INIT: handle_init, WorkflowState.ROUTING: handle_routing, WorkflowState.RETRIEVING: handle_retrieving, WorkflowState.TOOL_CALLING: handle_tool_calling, WorkflowState.GENERATING: handle_generating, } current_state ctx.state timeout_map { WorkflowState.ROUTING: config[timeout][routing], WorkflowState.TOOL_CALLING: config[timeout][tool_calling], WorkflowState.GENERATING: config[timeout][generating], } while current_state not in (WorkflowState.DONE, WorkflowState.FAILED, WorkflowState.TIMEOUT): handler handlers.get(current_state) if not handler: ctx.state WorkflowState.FAILED ctx.error_message funknown state: {current_state} break try: timeout timeout_map.get(current_state, 5) ctx await asyncio.wait_for(handler(ctx), timeouttimeout) current_state ctx.state except asyncio.TimeoutError: ctx.state WorkflowState.TIMEOUT ctx.error_message ftimeout at {current_state} except Exception as e: ctx.state WorkflowState.FAILED ctx.error_message str(e) # 这里可以记录指标按 session_id 聚合错误率 return ctx这个 while 循环最多转七次就会终态不会出现模型无限自循环的情况。这就是为什么我不喜欢让模型自己决定流程走向的原因——如果把后端循环的次数也交给模型判断超时和失控根本没法管理。3.3 意图路由与工具调用实现意图路由我采用“小模型 规则兜底”的双通道。先让一个轻量分类模型输出用户意图如果分类置信度低于阈值就回退到关键词规则。这样既保证准确率又不需要为了路由单独部署大模型。async def handle_routing(ctx: WorkflowContext, config) - WorkflowContext: # 这里用统一 client 调模型实际生产会用更小的分类模型 messages [ {role: system, content: 你是工单助手的意图分类器只输出 JSON格式为 {\intent\: \retrieve/query_ticket/faq/escalate\}}, {role: user, content: ctx.user_query} ] resp await llm_client.chat(messagesmessages, modelconfig[router_model], response_formatjson) intent_info parse_json_safe(resp.get(content, )) intent intent_info.get(intent, escalate) # 规则兜底如果模型分类置信度低或解析失败走规则逻辑 if 工单 in ctx.user_query and 查 in ctx.user_query: intent query_ticket elif 帮助 in ctx.user_query or 介绍 in ctx.user_query: intent faq ctx.route intent ctx.state WorkflowState.RETRIEVING if intent in (faq, retrieve) else WorkflowState.TOOL_CALLING return ctx工具调用这一块是生产坑最多的地方。我给工具定义统一的协议名称、描述、parameter schema、执行函数、可执行权限域。# tools.py def register_tool(name: str, description: str, schema: dict, permission: str): def decorator(func): TOOL_REGISTRY[name] { name: name, description: description, schema: schema, func: func, permission: permission, } return func return decorator register_tool( namequery_ticket, description根据工单编号查询工单状态, schema{ type: object, properties: {ticket_id: {type: string}}, required: [ticket_id], }, permissionuser, ) async def query_ticket(ticket_id: str): # 实际走工单服务接口 return {ticket_id: ticket_id, status: processing, updated_at: 2025-01-15 10:00:00}每次模型返回需要调用工具时我先做 JSON Schema 校验再检查权限域然后才真正执行。这两个检查要放在模型输出之后、外部调用之前顺序不能反。def validate_and_call(tool_name: str, arguments: dict, user_role: str) - dict: tool_def TOOL_REGISTRY.get(tool_name) if not tool_def: return {status: error, error: unknown_tool} if user_role not in tool_def[permission]: return {status: error, error: permission_denied} # 这里用 jsonschema 做校验生产环境不要自己手写校验逻辑 try: jsonschema.validate(instancearguments, schematool_def[schema]) except Exception as e: return {status: error, error: invalid_parameters, detail: str(e)} result asyncio.run(tool_def[func](**arguments)) return {status: ok, data: result}工具调用失败时我的处理策略是“重试一次 降级一次”不无限重试。很多生产事故就是重试风暴导致的模型第一次调用失败系统让模型重新生成模型又调了同一个接口又失败往复循环最后用户没等到答案系统先被接口限流打挂了。正确做法是最多重试一次第二次失败直接把错误信息拼接成降级提示让用户看到明确说明而不是永远转圈。3.4 生成层与反馈闭环当工具结果回来之后生成层要做的不是原样把结果丢给用户而是做一次信息筛选再让模型组织语言回复。我在生产项目中有一个经验很多人认为模型是自动生成回复的实际情况是手工设计回复格式的效果远好于让模型自由发挥。我通常会给生成层固定一个输出约束要么输出一句话加一个推荐动作要么输出一个结构化卡片。例如“工单查询结果”生成卡片。这样用户一眼能看到信息模型也不容易编造不存在的内容。async def handle_generating(ctx: WorkflowContext, config) - WorkflowContext: tool_summary summarize_tool_results(ctx.tool_results) history_window build_context_window(ctx.history, max_turns6) messages [ {role: system, content: GENERATE_SYSTEM_PROMPT}, *history_window, {role: user, content: f当前用户问题{ctx.user_query}\n工具返回摘要{tool_summary}}, ] resp await llm_client.chat(messagesmessages, modelconfig[generator_model], max_tokens900) ctx.final_answer resp.get(content, 抱歉我暂时没有拿到有效信息。) ctx.state WorkflowState.DONE return ctx反馈闭环决定了你的 Agent 能不能越用越好而不是上线之后一直躺平。生产级要求在每次 Done 之后把这次请求的入参、出参、路由、工具结果、用户后续行为点没点帮到、有没有转人工全部落库。然后定期整理成回归集用这套回归集做评测、微调、提示词优化验证。我见过很多团队把 Agent 上线之后只靠日志肉眼看效果这是典型的作坊做法。正确的反馈闭环至少要有两个东西结构化事件表里面每个请求一行包含所有关键字段人工标注通道产品或运营看到低质量回答可以打标标注结果进入下一轮评测。然后把这两部分数据整理成数据集定时跑一轮回归测试。4. 生产环境下你会踩到的坑4.1 上下文窗口溢出的三种表现上下文溢出不是只有“请求报错超出长度”这一种表现更多情况下是隐性的。第一种是显式溢出直接触发模型的 max_tokens 上限报异常。这个还算好发现日志一查就有。第二种是隐性膨胀上下文没超上限但太长导致响应变慢、成本升高而且窗口塞进去大量无关信息模型反而抓不住重点。第三种是信息稀释历史对话太长中间的关键信息被新的无关内容盖住模型开始遗忘早期用户提到的关键约束。这三个都得靠上下文管理解决我在 2.4 节已经讲了三种策略实操时建议线上先开摘要式因为它的性价比最高、改动最小。我自己的经验值是单个会话超过 8 轮必须触发历史压缩单次工具返回超过 1500 token必须做字段过滤。这两个阈值可以根据自己的模型和场景调但是不能不做限制。4.2 工具调用失败与重试风暴工具调用失败在生产里是家常便饭但很多时候根因不在工具本身而在模型生成参数上。举个例子模型把 ticket_id 生成了一个数字但接口需要字符串参数校验失败模型重新生成又生成了数字再失败形成死循环。这种情况你用重试解决不了要用的方法有两个。第一模型输出工具参数后做一层轻量规范化不光是校验还要做类型转换。数字可以转字符串、字符串可以尝试转时间格式规范不行再返回重新生成这样能提升不少成功率。第二限制模型生成工具参数的最大尝试次数一般设 2 到 3 次超过就放弃工具调用直接走兜底文案。我在前面已经说过重试只允许一次这里再强调一遍重试风暴比工具失败更伤系统。另外要注意工具调用是真实外部请求会真实产生费用和副作用所以每个工具的执行结果必须记录到日志包括请求参数和响应结果。这样一旦出问题你能知道是哪一步真正执行了而不是对着日志猜。4.3 幻觉治理从提示词到结果校验AI Agent 的幻觉和普通聊天机器人的幻觉不太一样它不只体现在回答内容上更危险的是虚构工具执行结果。模型可能会在没有查到工单的情况下直接说“工单已经在处理中”这种回答给用户的误导非常大。我的治理层次是这样的。第一层提示词强约束明确告诉模型“工具返回数据之外的信息一律不能出现”。第二层生成结果的实体校验如果回复里包含工单号、日期、金额这类关键实体系统要拿工具返回结果做包含校验发现不一致就重新生成一次。第三层置信度兜底如果这轮工具调用失败或上下文信息不足就预设一个保守回复模板不要让模型自由发挥。这三层叠加能把明显幻觉压到比较低的水平不过不会彻底消失线下要有质检通道。有一个具体做法很实用在生成层加一个“事实核对”步骤让模型在输出答案的同时输出三个字段answer、evidence、unsupported 标记。evidence 必须引用消息里真实存在的片段unsupported 表示回答中哪些内容没有得到工具结果支持。然后再用规则检查 unsupported 是否为 False。这一招简单有效能帮你在不引入额外大模型的情况下自动拦截一部分幻觉回复。4.4 成本与延迟平衡生产级 Agent 的账单通常由三块组成路由模型成本、生成模型成本、工具调用成本。我最开始做的时候只盯着生成模型后来发现路由模型每天都在被调用即使模型便宜积少成多也很可观。成本优化的优先级我建议这样排先砍 token 消耗再降模型档次最后才调整调用次数。token 消耗的大头在历史消息和工具返回结果这两块我前面已经强调了上下文压缩和结果过滤能省 30% 到 50% 的成本。延迟优化可以考虑做分支并行比如路由和初步检索同时启动等路由结果出来再用能省下一到两秒时间。工具调用如果是只读且耗时不长也可以把多个工具在生成之前并行执行但要特别注意并行的副作用问题写操作绝不能并行。延迟和成本不是一味调低模型就能解决的生产数据表明在关键分支上用一个更强模型比所有分支都用一个中等模型效果更好、总成本更低。因为你可以用强模型减少重试次数重试次数降了工具调用失败率也会跟着降。5. 从上线到持续迭代5.1 监控指标要盯这五个Agent 工作流的监控和普通接口监控不一样不能只看 QPS 和 P99至少要盯五个指标我直接列成一张速查表。指标定义需要盯的原因参考阈值端到端成功率到达 DONE 状态的请求占比最直观的系统健康度大于 95%工具调用失败率工具返回 error 的请求占比判断外部依赖稳定性小于 5%平均上下文长度生成层实际输入 token 数成本与延迟的先行指标按预算折算路由分错率用户后续转人工比例与路由标签的关联判断意图路由质量小于 10%兜底回复率触发降级或兜底文案的请求占比判断系统自动兜底能力小于 5%有了指标之后还要在指标后面挂告警。我个人的习惯是端到端成功率 5 分钟内连续低于 95% 就告警工具失败率单分钟内超过 15% 就立刻暂停该工具的调用权限走降级链路。暂停权限这个动作很容易忽略但它比告警更重要。工具连续失败说明它已经不稳定了继续让 Agent 调只会制造更多错误数据。5.2 回归测试集要怎么建Agent 的评估比传统接口测试复杂因为答案是开放的没法简单比对返回值。我建议分层去建测试集。第一层是确定性断言集输入问题断言输出里必须包含哪些实体。比如用户查工单输出必须包含工单号和状态断言脚本只需要做子串匹配。第二层是工具调用断言集验证给定的问题系统应该调用哪些工具、不应该调用哪些工具以及传参格式是否正确。这一层用日志里的 tool_calls 字段做断言比直接判答案更可靠。第三层是评判模型评估集没有固定答案的开放性问题用另一个更强大的模型按维度打分分数变化超过一定阈值就提醒人工复核。这三层测试集加起来不用很多最开始每个维度 30 到 50 条就够用。我强烈建议把测试集跑在每个改动的前置流程里不管是改提示词、换模型还是调整超时配置先跑一遍回归再决定要不要发。很多问题不是你改坏的而是你根本不知道这次改动会影响哪些分支。5.3 灰度发布与回滚Agent 系统的灰度发布有一点和普通服务很不一样普通服务回滚只要切流量Agent 系统回滚不只切代码还要考虑会话状态。试想一个已经在 ROUTING 状态的会话灰度切走之后它的上下文还在旧节点的内存里新节点根本感知不到用户就会莫名其妙的断线。我的建议是把工作流节点做成无状态设计。WorkflowContext 这个数据对象要支持序列化每次状态切换都写一份到存储里节点只负责从存储加载上下文再继续执行。这样灰度发布时即使同一会话落到不同实例上也能从存储里恢复上下文不会断线。灰度策略我一般按用户维度切先让 5% 的低风险流量跑到新版本观察半天指标再逐步放量到 20%、50%、100%。每个阶段只允许升不允许降除非触发前面讲的告警阈值。回滚时直接切回上一个稳定版因为上下文都在存储里不会出现状态断层。这一点在实际验证中深有体会——能否快速回滚的差距在事故面前就是几分钟和几小时的差距。我做完这套工作流之后最大的感悟是不要把 Agent 当成一个“模型问题”要把它当成一个“系统问题”。模型只是其中一个执行器真正决定系统能不能稳的是编排、状态、权限、观测这些工程细节。希望这篇能帮你在生产化这条路上少踩几个坑。后面有时间我可能再写一篇关于 Agent 评估集怎么设计更细的实践欢迎持续交流。