| """
|
| API调用重试机制
|
| 用于处理LLM等外部API调用的重试逻辑
|
| """
|
|
|
| import time
|
| import random
|
| import functools
|
| from typing import Callable, Any, Optional, Type, Tuple
|
| from ..utils.logger import get_logger
|
|
|
| logger = get_logger('mirofish.retry')
|
|
|
|
|
| def retry_with_backoff(
|
| max_retries: int = 3,
|
| initial_delay: float = 1.0,
|
| max_delay: float = 30.0,
|
| backoff_factor: float = 2.0,
|
| jitter: bool = True,
|
| exceptions: Tuple[Type[Exception], ...] = (Exception,),
|
| on_retry: Optional[Callable[[Exception, int], None]] = None
|
| ):
|
| """
|
| 带指数退避的重试装饰器
|
|
|
| Args:
|
| max_retries: 最大重试次数
|
| initial_delay: 初始延迟（秒）
|
| max_delay: 最大延迟（秒）
|
| backoff_factor: 退避因子
|
| jitter: 是否添加随机抖动
|
| exceptions: 需要重试的异常类型
|
| on_retry: 重试时的回调函数 (exception, retry_count)
|
|
|
| Usage:
|
| @retry_with_backoff(max_retries=3)
|
| def call_llm_api():
|
| ...
|
| """
|
| def decorator(func: Callable) -> Callable:
|
| @functools.wraps(func)
|
| def wrapper(*args, **kwargs) -> Any:
|
| last_exception = None
|
| delay = initial_delay
|
|
|
| for attempt in range(max_retries + 1):
|
| try:
|
| return func(*args, **kwargs)
|
|
|
| except exceptions as e:
|
| last_exception = e
|
|
|
| if attempt == max_retries:
|
| logger.error(f"函数 {func.__name__} 在 {max_retries} 次重试后仍失败: {str(e)}")
|
| raise
|
|
|
|
|
| current_delay = min(delay, max_delay)
|
| if jitter:
|
| current_delay = current_delay * (0.5 + random.random())
|
|
|
| logger.warning(
|
| f"函数 {func.__name__} 第 {attempt + 1} 次尝试失败: {str(e)}, "
|
| f"{current_delay:.1f}秒后重试..."
|
| )
|
|
|
| if on_retry:
|
| on_retry(e, attempt + 1)
|
|
|
| time.sleep(current_delay)
|
| delay *= backoff_factor
|
|
|
| raise last_exception
|
|
|
| return wrapper
|
| return decorator
|
|
|
|
|
| def retry_with_backoff_async(
|
| max_retries: int = 3,
|
| initial_delay: float = 1.0,
|
| max_delay: float = 30.0,
|
| backoff_factor: float = 2.0,
|
| jitter: bool = True,
|
| exceptions: Tuple[Type[Exception], ...] = (Exception,),
|
| on_retry: Optional[Callable[[Exception, int], None]] = None
|
| ):
|
| """
|
| 异步版本的重试装饰器
|
| """
|
| import asyncio
|
|
|
| def decorator(func: Callable) -> Callable:
|
| @functools.wraps(func)
|
| async def wrapper(*args, **kwargs) -> Any:
|
| last_exception = None
|
| delay = initial_delay
|
|
|
| for attempt in range(max_retries + 1):
|
| try:
|
| return await func(*args, **kwargs)
|
|
|
| except exceptions as e:
|
| last_exception = e
|
|
|
| if attempt == max_retries:
|
| logger.error(f"异步函数 {func.__name__} 在 {max_retries} 次重试后仍失败: {str(e)}")
|
| raise
|
|
|
| current_delay = min(delay, max_delay)
|
| if jitter:
|
| current_delay = current_delay * (0.5 + random.random())
|
|
|
| logger.warning(
|
| f"异步函数 {func.__name__} 第 {attempt + 1} 次尝试失败: {str(e)}, "
|
| f"{current_delay:.1f}秒后重试..."
|
| )
|
|
|
| if on_retry:
|
| on_retry(e, attempt + 1)
|
|
|
| await asyncio.sleep(current_delay)
|
| delay *= backoff_factor
|
|
|
| raise last_exception
|
|
|
| return wrapper
|
| return decorator
|
|
|
|
|
| class RetryableAPIClient:
|
| """
|
| 可重试的API客户端封装
|
| """
|
|
|
| def __init__(
|
| self,
|
| max_retries: int = 3,
|
| initial_delay: float = 1.0,
|
| max_delay: float = 30.0,
|
| backoff_factor: float = 2.0
|
| ):
|
| self.max_retries = max_retries
|
| self.initial_delay = initial_delay
|
| self.max_delay = max_delay
|
| self.backoff_factor = backoff_factor
|
|
|
| def call_with_retry(
|
| self,
|
| func: Callable,
|
| *args,
|
| exceptions: Tuple[Type[Exception], ...] = (Exception,),
|
| **kwargs
|
| ) -> Any:
|
| """
|
| 执行函数调用并在失败时重试
|
|
|
| Args:
|
| func: 要调用的函数
|
| *args: 函数参数
|
| exceptions: 需要重试的异常类型
|
| **kwargs: 函数关键字参数
|
|
|
| Returns:
|
| 函数返回值
|
| """
|
| last_exception = None
|
| delay = self.initial_delay
|
|
|
| for attempt in range(self.max_retries + 1):
|
| try:
|
| return func(*args, **kwargs)
|
|
|
| except exceptions as e:
|
| last_exception = e
|
|
|
| if attempt == self.max_retries:
|
| logger.error(f"API调用在 {self.max_retries} 次重试后仍失败: {str(e)}")
|
| raise
|
|
|
| current_delay = min(delay, self.max_delay)
|
| current_delay = current_delay * (0.5 + random.random())
|
|
|
| logger.warning(
|
| f"API调用第 {attempt + 1} 次尝试失败: {str(e)}, "
|
| f"{current_delay:.1f}秒后重试..."
|
| )
|
|
|
| time.sleep(current_delay)
|
| delay *= self.backoff_factor
|
|
|
| raise last_exception
|
|
|
| def call_batch_with_retry(
|
| self,
|
| items: list,
|
| process_func: Callable,
|
| exceptions: Tuple[Type[Exception], ...] = (Exception,),
|
| continue_on_failure: bool = True
|
| ) -> Tuple[list, list]:
|
| """
|
| 批量调用并对每个失败项单独重试
|
|
|
| Args:
|
| items: 要处理的项目列表
|
| process_func: 处理函数，接收单个item作为参数
|
| exceptions: 需要重试的异常类型
|
| continue_on_failure: 单项失败后是否继续处理其他项
|
|
|
| Returns:
|
| (成功结果列表, 失败项列表)
|
| """
|
| results = []
|
| failures = []
|
|
|
| for idx, item in enumerate(items):
|
| try:
|
| result = self.call_with_retry(
|
| process_func,
|
| item,
|
| exceptions=exceptions
|
| )
|
| results.append(result)
|
|
|
| except Exception as e:
|
| logger.error(f"处理第 {idx + 1} 项失败: {str(e)}")
|
| failures.append({
|
| "index": idx,
|
| "item": item,
|
| "error": str(e)
|
| })
|
|
|
| if not continue_on_failure:
|
| raise
|
|
|
| return results, failures
|
|
|
|
|
