大模型 API 异常处理与重试机制实战指南

在调用大模型 API 时，网络波动、服务限流或临时故障导致请求失败，是每个开发者都会遇到的日常。公有云大模型的 API 稳定性远低于传统的数据库或微服务，算力紧张的早高峰、模型服务商热更新时，响应延迟可能从几百毫秒飙升到数十秒，甚至直接抛出 502 错误。本文基于实际踩坑经验，手把手带你搭建一套可靠的异常处理与重试机制。

一、常见错误类型识别与场景解析

在动手写代码之前，先搞清楚哪些错误该重试、哪些不该重试。

HTTP 状态码维度：

状态码	含义	是否重试
429	请求过多，被限流	✅ 是（需配合退避）
500	内部服务器错误	✅ 是
502	网关错误	✅ 是
503	服务不可用	✅ 是
504	网关超时	✅ 是
400	客户端请求错误（参数问题）	❌ 否
401	未授权（Key 无效或过期）	❌ 否

异常类型维度：

• 网络层异常：ConnectionError（连接被拒绝）、ConnectTimeout（连接超时）、ReadTimeout（读取超时）——这些都是网络波动或服务端响应慢导致的，值得重试。
• 限流异常：RateLimitError（429）——服务端明确告诉你"请求太多了"，需要等待后再试。
• 服务端异常：APIError、InternalServerError（5xx）——服务端临时出问题，重试往往能解决。
• 客户端异常：BadRequestError（400）、AuthenticationError（401）——参数写错了或者 Key 不对，重试一万次也没用。

一个关键原则：只对"可恢复的临时性错误"进行重试。认证失败、参数错误这类"永久性失败"，重试只会浪费资源。

二、基础环境搭建与依赖安装

用虚拟环境隔离项目依赖，这是最基本的工程素养。

# 创建虚拟环境（Python 3.8+ 均可）
python -m venv llm-env

# 激活环境（Linux/Mac）
source llm-env/bin/activate
# 或（Windows）
llm-env\Scripts\activate

# 安装核心依赖
pip install openai requests tenacity python-dotenv

各依赖的作用：

• openai：OpenAI 官方 SDK，也兼容大多数国产模型（DeepSeek、通义千问等）
• requests：底层 HTTP 库，配合 urllib3 做更细粒度的重试控制
• tenacity：Python 最成熟的重试库，支持指数退避、异常过滤等
• python-dotenv：管理 API Key 等敏感配置，别把 Key 写死在代码里

创建 .env 文件存放密钥：

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.openai.com/v1

三、标准化异常捕获代码实现

先写一个最基础的调用函数，把异常捕获的架子搭好。

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

