对于日均处理数万次 AI 请求的企业而言，单一模型的局限性日益凸显。我曾在深圳一家 AI 创业团队负责架构设计，他们的智能客服系统每天需要处理超过 15 万次对话生成请求。团队最初依赖 OpenAI API，420ms 的平均延迟加上每月 4200 美元的账单，让 CTO 在季度评审时拍了桌子。这个真实案例促使我设计了一套完整的多模型 A/B 测试框架，并在迁移至 HolySheep AI 后，将延迟降低至 180ms，月账单压缩至 680 美元。本文将详细阐述这套框架的设计思路、代码实现与避坑指南。

一、业务背景与原方案痛点

深圳这家 AI 创业团队的智能客服系统服务于三家跨境电商客户，业务场景包括商品推荐、售后问答、物流查询等。原有架构采用单一 OpenAI GPT-4 模型，虽然响应质量尚可，但在两个维度上遇到了瓶颈。

延迟问题尤为突出。跨境电商的客服场景对响应速度极为敏感，用户期望在 2 秒内得到回复。但 OpenAI API 从香港节点绕转后，P99 延迟经常超过 800ms，部分时段甚至超时。团队曾尝试接入 Azure OpenAI Service 改善延迟，但跨境专线的月费高达 1200 美元，性价比并不理想。

成本压力同样不容忽视。GPT-4 的输入成本为 $30/MTok，输出成本为 $60/MTok，按照日均 1500 万 token 的处理量，月账单轻松突破 4000 美元。更棘手的是，不同业务场景对模型能力的要求差异很大——物流查询只需要简单的事实性回答，却要按 GPT-4 的价格付费，这显然是资源浪费。

CTO 提出的目标很明确：在保证响应质量的前提下，将延迟降低 50% 以上，成本降低 80% 以上。这个目标看似激进，但通过多模型 A/B 测试框架与智能路由，团队最终交出了延迟从 420ms 降至 180ms、成本从 $4200 降至 $680 的答卷。

二、技术方案设计

2.1 框架架构概览

多模型 A/B 测试框架的核心思路是：让多个模型同时处理相同的请求，根据预设的规则自动选择最优响应。框架分为四层：流量分发层、模型调用层、质量评估层和智能路由层。

流量分发层负责将请求按比例分流。我采用了基于请求特征的哈希分流策略，确保同一用户的请求尽量路由到同一模型，保证体验一致性。对于需要 A/B 测试的请求，按配置比例随机分配。

模型调用层封装了多个模型的调用接口。考虑到性价比，团队最终选择了 GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash 和 DeepSeek V3.2 的组合。GPT-4.1 用于高复杂度推理场景，Claude Sonnet 4.5 用于创意写作，Gemini 2.5 Flash 用于常规问答，DeepSeek V3.2 用于简单的事实查询。

质量评估层是框架的亮点。通过响应延迟、token 消耗、语义相似度、业务规则校验等多个维度综合打分，为每次响应生成质量报告。这些数据会汇总到监控看板，供团队分析各模型在不同场景下的表现。

智能路由层基于质量评估结果和业务规则，动态选择最优模型响应。路由策略支持多种模式：最低延迟模式、最低成本模式、最高质量模式，以及自定义权重模式。

2.2 核心代码实现

以下是框架的核心调用模块，采用 Python 异步架构以提升吞吐量：

import asyncio
import hashlib
import time
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import httpx

class Model(Enum):
    GPT4 = "gpt-4.1"
    CLAUDE = "claude-sonnet-4.5"
    GEMINI = "gemini-2.5-flash"
    DEEPSEEK = "deepseek-v3.2"

@dataclass
class ModelConfig:
    name: Model
    base_url: str = "https://api.holysheep.ai/v1"
    api_key: str = "YOUR_HOLYSHEEP_API_KEY"
    max_tokens: int = 2048
    temperature: float = 0.7
    input_cost_per_mtok: float = 0.0  # $/MTok
    output_cost_per_mtok: float = 0.0

@dataclass
class Response:
    model: Model
    content: str
    latency_ms: float
    input_tokens: int
    output_tokens: int
    quality_score: float = 0.0
    metadata: Dict = field(default_factory=dict)

