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

键盘鼓手苏苏

2883人浏览 · 2026-07-08 13:18:04

键盘鼓手苏苏 · 2026-07-08 13:18:04 发布

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

一、一个 tool call 失败，整个工作流白跑

Agent 做复杂任务时，通常拆成多步骤：搜索 → 分析 → 生成 → 审核，每步可能涉及多次 LLM 调用和工具调用。问题在于：如果第三步的工具调用抛了 500，整个工作流全部作废——没有重放能力，只能从头再跑。

这不仅浪费算力（前面走过的步骤白跑了），更糟糕的是非确定性：重新执行时 LLM 可能生成不同的中间结果，导致最终输出与重试前不一致。这对需要可复现性的场景是致命的——比如数据分析 Agent，两次运行得出不同的结论就完全不可信了。

实际损失数据：一个 5 步工作流，每步平均耗时 10 秒，LLM 调用成本约 $0.02/步。如果第 4 步失败，前 3 步的 $0.06 和 30 秒全部浪费。如果一天跑 1000 个工作流、失败率 5%，每天浪费 $3 和 500 秒算力——累积下来不是小数目。更关键的是时间浪费：一个 5 步工作流耗时 50 秒，失败后从头跑又是 50 秒，加上重试可能 100 秒。如果用断点续传，只重跑第 4 步，恢复时间约 10 秒——节省 40 秒。

断点续传（Checkpoint & Resume）的核心价值：在每一步成功的节点保存状态快照，失败时从最近的检查点恢复，而不是从头开始。这把"从头重跑"变成"从断点续跑"，时间和成本节省与失败位置成正比。

sequenceDiagram
    participant WF as 工作流引擎
    participant CK as 检查点存储
    participant LLM as LLM 服务
    participant TOOL as 工具服务
    
    WF->>WF: Step 1: 需求分析
    WF->>LLM: 分析请求
    LLM-->>WF: 分析结果
    WF->>CK: 写入检查点 checkpoint_1
    
    WF->>WF: Step 2: 信息检索
    WF->>TOOL: search(query)
    TOOL-->>WF: 搜索结果
    
    Note over WF,TOOL: Step 3 失败
    WF->>TOOL: fetch_detail(url)
    TOOL--xWF: 500 错误
    
    WF->>CK: 读取最近检查点 checkpoint_1
    CK-->>WF: 恢复状态
    WF->>TOOL: search(query) — 复用缓存
    TOOL-->>WF: 缓存命中, 跳过
    WF->>TOOL: fetch_detail(url) — 重试
    TOOL-->>WF: 成功
    WF->>CK: 写入检查点 checkpoint_2

二、检查点的设计原则

原则一：最小快照

检查点不是把整个内存 dump 到磁盘。只保存导致状态变化的决策结果，不保存中间推理过程。一个典型检查点的内容：已完成的步骤 ID、每步的输入输出、当前 LLM 对话摘要。

为什么只存结果不存过程？两个原因。一是存储成本：一个 LLM 推理的中间链可能包含数千 token，存下来检查点体积膨胀 10 倍。二是恢复效率：恢复时只需要结果就能继续下一步，中间推理链对后续步骤没有直接价值。如果把推理链也存下来，检查点从 1KB 膨胀到 10KB，Redis 写入延迟增加，但恢复后的执行效果完全一样——得不偿失。

最小快照的另一个优势：跨实例恢复。如果原实例挂了，另一个实例可以加载检查点继续执行——因为检查点只包含决策结果而非内存状态，没有实例绑定的数据。

原则二：幂等性保证

恢复时可能重新执行某些步骤。设计上必须保证：同一步骤以相同输入重复执行，结果等价。LLM 调用没有天然幂等性——同一个 prompt 两次调用可能返回不同内容。所以需要对 LLM 输出做确定性缓存。

具体做法：为每个 LLM 调用生成一个缓存 key = hash(prompt + model_id + temperature)。temperature > 0 时缓存无效（因为 LLM 输出不确定），temperature = 0 时可以缓存。但实际工程中 temperature = 0 也不保证完全确定性（受 GPU 并行计算浮点误差影响），所以缓存只是"近似幂等"——对于大多数场景够用了。

真正的幂等性要求：对于关键决策步骤（如"选择哪个数据源"），不应该依赖 LLM 的非确定性输出。应该在检查点中保存决策结果，恢复时直接使用保存的决策——而非重新让 LLM 做决策。这保证两次运行的决策路径完全一致。

原则三：回滚边界

