从 3 次超时到 0 人工介入：LLM 网关故障转移实战

问题背景：多厂商模型混用为什么不能只写 SDK 重试

在智能体工作流进入生产后，团队通常不会只绑定一家模型供应商：主力模型负责复杂推理，低成本模型处理摘要与分类，视觉模型处理多模态输入，某些场景还要接入私有部署模型。多厂商混用能降低成本、提升能力覆盖，但也把故障面从“应用是否可用”变成了“当前请求应走哪条模型链路”。

常见故障包括：供应商返回 429 限流、区域网络抖动、流式响应在首 token 后中断、模型版本行为漂移、长上下文请求超时，以及账号余额或配额异常。如果业务代码直接散落地调用各家 SDK，即使每个调用点都写了 retry，也很难统一回答三个问题：哪些错误值得重试、何时切换供应商、降级后如何保证输出契约仍可被工作流消费。

因此，需要在业务服务与模型供应商之间引入统一 LLM 网关。它不是简单的反向代理，而是负责协议归一化、多模型路由、密钥管理、配额隔离、成本统计、语义缓存、观测审计，以及本文重点讨论的 failover 与降级。

方案对比：SDK 重试、供应商高可用与统一网关

第一种方案是在业务代码中直接重试。优点是上线快，缺点是策略分散，无法跨进程共享熔断状态；同一个供应商已经全局不可用时，每个实例仍可能继续放大请求。

第二种方案是依赖单一供应商提供的多区域或多密钥高可用。它能解决部分基础设施问题，但无法覆盖跨厂商能力差异、成本切换、私有模型兜底，以及供应商级账号风险。

第三种方案是建设统一 LLM 网关：业务只面向统一的 Chat Completions 风格协议，网关根据任务类型、上下文长度、成本预算、实时健康度选择模型；遇到可转移错误时，在主模型、备用模型、低成本模型之间切换。该方案的代价是需要设计路由元数据、幂等策略、流式处理和可观测性，但它最适合多模型、多团队、多环境的长期架构。

架构设计：把路由、健康度与故障转移拆开

1. 路由策略不要只按模型名称选择

生产路由至少要包含四类信息：任务能力、输入约束、业务优先级和实时健康度。例如，代码生成要求工具调用稳定，摘要任务允许使用便宜模型；超长上下文不能路由到上下文窗口不足的模型；客服实时回复对首 token 延迟敏感，离线评测则可以排队等待更强模型。

建议把路由配置声明化，而不是写死在代码里。每个 target 声明供应商、模型、能力标签、上下文上限、超时、成本权重和降级链。网关根据请求的 task_type 选择候选集，再结合熔断器状态排序。

2. 区分可转移错误与不可转移错误

并不是所有失败都应该 failover。401、403、签名错误、请求体格式错误、内容被安全策略拒绝，通常重试或换厂商也不会成功；429、500、502、503、连接超时、上游网络中断，则适合切换。还要注意 429 可能是某个密钥限流，也可能是整账号配额耗尽，因此维度应区分 provider、model、api_key 与区域。

对于流式响应，错误可能发生在 HTTP 建立前，也可能发生在输出若干 chunk 后。前者可以透明切换；后者不能简单重放，因为用户可能已经看到文本。更稳妥的做法是：面向用户的流式请求设置较短的首 token 等待窗口，只有在首 token 前失败才切换；首 token 后失败则中断并标记事件，由上层决定是否提示“续写”或用非流式修复。

3. 熔断比无限重试更重要

重试风暴是 LLM 网关最容易制造的二次事故。熔断器应在滑动窗口内统计失败率、平均延迟、限流次数和配额错误，状态分为 closed、open、half_open。closed 正常放行；open 直接跳过该目标；half_open 只放少量探测请求，成功后恢复。

降级链也要设置总预算，例如总超时 12 秒、最多 2 次切换、同供应商最多重试 1 次。否则“主模型超时、备用模型排队、兜底模型再超时”会让用户等待更久。

可运行 Python 示例：实现最小可用 LLM 网关

下面示例不依赖第三方库，使用 Python 标准库模拟多供应商调用、熔断、路由与故障转移。你可以把 fake_call_llm 替换为真实 httpx 异步请求。

import asyncio
import random
import time
from dataclasses import dataclass, field
from typing import AsyncIterator


class LLMError(Exception):
    """模型调用基础异常。"""

class RateLimitError(LLMError):
    """可转移：限流。"""

class UpstreamTimeout(LLMError):
    """可转移：上游超时。"""

class AuthError(LLMError):
    """不可转移：密钥或权限错误。"""

class BadRequestError(LLMError):
    """不可转移：请求参数错误。"""


