Agent 异常恢复架构不是所有错误都能重试有些要回滚一、个性化深度引言周一早上看到告警群里的截图我后背一凉。用户对Agent说帮我取消所有订阅Agent理解为取消所有服务的订阅——包括正在使用的核心服务。它调了3个API取消了用户的全部6个订阅然后才意识到可能不对但这3个API中有2个不支持恢复。这是Agent异常恢复中最危险的一类错误操作已经部分执行无法简单重试也不能直接回滚。用户在线上遇到这种情况比模型回复不好更致命因为Agent的错误直接影响用户的真实资产。二、个性化原理剖析Agent 异常恢复需要区分四类错误每类的处理策略完全不同可重试错误网络超时、API 暂时不可用——直接重试即可不可重试但可回滚错误订单已创建但用户想取消——需要回滚操作部分执行不可回滚错误上文场景——需要补偿操作不可恢复错误用户账户被封禁——需要人工介入在实际执行流程中Agent 任务完成后系统会首先判断执行结果。若成功则直接完成任务若失败则根据上述错误分类进入相应的处理分支。对于可重试错误系统采用指数退避策略进行重试若超过最大次数则降级或转人工对于可回滚错误执行回滚操作成功则取消任务并通知用户失败则生成补偿任务对于部分执行不可回滚错误系统分析已执行操作若可补偿则生成补偿操作否则升级人工处理对于不可恢复错误记录详细上下文并通知用户及人工介入。生成的补偿任务与补偿操作最终都会进入补偿任务队列统一管理。见证奇迹的时刻在于补偿操作的设计。对于取消所有订阅这类不可完全回滚的场景系统需要识别已执行的不可逆操作如取消订阅 A、B评估危害程度核心服务订阅被取消危害等级 P0生成补偿操作自动重新订阅并在备注中说明这是系统误操作恢复发送通知给用户确认这套流程在后台自动执行用户看到的是系统检测到您的订阅可能被误操作已自动恢复。请确认以下订阅状态是否符合您的预期。——这比让用户自己去发现问题并联系客服要好得多。三、个性化代码实践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( actionaction, statusexecuted, resultresult )) 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), defaultHarmLevel.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操作绝不自动补偿必须人工介入。