企业官网建设流程全解析

Agent 异常恢复架构：不是所有错误都能重试，有些要回滚

一、个性化深度引言

周一早上看到告警群里的截图，我后背一凉。用户对Agent说"帮我取消所有订阅"，Agent理解为"取消所有服务的订阅"——包括正在使用的核心服务。它调了3个API取消了用户的全部6个订阅，然后才"意识到"可能不对，但这3个API中有2个不支持恢复。

这是Agent异常恢复中最危险的一类错误：操作已经部分执行，无法简单重试，也不能直接回滚。用户在线上遇到这种情况比模型回复不好更致命，因为Agent的错误直接影响用户的真实资产。

二、个性化原理剖析

Agent 异常恢复需要区分四类错误，每类的处理策略完全不同：

1. 可重试错误：网络超时、API 暂时不可用——直接重试即可
2. 不可重试但可回滚错误：订单已创建但用户想取消——需要回滚操作
3. 部分执行不可回滚错误：上文场景——需要补偿操作
4. 不可恢复错误：用户账户被封禁——需要人工介入

在实际执行流程中，Agent 任务完成后系统会首先判断执行结果。若成功则直接完成任务；若失败，则根据上述错误分类进入相应的处理分支。对于可重试错误，系统采用指数退避策略进行重试，若超过最大次数则降级或转人工；对于可回滚错误，执行回滚操作，成功则取消任务并通知用户，失败则生成补偿任务；对于部分执行不可回滚错误，系统分析已执行操作，若可补偿则生成补偿操作，否则升级人工处理；对于不可恢复错误，记录详细上下文并通知用户及人工介入。生成的补偿任务与补偿操作最终都会进入补偿任务队列统一管理。

见证奇迹的时刻在于补偿操作的设计。对于"取消所有订阅"这类不可完全回滚的场景，系统需要：

1. 识别已执行的不可逆操作（如取消订阅 A、B）
2. 评估危害程度（核心服务订阅被取消，危害等级 P0）
3. 生成补偿操作（自动重新订阅，并在备注中说明这是"系统误操作恢复"）
4. 发送通知给用户确认

这套流程在后台自动执行，用户看到的是："系统检测到您的订阅可能被误操作，已自动恢复。请确认以下订阅状态是否符合您的预期。"——这比让用户自己去发现问题并联系客服要好得多。

三、个性化代码实践