@dataclass
class QualityMetrics:
    latency_score: float      # 延迟得分 (越低越好)
    cost_score: float         # 成本得分 (越低越好)
    quality_score: float      # 质量得分 (语义相似度)
    final_score: float = 0.0  # 综合得分

class MultiModelRouter:
    def __init__(self, configs: List[ModelConfig]):
        self.models = {c.name: c for c in configs}
        self.client = httpx.AsyncClient(timeout=30.0)
        # HolySheep 汇率优势: ¥1=$1, 无损兑换
        self.exchange_rate = 1.0  # 相比官方 $1=¥7.3, 节省 >85%
    
    async def call_model(
        self, 
        config: ModelConfig, 
        prompt: str,
        system_prompt: str = "你是一个专业的AI客服助手。"
    ) -> Response:
        """调用单个模型并记录性能指标"""
        start_time = time.perf_counter()
        
        headers = {
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": config.name.value,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt}
            ],
            "max_tokens": config.max_tokens,
            "temperature": config.temperature
        }
        
        try:
            resp = await self.client.post(
                f"{config.base_url}/chat/completions",
                headers=headers,
                json=payload
            )
            resp.raise_for_status()
            data = resp.json()
            
            latency_ms = (time.perf_counter() - start_time) * 1000
            output_tokens = data["usage"]["completion_tokens"]
            input_tokens = data["usage"]["prompt_tokens"]
            
            return Response(
                model=config.name,
                content=data["choices"][0]["message"]["content"],
                latency_ms=latency_ms,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                metadata={"raw": data}
            )
        except httpx.HTTPStatusError as e:
            raise Exception(f"API调用失败 [{e.response.status_code}]: {e.response.text}")
    
    async def ab_test(
        self,
        prompt: str,
        test_ratio: float = 0.1,
        models: Optional[List[Model]] = None
    ) -> Dict[str, Response]:
        """
        A/B 测试模式: 所有模型并行调用，返回各模型响应
        test_ratio 参数控制测试流量的比例
        """
        if models is None:
            models = list(self.models.keys())
        
        tasks = [
            self.call_model(self.models[m], prompt) 
            for m in models
        ]
        
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        result = {}
        for model, resp in zip(models, responses):
            if isinstance(resp, Exception):
                result[model.value] = None
                print(f"模型 {model.value} 调用失败: {resp}")
            else:
                result[model.value] = resp
        
        return result


HolySheep 2026年主流模型定价参考

MODEL_PRICING = {
    "gpt-4.1": {"input": 3.0, "output": 8.0},        # $/MTok
    "claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
    "gemini-2.5-flash": {"input": 0.35, "output": 2.50},
    "deepseek-v3.2": {"input": 0.07, "output": 0.42}
}


实例化路由

router = MultiModelRouter([
    ModelConfig(Model.GPT4, input_cost_per_mtok=3.0, output_cost_per_mtok=8.0),
    ModelConfig(Model.CLAUDE, input_cost_per_mtok=3.0, output_cost_per_mtok=15.0),
    ModelConfig(Model.GEMINI, input_cost_per_mtok=0.35, output_cost_per_mtok=2.50),
    ModelConfig(Model.DEEPSEEK, input_cost_per_mtok=0.07, output_cost_per_mtok=0.42),
])

async def main():
    # 测试请求
    responses = await router.ab_test(
        prompt="请用简洁的语言解释量子计算的基本原理",
        test_ratio=1.0  # 100% 测试流量
    )
    
    for model_name, resp in responses.items():
        if resp:
            cost = (resp.input_tokens / 1e6 * MODEL_PRICING[model_name]["input"] +
                   resp.output_tokens / 1e6 * MODEL_PRICING[model_name]["output"])
            print(f"{model_name}: 延迟 {resp.latency_ms:.1f}ms, "
                  f"成本 ${cost:.4f}, 内容长度 {len(resp.content)} 字符")

asyncio.run(main())

三、质量评估与智能路由

框架的核心价值在于如何从多个响应中选择最优结果。我设计了一套多维度评分机制，综合考虑延迟、成本和质量三个维度。

from sentence_transformers import SentenceTransformer
import numpy as np

