Vous en avez marre de gérer séparément vos appels à GPT-4, Claude et Gemini ? Vous cherchez une solution unique qui vous fait économiser 85% sur vos factures API tout en offrant une latence inférieure à 50ms ? La réponse s'appelle HolySheep API聚合平台. Dans ce guide complet, je vais vous montrer comment configurer un système de load balancing multi-modèle professionnel qui路由自动切换最优供应商.
Comparatif des plateformes API — HolySheep vs Officielles vs Concurrents
| Critère | HolySheep API | API OpenAI | API Anthropic | Concurrents |
|---|---|---|---|---|
| Prix GPT-4.1 | $8/MTok | $60/MTok | N/A | $45-55/MTok |
| Prix Claude Sonnet 4.5 | $15/MTok | N/A | $18/MTok | $16-20/MTok |
| Prix Gemini 2.5 Flash | $2.50/MTok | N/A | N/A | $3-5/MTok |
| Prix DeepSeek V3.2 | $0.42/MTok | N/A | N/A | $0.50-1/MTok |
| Latence moyenne | <50ms ✅ | 200-500ms | 300-800ms | 100-300ms |
| Paiements | WeChat/Alipay/Cartes | Cartes internationales | Cartes internationales | Variables |
| Multi-modèle unifié | ✅ 20+ modèles | ❌ OpenAI only | ❌ Anthropic only | ⚠️ Limité |
| Load balancing intégré | ✅ Native | ❌ | ❌ | ⚠️ Basique |
| Crédits gratuits | ✅ Offerts | ❌ | ❌ | ⚠️ Limités |
| Profil idéal | Tous profils | Développeurs USA | Enterprise US | Variables |
Pour qui / Pour qui ce n'est pas fait
✅ Cette solution est faite pour vous si :
- Vous utilisez plusieurs modèles IA et souhaitez unifyr la gestion
- Vous cherchez à réduire vos coûts API de 85% ou plus
- Vous êtes basé en Chine ou en Asie et avez besoin de WeChat/Alipay
- Vous avez besoin d'une latence inférieure à 50ms pour vos applications temps réel
- Vous souhaitez un load balancing automatique entre fournisseurs
- Vous voulez éviter les blocages géographiques et les problèmes de cartes internationales
❌ Cette solution n'est pas faite pour vous si :
- Vous avez uniquement besoin d'API officielles avec support enterprise contractuel direct
- Vous êtes dans un pays où les regulations interdisent l'utilisation de proxies API
- Vous nécessitez des garanties SLA enterprise avec support 24/7 dédié
- Vous n'avez pas de besoins multi-modèles et une seule API suffit
Tarification et ROI
Avec les tarifs HolySheep 2026, le retour sur investissement est immédiat :
- GPT-4.1 : $8/MTok vs $60/MTok officiel = 86% d'économie
- Claude Sonnet 4.5 : $15/MTok vs $18/MTok officiel = 16% d'économie
- Gemini 2.5 Flash : $2.50/MTok — modèle ultra-économique pour tâches simples
- DeepSeek V3.2 : $0.42/MTok — le modèle le plus économique du marché
Exemple concret : Si vous consommez $1000/mois en API OpenAI, passer à HolySheep vous coûtera environ $133/mois — soit $867 économisés chaque mois, ou $10,404/an.
Pourquoi choisir HolySheep
En tant que développeur qui a testé des dizaines de solutions API, HolySheep se distingue par :
- Une平台 unique pour tous vos modèles — Plus besoin de gérer 10 clés API différentes
- Un système de load balancing intelligent qui route automatiquement vers le modèle optimal
- Une latence inférieure à 50ms grâce à l'infrastructure optimisée
- Des économies de 85%+ sur les modèles principaux
- Paiement local simplifié avec WeChat et Alipay
- Crédits gratuits pour tester la plateforme
Configuration du Load Balancing Multi-Modèle
Passons maintenant à la pratique. Je vais vous montrer comment configurer un système de load balancing professionnel avec HolySheep.
Prérequis
- Un compte HolySheep API — créez le vôtre ici
- Votre clé API HolySheep
- Python 3.8+ ou Node.js 18+
Installation du SDK
# Installation Python
pip install requests httpx
Installation Node.js
npm install axios node-fetchConfiguration de base avec Load Balancing
import requests
import json
from typing import List, Dict, Optional
import time
class HolySheepLoadBalancer:
"""
Load Balancer Multi-Modèle pour HolySheep API
Auteur: Équipe HolySheep AI
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Configuration des modèles avec leurs poids pour le load balancing
self.model_config = {
"gpt4": {
"model": "gpt-4.1",
"weight": 30, # 30% du trafic
"cost_per_mtok": 8.0,
"latency_threshold": 2000 # ms
},
"claude": {
"model": "claude-sonnet-4.5",
"weight": 25, # 25% du trafic
"cost_per_mtok": 15.0,
"latency_threshold": 2500
},
"gemini": {
"model": "gemini-2.5-flash",
"weight": 30, # 30% du trafic - économique
"cost_per_mtok": 2.50,
"latency_threshold": 1000
},
"deepseek": {
"model": "deepseek-v3.2",
"weight": 15, # 15% du trafic - très économique
"cost_per_mtok": 0.42,
"latency_threshold": 800
}
}
# Statistiques pour le monitoring
self.stats = {
model: {"requests": 0, "errors": 0, "total_latency": 0}
for model in self.model_config.keys()
}
def select_model_weighted(self) -> str:
"""Sélectionne un modèle selon les poids configurés"""
import random
total_weight = sum(cfg["weight"] for cfg in self.model_config.values())
rand = random.uniform(0, total_weight)
cumulative = 0
for model_name, config in self.model_config.items():
cumulative += config["weight"]
if rand <= cumulative:
return model_name
return "gemini" # fallback par défaut
def select_model_by_load(self) -> str:
"""Sélectionne le modèle avec la charge la plus faible"""
min_requests = float('inf')
selected = "gemini"
for model_name, stat in self.stats.items():
if stat["requests"] < min_requests:
min_requests = stat["requests"]
selected = model_name
return selected
def select_model_smart(self, task_type: str = "general") -> str:
"""
Sélection intelligente du modèle selon le type de tâche
Args:
task_type: "general", "coding", "reasoning", "fast", "cheap"
"""
if task_type == "coding":
return "claude" # Excellent pour le code
elif task_type == "reasoning":
return "gpt4" # Meilleure capacité de raisonnement
elif task_type == "fast":
return "gemini" # Latence minimale
elif task_type == "cheap":
return "deepseek" # Coût minimal
else:
return self.select_model_weighted()
def chat_completion(
self,
messages: List[Dict],
model_strategy: str = "weighted",
task_type: str = "general",
**kwargs
) -> Dict:
"""
Effectue un appel API avec load balancing
Args:
messages: Liste des messages [{"role": "user", "content": "..."}]
model_strategy: "weighted", "load", "smart"
task_type: "general", "coding", "reasoning", "fast", "cheap"
"""
# Sélection du modèle selon la stratégie
if model_strategy == "smart":
model_key = self.select_model_smart(task_type)
elif model_strategy == "load":
model_key = self.select_model_by_load()
else:
model_key = self.select_model_weighted()
model = self.model_config[model_key]["model"]
# Appel API
start_time = time.time()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": model,
"messages": messages,
**kwargs
},
timeout=30
)
latency = (time.time() - start_time) * 1000
if response.status_code == 200:
self.stats[model_key]["requests"] += 1
self.stats[model_key]["total_latency"] += latency
result = response.json()
result["_meta"] = {
"model_used": model,
"model_key": model_key,
"latency_ms": latency,
"cost_per_mtok": self.model_config[model_key]["cost_per_mtok"]
}
return result
else:
self.stats[model_key]["errors"] += 1
return {"error": response.text, "status_code": response.status_code}
except Exception as e:
self.stats[model_key]["errors"] += 1
return {"error": str(e)}
def get_stats(self) -> Dict:
"""Retourne les statistiques d'utilisation"""
report = {}
for model_key, stat in self.stats.items():
avg_latency = (
stat["total_latency"] / stat["requests"]
if stat["requests"] > 0 else 0
)
report[model_key] = {
**stat,
"avg_latency_ms": round(avg_latency, 2),
"error_rate": round(
stat["errors"] / (stat["requests"] + stat["errors"]) * 100, 2
) if (stat["requests"] + stat["errors"]) > 0 else 0
}
return report
============================================
EXEMPLE D'UTILISATION
============================================
Initialisation avec votre clé API
client = HolySheepLoadBalancer("YOUR_HOLYSHEEP_API_KEY")
Exemple 1: Load balancing pondéré classique
messages = [
{"role": "system", "content": "Tu es un assistant IA helpful."},
{"role": "user", "content": "Explique-moi le load balancing en termes simples."}
]
response = client.chat_completion(
messages,
model_strategy="weighted"
)
if "error" not in response:
print(f"Réponse: {response['choices'][0]['message']['content']}")
print(f"Modèle utilisé: {response['_meta']['model_used']}")
print(f"Latence: {response['_meta']['latency_ms']:.2f}ms")
else:
print(f"Erreur: {response['error']}")
Exemple 2: Sélection intelligente par type de tâche
code_response = client.chat_completion(
messages=[
{"role": "user", "content": "Écris une fonction Python pour calculer la factorielle."}
],
model_strategy="smart",
task_type="coding" # Utilise Claude pour le code
)
print(f"\nPour le code, modèle utilisé: {code_response['_meta']['model_used']}")Configuration Load Balancing Avancée avec Fallback
import requests
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Dict, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ModelEndpoint:
name: str
model: str
priority: int
max_retries: int
timeout: int
enabled: bool = True
class AdvancedLoadBalancer:
"""
Load Balancer Avancé avec:
- Circuit Breaker pattern
- Automatic fallback
- Rate limiting
- Cost optimization
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Configuration des endpoints avec priorité
self.endpoints: List[ModelEndpoint] = [
ModelEndpoint(
name="deepseek_cheapest",
model="deepseek-v3.2",
priority=1,
max_retries=3,
timeout=10
),
ModelEndpoint(
name="gemini_fast",
model="gemini-2.5-flash",
priority=2,
max_retries=2,
timeout=15
),
ModelEndpoint(
name="gpt4_premium",
model="gpt-4.1",
priority=3,
max_retries=2,
timeout=20
),
ModelEndpoint(
name="claude_coding",
model="claude-sonnet-4.5",
priority=4,
max_retries=2,
timeout=25
),
]
# Circuit breaker state
self.circuit_state = {ep.name: "closed" for ep in self.endpoints}
self.failure_count = {ep.name: 0 for ep in self.endpoints}
self.last_success = {ep.name: 0 for ep in self.endpoints}
# Configuration
self.failure_threshold = 5
self.recovery_timeout = 60 # secondes
def _check_circuit(self, endpoint_name: str) -> bool:
"""Vérifie si le circuit breaker est ouvert"""
state = self.circuit_state.get(endpoint_name, "closed")
if state == "open":
# Vérifie si assez de temps s'est écoulé pour réessayer
if self.last_success.get(endpoint_name, 0) + self.recovery_timeout < asyncio.get_event_loop().time():
self.circuit_state[endpoint_name] = "half-open"
logger.info(f"Circuit {endpoint_name} → half-open")
return True
return False
return True
def _record_success(self, endpoint_name: str):
"""Enregistre un succès et ferme le circuit"""
self.circuit_state[endpoint_name] = "closed"
self.failure_count[endpoint_name] = 0
self.last_success[endpoint_name] = asyncio.get_event_loop().time()
def _record_failure(self, endpoint_name: str):
"""Enregistre un échec et ouvre le circuit si nécessaire"""
self.failure_count[endpoint_name] = self.failure_count.get(endpoint_name, 0) + 1
if self.failure_count[endpoint_name] >= self.failure_threshold:
self.circuit_state[endpoint_name] = "open"
logger.warning(f"Circuit {endpoint_name} → OPEN (trop d'échecs)")
def get_available_endpoints(self, task_type: Optional[str] = None) -> List[ModelEndpoint]:
"""Retourne les endpoints disponibles triés par priorité"""
available = []
for ep in self.endpoints:
if not ep.enabled:
continue
if not self._check_circuit(ep.name):
continue
available.append(ep)
# Tri par priorité
available.sort(key=lambda x: x.priority)
return available
async def chat_completion_with_fallback(
self,
messages: List[Dict],
task_type: Optional[str] = None,
**kwargs
) -> Dict:
"""
Effectue un appel avec fallback automatique sur tous les endpoints disponibles
"""
endpoints = self.get_available_endpoints(task_type)
last_error = None
for endpoint in endpoints:
for attempt in range(endpoint.max_retries):
try:
logger.info(f"Tentative avec {endpoint.model} (tentative {attempt + 1})")
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": endpoint.model,
"messages": messages,
**kwargs
},
timeout=aiohttp.ClientTimeout(total=endpoint.timeout)
) as response:
if response.status == 200:
self._record_success(endpoint.name)
result = await response.json()
result["_meta"] = {
"model": endpoint.model,
"endpoint": endpoint.name,
"attempt": attempt + 1,
"latency_category": "low" if endpoint.priority <= 2 else "medium"
}
return result
elif response.status == 429:
# Rate limited, on essaie le suivant
logger.warning(f"Rate limited sur {endpoint.model}")
continue
else:
error_text = await response.text()
logger.error(f"Erreur {response.status} sur {endpoint.model}: {error_text}")
last_error = f"HTTP {response.status}: {error_text}"
break # On essaie l'endpoint suivant
except asyncio.TimeoutError:
logger.warning(f"Timeout sur {endpoint.model}")
last_error = f"Timeout après {endpoint.timeout}s"
continue
except Exception as e:
logger.error(f"Exception sur {endpoint.model}: {str(e)}")
last_error = str(e)
self._record_failure(endpoint.name)
break
# Tous les endpoints ont échoué
return {
"error": "Tous les endpoints ont échoué",
"last_error": last_error,
"available_endpoints": [ep.model for ep in endpoints],
"circuit_states": self.circuit_state
}
def get_health_report(self) -> Dict:
"""Génère un rapport de santé des endpoints"""
report = {
"endpoints": [],
"summary": {
"total": len(self.endpoints),
"healthy": 0,
"degraded": 0,
"unhealthy": 0
}
}
for ep in self.endpoints:
state = self.circuit_state.get(ep.name, "unknown")
failures = self.failure_count.get(ep.name, 0)
health = "healthy" if state == "closed" and failures == 0 else "degraded" if failures < self.failure_threshold else "unhealthy"
report["endpoints"].append({
"name": ep.name,
"model": ep.model,
"priority": ep.priority,
"state": state,
"failures": failures,
"health": health
})
report["summary"][f"{health}s"] += 1
return report
============================================
EXEMPLE D'UTILISATION AVANCÉE
============================================
async def main():
lb = AdvancedLoadBalancer("YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "user", "content": "Donne-moi un résumé des avantages de HolySheep API."}
]
# Avec fallback automatique
result = await lb.chat_completion_with_fallback(
messages,
task_type="general",
max_tokens=500,
temperature=0.7
)
if "error" not in result:
print(f"✅ Réponse obtenue avec {result['_meta']['model']}")
print(f" Tentatives: {result['_meta']['attempt']}")
print(f" Latence: {result['_meta']['latency_category']}")
print(f"\nRéponse: {result['choices'][0]['message']['content']}")
else:
print(f"❌ Erreur: {result['error']}")
print(f" Détail: {result['last_error']}")
# Rapport de santé
health = lb.get_health_report()
print(f"\n📊 Rapport de santé:")
print(f" Endpoints healthy: {health['summary']['healthy']}")
print(f" Endpoints degraded: {health['summary']['degraded']}")
print(f" Endpoints unhealthy: {health['summary']['unhealthy']}")
Exécution
if __name__ == "__main__":
asyncio.run(main())Intégration avec Round-Robin et Weights Personnalisés
const axios = require('axios');
class HolySheepRoundRobin {
/**
* Load Balancer Round-Robin avec poids personnalisables
* pour HolySheep API
*/
constructor(apiKey) {
this.apiKey = apiKey;
this.baseUrl = 'https://api.holysheep.ai/v1';
// Configuration des modèles avec leurs poids
this.models = [
{
name: 'gpt4',
model: 'gpt-4.1',
weight: 30,
currentWeight: 30,
pricePerMTok: 8.0,
avgLatency: 0
},
{
name: 'claude',
model: 'claude-sonnet-4.5',
weight: 25,
currentWeight: 25,
pricePerMTok: 15.0,
avgLatency: 0
},
{
name: 'gemini',
model: 'gemini-2.5-flash',
weight: 35,
currentWeight: 35,
pricePerMTok: 2.50,
avgLatency: 0
},
{
name: 'deepseek',
model: 'deepseek-v3.2',
weight: 10,
currentWeight: 10,
pricePerMTok: 0.42,
avgLatency: 0
}
];
this.requestCount = 0;
this.totalCost = 0;
this.totalTokens = 0;
}
// Algorithme Weighted Round-Robin
selectModel() {
let selectedModel = null;
let maxWeight = -1;
// Trouver le modèle avec le poids actuel le plus élevé
for (const model of this.models) {
if (model.currentWeight > maxWeight) {
maxWeight = model.currentWeight;
selectedModel = model;
}
}
// Réduire le poids du modèle sélectionné
selectedModel.currentWeight -= this.getTotalWeight();
// Augmenter les poids de tous les modèles
for (const model of this.models) {
model.currentWeight += model.weight;
}
return selectedModel;
}
getTotalWeight() {
return this.models.reduce((sum, m) => sum + m.weight, 0);
}
async chatCompletion(messages, options = {}) {
const model = this.selectModel();
const startTime = Date.now();
try {
const response = await axios.post(
${this.baseUrl}/chat/completions,
{
model: model.model,
messages: messages,
max_tokens: options.maxTokens || 1000,
temperature: options.temperature || 0.7
},
{
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
timeout: 30000
}
);
const latency = Date.now() - startTime;
// Mise à jour des statistiques
model.avgLatency = (model.avgLatency * 0.7) + (latency * 0.3);
this.requestCount++;
// Calcul du coût (estimation)
const inputTokens = response.data.usage?.prompt_tokens || 0;
const outputTokens = response.data.usage?.completion_tokens || 0;
const totalTokens = inputTokens + outputTokens;
const cost = (totalTokens / 1000000) * model.pricePerMTok;
this.totalCost += cost;
this.totalTokens += totalTokens;
return {
success: true,
data: response.data,
meta: {
model: model.model,
modelName: model.name,
latencyMs: latency,
avgModelLatency: Math.round(model.avgLatency),
estimatedCost: cost.toFixed(6),
totalCost: this.totalCost.toFixed(2),
totalTokens: this.totalTokens
}
};
} catch (error) {
console.error(Erreur avec ${model.name}:, error.message);
// Retry avec un autre modèle
const fallback = this.models.find(m => m.name !== model.name);
if (fallback) {
console.log(Fallback vers ${fallback.name});
return this.chatCompletion(messages, options);
}
return {
success: false,
error: error.message,
model: model.name
};
}
}
// Distribution based on cost optimization
async chatCompletionCostOptimized(messages, budgetPerRequest = 0.01) {
// Filtrer les modèles dans le budget
const affordableModels = this.models.filter(
m => m.pricePerMTok <= budgetPerRequest * 1000000
);
// Sélectionner le modèle le moins cher
affordableModels.sort((a, b) => a.pricePerMTok - b.pricePerMTok);
const selectedModel = affordableModels[0] || this.models[0];
return this.chatCompletion(messages);
}
// Distribution based on latency
async chatCompletionLatencyOptimized(messages) {
// Sélectionner le modèle avec la latence la plus faible
const sortedModels = [...this.models].sort(
(a, b) => a.avgLatency - b.avgLatency
);
const selectedModel = sortedModels[0];
return this.chatCompletion(messages);
}
getStats() {
return {
totalRequests: this.requestCount,
totalCost: this.totalCost.toFixed(4) + ' USD',
totalTokens: this.totalTokens,
models: this.models.map(m => ({
name: m.name,
model: m.model,
weight: m.weight,
avgLatency: Math.round(m.avgLatency) + 'ms',
pricePerMTok: '$' + m.pricePerMTok.toFixed(2)
})),
distribution: this.getDistribution()
};
}
getDistribution() {
const totalWeight = this.getTotalWeight();
return this.models.map(m => ({
name: m.name,
percentage: ((m.weight / totalWeight) * 100).toFixed(1) + '%'
}));
}
}
// ============================================
// EXEMPLE D'UTILISATION
// ============================================
const client = new HolySheepRoundRobin('YOUR_HOLYSHEEP_API_KEY');
async function runExamples() {
const messages = [
{ role: 'user', content: 'Explique-moi ce qu\'est le load balancing.' }
];
// Test avec distribution par défaut
console.log('=== Test 1: Distribution pondérée ===');
for (let i = 0; i < 5; i++) {
const result = await client.chatCompletion(messages);
if (result.success) {
console.log(Requête ${i + 1}: ${result.meta.model} (${result.meta.latencyMs}ms));
}
}
// Test avec optimisation coût
console.log('\n=== Test 2: Optimisation coût ===');
const costResult = await client.chatCompletionCostOptimized(messages, 0.005);
console.log(Modèle sélectionné: ${costResult.meta?.modelName || costResult.error});
// Test avec optimisation latence
console.log('\n=== Test 3: Optimisation latence ===');
const latencyResult = await client.chatCompletionLatencyOptimized(messages);
console.log(Modèle sélectionné: ${latencyResult.meta?.modelName || latencyResult.error});
// Statistiques finales
console.log('\n=== Statistiques finales ===');
const stats = client.getStats();
console.log(Total requêtes: ${stats.totalRequests});
console.log(Coût total: ${stats.totalCost});
console.log('Distribution:', stats.distribution);
}
runExamples().catch(console.error);Configuration des Stratégies de Routage
Selon votre cas d'utilisation, différentes stratégies de routing sont optimales :
- Task-based routing : Routing automatique selon le type de tâche (code → Claude, raisonnement → GPT-4)
- Cost-based routing : Priorité aux modèles économiques pour les tâches simples
- Latency-based routing : Priorité aux modèles les plus rapides
- Weighted round-robin : Distribution proportionnelle configurée manuellement
Erreurs courantes et solutions
Erreur 1: "401 Unauthorized" - Clé API invalide
# ❌ ERREUR: Clé API mal formatée ou invalide
Response: {"error": {"message": "Incorrect API key provided", "type": "invalid_request_error"}}
✅ SOLUTION: Vérifiez votre clé API et le format de l'en-tête
Mauvais format
headers = {"Authorization": "YOUR_HOLYSHEEP_API_KEY"} # Manquant "Bearer "
Bon format
headers = {"Authorization": f"Bearer {api_key}"}
Vérification de la clé
import requests
def verify_api_key(api_key: str) -> bool:
"""Vérifie si la clé API est valide"""
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
return response.status_code == 200
Test
if verify_api_key("YOUR_HOLYSHEEP_API_KEY"):
print("✅ Clé API valide")
else:
print("❌ Clé API invalide - Vérifiez sur https://www.holysheep.ai/register")Erreur 2: "429 Rate Limit Exceeded"
# ❌ ERREUR: Trop de requêtes simultanées
Response: {"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
✅ SOLUTION: Implémentez un système de rate limiting et de retry
import time
import asyncio
from collections import deque
class RateLimiter:
"""Rate limiter avec token bucket algorithm"""
def __init__(self, max_requests: int, time_window: int):
self.max_requests = max_requests
self.time_window = time_window # en secondes
self.requests = deque()
def is_allowed(self) -> bool:
"""Vérifie si une requête est autorisée"""
now = time.time()
# Supprimer les requêtes hors de la fenêtre
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
return len(self.requests) < self.max_requests
def record_request(self):