def call_llm(prompt: str, max_retries: int = 3) -> dict:
    """
    调用大模型 API，带基础异常捕获
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-4o",
        "messages": [{"role": "user", "content": prompt}]
    }
    
    for attempt in range(max_retries):
        try:
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()  # 4xx/5xx 会抛出 HTTPError
            return response.json()
            
        except requests.exceptions.Timeout as e:
            print(f"第 {attempt + 1} 次尝试超时: {e}")
            if attempt == max_retries - 1:
                raise
                
        except requests.exceptions.ConnectionError as e:
            print(f"第 {attempt + 1} 次尝试连接失败: {e}")
            if attempt == max_retries - 1:
                raise
                
        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code
            # 4xx 客户端错误不重试（除了 429）
            if 400 <= status_code < 500 and status_code != 429:
                raise  # 参数错误、认证错误，直接抛出
            print(f"第 {attempt + 1} 次尝试 HTTP {status_code}: {e}")
            if attempt == max_retries - 1:
                raise
                
        except Exception as e:
            print(f"第 {attempt + 1} 次尝试未知错误: {e}")
            if attempt == max_retries - 1:
                raise

这个实现的问题是：重试间隔固定，没有退避策略。如果服务端正忙，连续密集重试只会让情况更糟。下面我们用更专业的方案来解决。

四、指数退避重试策略详解

指数退避（Exponential Backoff）的核心思想很简单：每次重试的等待时间逐渐拉长——第1次等1秒，第2次等2秒，第3次等4秒，以此类推。

为什么要加抖动（Jitter）？

如果成百上千个客户端都在同一时间收到 429 错误，然后用相同的退避策略重试，它们会在同一时间再次涌向服务器——这就是"雪崩效应"。加入随机抖动后，每个客户端的重试时间略微错开，能有效分散压力。

推荐的退避公式：

delay = min(base_delay * (2 ^ attempt), max_delay)
# 加抖动后：
delay = random.uniform(0, delay)  # Full Jitter，AWS 推荐方案

用 tenacity 库可以轻松实现带抖动的指数退避：

from tenacity import (
    retry, 
    stop_after_attempt, 
    wait_exponential, 
    retry_if_exception_type,
    wait_random
)

@retry(
    stop=stop_after_attempt(3),  # 最多重试 3 次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 1s, 2s, 4s... 上限 10s
    retry=retry_if_exception_type(
        (requests.exceptions.Timeout, 
         requests.exceptions.ConnectionError,
         requests.exceptions.HTTPError)
    )
)
def call_llm_with_retry(prompt: str) -> dict:
    # ... 请求逻辑
    pass

tenacity 的 wait_exponential 默认不带抖动。如果需要加抖动，可以组合使用：

from tenacity import wait_exponential, wait_random

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10) + wait_random(0, 1),
    # 实际等待时间 = 指数退避时间 + 0~1 秒随机值
)
def call_llm_with_jitter(prompt: str) -> dict:
    pass

五、超时控制与请求取消机制

超时是必选项，不是可选项。 如果不设超时，一个卡住的 API 请求可能让你的服务线程永远挂起。

requests 库的超时分为两个维度：

response = requests.post(
    url,
    headers=headers,
    json=payload,
    timeout=(5, 30)  # (连接超时 5秒, 读取超时 30秒)
)

• 连接超时：客户端与服务器建立 TCP 连接的最大等待时间
• 读取超时：连接建立后，等待服务器返回数据的最大时间

对于大模型调用，建议：

• 连接超时：3-5 秒（网络不通就别等了）
• 读取超时：30-60 秒（大模型生成需要时间）
• 流式输出场景：首包超时单独控制

异步场景下的取消机制：

如果用 asyncio 做并发调用，可以通过 asyncio.timeout 控制超时，超时后主动取消任务：

import asyncio

async def call_llm_async(prompt: str):
    try:
        async with asyncio.timeout(30):
            # 实际的 API 调用
            return await client.chat.completions.create(...)
    except asyncio.TimeoutError:
        # 超时后自动取消任务
        raise

并发调用时需要注意：asyncio.gather 默认会在第一个异常时取消所有剩余任务。如果需要部分失败不影响其他任务，可以用 asyncio.gather(return_exceptions=True) 或 asyncio.Semaphore 控制并发数。

六、完整重试流程实战演示

把前面的知识点串起来，封装一个生产可用的调用函数：

import os
import requests
from dotenv import load_dotenv
from tenacity import (
    retry, 
    stop_after_attempt, 
    wait_exponential, 
    retry_if_exception,
    before_sleep_log
)
import logging

load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

def is_retryable_error(exception):
    """判断异常是否值得重试"""
    if isinstance(exception, requests.exceptions.Timeout):
        return True
    if isinstance(exception, requests.exceptions.ConnectionError):
        return True
    if isinstance(exception, requests.exceptions.HTTPError):
        # 429 和 5xx 重试，4xx（除429外）不重试
        status = exception.response.status_code
        return status == 429 or status >= 500
    return False

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception(is_retryable_error),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def call_llm_production(prompt: str, timeout: tuple = (5, 30)) -> dict:
    """
    生产环境调用大模型 API
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-4o",
        "messages": [{"role": "user", "content": prompt}]
    }
    
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload,
        timeout=timeout
    )
    response.raise_for_status()
    return response.json()

# 使用示例
if __name__ == "__main__":
    try:
        result = call_llm_production("介绍一下指数退避算法")
        print(result["choices"][0]["message"]["content"])
    except Exception as e:
        logger.error(f"API 调用最终失败: {e}")
        # 这里可以触发降级逻辑

关键设计点：

1. 重试条件精准控制：只对超时、连接错误、429 和 5xx 重试
2. 最大重试次数：3 次是经验值，避免无限循环
3. 重试前日志：before_sleep_log 会在每次重试前自动记录，方便排查
4. 超时分级：连接超时 5 秒，读取超时 30 秒

