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 库：

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 错误处理完整模板

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

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 秒。

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 间隔超时"。

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 降级策略：主模型挂了用备用模型

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

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

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 调用
• 半开状态（探测）：隔一段时间放一个请求进去探测，如果成功则恢复正常，失败则继续熔断

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 环境模拟这些场景，能在用户发现问题之前先行发现并修复。