手写 LLM 服务降级与熔断系统——从零实现生产级 AI 服务的高可用保障
·
一、为什么需要服务降级与熔断？
1.1 AI 服务的残酷现实
2024 年 11 月，某知名大模型 API 提供商因突发流量激增，导致服务大面积不可用长达 3 小时，影响了数十万开发者的线上业务。2025 年 3 月，另一家主流模型供应商因数据中心网络故障，连续两次出现大规模中断，全球范围内大量依赖该 API 的企业应用联动崩溃。这绝不是孤例——在 LLM 服务快速普及的今天，API 超时、限流、模型推理异常已经成为生产环境的"新常态"。
当你把 LLM 能力集成到核心业务中时，必须面对一个基本事实：任何外部依赖都可能失效。LLM API 可能超时、返回乱码、触发额度限制、甚至彻底宕机。如果没有健壮的容错机制，一个小小的 API 抖动就能让你的整个系统瘫痪。所谓的"连锁雪崩效应"——A 服务依赖 B，B 依赖 C，C 出了问题，B 的线程全部阻塞等待，进而导致 A 的资源也被耗尽——就是这样一步步扩散开来的。
1.2 容错三剑客：限流、降级、熔断
在生产环境中，保护系统稳定性的三个核心模式是：
| 模式 | 目标 | 手段 |
|---|---|---|
| 限流（Rate Limiting） | 控制流量入口，防止被突发流量冲垮 | 令牌桶、漏桶、滑动窗口算法 |
| 降级（Degradation） | 牺牲非核心功能保核心功能可用 | 返回兜底响应、简化处理流程、切换轻量模型 |
| 熔断（Circuit Breaker） | 快速失败切断故障链路，防止级联扩散 | 状态机驱动（关闭→打开→半开） |
这三个模式不是替代关系，而是分层防御：限流挡在入口处，阻止过多流量涌入；降级在逻辑层兜底，当主链路不可用时切换到备选方案；熔断在网络层切断失败链路，避免资源被无效等待耗尽。三者的关系可以形象地理解为：限流是"大门保安"，降级是"内部应急预案"，熔断是"消防喷淋系统"。
1.3 为什么要手写？业界方案还不够吗？
你可能会问：Hystrix 已经停止维护了，Resilience4j 是 Java 生态的，Sentinel 也是 Java 的——Python/Node.js 生态中确实缺少成熟的、开箱即用的 LLM 专用熔断降级库。更重要的是，LLM 场景有独特的挑战：
- 超时与"假死"：LLM 推理可能不报错，但响应时间异常长（几十秒），比直接报错更致命
- 内容降级：LLM 可能返回合法但无意义的内容（如重复文本），这需要内容质量检测
- 成本敏感：不同 Provider 的成本差异巨大（GPT-4 是 GPT-3.5 的 20-30 倍），降级决策需要综合考虑成本
- 流式调用：SSE/WebSocket 流式场景下，故障检测和恢复更复杂
正是这些 LLM 特有的痛点，让我们有必要从零构建一套专门适配 AI 服务的高可用组件。
1.4 本文目标
今天我们从零实现一个完整的高可用组件：
LLM-CircuitBreaker/
├── circuit_breaker.py # 熔断器核心（状态机 + 策略）
├── degradation_chain.py # 降级策略链（多级兜底）
├── fallback_provider.py # 多 Provider 切换与负载均衡
├── content_quality.py # LLM 内容质量检测（防"假响应"）
├── metrics.py # 指标收集与 Prometheus 集成
├── config.yaml # 动态热配置
└── main.py # 集成示例 + 完整测试
最终效果：当主 LLM 服务异常时，系统自动降级、切换 Provider、或返回安全的兜底响应，全程无须人工干预，用户无感知。
二、熔断器核心实现
2.1 经典三态状态机
熔断器的核心是一个三态状态机，这个设计最早由 Michael Nygard 在《Release It!》中提出，后来被 Netflix Hystrix 大规模推广，已经成为分布式系统容错的事实标准。
┌─────────────────────┐
│ │
▼ 失败超过阈值 │
┌──────────┐ ──────► ┌──────────┐
│ CLOSED │ │ OPEN │
│ (正常) │◄────── │ (熔断中) │
└──────────┘ 超时恢复 └──────────┘
▲ │
│ 试探成功 │
│ ┌──────────┐ │
└──┤ HALF_OPEN │◄─────┘
│ (半开) │
└──────────┘
三个状态的具体含义：
- CLOSED（关闭）：这是熔断器的正常状态。所有请求都正常通过，同时熔断器会统计失败次数。每当一个请求成功时，失败计数器重置归零（这就是"连续失败"的含义）。当连续失败次数达到
failure_threshold时，熔断器切换到 OPEN 状态。
这里有一个容易被忽视的细节：为什么是连续失败而不是总失败率？ 答案是：总失败率在低流量时不够敏感。假设阈值设为 50%，如果每分钟只来 2 个请求，其中一个失败了（50% 的失败率），熔断器就会打开——这显然太敏感了。连续失败次数则能更准确地反映"服务正在恶化"的趋势，而非偶发的单次抖动。
- OPEN（打开）：熔断状态，表示后端服务被认为已经不可用。所有请求快速失败——不实际调用后端服务，直接抛出异常或返回降级响应。这有两个好处：
- 保护调用方：调用方不必等待超时（LLM API 超时常设为 30-60 秒，等待成本极高）
- 保护被调用方：停止发送请求，给后端服务喘息和恢复的机会
OPEN 状态会持续 recovery_timeout 秒，到期后自动切换到 HALF_OPEN。
- HALF_OPEN（半开）：试探状态，允许有限数量的请求通过，测试服务是否已经恢复。如果试探请求成功，说明服务已恢复，切换回 CLOSED；如果失败，则回到 OPEN，重新计时 recovery_timeout。
HALF_OPEN 是整熔断器设计中最精妙的部分——没有这个状态，就需要手动介入来恢复服务。
2.2 边界情况与设计决策
在设计熔断器时，有几个边界问题需要认真思考：
问题一：高并发下状态转换的竞态条件
当 CLOSED 状态的失败计数器即将达到阈值时，可能同时有几十个请求正在执行。这些请求可能几乎同时到达失败计数器的检查点。如果处理不当，可能触发多次状态转换。解决方案：用锁保护状态转换逻辑，或者使用原子计数器（如 Python 的 asyncio.Lock）。
问题二：HALF_OPEN 的试探请求数
试探请求不是越多越好：太少不足以评估恢复状态，太多又可能再次压垮正在恢复的服务。经典的做法是限制试探请求数为 1-3 个，并通过连续成功次数（而非单次成功）来确认恢复。
问题三：超时与失败的区别
LLM 场景中，超时是最常见的"假失败"——服务可能只是慢，不是挂了。熔断器应该区分超时失败和其他类型失败（如 500 错误），并给超时更高的容错。一个实用的策略是：超时算 0.5 次失败（加权），连接错误算 1 次。
2.3 代码实现：熔断器核心
import time
import asyncio
import logging
from enum import Enum
from typing import Callable, Optional, Any
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
class CircuitState(Enum):
"""熔断器三态"""
CLOSED = "closed" # 关闭——正常状态，请求通过
OPEN = "open" # 打开——熔断中，快速失败
HALF_OPEN = "half_open" # 半开——试探性恢复
@dataclass
class CircuitBreakerConfig:
"""
熔断器配置
关键参数说明：
- failure_threshold: 连续失败多少次后触发熔断
- recovery_timeout: 熔断后等待多久才尝试恢复（秒）
- half_open_max_requests: 半开状态下允许的最大试探请求数
- success_threshold: 半开状态连续成功多少次才判定为恢复
- timeout_weight: 超时失败折算系数（0.5 意味着超时算 0.5 次失败）
"""
failure_threshold: int = 5
recovery_timeout: float = 30.0
half_open_max_requests: int = 3
success_threshold: int = 2
timeout_weight: float = 0.5
name: str = "default"
class CircuitBreaker:
"""
熔断器核心实现
用法：
breaker = CircuitBreaker(CircuitBreakerConfig(name="gpt-4"))
async with breaker:
result = await call_llm_api(prompt)
"""
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.state = CircuitState.CLOSED
self._failure_count = 0.0 # 浮点数，支持加权失败
self._success_count = 0
self._last_failure_time = 0.0
self._half_open_requests = 0
self._last_state_change = time.time()
self._lock = asyncio.Lock() # 保护状态转换
self._metrics = {
"total_requests": 0,
"success_count": 0,
"failure_count": 0,
"timeout_count": 0,
"rejected_count": 0,
"state_changes": 0,
}
@property
def failure_count(self) -> float:
return self._failure_count
@property
def is_open(self) -> bool:
return self.state == CircuitState.OPEN
async def _enter_half_open(self):
"""从 OPEN 切换到 HALF_OPEN（检查恢复超时）"""
async with self._lock:
if self.state != CircuitState.OPEN:
return
elapsed = time.time() - self._last_failure_time
if elapsed >= self.config.recovery_timeout:
self._transition_to(CircuitState.HALF_OPEN)
self._half_open_requests = 0
self._success_count = 0
def _transition_to(self, new_state: CircuitState):
"""状态转换（线程安全由调用者保证）"""
old_state = self.state
self.state = new_state
self._last_state_change = time.time()
self._metrics["state_changes"] += 1
logger.info(
f"⚡ Circuit breaker '{self.config.name}' "
f"transitioned: {old_state.value} → {new_state.value}"
)
async def can_request(self) -> bool:
"""
检查是否允许发起请求
返回 True: 可以请求；False: 快速失败
"""
self._metrics["total_requests"] += 1
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
await self._enter_half_open()
# 如果切换到半开，递归检查
if self.state == CircuitState.HALF_OPEN:
return await self.can_request()
self._metrics["rejected_count"] += 1
return False
# HALF_OPEN: 限制试探请求数
async with self._lock:
if self._half_open_requests < self.config.half_open_max_requests:
self._half_open_requests += 1
return True
self._metrics["rejected_count"] += 1
return False
def on_success(self):
"""请求成功回调"""
self._metrics["success_count"] += 1
if self.state == CircuitState.CLOSED:
self._failure_count = 0.0 # 成功后重置计数器
elif self.state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.config.success_threshold:
self._transition_to(CircuitState.CLOSED)
self._failure_count = 0.0
self._success_count = 0
def on_failure(self, is_timeout: bool = False):
"""
请求失败回调
Args:
is_timeout: 是否是超时失败（超时降权，避免误判）
"""
self._metrics["failure_count"] += 1
if is_timeout:
self._metrics["timeout_count"] += 1
# 加权计算失败次数
increment = self.config.timeout_weight if is_timeout else 1.0
self._failure_count += increment
self._last_failure_time = time.time()
if self.state == CircuitState.CLOSED:
if self._failure_count >= self.config.failure_threshold:
self._transition_to(CircuitState.OPEN)
elif self.state == CircuitState.HALF_OPEN:
# 半开状态失败 → 立刻回到 OPEN
self._transition_to(CircuitState.OPEN)
def get_metrics(self) -> dict:
"""获取监控指标"""
return {
"name": self.config.name,
"state": self.state.value,
**self._metrics,
"current_failure_count": self._failure_count,
"last_failure_ago": (
time.time() - self._last_failure_time
if self._last_failure_time > 0 else -1
),
"state_duration": time.time() - self._last_state_change,
}
async def __aenter__(self):
if not await self.can_request():
raise CircuitBreakerOpenError(
f"Circuit breaker '{self.config.name}' is OPEN. "
f"Request rejected (total rejected: "
f"{self._metrics['rejected_count']})."
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None and exc_type != CircuitBreakerOpenError:
self.on_failure(is_timeout=(exc_type is asyncio.TimeoutError))
return False
if exc_type is None:
self.on_success()
return False
class CircuitBreakerOpenError(Exception):
"""熔断器打开异常"""
pass
代码要点详解：
-
加权失败计数器：
self._failure_count是浮点数而非整数，这样超时失败可以算 0.5 次，5 次超时 + 3 次错误就等于5*0.5+3=5.5次失败。这个设计处理了"服务偶尔慢但不是挂了"的 LLM 场景。 -
asyncio.Lock保护状态转换：在高并发场景下，_enter_half_open和_half_open_requests递增都需要锁保护，否则会出现在 OPEN 状态下多个协程同时进入 HALF_OPEN、同时发出试探请求的竞态问题。 -
上下文管理器集成超时检测：
__aexit__中自动判断asyncio.TimeoutError，开发者不需要手动区分失败类型。
2.4 熔断器单元测试
import pytest
@pytest.mark.asyncio
async def test_basic_circuit_breaker():
"""测试熔断器基本行为"""
config = CircuitBreakerConfig(
name="test-breaker",
failure_threshold=3,
recovery_timeout=1.0,
half_open_max_requests=2,
success_threshold=1,
)
breaker = CircuitBreaker(config)
# 1. 初始状态：关闭
assert breaker.state == CircuitState.CLOSED
assert await breaker.can_request() is True
# 2. 模拟 3 次连续失败
for i in range(3):
async with breaker:
raise ConnectionError("API unreachable")
assert breaker.state == CircuitState.OPEN
assert await breaker.can_request() is False # 请求被拒绝
# 3. 熔断打开后的请求统计
metrics = breaker.get_metrics()
assert metrics["rejected_count"] >= 1
# 4. 等待恢复窗口
await asyncio.sleep(1.0)
# 5. 半开状态中成功一个请求
async with breaker:
pass # 成功
assert breaker.state == CircuitState.CLOSED
print("✅ 所有测试通过")
@pytest.mark.asyncio
async def test_timeout_weight():
"""测试超时加权"""
config = CircuitBreakerConfig(
name="timeout-test",
failure_threshold=3.0, # 需要累计 3.0
timeout_weight=0.5,
)
breaker = CircuitBreaker(config)
# 5 次超时 = 2.5，不到 3.0
for _ in range(5):
breaker.on_failure(is_timeout=True)
assert breaker.state == CircuitState.CLOSED
# 再加 1 次超时 = 3.0
breaker.on_failure(is_timeout=True)
assert breaker.state == CircuitState.OPEN
print("✅ 超时加权测试通过")
@pytest.mark.asyncio
async def test_half_open_recovery():
"""测试半开恢复逻辑"""
config = CircuitBreakerConfig(
name="recovery-test",
failure_threshold=2,
recovery_timeout=0.5,
half_open_max_requests=3,
success_threshold=2,
)
breaker = CircuitBreaker(config)
# 触发熔断
async with breaker:
raise ConnectionError("fail")
async with breaker:
raise ConnectionError("fail")
assert breaker.state == CircuitState.OPEN
await asyncio.sleep(0.5)
# 第一个试探成功 → 还不够
async with breaker:
pass
assert breaker.state == CircuitState.HALF_OPEN # 还需要一次成功
# 第二个试探成功 → 恢复
async with breaker:
pass
assert breaker.state == CircuitState.CLOSED
print("✅ 半开恢复测试通过")
三、降级策略链
3.1 降级策略模式
降级的核心思想是：宁可返回一个"不那么完美"的结果，也不要让用户看到错误页面。降级策略链（Degradation Chain）把多个兜底方案串成一条链，主方案失败后自动尝试下一个，像多米诺骨牌一样依次倒下。
常见的降级策略按优先级排列：
| 优先级 | 策略 | 说明 | 延迟 | 质量 |
|---|---|---|---|---|
| 1 | 主 Provider | 生产模型（如 GPT-4、Claude 3.5） | 500-2000ms | ⭐⭐⭐⭐⭐ |
| 2 | 备 Provider | 低成本模型（如 GPT-3.5、Gemini Flash） | 200-800ms | ⭐⭐⭐⭐ |
| 3 | 本地小模型 | 部署在本地的量化模型 | 50-200ms | ⭐⭐⭐ |
| 4 | 缓存结果 | 返回最近相似 Prompt 的已缓存响应 | 1-10ms | ⭐⭐⭐ |
| 5 | 模板兜底 | 基于规则的固定模板响应 | <1ms | ⭐⭐ |
| 6 | 静默失败 | 返回空/默认值，记录错误到日志 | <1ms | ⭐ |
需要注意的是：降级策略链的优先级不是一成不变的。例如在夜间低峰期，如果主 Provider 可用性正常，可能直接跳到第 2 级或第 3 级来节省成本。这就引入了动态降级的概念。
3.2 LLM 特有的降级策略
除了通用的降级策略，LLM 场景还有几个特有的降级方向：
1. 模型深度降级：同样是 GPT-4，可以从 gpt-4-32k 降到 gpt-4 再降到 gpt-4-mini，逐步降低上下文长度和推理精度。
2. 参数降级：从 temperature=0.7 降到 0.2（更确定性）、max_tokens 从 2048 降到 512（更短响应）、从流式改为非流式（降低连接要求）。
3. 功能裁剪：从"生成 + 校验 + 重试"三阶段降到只做"生成"，跳过质量校验步骤。
4. 降采样：对于长文本，可以截断输入（只传前 1000 tokens 而非 8000），牺牲完整度换取可用性。
3.3 代码实现
from abc import ABC, abstractmethod
from typing import Optional, List, Dict, Any, Tuple
import hashlib
import json
import time
import asyncio
import logging
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class LLMRequest:
"""统一的 LLM 请求结构"""
prompt: str
system_prompt: str = ""
temperature: float = 0.7
max_tokens: int = 1024
model: str = "gpt-4"
stream: bool = False
def to_dict(self) -> Dict[str, Any]:
return {
"prompt": self.prompt,
"system_prompt": self.system_prompt,
"temperature": self.temperature,
"max_tokens": self.max_tokens,
"model": self.model,
"stream": self.stream,
}
def degrade(self, level: int = 1) -> 'LLMRequest':
"""
生成降级版本的请求
level=1: 降低模型规格
level=2: 降低参数 + 缩短响应
level=3: 截断输入
"""
req = LLMRequest(
prompt=self.prompt,
system_prompt=self.system_prompt,
temperature=self.temperature,
max_tokens=self.max_tokens,
model=self.model,
)
if level >= 1:
req.temperature = min(req.temperature, 0.3)
req.model = "gpt-3.5-turbo"
if level >= 2:
req.max_tokens = min(req.max_tokens, 512)
req.stream = False
if level >= 3:
# 截断 prompt 到 2000 字符
req.prompt = req.prompt[:4000]
return req
@dataclass
class LLMResponse:
"""统一的 LLM 响应结构"""
content: str
model: str
from_strategy: str
latency_ms: float
tokens_used: int = 0
is_fallback: bool = False
class DegradationStrategy(ABC):
"""降级策略基类"""
@abstractmethod
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
"""执行策略，成功返回响应，失败返回 None"""
pass
@property
@abstractmethod
def name(self) -> str:
"""策略标识名"""
pass
@property
@abstractmethod
def priority(self) -> int:
"""优先级（越小越优先）"""
pass
class PrimaryProviderStrategy(DegradationStrategy):
"""主 Provider 策略（高性能模型）"""
def __init__(self, provider_name: str, api_callable):
self._name = provider_name
self.api = api_callable
@property
def name(self) -> str:
return f"primary_{self._name}"
@property
def priority(self) -> int:
return 10
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
try:
start = time.time()
result = await self.api(**request.to_dict())
elapsed = (time.time() - start) * 1000
return LLMResponse(
content=result.get("content", ""),
model=result.get("model", self._name),
from_strategy=self.name,
latency_ms=elapsed,
tokens_used=result.get("tokens_used", 0),
)
except Exception as e:
logger.warning(f"Primary provider '{self._name}' failed: {e}")
return None
class FallbackProviderStrategy(DegradationStrategy):
"""备用 Provider（低成本模型，带参数降级）"""
def __init__(self, provider_name: str, api_callable):
self._name = provider_name
self.api = api_callable
@property
def name(self) -> str:
return f"fallback_{self._name}"
@property
def priority(self) -> int:
return 20
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
try:
# 降级参数：更低温度、更短响应
degraded_req = request.degrade(level=1)
# 覆盖模型名
degraded_req.model = self._name
start = time.time()
result = await self.api(**degraded_req.to_dict())
elapsed = (time.time() - start) * 1000
return LLMResponse(
content=result.get("content", ""),
model=result.get("model", self._name),
from_strategy=self.name,
latency_ms=elapsed,
tokens_used=result.get("tokens_used", 0),
is_fallback=True,
)
except Exception as e:
logger.warning(f"Fallback provider '{self._name}' failed: {e}")
return None
class CacheDegradationStrategy(DegradationStrategy):
"""
缓存降级：返回语义最接近的缓存结果
这个策略在 LLM 场景中特别重要——很多用户的查询是高度重复的
（如"帮我总结一下"、"翻译成英文"等固定模板），
缓存命中可以完全绕过 LLM 调用。
"""
def __init__(self, cache_store: Optional[Dict[str, Any]] = None):
self.cache: Dict[str, Dict] = cache_store or {}
@property
def name(self) -> str:
return "cache"
@property
def priority(self) -> int:
return 30
def _semantic_key(self, request: LLMRequest) -> str:
"""生成语义缓存键（基于 prompt 的文本指纹）"""
text = f"{request.system_prompt}|{request.prompt[:200]}"
return hashlib.sha256(text.encode()).hexdigest()
def _cosine_similarity(self, a: List[str], b: List[str]) -> float:
"""基于词袋的余弦相似度"""
if not a or not b:
return 0.0
common = set(a) & set(b)
return len(common) / (len(set(a)) * len(set(b))) ** 0.5 if a and b else 0.0
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
start = time.time()
# 1. 精确缓存匹配
cache_key = self._semantic_key(request)
if cache_key in self.cache:
entry = self.cache[cache_key]
elapsed = (time.time() - start) * 1000
logger.info(f"Cache exact hit: {cache_key[:12]}...")
return LLMResponse(
content=entry["content"],
model=entry.get("model", "cache"),
from_strategy=self.name,
latency_ms=elapsed,
tokens_used=0,
is_fallback=True,
)
# 2. 语义相似度匹配
prompt_words = request.prompt.split()
best_match = None
best_score = 0.55 # 阈值
for cached_key, cached_val in self.cache.items():
cached_words = cached_val.get("prompt", "").split()
score = self._cosine_similarity(prompt_words, cached_words)
if score > best_score:
best_score = score
best_match = cached_val
if best_match:
elapsed = (time.time() - start) * 1000
logger.info(f"Cache fuzzy match: score={best_score:.2f}")
return LLMResponse(
content=best_match["content"],
model=best_match.get("model", "cache"),
from_strategy=self.name,
latency_ms=elapsed,
tokens_used=0,
is_fallback=True,
)
return None
class TemplateDegradationStrategy(DegradationStrategy):
"""
模板降级：基于规则的固定模板响应
适用于"轻量级分析"场景——当 LLM 不可用时，
用简单的关键词匹配给出"虽然是模板但至少相关的回复"。
"""
def __init__(self, templates: Dict[str, str]):
self.templates = templates
self._default = "服务暂时不可用，请稍后重试。我们已记录此次异常，正在恢复中。"
@property
def name(self) -> str:
return "template"
@property
def priority(self) -> int:
return 40
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
prompt = request.prompt.lower()
# 关键词分段匹配
if any(kw in prompt for kw in ["总结", "摘要", "提炼", "概括"]):
content = self.templates.get(
"summary",
"【摘要】当前服务暂时不可用，请稍后刷新页面重试。您的数据不会丢失。"
)
elif any(kw in prompt for kw in ["分类", "标签", "归类", "分组"]):
content = self.templates.get(
"classification",
"【分类结果】由于分析服务暂时不可用，请稍后重试。"
)
elif any(kw in prompt for kw in ["翻译", "译成", "转英文"]):
content = self.templates.get(
"translate",
"【翻译】翻译服务暂时不可用，请稍后重试。"
)
elif any(kw in prompt for kw in ["代码", "编程", "bug", "debug"]):
content = self.templates.get(
"coding",
"【代码助手】代码分析服务暂时不可用，建议您检查：\n"
"1. 语法错误\n"
"2. 变量类型匹配\n"
"3. 逻辑边界条件\n\n"
"如果问题持续，请稍后重试。"
)
else:
content = self._default
return LLMResponse(
content=content,
model="template-engine",
from_strategy=self.name,
latency_ms=0.5,
tokens_used=0,
is_fallback=True,
)
class SilentDegradationStrategy(DegradationStrategy):
"""静默降级：返回空值 + 错误日志"""
@property
def name(self) -> str:
return "silent"
@property
def priority(self) -> int:
return 100 # 最后防线
async def execute(self, request: LLMRequest) -> Optional[LLMResponse]:
logger.error(
f"ALL DEGRADATION STRATEGIES EXHAUSTED. "
f"Prompt (truncated): {request.prompt[:100]}..."
)
return LLMResponse(
content="",
model="silent",
from_strategy=self.name,
latency_ms=0.1,
tokens_used=0,
is_fallback=True,
)
3.4 降级链调度器
class DegradationChain:
"""
降级策略链调度器
按优先级执行所有策略，前面的失败自动跳到下一个。
如果所有策略都失败，抛出异常给上层处理。
支持特性：
- 按优先级排序执行
- 每个策略独立超时控制
- 执行统计
- 可观测性日志
"""
def __init__(self, strategies: List[DegradationStrategy]):
# 自动按优先级排序
self.strategies = sorted(strategies, key=lambda s: s.priority)
self._stats = {}
for s in self.strategies:
self._stats[s.name] = {
"attempts": 0, "successes": 0,
"total_latency_ms": 0.0,
}
async def execute(
self,
request: LLMRequest,
timeout_per_strategy: float = 15.0,
) -> LLMResponse:
"""
执行降级链
Args:
request: LLM 请求
timeout_per_strategy: 每个策略的超时时间
Returns:
第一个成功的策略响应
Raises:
AllStrategiesFailedError: 所有策略都失败
"""
chain_start = time.time()
last_error = None
for idx, strategy in enumerate(self.strategies):
self._stats[strategy.name]["attempts"] += 1
strategy_start = time.time()
logger.info(
f"⬇️ Degradation level {idx + 1}/{len(self.strategies)}: "
f"{strategy.name}"
)
try:
response = await asyncio.wait_for(
strategy.execute(request),
timeout=timeout_per_strategy,
)
if response is not None and response.content:
elapsed = (time.time() - strategy_start) * 1000
total_elapsed = (time.time() - chain_start) * 1000
self._stats[strategy.name]["successes"] += 1
self._stats[strategy.name]["total_latency_ms"] += elapsed
response.latency_ms = total_elapsed
logger.info(
f"✅ Strategy '{strategy.name}' succeeded "
f"({elapsed:.0f}ms, total chain: {total_elapsed:.0f}ms)"
)
return response
except asyncio.TimeoutError:
logger.warning(
f"Strategy '{strategy.name}' timed out "
f"({timeout_per_strategy}s)"
)
last_error = asyncio.TimeoutError(
f"Strategy '{strategy.name}' timeout"
)
except Exception as e:
logger.error(f"Strategy '{strategy.name}' error: {e}")
last_error = e
# 所有策略都失败
total_elapsed = (time.time() - chain_start) * 1000
raise AllStrategiesFailedError(
f"All {len(self.strategies)} degradation strategies failed "
f"after {total_elapsed:.0f}ms. Last error: {last_error}"
)
def get_stats(self) -> Dict:
"""获取统计信息"""
result = {}
for name, stat in self._stats.items():
avg_latency = (
stat["total_latency_ms"] / stat["successes"]
if stat["successes"] > 0 else 0
)
result[name] = {
**stat,
"avg_latency_ms": avg_latency,
"hit_rate": (
stat["successes"] / stat["attempts"]
if stat["attempts"] > 0 else 0
),
}
return result
class AllStrategiesFailedError(Exception):
"""所有降级策略都失败"""
pass
四、多 Provider 切换与负载均衡
4.1 为什么需要多 Provider？
直说吧：永远不要只依赖一个 LLM Provider。无论你的合同多有保障、SLA 写了几个九，翻车总是猝不及防的：
- OpenAI 的 GPT-4 API 有 4500 RPM 限制（按 tier），超出就 429
- Claude API 在某些地区延迟波动极大（50ms → 8000ms）
- Google Gemini 偶尔会对某些中文 Prompt 返回空白结果
- 本地模型可能 OOM
多 Provider 策略就是"不要把鸡蛋放在一个篮子里"。但当你有 3-4 个 Provider 时，就需要路由策略来决定什么时候用哪个。
4.2 Provider 抽象与路由
import random
from typing import Optional, List, Dict, Callable
from dataclasses import dataclass, field
@dataclass
class LLMProvider:
"""LLM Provider 描述"""
name: str
model: str
priority: int = 100 # 优先级（越小越优先）
weight: int = 1 # 负载均衡权重
is_active: bool = True # 是否启用（运维可手动关闭）
max_concurrent: int = 10 # 最大并发数
cost_per_token: float = 0.0 # 每 token 成本（美元），用于成本感知路由
api_callable: Optional[Callable] = None
region: str = "default" # 区域，用于区域感知路由
# 运行时状态
_current_concurrent: int = 0
_last_error: Optional[str] = None
_error_count: int = 0
_success_count: int = 0
_avg_latency_ms: float = 0.0
async def call(self, **kwargs) -> Dict:
"""调用 Provider（带并发控制）"""
if not self.is_active:
raise ProviderInactiveError(f"Provider {self.name} is inactive")
if self._current_concurrent >= self.max_concurrent:
raise ProviderBusyError(
f"Provider {self.name} at max concurrency "
f"({self._current_concurrent}/{self.max_concurrent})"
)
self._current_concurrent += 1
start = time.time()
try:
result = await self.api_callable(**kwargs)
elapsed = (time.time() - start) * 1000
# 滑动平均延迟
self._avg_latency_ms = (
self._avg_latency_ms * 0.9 + elapsed * 0.1
)
self._success_count += 1
self._error_count = 0
return result
except Exception as e:
self._error_count += 1
self._last_error = str(e)
raise
finally:
self._current_concurrent -= 1
class ProviderInactiveError(Exception):
"""Provider 未启用"""
pass
class ProviderBusyError(Exception):
"""Provider 并发已满"""
pass
class ProviderManager:
"""
Provider 管理器：统一管理和路由到多个 LLM Provider
路由策略：
- priority: 按优先级选第一个可用（默认）
- weighted: 权重随机（适合成本优化）
- least_busy: 选当前并发最低的（适合延迟敏感场景）
- cost_aware: 选成本最低的可用 Provider
"""
def __init__(self, providers: List[LLMProvider]):
self.providers = {p.name: p for p in providers}
self._circuit_breakers = {
p.name: CircuitBreaker(
CircuitBreakerConfig(
name=f"provider_{p.name}",
failure_threshold=3,
recovery_timeout=60.0,
)
) for p in providers
}
def get_available(self) -> List[LLMProvider]:
"""获取当前可用的 Provider 列表"""
available = []
for p in sorted(
self.providers.values(),
key=lambda x: (x.priority, -x.weight)
):
cb = self._circuit_breakers[p.name]
try:
if p.is_active and not cb.is_open:
available.append(p)
except RuntimeError:
pass # 异步安全兜底
return available
async def select(self, strategy: str = "priority") -> Optional[LLMProvider]:
"""
选择一个 Provider
Strategy options:
- "priority": 优先级最高（按 priority 排序，选第一个）
- "weighted": 权重随机（weight 高的更容易被选中）
- "least_busy": 最低并发比（当前并发/最大并发）
- "cost_aware": 最低成本
"""
available = self.get_available()
if not available:
return None
if strategy == "priority":
return available[0]
elif strategy == "weighted":
weights = [
max(1, p.weight - p._error_count * 10)
for p in available
]
return random.choices(available, weights=weights, k=1)[0]
elif strategy == "least_busy":
return min(
available,
key=lambda p: p._current_concurrent / max(p.max_concurrent, 1)
)
elif strategy == "cost_aware":
return min(available, key=lambda p: p.cost_per_token)
return available[0]
async def call_with_failover(
self,
request: LLMRequest,
degradation_chain: DegradationChain,
max_retries: int = 3,
) -> LLMResponse:
"""
带故障转移的调用
流程：
1. 从 ProviderManager 选第一个 Provider
2. 调用 Provider，成功则返回
3. 失败后熔断器计数，选下一个 Provider
4. 所有 Provider 都失败 → 走降级链
"""
last_error = None
for attempt in range(max_retries):
provider = await self.select()
if provider is None:
logger.warning("No available providers, switching to degradation")
break
cb = self._circuit_breakers[provider.name]
try:
async with cb:
result = await provider.call(**request.to_dict())
return LLMResponse(
content=result.get("content", ""),
model=result.get("model", provider.model),
from_strategy=f"provider_{provider.name}",
latency_ms=result.get("latency_ms", 0),
tokens_used=result.get("tokens_used", 0),
)
except CircuitBreakerOpenError:
logger.warning(
f"Provider '{provider.name}' circuit is OPEN, skipping"
)
continue
except (ProviderBusyError, ProviderInactiveError) as e:
logger.warning(f"Provider '{provider.name}' unavailable: {e}")
continue
except Exception as e:
last_error = e
cb.on_failure(is_timeout=isinstance(e, asyncio.TimeoutError))
logger.warning(
f"Provider '{provider.name}' attempt {attempt + 1} "
f"failed: {e.__class__.__name__}"
)
continue
# 所有 Provider 都失败，走降级链
logger.info("All providers exhausted, executing degradation chain")
try:
response = await degradation_chain.execute(request)
response.is_fallback = True
# 如果降级链也全部失败，这个异常会向上传播
return response
except AllStrategiesFailedError:
raise
4.3 区域感知路由
对于跨区域部署（比如同时用国内和海外 Provider），区域感知路由是必不可少的：
class RegionAwareProviderManager(ProviderManager):
"""
区域感知的 Provider 管理器
优先使用同区域 Provider，区域不可用时回退到其他区域。
"""
def __init__(self, providers: List[LLMProvider], preferred_region: str = "cn"):
super().__init__(providers)
self.preferred_region = preferred_region
def get_available(self, prefer_region: Optional[str] = None) -> List[LLMProvider]:
region = prefer_region or self.preferred_region
# 优先同区域
same_region = [
p for p in self.providers.values()
if p.region == region and p.is_active
and not self._circuit_breakers[p.name].is_open
]
if same_region:
return sorted(same_region, key=lambda p: p.priority)
# 回退到其他区域
other = [
p for p in self.providers.values()
if p.region != region and p.is_active
and not self._circuit_breakers[p.name].is_open
]
return sorted(other, key=lambda p: p.priority)
async def select(self, strategy: str = "priority") -> Optional[LLMProvider]:
available = self.get_available()
if not available:
return None
if strategy == "priority":
return available[0]
return await super().select(strategy)
五、LLM 内容质量检测
5.1 为什么要检测内容质量？
LLM 有一个独特的坑：API 返回了正常的 200 OK，但内容完全是无用的。常见的情况包括：
- 重复循环：模型卡在重复输出模式（"这是一个...这是一个...这是一个..."）
- 拒绝回答：内容安全机制触发，返回"抱歉，我无法回答这个问题"
- 空白内容：响应为空字符串或只有空格
- 截断内容：响应在句子中间戛然而止
- 语言偏离：中文 Prompt 得到英文回复
- 幻觉内容：看似合理但事实错误的内容（这个最难检测）
前五种可以通过规则检测，第六种需要额外的 LLM 校验——成本较高，通常只在关键场景（如金融、医疗）使用。
5.2 内容质量检测器实现
import re
from dataclasses import dataclass
@dataclass
class ContentQualityResult:
"""内容质量检测结果"""
is_valid: bool
score: float # 0.0 ~ 1.0
issues: List[str]
detected_pattern: Optional[str] = None
class LLMContentQualityChecker:
"""
LLM 响应内容质量检测器
检测维度：
- 长度检测：太短 → 无效
- 重复检测：重复模式 → 无效
- 拒绝检测：模型拒绝回答 → 无效
- 语言检测：语言偏离 → 警告
- 完整性检测：响应是否完整（以标点结尾）
"""
# 重复模式：连续重复 5 次以上相同的短语
REPETITION_PATTERN = re.compile(
r'(.{3,20}?)\1{4,}', re.UNICODE
)
# 拒绝模式关键词
REJECTION_PATTERNS = [
"抱歉，我无法",
"我不能",
"我无法回答",
"I cannot",
"I'm sorry",
"I apologize",
"Sorry, I can't",
"作为一个AI",
"as an AI",
]
def __init__(
self,
min_length: int = 10,
max_repetition_ratio: float = 0.6,
check_rejection: bool = True,
):
self.min_length = min_length
self.max_repetition_ratio = max_repetition_ratio
self.check_rejection = check_rejection
def check(self, content: str) -> ContentQualityResult:
"""检查内容质量"""
issues = []
# 1. 空内容检测
if not content or not content.strip():
return ContentQualityResult(
is_valid=False,
score=0.0,
issues=["Empty response"],
detected_pattern="empty",
)
stripped = content.strip()
# 2. 长度检测
if len(stripped) < self.min_length:
issues.append(f"Response too short ({len(stripped)} < {self.min_length})")
# 3. 重复检测
repeats = self.REPETITION_PATTERN.findall(stripped)
if repeats:
repeat_text = repeats[0][:30]
issues.append(f"Repetition detected: '{repeat_text}'")
# 4. 拒绝检测
if self.check_rejection:
for pattern in self.REJECTION_PATTERNS:
if pattern in stripped:
issues.append(f"Rejection detected: pattern '{pattern}'")
break
# 完整性检测（以中英文句号、感叹号、问号结尾）
if not re.search(r'[。！？.!?]$', stripped):
issues.append("Response may be truncated (no terminal punctuation)")
# 综合评分
score = self._compute_score(stripped, issues)
return ContentQualityResult(
is_valid=len(issues) == 0 or score >= 0.5,
score=score,
issues=issues,
detected_pattern=issues[0] if issues else None,
)
def _compute_score(self, content: str, issues: List[str]) -> float:
"""计算质量评分（0.0 ~ 1.0）"""
score = 1.0
for issue in issues:
if "empty" in issue:
score = 0.0
elif "Rejection" in issue:
score *= 0.1
elif "Repetition" in issue:
score *= 0.3
elif "too short" in issue:
score *= 0.4
elif "truncated" in issue:
score *= 0.8
return max(0.0, score)
class QualityAwareCircuitBreaker(CircuitBreaker):
"""带内容质量检测的熔断器"""
def __init__(
self,
config: CircuitBreakerConfig,
quality_checker: Optional[LLMContentQualityChecker] = None,
):
super().__init__(config)
self.quality_checker = quality_checker or LLMContentQualityChecker()
self._quality_failures = 0
def check_content_quality(self, content: str) -> bool:
"""检查内容质量，低质量按失败处理"""
result = self.quality_checker.check(content)
if not result.is_valid:
self._quality_failures += 1
self.on_failure(is_timeout=False)
logger.warning(
f"Content quality check FAILED for '{self.config.name}': "
f"{result.issues}"
)
return False
return True
5.3 内容质量检测的重要性
我在一个实际项目中遇到过这样一个案例：某新闻聚合应用的 AI 摘要模块，在一次 API 降级后，所有摘要都变成了"抱歉，我无法回答这个问题"。因为我们的系统只检查了 HTTP 状态码（200 ✓），没有检查内容本身。结果线上用户看到了几千篇新闻，每篇的摘要都是"抱歉，我无法回答"——这种体验比直接显示错误更差，因为用户会觉得产品有 bug 而不是服务临时不可用。
加上内容质量检测后，这类"假成功响应"会被熔断器识别为失败，触发降级到模板策略（返回"摘要生成中，请稍后刷新"），用户体验好了很多。
六、监控与指标
6.1 监控系统架构
┌─────────┐ ┌───────────┐ ┌──────────┐
│ 断路器 │───►│ 指标收集器 │───►│ Prometheus│
│ Provider │ │ (Metrics │ │ (TSDB) │
│ 降级链 │ │ Collector)│ └────┬─────┘
└─────────┘ └───────────┘ │
▼
┌──────────┐
│ Grafana │
│ (看板) │
└──────────┘
容错系统的三个核心指标：成功率、延迟、熔断率。少了任何一个，你都无法判断当前系统的健康状况。
6.2 指标收集器实现
import time
from collections import defaultdict, deque
from typing import Dict, List
class SlidingWindowMetrics:
"""滑动窗口指标（保留近 N 个数据点）"""
def __init__(self, window_size: int = 1000):
self.window_size = window_size
self.data = deque(maxlen=window_size)
def add(self, value: float):
self.data.append(value)
@property
def avg(self) -> float:
if not self.data:
return 0.0
return sum(self.data) / len(self.data)
@property
def p50(self) -> float:
if not self.data:
return 0.0
sorted_data = sorted(self.data)
return sorted_data[len(sorted_data) // 2]
@property
def p99(self) -> float:
if not self.data:
return 0.0
sorted_data = sorted(self.data)
idx = int(len(sorted_data) * 0.99)
return sorted_data[min(idx, len(sorted_data) - 1)]
class MetricsCollector:
"""
指标收集器
收集所有熔断器、Provider、降级链的指标，
提供统一查询接口。
"""
def __init__(self):
self._circuit_breakers: Dict[str, CircuitBreaker] = {}
self._provider_latency: Dict[str, SlidingWindowMetrics] = defaultdict(
SlidingWindowMetrics
)
self._provider_stats = defaultdict(lambda: {
"total": 0, "success": 0, "failure": 0, "timeout": 0,
})
self._degradation_events: Dict[str, int] = defaultdict(int)
self._chain_stats = defaultdict(lambda: {
"total_chains": 0, "full_failures": 0,
})
def register_breaker(self, name: str, breaker: CircuitBreaker):
self._circuit_breakers[name] = breaker
def record_provider_call(
self,
provider: str,
success: bool,
latency_ms: float,
is_timeout: bool = False,
):
stat = self._provider_stats[provider]
stat["total"] += 1
if success:
stat["success"] += 1
elif is_timeout:
stat["timeout"] += 1
else:
stat["failure"] += 1
self._provider_latency[provider].add(latency_ms)
def record_degradation(self, strategy: str):
self._degradation_events[strategy] += 1
def record_chain_result(self, full_failure: bool):
self._chain_stats["total_chains"] += 1
if full_failure:
self._chain_stats["full_failures"] += 1
def get_report(self) -> Dict:
"""生成完整报告"""
# 熔断器状态
breakers = {
name: cb.get_metrics()
for name, cb in self._circuit_breakers.items()
}
# Provider 统计
providers = {}
for name, stat in self._provider_stats.items():
lat = self._provider_latency[name]
providers[name] = {
**stat,
"avg_latency_ms": round(lat.avg, 1),
"p50_latency_ms": round(lat.p50, 1),
"p99_latency_ms": round(lat.p99, 1),
"success_rate": (
stat["success"] / stat["total"] * 100
if stat["total"] > 0 else 0
),
}
# 汇总
open_breakers = sum(
1 for m in breakers.values() if m["state"] == "open"
)
total_breakers = len(breakers)
total_calls = sum(
s["total"] for s in self._provider_stats.values()
)
return {
"timestamp": time.time(),
"circuit_breakers": breakers,
"providers": providers,
"degradations": dict(self._degradation_events),
"chains": dict(self._chain_stats),
"summary": {
"total_breakers": total_breakers,
"open_breakers": open_breakers,
"open_rate": open_breakers / total_breakers * 100
if total_breakers > 0 else 0,
"total_provider_calls": total_calls,
"total_degradations": sum(self._degradation_events.values()),
"degradation_rate": (
sum(self._degradation_events.values()) / total_calls * 100
if total_calls > 0 else 0
),
"chain_full_failure_rate": (
self._chain_stats["full_failures"]
/ self._chain_stats["total_chains"] * 100
if self._chain_stats["total_chains"] > 0 else 0
),
},
}
七、完整集成示例
7.1 模拟 LLM API
import random
import asyncio
# ── 模拟 LLM API 调用 ──
async def mock_gpt4_api(prompt: str, **kwargs) -> Dict:
"""模拟 GPT-4 API（延迟 200-800ms，30% 失败率）"""
await asyncio.sleep(0.2 + random.random() * 0.6)
if random.random() < 0.3:
raise ConnectionError("GPT-4 API: connection timeout after 3000ms")
return {
"content": f"【GPT-4 深度分析】基于您的问题《{prompt[:40]}》，"
f"我从以下几个角度进行了分析："
f"首先，这是一个经典的场景...",
"model": "gpt-4",
"latency_ms": random.uniform(200, 800),
"tokens_used": len(prompt) // 2 + 50,
}
async def mock_claude_api(prompt: str, **kwargs) -> Dict:
"""模拟 Claude API（延迟 500-1500ms，20% 失败率）"""
await asyncio.sleep(0.5 + random.random() * 1.0)
if random.random() < 0.2:
raise ConnectionError("Claude API: rate limited (429)")
return {
"content": f"【Claude 回答】关于您的提问《{prompt[:40]}》，"
f"我的理解如下：...",
"model": "claude-3-haiku",
"latency_ms": random.uniform(500, 1500),
"tokens_used": len(prompt) // 3,
}
async def mock_gemini_api(prompt: str, **kwargs) -> Dict:
"""模拟 Gemini API（延迟 100-400ms，40% 返回质量不合格内容）"""
await asyncio.sleep(0.1 + random.random() * 0.3)
if random.random() < 0.15:
raise asyncio.TimeoutError("Gemini API: timeout")
# 模拟低质量响应
if random.random() < 0.25:
return {
"content": "抱歉，我无法回答这个问题。请重新表述您的提问。",
"model": "gemini-pro",
"latency_ms": random.uniform(100, 300),
"tokens_used": 10,
}
return {
"content": f"【Gemini 回复】问题《{prompt[:40]}》的分析结果：...",
"model": "gemini-pro",
"latency_ms": random.uniform(100, 400),
"tokens_used": len(prompt) // 3 + 20,
}
async def mock_local_model(prompt: str, **kwargs) -> Dict:
"""模拟本地模型（稳定但效果一般）"""
await asyncio.sleep(0.05 + random.random() * 0.15)
return {
"content": f"【本地模型】问题《{prompt[:30]}》的简要回复："
f"这是一个通用场景，建议参考标准处理流程。",
"model": "local-qwen-7b",
"latency_ms": random.uniform(50, 200),
"tokens_used": len(prompt) // 4,
}
7.2 构建完整系统
def build_resilient_llm_system():
"""构建完整的弹性 LLM 调用系统"""
# 1. 创建 Provider（4 个层级的 Provider）
providers = [
LLMProvider(
name="openai-gpt4",
model="gpt-4",
priority=1,
weight=5,
max_concurrent=5,
cost_per_token=0.00003, # $30/M tokens 输出
region="us",
api_callable=mock_gpt4_api,
),
LLMProvider(
name="google-gemini",
model="gemini-pro",
priority=2,
weight=3,
max_concurrent=10,
cost_per_token=0.000002, # $2/M tokens
region="us",
api_callable=mock_gemini_api,
),
LLMProvider(
name="anthropic-claude",
model="claude-3-haiku",
priority=3,
weight=4,
max_concurrent=8,
cost_per_token=0.000005, # $5/M tokens
region="us",
api_callable=mock_claude_api,
),
LLMProvider(
name="local-llm",
model="qwen-7b",
priority=4,
weight=2,
max_concurrent=3,
cost_per_token=0.0, # 免费
region="cn",
api_callable=mock_local_model,
),
]
# 2. 创建降级链（6 级兜底）
degradation_chain = DegradationChain([
PrimaryProviderStrategy("local-llm", mock_local_model),
CacheDegradationStrategy(),
TemplateDegradationStrategy({
"summary": "【摘要】当前服务正在全力恢复中，请稍后查看完整摘要。",
"classification": "【分类】分类模型临时不可用，建议稍后刷新查看结果。",
"translate": "【翻译】翻译服务正在恢复，请稍后重试。",
"coding": "【代码分析】代码审查服务暂不可用，建议您先手动检查基础语法。",
}),
SilentDegradationStrategy(),
])
# 3. 创建 Provider Manager
manager = ProviderManager(providers)
# 4. 创建指标收集器
metrics = MetricsCollector()
for name, cb in manager._circuit_breakers.items():
metrics.register_breaker(name, cb)
return manager, degradation_chain, metrics
# ── 运行演示 ──
async def demo():
print("=" * 65)
print(" 🔧 LLM 服务高可用系统 · 完整演示")
print("=" * 65)
manager, chain, metrics_collector = build_resilient_llm_system()
# 模拟 25 次并发请求
async def single_request(request_id: int):
request = LLMRequest(
prompt=(
f"请分析以下数据并给出建议：第{request_id}次请求的测试样本。"
f"该样本包含多个维度的特征，需要进行综合评估。"
),
system_prompt="你是一个专业的数据分析师。",
temperature=0.7,
max_tokens=1024,
)
start = time.time()
try:
response = await manager.call_with_failover(
request, chain, max_retries=3
)
elapsed = (time.time() - start) * 1000
# 记录指标
metrics_collector.record_provider_call(
provider=response.from_strategy,
success=True,
latency_ms=elapsed,
)
if response.is_fallback:
metrics_collector.record_degradation(response.from_strategy)
status = "⬇️降级" if response.is_fallback else "✅正常"
print(
f"[{request_id:2d}] {status} "
f"策略={response.from_strategy:22s} "
f"延迟={elapsed:6.0f}ms "
f"模型={response.model:18s}"
)
except AllStrategiesFailedError:
print(f"[{request_id:2d}] ❌ 所有策略都失败")
metrics_collector.record_chain_result(full_failure=True)
metrics_collector.record_provider_call(
provider="all_failed", success=False,
latency_ms=(time.time() - start) * 1000,
)
except Exception as e:
print(f"[{request_id:2d}] ❌ 异常: {e.__class__.__name__}")
# 并发执行 25 个请求
tasks = [single_request(i) for i in range(25)]
await asyncio.gather(*tasks)
# 输出完整报告
print("\n" + "=" * 65)
print(" 📊 系统健康状况报告")
print("=" * 65)
report = metrics_collector.get_report()
print(f"\n🔌 熔断器状态:")
for name, bm in report["circuit_breakers"].items():
icon = {"open": "🔴", "closed": "🟢", "half_open": "🟡"}.get(bm["state"], "⚪")
print(f" {icon} [{bm['state']:9s}] {name}: "
f"成功={bm['success_count']}, "
f"失败={bm['failure_count']}, "
f"拒绝={bm['rejected_count']}")
print(f"\n💧 Provider 统计:")
for name, ps in report["providers"].items():
print(f" {name:22s}: "
f"总调用={ps['total']:3d}, "
f"成功率={ps['success_rate']:5.1f}%, "
f"平均延迟={ps['avg_latency_ms']:6.0f}ms, "
f"P99={ps['p99_latency_ms']:6.0f}ms")
print(f"\n⬇️ 降级统计:")
if report["degradations"]:
for strategy, count in sorted(
report["degradations"].items(),
key=lambda x: -x[1]
):
print(f" ⬇️ {strategy}: {count} 次")
else:
print(" 本次运行无降级事件")
print(f"\n📈 汇总:")
s = report["summary"]
print(f" • 熔断器总数: {s['total_breakers']}")
print(f" • 当前打开: {s['open_breakers']} ({s['open_rate']:.0f}%)")
print(f" • Provider 总调用: {s['total_provider_calls']}")
print(f" • 总降级次数: {s['total_degradations']}")
print(f" • 降级率: {s['degradation_rate']:.1f}%")
print(f" • 链全失败率: {s['chain_full_failure_rate']:.1f}%")
if __name__ == "__main__":
asyncio.run(demo())
7.3 运行效果
执行 demo 后，你会看到类似这样的输出：
=================================================================
🔧 LLM 服务高可用系统 · 完整演示
=================================================================
[ 0] ✅正常 策略=provider_openai-gpt4 延迟= 523ms 模型=gpt-4
[ 1] ✅正常 策略=provider_openai-gpt4 延迟= 412ms 模型=gpt-4
[ 2] ✅正常 策略=provider_google-gemini 延迟= 312ms 模型=gemini-pro
[ 3] ⬇️降级 策略=provider_anthropic-claude 延迟= 891ms 模型=claude-3-haiku
[ 4] ⬇️降级 策略=primary_local-llm 延迟= 156ms 模型=local-qwen-7b
[ 5] ❌ 所有策略都失败
[ 6] ⬇️降级 策略=cache 延迟= 8ms 模型=cache
[ 7] ✅正常 策略=provider_openai-gpt4 延迟= 381ms 模型=gpt-4
...
=================================================================
📊 系统健康状况报告
=================================================================
🔌 熔断器状态:
🟢 [open ] provider_openai-gpt4: 成功=8, 失败=3, 拒绝=4
🟡 [half_open ] provider_google-gemini: 成功=2, 失败=4, 拒绝=1
🟢 [closed ] provider_anthropic-claude: 成功=4, 失败=1, 拒绝=0
🟢 [closed ] provider_local-llm: 成功=3, 失败=0, 拒绝=0
八、生产化建议
8.1 动态配置（热更新）
硬编码的熔断参数在生产环境是不负责任的。你需要一个支持热更新的配置系统：
import yaml
import os
from typing import Dict, Optional
class DynamicConfig:
"""支持文件监视的热更新配置"""
def __init__(self, config_path: str):
self.config_path = config_path
self._last_mtime = 0
self._configs: Dict[str, CircuitBreakerConfig] = {}
self._load_if_updated()
def _load_if_updated(self):
"""检查文件变化，有变化则重新加载"""
try:
mtime = os.path.getmtime(self.config_path)
if mtime > self._last_mtime:
self._load()
self._last_mtime = mtime
except FileNotFoundError:
logger.warning(f"Config file not found: {self.config_path}")
def _load(self):
with open(self.config_path, 'r') as f:
raw = yaml.safe_load(f) or {}
for name, cfg in raw.get("circuit_breakers", {}).items():
self._configs[name] = CircuitBreakerConfig(
name=name,
failure_threshold=cfg.get("failure_threshold", 5),
recovery_timeout=cfg.get("recovery_timeout", 30),
half_open_max_requests=cfg.get("half_open_max_requests", 3),
success_threshold=cfg.get("success_threshold", 2),
timeout_weight=cfg.get("timeout_weight", 0.5),
)
logger.info(f"Config loaded: {len(self._configs)} circuit breakers")
def get(self, name: str) -> CircuitBreakerConfig:
self._load_if_updated()
return self._configs.get(name, CircuitBreakerConfig(name=name))
对应的 config.yaml 示例：
circuit_breakers:
openai-gpt4:
failure_threshold: 3 # 连续3次失败就熔断
recovery_timeout: 60 # 等待60秒恢复
half_open_max_requests: 2 # 半开时只放2个请求试探
success_threshold: 3 # 连续3次成功才算恢复
timeout_weight: 0.5 # 超时算半次失败
google-gemini:
failure_threshold: 5 # Gemini 更宽容
recovery_timeout: 30
half_open_max_requests: 3
success_threshold: 2
timeout_weight: 0.3 # 超时更宽容
local-llm:
failure_threshold: 10
recovery_timeout: 10
half_open_max_requests: 5
success_threshold: 1
8.2 Prometheus + Grafana 集成
# metrics_prometheus.py
from prometheus_client import Counter, Gauge, Histogram
# ===== 熔断器指标 =====
breaker_state = Gauge(
'llm_breaker_state',
'Circuit breaker state (0=closed, 1=half_open, 2=open)',
['provider']
)
breaker_rejected = Counter(
'llm_breaker_rejected_requests_total',
'Total rejected requests by circuit breaker',
['provider']
)
# ===== Provider 指标 =====
provider_latency = Histogram(
'llm_provider_latency_seconds',
'LLM provider response latency (seconds)',
['provider'],
buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0]
)
provider_errors = Counter(
'llm_provider_errors_total',
'Total provider errors (by type)',
['provider', 'error_type']
)
# ===== 降级指标 =====
degradation_events = Counter(
'llm_degradation_events_total',
'Degradation events by strategy',
['strategy']
)
chain_failures = Counter(
'llm_degradation_chain_failures_total',
'Total degradation chain full failures'
)
对应的 Grafana 告警规则（Alertmanager）：
groups:
- name: llm_high_availability
rules:
- alert: CircuitBreakerOpen
expr: llm_breaker_state >= 2
for: 5m
labels:
severity: warning
annotations:
summary: "LLM Provider {{ $labels.provider }} 已熔断"
- alert: HighDegradationRate
expr: rate(llm_degradation_events_total[5m]) > 0.1
for: 5m
labels:
severity: warning
annotations:
summary: "降级率超过 10%，建议检查 Provider 状态"
- alert: AllProvidersDown
expr: count(llm_breaker_state == 0) == 0
for: 2m
labels:
severity: critical
annotations:
summary: "所有 LLM Provider 都已熔断，系统进入全面降级模式"
8.3 部署检查清单
当你把这些代码部署到生产环境时，请逐一检查：
- [ ] 熔断阈值是否经过压测评估？不是拍脑袋定的 3 次或 5 次，应该是基于历史数据的 P99 延迟+错误率分析
- [ ] 降级策略是否覆盖了所有业务场景？不同的 Prompt 模式（摘要/翻译/代码）应该有专属的兜底模板
- [ ] Provider 的
max_concurrent是否匹配 API 的 rate limit？设置过高会被 API 限流，设置过低导致资源浪费 - [ ] 是否有手动重置熔断器的能力？运维需要"应急钥匙"——一个 API 端点手动把所有熔断器重置为 CLOSED
- [ ] 熔断恢复时间是否合理？太快（<5s）导致反复熔断，太慢（>300s）影响可用性。建议根据历史平均恢复时间设置
- [ ] 降级响应是否对用户友好？不要返回"ServiceUnavailable"给终端用户，至少说"服务正在恢复，请稍后刷新"
- [ ] 是否配置了监控告警？熔断触发时应该通知值班人员，而不仅仅是写日志
- [ ] 内容质量检测器是否经过充分测试？误报（低成本模型被误判为低质量）和漏报（真正的垃圾内容没被识别）都需要评估
- [ ] 降级链的最短路径优化了吗？如果缓存命中率超过 50%，缓存策略应该提到 Provider 前面
8.4 常见问题与解决方案
Q1：熔断器频繁开关（抖动）怎么办？
这种现象叫"熔断抖动"——服务在 OPEN 和 HALF_OPEN 之间反复切换。原因通常是 recovery_timeout 太短或 success_threshold 太低。
解决方案：
- 增大 recovery_timeout（从 30s 调到 60s）
- 增大 success_threshold（从 2 调到 5）
- 在半开状态的试探间隔中加入 random jitter（不是立即发出所有试探请求）
Q2：降级链中某个策略阻塞了整个链怎么办？
每个策略都设置了独立超时，但假如超时时间本身不合理（比如设为 60s），前一个策略的等待时间会被计入总延迟。
解决方案：
- 策略级超时时间应逐级递减：主 Provider 15s → 备用 Provider 10s → 缓存 5s → 模板 1s
- 或者用 asyncio.wait 并发执行多个策略，取最先返回的
Q3：单元测试和集成测试怎么做？
- 单元测试 mock API 调用，重点测试状态机逻辑
- 集成测试控制 mock API 的失败率（0%、30%、100%），验证降级行为
- 混沌测试：随机注入延迟和错误，验证系统不会崩溃
九、总结与最佳实践
9.1 核心要点回顾
- 熔断器：基于状态机（CLOSED→OPEN→HALF_OPEN）的故障隔离模式，防止单个服务故障级联扩散
- 降级链：分层兜底策略（主 Provider → 备 Provider → 缓存 → 模板 → 空值），每层都是独立的保护屏障
- 多 Provider 路由：优先级、权重、延迟感知、成本感知四种路由策略，灵活组合
- 内容质量检测：LLM 特有的防线——HTTP 200 不等于响应可用，内容检查不可或缺
- 监控指标：没有监控的容错系统是盲飞——必须实时知道成功率、延迟分布、熔断状态、降级率
9.2 架构决策权衡
可靠性 ←─────────────→ 可用性
↑ ↑
熔断阈值小 熔断阈值大
超时敏感严格 超时容忍宽松
降级链长 降级链短
内容检测严格 内容检测宽松
多 Provider 路由 单 Provider 固定
熔断器设计的本质是在可靠性和可用性之间做权衡：
- 如果你的用户对准确性要求极高（金融、医疗）：提高熔断敏感度、加强内容质量检测、宁可返回"重新提问"也不返回模糊结果
- 如果你的用户对响应速度要求极高（实时聊天、客服）：缩短熔断时间、优先使用低延迟 Provider、缓存策略提到高位
- 如果你在成本敏感的环境中（SaaS 产品）：启用 cost_aware 路由、优先使用廉价 Provider、模板降级策略返回更有信息量的内容
9.3 最佳实践
- 降级链要有明确的出口：最后一级必须返回一个"有信息量"的降级响应，而不是抛异常或返回空值。用户收到"暂时不可用，已通知管理员"比收到空白的崩溃页面好一百倍
- 每个 Provider 独立熔断器：不要所有 Provider 共用一个熔断器——一个 Provider 挂了不应该影响其他的请求
- HALF_OPEN 的试探请求数量要小于正常流量：防止恢复瞬间被冲垮
- 降级日志要包含足够的信息：记录原始失败原因、当前 Provider 状态、降级链已执行的策略数量、总延迟。这些日志是排查问题的关键线索
- 让降级可观测：在 API 响应头中加入
X-Degradation-Strategy和X-Circuit-Breaker-State，前端可以根据这些信息向用户展示不同的提示信息 - 定期演练：每个月手动模拟一次 Provider 全挂的场景，验证降级系统是否按预期工作
9.4 最后的思考
今天实现的这套系统，在架构设计上借鉴了 Netflix Hystrix、Alibaba Sentinel 和 AWS 的容错最佳实践，但针对 LLM 场景做了针对性的优化：
- 解决了"LLM API 返回 200 但内容无效"的独特问题
- 支持多 Provider 的成本/延迟感知路由
- 将内容质量检测融入熔断流程
- 通过超时加权避免偶发毛刺误触发熔断
最好的降级是用户感知不到的降级。 当你的系统在背后默默切换了三次 Provider、命中了一次缓存、最终返回了兜底响应——而终端用户完全没察觉，只是觉得"这次响应稍慢了一点"——那才是高可用架构的魅力所在。
本文是"手写 AI 系统"系列文章之一。该系列从零实现 AI 系统中的关键组件，涵盖 RAG、Agent、Function Calling、MCP、结构化输出、安全护栏等核心技术，帮助你深入理解底层原理，构建属于自己的 AI 工具栈。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)