2026 年"双十一"预售日深夜，我所在电商平台的 AI 客服系统遭遇了前所未有的并发冲击。凌晨 2 点，某头部主播的直播间突然爆单，AI 客服的并发请求量从日常的 200 QPS 飙升至 15,000 QPS，OpenAI API 的速率限制（Rate Limit）不断触发，响应延迟从 800ms 暴涨至 28 秒。用户投诉工单像雪片一样飞来，运营团队在凌晨 3 点紧急拉群，技术团队开始疯狂翻文档找解决方案。

这是我们决定从单一 OpenAI API 迁移到多模型 API 网关的导火索。在 HolySheep 的支持下，我们用了 3 周时间完成了一套完整的迁移架构，实现了多模型自动路由、成本降低 78%、P99 延迟稳定在 200ms 以内。本文将详细复盘这个过程，包含所有可直接落地的代码和配置。

为什么企业必须考虑多模型 API 网关

在那个灾难性的夜晚之后，我对 API 网关选型做了深入调研。核心问题有三个：

迁移到 HolySheep 多模型 API 网关后，我们构建了一套智能路由系统：根据请求类型自动选择最优模型——简单问答走 Gemini 2.5 Flash（$2.50/MTok），复杂推理走 Claude Sonnet 4.5（$15/MTok），而国内直连延迟稳定在 50ms 以内，彻底告别跨境抖动。

整体架构设计

我们的迁移架构分为四层：

┌─────────────────────────────────────────────────────────────────┐
│                         客户端层                                  │
│              (Web / App / 客服系统 / RAG 应用)                    │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                       API 网关层                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐   │
│  │  灰度路由     │  │   限流熔断    │  │     密钥管理         │   │
│  │  (按用户/地区) │  │  (令牌桶算法) │  │  (多 Key 轮询+兜底)  │   │
│  └──────────────┘  └──────────────┘  └──────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      模型路由层                                   │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐    │
│  │GPT-4.1  │ │Claude   │ │Gemini   │ │DeepSeek │ │  本地    │    │
│  │$8/MTok  │ │Sonnet4.5│ │2.5 Flash│ │V3.2     │ │  模型    │    │
│  │         │ │$15/MTok │ │$2.5/MTok│ │$0.42/MT │ │         │    │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘    │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    HolySheep API 网关                            │
│           https://api.holysheep.ai/v1  ·  国内 <50ms           │
└─────────────────────────────────────────────────────────────────┘

第一步：SDK 兼容层实现（零改动迁移）

很多团队不敢迁移的核心顾虑是"代码改动太大"。我们的方案是通过适配器模式，在不改动业务代码的情况下实现透明切换。

# openai_adapter.py

OpenAI SDK 兼容适配器 - 替换 base_url 即可切换到 HolySheep

import openai
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import httpx
import asyncio

class ModelTier(Enum):
    FAST = "gemini-2.0-flash"           # 快速响应 - $2.50/MTok
    BALANCED = "gpt-4.1"                 # 均衡选择 - $8/MTok
    REASONING = "claude-sonnet-4.5"      # 深度推理 - $15/MTok
    ECONOMY = "deepseek-v3.2"            # 极致性价比 - $0.42/MTok

@dataclass
class RoutableRequest:
    prompt_tokens: int
    requires_reasoning: bool = False
    user_tier: str = "free"  # free / pro / enterprise

