Python 如何给 AI API 实现自动重试与退避策略:应对临时网络抖动与限流

import timefrom openai import OpenAIclient = OpenAI(api_key="your-key", base_url="https://api.example.com/v1")def ask_with_simple_retry(prompt: str):    for i in range(3):        try:            return client.chat.completions.create(                model="gpt-4o-mini",                messages=[{"role": "user", "content": prompt}]            )        except Exception as e:            print(f"出错啦: {e}，正在重试...")            time.sleep(1) # 每次固定休息 1 秒    raise RuntimeError("重试 3 次后依然失败")

import timeimport randomimport loggingfrom openai import (    APIConnectionError,     APITimeoutError,     RateLimitError,     InternalServerError)# 定义哪些异常属于“可以重试”的临时故障RETRYABLE_EXCEPTIONS = (    APIConnectionError,    APITimeoutError,    RateLimitError,    InternalServerError,)def retry_with_exponential_backoff(    max_retries: int = 4,    base_delay: float = 2.0,    max_delay: float = 60.0,):    """    带指数退避和随机抖动的重试装饰器    """    def decorator(func):        def wrapper(*args, **kwargs):            retries = 0            while True:                try:                    return func(*args, **kwargs)                except RETRYABLE_EXCEPTIONS as e:                    retries += 1                    if retries > max_retries:                        logging.error(f"[重试耗尽] 已达到最大重试次数 {max_retries}，操作终止。")                        raise e                    # 计算指数退避时间: base_delay * (2 ^ (retries - 1))                    sleep_time = base_delay * (2 ** (retries - 1))                    # 加上 ±25% 的随机抖动                    jitter = sleep_time * 0.25 * (random.random() * 2 - 1)                    actual_sleep = min(max_delay, sleep_time + jitter)                    logging.warning(                        f"[临时故障] 捕获异常: {type(e).__name__}. "                        f"将在 {actual_sleep:.2f 秒后进行第 {retries}/{max_retries} 次重试..."                    )                    time.sleep(actual_sleep)                except Exception as e:                    # 对于 400 Bad Request 等无法恢复的错误，直接抛出，不进行重试                    logging.error(f"[不可恢复异常] 遇到致命错误，停止重试: {e}")                    raise e        return wrapper    return decorator

@retry_with_exponential_backoff(max_retries=3, base_delay=1.5)def safe_call_llm(prompt: str) -> str:    client = OpenAI(        api_key="your-api-key",        base_url="https://your-api-domain.com/v1",        timeout=15.0    )    response = client.chat.completions.create(        model="your-model-name",        messages=[{"role": "user", "content": prompt}]    )    return response.choices[0].message.content# 运行测试if __name__ == "__main__":    try:        result = safe_call_llm("你好，测试自动重试")        print("调用成功:", result)    except Exception as exc:        print("最终调用失败:", exc)

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typefrom openai import OpenAI, APIConnectionError, APITimeoutError, RateLimitError# 配置重试规则：# 1. 遇到指定网络/限流异常才重试# 2. 最多尝试 4 次# 3. 指数退避：初始等待 2 秒，最高等待 30 秒，包含随机抖动@retry(    retry=retry_if_exception_type((APIConnectionError, APITimeoutError, RateLimitError)),    stop=stop_after_attempt(4),    wait=wait_exponential(multiplier=1, min=2, max=30),    reraise=True)def call_api_with_tenacity(prompt: str):    client = OpenAI(        api_key="your-api-key",        base_url="https://your-api-domain.com/v1"    )    res = client.chat.completions.create(        model="your-model-name",        messages=[{"role": "user", "content": prompt}]    )    return res.choices[0].message.content