diff --git a/weclone/core/inference/online_infer.py b/weclone/core/inference/online_infer.py index 7a3b12d..20047ff 100644 --- a/weclone/core/inference/online_infer.py +++ b/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, diff --git a/weclone/data/utils.py b/weclone/data/utils.py index 1f742ce..fc42527 100644 --- a/weclone/data/utils.py +++ b/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: """公开方法，用于描述单张图片内容""" diff --git a/weclone/utils/retry.py b/weclone/utils/retry.py new file mode 100644 index 0000000..5f766e2 --- /dev/null +++ b/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], +)