class HolySheepAdapter:
    """
    OpenAI SDK 适配器 - 零代码改动迁移方案
    只需将 openai.OpenAI() 的 base_url 指向 HolySheep 即可
    """
    def __init__(
        self, 
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",  # HolySheep API 端点
        timeout: float = 30.0,
        max_retries: int = 3
    ):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url=base_url,  # 关键：指向 HolySheep
            timeout=timeout,
            max_retries=max_retries
        )
        self.fallback_keys: List[str] = []
        
    def _route_model(self, request: RoutableRequest) -> str:
        """智能模型路由策略"""
        if request.requires_reasoning:
            return ModelTier.REASONING.value
        elif request.prompt_tokens > 32000:
            # 超长上下文走 DeepSeek，性价比最高
            return ModelTier.ECONOMY.value
        elif request.user_tier == "pro":
            return ModelTier.BALANCED.value
        else:
            # 默认走 Gemini Flash，兼顾速度与成本
            return ModelTier.FAST.value
    
    def chat(
        self,
        messages: List[Dict[str, str]],
        model: Optional[str] = None,
        temperature: float = 0.7,
        **kwargs
    ) -> Dict[str, Any]:
        """
        兼容 OpenAI SDK 的 chat 接口
        使用示例:
            adapter = HolySheepAdapter(api_key="YOUR_HOLYSHEEP_API_KEY")
            response = adapter.chat(
                messages=[{"role": "user", "content": "Hello!"}]
            )
        """
        # 如果未指定模型，使用路由策略
        if not model:
            model = ModelTier.FAST.value
            
        response = self.client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            **kwargs
        )
        
        return {
            "id": response.id,
            "model": response.model,
            "choices": [{
                "message": response.choices[0].message.model_dump(),
                "finish_reason": response.choices[0].finish_reason
            }],
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }
        }
    
    async def achat(
        self,
        messages: List[Dict[str, str]],
        model: Optional[str] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """异步接口 - 用于高并发场景"""
        if not model:
            model = ModelTier.FAST.value
            
        response = await self.client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs
        )
        return response.model_dump()


使用示例 - 零改动从 OpenAI 切换到 HolySheep

def example_migration():
    # 旧的 OpenAI 代码（仅需修改 base_url）
    # client = openai.OpenAI(api_key="sk-xxx", base_url="https://api.openai.com/v1")
    
    # 新的 HolySheep 代码（99% 兼容）
    client = HolySheepAdapter(
        api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为你的 HolySheep Key
        base_url="https://api.holysheep.ai/v1"
    )
    
    response = client.chat(
        messages=[
            {"role": "system", "content": "你是一个专业的电商客服"},
            {"role": "user", "content": "双十一活动什么时候开始？"}
        ],
        temperature=0.7
    )
    
    print(f"响应: {response['choices'][0]['message']['content']}")
    print(f"Token 消耗: {response['usage']['total_tokens']}")

if __name__ == "__main__":
    example_migration()

第二步：密钥管理与多 Key 轮询

企业级应用中，单一 API Key 的限流往往无法满足需求。我们实现了基于 HolySheep 的多 Key 轮询 + 兜底机制。

# key_manager.py

企业级多 Key 管理系统 - 支持轮询、加权、兜底

from typing import List, Dict, Optional, Callable
from dataclasses import dataclass, field
from collections import deque
import asyncio
import time
import threading
from enum import Enum

class KeyStatus(Enum):
    ACTIVE = "active"
    RATE_LIMITED = "rate_limited"
    DEPLETED = "depleted"
    ERROR = "error"

@dataclass
class APIKey:
    key: str
    name: str
    weight: int = 1  # 权重，用于加权轮询
    rpm_limit: int = 500  # 每分钟请求限制
    tpm_limit: int = 100000  # 每分钟 Token 限制
    current_rpm: int = 0
    current_tpm: int = 0
    status: KeyStatus = KeyStatus.ACTIVE
    last_reset: float = field(default_factory=time.time)
    error_count: int = 0
    consecutive_errors: int = 0

