生产级AI Agent工作流：从状态机到可观测性的工程实践

我把一个内部用的 AI Agent 从 demo 状态推到生产环境，前后折腾了一个多月。这个过程里最深的体会是：构建生产级 AI Agent 工作流，真正的难点往往不在模型本身，而在模型之外的工程细节——状态怎么流转、上下文怎么管理、失败怎么重试、线上怎么观测。如果你也在搭类似的东西，这篇内容应该能帮你少走不少弯路。

我接下来会按自己的实际拆解路径来写：先讲清楚“生产级”到底比“能用”多了什么，再讲工作流的骨架设计、上下文与重试策略、可观测性，最后用一个客服工单助手的例子把整条链路串起来。涉及的代码会是简化过的伪代码，方便你迁移到自己的技术栈。

1. 从“能跑”到“敢上线”：生产级 AI Agent 工作流到底难在哪

1.1 一个“能用”的 Agent 和一个“敢上线”的 Agent，差的不只是准确率

很多人对 Agent 的认知是从聊天框开始的：给它一个任务，它能拆解、能调用工具、能返回一段像模像样的回答。Demo 阶段确实可以这样玩，因为输入的测试样本就那么几十条，任务范围窄，失败了重来一次就好了。但生产环境完全是另一回事。

生产环境意味着三件 Demo 阶段不需要面对的事：第一，输入分布是开放的，用户不会按你预想的句式提问；第二，失败是有成本的，可能是用户投诉，可能是资金损失，也可能是下游系统被错误的数据污染；第三，没人会守在旁边盯着日志，系统必须自己处理异常并给出可审计的记录。

我之前带过一个模拟项目，在一个客服场景里把“意图识别 + 知识库检索 + 回复生成”的 Agent 工作流从原型推到线上。原型的准确率看着有 90% 以上，但上了灰度之后，真实工单上的有效处理率只有 70% 左右。差的这 20 个点，几乎全部来自边界情况：工单信息不全、知识库没有对应条目、用户一句话里包含多个意图。模型还是那个模型，prompt 也差不多，问题全出在工作流没有为这些情况设计兜底路径。

1.2 我给“生产级”定的五个必考项

如果你问我什么才算生产级，我自己的判断标准是下面五条。每一条都对应一类具体的工程动作，而不是一句“要稳定”的空话。

这五条里面，我见过最多团队栽跟头的是第一条和第三条。模型调用不稳定大家都有心理准备，但“重复执行”和“token 失控”这种事，往往要到账单出来或者下游数据出现重复记录时才被发现。后续我会在每个环节展开具体怎么做，这里先记住一个原则：生产级 Agent 工作流的核心是“系统思维”，模型只是系统里一个可被替换的执行单元。

1.3 先接受一个现实：Agent 不可能永远自主

还有一个心态上的坎要过：不是所有环节都应该让模型自主决定。很多人推崇“给 Agent 一个目标，让它自己规划路径”，这种想法在封闭环境里很性感，但放到生产环境就变成了灾难——你不知道它下一步会调用哪个工具，也不知道它会不会在一个失败分支里反复打转。

我的做法是：能由代码决定的流程控制，绝不让模型决定；模型只负责它真正擅长的事，比如意图判断、内容生成、信息抽取。工作流的骨架是确定的，模型是骨架上的一个节点。这个思路贯穿整篇文章，也是我把一个 Agent 从“会聊天”变成“能干活”的关键转折。

2. 先定骨架再写代码：用状态机把 Agent 的执行路径锁死

2.1 为什么我放弃了让模型“自由发挥”

最开始我也追过 ReAct 那套思路：让模型自己思考、自己决定调什么工具、自己观察结果、再继续思考。Demo 里效果确实惊艳，它能自己绕过一个又一个坑。但上线以后我很快发现，自由发挥的代价是代价不可控。

典型的问题有三个：循环出不来回、工具参数编造、以及极端情况下的 token 浪费。有一次测试，模型在一个查询失败的分支里连续重试了 7 次相同的工具调用，每次都先输出一大段“分析”，然后再调一次，最后返回给用户一句“暂时无法处理”。那次单个任务消耗的 token 是正常路径的 12 倍。我后来查日志才发现，模型并不知道这个工具已经稳定失败，它只是按照“再试一次可能就好了”的惯性往下走。

所以我的结论是：自由发挥适合探索，不适合执行。生产级工作流需要把“下一步做什么”的控制权从模型手里拿回来。模型负责判断“这个用户意图是什么”，代码负责决定“意图识别之后应该进入哪个节点”。