不是所有步骤都值得回滚。副作用步骤（发送邮件、创建工单、写入数据库）一旦执行就不能撤销。这类步骤之前必须写检查点，且恢复时标记为"已完成"跳过重放。

为什么不能重放副作用步骤？因为外部系统的状态已经被改变了——重放意味着再发一封邮件、再创建一个工单。这不仅冗余，还可能造成业务错误（比如用户收到两封相同的邮件）。所以副作用步骤在恢复时必须跳过，而非重放——这是检查点系统中最关键的区分。

一个常见的错误：把所有步骤都标记为 IDEMPOTENT。这会导致副作用步骤被重放——发送两次邮件。正确的做法是在定义工作流时显式标注每个步骤的类型：IDEMPOTENT（可安全重放）、SIDE_EFFECT（不可重放）、DETERMINISTIC（结果确定可缓存）。

三、带检查点的工作流引擎

import json
import hashlib
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional
import redis.asyncio as redis


class StepType(Enum):
    IDEMPOTENT = "idempotent"        # 可安全重放
    SIDE_EFFECT = "side_effect"      # 有副作用，不可重放
    DETERMINISTIC = "deterministic"  # 相同输入保证相同输出


@dataclass
class StepResult:
    step_id: str
    step_type: StepType
    input_hash: str
    output: Any
    success: bool
    error: Optional[str] = None
    duration_ms: float = 0.0
    timestamp: float = field(default_factory=time.time)


@dataclass
class WorkflowState:
    workflow_id: str
    current_step: int
    completed_steps: list[StepResult] = field(default_factory=list)
    # conversation_summary 用于恢复 LLM 上下文
    conversation_summary: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)


class CheckpointWorkflowEngine:
    """支持断点续传的工作流引擎。
    
    架构决策：
    - 使用 Redis 存检查点而非本地文件，支持多实例共享
    - 检查点带 TTL（24h），过期自动清理，防止无限积累
    - 每步执行前检查是否存在有效检查点，优先恢复
    """

    def __init__(self, redis_url: str, checkpoint_ttl: int = 86400):
        self._redis = redis.from_url(redis_url)
        self._checkpoint_ttl = checkpoint_ttl
        self._llm_cache: dict[str, Any] = {}

    async def execute(
        self,
        workflow_id: str,
        steps: list[Callable],
        step_types: list[StepType],
        fallback: Optional[Callable] = None,
    ) -> WorkflowState:
        """执行工作流并自动处理断点续传。"""
        # 尝试加载已有状态
        state = await self._load_state(workflow_id)

        if state is None:
            state = WorkflowState(workflow_id=workflow_id, current_step=0)

        # 从断点开始执行
        while state.current_step < len(steps):
            step_fn = steps[state.current_step]
            step_type = step_types[state.current_step] if state.current_step < len(step_types) else StepType.IDEMPOTENT
            step_id = f"{workflow_id}:step_{state.current_step}"

            # 检查是否已经成功执行过（用于恢复场景）
            # 副作用步骤在恢复时跳过重放——只检查是否已完成
            if self._is_step_completed(state, step_id):
                state.current_step += 1
                continue

            start = time.monotonic()
            try:
                result = await step_fn()
                step_result = StepResult(
                    step_id=step_id,
                    step_type=step_type,
                    input_hash=self._hash_input(state),
                    output=result,
                    success=True,
                    duration_ms=(time.monotonic() - start) * 1000,
                )
            except Exception as e:
                step_result = StepResult(
                    step_id=step_id,
                    step_type=step_type,
                    input_hash=self._hash_input(state),
                    output=None,
                    success=False,
                    error=str(e),
                    duration_ms=(time.monotonic() - start) * 1000,
                )

                if fallback and state.current_step > 0:
                    # 有回退策略且之前有成功步骤: 回到上一个检查点
                    state.current_step -= 1
                    await self._save_state(state)
                    continue
                
                state.completed_steps.append(step_result)
                await self._save_state(state)
                # 不吞异常，让上层感知失败
                raise RuntimeError(f"Step {step_id} failed: {e}") from e

            state.completed_steps.append(step_result)
            state.current_step += 1

            # 每步完成后保存检查点
            await self._save_state(state)

        return state

    async def _save_state(self, state: WorkflowState) -> None:
        """持久化工作流状态到 Redis。
        
        使用 SETEX 自动过期，避免僵尸检查点积累。
        为什么不用 Redis 事务？因为单个 SETEX 命令已经原子性足够，
        多字段写入用 pipeline 即可，不需要 MULTI/EXEC 的隔离保证。
        """
        key = f"workflow:checkpoint:{state.workflow_id}"
        data = json.dumps(self._serialize_state(state))
        await self._redis.setex(key, self._checkpoint_ttl, data)

    async def _load_state(self, workflow_id: str) -> Optional[WorkflowState]:
        key = f"workflow:checkpoint:{workflow_id}"
        data = await self._redis.get(key)
        if data is None:
            return None
        return self._deserialize_state(json.loads(data))

    def _is_step_completed(self, state: WorkflowState, step_id: str) -> bool:
        return any(s.step_id == step_id and s.success for s in state.completed_steps)

    def _hash_input(self, state: WorkflowState) -> str:
        # 包含 conversation_summary 确保上下文变化时哈希也变化
        context = f"{state.workflow_id}:{state.current_step}:{state.conversation_summary}"
        return hashlib.sha256(context.encode()).hexdigest()[:16]

    def _serialize_state(self, state: WorkflowState) -> dict:
        return {
            "workflow_id": state.workflow_id,
            "current_step": state.current_step,
            "steps": [
                {
                    "step_id": s.step_id,
                    "step_type": s.step_type.value,
                    "input_hash": s.input_hash,
                    "success": s.success,
                    "error": s.error,
                }
                for s in state.completed_steps
            ],
            "summary": state.conversation_summary,
        }

    def _deserialize_state(self, data: dict) -> WorkflowState:
        state = WorkflowState(
            workflow_id=data["workflow_id"],
            current_step=data["current_step"],
            conversation_summary=data.get("summary", ""),
        )
        for s in data.get("steps", []):
            state.completed_steps.append(StepResult(
                step_id=s["step_id"],
                step_type=StepType(s["step_type"]),
                input_hash=s["input_hash"],
                output=None,
                success=s["success"],
                error=s.get("error"),
            ))
        return state

