feat(infer): add retry mechanism to API calls

Adds retry logic for OpenAI and image description API calls
to improve resilience against transient errors and rate
limiting. Implements exponential backoff with configurable
parameters.

weclone/core/inference/online_infer.py

@@ -1,5 +1,7 @@
 from openai import OpenAI
 
+from weclone.utils.retry import retry_openai_api
+
 
 class OnlineLLM:
     def __init__(self, api_key: str, base_url: str, model_name: str, default_system: str):
@@ -7,9 +9,10 @@ class OnlineLLM:
         self.base_url = base_url
         self.model_name = model_name
         self.default_system = default_system
-        self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)
+        self.client = OpenAI(api_key=self.api_key, base_url=self.base_url, max_retries=0)
 
     # TODO 需要做一个线程池进行并发
+    @retry_openai_api(max_retries=200, base_delay=30.0, max_delay=180.0)
     def chat(
         self,
         prompt_text,

weclone/data/utils.py

@@ -1,13 +1,13 @@
 import base64
 import concurrent.futures
 import os
-import time
 from pathlib import Path
 
 import requests
 
 from weclone.utils.config_models import WCMakeDatasetConfig
 from weclone.utils.log import logger
+from weclone.utils.retry import retry_on_http_error
 
 
 def check_image_file_exists(file_path: str) -> str | bool:
@@ -115,6 +115,14 @@ class ImageToTextProcessor:
             return "jpeg"
         return suffix
 
+    @retry_on_http_error(
+        max_retries=5,
+        base_delay=15.0,
+        max_delay=300.0,
+        backoff_factor=2.0,
+        retry_on_status=[429, 500, 502, 503, 504],
+        retry_on_exceptions=[requests.exceptions.RequestException, ConnectionError, TimeoutError],
+    )
     def _call_vision_api(self, image_path: str) -> str:
         """调用Vision API（增加了重试机制）"""
         base64_image = self._encode_image_to_base64(image_path)
@@ -143,49 +151,22 @@ class ImageToTextProcessor:
             "temperature": 0.1,
         }
 
-        # --- 重试逻辑 ---
-        max_retries = 5  # 最大重试次数
-        base_delay = 15  # 基础等待时间（秒）
+        response = requests.post(
+            f"{self.api_url}/chat/completions", headers=headers, json=payload, timeout=60
+        )
 
