作为 HolySheep AI 技术团队的核心架构师，我在过去三年帮助了超过 200 家企业完成 AI API 的迁移与优化。今天我想用我们服务的一个真实案例——深圳某 AI 创业团队的具体实践，来详细讲解如何构建健壮的 AI API 调用体系。如果你正在为 API 不稳定、成本失控而头疼，这篇文章将提供完整的工程解决方案。

业务背景：从频繁超时到丝滑调用

2024 年第三季度，我们接触了一家深圳的 AI 创业团队，他们主营智能客服与内容生成业务。团队 CTO 林工找到我们时，业务正处于爆发期：日均 API 调用量突破 50 万次，但系统稳定性却成了噩梦。深夜的告警电话、用户的投诉反馈、每月底的天价账单——这些问题严重制约着公司的发展。

原方案的三大痛点：

林工尝试过多种优化方案：更换代理服务商、增加本地缓存、调整超时参数。但每次改动都像在打补丁，系统复杂度急剧上升。问题的根源在于——缺乏系统级的错误处理与容错机制。这也是我们遇到的大多数客户的共性问题。

为什么选择 HolyShehep AI

在评估了多个方案后，林工的团队决定切换到 HolySheep AI。他们的 CTO 总结了三个核心决策因素：

更关键的是，HolySheep API 提供了完善的错误码体系和标准化的响应格式，为我们实现重试逻辑和熔断器提供了坚实基础。

整体架构设计

在开始代码实现之前，先理解我们的整体架构设计思路：

环境准备与基础配置

首先安装必要的依赖包，我们使用 Python 作为主要演示语言，生态丰富且易于理解：

# 安装核心依赖
pip install requests tenacity httpx slowapi

推荐使用 tenacity，它内置了指数退避和抖动支持

pip install "tenacity>=7.0.0"

然后配置 HolySheep API 的基础信息。注意：请勿使用任何包含 api.openai.com 或 api.anthropic.com 的配置，我们只使用 HolySheep 官方端点：

# config.py
import os


HolySheep AI 官方配置

HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": os.getenv("HOLYSHEEP_API_KEY"),  # 替换为你的密钥
    "model": "gpt-4.1",  # 或 claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2
    "timeout": 30,
    "max_retries": 3,
}


价格参考（2026年主流模型 output 价格 / MTok）

MODEL_PRICES = {
    "gpt-4.1": 8.0,           # $8 / MT
    "claude-sonnet-4.5": 15.0,  # $15 / MT
    "gemini-2.5-flash": 2.50,   # $2.50 / MT
    "deepseek-v3.2": 0.42,      # $0.42 / MT（成本最低）
}


熔断器配置

CIRCUIT_BREAKER_CONFIG = {
    "failure_threshold": 5,      # 连续失败5次后开启熔断
    "recovery_timeout": 60,      # 60秒后尝试半开
    "half_open_max_calls": 3,    # 半开状态最多允许3个请求
}

重试逻辑实现

重试是应对瞬时故障的第一道防线。但错误的重试策略比没有重试更糟糕——我见过太多系统因为暴力重试导致雪崩。下面是经过生产验证的重试实现：

# retry_handler.py
import time
import logging
from functools import wraps
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)

logger = logging.getLogger(__name__)


定义可重试的异常类型

class RetryableError(Exception):
    """可重试的错误基类"""
    pass

class RateLimitError(RetryableError):
    """限流错误 - 应该等待后重试"""
    pass

class TimeoutError(RetryableError):
    """超时错误 - 网络抖动导致"""
    pass

class ServiceUnavailableError(RetryableError):
    """服务不可用 - 短暂故障"""
    pass


HolySheep API 专用重试装饰器

def holy_sheep_retry(max_attempts=3, min_wait=1, max_wait=10):
    """
    HolySheep AI 专用重试装饰器
    
    参数:
        max_attempts: 最大重试次数
        min_wait: 最小等待时间（秒）
        max_wait: 最大等待时间（秒）
    
    实战经验：
    - 我们发现 AI API 的瞬时故障通常在3秒内恢复
    - 指数退避配合抖动可以避免多实例同时重试
    """
    return retry(
        stop=stop_after_attempt(max_attempts),
        wait=wait_exponential(
            multiplier=1,
            min=min_wait,
            max=max_wait,
            exp_base=2,  # 等待时间序列: 1s, 2s, 4s...
        ),
        retry=retry_if_exception_type(RetryableError),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )


