Agent 重试策略设计失败重试时要带原因不能盲目循环一、重试不是简单的再跑一遍是要带着诊断信息定向修复AI Agent执行任务时失败是常态外部服务超时、API返回错误、数据格式异常、模型输出不符合约束。遇到失败后的第一反应通常是再试一次但盲目重试大概率会再次失败因为根因没有解决。一个超时的API调用重试十次还是超时一个格式错误的模型输出重试十次还是格式错误。有效的重试策略必须携带失败原因超时则缩短输入长度或换服务端点格式错误则修改Prompt约束数据异常则跳过或降级处理。把失败原因喂给下一次重试见证奇迹的时刻是第二次重试带着错误分析修改了Prompt后一次通过。盲目重试有三个隐性风险。第一是资源浪费每次重试消耗推理Token和计算时间十次盲目重试的成本可能超过一次正确的定向修复。第二是状态累积某些失败会改变系统状态如数据库部分写入盲目重试可能导致重复操作或状态不一致。第三是延迟放大用户等待的时间是所有重试时间的总和十次重试的延迟远超一次定向修复。二、重试链路从失败分析到定向修复的闭环设计flowchart TD A[任务执行] -- B{执行结果} B -- 成功 -- C[返回结果] B -- 失败 -- D[失败分析提取原因和类型] D -- E{失败类型判断} E -- 超时 -- F[缩短输入或换端点] E -- 格式错误 -- G[修改Prompt约束] E -- 数据异常 -- H[跳过或降级] E -- 服务不可用 -- I[换备选Agent或延迟重试] F -- J[定向重试带原因和修改策略] G -- J H -- J I -- J J -- K{重试次数} K -- 未超限 -- A K -- 已超限 -- L[上报失败带完整诊断链]重试策略的核心转变是从再试一次到分析失败原因→修改策略→定向重试。每次重试前Agent需要回答三个问题失败的具体原因是什么超时、格式、数据、服务下次重试需要做什么修改缩短输入、修改约束、跳过数据、换服务这个修改是否合理修改是否会引入新问题定向修复也需要验证。修改Prompt后重试成功可能不是因为修改解决了根因而是偶然因素如模型温度波动。工程上应该记录每次重试的修改策略和结果形成诊断链方便事后分析重试策略的有效性。三、定向重试策略带原因和修改策略的闭环重试机制下面是Agent定向重试策略的实现。代码注释解释了设计原则。from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional import time class FailureCategory(Enum): 失败类型分类不同类型对应不同修复策略 TIMEOUT timeout # 超时缩短输入或换端点 FORMAT_ERROR format_error # 格式错误修改Prompt约束 DATA_ANOMALY data_anomaly # 数据异常跳过或降级 SERVICE_DOWN service_down # 服务不可用换备选Agent RATE_LIMIT rate_limit # 限流延迟后重试 LOGIC_ERROR logic_error # 逻辑错误修改执行步骤 # 设计原因分类是定向修复的基础 # 不分类只能盲目重试 dataclass class RetryContext: 重试上下文携带失败原因和修复策略 attempt_number: int 1 failure_category: Optional[FailureCategory] None failure_reason: str # 失败的具体描述 # 设计原因reason要包含具体信息而非泛泛描述 # 如API响应超时输入长度8567 tokens超时阈值5000ms # 而不是超时了 repair_strategy: str # 本次重试采取的修复策略 # 例如将输入截断至4000 tokens previous_attempts: list[dict] field(default_factorylist) # 记录所有前序尝试的策略和结果 # 见证奇迹的时刻第二次重试带着错误分析定向修复后一次通过 dataclass class RetryResult: 重试结果成功或带诊断链的失败上报 success: bool final_output: Any None total_attempts: int 0 diagnostic_chain: list[dict] field(default_factorylist) # 设计原因diagnostic_chain记录每次重试的完整过程 # 包含失败原因、修复策略和结果 class DirectedRetryStrategy: 定向重试策略分析失败原因→选择修复策略→定向重试 def __init__(self, max_retries: int 3, base_delay: float 1.0, max_delay: float 30.0): self.max_retries max_retries self.base_delay base_delay self.max_delay max_delay def analyze_failure(self, error: Exception, context: dict) - tuple[FailureCategory, str]: 失败分析提取类型和具体原因 error_str str(error) if timeout in error_str.lower() or timed out in error_str.lower(): input_len context.get(input_length, unknown) threshold context.get(timeout_ms, unknown) reason f超时输入长度{input_len}超时阈值{threshold}ms return FailureCategory.TIMEOUT, reason elif format in error_str.lower() or schema in error_str.lower(): expected context.get(expected_format, unknown) actual context.get(actual_format, unknown) reason f格式错误期望{expected}实际{actual} return FailureCategory.FORMAT_ERROR, reason elif 429 in error_str or rate limit in error_str.lower(): reason f限流错误{error_str} return FailureCategory.RATE_LIMIT, reason else: reason f未知错误{error_str} return FailureCategory.LOGIC_ERROR, reason def select_repair_strategy(self, category: FailureCategory, reason: str, context: RetryContext) - str: 修复策略选择不同失败类型对应不同修复动作 strategies { FailureCategory.TIMEOUT: # 设计原因超时修复是缩短输入而非等待更长时间 # 因为根因是输入太长而非时间不够 f将输入截断至当前长度的60%当前输入约{context.attempt_number * 0.6:.0f}倍, FailureCategory.FORMAT_ERROR: 修改Prompt增加格式约束和示例输出, FailureCategory.DATA_ANOMALY: 跳过异常数据项使用降级逻辑处理剩余数据, FailureCategory.SERVICE_DOWN: 切换到备选Agent或服务端点, FailureCategory.RATE_LIMIT: f延迟{self._compute_delay(context.attempt_number)}秒后重试, FailureCategory.LOGIC_ERROR: 修改执行步骤顺序增加中间校验, } return strategies.get(category, 无可用修复策略) def _compute_delay(self, attempt: int) - float: 指数退避延迟防止限流场景下密集重试 设计原因线性退避太慢指数退避在初始阶段快速试错 后续阶段给服务恢复留足时间 delay self.base_delay * (2 ** (attempt - 1)) return min(delay, self.max_delay) def should_retry(self, context: RetryContext) - bool: 判断是否应该继续重试超限则上报失败 if context.attempt_number self.max_retries: return False # 设计原因同一修复策略连续失败两次说明策略无效 # 应换策略而非用同一策略继续重试 if len(context.previous_attempts) 2: last_two_strategies [ a.get(strategy, ) for a in context.previous_attempts[-2:] ] if last_two_strategies[0] last_two_strategies[1]: return False # 同策略连续失败不再重试 return True四、重试权衡重试次数、修复策略多样性和延迟放大的三角约束定向重试策略的工程权衡有四个维度。第一是重试次数vs资源消耗每次重试消耗推理Token和时间大模型Agent一次推理成本可能几十美元3次重试就翻了3倍。重试上限应根据任务价值设定高价值任务核心业务流程允许5次重试低价值任务辅助信息查询限制2次。第二是修复策略多样性vs策略质量修复策略越多覆盖的失败类型越全但每个策略都需要验证有效性。未验证的策略可能比盲目重试更糟糕——比如超时场景下换备选Agent策略如果备选Agent也有超时问题重试只会浪费更多时间。第三是延迟放大vs用户体验重试次数和延迟成正比3次重试加上指数退避可能让用户等待30秒以上。实时交互场景聊天、语音的延迟上限是2-3秒超出后应直接告知用户失败原因而非继续重试。异步任务批量处理、数据分析可以允许更长重试延迟。第四是诊断链完整性vs日志成本每次重试记录详细的失败原因、修复策略和结果方便事后分析。但诊断链数据可能占用大量存储空间尤其是大批量任务场景。工程折中是保留最近3次重试的完整诊断链更早的记录只保留摘要。同策略连续失败需要特殊处理。如果同一修复策略连续两次失败说明策略本身有问题应切换到不同策略或直接上报失败。盲目坚持同一策略是重试策略中最常见的工程错误。五、总结Agent重试策略应从盲目循环转向定向修复每次重试携带失败原因和修复策略按失败类型超时、格式、数据、服务、限流、逻辑选择不同修复动作。超时修复应缩短输入而非增加等待时间限流重试使用指数退避。同一修复策略连续失败两次应切换策略或上报失败。重试次数按任务价值设定上限实时场景延迟超限应直接告知用户失败异步任务允许更长重试延迟诊断链保留最近3次完整记录。