-        for attempt in range(max_retries):
-            try:
-                response = requests.post(
-                    f"{self.api_url}/chat/completions", headers=headers, json=payload, timeout=60
-                )
-                if response.status_code == 200:
-                    pass
-                elif response.status_code in [429, 500, 502, 503, 504]:
-                    response.raise_for_status()
-                else:
-                    logger.error(f"API请求失败，状态码: {response.status_code}，原因: {response.reason}")
-                    return "[图片描述获取失败]"
-
-                result = response.json()
-
-                if "choices" in result and len(result["choices"]) > 0:
-                    content = result["choices"][0]["message"]["content"]
-                    return content.strip()
-                else:
-                    logger.warning(f"API响应格式异常: {result}")
-                    return "[图片描述获取失败：API格式错误]"
-
-            except requests.exceptions.RequestException as e:
-                logger.warning(f"API请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
-                if attempt < max_retries - 1:
-                    # 指数退避等待
-                    wait_time = base_delay * (2**attempt)
-                    logger.info(f"将在 {wait_time} 秒后重试...")
-                    time.sleep(wait_time)
-                else:
-                    logger.error(f"API请求在 {max_retries} 次尝试后最终失败: {image_path}")
-                    return "[图片描述获取失败：请求异常]"
-            except Exception as e:
-                logger.error(f"处理API响应时出现未知错误 {image_path}: {e}")
-                # 对于未知错误，可以选择不重试，直接返回
-                return "[图片描述获取失败：未知错误]"
-
-        # 如果循环结束仍未成功（理论上不会执行到这里，因为上面已有返回）
-        return "[图片描述获取失败：所有重试均失败]"
+        if response.status_code == 200:
+            result = response.json()
+            if "choices" in result and len(result["choices"]) > 0:
+                content = result["choices"][0]["message"]["content"]
+                return content.strip()
+            else:
+                logger.warning(f"API响应格式异常: {result}")
+                return "[图片描述获取失败：API格式错误]"
+        else:
+            logger.error(f"API请求失败，状态码: {response.status_code}，原因: {response.reason}")
+            response.raise_for_status()  # 触发重试机制
+            return "[图片描述获取失败]"
 
     def describe_image(self, image_path: str) -> str:
         """公开方法，用于描述单张图片内容"""

weclone/utils/retry.py

@@ -0,0 +1,218 @@
+import random
+import time
+from functools import wraps
+from typing import Callable, List, Optional
+
+from weclone.utils.log import logger
+
+
+def retry_on_http_error(
+    max_retries: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 60.0,
+    backoff_factor: float = 2.0,
+    jitter: bool = True,
+    retry_on_status: Optional[List[int]] = None,
+    retry_on_exceptions: Optional[List[type]] = None,
+):
+    """
+    HTTP请求重试装饰器，专门处理429状态码和其他网络错误
+
+    Args:
+        max_retries: 最大重试次数
+        base_delay: 基础延迟时间（秒）
+        max_delay: 最大延迟时间（秒）
+        backoff_factor: 退避因子，每次重试延迟时间乘以此因子
+        jitter: 是否添加随机抖动，避免雷群效应
+        retry_on_status: 需要重试的HTTP状态码列表，默认包含429, 500, 502, 503, 504
+        retry_on_exceptions: 需要重试的异常类型列表
+    """
+    if retry_on_status is None:
+        retry_on_status = [429, 500, 502, 503, 504]
+
+    if retry_on_exceptions is None:
+        retry_on_exceptions = [ConnectionError, TimeoutError]
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            for attempt in range(max_retries + 1):
+                try:
+                    result = func(*args, **kwargs)
+
+                    # 检查是否是HTTP响应对象
+                    if hasattr(result, "status_code"):
+                        if result.status_code in retry_on_status:
+                            if attempt < max_retries:
+                                delay = _calculate_delay(
+                                    attempt, base_delay, max_delay, backoff_factor, jitter
+                                )
+                                logger.warning(
+                                    f"HTTP请求返回状态码 {result.status_code}，"
+                                    f"第 {attempt + 1}/{max_retries + 1} 次尝试，"
+                                    f"将在 {delay:.2f} 秒后重试..."
+                                )
+                                time.sleep(delay)
+                                continue
+                            else:
+                                logger.error(
+                                    f"HTTP请求在 {max_retries + 1} 次尝试后最终失败，状态码: {result.status_code}"
+                                )
+                                return result
+
+                    return result
+
+                except Exception as e:
+                    should_retry_on_exception = any(
+                        isinstance(e, exc_type) for exc_type in retry_on_exceptions
+                    )
+
+                    if should_retry_on_exception and attempt < max_retries:
+                        delay = _calculate_delay(attempt, base_delay, max_delay, backoff_factor, jitter)
+                        logger.warning(
+                            f"请求异常: {type(e).__name__}: {e}，"
+                            f"第 {attempt + 1}/{max_retries + 1} 次尝试，"
+                            f"将在 {delay:.2f} 秒后重试..."
+                        )
+                        time.sleep(delay)
+                        continue
+                    elif should_retry_on_exception:
+                        logger.error(f"请求在 {max_retries + 1} 次尝试后最终失败: {type(e).__name__}: {e}")
+                        raise
+                    else:
+                        logger.error(f"未知错误，不进行重试: {type(e).__name__}: {e}")
+                        raise
+
+            return None  # 理论上不会执行到这里
+
+        return wrapper
+
+    return decorator
+
+
+def retry_openai_api(
+    max_retries: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 60.0,
+    backoff_factor: float = 2.0,
+    jitter: bool = True,
+):
+    """
+    专门用于OpenAI API调用的重试装饰器
+    处理OpenAI特有的异常类型
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            for attempt in range(max_retries + 1):
+                try:
+                    return func(*args, **kwargs)
+
+                except Exception as e:
+                    # 检查是否是速率限制或临时错误
+                    error_message = str(e).lower()
+                    should_retry = (
+                        "rate limit" in error_message
+                        or "429" in error_message
+                        or "too many requests" in error_message
+                        or "server error" in error_message
+                        or "timeout" in error_message
+                        or "connection" in error_message
+                    )
+
+                    if should_retry and attempt < max_retries:
+                        delay = _calculate_delay(attempt, base_delay, max_delay, backoff_factor, jitter)
+                        logger.warning(
+                            f"OpenAI API调用失败: {type(e).__name__}: {e}，"
+                            f"第 {attempt + 1}/{max_retries + 1} 次尝试，"
+                            f"将在 {delay:.2f} 秒后重试..."
+                        )
+                        time.sleep(delay)
+                        continue
+                    else:
+                        if attempt >= max_retries:
+                            logger.error(
+                                f"OpenAI API调用在 {max_retries + 1} 次尝试后最终失败: {type(e).__name__}: {e}"
+                            )
+                        raise
+
+            return None
+
+        return wrapper
+
+    return decorator
+
+
+def _calculate_delay(
+    attempt: int, base_delay: float, max_delay: float, backoff_factor: float, jitter: bool
+) -> float:
+    """计算重试延迟时间"""
+    delay = base_delay * (backoff_factor**attempt)
+    delay = min(delay, max_delay)
+
+    if jitter:
+        # 添加±20%的随机抖动
+        jitter_range = delay * 0.2
+        delay += random.uniform(-jitter_range, jitter_range)
+        delay = max(0, delay)  # 确保延迟不为负数
+
+    return delay
+
+
+class RetryConfig:
+    """重试配置类，用于统一管理重试参数"""
+
+    def __init__(
+        self,
+        max_retries: int = 3,
+        base_delay: float = 1.0,
+        max_delay: float = 60.0,
+        backoff_factor: float = 2.0,
+        jitter: bool = True,
+        retry_on_status: Optional[List[int]] = None,
+        retry_on_exceptions: Optional[List[type]] = None,
+    ):
+        self.max_retries = max_retries
+        self.base_delay = base_delay
+        self.max_delay = max_delay
+        self.backoff_factor = backoff_factor
+        self.jitter = jitter
+        self.retry_on_status = retry_on_status or [429, 500, 502, 503, 504]
+        self.retry_on_exceptions = retry_on_exceptions or [ConnectionError, TimeoutError]
+
+    def apply_to_function(self, func: Callable) -> Callable:
+        """将重试配置应用到函数上"""
+        return retry_on_http_error(
+            max_retries=self.max_retries,
+            base_delay=self.base_delay,
+            max_delay=self.max_delay,
+            backoff_factor=self.backoff_factor,
+            jitter=self.jitter,
+            retry_on_status=self.retry_on_status,
+            retry_on_exceptions=self.retry_on_exceptions,
+        )(func)
+
+
+# 预定义的重试配置
+AGGRESSIVE_RETRY = RetryConfig(
+    max_retries=5,
+    base_delay=0.5,
+    max_delay=30.0,
+    backoff_factor=1.5,
+)
+
+CONSERVATIVE_RETRY = RetryConfig(
+    max_retries=2,
+    base_delay=2.0,
+    max_delay=10.0,
+    backoff_factor=2.0,
+)
+
+API_RETRY = RetryConfig(
+    max_retries=3,
+    base_delay=1.0,
+    max_delay=60.0,
+    backoff_factor=2.0,
+    retry_on_status=[429, 500, 502, 503, 504],
+)