2.2 一个最小的状态机工作流长什么样

我习惯用状态机来表示 Agent 的执行路径，因为在状态机里，所有可能的状态转移都是提前枚举好的，不可能出现代码之外的意外路径。

一个典型的客服工单处理流程，我会拆成这样几个状态：

WORKFLOW_STATES = { "START": "工单进入", "INTENT_RECOGNITION": "意图识别", "INFO_COLLECTION": "信息补充收集", "KB_SEARCH": "知识库检索", "DRAFT_GENERATION": "回复草稿生成", "RISK_CHECK": "风险规则校验", "HUMAN_REVIEW": "人工审阅", "AUTO_SEND": "自动发送", "FINISHED": "完成", "FAILED": "失败", }

状态之间的转移条件提前写好：

TRANSITIONS = { "START": {"next": "INTENT_RECOGNITION"}, "INTENT_RECOGNITION": { "success": "KB_SEARCH", "need_more_info": "INFO_COLLECTION", "low_confidence": "HUMAN_REVIEW", }, "KB_SEARCH": { "found": "DRAFT_GENERATION", "not_found": "HUMAN_REVIEW", }, "DRAFT_GENERATION": { "success": "RISK_CHECK", "failure": "HUMAN_REVIEW", }, "RISK_CHECK": { "pass": "AUTO_SEND", "block": "HUMAN_REVIEW", }, "AUTO_SEND": {"success": "FINISHED", "failure": "HUMAN_REVIEW"}, "INFO_COLLECTION": {"success": "KB_SEARCH", "timeout": "HUMAN_REVIEW"}, "HUMAN_REVIEW": {"success": "FINISHED", "failure": "FAILED"}, }

别看这个结构简单，它解决了一个大问题：任何时刻你都能明确回答“这个 Agent 现在处于什么状态、下一步可能去哪”。排查问题的时候，只需要看状态字段，就能知道整个工作流走到哪一步了，不需要重新推演模型的思维链。

每个状态节点我建议都实现成独立的函数，输入是一个统一的上下文对象，输出是状态转移指令。这样后续想替换任何环节都很方便——比如把意图识别从模型 A 换成模型 B，只需要改这一个节点的实现。

2.3 工具层协议：让模型学会“用遥控器，而不是拆遥控器”

工作流里最容易被忽略的是工具调用的设计。很多人直接把函数写成工具暴露给模型，函数返回的是自然语言文本，导致模型需要自己“理解”一段话来决定下一步。这在流程简单时没问题，但一旦工具变多，问题就来了：模型可能会曲解返回文本，也可能把上一个工具的文本当成用户输入。

我的做法是给每个工具定义严格的输入输出协议。输入参数用 JSON Schema 约束，输出统一包成结构化对象，至少包含这几个字段：

class ToolResult: status: str # "success" / "error" / "empty" data: dict # 结构化数据或检索结果 error_code: str # 机器可读错误码，如 "TIMEOUT" error_message: str # 人类可读的错误描述 request_id: str # 关联到本次调用

模型的 prompt 里不写“你可以调用知识库接口”，而是写成“你可以使用 kb_search 工具，参数必须符合 schema”。工具返回的错误里，也尽量不要出现大段自然语言，而是用 error_code 告诉工作流引擎该走哪条分支。

这个设计看起来是多了一层包装，但换来了三个直接收益：第一，模型产生幻觉参数的几率下降，因为 schema 校验会在调用前拦截；第二，工作流引擎可以根据 error_code 决定重试还是降级，而不是把错误文本丢给模型再次判断；第三，所有调用都可以按 request_id 串联起来，方便后面的可观测性分析。

3. 生产环境真正让人失眠的三个问题：上下文、重试与并发

3.1 上下文窗口不是不够用，是没规划

大模型的上下文窗口正在变得越来越长，但“更长”不等于“够用”。如果你只是把历史对话、工具结果、业务数据一股脑拼进 prompt，很快就会发现两件事：一是单次请求的 token 成本快速上升，二是模型会被无关信息干扰，反而更容易出错。

我通常会按角色把上下文拆成四个区域：

对于历史对话，我很少把原始消息全量塞进去。更稳妥的做法是“滚动摘要 + 关键事实抽取”：每次处理完成后，用一次独立的模型调用，把当前上下文压缩成两三百字的摘要，并把关键事实提取成结构化字段。下一次请求只带上摘要和事实字段，不再带原始消息。