class QualityEvaluator:
    def __init__(self):
        # 使用轻量级模型进行语义相似度计算
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.reference_responses = {}  # 存储各场景的标准答案
    
    def set_reference(self, scenario: str, reference: str):
        """设置场景的标准响应，用于计算语义相似度"""
        self.reference_responses[scenario] = reference
    
    def calculate_quality_score(
        self,
        response: Response,
        reference: str,
        latency_weight: float = 0.3,
        cost_weight: float = 0.2,
        quality_weight: float = 0.5
    ) -> QualityMetrics:
        """计算综合质量得分"""
        
        # 1. 延迟评分 (越低越好，归一化到 0-100)
        # 假设 200ms 以内为满分，1000ms 以上为 0
        latency_score = max(0, 100 - (response.latency_ms - 200) / 8)
        
        # 2. 成本评分 (越低越好，基于 token 消耗)
        cost = (response.input_tokens * 0.001 + 
                response.output_tokens * 0.002)  # 简化的成本计算
        cost_score = max(0, 100 - cost * 10)
        
        # 3. 质量评分 (语义相似度)
        if reference and response.content:
            emb_ref = self.embedding_model.encode([reference])
            emb_resp = self.embedding_model.encode([response.content])
            similarity = np.dot(emb_ref, emb_resp.T)[0][0]
            quality_score = similarity * 100
        else:
            quality_score = 50  # 默认分数
        
        # 4. 综合得分 (可调权重)
        final_score = (
            latency_score * latency_weight +
            cost_score * cost_weight +
            quality_score * quality_weight
        )
        
        return QualityMetrics(
            latency_score=latency_score,
            cost_score=cost_score,
            quality_score=quality_score,
            final_score=final_score
        )
    
    def select_best_response(
        self,
        responses: Dict[str, Response],
        reference: str,
        strategy: str = "balanced"
    ) -> Optional[Response]:
        """
        根据策略选择最优响应
        
        策略选项:
        - "lowest_latency": 最低延迟
        - "lowest_cost": 最低成本
        - "highest_quality": 最高质量
        - "balanced": 均衡模式
        """
        if not responses:
            return None
        
        valid_responses = {k: v for k, v in responses.items() if v is not None}
        
        if not valid_responses:
            return None
        
        scored = []
        for model_name, resp in valid_responses.items():
            metrics = self.calculate_quality_score(resp, reference)
            resp.quality_score = metrics.final_score
            scored.append((model_name, resp, metrics))
        
        if strategy == "lowest_latency":
            return min(scored, key=lambda x: x[1].latency_ms)[1]
        elif strategy == "lowest_cost":
            return min(scored, key=lambda x: x[1].output_tokens)[1]
        elif strategy == "highest_quality":
            return max(scored, key=lambda x: x[2].quality_score)[1]
        else:  # balanced
            return max(scored, key=lambda x: x[2].final_score)[1]

