单点风险分析 LLM 服务链路中的每个环节都可能成为单点故障：
故障点 影响 概率 恢复时间 LLM 提供商 API 宕机 完全不可用 中 5-60min LLM 提供商限流 部分请求失败 高 1-5min 模型版本下线 特定模型不可用 低 需代码变更 网络抖动 延迟升高/超时 高 秒级 自建 GPU 节点故障 推理能力下降 中 10-30min Embedding 服务故障 RAG 检索失效 低 5-15min 核心原则：任何单一 LLM 提供商都不应成为系统的单点依赖。
多模型冗余 模型池设计 from dataclasses import dataclass, field from enum import Enum import asyncio from collections import defaultdict class ModelStatus(Enum): HEALTHY = "healthy" DEGRADED = "degraded" # 可用但质量/延迟有问题 UNHEALTHY = "unhealthy" @dataclass class ModelEndpoint: name: str provider: str # openai / anthropic / local priority: int # 1=首选, 2=备选... status: ModelStatus = ModelStatus.HEALTHY consecutive_errors: int = 0 last_error_time: float = 0 class ModelPool: def __init__(self): self.endpoints = [ ModelEndpoint("gpt-4o", "openai", priority=1), ModelEndpoint("claude-sonnet", "anthropic", priority=2), ModelEndpoint("llama-70b-local", "local", priority=3), ] self.circuit_breakers = defaultdict(CircuitBreaker) def get_available(self, exclude_degraded=False): """按优先级返回可用端点""" endpoints = sorted(self.endpoints, key=lambda e: e.priority) result = [] for ep in endpoints: if ep.status == ModelStatus.UNHEALTHY: continue if exclude_degraded and ep.status == ModelStatus.DEGRADED: continue result.append(ep) return result 故障检测与自动切换 class FailoverHandler: def __init__(self, pool: ModelPool, max_retries=3): self.pool = pool self.max_retries = max_retries async def complete_with_failover(self, messages, **kwargs): endpoints = self.pool.get_available() last_error = None for ep in endpoints: try: response = await self._call_with_timeout( ep, messages, timeout=30, **kwargs ) # 成功调用，重置错误计数 ep.consecutive_errors = 0 ep.status = ModelStatus.HEALTHY return response, ep.name except asyncio.TimeoutError: self._record_error(ep, "timeout") last_error = "timeout" except RateLimitError: self._record_error(ep, "rate_limit") last_error = "rate_limit" except Exception as e: self._record_error(ep, str(e)) last_error = str(e) # 所有端点都失败 raise AllEndpointsFailedError(f"Last error: {last_error}") def _record_error(self, endpoint: ModelEndpoint, reason: str): endpoint.consecutive_errors += 1 endpoint.last_error_time = time.time() if endpoint.consecutive_errors >= 5: endpoint.status = ModelStatus.UNHEALTHY logger.warning(f"Endpoint {endpoint.name} marked unhealthy: {reason}") elif endpoint.consecutive_errors >= 2: endpoint.status = ModelStatus.DEGRADED logger.info(f"Endpoint {endpoint.name} degraded: {reason}") async def _call_with_timeout(self, endpoint, messages, timeout, **kwargs): return await asyncio.wait_for( self._dispatch(endpoint, messages, **kwargs), timeout=timeout ) 优雅降级 当所有高端模型不可用时，系统应自动降级到更小/更便宜的模型，而非直接报错。
...