举个例子，一个客服工单如果来回沟通了 10 轮，原始文本可能有 5000 字。压缩成摘要后，可能只需要 400 字。代价是每轮多一次“摘要调用”，但这比每次请求都拖着 5000 字历史要便宜得多，模型的回复质量也更高。你需要根据任务类型去平衡，但原则是固定的：永远不要让上下文无限膨胀。

3.2 重试与超时：一次临时故障不应该拖垮整个工作流

模型接口不稳定是常态。网络抖动、限流、偶发 5xx，这些在生产环境里几乎每天都会遇到。如果没有重试机制，任何一次临时故障都会让一个工单直接失败；但如果不加区分地盲目重试，又可能出现重复扣费和重复写下游系统的问题。

我的经验是先把错误分成两类。可重试错误包括超时、限流、服务端 5xx 这类大概率是暂时性的错误，策略是指数退避重试，比如第一次等 1 秒，第二次等 2 秒，第三次等 4 秒，最多重试 3 次。不可重试错误包括参数校验失败、输入内容违规、业务规则拦截这类错误，它们重试一万次结果也一样，应该直接进入失败分支或人工处理通道。

更关键的是幂等控制。工作流里的每个任务都带一个 task_id，每次模型调用和工具调用也都带一个 call_id。重试的时候，工作流引擎携带相同的 call_id 去调用工具。下游系统只要实现“同一个 call_id 只处理一次”，就能防止重复发送短信、重复创建工单这类事故。这个设计不能等到出事后再补，必须在最开始就加进去。

def call_model_with_retry(task_id, call_id, payload, max_retries=3): for attempt in range(max_retries): try: return model_api(task_id=task_id, call_id=call_id, payload=payload) except RetryableError as e: wait_time = 2 ** attempt time.sleep(wait_time) raise WorkflowError("model call failed after retries")

这段伪代码的逻辑很简单，但实际接入时有个容易被忽略的细节：重试之间最好保存一份中间状态。比如在“生成回复草稿”这个节点，如果第一次调用已经生成了半段内容，重试时就不应该让模型重新从头写。一个可行的方法是在请求里带上“这是第几次重试生成”，或者把已生成的草稿作为上下文的一部分传给模型，让它基于已有内容完成补充。

3.3 并发与限流：当你同时跑 100 个 Agent 时

单人做 Demo 时，一次只处理一个任务，不会觉得并发是个问题。但生产环境里，消息队列里可能同时躺着几百个工单，你需要的是可控的并发，而不是一股脑全部发出请求。

我有一个比较朴素的并发控制方案：工作流引擎维护一个信号量，限制同时运行的 Agent 实例数量。比如最多同时运行 20 个实例，剩余的进入等待队列。每个实例内部，对模型 API 的调用还要再套一层限流器，避免瞬间打爆接口配额。

import asyncio MAX_CONCURRENCY = 20 semaphore = asyncio.Semaphore(MAX_CONCURRENCY) async def run_agent(task): async with semaphore: await execute_workflow(task)

除了并发数量，我还会给不同类型的任务设置优先级。比如“用户等待回复”的工单优先级高于“后台数据补全”的任务。实现方式很简单：消息队列里按优先级分组，高优先级队列的消费者线程数更多。等系统跑稳之后，还可以按模型响应延迟动态调整并发数，但一开始没必要做那么复杂，先固定一个安全的并发上限就够了。

另外一个容易忽略的点是成本预估。每个 Agent 实例在启动时，我会先估算它的 token 消耗预算。比如设置单任务预算为 20 万 token，当实际消耗超过预算的 80% 时，工作流自动切到降级模式：不再调用复杂模型，而是直接转人工。这个机制看起来粗暴，但在线上救过我很多次，尤其是遇到模型“死循环”式的调用时，预算上限就是止损线。

4. 可观测性：你凭什么知道 Agent 在“想”什么

4.1 把思维链变成结构化的执行记录

模型的黑盒属性没办法消除，但我们可以把它的每一步决策都转化成可检索的记录。我引入了一套事件日志协议：工作流中的每一次状态转移、模型调用、工具调用，都输出一条结构化事件。

事件至少包含这些字段：

EVENT = { "timestamp": "2025-01-01T12:00:00Z", "task_id": "task_12345", "workflow": "customer_service", "node": "INTENT_RECOGNITION", "action": "model_call", "model_input": {"text": "工单内容...", "token_count": 1200}, "model_output": {"intent": "refund", "confidence": 0.93}, "latency_ms": 850, "cost_usd": 0.002, "status": "success", }

