Agent 工作流重放与断点续传失败任务不必从头跑续篇场景痛点Agent执行一个10步工作流。第8步调用外部API超时失败。Agent从第1步重新执行。前7步的结果全部丢弃。第1步查询数据库耗时30秒。第3步调用AI模型生成报告耗时2分钟。第5步执行数据清洗耗时1分钟。前7步合计约4分钟。第8步超时后重跑4分钟的前置步骤浪费时间和token。更糟的情况第7步是发送邮件通知。重跑时又发了一封重复邮件。第2步是删除临时文件。重跑时文件已经不存在了——操作不可重复执行。核心矛盾步骤间的数据依赖与执行副作用。重跑不只是时间浪费还可能产生重复副作用发邮件、写数据库或失败副作用删除已删除的文件。底层机制与原理剖析断点续传的本质是执行状态的持久化与恢复。每个步骤完成后其输出数据和执行状态被保存。失败时从最后一个成功步骤继续而非从起点重新开始。关键机制步骤状态机。每个步骤有明确的状态pending → running → completed | failed | skipped。步骤完成后状态和输出被持久化到CheckpointStore。恢复时从completed状态的最晚步骤开始向后执行。副作用分类。步骤分为两类可重放纯计算无外部副作用和不可重放有外部副作用如发邮件、写数据库。可重放步骤可以安全重新执行。不可重放步骤必须用幂等性保证——即使重跑也不会产生重复效果。Checkpoint原子性。步骤状态持久化必须是原子操作。步骤执行了一半就崩溃——输出数据可能不完整。需要全部写完才标记completed的原子保证。步骤间数据依赖图。步骤4依赖步骤3的输出。恢复时必须确保步骤3的输出仍在CheckpointStore中——不是简单地从步骤4重新执行而是先验证前置依赖的完整性。生产级代码实现WorkflowCheckpointManager断点续传核心// workflow/checkpoint-manager.ts import { Redis } from ioredis; import { createHash } from crypto; type StepStatus pending | running | completed | failed | skipped; interface StepCheckpoint { stepId: string; stepName: string; status: StepStatus; output: any; // 步骤输出数据 startedAt: string; completedAt: string | null; retryCount: number; sideEffectType: replayable | idempotent | non_replayable; // 副作用分类 outputHash: string; // 输出数据的hash用于完整性校验 } interface WorkflowCheckpoint { workflowId: string; taskId: string; steps: StepCheckpoint[]; currentStepIndex: number; createdAt: string; updatedAt: string; globalRetryCount: number; maxGlobalRetries: number; } class WorkflowCheckpointManager { private redis: Redis; private readonly CHECKPOINT_TTL 7 * 24 * 3600; // checkpoint保留7天 constructor(redisUrl: string) { this.redis new Redis(redisUrl); } // 创建工作流checkpoint async createCheckpoint( workflowId: string, taskId: string, stepDefinitions: StepDefinition[] ): PromiseWorkflowCheckpoint { const checkpoint: WorkflowCheckpoint { workflowId, taskId, steps: stepDefinitions.map(def ({ stepId: def.id, stepName: def.name, status: pending, output: null, startedAt: , completedAt: null, retryCount: 0, sideEffectType: def.sideEffectType, outputHash: })), currentStepIndex: 0, createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), globalRetryCount: 0, maxGlobalRetries: 3 }; const key checkpoint:${workflowId}:${taskId}; await this.redis.setex(key, this.CHECKPOINT_TTL, JSON.stringify(checkpoint)); return checkpoint; } // 标记步骤开始执行 async markStepRunning(workflowId: string, taskId: string, stepIndex: number): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); checkpoint.steps[stepIndex].status running; checkpoint.steps[stepIndex].startedAt new Date().toISOString(); checkpoint.currentStepIndex stepIndex; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 标记步骤完成原子操作输出hash状态一起写 // 为什么要求原子性步骤输出写了一半就崩溃后续恢复读到不完整数据会引发错误 async markStepCompleted( workflowId: string, taskId: string, stepIndex: number, output: any ): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); // 计算输出hash用于完整性校验 const outputHash createHash(sha256) .update(JSON.stringify(output)) .digest(hex); checkpoint.steps[stepIndex].status completed; checkpoint.steps[stepIndex].output output; checkpoint.steps[stepIndex].completedAt new Date().toISOString(); checkpoint.steps[stepIndex].outputHash outputHash; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 标记步骤失败 async markStepFailed( workflowId: string, taskId: string, stepIndex: number, error: string ): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); checkpoint.steps[stepIndex].status failed; checkpoint.steps[stepIndex].retryCount; checkpoint.globalRetryCount; checkpoint.updatedAt new Date().toISOString(); // 将错误信息存入步骤输出用于调试 checkpoint.steps[stepIndex].output { error }; await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 从断点恢复找到最后一个completed步骤验证前置依赖完整性 async resumeFromCheckpoint( workflowId: string, taskId: string ): PromiseResumeResult { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); // 超过全局重试上限不再恢复 if (checkpoint.globalRetryCount checkpoint.maxGlobalRetries) { return { action: abort, reason: 全局重试次数已达上限${checkpoint.maxGlobalRetries}, checkpoint }; } // 找到恢复起点最后一个completed步骤的下一个步骤 let resumeFromIndex 0; for (let i checkpoint.steps.length - 1; i 0; i--) { if (checkpoint.steps[i].status completed) { // 验证输出完整性hash校验 const currentHash createHash(sha256) .update(JSON.stringify(checkpoint.steps[i].output)) .digest(hex); if (currentHash ! checkpoint.steps[i].outputHash) { // 输出数据被篡改或不完整——此步骤不可信从更早步骤恢复 // 为什么不直接报错退出数据不完整可能是因为Redis写入未完成 // 回退到更早的完整步骤是安全的恢复策略 checkpoint.steps[i].status failed; checkpoint.steps[i].output null; continue; } resumeFromIndex i 1; break; } } // 检查恢复起点的副作用类型 const resumeStep checkpoint.steps[resumeFromIndex]; if (resumeStep?.sideEffectType non_replayable) { // 不可重放步骤不能安全重试——转为人工介入模式 // 为什么不允许重试non_replayable步骤发邮件重试会重复发送 // 写数据库重试可能违反业务约束 return { action: human_intervention, reason: 步骤${resumeStep.stepName}为non_replayable类型不能安全重试, failedStepIndex: resumeFromIndex, checkpoint }; } // 清理后续步骤的状态——它们可能在上次执行中处于中间状态 for (let i resumeFromIndex; i checkpoint.steps.length; i) { if (checkpoint.steps[i].status ! pending) { checkpoint.steps[i].status pending; checkpoint.steps[i].output null; checkpoint.steps[i].outputHash ; } } checkpoint.currentStepIndex resumeFromIndex; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); // 构建恢复上下文提供前置步骤的输出数据 const context: Recordstring, any {}; for (let i 0; i resumeFromIndex; i) { if (checkpoint.steps[i].status completed) { context[checkpoint.steps[i].stepId] checkpoint.steps[i].output; } } return { action: resume, resumeFromIndex, context, // 前置步骤输出数据 checkpoint }; } // 获取指定步骤的缓存输出跳过该步骤的重新执行 async getCachedOutput( workflowId: string, taskId: string, stepIndex: number ): Promiseany { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) return null; const step checkpoint.steps[stepIndex]; if (step.status ! completed) return null; // hash校验确保数据完整性 const currentHash createHash(sha256) .update(JSON.stringify(step.output)) .digest(hex); if (currentHash ! step.outputHash) { return null; // 数据不完整不能使用缓存 } return step.output; } private async loadCheckpoint( workflowId: string, taskId: string ): PromiseWorkflowCheckpoint | null { const key checkpoint:${workflowId}:${taskId}; const data await this.redis.get(key); return data ? JSON.parse(data) : null; } private async saveCheckpoint( workflowId: string, taskId: string, checkpoint: WorkflowCheckpoint ): Promisevoid { const key checkpoint:${workflowId}:${taskId}; // 使用Redis的SETEXPIRE原子操作确保TTL不丢失 await this.redis.setex(key, this.CHECKPOINT_TTL, JSON.stringify(checkpoint)); } } interface ResumeResult { action: resume | abort | human_intervention; resumeFromIndex?: number; context?: Recordstring, any; reason?: string; failedStepIndex?: number; checkpoint: WorkflowCheckpoint; } interface StepDefinition { id: string; name: string; sideEffectType: replayable | idempotent | non_replayable; }ResumableWorkflowExecutor可恢复的工作流执行器// workflow/resumable-executor.ts import { WorkflowCheckpointManager, StepDefinition } from ./checkpoint-manager; class ResumableWorkflowExecutor { private stepHandlers: Mapstring, StepHandler new Map(); constructor(private checkpointManager: WorkflowCheckpointManager) {} // 注册步骤处理器 registerHandler(stepId: string, handler: StepHandler): void { this.stepHandlers.set(stepId, handler); } // 执行工作流新建或恢复 async execute( workflowId: string, taskId: string, steps: StepDefinition[], resumeFrom?: ResumeResult ): PromiseWorkflowResult { // 如果没有恢复上下文创建新checkpoint if (!resumeFrom) { await this.checkpointManager.createCheckpoint(workflowId, taskId, steps); } const startIndex resumeFrom?.resumeFromIndex ?? 0; const context resumeFrom?.context ?? {}; for (let i startIndex; i steps.length; i) { const step steps[i]; // 标记步骤running await this.checkpointManager.markStepRunning(workflowId, taskId, i); const handler this.stepHandlers.get(step.id); if (!handler) { await this.checkpointManager.markStepFailed(workflowId, taskId, i, 未注册步骤处理器: ${step.id}); break; } try { // 执行步骤传入前置步骤的输出作为上下文 // 为什么传入context而非让步骤自己去checkpoint取减少步骤对checkpoint系统的依赖 // 步骤只需要关心输入数据不需要关心持久化机制 const output await handler.execute(context, { timeoutMs: handler.timeoutMs ?? 30000, retryOnTimeout: step.sideEffectType ! non_replayable }); // 标记完成并缓存输出 await this.checkpointManager.markStepCompleted(workflowId, taskId, i, output); // 将输出加入上下文供后续步骤使用 context[step.id] output; } catch (error: any) { await this.checkpointManager.markStepFailed(workflowId, taskId, i, error.message); // 尝试从断点恢复 const resumeResult await this.checkpointManager.resumeFromCheckpoint( workflowId, taskId ); if (resumeResult.action resume) { // 自动重试从断点恢复后重新执行 // 为什么递归调用而非循环继续恢复可能需要重新初始化上下文 // 递归调用确保状态干净 return this.execute(workflowId, taskId, steps, resumeResult); } else if (resumeResult.action human_intervention) { // 不可自动恢复返回人工介入请求 return { status: needs_human_intervention, failedStep: resumeResult.failedStepIndex!, reason: resumeResult.reason, checkpoint: resumeResult.checkpoint }; } else { // 全局重试上限任务失败 return { status: failed, reason: resumeResult.reason, checkpoint: resumeResult.checkpoint }; } } } return { status: completed, context }; } } interface StepHandler { timeoutMs?: number; execute(context: Recordstring, any, options: { timeoutMs: number; retryOnTimeout: boolean }): Promiseany; } interface WorkflowResult { status: completed | failed | needs_human_intervention; context?: Recordstring, any; reason?: string; failedStep?: number; checkpoint?: WorkflowCheckpoint; }幂等性保障装饰器// workflow/idempotent-decorator.ts import { Redis } from ioredis; // 为non_replayable步骤添加幂等性保障 // 为什么用装饰器而非侵入式修改步骤处理器不应该知道幂等性机制的存在 // 保持业务逻辑与工程机制的分离 class IdempotentDecorator implements StepHandler { private redis: Redis; constructor( private innerHandler: StepHandler, redisUrl: string ) { this.redis new Redis(redisUrl); } async execute( context: Recordstring, any, options: { timeoutMs: number; retryOnTimeout: boolean } ): Promiseany { // 生成幂等性key基于步骤输入的hash // 为什么基于输入hash而非步骤ID同一步骤可能用不同参数调用 // 幂等性保证的是相同输入只执行一次 const idempotencyKey this.generateIdempotencyKey(context); // 检查是否已执行 const existingResult await this.redis.get(idempotent:${idempotencyKey}); if (existingResult) { // 已经执行过直接返回缓存结果 // 为什么不重新执行发邮件、写数据库这类操作重复执行会产生业务副作用 return JSON.parse(existingResult); } // 执行步骤 const result await this.innerHandler.execute(context, options); // 缓存结果保留24小时 // 为什么24小时而非永久幂等性保证的是同一次工作流执行中不重复 // 跨执行不需要保证下次工作流是新的业务场景 await this.redis.setex(idempotent:${idempotencyKey}, 86400, JSON.stringify(result)); return result; } private generateIdempotencyKey(context: Recordstring, any): string { // 只使用步骤直接相关的输入参数生成key而非全部上下文 // 为什么不全用context上下文包含所有前置步骤输出hash变化频繁 // 导致幂等性key不稳定缓存命中率低 const relevantInputs Object.entries(context) .filter(([key]) key.startsWith(step_) || key input) .sort(([a], [b]) a.localeCompare(b)); return createHash(sha256) .update(JSON.stringify(relevantInputs)) .digest(hex); } }边界分析与架构权衡Checkpoint存储选型Redis读写快1ms但数据有TTL7天后checkpoint丢失。适合短期任务。PostgreSQL持久化可靠但读写慢5~10ms。适合长期任务跨天执行的工作流。混合策略步骤执行期间用Redis暂存快工作流完成后迁移到PostgreSQL归档持久。归档数据用于事后审计和调试。大输出数据的Checkpoint成本步骤3AI生成报告的输出可能是10KB的文本。步骤6图片处理的输出可能是5MB的二进制数据。Checkpoint存储5MB数据在Redis中代价高。解决方案大输出存对象存储S3/OSSCheckpoint只存引用URL。// 大输出存储策略 async markStepCompleted(stepIndex: number, output: any): Promisevoid { const outputSize Buffer.byteLength(JSON.stringify(output)); if (outputSize 100_000) { // 超过100KB // 存入对象存储Checkpoint只保留URL const url await this.objectStore.upload( checkpoint-output/${this.workflowId}/${stepIndex}, JSON.stringify(output) ); // hash仍然基于完整数据计算完整性校验不受影响 const outputHash createHash(sha256) .update(JSON.stringify(output)) .digest(hex); checkpoint.steps[stepIndex].output { __ref: url, __hash: outputHash }; } else { checkpoint.steps[stepIndex].output output; } }并行步骤的Checkpoint问题某些工作流有并行步骤步骤3a和3b同时执行。两个并行步骤同时写Checkpoint——并发冲突。Redis的SET操作是原子的但两个步骤同时更新同一个checkpoint JSON会互相覆盖。解决方案每个步骤的checkpoint独立存储key checkpoint:${workflowId}:${taskId}:step:${stepIndex}而非所有步骤共享一个JSON。恢复时聚合所有步骤的状态。代价是恢复操作变慢需要读取N个key但消除了并发冲突。断点续传与版本兼容工作流定义在v1版本有10个步骤。执行到第5步时失败。开发者在v2版本修改了步骤定义合并了步骤3和4。此时v1的checkpoint无法映射到v2的步骤定义。解决方案checkpoint中保存步骤定义的版本号。恢复时如果版本不匹配触发完整重跑而非断点续传。保守但安全——映射错误的后果比重新执行更严重。超时与无限挂起的步骤步骤4调用外部APIAPI没有响应也没有超时。步骤永远停在running状态。checkpoint永远等不到这个步骤的完成或失败标记。解决全局超时守护线程。// 超时守护 async watchdog(workflowId: string, taskId: string, maxStepDurationMs: number): void { const checkpoint await this.loadCheckpoint(workflowId, taskId); for (const step of checkpoint.steps) { if (step.status running) { const elapsed Date.now() - new Date(step.startedAt).getTime(); if (elapsed maxStepDurationMs) { // 超时步骤标记为failed await this.markStepFailed(workflowId, taskId, checkpoint.steps.indexOf(step), 步骤超时${elapsed}ms ${maxStepDurationMs}ms ); } } } }总结断点续传把工作流从全量重跑变成增量恢复。核心设计每个步骤完成后状态和输出持久化到CheckpointStore。恢复时从最后一个completed步骤向后继续。步骤分三类副作用replayable可重放纯计算、idempotent幂等有副作用但重复安全、non_replayable不可重放重复执行有业务风险。non_replayable步骤必须用幂等性装饰器保障。Checkpoint原子性保证输出数据和hash一起写入完整性校验防止读到半写数据。步骤间数据通过context传递步骤处理器不需要关心Checkpoint机制——业务逻辑与工程机制分离。大输出数据存对象存储Checkpoint只存引用URL。Redis存10KB以内的小数据。并行步骤独立存储Checkpoint避免并发冲突。全局重试上限防止无限循环。超时守护防止步骤永远挂在running状态。断点续传不是锦上添花——是生产级Agent工作流的基本要求。没有断点续传的工作流每次失败都从头跑时间和token的浪费随步骤数线性增长。10步工作流节省4分钟50步工作流节省20分钟。这是工程效率的底线要求。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0730 资料来源索引并在发布前将具体来源贴到对应断言之后。
Agent 工作流重放与断点续传：失败任务不必从头跑（续篇）
Agent 工作流重放与断点续传失败任务不必从头跑续篇场景痛点Agent执行一个10步工作流。第8步调用外部API超时失败。Agent从第1步重新执行。前7步的结果全部丢弃。第1步查询数据库耗时30秒。第3步调用AI模型生成报告耗时2分钟。第5步执行数据清洗耗时1分钟。前7步合计约4分钟。第8步超时后重跑4分钟的前置步骤浪费时间和token。更糟的情况第7步是发送邮件通知。重跑时又发了一封重复邮件。第2步是删除临时文件。重跑时文件已经不存在了——操作不可重复执行。核心矛盾步骤间的数据依赖与执行副作用。重跑不只是时间浪费还可能产生重复副作用发邮件、写数据库或失败副作用删除已删除的文件。底层机制与原理剖析断点续传的本质是执行状态的持久化与恢复。每个步骤完成后其输出数据和执行状态被保存。失败时从最后一个成功步骤继续而非从起点重新开始。关键机制步骤状态机。每个步骤有明确的状态pending → running → completed | failed | skipped。步骤完成后状态和输出被持久化到CheckpointStore。恢复时从completed状态的最晚步骤开始向后执行。副作用分类。步骤分为两类可重放纯计算无外部副作用和不可重放有外部副作用如发邮件、写数据库。可重放步骤可以安全重新执行。不可重放步骤必须用幂等性保证——即使重跑也不会产生重复效果。Checkpoint原子性。步骤状态持久化必须是原子操作。步骤执行了一半就崩溃——输出数据可能不完整。需要全部写完才标记completed的原子保证。步骤间数据依赖图。步骤4依赖步骤3的输出。恢复时必须确保步骤3的输出仍在CheckpointStore中——不是简单地从步骤4重新执行而是先验证前置依赖的完整性。生产级代码实现WorkflowCheckpointManager断点续传核心// workflow/checkpoint-manager.ts import { Redis } from ioredis; import { createHash } from crypto; type StepStatus pending | running | completed | failed | skipped; interface StepCheckpoint { stepId: string; stepName: string; status: StepStatus; output: any; // 步骤输出数据 startedAt: string; completedAt: string | null; retryCount: number; sideEffectType: replayable | idempotent | non_replayable; // 副作用分类 outputHash: string; // 输出数据的hash用于完整性校验 } interface WorkflowCheckpoint { workflowId: string; taskId: string; steps: StepCheckpoint[]; currentStepIndex: number; createdAt: string; updatedAt: string; globalRetryCount: number; maxGlobalRetries: number; } class WorkflowCheckpointManager { private redis: Redis; private readonly CHECKPOINT_TTL 7 * 24 * 3600; // checkpoint保留7天 constructor(redisUrl: string) { this.redis new Redis(redisUrl); } // 创建工作流checkpoint async createCheckpoint( workflowId: string, taskId: string, stepDefinitions: StepDefinition[] ): PromiseWorkflowCheckpoint { const checkpoint: WorkflowCheckpoint { workflowId, taskId, steps: stepDefinitions.map(def ({ stepId: def.id, stepName: def.name, status: pending, output: null, startedAt: , completedAt: null, retryCount: 0, sideEffectType: def.sideEffectType, outputHash: })), currentStepIndex: 0, createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), globalRetryCount: 0, maxGlobalRetries: 3 }; const key checkpoint:${workflowId}:${taskId}; await this.redis.setex(key, this.CHECKPOINT_TTL, JSON.stringify(checkpoint)); return checkpoint; } // 标记步骤开始执行 async markStepRunning(workflowId: string, taskId: string, stepIndex: number): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); checkpoint.steps[stepIndex].status running; checkpoint.steps[stepIndex].startedAt new Date().toISOString(); checkpoint.currentStepIndex stepIndex; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 标记步骤完成原子操作输出hash状态一起写 // 为什么要求原子性步骤输出写了一半就崩溃后续恢复读到不完整数据会引发错误 async markStepCompleted( workflowId: string, taskId: string, stepIndex: number, output: any ): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); // 计算输出hash用于完整性校验 const outputHash createHash(sha256) .update(JSON.stringify(output)) .digest(hex); checkpoint.steps[stepIndex].status completed; checkpoint.steps[stepIndex].output output; checkpoint.steps[stepIndex].completedAt new Date().toISOString(); checkpoint.steps[stepIndex].outputHash outputHash; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 标记步骤失败 async markStepFailed( workflowId: string, taskId: string, stepIndex: number, error: string ): Promisevoid { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); checkpoint.steps[stepIndex].status failed; checkpoint.steps[stepIndex].retryCount; checkpoint.globalRetryCount; checkpoint.updatedAt new Date().toISOString(); // 将错误信息存入步骤输出用于调试 checkpoint.steps[stepIndex].output { error }; await this.saveCheckpoint(workflowId, taskId, checkpoint); } // 从断点恢复找到最后一个completed步骤验证前置依赖完整性 async resumeFromCheckpoint( workflowId: string, taskId: string ): PromiseResumeResult { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) throw new Error(Checkpoint不存在); // 超过全局重试上限不再恢复 if (checkpoint.globalRetryCount checkpoint.maxGlobalRetries) { return { action: abort, reason: 全局重试次数已达上限${checkpoint.maxGlobalRetries}, checkpoint }; } // 找到恢复起点最后一个completed步骤的下一个步骤 let resumeFromIndex 0; for (let i checkpoint.steps.length - 1; i 0; i--) { if (checkpoint.steps[i].status completed) { // 验证输出完整性hash校验 const currentHash createHash(sha256) .update(JSON.stringify(checkpoint.steps[i].output)) .digest(hex); if (currentHash ! checkpoint.steps[i].outputHash) { // 输出数据被篡改或不完整——此步骤不可信从更早步骤恢复 // 为什么不直接报错退出数据不完整可能是因为Redis写入未完成 // 回退到更早的完整步骤是安全的恢复策略 checkpoint.steps[i].status failed; checkpoint.steps[i].output null; continue; } resumeFromIndex i 1; break; } } // 检查恢复起点的副作用类型 const resumeStep checkpoint.steps[resumeFromIndex]; if (resumeStep?.sideEffectType non_replayable) { // 不可重放步骤不能安全重试——转为人工介入模式 // 为什么不允许重试non_replayable步骤发邮件重试会重复发送 // 写数据库重试可能违反业务约束 return { action: human_intervention, reason: 步骤${resumeStep.stepName}为non_replayable类型不能安全重试, failedStepIndex: resumeFromIndex, checkpoint }; } // 清理后续步骤的状态——它们可能在上次执行中处于中间状态 for (let i resumeFromIndex; i checkpoint.steps.length; i) { if (checkpoint.steps[i].status ! pending) { checkpoint.steps[i].status pending; checkpoint.steps[i].output null; checkpoint.steps[i].outputHash ; } } checkpoint.currentStepIndex resumeFromIndex; checkpoint.updatedAt new Date().toISOString(); await this.saveCheckpoint(workflowId, taskId, checkpoint); // 构建恢复上下文提供前置步骤的输出数据 const context: Recordstring, any {}; for (let i 0; i resumeFromIndex; i) { if (checkpoint.steps[i].status completed) { context[checkpoint.steps[i].stepId] checkpoint.steps[i].output; } } return { action: resume, resumeFromIndex, context, // 前置步骤输出数据 checkpoint }; } // 获取指定步骤的缓存输出跳过该步骤的重新执行 async getCachedOutput( workflowId: string, taskId: string, stepIndex: number ): Promiseany { const checkpoint await this.loadCheckpoint(workflowId, taskId); if (!checkpoint) return null; const step checkpoint.steps[stepIndex]; if (step.status ! completed) return null; // hash校验确保数据完整性 const currentHash createHash(sha256) .update(JSON.stringify(step.output)) .digest(hex); if (currentHash ! step.outputHash) { return null; // 数据不完整不能使用缓存 } return step.output; } private async loadCheckpoint( workflowId: string, taskId: string ): PromiseWorkflowCheckpoint | null { const key checkpoint:${workflowId}:${taskId}; const data await this.redis.get(key); return data ? JSON.parse(data) : null; } private async saveCheckpoint( workflowId: string, taskId: string, checkpoint: WorkflowCheckpoint ): Promisevoid { const key checkpoint:${workflowId}:${taskId}; // 使用Redis的SETEXPIRE原子操作确保TTL不丢失 await this.redis.setex(key, this.CHECKPOINT_TTL, JSON.stringify(checkpoint)); } } interface ResumeResult { action: resume | abort | human_intervention; resumeFromIndex?: number; context?: Recordstring, any; reason?: string; failedStepIndex?: number; checkpoint: WorkflowCheckpoint; } interface StepDefinition { id: string; name: string; sideEffectType: replayable | idempotent | non_replayable; }ResumableWorkflowExecutor可恢复的工作流执行器// workflow/resumable-executor.ts import { WorkflowCheckpointManager, StepDefinition } from ./checkpoint-manager; class ResumableWorkflowExecutor { private stepHandlers: Mapstring, StepHandler new Map(); constructor(private checkpointManager: WorkflowCheckpointManager) {} // 注册步骤处理器 registerHandler(stepId: string, handler: StepHandler): void { this.stepHandlers.set(stepId, handler); } // 执行工作流新建或恢复 async execute( workflowId: string, taskId: string, steps: StepDefinition[], resumeFrom?: ResumeResult ): PromiseWorkflowResult { // 如果没有恢复上下文创建新checkpoint if (!resumeFrom) { await this.checkpointManager.createCheckpoint(workflowId, taskId, steps); } const startIndex resumeFrom?.resumeFromIndex ?? 0; const context resumeFrom?.context ?? {}; for (let i startIndex; i steps.length; i) { const step steps[i]; // 标记步骤running await this.checkpointManager.markStepRunning(workflowId, taskId, i); const handler this.stepHandlers.get(step.id); if (!handler) { await this.checkpointManager.markStepFailed(workflowId, taskId, i, 未注册步骤处理器: ${step.id}); break; } try { // 执行步骤传入前置步骤的输出作为上下文 // 为什么传入context而非让步骤自己去checkpoint取减少步骤对checkpoint系统的依赖 // 步骤只需要关心输入数据不需要关心持久化机制 const output await handler.execute(context, { timeoutMs: handler.timeoutMs ?? 30000, retryOnTimeout: step.sideEffectType ! non_replayable }); // 标记完成并缓存输出 await this.checkpointManager.markStepCompleted(workflowId, taskId, i, output); // 将输出加入上下文供后续步骤使用 context[step.id] output; } catch (error: any) { await this.checkpointManager.markStepFailed(workflowId, taskId, i, error.message); // 尝试从断点恢复 const resumeResult await this.checkpointManager.resumeFromCheckpoint( workflowId, taskId ); if (resumeResult.action resume) { // 自动重试从断点恢复后重新执行 // 为什么递归调用而非循环继续恢复可能需要重新初始化上下文 // 递归调用确保状态干净 return this.execute(workflowId, taskId, steps, resumeResult); } else if (resumeResult.action human_intervention) { // 不可自动恢复返回人工介入请求 return { status: needs_human_intervention, failedStep: resumeResult.failedStepIndex!, reason: resumeResult.reason, checkpoint: resumeResult.checkpoint }; } else { // 全局重试上限任务失败 return { status: failed, reason: resumeResult.reason, checkpoint: resumeResult.checkpoint }; } } } return { status: completed, context }; } } interface StepHandler { timeoutMs?: number; execute(context: Recordstring, any, options: { timeoutMs: number; retryOnTimeout: boolean }): Promiseany; } interface WorkflowResult { status: completed | failed | needs_human_intervention; context?: Recordstring, any; reason?: string; failedStep?: number; checkpoint?: WorkflowCheckpoint; }幂等性保障装饰器// workflow/idempotent-decorator.ts import { Redis } from ioredis; // 为non_replayable步骤添加幂等性保障 // 为什么用装饰器而非侵入式修改步骤处理器不应该知道幂等性机制的存在 // 保持业务逻辑与工程机制的分离 class IdempotentDecorator implements StepHandler { private redis: Redis; constructor( private innerHandler: StepHandler, redisUrl: string ) { this.redis new Redis(redisUrl); } async execute( context: Recordstring, any, options: { timeoutMs: number; retryOnTimeout: boolean } ): Promiseany { // 生成幂等性key基于步骤输入的hash // 为什么基于输入hash而非步骤ID同一步骤可能用不同参数调用 // 幂等性保证的是相同输入只执行一次 const idempotencyKey this.generateIdempotencyKey(context); // 检查是否已执行 const existingResult await this.redis.get(idempotent:${idempotencyKey}); if (existingResult) { // 已经执行过直接返回缓存结果 // 为什么不重新执行发邮件、写数据库这类操作重复执行会产生业务副作用 return JSON.parse(existingResult); } // 执行步骤 const result await this.innerHandler.execute(context, options); // 缓存结果保留24小时 // 为什么24小时而非永久幂等性保证的是同一次工作流执行中不重复 // 跨执行不需要保证下次工作流是新的业务场景 await this.redis.setex(idempotent:${idempotencyKey}, 86400, JSON.stringify(result)); return result; } private generateIdempotencyKey(context: Recordstring, any): string { // 只使用步骤直接相关的输入参数生成key而非全部上下文 // 为什么不全用context上下文包含所有前置步骤输出hash变化频繁 // 导致幂等性key不稳定缓存命中率低 const relevantInputs Object.entries(context) .filter(([key]) key.startsWith(step_) || key input) .sort(([a], [b]) a.localeCompare(b)); return createHash(sha256) .update(JSON.stringify(relevantInputs)) .digest(hex); } }边界分析与架构权衡Checkpoint存储选型Redis读写快1ms但数据有TTL7天后checkpoint丢失。适合短期任务。PostgreSQL持久化可靠但读写慢5~10ms。适合长期任务跨天执行的工作流。混合策略步骤执行期间用Redis暂存快工作流完成后迁移到PostgreSQL归档持久。归档数据用于事后审计和调试。大输出数据的Checkpoint成本步骤3AI生成报告的输出可能是10KB的文本。步骤6图片处理的输出可能是5MB的二进制数据。Checkpoint存储5MB数据在Redis中代价高。解决方案大输出存对象存储S3/OSSCheckpoint只存引用URL。// 大输出存储策略 async markStepCompleted(stepIndex: number, output: any): Promisevoid { const outputSize Buffer.byteLength(JSON.stringify(output)); if (outputSize 100_000) { // 超过100KB // 存入对象存储Checkpoint只保留URL const url await this.objectStore.upload( checkpoint-output/${this.workflowId}/${stepIndex}, JSON.stringify(output) ); // hash仍然基于完整数据计算完整性校验不受影响 const outputHash createHash(sha256) .update(JSON.stringify(output)) .digest(hex); checkpoint.steps[stepIndex].output { __ref: url, __hash: outputHash }; } else { checkpoint.steps[stepIndex].output output; } }并行步骤的Checkpoint问题某些工作流有并行步骤步骤3a和3b同时执行。两个并行步骤同时写Checkpoint——并发冲突。Redis的SET操作是原子的但两个步骤同时更新同一个checkpoint JSON会互相覆盖。解决方案每个步骤的checkpoint独立存储key checkpoint:${workflowId}:${taskId}:step:${stepIndex}而非所有步骤共享一个JSON。恢复时聚合所有步骤的状态。代价是恢复操作变慢需要读取N个key但消除了并发冲突。断点续传与版本兼容工作流定义在v1版本有10个步骤。执行到第5步时失败。开发者在v2版本修改了步骤定义合并了步骤3和4。此时v1的checkpoint无法映射到v2的步骤定义。解决方案checkpoint中保存步骤定义的版本号。恢复时如果版本不匹配触发完整重跑而非断点续传。保守但安全——映射错误的后果比重新执行更严重。超时与无限挂起的步骤步骤4调用外部APIAPI没有响应也没有超时。步骤永远停在running状态。checkpoint永远等不到这个步骤的完成或失败标记。解决全局超时守护线程。// 超时守护 async watchdog(workflowId: string, taskId: string, maxStepDurationMs: number): void { const checkpoint await this.loadCheckpoint(workflowId, taskId); for (const step of checkpoint.steps) { if (step.status running) { const elapsed Date.now() - new Date(step.startedAt).getTime(); if (elapsed maxStepDurationMs) { // 超时步骤标记为failed await this.markStepFailed(workflowId, taskId, checkpoint.steps.indexOf(step), 步骤超时${elapsed}ms ${maxStepDurationMs}ms ); } } } }总结断点续传把工作流从全量重跑变成增量恢复。核心设计每个步骤完成后状态和输出持久化到CheckpointStore。恢复时从最后一个completed步骤向后继续。步骤分三类副作用replayable可重放纯计算、idempotent幂等有副作用但重复安全、non_replayable不可重放重复执行有业务风险。non_replayable步骤必须用幂等性装饰器保障。Checkpoint原子性保证输出数据和hash一起写入完整性校验防止读到半写数据。步骤间数据通过context传递步骤处理器不需要关心Checkpoint机制——业务逻辑与工程机制分离。大输出数据存对象存储Checkpoint只存引用URL。Redis存10KB以内的小数据。并行步骤独立存储Checkpoint避免并发冲突。全局重试上限防止无限循环。超时守护防止步骤永远挂在running状态。断点续传不是锦上添花——是生产级Agent工作流的基本要求。没有断点续传的工作流每次失败都从头跑时间和token的浪费随步骤数线性增长。10步工作流节省4分钟50步工作流节省20分钟。这是工程效率的底线要求。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0730 资料来源索引并在发布前将具体来源贴到对应断言之后。
