Agent 故障回滚机制任务执行到一半失败了怎么优雅地恢复到安全状态一、深度引言与场景痛点大家好我是赵咕咕。你的 Agent 在执行一个多步骤任务先调用了数据库 A 插入了一条记录又调用了外部 API B 发送了通知正准备调用服务 C 更新状态时——网络超时了。现在问题来了数据库 A 的记录已经写进去了API B 的通知也发出去了但任务整体失败了。用户看到的是任务失败请重试。重试后数据库里又多了一条一模一样的记录。这就是分布式任务中最棘手的部分失败问题。Agent 的多步骤操作不是原子的——你不能简单地回滚因为有些副作用如通知、扣款是不可逆的。这篇文章我们来设计 Agent 的故障回滚机制让任务即使中途失败也能优雅地恢复到安全状态不让用户看到半成品。二、底层机制与原理深度剖析Agent 故障回滚的核心策略是Saga 模式把一个长事务拆成多个局部事务每个局部事务都有一个对应的补偿操作。Saga 有两种实现编排式Choreography每个步骤完成后触发下一个失败时也自己触发补偿编排式Orchestration由中心协调器控制步骤的执行和回滚Agent 场景推荐使用编排式因为 Agent 本来就是一个天然的编排器stateDiagram-v2 [*] -- 准备阶段: 收到任务 准备阶段 -- Step1_执行: 初始化上下文 Step1_执行 -- Step1_成功: 操作完成 Step1_执行 -- 补偿阶段: 操作失败 Step1_成功 -- Step2_执行: 继续下一步 Step2_执行 -- Step2_成功: 操作完成 Step2_执行 -- 补偿阶段: 操作失败 Step2_成功 -- Step3_执行: 继续下一步 Step3_执行 -- Step3_成功: 操作完成 Step3_执行 -- 补偿阶段: 操作失败 Step3_成功 -- 完成阶段: 全部完成 补偿阶段 -- StepN_补偿: 逆序补偿 StepN_补偿 -- StepN-1_补偿: 继续补偿 StepN-1_补偿 -- Step1_补偿: 继续补偿 Step1_补偿 -- 失败终态: 补偿完成 失败终态 -- 准备阶段: 可选择重试 完成阶段 -- [*]: 任务完成 note right of 补偿阶段 逆序执行补偿操作 幂等性保证安全 end note note right of 完成阶段 任务状态持久化 通知用户结果 end note关键设计原则幂等性每个操作可以被反复执行而结果不变。用唯一 ID 做去重检查。补偿而非回滚已经发送的通知没法撤回只能发送新的通知更正。逆序补偿从最后成功的一步开始向前补偿而不是从第一步。补偿也可能失败补偿操作本身也可能失败需要重试机制。三、生产级代码实现下面是 Agent Saga 回滚机制的生产级实现from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Optional, Callable, Awaitable from enum import Enum import time import uuid import json class StepStatus(Enum): 步骤状态 PENDING pending # 待执行 RUNNING running # 执行中 SUCCEEDED succeeded # 已成功 FAILED failed # 已失败 COMPENSATING compensating # 补偿中 COMPENSATED compensated # 已补偿 COMPENSATE_FAILED compensate_failed # 补偿失败 dataclass class SagaStep: Saga 中的一个步骤 step_id: str name: str execute: Callable[[dict], Awaitable[dict]] # 执行函数 compensate: Callable[[dict], Awaitable[dict]] # 补偿函数 max_retries: int 3 retry_delay_seconds: float 1.0 status: StepStatus StepStatus.PENDING result: Optional[dict] None error: Optional[str] None retry_count: int 0 dataclass class SagaContext: Saga 执行上下文在步骤间传递数据 saga_id: str field(default_factorylambda: str(uuid.uuid4())[:8]) task_id: str data: dict field(default_factorydict) created_at: float field(default_factorytime.time) def get(self, key: str, defaultNone): return self.data.get(key, default) def set(self, key: str, value): self.data[key] value class SagaOrchestrator: Saga 编排器控制步骤执行和回滚 def __init__(self): self._steps: list[SagaStep] [] def add_step(self, step: SagaStep) - SagaOrchestrator: 添加步骤链式调用 self._steps.append(step) return self async def execute( self, context: SagaContext ) - tuple[bool, SagaContext]: 执行 Saga 返回: (是否全部成功, 执行上下文) completed_steps: list[SagaStep] [] try: for step in self._steps: success await self._run_step(step, context) if success: completed_steps.append(step) else: # 失败了启动补偿 await self._compensate( completed_steps, context, failed_stepstep ) return False, context except Exception as e: print(fSaga 执行异常中断: {e}) await self._compensate(completed_steps, context) return False, context # 全部成功 return True, context async def _run_step( self, step: SagaStep, context: SagaContext ) - bool: 执行单个步骤带重试 step.status StepStatus.RUNNING step.retry_count 0 while step.retry_count step.max_retries: try: step.result await step.execute(context) step.status StepStatus.SUCCEEDED print( f [{context.saga_id}] ✅ {step.name} 执行成功 f{f (重试 {step.retry_count} 次后) if step.retry_count else } ) return True except Exception as e: step.retry_count 1 step.error str(e) if step.retry_count step.max_retries: print( f [{context.saga_id}] ⚠️ {step.name} 失败 f第 {step.retry_count}/{step.max_retries} 次重试... ) await asyncio.sleep(step.retry_delay_seconds) else: step.status StepStatus.FAILED print( f [{context.saga_id}] ❌ {step.name} 最终失败: {e} ) return False return False async def _compensate( self, completed_steps: list[SagaStep], context: SagaContext, failed_step: Optional[SagaStep] None, ) - None: 逆序补偿已完成的步骤 if not completed_steps: print(f [{context.saga_id}] 无步骤需要补偿) return print( f [{context.saga_id}] 开始补偿 f{len(completed_steps)} 个已完成的步骤 ) # 逆序补偿 for step in reversed(completed_steps): await self._compensate_step(step, context) async def _compensate_step( self, step: SagaStep, context: SagaContext ) - bool: 补偿单个步骤带重试 step.status StepStatus.COMPENSATING for attempt in range(step.max_retries 1): try: await step.compensate(context) step.status StepStatus.COMPENSATED print( f [{context.saga_id}] ↩️ {step.name} 补偿成功 f{f (重试 {attempt} 次后) if attempt else } ) return True except Exception as e: if attempt step.max_retries: print( f [{context.saga_id}] ⚠️ {step.name} 补偿失败 f第 {attempt 1}/{step.max_retries} 次重试... ) await asyncio.sleep(step.retry_delay_seconds * 2) else: step.status StepStatus.COMPENSATE_FAILED step.error f补偿也失败了: {e} print( f [{context.saga_id}] {step.name} f补偿最终失败需要人工介入: {e} ) return False return False def status_summary(self) - list[dict]: 获取步骤状态摘要 return [ { step: s.name, status: s.status.value, error: s.error, retries: s.retry_count, } for s in self._steps ] # 使用示例模拟 Agent 多步骤任务 async def create_order(context: SagaContext) - dict: 步骤 1创建订单 order_id fORD-{context.saga_id} await asyncio.sleep(0.2) context.set(order_id, order_id) return {order_id: order_id, status: created} async def compensate_order(context: SagaContext) - dict: 步骤 1 的补偿取消订单 order_id context.get(order_id, unknown) await asyncio.sleep(0.1) print(f → 补偿: 取消订单 {order_id}) return {status: cancelled} async def deduct_inventory(context: SagaContext) - dict: 步骤 2扣减库存 order_id context.get(order_id) await asyncio.sleep(0.2) return {inventory: deducted, order_id: order_id} async def compensate_inventory(context: SagaContext) - dict: 步骤 2 的补偿恢复库存 await asyncio.sleep(0.1) print(f → 补偿: 恢复库存) return {status: restored} async def send_notification(context: SagaContext) - dict: 步骤 3发送通知不可逆操作 order_id context.get(order_id) # 模拟有 30% 概率失败 import random if random.random() 0.3: raise Exception(通知服务超时) await asyncio.sleep(0.1) print(f → 发送通知: 订单 {order_id} 已创建) return {notification: sent} async def compensate_notification(context: SagaContext) - dict: 步骤 3 的补偿发送更正通知非撤回 order_id context.get(order_id) await asyncio.sleep(0.1) print(f → 补偿: 发送更正通知 订单 {order_id} 已取消) return {status: corrected} async def main(): # 构建 Saga saga SagaOrchestrator() saga.add_step(SagaStep( step_idstep_1, name创建订单, executecreate_order, compensatecompensate_order, )) saga.add_step(SagaStep( step_idstep_2, name扣减库存, executededuct_inventory, compensatecompensate_inventory, )) saga.add_step(SagaStep( step_idstep_3, name发送通知, executesend_notification, compensatecompensate_notification, max_retries2, )) # 执行 context SagaContext(task_idTASK-001) print(f\n Saga {context.saga_id} 开始执行 \n) success, ctx await saga.execute(context) print(f\n 结果 ) print(f成功: {success}) print(f步骤状态:) for s in saga.status_summary(): icon { succeeded: ✅, failed: ❌, compensated: ↩️, compensate_failed: , } print(f {icon.get(s[status], )} {s[step]}: {s[status]}) asyncio.run(main())四、边界分析与架构权衡Saga 模式虽好有几个需要认真考虑的边界补偿的不可逆操作。有些操作根本不能补偿——比如发出的短信、推送给第三方的数据、已经执行的支付。对于这类操作需要在设计阶段就识别出来放在 Saga 的最后执行。或者用两阶段确认先预留资源等所有步骤成功后再真正提交。补偿操作的幂等性。如果补偿过程中也失败了重新执行补偿时会不会造成过度补偿比如恢复库存时如果执行了两次库存变成了超额。每个补偿操作必须被设计为幂等的——用唯一 ID 做去重。Saga 状态的持久化。如果 Saga 编排器在执行过程中崩溃了需要能从持久化状态中恢复。在生产环境中每个步骤的状态和上下文数据都需要序列化到 Redis 或数据库中。恢复时从最后一个已完成的步骤继续执行或补偿。长时间运行 Saga 的并发。如果 Saga 执行时间很长数分钟以上期间其他 Saga 可能修改了相同的数据。需要乐观锁或分布式锁来防止并发冲突。比如扣减库存前先检查当前库存是否还够。嵌套 Saga 的传递问题。Agent 可能在一个 Saga 步骤中调用了另一个子 Agent子 Agent 内部也有自己的 Saga。这种嵌套 Saga 的补偿需要对齐边界——子 Saga 的失败是否触发父 Saga 的补偿告警与人工介入。当补偿也失败时不能静默处理。需要触发告警并创建人工工单。同时保留完整的执行记录让运维人员知道当时的上下文。五、总结Agent 的故障回滚不是一个可选的优雅特性而是生产环境的必备条件。没有它一次网络抖动就可能导致数据不一致。核心要点用 Saga 模式将长任务拆为小步骤每个步骤有对应的补偿操作失败时逆序补偿而非正序每个操作和补偿都必须幂等将不可逆操作放在 Saga 的最后执行补偿也失败时必须告警并等待人工介入能优雅失败的 Agent比不会失败的 Agent 更重要。——因为不会失败的系统不存在但会失败后能恢复的系统才是真正可用的。
