AI 应用的错误处理与重试策略

LLM API 在生产环境中面临多种失败场景：限流、超时、连接失败等。不同类型的错误需要不同的处理策略——盲目重试不仅无效，还会浪费配额、加剧服务器压力。本章系统介绍 LLM API 的错误分类、重试策略、超时配置、降级方案和熔断机制。

1.1 LLM API 的常见错误类型

AI应用错误处理决策树——按错误类型分类，匹配重试、降级、熔断、告警策略

明确错误类型是制定处理策略的前提。

RateLimitError（限流错误），最常见。API 提供商对每个账号都有请求频率限制，请求太快就会触发。OpenAI 的报错是 429，通常过一会儿就能重试。解法：等一段时间再重试，不要立刻重发。

APITimeoutError（超时），API 响应时间过长，客户端主动断开。LLM 生成 token 本来就慢，加上网络延迟，这个问题不少见。解法：设置合理的 timeout，超时后重试。

APIConnectionError（连接失败），网络层面的问题，TCP 连接建立失败或中断。通常是瞬时的网络抖动。解法：重试几次，多数情况下能恢复。

AuthenticationError（认证失败），API Key 无效或过期，返回 401。这种错误不应该重试——重试也没用，Key 本身有问题。解法：立刻报错，告警通知，人工介入。

InvalidRequestError（请求无效），请求参数本身有问题，最常见的是 token 超限（context window 满了）。返回 400。这种错误重试也没用，需要先处理请求本身。解法：不重试，截断输入或分批处理。

归纳一下：可重试的是 429、超时、连接失败；不可重试的是认证失败和请求参数错误。

1.2 重试策略：指数退避加随机抖动

固定间隔重试存在"雷群效应"（Thundering Herd）问题：大量请求同时失败后，如果以固定间隔重试，这些请求会在同一时刻再次涌入，可能再次打垮服务器，形成失败-重试-失败的恶性循环。

正确的方案是指数退避（Exponential Backoff，每次重试等待时间翻倍：1秒、2秒、4秒……避免所有请求同时重试）加随机抖动（Jitter，在等待时间上加随机偏移，让不同请求错开重试时间）：

第 1 次失败等 1 秒
第 2 次失败等 2 秒
第 3 次失败等 4 秒
每次等待时间再加一个随机的抖动值（比如 0-1 秒的随机数）

这样重试间隔越来越长，不同客户端的重试时间点也被打散，服务器压力分布均匀，能从故障中平稳恢复。

Python 中实现这个方案推荐使用 tenacity 库：

python

import os
import logging
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError

logger = logging.getLogger(__name__)

client = OpenAI(base_url="https://api.deepseek.com", api_key=os.getenv("DEEPSEEK_API_KEY"))

@retry(
    # 只对这三类错误重试，认证失败和请求无效不重试
    retry=retry_if_exception_type((RateLimitError, APITimeoutError, APIConnectionError)),
    # 最多重试 4 次（加上第一次共 5 次）
    stop=stop_after_attempt(4),
    # 指数退避：最少等 1 秒，最多等 30 秒，乘数 2
    wait=wait_exponential(multiplier=1, min=1, max=30),
    # 每次重试前打日志，方便排查
    before_sleep=before_sleep_log(logger, logging.WARNING),
    # 重试全部失败后重新抛出原始异常，而不是 tenacity 的异常
    reraise=True,
)
def call_llm_with_retry(messages: list, model: str = "deepseek-chat") -> str:
    """带重试的 LLM 调用封装"""
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        timeout=30,  # 单次调用超时 30 秒
    )
    return response.choices[0].message.content

wait_exponential 默认已经内置了随机抖动，不需要额外处理。

retry_if_exception_type 指定哪些异常才触发重试，这一条非常重要。如果不加限制，AuthenticationError 也会重试 4 次，毫无意义还浪费时间。

1.3 生产级 LLM API 错误处理完整模板

以下是经过生产验证的完整错误处理模板，涵盖所有常见故障场景：

python

import asyncio
import logging
import os
import time
from typing import Optional
from openai import AsyncOpenAI, RateLimitError, APIStatusError, APIConnectionError

logger = logging.getLogger(__name__)

class LLMAPIError(Exception):
    """LLM API 调用失败的统一异常"""
    def __init__(self, message: str, status_code: Optional[int] = None, retryable: bool = False):
        super().__init__(message)
        self.status_code = status_code
        self.retryable = retryable

