AI Work 并不是一个新模型也不是某家大厂发布的某个 App而是把大模型能力编排进业务流程后形成的一套可运行、可维护、可观测的工作流体系。最近关于“国产 AI Work”的讨论明显增多也有人用“悟空四个月‘消失’”来形容某类 AI 产品在热度消退后没有持续迭代。单看这个现象会觉得是运营问题或模型不够好但从工程视角观察更接近真相的原因是产品没有把 AI 能力沉淀成可稳定流转的业务流程也就是缺少 AI Work。模型能力决定 AI 的下限AI Work 决定 AI 的上限。单次对话能写出不错的文案这靠模型让上千条待办工单自动完成分类、摘要、分派、回访、结案并且每一步都可追踪、可回滚、可审计这必须依赖 AI Work。下面先厘清 AI Work 与传统自动化的边界再分析大厂重视它的工程原因然后用一个最小可运行示例演示如何实现最后补充参数设计、排错路径和生产落地清单。1. 先厘清边界AI Work 不是聊天机器人也不是作业脚本1.1 通俗理解AI Work 是给大模型装上一套业务流程普通聊天机器人的逻辑是“一问一答”用户输入文本模型返回文本流程结束。AI Work 更像一条组装好的流水线。它会先判断当前任务是什么决定调用哪套提示词再调用模型生成中间结果然后通过规则校验结果是否合格不合格的进入重试或人工处理最后把结果写入数据库、消息队列或报表系统。整个过程不是一次模型调用而是多次模型调用和规则判断的组合。举个例子客服工单处理如果只靠聊天机器人用户在对话框里问“怎么退货”模型能回答一段退货政策。但 AI Work 会把这个动作拆成读取工单文本调用模型提取订单号、商品、退货原因再调用规则判断是否满足退货条件不满足则生成驳回文案满足则创建退货单并通知仓库。每一步都有输入、输出、校验和异常分支。从使用场景看AI Work 适合输入量大、规则复杂、结果需要进入其他系统的任务比如工单分类、文档审核、舆情摘要、数据清洗、代码提交描述生成等。它不追求一次对话惊艳而追求一批任务稳定完成。1.2 从单模型调用到 AI Work为什么流程比单个提示词重要单次模型调用解决的是“生成”AI Work 解决的是“稳定地生成并交付”。在实际业务中一次生成很少能直接成为最终结果。生产环境需要面对格式错误、字段缺失、置信度不足、调用超时、限流、重复提交等问题。这些问题无法靠更换一个更大模型解决必须通过流程设计来兜底。单模型调用可以写成response llm_client.chat(prompt) text response.content这种写法适合跑通原型。一旦进入生产就需要把上面这行代码扩展成包含输入校验、调用参数、超时控制、输出解析、结果校验、失败重试、日志记录的完整 AI Work。工作流的价值不是让模型变聪明而是让不稳定的模型输出在业务边界内稳定落地。大厂重视 AI Work 的核心原因也在这里模型能力可以通过采购或自研获得但把模型接进业务系统的工程能力只能靠团队自己沉淀。同一个大模型接口有人只能做出聊天 Demo有人能做出自动化运营系统差别就在工作流设计。1.3 AI Work 与 RPA、传统工作流的区别判断一个系统是否属于 AI Work可以从三个特征来看第一流程中存在大模型生成或决策节点第二生成结果会被后续步骤校验、转换和消费第三执行链路能够被跟踪和恢复。传统工作流和 RPA 也强调流程但决策来自预定义规则几乎不依赖模型泛化能力。AI Work 引入了模型节点之后流程的复杂度从“分支多”变成“不确定性高”因此需要额外的工程保护。维度传统工作流RPAAI Work处理对象结构化数据和固定接口界面操作和模拟点击文本、语义、决策和外部工具混合规则来源预定义业务规则预定义操作步骤模型生成 人工规则校验异常处理条件分支固定重试重试、校验、兜底、人工介入最怕的问题流程节点变化界面元素变化模型输出不稳定典型落地审批流、订单流报表自动化、系统切换工单、审核、摘要、辅助决策从表格可以看到AI Work 并不是替代 RPA 或传统工作流而是在它们之上增加了“模型决策节点”。这带来的好处是能处理非结构化输入坏处是必须为模型的不确定性设计额外的保护机制。2. 大厂为什么把目光移到 AI Work四个工程原因2.1 模型能力被拉平工作流成为落地差异点过去几年产品竞争重点在大模型本身谁的参数多、谁的榜单高谁就有优势。但模型能力发展到一定阶段后开源模型和商业模型之间的差距逐渐缩小单纯提供“模型调用”已经很难形成产品壁垒。此时真正决定用户体验的是模型输出之后产品如何解析、校验、过滤、排序、回退和转人工。这些逻辑都长在 AI Work 里。同样一个摘要模型A 产品直接把模型输出展示给用户B 产品会先判断摘要是否完整、是否包含敏感词、是否覆盖关键指标异常时自动重试或降级。用户感知到的差距不是模型聪明程度而是流程的成熟度。所以大厂强调 AI Work本质是把竞争从“谁的模型更强”转向“谁能把模型用得更好”。这更接近软件工程问题也更适合团队长期积累。2.2 AI Work 可以把不可控输出变成可控流程大模型输出具有概率性同一个提示词在不同时间可能返回不同内容。如果业务直接把模型结果写入数据库大概率会出现字段缺失、格式错误、逻辑冲突等问题。AI Work 的处理方式是给模型输出加约束。常见的做法包括提示词中明确要求返回 JSON并给出字段示例使用结构化输出或函数调用能力让模型按 Schema 返回模型返回后用代码校验必填字段、类型、范围、枚举值校验失败时重新调用模型进行“修复式追问”而不是直接抛错连续多次失败后转入人工处理通道。这些约束保证进入下游系统的数据完整、可信。AI Work 的本质就是把不可控的模型生成行为用工程手段约束到业务可接受的范围内。2.3 成本与延迟路由、缓存、并发控制大模型调用不是免费的成本由 token 数量决定延迟由模型大小和排队情况决定。如果不做工作流所有请求都发给最大模型成本会快速失控。AI Work 可以在流程中设计分级路由route_rules: - condition: intent greeting model: small-fast-model max_tokens: 64 - condition: intent complex_query model: large-model max_tokens: 1024 - default_model: medium-model简单问题走小模型复杂问题走大模型重复问题走缓存。语义缓存可以把相同问题的历史回答直接返回省去一次模型调用。再加上并发控制在合理范围内让系统不会因为瞬时流量被限流整体成本更可控。延迟方面AI Work 可以做到部分步骤并行。比如文档审核工作流可以先并行抽取标题、作者、风险词最后再汇总。单步模型调用延迟没有变但整体任务耗时明显下降。2.4 合规和审计每步可追踪、可回滚业务系统一旦接入 AI就会出现一个很难回答的问题这个结果是谁生成的依据是什么如果出错如何定位AI Work 通过流程编排天然记录每一步的输入、输出、模型、参数和异常信息。在审计场景下这非常关键。比如银行对公开户材料审核模型判断“资料齐全”系统必须能追溯是哪次调用、哪个模型、哪些字段、置信度多少。AI Work 把每一步执行结果和上下文串起来相当于给 AI 决策加了一条完整时间线。可回滚也很重要。当新流程版本出现明显错误时可以保留旧版本的执行逻辑通过配置中心快速切换而不是重新发版。这也是大厂在设计 AI Work 平台时会把版本管理放在第一优先级的原因。3. 从零搭建一个最小 AI Work 示例3.1 最小示例解决什么问题先直接看代码再解释设计。下面这个 Python 示例会完成一条客服工单的自动处理输入工单文本AI Work 自动完成预处理、模型调用、结果校验、落库展示。模型调用超时会自动重试结果置信度过低会让任务失败所有状态保存在上下文对象中方便后续替换成数据库。示例目标是体现 AI Work 的四个关键能力用状态机表达任务生命周期用上下文对象在步骤之间传递数据用异常和重试处理模型不稳定用校验步骤拦截不合格结果。3.2 目录与运行环境运行环境只有两个要求Python 3.10 以上无第三方依赖。目录结构保持最小ai-work-demo/ ├── workflow_demo.py └── README.md把全部代码放在workflow_demo.py便于在一篇文章里讲清楚。实际项目中会把步骤拆到不同模块但核心思想不变。3.3 用状态机和上下文表达工作流状态机解决“任务处于什么阶段”的问题。这里定义任务级别的状态和步骤级别的状态from enum import Enum class WorkflowStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed class StepStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed RETRYING retrying上下文对象解决“步骤之间如何传数据”的问题。不要把中间结果存在全局变量里应该统一放在上下文对象中这样每个步骤都能读取和写入排错时也容易打印from dataclasses import dataclass, field from typing import Any dataclass class WorkflowContext: input_data: dict records: dict field(default_factorydict) workflow_status: WorkflowStatus WorkflowStatus.PENDING current_step: str step_status: StepStatus StepStatus.PENDING error: str records字典会保存每个步骤的关键输出。没有它后续步骤只能重复调用前面的逻辑工作流就退化成顺序脚本。3.4 核心代码实现先写一个模拟大模型调用的函数。它本身带有重试、退避和模拟异常逻辑import json import random import time RETRYABLE_ERRORS (TimeoutError, ConnectionError) def llm_call_with_retry(prompt: str, max_retries: int 3, base_delay: float 0.5) - str: for attempt in range(max_retries): try: print(f[llm] 第 {attempt 1} 次调用prompt 长度 {len(prompt)}) # 模拟模型接口偶尔超时 if random.random() 0.15: raise TimeoutError(模拟模型接口超时) # 正常情况返回一段 JSON 字符串 return json.dumps({ summary: 用户申请退货需要确认商品是否已寄回, category: 售后, confidence: 0.96 }, ensure_asciiFalse) except RETRYABLE_ERRORS as e: if attempt max_retries - 1: raise sleep_time base_delay * (2 ** attempt) random.uniform(0, 0.3) print(f[llm] 调用异常{e}{sleep_time:.2f}s 后重试) time.sleep(sleep_time)接下来定义四个步骤。每个步骤都接收同一个WorkflowContext对象并往里写入结果def step_preprocess(ctx: WorkflowContext) - None: raw ctx.input_data.get(ticket, ).strip() if not raw: raise ValueError(工单内容不能为空) ctx.records[normalized_ticket] raw def step_llm_summary(ctx: WorkflowContext) - None: prompt ( 请对下面的客服工单做摘要和分类只返回 JSON 字段包括 summary、category、confidence。\n f工单内容{ctx.records[normalized_ticket]} ) response llm_call_with_retry(prompt) try: data json.loads(response) except json.JSONDecodeError as e: raise ValueError(f模型输出不是合法 JSON{response}) from e if not isinstance(data, dict): raise ValueError(模型输出不是 JSON 对象) if summary not in data or category not in data or confidence not in data: raise ValueError(模型输出缺少 summary/category/confidence 字段) ctx.records[llm_result] data def step_validate(ctx: WorkflowContext) - None: data ctx.records.get(llm_result, {}) if float(data.get(confidence, 0)) 0.9: raise RuntimeError(置信度过低需要人工复核) ctx.records[validated] True def step_save(ctx: WorkflowContext) - None: # 实际项目里写数据库或消息队列 ctx.records[saved_id] TICKET-0001工作流引擎只需要一个简单的循环顺序执行步骤任何异常都会把整个任务标记为失败并记录当前步骤和错误信息class LLMWorkflow: def __init__(self, steps): self.steps steps def run(self, ctx: WorkflowContext) - WorkflowContext: ctx.workflow_status WorkflowStatus.RUNNING for step in self.steps: ctx.current_step step.__name__ ctx.step_status StepStatus.RUNNING print(f[workflow] 开始步骤{ctx.current_step}) try: step(ctx) ctx.step_status StepStatus.SUCCESS print(f[workflow] 步骤完成{ctx.current_step}) except Exception as e: ctx.step_status StepStatus.FAILED ctx.error f{ctx.current_step}: {e} ctx.workflow_status WorkflowStatus.FAILED return ctx ctx.workflow_status WorkflowStatus.SUCCESS return ctx最后是入口。输入一条工单创建上下文执行工作流输出结果if __name__ __main__: workflow LLMWorkflow([ step_preprocess, step_llm_summary, step_validate, step_save, ]) ctx WorkflowContext(input_data{ ticket: 我上周买的杯子碎了想申请退货。 }) workflow.run(ctx) print(json.dumps({ workflow_status: ctx.workflow_status.value, current_step: ctx.current_step, records: ctx.records, error: ctx.error }, ensure_asciiFalse, indent2))3.5 运行验证和预期输出在项目目录下执行python workflow_demo.py正常情况会看到类似输出[workflow] 开始步骤step_preprocess [workflow] 步骤完成step_preprocess [workflow] 开始步骤step_llm_summary [llm] 第 1 次调用prompt 长度 52 [workflow] 步骤完成step_llm_summary [workflow] 开始步骤step_validate [workflow] 步骤完成step_validate [workflow] 开始步骤step_save [workflow] 步骤完成step_save { workflow_status: success, current_step: step_save, records: { normalized_ticket: 我上周买的杯子碎了想申请退货。, llm_result: { summary: 用户申请退货需要确认商品是否已寄回, category: 售后, confidence: 0.96 }, validated: true, saved_id: TICKET-0001 }, error: }如果模型调用连续失败或结果校验不通过输出中workflow_status会变成failederror会记录具体哪一步出了什么问题。这就验证了 AI Work 的异常处理链路。注意不要只验证程序能启动还要故意把confidence改小、把模型返回改成非 JSON观察工作流是否正确进入失败分支。只有失败分支可控才能算真正的 AI Work。4. 关键设计超时、重试、参数和并发控制4.1 模型调用参数速查表模型调用参数不是随便填的它们直接影响 AI Work 的成功率、延迟和成本。下面是一份常用参数速查表参数含义常见值调大影响调小影响temperature控制结果随机性分类任务 0.2写作任务 0.7结果更发散更有创造性输出更稳定、更保守max_tokens单次生成最大 token 数512 到 2048能覆盖更长输出但耗时和成本上升可能截断导致 JSON 不完整timeout单次请求超时时间30 到 120 秒避免误判慢请求但任务堆积风险增加更容易失败适合快速降级top_p核采样概率0.9 左右候选词更多输出更集中max_retries失败重试次数2 到 3 次提高成功率但放大成本和延迟成功率下降重点是timeout和max_retries。很多人只配置max_tokens却忘记了超时。结果模型接口无响应时工作流一直卡住大量任务堆在队列里。超时应该理解为“整体任务超时”不只是“单次请求超时”。4.2 重试策略要兼顾成功率和成本重试不是简单地把同一个请求再发一次。直接重发可能造成重复扣费也可能放大故障期间的调用压力。推荐的做法是采用指数退避并加入随机抖动def get_retry_delay(attempt: int, base_delay: float 1.0) - float: return base_delay * (2 ** attempt) random.uniform(0, 0.5)同时每个请求都要带上唯一的请求 ID。这样即使客户端重试服务端也能识别出同一业务请求避免重复生成、重复写库。请求 ID 在日志排查中也起到串联作用。重试还要区分错误类型。网络超时可以重试鉴权失败、参数错误这种 4xx 错误重试多少次都不会成功应该直接失败进入人工处理。示例代码中RETRYABLE_ERRORS只包含TimeoutError和ConnectionError就是这个原因。4.3 并发控制避免限流保留排队语义大模型接口通常有 QPS 或 token 速率限制。AI Work 如果同时发出大量请求很快会被限流造成大量重试反而更慢。简单的做法是使用 Python 信号量控制并发数import threading semaphore threading.Semaphore(5) def call_with_limit(prompt: str) - str: with semaphore: return llm_call_with_retry(prompt)生产环境更推荐用消息队列或任务队列来控制消费速度。任务进入队列后按固定的并发数消费每个任务都有独立状态。这样即使模型接口变慢也只是队列积压不会导致后端被击穿。学习环境直接串行调用即可重点是理解状态流转。生产环境再加入并发不要在 Demo 阶段过度设计。5. 常见问题与排查链路5.1 任务一直处于“运行中”现象AI Work 的某个任务在数据库里长期处于 RUNNING 状态没有结束。可能原因模型调用没有设置超时请求一直等在那里工作流进程崩溃但状态没有更新步骤之间出现死锁或无限循环重试次数过大加上指数退避导致单任务耗时非常长。检查方式先看任务日志里最后一次心跳时间确认任务是否还在执行查看进程日志搜索当前步骤名称统计单次模型调用的平均耗时和 P95 耗时判断是否异常。处理建议给每一步设置独立超时和整个工作流的整体超时状态更新使用“最后更新时间”字段超时未更新的任务由定时器标记为失败并告警避免在步骤里写 while True 循环必须设置最大循环次数。5.2 模型返回格式不稳定现象模型返回内容有时是合法 JSON有时是普通文本有时 JSON 解析成功但字段缺失。可能原因提示词没有明确要求 JSON 结构max_tokens太小输出被截断模型本身对不同输入理解不一致直接使用字符串切割解析没有用json.loads。检查方式打印原始响应确认是截断还是格式错误把截断响应与max_tokens对比检查是否到达长度上限记录连续失败的样本观察失败集中在哪种输入。处理建议提示词里给出 JSON 示例并使用“只返回 JSON不要解释”优先使用模型平台提供的结构化输出或函数调用能力解析失败时把原始响应和错误信息一起放入失败队列人工处理。5.3 重试导致重复执行现象模型调用重试成功但业务系统写了多条相同记录或者用户收到多次通知。可能原因重试的不是幂等请求第一次调用已经成功只是响应超时客户端误判为失败后重试写库操作没有唯一键约束。检查方式查看日志中同一个业务请求 ID 对应的调用次数在数据库查request_id或order_no是否重复检查工作流步骤是否是纯函数是否每次执行都会产生副作用。处理建议每次任务生成一个request_id并在写库前检查唯一键写库和通知操作放在流程末尾并使用事务重试逻辑只对可重试错误生效不能对所有异常无条件重试。5.4 排查清单现象可能原因检查方式处理建议任务一直 PENDING队列消费线程挂起查看消费者日志和线程数给任务设置整体超时任务失败但没有详细日志异常被吞掉检查 except 是否记录 error至少记录异常类型和堆栈信息模型返回非 JSON提示词未约束或输出被截断打印原始响应使用结构化输出或修复式追问重试导致重复写库缺少幂等键查库中 request_id 重复情况写库前检查唯一键并发一高就被限流所有任务同时调模型查看限流错误码用队列或信号量控制并发新版本流程出问题缺少版本控制查看当前配置是否指向新逻辑保留旧版本快速回切6. 从 Demo 到生产落地 AI Work 的工程要求6.1 状态持久化不能少示例中的WorkflowContext只存在于内存里进程一重启就丢失。生产环境必须把任务状态存到数据库让每个任务都能从断点继续或者至少能定位到失败原因。一张最简任务表可以这样设计CREATE TABLE ai_work_task ( id BIGINT PRIMARY KEY AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, workflow_name VARCHAR(64) NOT NULL, input_data JSON, status VARCHAR(16) NOT NULL, current_step VARCHAR(64), result JSON, error TEXT, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_status_update_time (status, update_time), UNIQUE KEY uk_request_id (request_id) );request_id做唯一键既保证幂等也方便日志查询。status用于标记任务处于哪个阶段current_step用于在失败时定位具体步骤。6.2 日志和监控要覆盖调用全链路AI Work 的日志不能只记录“成功了”或“失败了”至少要包含这些字段request_id, workflow_name, step_name, model_name, prompt_tokens, completion_tokens, latency_ms, status, error_type, error_message, timestamp这些字段可以回答四个问题哪个任务出了问题卡在哪一步模型调用花了多长时间、消耗了多少 token错误是重试后恢复还是彻底失败。监控指标建议至少覆盖任务成功率和平均耗时每个步骤的成功率模型调用错误率和重试率队列积压数量token 消耗成本。当某个步骤成功率低于阈值时应该触发告警而不是等用户投诉。6.3 灰度上线先处理低风险任务再逐步放量AI Work 直接全量上线风险很高。模型判断一旦出现系统性错误所有任务都会受影响。推荐分三步走第一阶段只处理内部测试任务关闭真实写入结果以人工核对为主第二阶段选择低风险任务放量比如摘要生成、内部工单分类第三阶段接入有资金、合规影响的任务并增加人工抽检和回滚开关。灰度过程中要对比人工处理结果和 AI Work 结果的差异记录失败样本。只有当失败样本比例下降到可接受范围才继续放大流量。6.4 学习环境与生产环境的差异维度学习环境生产环境状态存储内存对象数据库或消息队列异常处理打印异常重试、死信、人工介入日志控制台输出日志平台 链路追踪参数配置写在代码里配置中心动态调整权限本机 API Key密钥管理、最小权限监控手动查看指标 告警 看板上线方式直接运行灰度发布 快速回滚学习环境跑通的是逻辑生产环境要解决的是可靠性。不要把两者混为一谈。7. 最佳实践和下一步扩展方向7.1 可直接套用的检查清单AI Work 落地前可以按这份清单逐项检查每个任务是否有唯一request_id单次模型调用是否设置了超时重试是否使用指数退避 随机抖动重试是否只针对可恢复错误模型输出是否有格式校验关键字段缺失时是否进入失败分支或人工通道工作流状态是否持久化失败任务是否能定位到具体步骤是否记录模型、token、耗时、错误类型是否有灰度开关和版本回滚方案是否有并发控制避免接口限流是否对敏感数据做脱敏和权限控制。把这些检查项固化到代码审查和发布流程里比靠个人经验更可靠。7.2 从线性工作流到 Agent 循环本文示例是线性工作流步骤从上到下执行一次。真实业务中模型第一次输出可能不够好需要让模型根据反馈迭代。这就进入 Agent 循环常见模式有 ReAct 和 Plan-Execute。ReAct 模式里模型先思考、再决定调用工具、观察结果后继续思考直到完成任务。难度和工作量都比线性工作流大很多但能处理更开放的问题。建议先掌握线性 AI Work再逐步引入循环和工具调用。7.3 人工审批节点和知识库增强很多业务场景不能完全自动需要插入人工审批节点。AI Work 引擎应该支持“任务挂起-等待人工处理-继续执行”的状态。比如风险较高的退款申请AI 只做分类和摘要最终审批权保留给人。知识库增强是另一个常见的扩展方向。模型生成前先检索相关文档把检索结果拼进提示词能显著提高专业领域准确性。AI Work 不是在每个步骤都硬调一次模型而是要把检索、判据、生成、校验组合成一条完整链路。从“悟空四个月消失”这个观察回到工程本身可以看到一个很朴素的结论AI 产品的竞争力不在一次生成结果而在能否稳定地把结果送进业务系统、能否在失败时快速恢复、能否在复盘时还原完整决策链路。这些能力需要主动设计无法靠模型升级自动获得。先从小任务开始把一个客服工单、一份审核材料、一类舆情摘要处理稳定再逐步扩大 AI Work 的覆盖范围是比较稳妥的路径。 AI Work：把大模型能力编排进业务流程的工程实践
AI Work 并不是一个新模型也不是某家大厂发布的某个 App而是把大模型能力编排进业务流程后形成的一套可运行、可维护、可观测的工作流体系。最近关于“国产 AI Work”的讨论明显增多也有人用“悟空四个月‘消失’”来形容某类 AI 产品在热度消退后没有持续迭代。单看这个现象会觉得是运营问题或模型不够好但从工程视角观察更接近真相的原因是产品没有把 AI 能力沉淀成可稳定流转的业务流程也就是缺少 AI Work。模型能力决定 AI 的下限AI Work 决定 AI 的上限。单次对话能写出不错的文案这靠模型让上千条待办工单自动完成分类、摘要、分派、回访、结案并且每一步都可追踪、可回滚、可审计这必须依赖 AI Work。下面先厘清 AI Work 与传统自动化的边界再分析大厂重视它的工程原因然后用一个最小可运行示例演示如何实现最后补充参数设计、排错路径和生产落地清单。1. 先厘清边界AI Work 不是聊天机器人也不是作业脚本1.1 通俗理解AI Work 是给大模型装上一套业务流程普通聊天机器人的逻辑是“一问一答”用户输入文本模型返回文本流程结束。AI Work 更像一条组装好的流水线。它会先判断当前任务是什么决定调用哪套提示词再调用模型生成中间结果然后通过规则校验结果是否合格不合格的进入重试或人工处理最后把结果写入数据库、消息队列或报表系统。整个过程不是一次模型调用而是多次模型调用和规则判断的组合。举个例子客服工单处理如果只靠聊天机器人用户在对话框里问“怎么退货”模型能回答一段退货政策。但 AI Work 会把这个动作拆成读取工单文本调用模型提取订单号、商品、退货原因再调用规则判断是否满足退货条件不满足则生成驳回文案满足则创建退货单并通知仓库。每一步都有输入、输出、校验和异常分支。从使用场景看AI Work 适合输入量大、规则复杂、结果需要进入其他系统的任务比如工单分类、文档审核、舆情摘要、数据清洗、代码提交描述生成等。它不追求一次对话惊艳而追求一批任务稳定完成。1.2 从单模型调用到 AI Work为什么流程比单个提示词重要单次模型调用解决的是“生成”AI Work 解决的是“稳定地生成并交付”。在实际业务中一次生成很少能直接成为最终结果。生产环境需要面对格式错误、字段缺失、置信度不足、调用超时、限流、重复提交等问题。这些问题无法靠更换一个更大模型解决必须通过流程设计来兜底。单模型调用可以写成response llm_client.chat(prompt) text response.content这种写法适合跑通原型。一旦进入生产就需要把上面这行代码扩展成包含输入校验、调用参数、超时控制、输出解析、结果校验、失败重试、日志记录的完整 AI Work。工作流的价值不是让模型变聪明而是让不稳定的模型输出在业务边界内稳定落地。大厂重视 AI Work 的核心原因也在这里模型能力可以通过采购或自研获得但把模型接进业务系统的工程能力只能靠团队自己沉淀。同一个大模型接口有人只能做出聊天 Demo有人能做出自动化运营系统差别就在工作流设计。1.3 AI Work 与 RPA、传统工作流的区别判断一个系统是否属于 AI Work可以从三个特征来看第一流程中存在大模型生成或决策节点第二生成结果会被后续步骤校验、转换和消费第三执行链路能够被跟踪和恢复。传统工作流和 RPA 也强调流程但决策来自预定义规则几乎不依赖模型泛化能力。AI Work 引入了模型节点之后流程的复杂度从“分支多”变成“不确定性高”因此需要额外的工程保护。维度传统工作流RPAAI Work处理对象结构化数据和固定接口界面操作和模拟点击文本、语义、决策和外部工具混合规则来源预定义业务规则预定义操作步骤模型生成 人工规则校验异常处理条件分支固定重试重试、校验、兜底、人工介入最怕的问题流程节点变化界面元素变化模型输出不稳定典型落地审批流、订单流报表自动化、系统切换工单、审核、摘要、辅助决策从表格可以看到AI Work 并不是替代 RPA 或传统工作流而是在它们之上增加了“模型决策节点”。这带来的好处是能处理非结构化输入坏处是必须为模型的不确定性设计额外的保护机制。2. 大厂为什么把目光移到 AI Work四个工程原因2.1 模型能力被拉平工作流成为落地差异点过去几年产品竞争重点在大模型本身谁的参数多、谁的榜单高谁就有优势。但模型能力发展到一定阶段后开源模型和商业模型之间的差距逐渐缩小单纯提供“模型调用”已经很难形成产品壁垒。此时真正决定用户体验的是模型输出之后产品如何解析、校验、过滤、排序、回退和转人工。这些逻辑都长在 AI Work 里。同样一个摘要模型A 产品直接把模型输出展示给用户B 产品会先判断摘要是否完整、是否包含敏感词、是否覆盖关键指标异常时自动重试或降级。用户感知到的差距不是模型聪明程度而是流程的成熟度。所以大厂强调 AI Work本质是把竞争从“谁的模型更强”转向“谁能把模型用得更好”。这更接近软件工程问题也更适合团队长期积累。2.2 AI Work 可以把不可控输出变成可控流程大模型输出具有概率性同一个提示词在不同时间可能返回不同内容。如果业务直接把模型结果写入数据库大概率会出现字段缺失、格式错误、逻辑冲突等问题。AI Work 的处理方式是给模型输出加约束。常见的做法包括提示词中明确要求返回 JSON并给出字段示例使用结构化输出或函数调用能力让模型按 Schema 返回模型返回后用代码校验必填字段、类型、范围、枚举值校验失败时重新调用模型进行“修复式追问”而不是直接抛错连续多次失败后转入人工处理通道。这些约束保证进入下游系统的数据完整、可信。AI Work 的本质就是把不可控的模型生成行为用工程手段约束到业务可接受的范围内。2.3 成本与延迟路由、缓存、并发控制大模型调用不是免费的成本由 token 数量决定延迟由模型大小和排队情况决定。如果不做工作流所有请求都发给最大模型成本会快速失控。AI Work 可以在流程中设计分级路由route_rules: - condition: intent greeting model: small-fast-model max_tokens: 64 - condition: intent complex_query model: large-model max_tokens: 1024 - default_model: medium-model简单问题走小模型复杂问题走大模型重复问题走缓存。语义缓存可以把相同问题的历史回答直接返回省去一次模型调用。再加上并发控制在合理范围内让系统不会因为瞬时流量被限流整体成本更可控。延迟方面AI Work 可以做到部分步骤并行。比如文档审核工作流可以先并行抽取标题、作者、风险词最后再汇总。单步模型调用延迟没有变但整体任务耗时明显下降。2.4 合规和审计每步可追踪、可回滚业务系统一旦接入 AI就会出现一个很难回答的问题这个结果是谁生成的依据是什么如果出错如何定位AI Work 通过流程编排天然记录每一步的输入、输出、模型、参数和异常信息。在审计场景下这非常关键。比如银行对公开户材料审核模型判断“资料齐全”系统必须能追溯是哪次调用、哪个模型、哪些字段、置信度多少。AI Work 把每一步执行结果和上下文串起来相当于给 AI 决策加了一条完整时间线。可回滚也很重要。当新流程版本出现明显错误时可以保留旧版本的执行逻辑通过配置中心快速切换而不是重新发版。这也是大厂在设计 AI Work 平台时会把版本管理放在第一优先级的原因。3. 从零搭建一个最小 AI Work 示例3.1 最小示例解决什么问题先直接看代码再解释设计。下面这个 Python 示例会完成一条客服工单的自动处理输入工单文本AI Work 自动完成预处理、模型调用、结果校验、落库展示。模型调用超时会自动重试结果置信度过低会让任务失败所有状态保存在上下文对象中方便后续替换成数据库。示例目标是体现 AI Work 的四个关键能力用状态机表达任务生命周期用上下文对象在步骤之间传递数据用异常和重试处理模型不稳定用校验步骤拦截不合格结果。3.2 目录与运行环境运行环境只有两个要求Python 3.10 以上无第三方依赖。目录结构保持最小ai-work-demo/ ├── workflow_demo.py └── README.md把全部代码放在workflow_demo.py便于在一篇文章里讲清楚。实际项目中会把步骤拆到不同模块但核心思想不变。3.3 用状态机和上下文表达工作流状态机解决“任务处于什么阶段”的问题。这里定义任务级别的状态和步骤级别的状态from enum import Enum class WorkflowStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed class StepStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed RETRYING retrying上下文对象解决“步骤之间如何传数据”的问题。不要把中间结果存在全局变量里应该统一放在上下文对象中这样每个步骤都能读取和写入排错时也容易打印from dataclasses import dataclass, field from typing import Any dataclass class WorkflowContext: input_data: dict records: dict field(default_factorydict) workflow_status: WorkflowStatus WorkflowStatus.PENDING current_step: str step_status: StepStatus StepStatus.PENDING error: str records字典会保存每个步骤的关键输出。没有它后续步骤只能重复调用前面的逻辑工作流就退化成顺序脚本。3.4 核心代码实现先写一个模拟大模型调用的函数。它本身带有重试、退避和模拟异常逻辑import json import random import time RETRYABLE_ERRORS (TimeoutError, ConnectionError) def llm_call_with_retry(prompt: str, max_retries: int 3, base_delay: float 0.5) - str: for attempt in range(max_retries): try: print(f[llm] 第 {attempt 1} 次调用prompt 长度 {len(prompt)}) # 模拟模型接口偶尔超时 if random.random() 0.15: raise TimeoutError(模拟模型接口超时) # 正常情况返回一段 JSON 字符串 return json.dumps({ summary: 用户申请退货需要确认商品是否已寄回, category: 售后, confidence: 0.96 }, ensure_asciiFalse) except RETRYABLE_ERRORS as e: if attempt max_retries - 1: raise sleep_time base_delay * (2 ** attempt) random.uniform(0, 0.3) print(f[llm] 调用异常{e}{sleep_time:.2f}s 后重试) time.sleep(sleep_time)接下来定义四个步骤。每个步骤都接收同一个WorkflowContext对象并往里写入结果def step_preprocess(ctx: WorkflowContext) - None: raw ctx.input_data.get(ticket, ).strip() if not raw: raise ValueError(工单内容不能为空) ctx.records[normalized_ticket] raw def step_llm_summary(ctx: WorkflowContext) - None: prompt ( 请对下面的客服工单做摘要和分类只返回 JSON 字段包括 summary、category、confidence。\n f工单内容{ctx.records[normalized_ticket]} ) response llm_call_with_retry(prompt) try: data json.loads(response) except json.JSONDecodeError as e: raise ValueError(f模型输出不是合法 JSON{response}) from e if not isinstance(data, dict): raise ValueError(模型输出不是 JSON 对象) if summary not in data or category not in data or confidence not in data: raise ValueError(模型输出缺少 summary/category/confidence 字段) ctx.records[llm_result] data def step_validate(ctx: WorkflowContext) - None: data ctx.records.get(llm_result, {}) if float(data.get(confidence, 0)) 0.9: raise RuntimeError(置信度过低需要人工复核) ctx.records[validated] True def step_save(ctx: WorkflowContext) - None: # 实际项目里写数据库或消息队列 ctx.records[saved_id] TICKET-0001工作流引擎只需要一个简单的循环顺序执行步骤任何异常都会把整个任务标记为失败并记录当前步骤和错误信息class LLMWorkflow: def __init__(self, steps): self.steps steps def run(self, ctx: WorkflowContext) - WorkflowContext: ctx.workflow_status WorkflowStatus.RUNNING for step in self.steps: ctx.current_step step.__name__ ctx.step_status StepStatus.RUNNING print(f[workflow] 开始步骤{ctx.current_step}) try: step(ctx) ctx.step_status StepStatus.SUCCESS print(f[workflow] 步骤完成{ctx.current_step}) except Exception as e: ctx.step_status StepStatus.FAILED ctx.error f{ctx.current_step}: {e} ctx.workflow_status WorkflowStatus.FAILED return ctx ctx.workflow_status WorkflowStatus.SUCCESS return ctx最后是入口。输入一条工单创建上下文执行工作流输出结果if __name__ __main__: workflow LLMWorkflow([ step_preprocess, step_llm_summary, step_validate, step_save, ]) ctx WorkflowContext(input_data{ ticket: 我上周买的杯子碎了想申请退货。 }) workflow.run(ctx) print(json.dumps({ workflow_status: ctx.workflow_status.value, current_step: ctx.current_step, records: ctx.records, error: ctx.error }, ensure_asciiFalse, indent2))3.5 运行验证和预期输出在项目目录下执行python workflow_demo.py正常情况会看到类似输出[workflow] 开始步骤step_preprocess [workflow] 步骤完成step_preprocess [workflow] 开始步骤step_llm_summary [llm] 第 1 次调用prompt 长度 52 [workflow] 步骤完成step_llm_summary [workflow] 开始步骤step_validate [workflow] 步骤完成step_validate [workflow] 开始步骤step_save [workflow] 步骤完成step_save { workflow_status: success, current_step: step_save, records: { normalized_ticket: 我上周买的杯子碎了想申请退货。, llm_result: { summary: 用户申请退货需要确认商品是否已寄回, category: 售后, confidence: 0.96 }, validated: true, saved_id: TICKET-0001 }, error: }如果模型调用连续失败或结果校验不通过输出中workflow_status会变成failederror会记录具体哪一步出了什么问题。这就验证了 AI Work 的异常处理链路。注意不要只验证程序能启动还要故意把confidence改小、把模型返回改成非 JSON观察工作流是否正确进入失败分支。只有失败分支可控才能算真正的 AI Work。4. 关键设计超时、重试、参数和并发控制4.1 模型调用参数速查表模型调用参数不是随便填的它们直接影响 AI Work 的成功率、延迟和成本。下面是一份常用参数速查表参数含义常见值调大影响调小影响temperature控制结果随机性分类任务 0.2写作任务 0.7结果更发散更有创造性输出更稳定、更保守max_tokens单次生成最大 token 数512 到 2048能覆盖更长输出但耗时和成本上升可能截断导致 JSON 不完整timeout单次请求超时时间30 到 120 秒避免误判慢请求但任务堆积风险增加更容易失败适合快速降级top_p核采样概率0.9 左右候选词更多输出更集中max_retries失败重试次数2 到 3 次提高成功率但放大成本和延迟成功率下降重点是timeout和max_retries。很多人只配置max_tokens却忘记了超时。结果模型接口无响应时工作流一直卡住大量任务堆在队列里。超时应该理解为“整体任务超时”不只是“单次请求超时”。4.2 重试策略要兼顾成功率和成本重试不是简单地把同一个请求再发一次。直接重发可能造成重复扣费也可能放大故障期间的调用压力。推荐的做法是采用指数退避并加入随机抖动def get_retry_delay(attempt: int, base_delay: float 1.0) - float: return base_delay * (2 ** attempt) random.uniform(0, 0.5)同时每个请求都要带上唯一的请求 ID。这样即使客户端重试服务端也能识别出同一业务请求避免重复生成、重复写库。请求 ID 在日志排查中也起到串联作用。重试还要区分错误类型。网络超时可以重试鉴权失败、参数错误这种 4xx 错误重试多少次都不会成功应该直接失败进入人工处理。示例代码中RETRYABLE_ERRORS只包含TimeoutError和ConnectionError就是这个原因。4.3 并发控制避免限流保留排队语义大模型接口通常有 QPS 或 token 速率限制。AI Work 如果同时发出大量请求很快会被限流造成大量重试反而更慢。简单的做法是使用 Python 信号量控制并发数import threading semaphore threading.Semaphore(5) def call_with_limit(prompt: str) - str: with semaphore: return llm_call_with_retry(prompt)生产环境更推荐用消息队列或任务队列来控制消费速度。任务进入队列后按固定的并发数消费每个任务都有独立状态。这样即使模型接口变慢也只是队列积压不会导致后端被击穿。学习环境直接串行调用即可重点是理解状态流转。生产环境再加入并发不要在 Demo 阶段过度设计。5. 常见问题与排查链路5.1 任务一直处于“运行中”现象AI Work 的某个任务在数据库里长期处于 RUNNING 状态没有结束。可能原因模型调用没有设置超时请求一直等在那里工作流进程崩溃但状态没有更新步骤之间出现死锁或无限循环重试次数过大加上指数退避导致单任务耗时非常长。检查方式先看任务日志里最后一次心跳时间确认任务是否还在执行查看进程日志搜索当前步骤名称统计单次模型调用的平均耗时和 P95 耗时判断是否异常。处理建议给每一步设置独立超时和整个工作流的整体超时状态更新使用“最后更新时间”字段超时未更新的任务由定时器标记为失败并告警避免在步骤里写 while True 循环必须设置最大循环次数。5.2 模型返回格式不稳定现象模型返回内容有时是合法 JSON有时是普通文本有时 JSON 解析成功但字段缺失。可能原因提示词没有明确要求 JSON 结构max_tokens太小输出被截断模型本身对不同输入理解不一致直接使用字符串切割解析没有用json.loads。检查方式打印原始响应确认是截断还是格式错误把截断响应与max_tokens对比检查是否到达长度上限记录连续失败的样本观察失败集中在哪种输入。处理建议提示词里给出 JSON 示例并使用“只返回 JSON不要解释”优先使用模型平台提供的结构化输出或函数调用能力解析失败时把原始响应和错误信息一起放入失败队列人工处理。5.3 重试导致重复执行现象模型调用重试成功但业务系统写了多条相同记录或者用户收到多次通知。可能原因重试的不是幂等请求第一次调用已经成功只是响应超时客户端误判为失败后重试写库操作没有唯一键约束。检查方式查看日志中同一个业务请求 ID 对应的调用次数在数据库查request_id或order_no是否重复检查工作流步骤是否是纯函数是否每次执行都会产生副作用。处理建议每次任务生成一个request_id并在写库前检查唯一键写库和通知操作放在流程末尾并使用事务重试逻辑只对可重试错误生效不能对所有异常无条件重试。5.4 排查清单现象可能原因检查方式处理建议任务一直 PENDING队列消费线程挂起查看消费者日志和线程数给任务设置整体超时任务失败但没有详细日志异常被吞掉检查 except 是否记录 error至少记录异常类型和堆栈信息模型返回非 JSON提示词未约束或输出被截断打印原始响应使用结构化输出或修复式追问重试导致重复写库缺少幂等键查库中 request_id 重复情况写库前检查唯一键并发一高就被限流所有任务同时调模型查看限流错误码用队列或信号量控制并发新版本流程出问题缺少版本控制查看当前配置是否指向新逻辑保留旧版本快速回切6. 从 Demo 到生产落地 AI Work 的工程要求6.1 状态持久化不能少示例中的WorkflowContext只存在于内存里进程一重启就丢失。生产环境必须把任务状态存到数据库让每个任务都能从断点继续或者至少能定位到失败原因。一张最简任务表可以这样设计CREATE TABLE ai_work_task ( id BIGINT PRIMARY KEY AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, workflow_name VARCHAR(64) NOT NULL, input_data JSON, status VARCHAR(16) NOT NULL, current_step VARCHAR(64), result JSON, error TEXT, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_status_update_time (status, update_time), UNIQUE KEY uk_request_id (request_id) );request_id做唯一键既保证幂等也方便日志查询。status用于标记任务处于哪个阶段current_step用于在失败时定位具体步骤。6.2 日志和监控要覆盖调用全链路AI Work 的日志不能只记录“成功了”或“失败了”至少要包含这些字段request_id, workflow_name, step_name, model_name, prompt_tokens, completion_tokens, latency_ms, status, error_type, error_message, timestamp这些字段可以回答四个问题哪个任务出了问题卡在哪一步模型调用花了多长时间、消耗了多少 token错误是重试后恢复还是彻底失败。监控指标建议至少覆盖任务成功率和平均耗时每个步骤的成功率模型调用错误率和重试率队列积压数量token 消耗成本。当某个步骤成功率低于阈值时应该触发告警而不是等用户投诉。6.3 灰度上线先处理低风险任务再逐步放量AI Work 直接全量上线风险很高。模型判断一旦出现系统性错误所有任务都会受影响。推荐分三步走第一阶段只处理内部测试任务关闭真实写入结果以人工核对为主第二阶段选择低风险任务放量比如摘要生成、内部工单分类第三阶段接入有资金、合规影响的任务并增加人工抽检和回滚开关。灰度过程中要对比人工处理结果和 AI Work 结果的差异记录失败样本。只有当失败样本比例下降到可接受范围才继续放大流量。6.4 学习环境与生产环境的差异维度学习环境生产环境状态存储内存对象数据库或消息队列异常处理打印异常重试、死信、人工介入日志控制台输出日志平台 链路追踪参数配置写在代码里配置中心动态调整权限本机 API Key密钥管理、最小权限监控手动查看指标 告警 看板上线方式直接运行灰度发布 快速回滚学习环境跑通的是逻辑生产环境要解决的是可靠性。不要把两者混为一谈。7. 最佳实践和下一步扩展方向7.1 可直接套用的检查清单AI Work 落地前可以按这份清单逐项检查每个任务是否有唯一request_id单次模型调用是否设置了超时重试是否使用指数退避 随机抖动重试是否只针对可恢复错误模型输出是否有格式校验关键字段缺失时是否进入失败分支或人工通道工作流状态是否持久化失败任务是否能定位到具体步骤是否记录模型、token、耗时、错误类型是否有灰度开关和版本回滚方案是否有并发控制避免接口限流是否对敏感数据做脱敏和权限控制。把这些检查项固化到代码审查和发布流程里比靠个人经验更可靠。7.2 从线性工作流到 Agent 循环本文示例是线性工作流步骤从上到下执行一次。真实业务中模型第一次输出可能不够好需要让模型根据反馈迭代。这就进入 Agent 循环常见模式有 ReAct 和 Plan-Execute。ReAct 模式里模型先思考、再决定调用工具、观察结果后继续思考直到完成任务。难度和工作量都比线性工作流大很多但能处理更开放的问题。建议先掌握线性 AI Work再逐步引入循环和工具调用。7.3 人工审批节点和知识库增强很多业务场景不能完全自动需要插入人工审批节点。AI Work 引擎应该支持“任务挂起-等待人工处理-继续执行”的状态。比如风险较高的退款申请AI 只做分类和摘要最终审批权保留给人。知识库增强是另一个常见的扩展方向。模型生成前先检索相关文档把检索结果拼进提示词能显著提高专业领域准确性。AI Work 不是在每个步骤都硬调一次模型而是要把检索、判据、生成、校验组合成一条完整链路。从“悟空四个月消失”这个观察回到工程本身可以看到一个很朴素的结论AI 产品的竞争力不在一次生成结果而在能否稳定地把结果送进业务系统、能否在失败时快速恢复、能否在复盘时还原完整决策链路。这些能力需要主动设计无法靠模型升级自动获得。先从小任务开始把一个客服工单、一份审核材料、一类舆情摘要处理稳定再逐步扩大 AI Work 的覆盖范围是比较稳妥的路径。