Agent 工作流的异常处理：失败可恢复的执行设计
Agent 工作流的异常处理：失败可恢复的执行设计
一、多步执行的脏状态陷阱
Agent 一次任务往往分多步：查数据、调工具、写结果。走到第三步失败了，前两步的副作用已落地。数据库写了一半，文件改了一半，外部 API 调了。重跑？
从头来，副作用重复执行。不跑？脏状态留在系统里，下次接着错。进退两难，是 Agent 落地最痛的坑。
和单次推理不同，多步执行是"有状态"的。状态没管好，失败就不可恢复。本文探讨让 Agent 失败可回滚的执行设计。
二、事务式执行与检查点机制
可恢复的核心是"事务思维"。每步执行前，先存检查点（checkpoint）。检查点记录：当前状态、已完成的步骤、待补偿动作。失败时，按检查点回放补偿动作。
补偿是正向操作的反函数：写了就删，加了就减。不是所有操作都可补偿，不可补偿的必须前置校验。下面是事务式执行的链路：
flowchart TD A[Agent 任务] --> B[存检查点] B --> C[执行步骤] C --> D{成功?} D -->|是| E[更新检查点] E --> F{还有步骤?} F -->|是| B F -->|否| G[任务完成] D -->|否| H[读检查点] H --> I[执行补偿动作] I --> J[回滚到安全态] J --> K[断点续跑或告警] style G fill:#e8f5e9 style K fill:#ffebee关键在"补偿的完备性"。不可补偿的操作（如发短信、转账）不能靠事后撤销。必须前置：要么改设计为可补偿，要么用两阶段提交。补偿设计不到位，回滚就是空谈。
补偿与回滚不是一回事。回滚是数据库事务的原生能力，依赖日志恢复到旧版本。补偿是应用层的正向操作，用“反向动作”抵消已产生的副作用。数据库写可以回滚，发出去的短信只能补偿，再发一条撤销通知。
补偿必须逆序执行。先执行的步骤副作用在底层，后执行的在上层。逆序撤销才不会破坏中间状态。比如先建文件再写内容，补偿要先删内容再删文件，顺序反了会报错。
检查点要存外部。进程内存里的检查点，进程一崩就没了。存数据库或对象存储，进程重启后能读回来续跑。检查点要带版本号，格式升级后旧检查点还能解析。
三、生产级带检查点的执行器
下面用 Python 实现一个带检查点与补偿的 Agent 执行器。
from dataclasses import dataclass, field from typing import Callable @dataclass class Step: """一个执行步骤：含正向动作与补偿动作""" name: str do: Callable[[], None] # 补偿是正向的反函数，失败时按完成步骤逆序调用 undo: Callable[[], None] = lambda: None @dataclass class Checkpoint: """检查点：记录已完成步骤，供失败时回滚与续跑""" completed: list[str] = field(default_factory=list) def save(self, name: str) -> None: if name not in self.completed: self.completed.append(name) def run(steps: list[Step], cp: Checkpoint) -> bool: """事务式执行：成功推进，失败按已完成步骤逆序补偿""" for step in steps: # 跳过已完成步骤，支持断点续跑 if step.name in cp.completed: continue try: step.do() cp.save(step.name) except Exception as exc: # 失败时逆序补偿已完成的步骤，回滚到安全态 for done in reversed(cp.completed): for s in steps: if s.name == done: try: s.undo() except Exception: # 补偿失败要告警，不能静默吞掉 print(f"补偿失败: {s.name}") print(f"任务失败于 {step.name}: {exc}") return False return True if __name__ == "__main__": cp = Checkpoint() def write_db(): print("写入数据库") def rollback_db(): print("回滚数据库") def call_api(): raise RuntimeError("API 不可用") steps = [ Step("write_db", write_db, rollback_db), Step("call_api", call_api), ] # 第二步失败，第一步的补偿会被自动触发 ok = run(steps, cp) print("成功" if ok else "已回滚")真实系统会把检查点持久化到外部存储。进程崩了重启后，读检查点续跑，而非从头来。补偿动作要做幂等，重试不产生重复副作用。幂等靠业务 ID 兜底。
每个副作用操作带唯一键，执行前先查是否已做过。已做过直接跳过，未做过才执行。补偿动作同理，带键查重，避免重复撤销把正确的状态也撤掉。检查点写入要原子。
检查点与副作用操作不能跨网络分两次写，否则中间崩了会出现“副作用做了但检查点没记”的脏状态。应先写检查点标记“待执行”，再执行副作用，最后更新检查点为“已完成”。这套写前日志模式，借鉴自数据库的 WAL。
四、Agent 工作流的异常处理的代价与边界
事务式执行稳妥，但不是银弹。
补偿的不可行性。发短信、发邮件、真实扣款。这些操作无法撤销，补偿无从设计。必须前置校验，或用 saga 模式拆成可补偿的子事务。
检查点的存储成本。每步存检查点有 I/O 开销。高频小步任务，检查点开销可能抵消 Agent 的效率。应在关键节点存，而非每步都存。
补偿失败的雪崩。补偿本身也可能失败。补偿失败不告警，脏状态更深。必须对补偿失败做告警，并留人工介入入口。
幂等性的要求。续跑时已执行步骤可能重复执行。正向与补偿动作都必须幂等。否则重试一次，副作用翻倍。
事务式执行的"补偿完备性审计"要做在前面。上线前逐步骤检查：这个操作能否撤销？撤销失败怎么办？答不上来的步骤就是定时炸弹。
另一个被忽视的点是"检查点的可见性"：检查点存在哪、什么格式、怎么读，要文档化。否则故障时想人工续跑，连状态都看不懂。最后，对外部系统的调用要设超时与重试上限，避免一步卡死整个事务，补偿动作也跟着永远等不到触发，脏状态无限期残留。
五、总结
Agent 可恢复执行，本质是用"检查点 + 补偿"换"失败可回滚"。机制上每步存检查点，失败时逆序执行补偿动作。工程上补偿需幂等，不可补偿操作前置校验。落地路线：先识别每步的可补偿性；关键节点存检查点；失败逆序补偿并告警；持久化检查点支持断点续跑。Agent 可以失败，但失败后系统必须能回到安全态。