企业官网建设流程全解析

Agent 故障回滚机制：任务执行到一半失败了，怎么优雅地恢复到安全状态

一、深度引言与场景痛点

大家好，我是赵咕咕。

你的 Agent 在执行一个多步骤任务：先调用了数据库 A 插入了一条记录，又调用了外部 API B 发送了通知，正准备调用服务 C 更新状态时——网络超时了。

现在问题来了：数据库 A 的记录已经写进去了，API B 的通知也发出去了，但任务整体失败了。用户看到的是"任务失败，请重试"。重试后，数据库里又多了一条一模一样的记录。

这就是分布式任务中最棘手的"部分失败"问题。Agent 的多步骤操作不是原子的——你不能简单地"回滚"，因为有些副作用（如通知、扣款）是不可逆的。

这篇文章我们来设计 Agent 的故障回滚机制，让任务即使中途失败，也能优雅地恢复到安全状态，不让用户看到"半成品"。

二、底层机制与原理深度剖析

Agent 故障回滚的核心策略是Saga 模式：把一个长事务拆成多个局部事务，每个局部事务都有一个对应的补偿操作。

Saga 有两种实现：

1. 编排式（Choreography）：每个步骤完成后触发下一个，失败时也自己触发补偿
2. 编排式（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

关键设计原则：

1. 幂等性：每个操作可以被反复执行而结果不变。用唯一 ID 做去重检查。
2. 补偿而非回滚：已经发送的通知没法"撤回"，只能发送新的通知更正。
3. 逆序补偿：从最后成功的一步开始向前补偿，而不是从第一步。
4. 补偿也可能失败：补偿操作本身也可能失败，需要重试机制。

三、生产级代码实现

下面是 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_factory=lambda: str(uuid.uuid4())[:8]) task_id: str = "" data: dict = field(default_factory=dict) created_at: float = field(default_factory=time.time) def get(self, key: str, default=None): 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_step=step ) return False, context except Exception as e: print(f"Saga 执行异常中断: {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 = f"ORD-{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_id="step_1", name="创建订单", execute=create_order, compensate=compensate_order, )) saga.add_step(SagaStep( step_id="step_2", name="扣减库存", execute=deduct_inventory, compensate=compensate_inventory, )) saga.add_step(SagaStep( step_id="step_3", name="发送通知", execute=send_notification, compensate=compensate_notification, max_retries=2, )) # 执行 context = SagaContext(task_id="TASK-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 的故障回滚不是一个"可选的优雅特性"，而是"生产环境的必备条件"。没有它，一次网络抖动就可能导致数据不一致。

核心要点：

1. 用 Saga 模式将长任务拆为小步骤，每个步骤有对应的补偿操作
2. 失败时逆序补偿，而非正序
3. 每个操作和补偿都必须幂等
4. 将不可逆操作放在 Saga 的最后执行
5. 补偿也失败时，必须告警并等待人工介入

能优雅失败的 Agent，比不会失败的 Agent 更重要。——因为不会失败的系统不存在，但会失败后能恢复的系统才是真正可用的。