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 膨胀到 10KBRedis 写入延迟增加但恢复后的执行效果完全一样——得不偿失。最小快照的另一个优势跨实例恢复。如果原实例挂了另一个实例可以加载检查点继续执行——因为检查点只包含决策结果而非内存状态没有实例绑定的数据。原则二幂等性保证恢复时可能重新执行某些步骤。设计上必须保证同一步骤以相同输入重复执行结果等价。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_factorytime.time) dataclass class WorkflowState: workflow_id: str current_step: int completed_steps: list[StepResult] field(default_factorylist) # conversation_summary 用于恢复 LLM 上下文 conversation_summary: str metadata: dict[str, Any] field(default_factorydict) class CheckpointWorkflowEngine: 支持断点续传的工作流引擎。 架构决策 - 使用 Redis 存检查点而非本地文件支持多实例共享 - 检查点带 TTL24h过期自动清理防止无限积累 - 每步执行前检查是否存在有效检查点优先恢复 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_idworkflow_id, current_step0) # 从断点开始执行 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_idstep_id, step_typestep_type, input_hashself._hash_input(state), outputresult, successTrue, duration_ms(time.monotonic() - start) * 1000, ) except Exception as e: step_result StepResult( step_idstep_id, step_typestep_type, input_hashself._hash_input(state), outputNone, successFalse, errorstr(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(fStep {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 fworkflow: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 fworkflow: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_iddata[workflow_id], current_stepdata[current_step], conversation_summarydata.get(summary, ), ) for s in data.get(steps, []): state.completed_steps.append(StepResult( step_ids[step_id], step_typeStepType(s[step_type]), input_hashs[input_hash], outputNone, successs[success], errors.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。
Agent 工作流重放与断点续传：失败任务不必从头跑
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 膨胀到 10KBRedis 写入延迟增加但恢复后的执行效果完全一样——得不偿失。最小快照的另一个优势跨实例恢复。如果原实例挂了另一个实例可以加载检查点继续执行——因为检查点只包含决策结果而非内存状态没有实例绑定的数据。原则二幂等性保证恢复时可能重新执行某些步骤。设计上必须保证同一步骤以相同输入重复执行结果等价。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_factorytime.time) dataclass class WorkflowState: workflow_id: str current_step: int completed_steps: list[StepResult] field(default_factorylist) # conversation_summary 用于恢复 LLM 上下文 conversation_summary: str metadata: dict[str, Any] field(default_factorydict) class CheckpointWorkflowEngine: 支持断点续传的工作流引擎。 架构决策 - 使用 Redis 存检查点而非本地文件支持多实例共享 - 检查点带 TTL24h过期自动清理防止无限积累 - 每步执行前检查是否存在有效检查点优先恢复 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_idworkflow_id, current_step0) # 从断点开始执行 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_idstep_id, step_typestep_type, input_hashself._hash_input(state), outputresult, successTrue, duration_ms(time.monotonic() - start) * 1000, ) except Exception as e: step_result StepResult( step_idstep_id, step_typestep_type, input_hashself._hash_input(state), outputNone, successFalse, errorstr(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(fStep {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 fworkflow: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 fworkflow: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_iddata[workflow_id], current_stepdata[current_step], conversation_summarydata.get(summary, ), ) for s in data.get(steps, []): state.completed_steps.append(StepResult( step_ids[step_id], step_typeStepType(s[step_type]), input_hashs[input_hash], outputNone, successs[success], errors.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。