class KeyManager:
    """
    企业级密钥管理器
    
    功能特性:
    - 加权轮询：根据 Key 权重分配请求
    - 自动熔断：连续错误自动禁用
    - 速率保护：RPM/TPM 双维度限流
    - 兜底机制：主 Key 耗尽自动切换
    """
    
    def __init__(self, fallback_url: str = "https://api.holysheep.ai/v1"):
        self.keys: List[APIKey] = []
        self.current_index = 0
        self.fallback_url = fallback_url
        self._lock = threading.RLock()
        self._reset_interval = 60  # 60秒重置窗口
        
        # 熔断配置
        self.max_consecutive_errors = 5
        self.circuit_break_duration = 300  # 5分钟熔断
        
    def add_key(
        self, 
        key: str, 
        name: str = "default",
        weight: int = 1,
        rpm_limit: int = 500,
        tpm_limit: int = 100000
    ):
        """添加 API Key"""
        api_key = APIKey(
            key=key,
            name=name,
            weight=weight,
            rpm_limit=rpm_limit,
            tpm_limit=tpm_limit
        )
        self.keys.append(api_key)
        print(f"✅ 已添加 Key: {name}, 权重: {weight}, RPM: {rpm_limit}, TPM: {tpm_limit}")
        
    def _reset_counters(self, key: APIKey):
        """重置计数器"""
        now = time.time()
        if now - key.last_reset >= self._reset_interval:
            key.current_rpm = 0
            key.current_tpm = 0
            key.last_reset = now
            
    def _check_rate_limit(self, key: APIKey, tokens: int = 0) -> bool:
        """检查是否触发限流"""
        self._reset_counters(key)
        
        # 检查 RPM
        if key.current_rpm >= key.rpm_limit:
            key.status = KeyStatus.RATE_LIMITED
            return False
            
        # 检查 TPM
        if key.current_tpm + tokens > key.tpm_limit:
            key.status = KeyStatus.RATE_LIMITED
            return False
            
        return True
    
    def get_available_key(self, tokens: int = 0) -> Optional[APIKey]:
        """
        获取可用的 Key（加权轮询 + 熔断保护）
        
        返回: 可用的 APIKey 或 None
        """
        with self._lock:
            # 先检查是否有刚恢复的 Key
            for key in self.keys:
                if (key.status == KeyStatus.RATE_LIMITED or 
                    key.status == KeyStatus.ERROR):
                    self._check_recovery(key)
                    
            # 加权轮询选择
            candidates = []
            for key in self.keys:
                if key.status == KeyStatus.ACTIVE:
                    if self._check_rate_limit(key, tokens):
                        # 根据权重添加候选
                        for _ in range(key.weight):
                            candidates.append(key)
                            
            if not candidates:
                return None
                
            # 轮询选择
            selected = candidates[self.current_index % len(candidates)]
            self.current_index += 1
            
            return selected
            
    def record_usage(self, key: APIKey, tokens: int = 0):
        """记录 Key 使用量"""
        with self._lock:
            key.current_rpm += 1
            key.current_tpm += tokens
            
    def record_error(self, key: APIKey, is_consecutive: bool = True):
        """记录错误"""
        with self._lock:
            key.error_count += 1
            if is_consecutive:
                key.consecutive_errors += 1
                
            if key.consecutive_errors >= self.max_consecutive_errors:
                key.status = KeyStatus.ERROR
                print(f"⚠️ Key {key.name} 触发熔断（连续{key.consecutive_errors}次错误）")
                
    def record_success(self, key: APIKey):
        """记录成功调用"""
        with self._lock:
            key.consecutive_errors = 0
            
    def _check_recovery(self, key: APIKey):
        """检查 Key 是否恢复"""
        # 实现自动恢复逻辑
        if key.status == KeyStatus.ERROR:
            # 可以在这里添加时间检查逻辑
            key.status = KeyStatus.ACTIVE
            key.consecutive_errors = 0
            print(f"✅ Key {key.name} 已自动恢复")
            
    def get_status_report(self) -> Dict:
        """获取 Key 状态报告"""
        return {
            "total_keys": len(self.keys),
            "active_keys": sum(1 for k in self.keys if k.status == KeyStatus.ACTIVE),
            "keys": [{
                "name": k.name,
                "status": k.status.value,
                "rpm": f"{k.current_rpm}/{k.rpm_limit}",
                "tpm": f"{k.current_tpm}/{k.tpm_limit}",
                "errors": k.consecutive_errors
            } for k in self.keys]
        }


使用示例