class SmartRouter:
    """智能路由：基于业务规则和实时指标动态选择模型"""
    
    def __init__(self, router: MultiModelRouter, evaluator: QualityEvaluator):
        self.router = router
        self.evaluator = evaluator
        self.performance_history = {}  # 模型性能历史
    
    async def route(self, request: Dict) -> Response:
        """
        核心路由逻辑：
        1. 判断场景类型
        2. 根据业务规则初筛可用模型
        3. 执行 A/B 测试
        4. 根据策略选择最优响应
        """
        scenario = request.get("scenario", "general")
        prompt = request["prompt"]
        user_id = request.get("user_id", "anonymous")
        
        # 根据场景选择候选模型
        candidate_models = self._get_candidates_by_scenario(scenario)
        
        # 决定是否启用 A/B 测试
        enable_ab = self._should_enable_ab(user_id, scenario)
        
        if enable_ab:
            # A/B 测试模式
            responses = await self.router.ab_test(
                prompt=prompt,
                models=candidate_models
            )
            
            reference = self._get_reference(scenario)
            strategy = self._get_strategy(scenario)
            return self.evaluator.select_best_response(
                responses, reference, strategy
            )
        else:
            # 直接路由模式 (生产流量)
            primary_model = self._get_primary_model(scenario)
            return await self.router.call_model(
                self.router.models[primary_model],
                prompt
            )
    
    def _get_candidates_by_scenario(self, scenario: str) -> List[Model]:
        """根据场景确定候选模型列表"""
        scenarios = {
            "logistics": [Model.DEEPSEEK, Model.GEMINI],        # 物流查询用便宜的
            "product_recommend": [Model.GPT4, Model.CLAUDE],     # 推荐用高质量的
            "refund": [Model.GEMINI, Model.DEEPSEEK],            # 退款咨询用快速的
            "general": [Model.GPT4, Model.GEMINI, Model.DEEPSEEK]
        }
        return scenarios.get(scenario, scenarios["general"])
    
    def _should_enable_ab(self, user_id: str, scenario: str) -> bool:
        """判断是否启用 A/B 测试"""
        # 按用户 ID 哈希决定分组，保证用户体验一致性
        hash_val = int(hashlib.md5(f"{user_id}:{scenario}".encode()).hexdigest(), 16)
        return (hash_val % 100) < 20  # 20% 流量进入 A/B 测试
    
    def _get_reference(self, scenario: str) -> str:
        """获取场景的标准答案"""
        # 实际应用中应从数据库或缓存获取
        return ""
    
    def _get_strategy(self, scenario: str) -> str:
        """获取场景的路由策略"""
        strategies = {
            "logistics": "lowest_cost",
            "product_recommend": "highest_quality",
            "refund": "lowest_latency",
            "general": "balanced"
        }
        return strategies.get(scenario, "balanced")
    
    def _get_primary_model(self, scenario: str) -> Model:
        """获取场景的首选模型 (非 A/B 测试流量)"""
        primary = {
            "logistics": Model.DEEPSEEK,
            "product_recommend": Model.GPT4,
            "refund": Model.GEMINI,
            "general": Model.GPT4
        }
        return primary.get(scenario, Model.GPT4)

四、灰度迁移与监控

4.1 从 OpenAI 到 HolySheep 的平滑迁移

迁移过程分为三个阶段：

阶段一：并行验证（第 1-7 天）。在不影响现有业务的前提下，将 10% 的流量切换到 HolySheep API。通过对比两个平台的响应质量、延迟和成本，验证迁移可行性。这个阶段的关键是建立完善的监控体系，确保能够及时发现问题。

阶段二：灰度放量（第 8-21 天）。将 HolySheep 的流量占比逐步提升至 50%。通过 A/B 测试框架对比两平台在真实业务场景下的表现，同时根据数据反馈微调路由策略。团队发现 DeepSeek V3.2 在物流查询场景下的表现与 GPT-4 相当，但成本仅为后者的 1/20，这个发现为后续优化提供了依据。

阶段三：全量切换（第 22-30 天）。在确认 HolySheep 稳定可靠后，将全部流量切换至新平台。同时保留 OpenAI 作为备用，当 HolySheep 出现异常时自动降级。

