Prompts-MCP 686 skills · ai / design / design-pattern / framework / fundamentals / habit / lang / tech-selection /mcp/sse
framework framework/fastapi/llm/error-fallback.md medium

fastapi-llm-error-fallback

LLM 降级与重试 — 主 provider 失败切备用 + 指数退避。Use when 写 Python 后端代码 / 评审涉及 `error-fallback` 的 PR。

降级fallbackretrybackoff
paths
  • backend/services/**/*.py
  • py/services/**/*.py

LLM · 降级与重试

策略

 (qwen-plus)
   失败
重试 指数退避最多 2 
   仍失败
切备 (gpt-4o-mini)
   失败
重试 最多 1 
   仍失败
通知前端暂时不可用

实现模板

# backend/services/llm_client.py
import asyncio
from typing import AsyncGenerator
from core.config import llm_settings

async def call_with_fallback(
    prompt: str,
    *,
    max_retries: int = 2,
) -> AsyncGenerator[str, None]:
    """主 → 备的流式降级"""

    # 1. 尝试主
    try:
        async for chunk in _call_with_retry("qwen", prompt, max_retries):
            yield chunk
        return
    except Exception as e:
        logger.warning(f"主 provider 失败,切备用: {e}")

    if not llm_settings.fallback_enabled:
        raise ApiException(msg="LLM 主服务不可用", code=503)

    # 2. 切备
    try:
        async for chunk in _call_with_retry("openai", prompt, max_retries=1):
            yield chunk
    except Exception as e:
        logger.exception(f"备用 provider 也失败: {e}")
        raise ApiException(msg="生成服务暂时不可用,请稍后重试", code=503)


async def _call_with_retry(
    provider: str,
    prompt: str,
    max_retries: int,
) -> AsyncGenerator[str, None]:
    """单 provider 指数退避重试"""
    for attempt in range(max_retries + 1):
        try:
            llm = get_llm(provider)
            async for chunk in llm.astream(prompt):
                yield chunk
            return
        except (TimeoutError, ConnectionError, RateLimitError) as e:
            if attempt >= max_retries:
                raise
            wait = 2 ** attempt
            logger.warning(f"{provider} 重试 {attempt + 1}/{max_retries}{wait}s: {e}")
            await asyncio.sleep(wait)

不重试的异常

异常 重试?
TimeoutError
ConnectionError
RateLimitError ✅(按 retry-after 等待)
APIError(5xx)
AuthenticationError ❌(凭证错)
InvalidRequestError(4xx 非 429) ❌(参数错)
ContentFilterError ❌(违规)

何时不降级

  • 用户明确指定 provider("用 OpenAI 生成")→ 不切换
  • 已扣过积分的二次重试 → 通知用户失败而非自动重生成
  • 流式生成中途切换 → 实现复杂,不切换(让用户主动重试)

自检

  • [ ] 主 / 备清晰?
  • [ ] 重试用指数退避?
  • [ ] 区分可重试和不可重试的异常?
  • [ ] 降级前提示用户(前端用 warning toast)?
  • [ ] 最终失败 SSE error 帧通知前端?

相关