def example_key_management():
    manager = KeyManager()
    
    # 添加多个 Key（模拟企业多账号场景）
    manager.add_key(
        key="YOUR_HOLYSHEEP_API_KEY_1",
        name="主账号-生产",
        weight=3,      # 高权重
        rpm_limit=1000,
        tpm_limit=200000
    )
    
    manager.add_key(
        key="YOUR_HOLYSHEEP_API_KEY_2", 
        name="备用账号",
        weight=1,      # 低权重
        rpm_limit=500,
        tpm_limit=100000
    )
    
    # 模拟请求
    for i in range(5):
        key = manager.get_available_key(tokens=500)
        if key:
            print(f"请求 {i+1}: 使用 Key {key.name}")
            manager.record_usage(key, tokens=500)
            manager.record_success(key)
        else:
            print(f"请求 {i+1}: 所有 Key 均不可用，触发降级")
            
    # 状态报告
    print("\n📊 Key 状态报告:")
    print(manager.get_status_report())

if __name__ == "__main__":
    example_key_management()

第三步：限流熔断与高可用设计

在电商大促场景下，限流熔断机制是保护系统的最后一道防线。我们的方案基于令牌桶算法，支持多维度限流。

# rate_limiter.py

令牌桶限流器 + 熔断器实现

import time
import asyncio
from typing import Dict, Optional, Callable, Any
from dataclasses import dataclass, field
from collections import defaultdict
from enum import Enum
import threading
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 熔断
    HALF_OPEN = "half_open"  # 半开