# 灰度控制器示例
class CanaryController:
    """灰度发布控制器"""
    
    def __init__(self, holy_sheep_config: ModelConfig, openai_config: ModelConfig):
        self.holy_sheep = holy_sheep_config
        self.openai = openai_config
        self.weights = {"holy_sheep": 0.1, "openai": 0.9}  # 初始权重
        self.metrics = {"holy_sheep": [], "openai": []}
    
    async def call(self, prompt: str, user_id: str) -> Response:
        """根据灰度权重选择后端"""
        # 按用户 ID 哈希分流，保证一致性
        hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
        use_holy_sheep = (hash_val % 100) < (self.weights["holy_sheep"] * 100)
        
        config = self.holy_sheep if use_holy_sheep else self.openai
        backend = "holy_sheep" if use_holy_sheep else "openai"
        
        try:
            if use_holy_sheep:
                resp = await self._call_holy_sheep(prompt)
            else:
                resp = await self._call_openai(prompt)
            
            self.record_metric(backend, resp, success=True)
            return resp
        except Exception as e:
            self.record_metric(backend, None, success=False, error=str(e))
            # 降级逻辑
            fallback_config = self.openai if use_holy_sheep else self.holy_sheep
            return await self._call_with_config(fallback_config, prompt)
    
    async def _call_holy_sheep(self, prompt: str) -> Response:
        """调用 HolySheep API"""
        return await self._call_with_config(self.holy_sheep, prompt)
    
    async def _call_openai(self, prompt: str) -> Response:
        """调用 OpenAI API (仅用于对比验证)"""
        return await self._call_with_config(self.openai, prompt)
    
    async def _call_with_config(self, config: ModelConfig, prompt: str) -> Response:
        """通用调用逻辑"""
        headers = {
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": config.name.value,
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": config.max_tokens
        }
        
        start = time.perf_counter()
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{config.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30.0
            )
            resp.raise_for_status()
            data = resp.json()
        
        return Response(
            model=config.name,
            content=data["choices"][0]["message"]["content"],
            latency_ms=(time.perf_counter() - start) * 1000,
            input_tokens=data["usage"]["prompt_tokens"],
            output_tokens=data["usage"]["completion_tokens"]
        )
    
    def record_metric(self, backend: str, response: Optional[Response], 
                      success: bool, error: str = None):
        """记录指标用于后续分析"""
        self.metrics[backend].append({
            "timestamp": time.time(),
            "success": success,
            "latency": response.latency_ms if response else None,
            "error": error
        })
    
    def adjust_weights(self):
        """根据监控数据动态调整灰度权重"""
        for backend in ["holy_sheep", "openai"]:
            records = self.metrics[backend][-100:]  # 最近 100 次
            if not records:
                continue
            
            success_rate = sum(1 for r in records if r["success"]) / len(records)
            avg_latency = sum(r["latency"] for r in records if r["latency"]) / len(records)
            
            # 如果 HolySheep 成功率 >99% 且延迟更低，增加权重
            if backend == "holy_sheep" and success_rate > 0.99 and avg_latency < 300:
                self.weights["holy_sheep"] = min(1.0, self.weights["holy_sheep"] + 0.1)
                self.weights["openai"] = max(0.0, self.weights["openai"] - 0.1)
                print(f"调整权重: HolySheep {self.weights['holy_sheep']:.0%}, "
                      f"OpenAI {self.weights['openai']:.0%}")


初始化配置


注意: 这里使用 HolySheep 作为主后端

canary = CanaryController(
    holy_sheep_config=ModelConfig(
        name=Model.GPT4,
        base_url="https://api.holysheep.ai/v1",  # HolySheep 直连国内
        api_key="YOUR_HOLYSHEEP_API_KEY"
    ),
    openai_config=ModelConfig(
        name=Model.GPT4,
        base_url="https://api.openai.com/v1",  # 仅用于对比验证
        api_key="YOUR_OPENAI_API_KEY"
    )
)

4.2 监控与告警

生产环境的监控是框架稳定运行的关键。建议关注以下核心指标：

团队在监控看板中设置了以下告警规则：单一模型连续 5 次调用失败、整体成功率低于 99.5%、P99 延迟超过 1 秒、成本环比增长超过 20%。这些阈值可以根据实际业务调整。

五、30 天运营数据与成本分析

框架上线 30 天后的数据远超预期：

指标 迁移前 (OpenAI) 迁移后 (HolySheep) 改善幅度
平均延迟 420ms 180ms ↓57%
P99 延迟 850ms 320ms ↓62%
月 Token 消耗 450 亿 480 亿 ↑7%
月度账单 $4,200 $680 ↓84%
API 成功率 99.2% 99.8% ↑0.6%
用户满意度 4.1/5 4.6/5 ↑12%

成本的显著降低主要来自三个方面：首先是 HolySheep 的汇率优势，人民币充值按 ¥1=$1 结算，相比官方 $1=¥7.3 的汇率，节省超过 85%；其次是智能路由策略将 60% 的简单查询路由到 DeepSeek V3.2，成本仅为 GPT-4 的 5%；最后是流量优化，通过 prompt 压缩和缓存机制减少了 15% 的 token 消耗。

六、价格与回本测算

模型 输入价格 ($/MTok) 输出价格 ($/MTok) 适用场景 性价比评分
GPT-4.1 $3.00 $8.00 复杂推理、代码生成 ★★★
Claude Sonnet 4.5 $3.00 $15.00 创意写作、长文本分析 ★★
Gemini 2.5 Flash $0.35 $2.50 常规问答、批量处理 ★★★★★
DeepSeek V3.2 $0.07 $0.42 简单查询、事实性回答 ★★★★★