请求级别的重试处理

import requests

class HolySheepClient:
    """HolySheep AI API 客户端，包含完整的错误处理"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
    
    def _handle_response_error(self, response: requests.Response) -> None:
        """根据 HTTP 状态码判断是否可重试"""
        status_code = response.status_code
        
        if status_code == 429:
            # 限流 - 必须重试
            raise RateLimitError(
                f"Rate limited. Retry-After: {response.headers.get('Retry-After', 'unknown')}"
            )
        elif status_code == 500 or status_code == 502 or status_code == 503:
            # 服务端错误 - 可重试
            raise ServiceUnavailableError(f"Server error: {status_code}")
        elif status_code == 408:
            # 请求超时 - 可重试
            raise TimeoutError("Request timeout")
        elif status_code == 401 or status_code == 403:
            # 认证错误 - 不重试
            raise PermissionError(f"Auth error: {status_code}")
    
    @holy_sheep_retry(max_attempts=3, min_wait=1, max_wait=8)
    def chat_completion(self, messages: list, model: str = "gpt-4.1", **kwargs):
        """
        调用 HolySheep AI 聊天完成接口
        
        Args:
            messages: 消息列表，格式同 OpenAI
            model: 模型名称
            **kwargs: 其他参数如 temperature, max_tokens
        
        Returns:
            dict: API 响应结果
        
        实战经验：
        - 我们在生产环境发现，大部分临时故障在第一次重试时就恢复了
        - 第二次重试恢复率约 95%，第三次基本没有必要
        - 但保留3次是为了应对极端网络抖动情况
        """
        try:
            response = self.session.post(
                f"{self.base_url}/chat/completions",
                json={
                    "model": model,
                    "messages": messages,
                    **kwargs,
                },
                timeout=30,
            )
            
            # 处理业务层面的错误
            if response.status_code != 200:
                self._handle_response_error(response)
            
            return response.json()
            
        except requests.exceptions.Timeout:
            raise TimeoutError("Request timeout after 30s")
        except requests.exceptions.ConnectionError as e:
            # 网络错误 - 可重试
            raise TimeoutError(f"Connection error: {str(e)}")


使用示例

if __name__ == "__main__":
    import os
    
    client = HolySheepClient(
        api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    )
    
    messages = [
        {"role": "system", "content": "你是一个专业的客服助手"},
        {"role": "user", "content": "我想了解产品的价格和功能"},
    ]
    
    try:
        result = client.chat_completion(messages, model="deepseek-v3.2")
        print(f"响应: {result['choices'][0]['message']['content']}")
    except RetryableError as e:
        logger.error(f"重试耗尽仍失败: {e}")
    except PermissionError as e:
        logger.error(f"认证错误，请检查 API Key: {e}")

熔断器模式实现

熔断器是防止级联故障的关键组件。当某个服务的错误率超过阈值时，熔断器会"跳闸"，快速拒绝后续请求，避免资源耗尽和雪崩效应。我见过太多团队因为没有熔断器，在上游 API 故障时整个系统崩溃的场景。

# circuit_breaker.py
import time
import threading
from enum import Enum
from typing import Callable, Any
from dataclasses import dataclass, field
from collections import defaultdict
import logging

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 设为 5，连续5次失败才触发熔断
    - recovery_timeout 设为 60秒，给上游服务足够的恢复时间
    - 半开状态限制并发数，避免同时涌入大量请求
    """
    
    name: str
    failure_threshold: int = 5      # 触发熔断的连续失败次数
    recovery_timeout: int = 60       # 熔断持续时间（秒）
    half_open_max_calls: int = 3    # 半开状态允许的试探请求数
    
    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    _failure_count: int = field(default=0, init=False)
    _success_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0.0, init=False)
    _half_open_calls: int = field(default=0, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)
    
    @property
    def state(self) -> CircuitState:
        """获取当前状态，必要时进行状态转换"""
        with self._lock:
            if self._state == CircuitState.OPEN:
                # 检查是否应该进入半开状态
                if time.time() - self._last_failure_time >= self.recovery_timeout:
                    logger.info(f"CircuitBreaker '{self.name}': OPEN → HALF_OPEN")
                    self._state = CircuitState.HALF_OPEN
                    self._half_open_calls = 0
            return self._state
    
    def can_execute(self) -> bool:
        """检查是否可以执行请求"""
        current_state = self.state
        
        if current_state == CircuitState.CLOSED:
            return True
        
        if current_state == CircuitState.OPEN:
            return False
        
        # 半开状态，限制并发数
        if current_state == CircuitState.HALF_OPEN:
            with self._lock:
                if self._half_open_calls < self.half_open_max_calls:
                    self._half_open_calls += 1
                    return True
                return False
    
    def record_success(self) -> None:
        """记录成功调用"""
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                # 连续成功次数超过阈值则关闭熔断
                if self._success_count >= 2:  # 2次成功即关闭
                    logger.info(f"CircuitBreaker '{self.name}': HALF_OPEN → CLOSED")
                    self._state = CircuitState.CLOSED
                    self._failure_count = 0
                    self._success_count = 0
            elif self._state == CircuitState.CLOSED:
                # 成功时重置失败计数
                self._failure_count = max(0, self._failure_count - 1)
    
    def record_failure(self) -> None:
        """记录失败调用"""
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.time()
            
            if self._state == CircuitState.HALF_OPEN:
                # 半开状态失败，立即打开熔断
                logger.warning(f"CircuitBreaker '{self.name}': HALF_OPEN → OPEN")
                self._state = CircuitState.OPEN
                self._success_count = 0
            
            elif self._state == CircuitState.CLOSED:
                if self._failure_count >= self.failure_threshold:
                    logger.warning(
                        f"CircuitBreaker '{self.name}': CLOSED → OPEN "
                        f"(failures: {self._failure_count})"
                    )
                    self._state = CircuitState.OPEN
    
    def get_stats(self) -> dict:
        """获取熔断器统计信息"""
        with self._lock:
            return {
                "name": self.name,
                "state": self._state.value,
                "failure_count": self._failure_count,
                "last_failure_time": self._last_failure_time,
            }


