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 资料来源索引，并在发布前将具体来源贴到对应断言之后。

# 云原生 # kubernetes
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

深入理解 AI Agent · AGENT #02：Agent 的四大核心机制，拆开来看每个都在干什么

avatar AI Agent技术社区

MCP协议开发实战：从零搭建AI Agent工具链

本文介绍了MCP协议在AI工具链中的应用与实践。MCP协议旨在解决异构工具集成难题，为AI模型提供标准化、安全的工具调用接口。文章详细解析了MCP的核心架构，包括Server、Client、Transport等组件，以及通信流程与安全模型。通过实战示例，指导读者开发计算器和文件系统工具Server，并集成到Claude等AI客户端中。进阶部分展示了如何构建复杂工具链，包括API集成、数据库查询和多

avatar AI Agent技术社区
cover

AI Tool 设计：粒度、参数与错误恢复怎么做

avatar AI Agent技术社区