1.将错误分为可重试错误与不可重试错误2类型特征例子可重试Transient临时性、随时间可能恢复网络超时、限流429、服务暂时不可用503不可重试Permanent本质性问题重试没用认证失败401、参数错误400、资源不存在4043异常继承体系设计class AgentError(Exception): 所有 Agent 相关错误的基类 pass class RetryableError(AgentError): 标记为可重试的错误基类 pass class NonRetryableError(AgentError): 标记为不可重试的错误基类 pass # ---- 可重试错误 ---- class RateLimitError(RetryableError): 触发限流等待后可以重试 def __init__(self, message: str, retry_after: float None): super().__init__(message) self.retry_after retry_after # 服务端建议的等待时间 class TimeoutError(RetryableError): 请求超时网络问题可能是临时的 pass class ServiceUnavailableError(RetryableError): 服务暂时不可用503 pass # ---- 不可重试错误 ---- class AuthenticationError(NonRetryableError): 认证失败重试无意义 pass class InvalidRequestError(NonRetryableError): 请求参数错误重试无意义 pass class ResourceNotFoundError(NonRetryableError): 资源不存在404 pass4 升级版retry代码import asyncio import functools import logging import random from typing import Callable, TypeVar logger logging.getLogger(__name__) T TypeVar(T) def retry( max_attempts: int 3, base_delay: float 1.0, max_delay: float 30.0, exponential_base: float 2.0, jitter: bool True, ): 智能重试装饰器 - 只重试 RetryableError 及其子类 - NonRetryableError 直接抛出不浪费时间 - 指数退避 随机抖动避免雪崩 def decorator(func: Callable[..., T]) - Callable[..., T]: functools.wraps(func) async def wrapper(*args, **kwargs) - T: last_exception None for attempt in range(1, max_attempts 1): try: return await func(*args, **kwargs) except NonRetryableError: # 不可重试错误立刻抛出 logger.error( f{func.__name__} failed with non-retryable error, fgiving up immediately ) raise except RetryableError as e: last_exception e if attempt max_attempts: logger.error( f{func.__name__} failed after {max_attempts} attempts ) raise delay _compute_delay( e, attempt, base_delay, max_delay, exponential_base, jitter ) logger.warning( f{func.__name__} attempt {attempt} failed f({type(e).__name__}), retrying in {delay:.2f}s ) await asyncio.sleep(delay) raise last_exception return wrapper return decorator def _compute_delay( error: RetryableError, attempt: int, base_delay: float, max_delay: float, exponential_base: float, jitter: bool, ) - float: # 如果服务端明确告知等待时间优先用它 if isinstance(error, RateLimitError) and error.retry_after: return min(error.retry_after, max_delay) # 否则用指数退避 delay min(base_delay * (exponential_base ** (attempt - 1)), max_delay) if jitter: delay delay * (0.5 random.random()) # 50%~150% 随机抖动 return delay