def circuit_breaker_protect(
    circuit_breaker: CircuitBreaker,
    fallback: Any = None,
):
    """
    熔断器保护装饰器
    
    Args:
        circuit_breaker: 熔断器实例
        fallback: 熔断时的降级处理函数
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            if not circuit_breaker.can_execute():
                logger.warning(
                    f"CircuitBreaker '{circuit_breaker.name}' is OPEN, "
                    f"using fallback"
                )
                if fallback:
                    return fallback(*args, **kwargs)
                raise CircuitOpenError(
                    f"Circuit breaker '{circuit_breaker.name}' is open"
                )
            
            try:
                result = func(*args, **kwargs)
                circuit_breaker.record_success()
                return result
            except Exception as e:
                circuit_breaker.record_failure()
                raise
        
        return wrapper
    return decorator


class CircuitOpenError(Exception):
    """熔断器开启异常"""
    pass



实际使用示例

class AIService:
    """集成熔断器的 AI 服务"""
    
    def __init__(self, api_key: str):
        self.client = HolySheepClient(api_key)
        # 为不同模型创建独立的熔断器
        self.circuit_breakers = {
            "gpt-4.1": CircuitBreaker("gpt-4.1", failure_threshold=5),
            "deepseek-v3.2": CircuitBreaker("deepseek-v3.2", failure_threshold=3),
        }
    
    def _get_circuit_breaker(self, model: str) -> CircuitBreaker:
        """获取对应模型的熔断器"""
        return self.circuit_breakers.get(
            model,
            CircuitBreaker(model, failure_threshold=5)
        )
    
    def generate_response(
        self,
        messages: list,
        model: str = "deepseek-v3.2",
        use_fallback_model: bool = True,
    ) -> dict:
        """
        生成回复，支持熔断和模型降级
        
        实战经验：
        - 当主模型熔断时，自动切换到备选模型（如从 gpt-4.1 切换到 deepseek-v3.2）
        - deepseek-v3.2 价格仅 $0.42/MT，是成本最低的选择
        - 降级策略可以在保证服务可用性的同时节省 80%+ 成本
        """
        circuit = self._get_circuit_breaker(model)
        
        try:
            return self.client.chat_completion(messages, model=model)
        except (RetryableError, ServiceUnavailableError) as e:
            logger.error(f"Model {model} failed: {e}")
            
            # 尝试降级到备选模型
            if use_fallback_model and model != "deepseek-v3.2":
                fallback_model = "deepseek-v3.2"
                logger.info(f"Falling back to {fallback_model}")
                fallback_circuit = self._get_circuit_breaker(fallback_model)
                
                if fallback_circuit.can_execute():
                    try:
                        result = self.client.chat_completion(
                            messages,
                            model=fallback_model
                        )
                        fallback_circuit.record_success()
                        return result
                    except Exception:
                        fallback_circuit.record_failure()
            
            raise



熔断器监控（生产环境建议集成到 Prometheus）

def monitor_circuits(service: AIService):
    """定期输出熔断器状态"""
    while True:
        for name, cb in service.circuit_breakers.items():
            stats = cb.get_stats()
            if stats["state"] != "closed":
                logger.warning(f"Circuit breaker alert: {stats}")
        time.sleep(10)

完整集成示例

将重试逻辑、熔断器和 HolySheep API 集成在一起，形成完整的错误处理方案：

# main.py - 完整集成示例
import os
import logging
from datetime import datetime
from typing import Optional


配置日志

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class ResilientAIProvider:
    """
    弹性 AI 供应商
    
    整合重试、熔断、降级、监控的完整解决方案
    针对 HolySheep AI 进行了优化配置
    """
    
    def __init__(self, api_key: str):
        self.client = HolySheepClient(api_key)
        self.circuit_breaker = CircuitBreaker(
            name="holy_sheep_main",
            failure_threshold=5,
            recovery_timeout=60,
        )
        
        # 降级策略：按成本和稳定性排序
        self.models = [
            {"name": "deepseek-v3.2", "cost": 0.42, "priority": 1},   # 最低成本
            {"name": "gemini-2.5-flash", "cost": 2.50, "priority": 2},
            {"name": "gpt-4.1", "cost": 8.0, "priority": 3},
            {"name": "claude-sonnet-4.5", "cost": 15.0, "priority": 4},
        ]
        
        # 统计指标
        self.stats = {
            "total_calls": 0,
            "successful_calls": 0,
            "failed_calls": 0,
            "fallback_calls": 0,
            "circuit_open_count": 0,
        }
    
    def _get_fallback_model(self, current_model: str) -> Optional[dict]:
        """获取降级模型"""
        current_priority = None
        for m in self.models:
            if m["name"] == current_model:
                current_priority = m["priority"]
                break
        
        for m in sorted(self.models, key=lambda x: x["priority"]):
            if m["priority"] > (current_priority or 0):
                return m
        return None
    
    @holy_sheep_retry(max_attempts=3, min_wait=1, max_wait=8)
    def _call_api(self, messages: list, model: str) -> dict:
        """带重试的 API 调用"""
        return self.client.chat_completion(messages, model=model)
    
    def generate(
        self,
        messages: list,
        model: str = "deepseek-v3.2",
        enable_fallback: bool = True,
    ) -> dict:
        """
        生成回复的主入口
        
        特性:
        1. 自动重试（指数退避）
        2. 熔断器保护
        3. 模型降级
        4. 详细日志记录
        """
        self.stats["total_calls"] += 1
        start_time = datetime.now()
        
        # 检查熔断器
        if not self.circuit_breaker.can_execute():
            logger.warning("Circuit breaker is OPEN, attempting fallback")
            self.stats["circuit_open_count"] += 1
            
            if not enable_fallback:
                raise CircuitOpenError("Service temporarily unavailable")
            
            fallback = self._get_fallback_model(model)
            if fallback:
                model = fallback["name"]
                logger.info(f"Using fallback model: {model}")
        
        try:
            result = self._call_api(messages, model)
            self.circuit_breaker.record_success()
            self.stats["successful_calls"] += 1
            
            latency = (datetime.now() - start_time).total_seconds() * 1000
            logger.info(f"Success: model={model}, latency={latency:.2f}ms")
            
            return result
            
        except (RateLimitError, TimeoutError, ServiceUnavailableError) as e:
            self.circuit_breaker.record_failure()
            
            if enable_fallback and model != "deepseek-v3.2":
                self.stats["fallback_calls"] += 1
                logger.warning(f"Primary model failed, trying fallback: {e}")
                
                fallback = self._get_fallback_model(model)
                if fallback:
                    try:
                        result = self._call_api(
                            messages,
                            model=fallback["name"]
                        )
                        self.stats["successful_calls"] += 1
                        return result
                    except Exception:
                        pass
            
            self.stats["failed_calls"] += 1
            logger.error(f"All models failed: {e}")
            raise
        
        except Exception as e:
            self.circuit_breaker.record_failure()
            self.stats["failed_calls"] += 1
            logger.error(f"Unexpected error: {e}")
            raise
    
    def get_stats(self) -> dict:
        """获取服务统计"""
        total = self.stats["total_calls"]
        success_rate = (
            self.stats["successful_calls"] / total * 100
            if total > 0 else 0
        )
        
        return {
            **self.stats,
            "success_rate": f"{success_rate:.2f}%",
            "circuit_state": self.circuit_breaker.state.value,
        }



生产环境使用示例

if __name__ == "__main__":
    # 初始化（从环境变量读取 API Key）
    api_key = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    provider = ResilientAIProvider(api_key)
    
    # 模拟请求
    test_messages = [
        {"role": "user", "content": "请介绍一下深圳的气候特点"}
    ]
    
    try:
        response = provider.generate(
            messages=test_messages,
            model="deepseek-v3.2",  # 成本最优选择
            enable_fallback=True,
        )
        print(f"Response: {response['choices'][0]['message']['content']}")
        
        # 打印统计
        print(f"\nService Stats:")
        for key, value in provider.get_stats().items():
            print(f"  {key}: {value}")
            
    except Exception as e:
        logger.error(f"Request failed after all retries: {e}")

灰度切换与密钥轮换策略

从原有方案切换到 HolySheep AI 时，建议采用灰度发布策略，逐步将流量迁移到新系统。这样可以及时发现问题并回滚，将业务影响降到最低。

密钥轮换建议：

# key_rotation.py - 多密钥负载均衡与自动轮换
import os
import time
import threading
from collections import defaultdict
from typing import List, Optional
import random


class KeyPool:
    """
    API Key 池，支持负载均衡和自动轮换
    
    实战经验：
    - 使用 Key 池可以避免单 Key 限流
    - 配合熔断器，可以自动跳过有问题的 Key
    - 建议保留 2-3 个活跃 Key
    """
    
    def __init__(self, keys: List[str]):
        self.keys = keys
        self.current_index = 0
        self.key_health = defaultdict(lambda: {"failures": 0, "last_failure": 0})
        self._lock = threading.Lock()
        
        # 每个 Key 的故障阈值
        self.failure_threshold = 10
        self.failure_cooldown = 300  # 5分钟冷却期
    
    def get_healthy_key(self) -> Optional[str]:
        """获取一个健康的 Key"""
        with self._lock:
            now = time.time()
            
            # 尝试所有 Key，找一个健康的
            for _ in range(len(self.keys)):
                self.current_index = (self.current_index + 1) % len(self.keys)
                key = self.keys[self.current_index]
                
                health = self.key_health[key]
                
                # 检查是否在冷却期
                if now - health["last_failure"] < self.failure_cooldown:
                    continue
                
                # 检查失败次数
                if health["failures"] >= self.failure_threshold:
                    continue
                
                return key
            
            return None  # 所有 Key 都不健康
    
    def record_success(self, key: str):
        """记录成功"""
        with self._lock:
            self.key_health[key]["failures"] = 0
    
    def record_failure(self, key: str):
        """记录失败"""
        with self._lock:
            health = self.key_health[key]
            health["failures"] += 1
            health["last_failure"] = time.time()
    
    def get_stats(self) -> dict:
        """获取所有 Key 的健康状态"""
        with self._lock:
            return {
                key: dict(self.key_health[key])
                for key in self.keys
            }



使用示例

if __name__ == "__main__":
    # 从环境变量读取多个 Key（用逗号分隔）
    keys_str = os.getenv("HOLYSHEEP_API_KEYS", "YOUR_KEY_1,YOUR_KEY_2")
    key_pool = KeyPool(keys_str.split(","))
    
    # 负载均衡获取 Key
    for i in range(5):
        key = key_pool.get_healthy_key()
        print(f"Request {i+1} using key: {key[:10]}...")

上线后 30 天性能数据

深圳这家 AI 创业团队在完成迁移后，我们持续跟踪了 30 天的运营数据，效果远超预期：

指标迁移前迁移后提升幅度
平均延迟420ms28ms降93%
P99 延迟2,100ms85ms降96%
月均账单$4,200$680降84%
可用性96.5%99.8%+3.3%
故障恢复时间45分钟3分钟降93%
无效重试率12%0.5%降96%

林工反馈说："迁移到 HolySheep 后，最大的感受是稳定性和成本的双重优化。以前每月 $4,200 的账单让我们压力很大，现在 $680 就能覆盖同样的业务量，而且服务更稳定了。"

成本优化建议

基于我们服务 200+ 企业的经验，以下是成本优化的最佳实践：

常见报错排查

在集成 HolySheep AI API 时，以下是开发者最常遇到的 6 个问题及其解决方案：

1. AuthenticationError: Invalid API Key

错误信息：{"error": {"message": "Invalid API Key", "type": "invalid_request_error"}}

可能原因：

解决方案：

# 检查 API Key 格式
import os


正确的 Key 格式：sk-hs- 开头

api_key = os.getenv("HOLYSHEEP_API_KEY")
print(f"Key length: {len(api_key)}")
print(f"Key prefix: {api_key[:10]}...")


验证 Key 是否有效

from config import HOLYSHEEP_CONFIG
client = HolySheepClient(api_key=api_key)

try:
    # 测试调用
    client.chat_completion([
        {"role": "user", "content": "test"}
    ], model="deepseek-v3.2")
    print("✓ API Key valid")
except Exception as e:
    print(f"✗ API Key error: {e}")

2. RateLimitError: 请求频率超限

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

可能原因：

解决方案：

# 实现请求限流
import time
import threading
from collections import deque

class RateLimiter:
    """令牌桶限流器"""
    
    def __init__(self, max_calls: int, period: float):
        """
        Args:
            max_calls: period 时间内的最大调用次数
            period: 时间周期（秒）
        """
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
        self._lock = threading.Lock()
    
    def acquire(self) -> float:
        """
        获取许可，如果需要等待则返回等待时间
        
        Returns:
            实际等待时间（秒）
        """
        with self._lock:
            now = time.time()
            
            # 清理过期的调用记录
            while self.calls and self.calls[0] < now - self.period:
                self.calls.popleft()
            
            if len(self.calls) < self.max_calls:
                self.calls.append(now)
                return 0
            
            # 需要等待
            wait_time = self.calls[0] + self.period - now
            return wait_time
    
    def __enter__(self):
        wait = self.acquire()
        if wait > 0:
            time.sleep(wait)
        return self



使用限流器

limiter = RateLimiter(max_calls=50, period=60)  # 50 QPM

with limiter:
    result = client.chat_completion(messages, model="deepseek-v3.2")

3. TimeoutError: 请求超时

错误信息：TimeoutError: Request timeout after 30s

可能原因：

解决方案：

# 1. 检查网络连通性
import socket

def check_h联通性():
    host = "api.holys

相关资源
相关文章