async def call_llm_with_retry(
    client: AsyncOpenAI,
    messages: list,
    model: str = "deepseek-chat",
    max_tokens: int = 1024,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> str:
    """
    带完整错误处理和指数退避重试的 LLM 调用。

    错误分类：
    - 429 限流：可重试，等待更长时间
    - 500/529 服务器错误：可重试
    - 400 请求错误：不可重试（参数有问题）
    - 401 认证失败：不可重试（检查 API Key）
    - 超时：可重试
    """
    last_exception = None

    for attempt in range(max_retries):
        try:
            response = await asyncio.wait_for(
                client.chat.completions.create(
                    model=model,
                    max_tokens=max_tokens,
                    messages=messages,
                ),
                timeout=30.0  # 30秒超时
            )

            # 验证响应不为空
            if not response.choices:
                raise LLMAPIError("API 返回空响应", retryable=True)

            return response.choices[0].message.content

        except asyncio.TimeoutError:
            last_exception = LLMAPIError("API 调用超时（30s）", retryable=True)
            logger.warning(f"LLM 调用超时，第 {attempt + 1}/{max_retries} 次尝试")

        except RateLimitError as e:
            # 429：限流，需要等待更长时间
            last_exception = LLMAPIError(f"API 限流: {e}", status_code=429, retryable=True)
            wait_time = base_delay * (2 ** attempt) + 5  # 额外等5秒
            logger.warning(f"API 限流，等待 {wait_time:.1f}s 后重试")
            await asyncio.sleep(wait_time)
            continue

        except APIStatusError as e:
            if e.status_code in (500, 529):
                # 服务器错误，可重试
                last_exception = LLMAPIError(f"API 服务器错误: {e}", status_code=e.status_code, retryable=True)
                logger.warning(f"API 服务器错误 {e.status_code}，第 {attempt + 1}/{max_retries} 次尝试")
            elif e.status_code == 400:
                # 请求参数错误，不可重试
                logger.error(f"API 请求参数错误: {e}")
                raise LLMAPIError(f"请求参数错误: {e}", status_code=400, retryable=False)
            elif e.status_code == 401:
                # 认证失败，不可重试
                logger.error("API Key 无效或已过期")
                raise LLMAPIError("API Key 无效，请检查配置", status_code=401, retryable=False)
            else:
                last_exception = LLMAPIError(f"API 错误 {e.status_code}: {e}", status_code=e.status_code, retryable=True)

        except APIConnectionError as e:
            # 网络连接问题，可重试
            last_exception = LLMAPIError(f"网络连接失败: {e}", retryable=True)
            logger.warning(f"网络连接失败，第 {attempt + 1}/{max_retries} 次尝试")

        # 指数退避：1s, 2s, 4s...
        if attempt < max_retries - 1:
            wait_time = base_delay * (2 ** attempt)
            logger.info(f"等待 {wait_time}s 后重试...")
            await asyncio.sleep(wait_time)

    # 所有重试都失败
    logger.error(f"LLM 调用在 {max_retries} 次重试后失败: {last_exception}")
    raise last_exception or LLMAPIError("未知错误")


# 使用示例
async def main():
    client = AsyncOpenAI(
        base_url="https://api.deepseek.com",
        api_key=os.environ.get("DEEPSEEK_API_KEY"),
    )
    try:
        result = await call_llm_with_retry(
            client=client,
            messages=[{"role": "user", "content": "你好"}],
        )
        print(result)
    except LLMAPIError as e:
        if e.retryable:
            print(f"临时错误，请稍后重试: {e}")
        else:
            print(f"配置错误，请检查设置: {e}")

关键设计决策说明：

错误类型	HTTP状态码	是否重试	原因
限流	429	重试+额外等待	等待配额恢复
服务器错误	500/529	重试	临时故障
超时	-	重试	网络抖动
请求错误	400	不重试	参数有问题，重试无意义
认证失败	401	不重试	需要修复配置
空响应	-	重试	偶发异常

与前面 tenacity 方案的区别：这个模板直接控制每种错误的处理逻辑，不依赖第三方库，更适合对错误行为有精确要求的场景；tenacity 方案代码更简洁，适合快速集成。两种方式可以根据项目情况选择。

1.4 超时设置：两个维度

超时有两种语义，要分开处理。

连接超时：建立 TCP 连接的等待时间。如果服务器根本没响应，这个超时决定等待多久才放弃。通常设 5-10 秒够了。

读取超时：连接建立之后，等待服务器返回数据的时间。LLM 生成内容慢，这个值要设大一些，建议 30-60 秒。

python

from openai import OpenAI
import httpx

# 分别设置连接超时和读取超时
client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    timeout=httpx.Timeout(
        connect=5.0,    # 建立连接最多等 5 秒
        read=60.0,      # 等待响应最多等 60 秒
        write=5.0,
        pool=5.0,
    )
)

流式输出的超时是另一个需要注意的场景。流式输出时，第一个 token 可能要等 3-5 秒，但之后每个 token 很快。处理流式超时的思路是：分开控制"首个 token 等待超时"和"token 间隔超时"。

python

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(base_url="https://api.deepseek.com", api_key=os.getenv("DEEPSEEK_API_KEY"))

