LLM API 熔断器模式：将 SRE 模式应用于 AI 基础设施

作者
  • avatar
    姓名
    Nino
    职业
    Senior Tech Editor

你终于发布了核心 AI 功能。用户不断涌入，反馈极佳，留存率稳步上升。然而，在周二下午 2 点，OpenAI 对每一次调用都返回了 429 Too Many Requests 错误。几分钟内，你的整个产品陷入瘫痪，支持队列爆满，社交媒体上充斥着用户的抱怨。如果你在 LLM API 上构建过任何产品，你一定能体会到这种切肤之痛。

讽刺的是，早在十五年前，我们就已经在分布式系统中解决了这个问题。熔断器（Circuit Breakers）、健康检查（Health Checks）和故障转移链（Failover Chains）是微服务架构中的标准配置。然而，当今大多数 LLM 集成仍然异常脆弱——通常只是裸露的 API 调用，加上一个简单的 try/except 块和一份“祈祷”。作为一名在大型生产系统中构建可靠性（SRE）十余年的工程师，我惊讶地发现，在 AI 基础设施领域，我们的工程实践竟然退步了。

当你使用像 n1n.ai 这样的高性能 API 聚合器时，你已经拥有了接入多个后端的能力，但你仍然需要一个逻辑层来优雅地处理故障。本文将介绍在生产环境中真正有效的架构模式。

为什么 LLM API 需要不同的可靠性策略

标准的 REST API 通常是可预测的。如果数据库变慢，它通常会一直慢到负载降低。然而，LLM 供应商引入了传统重试逻辑无法有效处理的独特故障模式：

  1. 激进且不可预测的速率限制：不同于标准的 SaaS API，LLM 的限制（TPM/RPM）会根据供应商的内部集群负载而波动。OpenAI 的 429 错误可能会在对话中途毫无预兆地出现。
  2. 极端的延迟波动：在典型的微服务中，500ms 的响应已经算“慢”了。但在 LLM 世界中，同一个提示词在这一分钟可能只需 800ms，而下一分钟可能因为 Claude 3.5 Sonnet 或 OpenAI o3 的队列深度而需要 15 秒。
  3. 频繁的区域性停机：在过去的一年里，每个主要的供应商都经历过数小时的停机。依赖单一供应商就是单点故障（SPOF）。
  4. 无声的性能退化：在高负载期间，某些供应商可能会将流量路由到模型的量化版本。API 虽然返回了 200 OK，但响应质量明显下降。

大多数开发人员尝试使用“基于祈祷的可靠性”来处理这些问题：

# “基于祈祷”的可靠性方法
for attempt in range(3):
    try:
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        return response
    except Exception:
        time.sleep(2 ** attempt) # 指数退避
raise Exception("所有重试均失败")

这比什么都不做要好，但它存在严重缺陷。你迫使用户等待每一次重试（可能导致 15 秒以上的延迟）。你正在持续轰击一个已经陷入困境的供应商，加剧了“惊群效应”。最重要的是，你的系统没有从失败中学习。

熔断器模式（Circuit Breaker Pattern）

熔断器模式源自电气工程。当电流超过安全水平时，断路器会跳闸并停止电流流动以防止火灾。在软件中，这意味着：如果一个服务持续失败，立即停止向其发送流量，而不是等待每个请求超时。

通过集成 n1n.ai，你可以轻松地在熔断器跳闸时切换模型。一个健壮的熔断器状态机如下：

  • CLOSED（关闭/健康）：请求正常流动。如果请求失败，增加错误计数。
  • OPEN（开启/断路）：如果错误达到阈值（例如 1 分钟内失败 5 次），熔断器跳闸至 OPEN 状态。后续所有请求将立即失败，甚至不会调用 API。这通过“快速失败”保护了用户体验。
  • HALF-OPEN（半开/测试）：经过一段“冷却期”后，熔断器允许恰好一个探测请求通过。如果成功，则关闭熔断器；如果失败，则重新计时。

Python 代码实现

class CircuitBreaker:
    def __init__(self, failure_threshold=3, cool_down_seconds=30):
        self.state = "CLOSED"
        self.failure_count = 0
        self.threshold = failure_threshold
        self.cool_down = cool_down_seconds
        self.last_failure_time = None

    def can_execute(self):
        if self.state == "CLOSED":
            return True

        if self.state == "OPEN":
            # 检查冷却期是否已过
            elapsed = time.time() - self.last_failure_time
            if elapsed >= self.cool_down:
                self.state = "HALF_OPEN"
                return True  # 允许一个探测请求
            return False  # 快速失败

        return False

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.threshold:
            self.state = "OPEN"