from enum import Enum from dataclasses import dataclass, field from typing import List, Optional, Callable import asyncio class ErrorCategory(Enum): RETRYABLE = "retryable" # 可重试 ROLLBACKABLE = "rollbackable" # 可回滚 PARTIALLY_COMMITTED = "partial" # 部分执行 UNRECOVERABLE = "unrecoverable" # 不可恢复 class HarmLevel(Enum): P0_CRITICAL = 0 # 影响资产/安全 P1_MAJOR = 1 # 影响功能 P2_MINOR = 2 # 影响体验 @dataclass class AgentAction: action_id: str description: str execute: Callable rollback: Optional[Callable] = None # 回滚函数 is_idempotent: bool = False # 是否幂等 harm_level: HarmLevel = HarmLevel.P2_MINOR @dataclass class ActionRecord: action: AgentAction status: str # "pending" / "executed" / "rolled_back" result: Optional[dict] = None class AgentRecoveryEngine: """Agent 异常恢复引擎""" def __init__(self, notification_service): self.notify = notification_service self.max_retries = 3 async def execute_with_recovery( self, actions: List[AgentAction], user_id: str ): """执行Action序列，遇到失败时自动恢复""" executed: List[ActionRecord] = [] for i, action in enumerate(actions): try: result = await action.execute() executed.append(ActionRecord( action=action, status="executed", result=result )) except Exception as e: # 设计原因：失败时分析已执行的操作列表 # 这是整个恢复逻辑的入口 await self._recover(executed, action, e, user_id) raise # 恢复后仍抛出异常让上层感知 return executed async def _recover( self, executed: List[ActionRecord], failed_action: AgentAction, error: Exception, user_id: str ): """核心恢复逻辑""" category = self._classify_error(error, failed_action) if category == ErrorCategory.RETRYABLE: # 设计原因：重试前检查幂等性 # 非幂等操作（如创建订单）重试可能产生重复 if not failed_action.is_idempotent: await self._handle_partial_commit( executed, failed_action, user_id ) return # 设计原因：指数退避避免对下游造成冲击 await self._retry_with_backoff(failed_action) elif category == ErrorCategory.ROLLBACKABLE: # 设计原因：回滚已执行的操作（逆向顺序） # 假设Action1创建订单，Action2发送通知 # 如果Action2失败，需要先回滚通知（无意义），再回滚订单 rollback_errors = [] for record in reversed(executed): if record.action.rollback: try: await record.action.rollback() record.status = "rolled_back" except Exception as rb_error: rollback_errors.append(rb_error) if rollback_errors: # 设计原因：回滚本身失败了，需要生成补偿任务 await self._generate_compensation(executed, user_id) elif category == ErrorCategory.PARTIALLY_COMMITTED: await self._handle_partial_commit(executed, failed_action, user_id) elif category == ErrorCategory.UNRECOVERABLE: # 设计原因：不可恢复错误记录完整上下文并升级 await self._escalate_to_human( executed, failed_action, error, user_id ) async def _handle_partial_commit( self, executed: List[ActionRecord], failed_action: AgentAction, user_id: str ): """处理部分提交场景""" # 设计原因：评估已执行操作的危害等级 max_harm = max( (r.action.harm_level.value for r in executed), default=HarmLevel.P2_MINOR.value ) if max_harm <= HarmLevel.P1_MAJOR.value: # 设计原因：低危害操作自动补偿 compensation = self._build_compensation(executed) if compensation: await self.notify.send(user_id, { "type": "auto_recovery", "message": "检测到操作异常，已自动处理", "actions_taken": compensation, }) else: # 设计原因：高危害操作必须人工确认 await self._escalate_to_human( executed, failed_action, Exception("高危操作部分执行"), user_id ) def _classify_error( self, error: Exception, action: AgentAction ) -> ErrorCategory: """错误分类""" # 设计原因：基于错误类型和Action属性协同判断 if isinstance(error, (TimeoutError, ConnectionError)): return ErrorCategory.RETRYABLE if action.rollback is not None: return ErrorCategory.ROLLBACKABLE # 设计原因：有已执行操作但无法回滚 = 部分提交 return ErrorCategory.PARTIALLY_COMMITTED async def _retry_with_backoff(self, action: AgentAction): """带指数退避的重试""" for attempt in range(self.max_retries): try: delay = min(2 ** attempt, 30) # 1s, 2s, 4s, 8s, ..., max 30s await asyncio.sleep(delay) return await action.execute() except Exception: if attempt == self.max_retries - 1: raise

四、个性化边界权衡

回滚的假设前提：回滚操作的可靠性取决于下游API是否真正支持"取消"语义。很多SaaS API的"取消"操作实际上是"标记为取消"而非真正回退状态。设计Agent Action时，rollback函数的实现需要充分考虑下游的语义差异。

幂等性设计的成本：让每个Action都实现幂等（如用idempotency_key）会增加开发复杂度。最低要求是P0操作（支付、删除）必须幂等，P1/P2操作可以容错。一个实际的做法是在Action层面加一层"幂等性wrapper"：先查操作是否已执行，已执行则直接返回结果。

自动补偿 vs 人工介入的决策：自动补偿速度快、用户体验好，但可能做出错误判断。目前的原则：明确可逆的操作自动补偿，模糊操作人工确认，涉及金额的操作绝不自动补偿。

恢复过程的信息透明：用户应该被告知发生了什么以及系统做了什么。但"告诉太多"可能引起不必要的恐慌。通知文案需要在"告知足够信息"和"不过度惊吓"之间找平衡。

五、总结

Agent异常恢复架构将错误分为可重试、可回滚、部分执行、不可恢复四类，分别采用重试、逆向回滚、补偿操作、人工升级的策略。幂等性设计是重试安全的基础。部分执行且不可回滚的场景需要自动补偿+用户通知。涉及金额和安全的P0操作绝不自动补偿，必须人工介入。