Agent 的回滚与补偿机制分布式事务思维在多步骤 Agent 中的应用一、深度引言与场景痛点多步骤 Agent 工作中最常见的一个事故现场Agent 执行了四个步骤——查询库存成功、创建订单成功、扣减优惠券成功、调用支付失败。然后呢订单已经创建了优惠券已经扣了用户的钱还没付。Agent 报错了用户傻了数据库里多了一条幽灵订单永远挂在待支付状态。这个问题的本质是多步骤 Agent 的每一步都可能对系统产生副作用写数据库、调 API、发消息但 Agent 原生没有任何事务性保障。传统 Web 应用里多个写操作可以包在一个数据库事务里要么全成功要么全回滚。但 Agent 的步骤通常跨越多个独立的系统——库存系统、订单系统、优惠券系统、支付系统——它们不在同一个数据库事务里没法用 ACID 事务。分布式事务里的 Saga 模式解决了类似的问题把一个长事务拆成一系列本地事务每个本地事务都有一个对应的补偿事务。如果某个步骤失败就按相反顺序执行之前成功步骤的补偿事务把系统恢复到一致状态。这种正向操作 补偿回滚的思路可以直接应用到多步骤 Agent 的编排中。二、底层机制与原理深度剖析flowchart TD START((开始)) -- S1[步骤 1: 查询库存br/side_effect: 无br/compensation: 无] S1 -- S1_OK{步骤 1 成功?} S1_OK --|是| S2[步骤 2: 创建订单br/side_effect: 写入订单表br/compensation: 取消订单] S1_OK --|否| FAIL[终止库存不足br/无需回滚] S2 -- S2_OK{步骤 2 成功?} S2_OK --|是| S3[步骤 3: 锁定优惠券br/side_effect: 扣减优惠券余额br/compensation: 释放优惠券] S2_OK --|否| ROLLBACK_AFTER_S2[补偿无br/步骤 2 失败S2 未执行] ROLLBACK_AFTER_S2 -- FAIL_2[终止订单创建失败] S3 -- S3_OK{步骤 3 成功?} S3_OK --|是| S4[步骤 4: 发起支付br/side_effect: 调用支付网关br/compensation: 发起退款] S3_OK --|否| ROLLBACK_S2[执行补偿 2: 取消订单] ROLLBACK_S2 -- ROLLBACK_CHECK_2{补偿成功?} ROLLBACK_CHECK_2 --|是| FAIL_3[终止数据已回滚] ROLLBACK_CHECK_2 --|否| ALERT[告警补偿失败br/需要人工介入] S4 -- S4_OK{步骤 4 成功?} S4_OK --|是| SUCCESS[全部完成] S4_OK --|否| ROLLBACK_S3[执行补偿 3: 释放优惠券] ROLLBACK_S3 -- ROLLBACK_3A{补偿 3 成功?} ROLLBACK_3A --|是| ROLLBACK_S2B[执行补偿 2: 取消订单] ROLLBACK_3A --|否| ALERT_S3[告警优惠券补偿失败] ROLLBACK_S2B -- ROLLBACK_3B{补偿 2 成功?} ROLLBACK_3B --|是| FAIL_4[终止数据已回滚] ROLLBACK_3B --|否| ALERT_FINAL[告警多重补偿失败br/需紧急人工介入] style S1 fill:#e8f5e9,stroke:#2e7d32 style S2 fill:#e8f5e9,stroke:#2e7d32 style S3 fill:#e8f5e9,stroke:#2e7d32 style S4 fill:#e8f5e9,stroke:#2e7d32 style ROLLBACK_S2 fill:#ffebee,stroke:#c62828 style ROLLBACK_S3 fill:#ffebee,stroke:#c62828 style ROLLBACK_S2B fill:#ffebee,stroke:#c62828 style ALERT fill:#f44336,stroke:#b71c1c,color:#fff style ALERT_S3 fill:#f44336,stroke:#b71c1c,color:#fff style ALERT_FINAL fill:#b71c1c,stroke:#880e4f,color:#fff style SUCCESS fill:#4caf50,stroke:#1b5e20,color:#fffSaga 模式的核心是每个正向步骤配一个补偿步骤。创建订单的补偿是取消订单锁定优惠券的补偿是释放优惠券发起支付的补偿是发起退款。注意补偿不等于撤销——已创建的订单不能真的删除审计要求只能标记为已取消已扣减的优惠券也不能直接恢复可能有并发问题需要增加余额。回滚的执行顺序是正向步骤的逆序——最后成功的先回滚。步骤 4 失败时先执行步骤 3 的补偿释放优惠券再执行步骤 2 的补偿取消订单。这个逆序很重要如果先取消订单再释放优惠券中间出现新的失败会导致优惠券永久丢失——先释放优惠券即使取消订单失败损失也只是一条已取消但优惠券没释放的孤立记录影响范围更小。补偿本身也可能失败。这是一种双重故障场景——原始操作失败了补偿操作也失败了。Saga 模式无法自动处理双重故障必须升级到人工介入。在 Agent 场景里升级方式可以是发送告警到运维群、创建工单、暂停该用户的其他操作。代码层面要做好补偿幂等——如果补偿因为网络超时被重试了两次实际只执行了一次释放优惠券。三、生产级代码实现import asyncio import logging from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Callable, Awaitable logger logging.getLogger(__name__) class StepStatus(str, Enum): PENDING pending SUCCESS success FAILED failed COMPENSATING compensating COMPENSATED compensated COMPENSATION_FAILED compensation_failed dataclass class SagaStep: Saga 中的一个步骤 name: str execute: Callable[[dict[str, Any]], Awaitable[dict[str, Any]]] compensate: Callable[[dict[str, Any]], Awaitable[dict[str, Any]]] | None None # 运行态 status: StepStatus StepStatus.PENDING result: dict[str, Any] | None None error: str | None None started_at: str | None None completed_at: str | None None dataclass class SagaContext: Saga 共享上下文——步骤间传递数据 data: dict[str, Any] field(default_factorydict) step_results: list[dict[str, Any]] field(default_factorylist) dataclass class SagaResult: Saga 执行结果 success: bool completed_steps: list[str] failed_step: str | None None compensated_steps: list[str] field(default_factorylist) compensation_failed: list[str] field(default_factorylist) total_time_ms: float 0 class SagaExecutor: Agent Saga 事务执行器 def __init__( self, steps: list[SagaStep], alert_fn: Callable[[str, dict[str, Any]], Awaitable[None]] | None None, ) - None: self._steps steps self._alert_fn alert_fn async def execute(self, initial_context: dict[str, Any] | None None) - SagaResult: 执行 Saga正向执行 → 失败时逆向补偿 import time t_start time.perf_counter() ctx SagaContext(datainitial_context or {}) completed_steps: list[SagaStep] [] # ── 正向执行 ── for step in self._steps: step.started_at datetime.now(timezone.utc).isoformat() try: # 将前面步骤的结果合并入上下文 for prev in completed_steps: if prev.result: ctx.data.update(prev.result) result await step.execute(ctx.data) step.result result step.status StepStatus.SUCCESS step.completed_at datetime.now(timezone.utc).isoformat() ctx.step_results.append({step: step.name, status: success, result: result}) completed_steps.append(step) logger.info(Step [%s] succeeded, step.name) except Exception as e: step.status StepStatus.FAILED step.error str(e) logger.error(Step [%s] failed: %s, step.name, e) # ── 逆向补偿 ── compensation_result await self._compensate(completed_steps, ctx) return SagaResult( successFalse, completed_steps[s.name for s in completed_steps], failed_stepstep.name, compensated_stepscompensation_result[compensated], compensation_failedcompensation_result[failed], total_time_ms(time.perf_counter() - t_start) * 1000, ) # 全部成功 return SagaResult( successTrue, completed_steps[s.name for s in self._steps], total_time_ms(time.perf_counter() - t_start) * 1000, ) async def _compensate( self, completed_steps: list[SagaStep], ctx: SagaContext ) - dict[str, list[str]]: 逆向执行补偿操作 compensated: list[str] [] failed: list[str] [] # 逆序遍历已完成的步骤 for step in reversed(completed_steps): if step.compensate is None: logger.info(Step [%s] has no compensation, skipping, step.name) continue step.status StepStatus.COMPENSATING try: await step.compensate(ctx.data) step.status StepStatus.COMPENSATED compensated.append(step.name) logger.info(Compensation for [%s] succeeded, step.name) except Exception as e: step.status StepStatus.COMPENSATION_FAILED failed.append(step.name) logger.error(Compensation for [%s] failed: %s, step.name, e) # 补偿失败 → 告警 if self._alert_fn: try: await self._alert_fn( fAgent Saga 补偿失败: step{step.name}, error{e}, { failed_step: step.name, error: str(e), context: ctx.data, }, ) except Exception: logger.exception(Alert function itself failed) return {compensated: compensated, failed: failed} # ── 使用示例 ── async def main() - None: # 定义 Saga 步骤 async def check_inventory(ctx: dict[str, Any]) - dict[str, Any]: 步骤 1: 查询库存无副作用不需要补偿 await asyncio.sleep(0.1) return {stock_available: True, item_id: ctx.get(item_id)} async def create_order(ctx: dict[str, Any]) - dict[str, Any]: 步骤 2: 创建订单 await asyncio.sleep(0.2) # 生产环境INSERT INTO orders... order_id fORD-{hash(str(ctx)) % 10000:05d} return {order_id: order_id, status: created} async def cancel_order(ctx: dict[str, Any]) - dict[str, Any]: 步骤 2 的补偿取消订单 order_id ctx.get(order_id, unknown) # 生产环境UPDATE orders SET statuscancelled WHERE id? await asyncio.sleep(0.1) return {order_id: order_id, status: cancelled} async def lock_coupon(ctx: dict[str, Any]) - dict[str, Any]: 步骤 3: 锁定优惠券 await asyncio.sleep(0.15) # 生产环境UPDATE coupons SET statuslocked WHERE user_id? return {coupon_locked: True, coupon_id: CPN-001} async def release_coupon(ctx: dict[str, Any]) - dict[str, Any]: 步骤 3 的补偿释放优惠券 await asyncio.sleep(0.1) return {coupon_released: True} async def process_payment(ctx: dict[str, Any]) - dict[str, Any]: 步骤 4: 发起支付故意模拟失败 await asyncio.sleep(0.3) # 模拟支付失败 if ctx.get(simulate_payment_failure): raise RuntimeError(支付网关超时) return {payment_id: PAY-001, status: paid} async def refund_payment(ctx: dict[str, Any]) - dict[str, Any]: 步骤 4 的补偿退款 await asyncio.sleep(0.2) return {refund_id: REF-001} async def alert(msg: str, details: dict[str, Any]) - None: 告警函数 logger.warning(ALERT: %s | 详情: %s, msg, details) # 构建 Saga steps [ SagaStep(name查库存, executecheck_inventory), SagaStep(name创建订单, executecreate_order, compensatecancel_order), SagaStep(name锁定优惠券, executelock_coupon, compensaterelease_coupon), SagaStep(name发起支付, executeprocess_payment, compensaterefund_payment), ] executor SagaExecutor(stepssteps, alert_fnalert) # 测试 1: 全部成功 print( 测试 1: 正常流程 ) result1 await executor.execute({item_id: ITEM-001, simulate_payment_failure: False}) print(f成功: {result1.success}, 补偿步骤: {result1.compensated_steps}) print(f总耗时: {result1.total_time_ms:.0f}ms) # 重置步骤状态 for s in steps: s.status StepStatus.PENDING s.result None s.error None # 测试 2: 支付失败 → 触发回滚 print(\n 测试 2: 支付失败 → 回滚 ) result2 await executor.execute({item_id: ITEM-002, simulate_payment_failure: True}) print(f成功: {result2.success}) print(f已完成步骤: {result2.completed_steps}) print(f失败步骤: {result2.failed_step}) print(f已补偿: {result2.compensated_steps}) print(f补偿失败: {result2.compensation_failed}) if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())SagaExecutor的正向执行是顺序的Agent 的步骤通常有依赖关系补偿是逆序的。ctx.data作为步骤间的共享上下文上游步骤的结果通过ctx.data.update(prev.result)传递给下游。关键设计第一个步骤check_inventory没有 compensate 函数——因为查询操作没有副作用不需要回滚。Saga 模式只关心有副作用的步骤写数据库、调外部 API只读步骤天然是安全的。补偿失败的告警机制通过alert_fn注入——补偿失败是需要人工介入的信号自动重试可能导致问题扩大化比如释放优惠券两次。告警中携带了上下文和错误信息运维可以据此判断是需要手动补偿还是忽略如果优惠券已经因为超时自动释放了重复释放的失败可以忽略。四、边界分析与架构权衡Saga 最大的挑战是补偿的完备性——你能保证每一步都有补偿且补偿能真实还原系统状态吗有些操作从本质上就是不可逆的发送了一封邮件、推送了一条短信通知、触发了一次外部 webhook。这些操作的补偿只能是发送第二封邮件说前面那封请忽略——用户体验很差。对于这类不可逆操作建议把它们的执行放到 Saga 的最后一步——因为如果前面的步骤回滚这些操作根本没执行。幂等性是补偿的基石。如果补偿操作释放优惠券不是幂等的——执行两次会导致优惠券余额被加了两次——那么重试补偿就可能造成二次伤害。设计补偿操作时应该用设置目标状态而非增减数据的方式实现UPDATE coupons SET balance balance 1非幂等→UPDATE coupons SET status available WHERE id ? AND status locked幂等。Saga 模式对业务代码的侵入性是另一个边界。为了支持补偿每个业务操作都需要额外实现一个补偿接口且补偿逻辑需要和正向逻辑保持一致的演进——正向操作改了字段补偿也要同步改。如果业务迭代快补偿代码很容易腐化。降低侵入的方案是事件溯源Event Sourcing替代补偿——不修改状态只追加事件通过重放事件来恢复状态。但事件溯源的实现复杂度高于 Saga适合对一致性要求极高的金融场景。五、总结Agent 的回滚与补偿机制借鉴了分布式事务中 Saga 模式的核心思想每个有副作用的步骤配一个补偿步骤失败时逆序执行补偿。实现上SagaExecutor管理正向执行和逆向补偿补偿失败的告警升级到人工介入。关键设计准则只读步骤不需要补偿、补偿必须是幂等的、不可逆操作放在最后一步、补偿失败不应该自动重试。Saga 不是银弹——它增加了业务代码的维护成本但对于跨多个独立系统的多步骤 Agent它是目前最实用的最终一致性方案。