核心逻辑在于：当熔断器开启时，失败发生在微秒级。你的用户会得到一个快速的备选方案，而不是看着加载图标转 30 秒，同时后台还在不断重试已经挂掉的端点。

故障转移链：多模型策略

熔断器只能告诉你某个东西坏了，你仍然需要一个地方来发送流量。这就是 故障转移链（Failover Chains） 的用武之地。不要依赖单一模型，而是定义一个有序的目标列表。

使用 n1n.ai 这样的服务可以让你在一个接口下统一管理 OpenAI、Anthropic、Google 和 DeepSeek 等多个供应商，使得故障转移链的实现变得非常简单。

failover_chain = [
    {"model": "gpt-4o", "provider": "openai"},
    {"model": "claude-3-5-sonnet", "provider": "anthropic"},
    {"model": "deepseek-v3", "provider": "deepseek"},
]

async def route_with_failover(messages, chain):
    for target in chain:
        model = target["model"]
        breaker = get_circuit_breaker(model)

        if not breaker.can_execute():
            continue

        try:
            start = time.monotonic()
            response = await call_model(model, messages)
            latency = time.monotonic() - start

            breaker.record_success()
            update_latency_tracker(model, latency)
            return response
        except Exception:
            breaker.record_failure()
            continue
    raise AllProvidersFailedError()

基于指数平滑的动态路由

静态故障转移链是一个好的开始，但模型性能是动态变化的。早上 8 点很快的模型，到了中午可能会变得非常缓慢。为了解决这个问题，我们使用 指数平滑（Exponential Smoothing） 来跟踪实时延迟。简单的平均值是不够的，因为它给昨天的旧数据和今天的实时数据赋予了相同的权重。

def update_latency(model, new_latency_ms):
    alpha = 0.2  # 新测量值的权重
    current_avg = get_avg_latency(model)

    # 最近的测量值更重要
    smoothed = (current_avg * (1 - alpha)) + (new_latency_ms * alpha)
    set_avg_latency(model, smoothed)
    return smoothed

通过设置 alpha 为 0.2，一个从缓慢期恢复的模型将在 5-6 个请求内显示出延迟的改善。这使得你的路由逻辑能够在任何给定时刻自动选择最快的健康模型。

高级加权路由策略

在生产环境中，你并不总是想要最快的模型。有时你想要最便宜的，或者对于 RAG（检索增强生成）等特定任务最可靠的模型。你可以为不同的业务目标定义权重向量：

ROUTING_STRATEGIES = {
    "balanced":    {"success_rate": 0.4, "latency": 0.3, "cost": 0.3},
    "speed":       {"success_rate": 0.2, "latency": 0.6, "cost": 0.2},
    "cost":        {"success_rate": 0.2, "latency": 0.2, "cost": 0.6},
}

这种架构允许你为后台批处理任务设置 optimization_goal: "cost"，而为实时聊天界面设置 optimization_goal: "speed"，所有这些都运行在相同的基础设施之上。

回放测试：安全网

最容易被忽略的模式是 回放测试（Replay Testing）。在部署路由策略的更改之前，你应该通过新算法回放历史流量，观察其表现。这可以防止“我更新了配置，结果延迟翻倍”的情况发生。

通过记录每个请求的模型、延迟和成功状态，你可以在几秒钟内模拟出一周的流量情况。这让团队有信心在不冒险影响生产环境的情况下迭代基础设施。

总结

构建一个具有弹性的 AI 产品需要超越简单的 API 调用。通过实现熔断器，你保护了用户免受挂起请求的影响；通过使用故障转移链，你消除了单点故障；通过利用动态路由，你优化了成本和性能。

手动集成这些模式可能非常复杂，这就是为什么许多开发人员选择 n1n.ai。它提供了实现这些 SRE 模式所需的高速、多模型骨干网，且开销极小。无论你是使用 LangChain、LlamaIndex 还是自定义代码，可靠性都应该是你 AI 技术栈中的一等公民。

在 n1n.ai 获取免费 API 密钥。

参考来源：https://dev.to/deep_mehta_b12a764b0ec6b5/circuit-breakers-for-llm-apis-applying-sre-patterns-to-ai-infrastructure-3pbo