这些事件可以写入本地日志文件，也可以推送到日志系统。关键是要保证两点：第一，事件格式要稳定，字段不能随意变动，否则后续分析会非常痛苦；第二，事件之间要有可串联的 ID，也就是 task_id 和 call_id，这样你可以按任务把所有事件拉出来，完整还原一次处理过程。

有了这套日志之后，排查问题的方式就变了。以前是“用户反馈回答不对，你猜模型为什么错”，现在是“拉出该任务的事件链，看一下意图识别结果是什么、知识库检索到了什么、生成草稿时上下文里包含哪些内容”。问题往往一眼就能定位。

4.2 评估一个 Agent 工作流，不能只盯着单次准确率

上线之后，我会同时盯三类指标：效果指标、成本指标、稳定性指标。

效果指标里，最核心的是“一次解决率”——多少任务在不需要人工介入的情况下被完整处理完。这个指标比单轮准确率更接近业务价值，因为一个 Agent 即使单次回答准确，如果它让用户来回确认三次，实际体验也是差的。

成本指标里，我主要看两个数：单任务平均成本和 token 消耗趋势。单任务成本如果出现突然上涨，通常意味着某个节点开始产生多余调用，比如知识库检索反复失败导致重试。token 消耗趋势反映的是上下文管理是否失控，如果每天平均 token 持续上涨，说明某个分支的上下文压缩没有生效。

稳定性指标里，除了平均延迟，我还会统计“无效回复率”——模型生成了内容，但因为内容为空、被规则拦截或格式非法，最终没有发出去。这个指标很有欺骗性，因为工作流表面上是成功的，实际上用户体验是断的。

评估不能全靠线上指标，还需要一套离线评测集。我会提前整理几百条真实脱敏工单，给每条打上“期望处理路径”和“期望回复要点”的标签。每次改 prompt 或调整工作流结构，都在这个评测集上跑一遍，对比前后效果。有了这个集子，改动时就不会“凭感觉上线”。

5. 端到端示例：一个客服工单分类与回复助手

5.1 场景设定与初始指标

我用一个简化但完整的例子来说明：某平台要做一个客服工单助手。工单内容包括用户问题和一些后台字段，工作流需要判断意图、检索知识库、生成回复草稿，再决定是自动发送还是转人工。

我给这个工作流定下的目标很简单：一次解决率不低于 75%，转人工率不超过 30%，单任务平均成本不超过 0.05 美元。这三个指标放在一起，能有效防止“只管准确率不管成本”或者“为了降低成本疯狂转人工”这两种偏科现象。

5.2 工作流定义和关键代码

工作流用 YAML 定义节点和状态转移，具体执行交给调度引擎。定义文件大致长这样：

workflow: customer_service triggers: - event: new_ticket nodes: - id: intent_recognition type: model_call model: default_llm system_prompt: prompts/intent_recognition_v3.txt output_schema: schemas/intent.json transitions: success: kb_search low_confidence: human_review - id: kb_search type: tool_call tool: kb_search timeout_sec: 3 retry: 1 transitions: found: draft_generation not_found: human_review - id: draft_generation type: model_call model: default_llm system_prompt: prompts/draft_generation_v2.txt transitions: success: risk_check failure: human_review - id: risk_check type: rule rules: - contains_sensitive_word - amount_exceeds_limit transitions: pass: auto_send block: human_review - id: auto_send type: tool_call tool: send_reply transitions: success: finished failure: human_review

对应的工作流节点函数，简化后是下面这种形式：

def run_node(node_id, ctx): if node_id == "intent_recognition": result = call_model(ctx.prompt, ctx.task_text) if result.confidence < 0.6: return {"transition": "low_confidence", "data": result} return {"transition": "success", "data": result} if node_id == "kb_search": docs = kb_search(ctx.intent, ctx.task_text) if not docs: return {"transition": "not_found", "data": None} return {"transition": "found", "data": docs} if node_id == "risk_check": if contains_sensitive_word(ctx.draft) or ctx.amount > 10000: return {"transition": "block", "data": None} return {"transition": "pass", "data": None}

你可能会觉得这个代码太简单了。但我的经验是，生产级工作流恰恰需要这种简单。节点函数只做一件事，输入输出都是明确的数据结构，不藏任何隐式逻辑。真正复杂的地方在边缘处理：比如 model_call 失败时走哪个分支、文档检索时间超过 3 秒怎么降级、生成内容为空时是重试还是转人工，这些都必须提前定义好。

5.3 上线后暴露出的三个问题

这个工作流上线第一周，我从日志里发现了三个典型问题，你也可以把它们当成设计时的检查清单。

