AI 自动化工作流设计：从单次调用到多步编排的可靠性实践
AI 自动化工作流设计：从单次调用到多步编排的可靠性实践
一、单次调用到多步编排：AI 工作流的复杂性跃迁
当 AI 能力从单次问答走向多步骤自动化工作流时，系统的复杂度会发生质变。一个典型的内容审核工作流可能包含：文本提取、敏感词检测、语义分析、审核决策生成、结果通知五个步骤。每个步骤都依赖上一步的输出，而每一步都可能失败。
单次 AI 调用的失败处理相对简单——重试或降级即可。但在多步工作流中，第三步失败后，是否需要重试前两步？如果第五步的模型输出格式异常，是否需要从第四步重新开始？这些问题在单次调用场景中不存在，但在工作流场景中必须明确回答。
更关键的是，多步工作流的成本是累加的。一个五步工作流，每步平均消耗 2000 Token，单次执行就需要 10000 Token。如果缺乏执行控制，一个异常触发的工作流循环可能在几分钟内消耗掉整月的 API 预算。因此，AI 自动化工作流的设计，核心不是"如何编排步骤"，而是"如何安全地编排步骤"。
二、工作流引擎的状态机模型
AI 工作流的本质是一个状态机：每个步骤是一个状态，步骤间的依赖关系定义了状态转移规则。引入状态机模型后，工作流具备了可观测性（当前执行到哪一步）、可恢复性（从任意步骤重新开始）和可审计性（每步的输入输出都有记录）。
stateDiagram-v2 [*] --> 初始化: 触发工作流 初始化 --> 数据采集: 参数校验通过 初始化 --> 失败: 参数校验失败 数据采集 --> AI预处理: 数据获取成功 数据采集 --> 重试队列: 获取超时 重试队列 --> 数据采集: 重试触发 AI预处理 --> AI推理: 预处理完成 AI预处理 --> 失败: 数据格式异常 AI推理 --> 结果校验: 推理完成 AI推理 --> 降级推理: 主模型超时 降级推理 --> 结果校验: 降级模型返回 结果校验 --> 结果输出: 校验通过 结果校验 --> AI推理: 结果不合格(重试) 结果校验 --> 失败: 重试次数耗尽 结果输出 --> 通知: 输出完成 通知 --> [*]: 工作流结束 失败 --> [*]: 记录日志并告警状态机模型的核心优势在于：每个状态的退出条件是明确的。数据采集超时后进入重试队列，而非无限等待；AI 推理超时后切换到降级模型，而非直接失败；结果校验不通过时，可以选择重试推理而非从头开始。这种细粒度的错误处理，是工作流可靠性的基础。
三、生产级工作流引擎的代码实现
3.1 工作流定义与状态管理
type WorkflowStatus = 'pending' | 'running' | 'paused' | 'completed' | 'failed'; interface WorkflowStep<TInput, TOutput> { name: string; execute: (input: TInput, context: WorkflowContext) => Promise<TOutput>; // 超时时间，超时后自动标记为失败 timeoutMs: number; // 最大重试次数 maxRetries: number; // 是否可跳过（跳过后使用默认值继续） skippable: boolean; // 跳过时的默认输出 fallbackOutput?: TOutput; } interface WorkflowContext { workflowId: string; currentStep: number; totalSteps: number; retryCount: number; metadata: Record<string, unknown>; // 步骤执行记录，用于审计和恢复 stepHistory: Array<{ stepName: string; startedAt: number; completedAt: number; status: 'success' | 'failed' | 'skipped'; error?: string; }>; } interface WorkflowDefinition { name: string; steps: WorkflowStep<unknown, unknown>[]; // 全局超时：整个工作流的最大执行时间 globalTimeoutMs: number; // 全局 Token 预算：累计消耗超过此值时中止 maxTokenBudget: number; }3.2 工作流执行引擎
class WorkflowEngine { private tokenUsage: number = 0; /** * 执行工作流，支持断点恢复和超时控制 * 设计考量： * - 每步执行前检查全局预算，防止成本失控 * - 步骤失败时根据配置决定重试/跳过/中止 * - 执行记录持久化，支持从任意步骤恢复 */ async execute( definition: WorkflowDefinition, initialInput: unknown, resumeFromStep?: number ): Promise<{ status: WorkflowStatus; output: unknown; context: WorkflowContext }> { const context: WorkflowContext = { workflowId: crypto.randomUUID(), currentStep: resumeFromStep ?? 0, totalSteps: definition.steps.length, retryCount: 0, metadata: {}, stepHistory: [], }; let currentInput = initialInput; const globalStart = Date.now(); for (let i = context.currentStep; i < definition.steps.length; i++) { const step = definition.steps[i]; context.currentStep = i; // 全局超时检查 if (Date.now() - globalStart > definition.globalTimeoutMs) { return this.fail(context, '工作流全局超时'); } // Token 预算检查 if (this.tokenUsage > definition.maxTokenBudget) { return this.fail(context, `Token 预算耗尽：已消耗 ${this.tokenUsage}`); } const stepResult = await this.executeStep(step, currentInput, context); if (stepResult.status === 'success') { currentInput = stepResult.output; this.tokenUsage += stepResult.tokensUsed ?? 0; } else if (stepResult.status === 'skipped') { // 使用降级输出继续执行 currentInput = step.fallbackOutput; } else { // 步骤彻底失败 return this.fail(context, `步骤 ${step.name} 执行失败：${stepResult.error}`); } } context.currentStep = definition.steps.length; return { status: 'completed', output: currentInput, context }; } private async executeStep( step: WorkflowStep<unknown, unknown>, input: unknown, context: WorkflowContext ): Promise<{ status: 'success' | 'failed' | 'skipped'; output?: unknown; error?: string; tokensUsed?: number }> { let lastError: string = ''; const stepStart = Date.now(); for (let attempt = 0; attempt <= step.maxRetries; attempt++) { try { // 单步超时控制 const result = await Promise.race([ step.execute(input, context), new Promise<never>((_, reject) => setTimeout(() => reject(new Error('步骤超时')), step.timeoutMs) ), ]); context.stepHistory.push({ stepName: step.name, startedAt: stepStart, completedAt: Date.now(), status: 'success', }); return { status: 'success', output: result }; } catch (error) { lastError = (error as Error).message; // 重试前等待，避免立即重试加重服务端压力 if (attempt < step.maxRetries) { await new Promise(resolve => setTimeout(resolve, 1000 * Math.pow(2, attempt))); } } } // 重试耗尽，判断是否可跳过 if (step.skippable && step.fallbackOutput !== undefined) { context.stepHistory.push({ stepName: step.name, startedAt: stepStart, completedAt: Date.now(), status: 'skipped', error: lastError, }); return { status: 'skipped' }; } context.stepHistory.push({ stepName: step.name, startedAt: stepStart, completedAt: Date.now(), status: 'failed', error: lastError, }); return { status: 'failed', error: lastError }; } private fail(context: WorkflowContext, reason: string) { return { status: 'failed' as WorkflowStatus, output: null, context }; } }3.3 内容审核工作流的具体编排
const contentModerationWorkflow: WorkflowDefinition = { name: '内容审核工作流', globalTimeoutMs: 60000, // 全局 60 秒超时 maxTokenBudget: 15000, // 单次执行最多消耗 15000 Token steps: [ { name: '文本提取', execute: async (input, ctx) => { const { content, format } = input as { content: string; format: string }; // HTML 内容清洗，提取纯文本 if (format === 'html') { return content.replace(/<[^>]*>/g, '').trim(); } return content; }, timeoutMs: 5000, maxRetries: 1, skippable: false, }, { name: 'AI 敏感词检测', execute: async (text, ctx) => { // 调用大模型进行语义级敏感词检测 const result = await callModel('gpt-4o-mini', text as string, '检测文本中是否包含敏感内容'); return { text, sensitiveResult: result }; }, timeoutMs: 10000, maxRetries: 2, skippable: true, fallbackOutput: { text: '', sensitiveResult: { flagged: false, reason: '检测服务不可用' } }, }, { name: '审核决策生成', execute: async (data, ctx) => { const { text, sensitiveResult } = data as { text: string; sensitiveResult: unknown }; if ((sensitiveResult as { flagged: boolean }).flagged) { return { decision: 'reject', reason: '内容包含敏感信息' }; } return { decision: 'approve', reason: '内容审核通过' }; }, timeoutMs: 3000, maxRetries: 0, skippable: false, }, ], };四、工作流编排的隐性风险与适用边界
状态爆炸问题：当工作流步骤超过 10 个，且步骤间存在条件分支时，状态机的状态数量会呈指数增长。一个包含 3 个条件分支的 10 步工作流，可能的状态路径多达数千条。此时，状态机模型的可读性会急剧下降，维护成本远超收益。对于这种复杂场景，应考虑使用 DAG（有向无环图）模型替代线性状态机，或者将工作流拆分为多个子工作流。
成本控制的精度问题：Token 预算是粗粒度的成本控制手段，但实际场景中，不同步骤的 Token 消耗差异巨大。文本提取步骤几乎不消耗 Token，而 AI 推理步骤可能消耗数千 Token。如果统一预算，简单步骤会"浪费"预算配额。更精细的做法是为每个步骤设置独立的 Token 预算，但这也增加了配置复杂度。
断点恢复的一致性风险：从某个步骤恢复执行时，前序步骤的输出可能已经过期。例如，数据采集步骤获取的是 10 分钟前的数据，从第三步恢复时，这些数据可能已经不再准确。对于时效性要求高的工作流，断点恢复可能引入数据一致性问题，需要根据业务场景决定是否允许恢复。
工作流 vs 简单管道：并非所有多步骤场景都需要工作流引擎。如果一个流程是严格线性的、无需重试和断点恢复，一个简单的 Promise 链就足够了。工作流引擎的引入成本包括：状态持久化、步骤定义配置、执行日志存储。当流程步骤少于 5 个且失败率极低时，这些成本可能不值得支付。
五、总结
AI 自动化工作流的设计核心，是在编排灵活性与执行可靠性之间取得平衡。状态机模型为工作流提供了可观测、可恢复、可审计的执行框架，但同时也引入了状态管理和配置复杂度。在实际落地中，工作流引擎的引入时机应基于具体的业务需求：当流程步骤超过 5 个、存在条件分支、需要断点恢复或成本控制时，工作流引擎的价值才会显现。
落地建议：第一步，从线性管道开始，用 Promise 链串联步骤，验证流程逻辑的正确性；第二步，当需要重试和超时控制时，引入步骤级别的错误处理；第三步，当需要断点恢复和成本控制时，再升级为完整的工作流引擎。渐进式演进可以避免过早引入不必要的复杂度。