1.将错误分为可重试错误与不可重试错误

2

3异常继承体系设计

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）""" pass

4 升级版@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, " f"giving 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