@dataclass
class CircuitBreaker:
    """
    熔断器实现
    
    状态转换:
    CLOSED -> OPEN: 失败率超过阈值
    OPEN -> HALF_OPEN: 熔断超时后
    HALF_OPEN -> CLOSED: 试探成功
    HALF_OPEN -> OPEN: 试探失败
    """
    failure_threshold: float = 0.5   # 触发熔断的失败率
    success_threshold: int = 3        # 恢复需要的成功次数
    timeout: float = 30.0             # 熔断超时时间（秒）
    volume_threshold: int = 10        # 最小请求量
    
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    total_count: int = 0
    last_failure_time: float = field(default_factory=time.time)
    
    def record_success(self):
        self.total_count += 1
        self.failure_count = max(0, self.failure_count - 1)
        
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self._reset()
                
    def record_failure(self):
        self.total_count += 1
        self.failure_count += 1
        
        # 判断是否需要熔断
        if (self.total_count >= self.volume_threshold and 
            self.failure_count / self.total_count >= self.failure_threshold):
            self._trip()
            
    def _trip(self):
        self.state = CircuitState.OPEN
        self.last_failure_time = time.time()
        logger.warning("🔴 熔断器打开")
        
    def _reset(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.total_count = 0
        logger.info("🟢 熔断器恢复")
        
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
            
        if self.state == CircuitState.OPEN:
            # 检查超时
            if time.time() - self.last_failure_time >= self.timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                logger.info("🟡 熔断器进入半开状态")
                return True
            return False
            
        # HALF_OPEN 状态允许试探请求
        return True


class TokenBucket:
    """
    令牌桶限流器
    
    特性:
    - 支持突发流量
    - 精确的速率控制
    - 线程安全
    """
    def __init__(self, rate: float, capacity: int):
        """
        Args:
            rate: 每秒产生的令牌数
            capacity: 桶的容量
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        self._lock = threading.Lock()
        
    def _refill(self):
        """补充令牌"""
        now = time.time()
        elapsed = now - self.last_update
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_update = now
        
    def acquire(self, tokens: int = 1, blocking: bool = False, timeout: float = 5.0) -> bool:
        """
        获取令牌
        
        Args:
            tokens: 需要获取的令牌数
            blocking: 是否阻塞等待
            timeout: 阻塞超时时间
            
        Returns:
            是否成功获取令牌
        """
        deadline = time.time() + timeout
        
        while True:
            with self._lock:
                self._refill()
                
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
                    
                if not blocking:
                    return False
                    
            # 非阻塞或超时
            if time.time() >= deadline:
                return False
                
            # 等待一段时间后重试
            wait_time = tokens / self.rate
            time.sleep(min(wait_time, 0.1))
            
    def get_available(self) -> int:
        """获取当前可用令牌数"""
        with self._lock:
            self._refill()
            return int(self.tokens)


class RateLimiter:
    """
    多维度限流器
    
    支持:
    - 全局限流
    - 用户维度限流
    - 接口维度限流
    - 模型维度限流
    """
    def __init__(self):
        # 全局限流器 (10000 QPS)
        self.global_limiter = TokenBucket(rate=10000, capacity=10000)
        
        # 用户维度限流器
        self.user_limiters: Dict[str, TokenBucket] = {}
        self.user_default_rate = 100  # 每用户 100 QPS
        self.user_default_capacity = 100
        
        # 模型维度限流器
        self.model_limiters: Dict[str, TokenBucket] = {}
        
        # 熔断器
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        
        self._lock = threading.Lock()
        
    def check_limit(
        self, 
        user_id: str,
        model: str,
        tokens: int = 1
    ) -> tuple[bool, str]:
        """
        检查是否允许请求
        
        Returns:
            (is_allowed, reason)
        """
        # 1. 检查全局限流
        if not self.global_limiter.acquire(tokens):
            return False, "global_rate_limit"
            
        # 2. 检查用户限流
        with self._lock:
            if user_id not in self.user_limiters:
                self.user_limiters[user_id] = TokenBucket(
                    rate=self.user_default_rate,
                    capacity=self.user_default_capacity
                )
                
        if not self.user_limiters[user_id].acquire(tokens):
            return False, f"user_rate_limit:{user_id}"
            
        # 3. 检查模型限流
        with self._lock:
            if model not in self.model_limiters:
                # 根据模型设置不同的限流
                model_rates = {
                    "gpt-4.1": (50, 50),           # 贵模型，低限流
                    "claude-sonnet-4.5": (30, 30),
                    "gemini-2.0-flash": (500, 500),  # 便宜模型，高限流
                    "deepseek-v3.2": (200, 200)
                }
                rate, capacity = model_rates.get(model, (100, 100))
                self.model_limiters[model] = TokenBucket(rate=rate, capacity=capacity)
                
        if not self.model_limiters[model].acquire(tokens):
            return False, f"model_rate_limit:{model}"
            
        # 4. 检查熔断器
        if model in self.circuit_breakers:
            if not self.circuit_breakers[model].can_execute():
                return False, f"circuit_open:{model}"
                
        return True, "allowed"
        
    def record_success(self, model: str):
        """记录成功"""
        if model not in self.circuit_breakers:
            self.circuit_breakers[model] = CircuitBreaker()
        self.circuit_breakers[model].record_success()
        
    def record_failure(self, model: str):
        """记录失败"""
        if model not in self.circuit_breakers:
            self.circuit_breakers[model] = CircuitBreaker()
        self.circuit_breakers[model].record_failure()
        
    def get_status(self) -> Dict[str, Any]:
        """获取限流器状态"""
        return {
            "global_available": self.global_limiter.get_available(),
            "active_users": len(self.user_limiters),
            "models": {
                model: {
                    "available": limiter.get_available(),
                    "circuit_state": self.circuit_breakers.get(model, {}).state.value 
                    if model in self.circuit_breakers else "none"
                }
                for model, limiter in self.model_limiters.items()
            }
        }



使用示例

def example_rate_limiter():
    limiter = RateLimiter()
    
    # 模拟 1000 个并发请求
    successes = 0
    failures = 0
    
    for i in range(1000):
        allowed, reason = limiter.check_limit(
            user_id=f"user_{i % 100}",
            model="gemini-2.0-flash",
            tokens=100
        )
        
        if allowed:
            successes += 1
            limiter.record_success("gemini-2.0-flash")
        else:
            failures += 1
            
    print(f"✅ 成功: {successes}, ❌ 拒绝: {failures}")
    print(f"\n📊 限流器状态: {limiter.get_status()}")

if __name__ == "__main__":
    example_rate_limiter()

第四步：灰度切换与流量染色

从 OpenAI 迁移到 HolySheep 不能一步到位，需要灰度验证。我们的方案支持按用户比例、地区、请求类型等多个维度进行流量染色。

# traffic_router.py

灰度流量路由系统

import hashlib
import random
from typing import Dict, List, Optional, Callable, Any
from dataclasses import dataclass
from enum import Enum
import json
import time

class TrafficStrategy(Enum):
    BLUE_GREEN = "blue_green"        # 蓝绿部署
    CANARY = "canary"                # 金丝雀发布
    FEATURE_FLAG = "feature_flag"    # 特性开关
    AB_TEST = "ab_test"              # A/B 测试

@dataclass
class RouteRule:
    """路由规则"""
    name: str
    strategy: TrafficStrategy
    target_provider: str  # "openai" 或 "holysheep"
    percentage: float     # 流量比例 (0-100)
    conditions: Dict[str, Any] = None  # 额外条件
    
@dataclass
class TrafficMetrics:
    """流量指标"""
    provider: str
    request_count: int = 0
    success_count: int = 0
    failure_count: int = 0
    avg_latency: float = 0
    total_latency: float = 0
    
    @property
    def success_rate(self) -> float:
        if self.request_count == 0:
            return 0
        return self.success_count / self.request_count
    
    @property
    def avg_latency_ms(self) -> float:
        if self.request_count == 0:
            return 0
        return (self.total_latency / self.request_count) * 1000


class GrayReleaseManager:
    """
    灰度发布管理器
    
    支持:
    - 多维度流量染色
    - 实时指标监控
    - 自动回滚
    - 配置热更新
    """
    def __init__(self):
        self.rules: List[RouteRule] = []
        self.metrics: Dict[str, TrafficMetrics] = {
            "openai": TrafficMetrics(provider="openai"),
            "holysheep": TrafficMetrics(provider="holysheep")
        }
        self.default_provider = "holysheep"  # 默认走 HolySheep
        
        # 告警阈值
        self.p99_latency_threshold = 2000  # ms
        self.success_rate_threshold = 0.95
        self.error_rate_threshold = 0.05
        
        # 缓存最近的请求用于计算
        self.recent_latencies: Dict[str, List[float]] = {
            "openai": [],
            "holysheep": []
        }
        
    def add_rule(self, rule: RouteRule):
        """添加路由规则"""
        self.rules.append(rule)
        print(f"✅ 添加路由规则: {rule.name}, 目标: {rule.target_provider}, 比例: {rule.percentage}%")
        
    def _hash_user_id(self, user_id: str, salt: str = "") -> float:
        """一致性哈希 - 确保同一用户始终路由到同一目标"""
        combined = f"{user_id}{salt}"
        hash_value = hashlib.md5(combined.encode()).hexdigest()
        return int(hash_value[:8], 16) / 0xFFFFFFFF
        
    def _check_conditions(self, rule: RouteRule, context: Dict) -> bool:
        """检查额外条件"""
        if not rule.conditions:
            return True
            
        conditions = rule.conditions
        
        # 地区条件
        if "regions" in conditions:
            if context.get("region") not in conditions["regions"]:
                return False
                
        # 用户等级条件
        if "user_tiers" in conditions:
            if context.get("user_tier") not in conditions["user_tiers"]:
                return False
                
        # 请求类型条件
        if "request_types" in conditions:
            if context.get("request_type") not in conditions["request_types"]:
                return False
                
        return True
        
    def route(self, context: Dict) -> str:
        """
        路由决策
        
        Args:
            context: 包含 user_id, region, user_tier 等信息
            
        Returns:
            目标 provider ("openai" 或 "holysheep")
        """
        user_id = context.get("user_id", "")
        
        # 按优先级检查规则
        for rule in sorted(self.rules, key=lambda r: -r.percentage):
            if not self._check_conditions(rule, context):
                continue
                
            # 使用一致性哈希
            hash_value = self._hash_user_id(user_id, salt=rule.name)
            if hash_value * 100 < rule.percentage:
                return rule.target_provider
                
        return self.default_provider
        
    def record_request(
        self, 
        provider: str, 
        latency: float, 
        success: bool,
        tokens: int = 0
    ):
        """记录请求指标"""
        if provider not in self.metrics:
            self.metrics[provider] = TrafficMetrics(provider=provider)
            
        metrics = self.metrics[provider]
        metrics.request_count += 1
        metrics.total_latency += latency
        
        if success:
            metrics.success_count += 1
        else:
            metrics.failure_count += 1
            
        # 记录最近延迟（用于 P99 计算）
        if provider not in self.recent_latencies:
            self.recent_latencies[provider] = []
        self.recent_latencies[provider].append(latency)
        # 保留最近 1000 个样本
        self.recent_latencies[provider] = self.recent_latencies[provider][-1000:]
        
        # 更新平均延迟
        metrics.avg_latency = metrics.total_latency / metrics.request_count
        
    def should_rollback(self, provider: str) -> tuple[bool, str]:
        """
        检查是否需要回滚
        
        Returns:
            (should_rollback, reason)
        """
        if provider not in self.metrics:
            return False, ""
            
        metrics = self.metrics[provider]
        
        # 检查请求量
        if metrics.request_count < 100:
            return False, "样本不足"
            
        # 检查成功率
        if metrics.success_rate < self.success_rate_threshold:
            return True, f"成功率 {metrics.success_rate:.2%} 低于阈值 {self.success_rate_threshold:.2%}"
            
        # 检查 P99 延迟
        if self.recent_latencies.get(provider):
            sorted_latencies = sorted(self.recent_latencies[provider])
            p99_index = int(len(sorted_latencies) * 0.99)
            p99_latency = sorted_latencies[p99_index] * 1000  # 转换为 ms
            if p99_latency > self.p99_latency_threshold:
                return True, f"P99 延迟 {p99_latency:.0f}ms 超过阈值 {self.p99_latency_threshold}ms"
                
        return False, ""
        
    def get_report(self) -> Dict:
        """获取灰度报告"""
        return {
            "providers": {
                provider: {
                    "request_count": m.request_count,
                    "success_rate": f"{m.success_rate:.2%}",
                    "avg_latency_ms": f"{m.avg_latency_ms:.1f}",
                    "should_rollback": self.should_rollback(provider)[0]
                }
                for provider, m in self.metrics.items()
            },
            "rules": [
                {
                    "name": r.name,
                    "strategy": r.strategy.value,
                    "target": r.target_provider,
                    "percentage": f"{r.percentage}%"
                }
                for r in self.rules
            ]
        }



使用示例

def example_gray_release():
    manager = GrayReleaseManager()
    
    # 规则 1: 10% 流量走 OpenAI（回滚对比组）
    manager.add_rule(RouteRule(
        name="baseline_comparison",
        strategy=TrafficStrategy.CANARY,
        target_provider="openai",
        percentage=10
    ))
    
    # 规则 2: VIP 用户 100% 走 HolySheep
    manager.add_rule(RouteRule(
        name="vip_users",
        strategy=TrafficStrategy.FEATURE_FLAG,
        target_provider="holysheep",
        percentage=100,
        conditions={"user_tiers": ["vip", "enterprise"]}
    ))
    
    # 规则 3: 国内用户 100% 走 HolySheep
    manager.add_rule(RouteRule(
        name="china_users",
        strategy=TrafficStrategy.FEATURE_FLAG,
        target_provider="holysheep",
        percentage=100,
        conditions={"regions": ["cn", "hk", "tw"]}
    ))
    
    # 模拟流量
    print("\n🚀 开始灰度测试...")
    for i in range(1000):
        user_id = f"user_{i}"
        context = {
            "user_id": user_id,
            "region": "cn" if

相关资源
相关文章