四、断点续传的代价

存储成本

Redis 中每个检查点约 1-5KB。日均 10 万次工作流执行，存储开销约 500MB——可接受。但不要用关系型数据库存检查点，写入频率高会导致主从延迟。MySQL 的 WAL 写入在高频场景下会产生 binlog 延迟，而 Redis 的纯内存操作延迟 < 1ms。

缓存一致性问题

LLM 输出的缓存需要区分模型版本。GPT-4 和 GPT-4o 对同一 prompt 的输出可能不同——这不是理论上的可能性，而是实践中确实会发生的。缓存 key 必须包含模型标识和版本号，否则跨版本恢复会导致结果不一致。

更深层的缓存问题：temperature 参数。temperature = 0 时 LLM 输出相对确定（但非绝对），可以缓存。temperature > 0 时输出不确定，缓存无效。如果缓存策略把 temperature = 0.01 的输出也缓存了，恢复后可能得到与原始执行不同的结果——因为 0.01 仍然有微小随机性。

不适用场景

单步工作流（只有一步，失败就失败，不需要检查点）
实时性要求 < 100ms（写 Redis 增加 2-5ms 延迟）
对不可重现性有要求的场景（A/B 测试等，需要每次运行结果不同）

Trade-off 讨论

检查点频率 vs 存储成本：每步写检查点是最安全的，但写入次数最多。可以改为"只在副作用步骤前写检查点"——减少写入，但失败时可能多回退一步。推荐：关键步骤（副作用前、决策点）写检查点，普通步骤可以省略。
缓存精度 vs 性能：temperature = 0 时缓存是安全的，但禁用随机性可能降低回答质量。temperature > 0 时不缓存，但恢复后可能得到不同结果。这是一个无法完美解决的 trade-off。
TTL 长短 vs 恢复窗口：24h TTL 适合大多数场景，但长时间运行的批处理任务可能需要更长 TTL。TTL 过短会导致任务还没恢复就过期了。推荐：按工作流类型设置不同 TTL——实时对话 1h、批处理任务 72h。

五、总结

断点续传的本质是把"无状态执行"变成"有状态恢复"。这个转换的代价是每次执行都要写一次 Redis，但换来的是失败瞬间的秒级恢复——对于多步骤 Agent 工作流，这个取舍是值得的。检查点设计上遵循最小快照原则，只存结果不存过程，才能在存储开销和恢复速度之间找到平衡。副作用步骤不可重放是系统中最需要小心处理的边界——搞错这一点会导致业务错误而非技术错误。最终的收益：5 步工作流第 4 步失败，恢复时间从 50 秒降到 10 秒，成本从 $0.10 降到 $0.02。