七、日志记录与故障排查技巧

日志是排查问题的命脉，但很多人写日志只记"调用了什么 API"，不记关键上下文。

每次 API 调用至少记录这些字段：

import time
import uuid
import logging

logger = logging.getLogger(__name__)

def call_with_logging(prompt: str):
    request_id = str(uuid.uuid4())[:8]
    start_time = time.time()
    
    logger.info(
        f"[{request_id}] 开始调用 LLM API",
        extra={
            "prompt_length": len(prompt),
            "model": "gpt-4o"
        }
    )
    
    try:
        result = call_llm_production(prompt)
        elapsed = time.time() - start_time
        tokens_used = result.get("usage", {}).get("total_tokens", 0)
        
        logger.info(
            f"[{request_id}] 调用成功",
            extra={
                "elapsed": elapsed,
                "tokens": tokens_used,
                "status": "success"
            }
        )
        return result
        
    except Exception as e:
        elapsed = time.time() - start_time
        logger.error(
            f"[{request_id}] 调用失败: {e}",
            extra={
                "elapsed": elapsed,
                "error_type": type(e).__name__,
                "status": "failed"
            }
        )
        raise

故障排查的几个实用技巧：

1. 分阶段记录：如果调用链路长（认证→参数校验→模型推理→返回），在每个阶段埋点，快速定位问题出在哪一环。
2. 记录原始请求和响应：出问题时可以复现，但注意脱敏——不要记录完整的用户输入输出到日志。
3. 记录 Retry-After 头：收到 429 时，服务端可能返回 Retry-After 头指明建议等待时间，记录下来有助于优化退避策略。
4. 关联请求 ID：用 uuid 给每次调用生成唯一 ID，方便在日志中串联整个调用链路。

八、生产环境稳定性优化建议

8.1 熔断器模式

重试机制处理的是"临时性故障"，但如果服务已经彻底挂了，继续重试只是在浪费资源。熔断器的思路是：当错误率达到阈值（比如最近 1 分钟失败率超过 50%），直接"熔断"——不再发起请求，快速失败，给服务端恢复的时间。

Python 中可以用 circuitbreaker 库或自己实现一个简单的滑动窗口计数器。

8.2 降级与 Fallback

当主模型不可用时，自动切换到备用模型。常见做法：

• 同级别降级：GPT-4 不行换 Claude
• 成本降级：大模型不行换小模型（如 GPT-4 → GPT-3.5）
• 本地兜底：所有云端模型都挂了，用本地部署的小参数模型提供基础服务

def call_with_fallback(prompt: str):
    models = ["gpt-4o", "claude-3-sonnet", "gpt-3.5-turbo"]
    for model in models:
        try:
            return call_model(model, prompt)
        except Exception:
            logger.warning(f"模型 {model} 不可用，尝试下一个")
            continue
    raise Exception("所有模型均不可用")

8.3 并发控制

大模型 API 通常有 QPM（每分钟请求数）和 TPM（每分钟 Token 数）限制。高并发场景下，用信号量控制并发数：

import asyncio

semaphore = asyncio.Semaphore(10)  # 最多 10 个并发

async def call_with_semaphore(prompt: str):
    async with semaphore:
        return await call_llm_async(prompt)

8.4 预算告警

大模型的成本是按 Token 计算的，一次不经意的循环调用或逻辑 Bug 可能让账单爆炸。务必：

• 在代码中通过 max_tokens 限制输出长度
• 对输入文本做截断，避免传入超长内容
• 在云服务平台设置预算告警阈值

8.5 配置外部化

把重试次数、超时时间、退避参数等放在配置文件或环境变量中，不要硬编码。这样线上出问题时可以动态调整，不用重新发布代码。

WEB项目地址：演示地址
安卓APP下载地址：演示地址

写在最后

大模型 API 调用从"能调通"到"调得稳"，中间隔着的是对异常处理的重视程度。服务中断在大模型生态里不是意外，而是常态。把重试、超时、熔断、降级这些基本功做好，你的服务才能在各种抖动中依然保持可用。

记住三个核心原则：

1. 只重试可恢复的错误——别在认证失败上浪费时间
2. 用指数退避 + 抖动——别让重试变成二次攻击
3. 设置上限——最大重试次数、超时时间、并发数，都得有天花板