以月处理 500 亿 token 的中等规模业务为例，采用 HolySheep 的多模型架构后，月成本约为 $680。如果继续使用单一 GPT-4 模型，同样规模的成本将超过 $4,200。这意味着每年可节省超过 4 万美元。

框架本身的开发和运维成本可控：一位后端工程师用两周时间完成开发，后续每周约需 4 小时维护。按月薪 2 万元计算，年度人力成本约 10 万元。相比每年节省的 4 万美元（约 28 万人民币），ROI 超过 180%。

七、适合谁与不适合谁

适合的场景

不适合的场景

八、为什么选 HolySheep

市场上存在多家大模型 API 中转服务商，选择 HolySheep 有以下核心优势：

汇率优势是首要考量。HolySheep 采用 ¥1=$1 的无损汇率，相比官方 $1=¥7.3 的汇率，节省超过 85%。对于月消耗数千美元的业务，这个差异意味着每年数万人民币的节省。同时支持微信、支付宝充值，对国内开发者非常友好。

国内直连延迟低是第二优势。HolySheep 在国内部署了接入节点，从大陆访问延迟可以控制在 50ms 以内。相比之下，直接调用 OpenAI 或 Anthropic API 需要绕道海外，延迟通常在 200-500ms，稳定性也难以保证。

模型覆盖全面满足一站式需求。支持 GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2 等 2026 年主流模型，无需对接多个供应商，降低集成复杂度。

注册即可试用降低决策门槛。新用户注册即可获得免费额度，可以在生产环境验证前先进行技术评估。

常见报错排查

错误一：401 Authentication Error

错误信息：{"error": {"message": "Incorrect API key provided", "type": "invalid_request_error", "code": "401"}}

常见原因：API Key 填写错误或已过期。检查 base_url 是否正确设置为 https://api.holysheep.ai/v1，API Key 格式是否为 sk-... 开头。

解决代码：

# 正确配置示例
import os


方式一：直接设置环境变量

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"


方式二：传入配置对象

config = ModelConfig(
    name=Model.GPT4,
    base_url="https://api.holysheep.ai/v1",  # 注意结尾无斜杠
    api_key="YOUR_HOLYSHEEP_API_KEY"
)


验证连接

async def verify_connection():
    client = httpx.AsyncClient()
    resp = await client.get(
        "https://api.holysheep.ai/v1/models",
        headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
    )
    print(resp.json())  # 应返回可用模型列表
    await client.aclose()

错误二：429 Rate Limit Exceeded

错误信息：{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error", "code": 429}}

常见原因：请求频率超出套餐限制，或短期内 token 消耗过大。

解决代码：

import asyncio
from typing import Optional

class RateLimitedClient:
    """带速率限制的 API 客户端"""
    
    def __init__(self, config: ModelConfig, max_rpm: int = 60):
        self.config = config
        self.max_rpm = max_rpm
        self.request_times = []
        self._lock = asyncio.Lock()
    
    async def call_with_retry(
        self, 
        prompt: str, 
        max_retries: int = 3,
        backoff_base: float = 1.0
    ) -> Optional[Response]:
        """带指数退避的调用"""
        for attempt in range(max_retries):
            try:
                return await self._call(prompt)
            except Exception as e:
                if "429" in str(e) and attempt < max_retries - 1:
                    # 指数退避等待
                    wait_time = backoff_base * (2 ** attempt)
                    print(f"触发限流，等待 {wait_time} 秒后重试...")
                    await asyncio.sleep(wait_time)
                else:
                    raise e
        return None
    
    async def _call(self, prompt: str) -> Response:
        """速率限制内的调用"""
        async with self._lock:
            now = time.time()
            # 清理超过 1 分钟的记录
            self.request_times = [t for t in self.request_times if now - t < 60]
            
            if len(self.request_times) >= self.max_rpm:
                # 需要等待
                oldest = self.request_times[0]
                wait = 60 - (now - oldest

相关资源
相关文章