独立产品 AI 工作流编排架构设计多步骤 AI 任务的依赖管理与容错一、独立产品 AI 化的策略性挑战从单次调用到工作流编排的复杂度跃迁独立产品引入 AI 能力的第一阶段通常是单次调用用户输入 → 调用 LLM API → 返回结果。这个模式足以支撑简单的文本生成、翻译或摘要功能。但当一个产品需要实现上传产品截图 → AI 分析设计问题 → 生成优化建议 → 输出设计规范 → 导出 Figma 插件配置这样的多步骤链路时单次调用的架构会迅速崩塌。多步骤 AI 工作流的核心挑战包括任务依赖管理步骤 2 的输入依赖于步骤 1 的输出如果步骤 1 返回格式不符合预期整个链路中断。中间状态持久化用户需要看到每一步的进度、中间结果并且支持从任意步骤重试。容错与降级LLM 调用存在一定的不确定性响应超时、格式错误、内容审核拦截需要有兜底策略。成本控制多步骤调用可能产生数倍于单次调用的 Token 消耗需要合理的中断和缓存机制。并发编排部分步骤可以并行执行如图像分析和文本分析需要在 DAG有向无环图中识别并行机会。传统方案通常直接在业务代码中用await串联调用但这种写法在步骤超过 3 步时就会变得难以维护——错误处理混杂在业务逻辑中、重试策略分散在各处、步骤间状态传递依赖闭包变量。二、AI 工作流编排的核心抽象2.1 步骤Step的定义每一个 AI 工作流步骤需要包含以下维度输入/输出定义使用 TypeScript 泛型约束步骤间的数据传递类型。执行策略最大重试次数、退避间隔、超时时间、是否可跳过。降级函数当主逻辑失败时执行的备选方案。例如 LLM 调用失败时返回缓存结果或预设默认值。依赖声明明确声明该步骤依赖哪些前置步骤的输出。2.2 DAG 的执行调度工作流的依赖关系构成一个有向无环图DAG。调度器通过拓扑排序确定执行顺序——入度为 0 的节点可以立即执行当一个节点执行完成后将其后继节点的入度减 1入度变为 0 的后继节点进入就绪队列。并行执行的关键在于正确识别独立子图。两个节点如果没有直接或间接的依赖关系它们分属不同的拓扑层级可以并行调度。对于 I/O 密集型的 AI 调用如同时向两个不同的 LLM 服务发送请求并行执行可以显著缩短总耗时。2.3 中间状态与检查点工作流执行过程中需要持久化每个步骤的状态。状态数据应包含步骤标识和执行状态pending / running / completed / failed / skipped步骤的输出数据用于传递给下游步骤执行时间戳和重试次数错误信息用于调试和重试决策这些数据可以存储在 Redis 或文件系统中。当工作流中断后用户可以从上次失败或最后一个成功步骤恢复执行而不必从头开始——这对 Token 成本控制至关重要。三、轻量级 AI 工作流引擎的实现/** * 独立产品 AI 工作流编排引擎 * 支持 DAG 依赖管理、并行调度、断点续传和降级策略 */ type StepStatus pending | running | completed | failed | skipped; interface StepDefinitionTInput, TOutput { id: string; name: string; /** 依赖的前置步骤 ID 列表 */ dependsOn: string[]; /** 步骤执行的最大重试次数 */ maxRetries?: number; /** 重试退避间隔毫秒 */ retryBackoffMs?: number; /** 步骤超时时间毫秒 */ timeoutMs?: number; /** 步骤失败时是否可以跳过 */ skippable?: boolean; /** 输入数据转换从上游输出中提取本步骤所需输入 */ extractInput: (context: WorkflowContext) TInput; /** 核心执行函数 */ execute: (input: TInput) PromiseTOutput; /** 降级函数execute 失败时的备选方案 */ fallback?: (input: TInput, error: Error) PromiseTOutput; } interface StepResult { stepId: string; status: StepStatus; output?: unknown; error?: string; startedAt: number; completedAt: number; retryCount: number; } interface WorkflowContext { /** 步骤输出数据映射表 */ results: Mapstring, StepResult; } interface WorkflowDefinition { id: string; steps: StepDefinitionany, any[]; /** 全局超时毫秒 */ globalTimeoutMs?: number; } interface WorkflowState { workflowId: string; status: running | completed | failed; steps: Mapstring, StepResult; startedAt: number; } class AIWorkflowEngine { private runningWorkflows new Mapstring, AbortController(); /** * 执行工作流 */ async execute( definition: WorkflowDefinition, options: { /** 断点续传从已有的 workflowState 恢复 */ resumeFrom?: WorkflowState; } {} ): PromiseWorkflowState { const abortController new AbortController(); this.runningWorkflows.set(definition.id, abortController); // 构建步骤索引表 const stepMap new Mapstring, StepDefinitionany, any(); for (const step of definition.steps) { stepMap.set(step.id, step); } // 初始化上下文和状态 const context: WorkflowContext { results: new Map() }; const stepResults new Mapstring, StepResult(); // 断点续传恢复已完成步骤的结果 if (options.resumeFrom) { for (const [id, result] of options.resumeFrom.steps) { if (result.status completed || result.status skipped) { stepResults.set(id, result); context.results.set(id, result); } } } // 计算入度依赖计数 const inDegree new Mapstring, number(); const successors new Mapstring, string[](); // 下游步骤 for (const step of definition.steps) { // 断点续传已完成的步骤不需要重新执行 if (stepResults.has(step.id)) { inDegree.set(step.id, 0); successors.set(step.id, []); continue; } inDegree.set(step.id, step.dependsOn.length); if (!successors.has(step.id)) { successors.set(step.id, []); } } // 构建后继关系 for (const step of definition.steps) { for (const depId of step.dependsOn) { const succ successors.get(depId) || []; succ.push(step.id); successors.set(depId, succ); } } // 找到所有入度为 0 的节点就绪队列 const readyQueue: string[] []; for (const [stepId, degree] of inDegree) { if (degree 0 !stepResults.has(stepId)) { readyQueue.push(stepId); } } const startedAt options.resumeFrom?.startedAt ?? Date.now(); let hasFailure false; // 核心调度循环只要还有就绪或运行中的步骤就继续 while (readyQueue.length 0 || this.hasRunningStep(context)) { // 检查全局超时 if ( definition.globalTimeoutMs Date.now() - startedAt definition.globalTimeoutMs ) { throw new Error(工作流执行超时${definition.globalTimeoutMs}ms); } if (abortController.signal.aborted) { throw new Error(工作流已被取消); } // 从就绪队列中取步骤并行执行 const batch: string[] []; while (readyQueue.length 0) { const id readyQueue.shift()!; batch.push(id); } if (batch.length 0) { const promises batch.map((stepId) this.executeStep(stepMap.get(stepId)!, context, abortController.signal) ); const results await Promise.allSettled(promises); for (let i 0; i batch.length; i) { const stepId batch[i]; const result results[i]; if (result.status fulfilled) { const stepResult result.value; stepResults.set(stepId, stepResult); context.results.set(stepId, stepResult); if (stepResult.status failed) { hasFailure true; // 如果某步骤失败且不可跳过整个工作流标记为失败 const step stepMap.get(stepId)!; if (!step.skippable) { return this.buildFailedState(definition.id, stepResults, startedAt); } } // 将后继步骤的入度减 1 for (const succId of successors.get(stepId) || []) { const currentDegree (inDegree.get(succId) || 0) - 1; inDegree.set(succId, currentDegree); if (currentDegree 0) { readyQueue.push(succId); } } } else { // Promise 被拒绝系统级错误 hasFailure true; const error result.reason instanceof Error ? result.reason : new Error(String(result.reason)); const stepResult: StepResult { stepId, status: failed, error: error.message, startedAt: 0, completedAt: Date.now(), retryCount: 0, }; stepResults.set(stepId, stepResult); context.results.set(stepId, stepResult); } } } // 如果就绪队列为空且没有失败短暂等待检查是否有新节点就绪 // 处理所有节点完成的情况 if (readyQueue.length 0 !hasFailure this.allCompleted(stepResults, definition.steps)) { break; } } // 最终状态 return { workflowId: definition.id, status: hasFailure ? failed : completed, steps: stepResults, startedAt, }; } /** * 执行单个步骤含重试和降级 */ private async executeStep( step: StepDefinitionany, any, context: WorkflowContext, signal: AbortSignal ): PromiseStepResult { const maxRetries step.maxRetries ?? 2; const backoff step.retryBackoffMs ?? 1000; const startedAt Date.now(); let lastError: Error | undefined; for (let attempt 0; attempt maxRetries; attempt) { if (signal.aborted) { return { stepId: step.id, status: failed, error: 工作流已被取消, startedAt, completedAt: Date.now(), retryCount: attempt, }; } try { // 从上下文中提取本步骤的输入 const input step.extractInput(context); // 执行核心逻辑添加超时控制 const output await this.withTimeout( step.execute(input), step.timeoutMs ?? 60_000, signal ); return { stepId: step.id, status: completed, output, startedAt, completedAt: Date.now(), retryCount: attempt, }; } catch (error) { lastError error instanceof Error ? error : new Error(String(error)); if (attempt maxRetries) { // 退避等待后重试 await this.sleep(backoff * Math.pow(2, attempt)); } } } // 所有重试都失败尝试降级 if (step.fallback) { try { const input step.extractInput(context); const fallbackOutput await step.fallback(input, lastError!); return { stepId: step.id, status: completed, // 降级成功视为完成 output: fallbackOutput, startedAt, completedAt: Date.now(), retryCount: maxRetries 1, }; } catch (fallbackError) { lastError fallbackError instanceof Error ? fallbackError : new Error(String(fallbackError)); } } return { stepId: step.id, status: failed, error: lastError?.message ?? 未知错误, startedAt, completedAt: Date.now(), retryCount: maxRetries 1, }; } /** * 带超时和取消信号的 Promise 包装 */ private async withTimeoutT( promise: PromiseT, timeoutMs: number, signal: AbortSignal ): PromiseT { return new PromiseT((resolve, reject) { const timer setTimeout(() reject(new Error(步骤执行超时)), timeoutMs); const onAbort () { clearTimeout(timer); reject(new Error(工作流已取消)); }; signal.addEventListener(abort, onAbort, { once: true }); promise .then((result) { clearTimeout(timer); signal.removeEventListener(abort, onAbort); resolve(result); }) .catch((error) { clearTimeout(timer); signal.removeEventListener(abort, onAbort); reject(error); }); }); } /** * 取消运行中的工作流 */ cancel(workflowId: string): void { const controller this.runningWorkflows.get(workflowId); if (controller) { controller.abort(); this.runningWorkflows.delete(workflowId); } } /** * 检查是否还有步骤在执行中 */ private hasRunningStep(context: WorkflowContext): boolean { // 实际实现中需要维护运行中步骤的计数器 // 这里为简化不做实时追踪 return false; } /** * 检查所有步骤是否已完成 */ private allCompleted( results: Mapstring, StepResult, steps: StepDefinitionany, any[] ): boolean { return steps.every((s) { const r results.get(s.id); return r (r.status completed || r.status skipped || r.status failed); }); } private buildFailedState( id: string, results: Mapstring, StepResult, startedAt: number ): WorkflowState { return { workflowId: id, status: failed, steps: results, startedAt, }; } private sleep(ms: number): Promisevoid { return new Promise((resolve) setTimeout(resolve, ms)); } } // ---- 使用示例截图分析工作流 ---- async function createScreenshotAnalysisWorkflow() { const engine new AIWorkflowEngine(); const workflow: WorkflowDefinition { id: screenshot-analysis, globalTimeoutMs: 120_000, steps: [ { id: analyze-ui, name: AI 分析截图 UI 结构, dependsOn: [], maxRetries: 2, timeoutMs: 30_000, extractInput: (ctx) ({ imageUrl: https://user-upload.example.com/screenshot.png, }), execute: async (input) { // 调用视觉模型分析截图 return { components: [Header, Card, Footer], layout: vertical-stack }; }, fallback: async (input, error) { // LLM 不可用时返回缓存的预设分析结果 return { components: [], layout: unknown, fallback: true }; }, }, { id: generate-suggestions, name: 生成优化建议, dependsOn: [analyze-ui], maxRetries: 1, extractInput: (ctx) { const uiResult ctx.results.get(analyze-ui)!; return { components: uiResult.output }; }, execute: async (input) { return { suggestions: [增加间距, 统一圆角] }; }, }, ], }; return engine.execute(workflow); } export { AIWorkflowEngine }; export type { StepDefinition, WorkflowDefinition, WorkflowState, WorkflowContext, StepResult };四、工作流编排的边界条件与失效模式4.1 步骤间类型安全当步骤 A 的输出类型变更时步骤 B 的输入提取逻辑不会自动感知到。纯 TypeScript 的类型系统无法在运行时保证这种跨步骤的类型一致性。两个补救措施使用 Zod 或 io-ts 在每个步骤的输入提取函数中做运行时校验将不符合预期的上游输出转化为明确的错误消息。为工作流定义编写集成测试验证完整链路的数据流转——从第一个步骤的输出断言到最后一个步骤的输入断言。4.2 循环依赖检测在 DAG 的有效性校验中循环依赖的检测是启动前必须执行的安全检查。如果步骤 A 依赖步骤 B步骤 B 又依赖步骤 A拓扑排序将无法完成调度器会陷入死锁。实现一个基于 DFS 的环检测算法在工作流注册阶段就拒绝包含环的 DAG 定义。4.3 部分成功场景的处理工作流中如果前 3 个步骤成功、第 4 个步骤失败已产生的结果和消耗的 Token 是否应该被丢弃这取决于具体业务语义对于生成式任务如生成长文中间结果仍有价值可以标记为部分完成供用户查看对于事务性任务如订单处理需要全有或全无的回滚语义。五、总结AI 工作流编排的核心在于将多步骤 LLM 调用的依赖关系建模为 DAG通过拓扑排序实现自动调度并在步骤层面提供重试、降级和超时控制。对于独立产品不需要引入 LangChain 或 Temporal 等重型框架——一个自实现的 DAG 调度器配合 Redis 状态存储足以支撑数十个步骤的 AI 任务编排。落地路线从最关键的三个能力开始步骤的依赖声明与拓扑排序第一优先级、中间状态的持久化与断点续传降低重试成本、以及步骤级的降级策略保障用户体验。并发调度和全局超时可以在后续版本中引入作为性能优化而非功能阻塞项。