RETRIABLE_ERRORS = (RateLimitError, UpstreamTimeout)


@dataclass
class ModelTarget:
    provider: str
    model: str
    capabilities: set[str]
    max_context: int
    timeout: float
    cost_weight: float


@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    cooldown_seconds: float = 20.0
    failures: int = 0
    opened_at: float | None = None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if time.monotonic() - self.opened_at >= self.cooldown_seconds:
            # half-open：允许一次探测；真实系统应增加状态锁
            return True
        return False

    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None

    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.opened_at = time.monotonic()


# 实际项目中建议用 Redis 维护共享熔断状态；本地演示使用内存字典
BREAKERS: dict[str, CircuitBreaker] = {}

ROUTES: dict[str, list[ModelTarget]] = {
    "agent": [
        ModelTarget("vendor-a", "a-reasoner", {"tools", "reasoning"}, 128000, 4.0, 1.0),
        ModelTarget("vendor-b", "b-fast-reason", {"tools", "reasoning"}, 96000, 3.0, 0.8),
        ModelTarget("local", "local-safe", {"chat"}, 32000, 2.0, 0.1),
    ],
    "summary": [
        ModelTarget("vendor-b", "b-mini", {"chat", "summary"}, 32000, 2.0, 0.25),
        ModelTarget("local", "local-safe", {"chat", "summary"}, 32000, 2.0, 0.1),
    ],
}


def select_candidates(task_type: str, input_tokens: int) -> list[ModelTarget]:
    """按任务、上下文窗口和熔断器筛选候选模型。"""
    candidates = []
    for target in ROUTES[task_type]:
        if input_tokens > target.max_context:
            continue
        breaker = BREAKERS.setdefault(
            f"{target.provider}:{target.model}", CircuitBreaker()
        )
        if breaker.allow():
            candidates.append(target)
    return candidates


async def fake_call_llm(target: ModelTarget, messages: list[dict]) -> dict:
    """模拟供应商响应；生产中替换为统一协议后的 HTTP 调用。"""
    await asyncio.sleep(0.05)
    r = random.random()

    # 故意制造部分供应商故障，便于观察 failover
    if target.provider == "vendor-a" and r < 0.6:
        raise RateLimitError("vendor-a rate limited")
    if target.provider == "vendor-b" and r < 0.25:
        raise UpstreamTimeout("vendor-b timeout")

    return {
        "provider": target.provider,
        "model": target.model,
        "content": f"由 {target.model} 生成：{messages[-1]['content']}",
    }


async def chat_with_failover(
    task_type: str,
    messages: list[dict],
    input_tokens: int,
    max_switch: int = 2,
) -> dict:
    candidates = select_candidates(task_type, input_tokens)
    if not candidates:
        return {
            "degraded": True,
            "content": "当前模型服务暂不可用，请稍后重试，或缩短输入后再试。",
        }

    attempted = []
    for target in candidates[: max_switch + 1]:
        key = f"{target.provider}:{target.model}"
        breaker = BREAKERS[key]
        attempted.append(key)
        try:
            async with asyncio.timeout(target.timeout):
                result = await fake_call_llm(target, messages)
            breaker.record_success()
            result["attempted"] = attempted
            result["degraded"] = len(attempted) > 1
            return result
        except RETRIABLE_ERRORS as exc:
            breaker.record_failure()
            print(f"[failover] {key} 可转移失败：{exc}")
        except AuthError as exc:
            # 密钥问题通常应告警，而不是继续盲目切换所有模型
            print(f"[alert] {key} 鉴权失败：{exc}")
            raise
        except BadRequestError:
            # 参数错误换供应商也无意义，应直接暴露给调用方修正
            raise
        except TimeoutError:
            breaker.record_failure()
            print(f"[failover] {key} 超过 {target.timeout}s")

    return {
        "degraded": True,
        "attempted": attempted,
        "content": "主备模型均不可用，已返回安全兜底结果。",
    }


async def stream_with_safe_failover(
    task_type: str,
    messages: list[dict],
    input_tokens: int,
) -> AsyncIterator[str]:
    """
    流式故障转移演示：只在首 token 前切换。
    一旦已经向客户端输出内容，就不再透明重放，避免重复文本。
    """
    candidates = select_candidates(task_type, input_tokens)
    started = False

    for target in candidates:
        try:
            async with asyncio.timeout(target.timeout):
                # 真实场景：这里读取 SSE，首个 chunk 到达前仍可切换
                first_chunk = await asyncio.wait_for(
                    asyncio.sleep(0.01, result=f"[{target.model}] "),
                    timeout=0.2,
                )
                started = True
                yield first_chunk
                yield "这是已经开始输出后的流式内容。"
                return
        except Exception as exc:
            if started:
                yield "\n[系统提示] 输出中断，请点击续写，避免重复生成。"
                print(f"[stream] 首 token 后失败，不再透明切换：{exc}")
                return
            print(f"[stream] 首 token 前失败，切换下一个模型：{exc}")

    yield "[系统提示] 当前所有模型通道繁忙，请稍后重试。"