async def stream_with_timeout(prompt: str, first_token_timeout: float = 10.0):
    """带首 token 超时的流式输出"""
    stream = await async_client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        timeout=60.0,
    )

    first_token_received = False
    full_content = []

    try:
        async for chunk in asyncio.wait_for(
            _iterate_stream(stream, full_content),
            timeout=first_token_timeout if not first_token_received else None,
        ):
            first_token_received = True
            yield chunk
    except asyncio.TimeoutError:
        raise LLMAPIError("等待首个 token 超时", retryable=True)

async def _iterate_stream(stream, collector: list):
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            collector.append(delta)
            yield delta

1.5 降级策略：主模型挂了用备用模型

单一模型依赖是一个风险点。当主模型出现区域性故障时，如果没有备用方案，整个服务就停了。

设计思路：主模型失败时，降级到备用模型。备用模型通常是更小、更便宜的版本，或者来自不同提供商。

python

import os
from openai import OpenAI, APIError

openai_client = OpenAI(base_url="https://api.deepseek.com", api_key=os.getenv("DEEPSEEK_API_KEY"))
deepseek_client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

def call_with_fallback(user_message: str) -> dict:
    """
    主模型：GPT-4o
    降级一：GPT-4o-mini
    降级二：DeepSeek
    """
    strategies = [
        ("gpt-4o", _call_openai),
        ("gpt-4o-mini", _call_openai),
        ("deepseek-chat", _call_deepseek),
    ]

    last_error = None
    for model_name, call_fn in strategies:
        try:
            result = call_fn(user_message, model_name)
            return {"content": result, "model_used": model_name}
        except Exception as e:
            logger.warning(f"模型 {model_name} 失败: {type(e).__name__}: {e}")
            last_error = e
            continue

    # 全部失败，返回友好错误，不暴露技术细节
    logger.error(f"所有模型均失败，最后一个错误: {last_error}")
    return {
        "content": "服务暂时不可用，请稍后再试。",
        "model_used": None,
        "error": True,
    }

def _call_openai(message: str, model: str) -> str:
    response = openai_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        timeout=30,
    )
    return response.choices[0].message.content

def _call_deepseek(message: str, model: str) -> str:
    response = deepseek_client.chat.completions.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
        timeout=30,
    )
    return response.choices[0].message.content

对用户暴露的错误信息应使用友好提示（"服务暂时不可用"），而非技术细节（APIConnectionError: Connection refused to api.openai.com）。技术细节暴露给用户没有意义，还可能引发安全问题。

1.6 熔断器：防止雪崩

重试和降级处理的是单次失败。熔断（Circuit Breaker）处理的是持续性故障。

当 API 提供商持续返回 500 时，若每次请求都要等 30 秒超时再重试 4 次，大量并发请求会耗尽线程池，导致服务自身也挂掉。熔断器的逻辑是：

关闭状态（正常）：请求正常放行
开启状态（熔断）：连续失败 N 次后，直接拒绝后续请求，不再发出 API 调用
半开状态（探测）：隔一段时间放一个请求进去探测，如果成功则恢复正常，失败则继续熔断

python

import time
from threading import Lock
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 熔断中
    HALF_OPEN = "half_open" # 探测中

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,     # 连续失败几次触发熔断
        recovery_timeout: float = 60.0, # 熔断后多少秒开始探测
        success_threshold: int = 2,     # 探测成功几次才恢复正常
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self._lock = Lock()

    def call(self, func, *args, **kwargs):
        with self._lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                    logger.info("熔断器进入半开状态，开始探测")
                else:
                    raise RuntimeError("服务熔断中，请稍后重试")

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    logger.info("熔断器恢复正常状态")
            elif self.state == CircuitState.CLOSED:
                self.failure_count = 0

    def _on_failure(self):
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if (self.state in (CircuitState.CLOSED, CircuitState.HALF_OPEN)
                    and self.failure_count >= self.failure_threshold):
                self.state = CircuitState.OPEN
                logger.warning(f"熔断器开启，连续失败 {self.failure_count} 次")


# 使用示例
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

def safe_call_llm(message: str) -> str:
    try:
        return breaker.call(call_llm_with_retry, [{"role": "user", "content": message}])
    except RuntimeError as e:
        # 熔断状态，快速失败
        return "服务暂时不可用，请稍后再试。"

熔断器和重试配合使用：重试处理偶发错误，熔断处理持续故障。有了这两层，AI 应用在 API 出问题时能做到快速失败，而不是慢性等待。

1.7 上线前的验证清单

LLM API 本身比普通 HTTP 服务脆弱得多：响应慢、偶发超时、有限流配额、价格敏感。这些问题在生产环境中每天都会出现。

上线前至少要验证以下场景：

API Key 故障时用户看到的是什么
限流触发时用户等待多久
主模型挂了备用模型能否正常接管
连续故障时熔断器是否生效

在 staging 环境模拟这些场景，能在用户发现问题之前先行发现并修复。