第一个问题是“意图识别低置信度被过早转人工”。我原本把置信度阈值设为 0.6，结果有一类“催退款”的工单因为表达比较委婉，经常低于这个阈值。直接结果是转人工率超过 40%，脱离了目标。后来我不再看整体阈值，而是按意图类型分别设阈值：退款类意图阈值降到 0.45，投诉类意图保持 0.6。转人工率立刻降下来了。

第二个问题是“知识库检索超时被当成 not_found”。我最初的实现里，检索超时和检索无结果都返回 not_found，导致一批本可以自动回复的工单被白白转人工。后来我把状态拆成了 found、not_found、error 三种，error 走重试，重试两次仍然失败才转人工。这么一改，自动处理率提升了大约 8 个点。

第三个问题是“回复草稿里出现模型幻觉信息”。比如知识库里没有明确退款时间，模型却生成了“将在 3 个工作日内退款”。这个问题的根因是生成草稿时的系统提示里没有强约束“未检索到的信息不允许补充”。我在 prompt 里加了一条硬性规则，并在风险校验节点加了关键词扫描：草稿中出现知识库中不存在的期限数字，直接转人工。准确率和安全性同时得到了改善。

这个例子想说明的是：工作流不是搭完就结束了，它是一个需要根据线上日志不断调整的系统。你会反复在“自动处理率高但风险也高”和“转人工多但安全”之间找平衡。日志体系越完善，这个调整过程就越有依据。

6. 容易被忽略的工程细节：权限、版本与灰度

6.1 权限边界：Agent 只应该有它该有的钥匙

Agent 能调用的工具越多，风险面就越大。我见过一个很危险的实践：为了开发方便，让 Agent 直接调用一个封装了完整数据库访问权限的函数。虽然工具名写的是“查询用户信息”，但底层的连接串拥有整个库的读写权限。一旦模型在参数上产生幻觉，或者 prompt 注入发生，后果很难收拾。

我的原则是：每个 Agent 对应一个最小权限账号。比如客服助手只能通过一个只读接口访问客户信息的脱敏视图，不能直接连数据库；发送消息的工具要经过风险校验节点，并且有频控限制。你可以在工具层做一层身份隔离，让每个工作流使用不同的服务账号，这样即便某一个 Agent 被攻破，影响范围也是可控的。

6.2 prompt 版本化与灰度发布：改 prompt 等于改代码

很多人把 prompt 当成“写几句话的事”，改完直接上线。这是生产级工作流的大忌。Prompt 对输出行为的影响非常敏感，可能一句措辞的调整就会让模型在边界情况下的表现完全不同。我把 prompt 当成代码管理：每个版本的 prompt 有独立文件名和版本号，工作流配置里明确引用哪个版本，日志里记录的模型调用也带上 prompt 版本字段。

上线流程上，我会先做小流量灰度。比如新版本 prompt 先承担 10% 的流量，对比旧版本在转人工率、无效回复率、单任务成本上的差异，确认没有明显劣化后，再逐步放大到 50%、100%。一旦发现问题，可以快速切回旧版本，因为旧版本的文件还在，配置回滚只需要改一个版本号。

还有一个细节：当工作流结构发生变化时，比如新增了一个节点，我会同步保留旧的工作流定义一段时间。线上任务有可能在切换过程中处于中间状态，如果新流程不能兼容旧状态，就需要一个迁移脚本。这个听起来很基础，但很多人上线新版本时只想着“旧的直接推倒重来”，结果卡在了半路状态上。

6.3 最后说几句个人体会

搭完这套东西之后，我最大的感受是：生产级 AI Agent 工作流，本质上是把“模型的不确定性”关进一个由工程规则围起来的笼子里。模型负责它擅长的语言理解和生成，工程系统负责流程、重试、权限、成本和可观测性。这两者不是谁替代谁，而是各管一段。

如果你正准备从 Demo 阶段往生产环境推，我的建议是先控制范围。不要一开始就做“万能助理”，先选一个业务边界清晰、失败成本可控的场景，比如客服工单、内容审核辅助、报表生成。等你在一个场景里把状态机、重试、日志、评估这套基建跑通，再横向复制到其他场景，速度会快很多。

还有一个经验是：工具不是越多越好。每接入一个工具，就意味着多一类参数校验、多一种错误处理、多一个权限管控点。初期宁可让 Agent 少干活，也要保证它干的每件事都是可控的。先把骨架做扎实，再谈智能，这句话我建议你记下来。