Agent 工作流重放与断点续传：失败任务不必从头跑（续篇）

场景痛点

Agent执行一个10步工作流。第8步调用外部API超时失败。Agent从第1步重新执行。前7步的结果全部丢弃。

第1步查询数据库耗时30秒。第3步调用AI模型生成报告耗时2分钟。第5步执行数据清洗耗时1分钟。前7步合计约4分钟。第8步超时后重跑4分钟的前置步骤，浪费时间和token。

更糟的情况：第7步是"发送邮件通知"。重跑时又发了一封重复邮件。第2步是"删除临时文件"。重跑时文件已经不存在了——操作不可重复执行。

核心矛盾：步骤间的数据依赖与执行副作用。重跑不只是时间浪费，还可能产生重复副作用（发邮件、写数据库）或失败副作用（删除已删除的文件）。

底层机制与原理剖析

断点续传的本质是执行状态的持久化与恢复。每个步骤完成后，其输出数据和执行状态被保存。失败时从最后一个成功步骤继续，而非从起点重新开始。

关键机制：

1. 步骤状态机。每个步骤有明确的状态：pending → running → completed | failed | skipped。步骤完成后状态和输出被持久化到CheckpointStore。恢复时从completed状态的最晚步骤开始向后执行。

2. 副作用分类。步骤分为两类：可重放（纯计算，无外部副作用）和不可重放（有外部副作用如发邮件、写数据库）。可重放步骤可以安全重新执行。不可重放步骤必须用幂等性保证——即使重跑也不会产生重复效果。

3. Checkpoint原子性。步骤状态持久化必须是原子操作。步骤执行了一半就崩溃——输出数据可能不完整。需要"全部写完才标记completed"的原子保证。

4. 步骤间数据依赖图。步骤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[] ): Promise<WorkflowCheckpoint> { 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): Promise<void> { 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 ): Promise<void> { 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 ): Promise<void> { 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 ): Promise<ResumeResult> { 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: Record<string, 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 ): Promise<any> { 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 ): Promise<WorkflowCheckpoint | 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 ): Promise<void> { const key = `checkpoint:${workflowId}:${taskId}`; // 使用Redis的SET+EXPIRE原子操作确保TTL不丢失 await this.redis.setex(key, this.CHECKPOINT_TTL, JSON.stringify(checkpoint)); } } interface ResumeResult { action: 'resume' | 'abort' | 'human_intervention'; resumeFromIndex?: number; context?: Record<string, 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: Map<string, 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 ): Promise<WorkflowResult> { // 如果没有恢复上下文，创建新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: Record<string, any>, options: { timeoutMs: number; retryOnTimeout: boolean }): Promise<any>; } interface WorkflowResult { status: 'completed' | 'failed' | 'needs_human_intervention'; context?: Record<string, 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: Record<string, any>, options: { timeoutMs: number; retryOnTimeout: boolean } ): Promise<any> { // 生成幂等性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: Record<string, 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），但数据有TTL，7天后checkpoint丢失。适合短期任务。

PostgreSQL：持久化可靠，但读写慢（5~10ms）。适合长期任务（跨天执行的工作流）。

混合策略：步骤执行期间用Redis暂存（快），工作流完成后迁移到PostgreSQL归档（持久）。归档数据用于事后审计和调试。

大输出数据的Checkpoint成本

步骤3（AI生成报告）的输出可能是10KB的文本。步骤6（图片处理）的输出可能是5MB的二进制数据。Checkpoint存储5MB数据在Redis中代价高。

解决方案：大输出存对象存储（S3/OSS），Checkpoint只存引用URL。

// 大输出存储策略 async markStepCompleted(stepIndex: number, output: any): Promise<void> { 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调用外部API，API没有响应也没有超时。步骤永远停在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` ); } } } }

总结

断点续传把工作流从"全量重跑"变成"增量恢复"。核心设计：

1. 每个步骤完成后状态和输出持久化到CheckpointStore。恢复时从最后一个completed步骤向后继续。
2. 步骤分三类副作用：replayable（可重放，纯计算）、idempotent（幂等，有副作用但重复安全）、non_replayable（不可重放，重复执行有业务风险）。non_replayable步骤必须用幂等性装饰器保障。
3. Checkpoint原子性保证：输出数据和hash一起写入，完整性校验防止读到半写数据。
4. 步骤间数据通过context传递，步骤处理器不需要关心Checkpoint机制——业务逻辑与工程机制分离。
5. 大输出数据存对象存储，Checkpoint只存引用URL。Redis存10KB以内的小数据。
6. 并行步骤独立存储Checkpoint，避免并发冲突。
7. 全局重试上限防止无限循环。超时守护防止步骤永远挂在running状态。

断点续传不是锦上添花——是生产级Agent工作流的基本要求。没有断点续传的工作流，每次失败都从头跑，时间和token的浪费随步骤数线性增长。10步工作流节省4分钟，50步工作流节省20分钟。这是工程效率的底线要求。

资料说明

本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论，不应视为行业事实。可参考 0730 资料来源索引，并在发布前将具体来源贴到对应断言之后。