async def main():
    messages = [{"role": "user", "content": "帮我分析这个智能体任务的下一步动作"}]
    for i in range(5):
        result = await chat_with_failover("agent", messages, input_tokens=2000)
        print(i, result)

    async for chunk in stream_with_safe_failover("summary", messages, 1000):
        print(chunk, end="")


if __name__ == "__main__":
    asyncio.run(main())

这段代码体现了三个关键原则：候选模型由声明式配置产生；只有可转移错误才进入 failover；熔断器会持续跳过异常目标。真实系统还应把 BREAKERS 放到 Redis、etcd 或网关共享内存中，并为每个请求生成 request_id，把供应商返回的 request_id、模型版本、token 用量、切换原因全部写入 tracing 与审计日志。

生产落地：观测、预算与响应契约

统一网关需要暴露的不只是延迟和错误率，还应包括每个 task_type 的模型命中率、降级率、首 token 延迟、每千 token 成本、重试次数、熔断器开启次数，以及“降级后用户是否完成任务”。如果只看 HTTP 200，可能会发现网关看似成功，实际大量请求已从强推理模型降到小模型，智能体工具调用准确率明显下降。

降级不是“随便换一个能回答的模型”。涉及 JSON Schema、函数调用、结构化输出的链路，应在候选配置中标注 tools、json_mode、vision 等能力，并在切换后复用同一套输出修复器。必要时可以增加一次轻量校验：若备用模型没有稳定函数调用能力，则由网关将自然语言结果转换为内部工具协议，而不是把不确定文本直接丢给工作流。

成本控制也应进入路由决策。可以为每类任务设置最大成本和降级优先级：实时客服优先低延迟，内部分析优先质量，批量任务允许使用更便宜模型并接受排队。故障期间还应动态调整，例如主力供应商连续限流时，将非关键流量主动切到备用通道，给核心链路保留配额。

避坑总结：多模型 failover 的高频事故

坑 1：把所有异常都当作“可重试”

现象是日志中出现大量 400 或 403，但网关仍连续切换三家供应商，最终返回超时，用户等待时间翻倍。修正方法是建立错误分类表：鉴权、参数、内容审核类错误直接失败并告警；429、5xx、连接失败、读取超时才允许转移。还要把供应商错误码映射成内部标准错误码，避免业务方依赖各家 SDK 的异常类型。

坑 2：流式输出中途失败后直接重放请求

现象是界面先出现一段答案，故障转移后又从头输出另一段答案，用户看到重复甚至互相矛盾的内容。修正方法是以首 token 为边界：首 token 前可以透明切换；首 token 后只记录中断事件，并提供续写、重试或人工降级入口。若必须自动修复，应把已输出文本作为上下文传给备用模型，让其“接着写”，并明确拼接边界。

坑 3：每个应用实例各自维护熔断器

现象是某供应商已经全局 503，但各实例仍轮流把流量打过去，恢复时间很长。修正方法是把 provider、model、key、region 维度的熔断状态放入共享存储，并配合随机抖动的半开探测。多实例同时探测时要加锁或采样，避免供应商刚恢复又被探测流量打爆。

坑 4：降级模型能力不满足输出契约

现象是非流式接口返回 200，但智能体期待的 tool_calls 字段消失，工作流在下一步抛出空指针。修正方法是路由配置必须声明能力矩阵，候选模型不支持函数调用时不能进入关键链路；同时在网关层做 Schema 校验和一次修复，仍失败则返回可被业务识别的结构化错误，而不是伪装成成功响应。

坑 5：没有总超时和总切换次数

现象是主模型 5 秒超时、备用模型重试 3 次、兜底模型再等待 8 秒，用户 20 秒后才收到失败提示。修正方法是在请求入口设置端到端预算，并按剩余时间选择候选。例如总预算 10 秒，主模型最多 4 秒，第一次切换最多 3 秒，兜底最多 2 秒；预算耗尽立即返回安全响应，同时保留异步补偿或稍后重试能力。

统一 LLM 网关的核心价值，是把“模型不稳定”从业务代码里的随机异常，转化为可配置、可观测、可演练的系统行为。真正可靠的多模型架构不是永远不失败，而是在供应商限流、网络抖动和模型能力变化时，能够用一致的策略快速切换，并让用户和上层工作流始终得到可预期的结果。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。