Introduction : Ma Propre Galère avec les Rate Limits
Il y a deux ans, j'ai déployé ma première application SaaS utilisant l'IA générative. En trois jours, mon service a été complètement bloqué par les limites de taux de l'API officielle. Trois jours de travail perdus, des clients mécontents, et une leçon inoubliable sur l'importance critique de gérer correctement les limitations de fréquence.
Aujourd'hui, après avoir testé des dizaines de configurations et optimisé des centaines de milliers d'appels API, je vais vous partager toutes les stratégies que j'ai apprises à la dure. Et je vous montrerai comment HolySheep AI a transformé ma façon de concevoir des applications IA robustes.
Tableau Comparatif : HolySheep vs API Officielle vs Services Relais
| Critère | HolySheep AI | API OpenAI | API Anthropic | Services Relais Génériques |
|---|---|---|---|---|
| Taux de change | ¥1 = $1 (parité) | USD uniquement | USD uniquement | Variable, souvent défavorable |
| Latence médiane | <50ms | 200-500ms | 300-800ms | 100-300ms |
| Rate Limit (RPM) | 10 000/min | 500/min (GPT-4) | 200/min (Claude) | Variable |
| GPT-4.1 prix/1M tokens | $8 (≈¥8) | $8 | - | $9-12 |
| Claude Sonnet 4.5 prix/1M tokens | $15 (≈¥15) | - | $15 | $16-20 |
| Gemini 2.5 Flash prix/1M tokens | $2.50 (≈¥2.50) | - | - | $3-5 |
| DeepSeek V3.2 prix/1M tokens | $0.42 (≈¥0.42) | - | - | $0.50-1 |
| Crédits gratuits | ✓ Oui | ✗ Non | ✗ Non | Rarement |
| Paiement | WeChat/Alipay/USD | Carte internationale | Carte internationale | Variable |
Comprendre les Types de Limites de Fréquence
1. Limites par Requête par Minute (RPM)
La限制 la plus commune. Elle définit combien de requêtes vous pouvez envoyer par minute. Pour HolySheep AI, cette limite est de 10 000 RPM, ce qui est 20 fois supérieur à l'API officielle OpenAI.
2. Limites par Tokens par Minute (TPM)
Cette limite contrôle le volume total de tokens (entrée + sortie) que vous pouvez consommer par minute. Elle est particulièrement importante pour les modèles puissants comme GPT-4.1 ou Claude Sonnet 4.5.
3. Limites par Requêtes par Jour (RPD)
Certains services imposent des limites journalières. HolySheep AI offre des limites journalières très généreuses adaptées aux besoins professionnels.
Stratégie #1 : Implémentation d'un Queue Manager avec Exponential Backoff
Après des mois de tests, j'ai développé ce système de file d'attente qui a réduit mes échecs d'API de 95%. L'astuce est d'implémenter un backoff exponentiel intelligent qui s'adapte automatiquement à la charge du serveur.
#!/usr/bin/env python3
"""
Système de gestion de requêtes API avec Rate Limiting intelligent
Conçu pour HolySheep AI avec support multi-modèles
"""
import time
import asyncio
from datetime import datetime, timedelta
from collections import deque
from typing import Optional, Dict, Any
import aiohttp
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HolySheepRateLimiter:
"""
Gestionnaire de rate limiting optimisé pour l'API HolySheep AI.
Implémente un algorithme de Token Bucket avec backoff exponentiel.
"""
def __init__(
self,
rpm_limit: int = 10000,
tpm_limit: int = 1000000,
base_url: str = "https://api.holysheep.ai/v1"
):
self.base_url = base_url
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
# Token buckets pour tracking
self.request_tokens = deque(maxlen=rpm_limit)
self.token_usage = deque() # (timestamp, count)
# Configuration backoff exponentiel
self.max_retries = 5
self.base_delay = 1.0
self.max_delay = 60.0
# Statistiques
self.total_requests = 0
self.failed_requests = 0
self.successful_requests = 0
def _clean_old_tokens(self, tokens_deque: deque) -> None:
"""Supprime les tokens expirés (fenêtre de 60 secondes)"""
cutoff = datetime.now() - timedelta(seconds=60)
while tokens_deque and tokens_deque[0] < cutoff:
tokens_deque.popleft()
def _can_make_request(self) -> bool:
"""Vérifie si une requête peut être faite selon les limites RPM"""
self._clean_old_tokens(self.request_tokens)
return len(self.request_tokens) < self.rpm_limit
def _calculate_wait_time(self) -> float:
"""Calcule le temps d'attente avant la prochaine requête"""
self._clean_old_tokens(self.request_tokens)
if not self.request_tokens:
return 0.0
oldest = self.request_tokens[0]
elapsed = (datetime.now() - oldest).total_seconds()
wait_time = max(0, 60 - elapsed)
return wait_time
async def _get_headers(self, api_key: str) -> Dict[str, str]:
"""Génère les headers d'authentification"""
return {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def chat_completion(
self,
api_key: str,
model: str,
messages: list,
max_retries: Optional[int] = None
) -> Dict[str, Any]:
"""
Envoie une requête de chat completion avec gestion intelligente du rate limit.
Args:
api_key: Clé API HolySheep AI
model: Nom du modèle (ex: gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash)
messages: Liste des messages au format OpenAI
max_retries: Nombre maximum de tentatives (par défaut: 5)
Returns:
Réponse de l'API au format standardisé
Raises:
Exception: Après épuisement des tentatives
"""
max_retries = max_retries or self.max_retries
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 4096
}
for attempt in range(max_retries):
# Vérification de la limite de taux
while not self._can_make_request():
wait_time = self._calculate_wait_time()
logger.info(f"Rate limit atteint. Attente de {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
# Enregistrement de la requête
self.request_tokens.append(datetime.now())
self.total_requests += 1
try:
headers = await self._get_headers(api_key)
async with aiohttp.ClientSession() as session:
async with session.post(
endpoint,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=120)
) as response:
if response.status == 200:
result = await response.json()
self.successful_requests += 1
# Extraction des tokens utilisés
usage = result.get('usage', {})
prompt_tokens = usage.get('prompt_tokens', 0)
completion_tokens = usage.get('completion_tokens', 0)
logger.info(
f"✓ Requête réussie | "
f"Model: {model} | "
f"Tokens: {prompt_tokens + completion_tokens}"
)
return result
elif response.status == 429:
# Rate limit atteint - backoff exponentiel
self.failed_requests += 1
delay = min(
self.base_delay * (2 ** attempt) +
(datetime.now().timestamp() % 1),
self.max_delay
)
logger.warning(
f"⚠ Rate limit HTTP 429 | "
f"Tentative {attempt + 1}/{max_retries} | "
f"Attente: {delay:.2f}s"
)
await asyncio.sleep(delay)
elif response.status == 401:
raise Exception("Clé API invalide. Vérifiez votre clé HolySheep.")
elif response.status == 500:
# Erreur serveur - retry
self.failed_requests += 1
delay = self.base_delay * (2 ** attempt)
logger.warning(f"⚠ Erreur serveur 500 | Retry dans {delay}s")
await asyncio.sleep(delay)
else:
error_text = await response.text()
raise Exception(f"Erreur API {response.status}: {error_text}")
except aiohttp.ClientError as e:
self.failed_requests += 1
delay = self.base_delay * (2 ** attempt)
logger.error(f"✗ Erreur connexion: {e} | Retry dans {delay}s")
await asyncio.sleep(delay)
raise Exception(
f"Échec après {max_retries} tentatives. "
f"Stats: {self.successful_requests}/{self.total_requests} réussites"
)
def get_stats(self) -> Dict[str, Any]:
"""Retourne les statistiques d'utilisation"""
success_rate = (
self.successful_requests / self.total_requests * 100
if self.total_requests > 0 else 0
)
return {
"total_requests": self.total_requests,
"successful_requests": self.successful_requests,
"failed_requests": self.failed_requests,
"success_rate": f"{success_rate:.1f}%",
"current_rpm": len(self.request_tokens),
"rpm_limit": self.rpm_limit
}
Exemple d'utilisation
async def main():
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# Initialisation du rate limiter
limiter = HolySheepRateLimiter(rpm_limit=10000)
# Exemple de conversation
messages = [
{"role": "system", "content": "Tu es un assistant IA expert."},
{"role": "user", "content": "Explique les avantages de HolySheep AI en français."}
]
# Test avec différents modèles
models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"]
for model in models:
try:
response = await limiter.chat_completion(
api_key=API_KEY,
model=model,
messages=messages
)
print(f"\n=== Réponse {model} ===")
print(response['choices'][0]['message']['content'])
except Exception as e:
print(f"\n✗ Erreur avec {model}: {e}")
# Affichage des statistiques
print(f"\n=== Statistiques ===")
print(limiter.get_stats())
if __name__ == "__main__":
asyncio.run(main())
Stratégie #2 : Batch Processing avec Gestion Intelligente des Tokens
Pour les applications qui traitent de grandes quantités de données, le traitement par lots est essentiel. Cette implémentation optimise l'utilisation des tokens tout en respectant scrupuleusement les limites de l'API.
#!/usr/bin/env python3
"""
Système de traitement par lots (Batching) pour requêtes API massives
Optimisé pour HolySheep AI avec gestion avancée des tokens
"""
import asyncio
import tiktoken
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
import hashlib
@dataclass
class BatchItem:
"""Représente un élément individuel dans un batch"""
id: str
messages: List[Dict[str, str]]
metadata: Dict[str, Any] = field(default_factory=dict)
priority: int = 0 # Plus élevé = plus prioritaire
@dataclass
class Batch:
"""Représente un lot de requêtes à traiter ensemble"""
items: List[BatchItem]
total_tokens: int = 0
created_at: datetime = field(default_factory=datetime.now)
def __lt__(self, other: 'Batch'):
return self.created_at < other.created_at
class TokenBatcher:
"""
Gestionnaire intelligent de batching par tokens.
Optimise le throughput en regroupant les requêtes efficacement.
"""
# Limites par modèle (tokens max par requête)
MODEL_LIMITS = {
"gpt-4.1": 128000,
"gpt-4.1-turbo": 128000,
"claude-sonnet-4.5": 200000,
"claude-opus-4": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3.2": 64000,
}
# Limites de batching (par modèle)
BATCH_SIZES = {
"gpt-4.1": 20,
"gpt-4.1-turbo": 20,
"claude-sonnet-4.5": 15,
"claude-opus-4": 10,
"gemini-2.5-flash": 50,
"deepseek-v3.2": 30,
}
def __init__(
self,
model: str = "gpt-4.1",
max_batch_tokens: int = 100000,
max_batch_size: Optional[int] = None,
max_wait_seconds: float = 5.0
):
self.model = model
self.max_batch_tokens = max_batch_tokens
self.max_batch_size = max_batch_size or self.BATCH_SIZES.get(model, 20)
self.max_wait_seconds = max_wait_seconds
# File d'attente des items en attente
self.pending_items: List[BatchItem] = []
# File des batches formés
self.batches: List[Batch] = []
# Encodage pour calcul des tokens
try:
self.encoding = tiktoken.get_encoding("cl100k_base")
except:
self.encoding = None
# Statistiques
self.total_processed = 0
self.total_tokens_used = 0
def _calculate_tokens(self, messages: List[Dict[str, str]]) -> int:
"""Calcule le nombre de tokens pour un ensemble de messages"""
if self.encoding:
text = json.dumps(messages)
return len(self.encoding.encode(text))
# Approximation simple
return sum(len(str(m)) for m in messages) // 4
def _estimate_output_tokens(self, messages: List[Dict[str, str]]) -> int:
"""Estime les tokens de sortie attendus"""
input_tokens = self._calculate_tokens(messages)
return min(input_tokens * 2, 4096) # Estimation conservative
def add_item(self, messages: List[Dict[str, str]], metadata: Dict = None) -> str:
"""
Ajoute un nouvel item à la file d'attente.
Args:
messages: Liste des messages pour cette requête
metadata: Métadonnées optionnelles
Returns:
ID unique de l'item pour suivi
"""
item_id = hashlib.md5(
f"{datetime.now().timestamp()}{messages}".encode()
).hexdigest()[:12]
item = BatchItem(
id=item_id,
messages=messages,
metadata=metadata or {},
priority=0
)
self.pending_items.append(item)
self._form_batches()
return item_id
def _form_batches(self) -> None:
"""Forme des batches optimaux en fonction des limites"""
current_batch_items = []
current_tokens = 0
# Trier par priorité (et timestamp comme tie-breaker)
sorted_items = sorted(
self.pending_items,
key=lambda x: (-x.priority, x.id)
)
for item in sorted_items:
item_input_tokens = self._calculate_tokens(item.messages)
item_output_tokens = self._estimate_output_tokens(item.messages)
item_total_tokens = item_input_tokens + item_output_tokens
model_limit = self.MODEL_LIMITS.get(self.model, 128000)
# Vérifier si l'item peut être traité seul
if item_total_tokens > model_limit:
# Item trop grand - à subdiviser ou rejeter
continue
# Vérifier si on peut ajouter à ce batch
can_add = (
len(current_batch_items) < self.max_batch_size and
current_tokens + item_total_tokens <= self.max_batch_tokens
)
if can_add:
current_batch_items.append(item)
current_tokens += item_total_tokens
else:
# Terminer le batch courant et en commencer un nouveau
if current_batch_items:
self.batches.append(Batch(
items=current_batch_items,
total_tokens=current_tokens
))
current_batch_items = [item]
current_tokens = item_total_tokens
# Ajouter le dernier batch
if current_batch_items:
self.batches.append(Batch(
items=current_batch_items,
total_tokens=current_tokens
))
# Retirer les items traités des pending
processed_ids = set()
for batch in self.batches:
for item in batch.items:
processed_ids.add(item.id)
self.pending_items = [
item for item in self.pending_items
if item.id not in processed_ids
]
# Trier les batches par ordre de création
self.batches.sort()
def get_ready_batch(self) -> Optional[Batch]:
"""
Retourne un batch prêt pour traitement si les conditions sont remplies.
Retourne None si aucun batch n'est encore prêt.
"""
if not self.batches:
return None
# Vérifier si le batch le plus ancien est prêt
oldest_batch = self.batches[0]
wait_time = (datetime.now() - oldest_batch.created_at).total_seconds()
# Batch prêt si assez d'items OU temps d'attente max atteint
is_ready = (
len(oldest_batch.items) >= max(1, self.max_batch_size // 2) or
wait_time >= self.max_wait_seconds
)
if is_ready:
return self.batches.pop(0)
return None
def get_pending_count(self) -> int:
"""Retourne le nombre d'items en attente"""
return len(self.pending_items)
def get_batch_count(self) -> int:
"""Retourne le nombre de batches en attente"""
return len(self.batches)
class BatchProcessor:
"""
Processeur de batches avec intégration HolySheep AI.
Gère le traitement parallèle tout en respectant les rate limits.
"""
def __init__(
self,
api_key: str,
model: str = "gpt-4.1",
max_concurrent_batches: int = 5,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.model = model
self.max_concurrent_batches = max_concurrent_batches
self.base_url = base_url
self.batcher = TokenBatcher(model=model)
self.rate_limiter = None # Sera initialisé plus tard
self.results: Dict[str, Any] = {}
async def process_stream(self) -> Dict[str, Any]:
"""
Traite les items en flux continu.
Cette méthode doit être appelée régulièrement pour ajouter des items.
"""
while True:
batch = self.batcher.get_ready_batch()
if batch:
await self._process_batch(batch)
# Attente courte avant de vérifier à nouveau
await asyncio.sleep(0.5)
# Sortir si plus rien à traiter
if not batch and self.batcher.get_pending_count() == 0:
break
return self.results
async def _process_batch(self, batch: Batch) -> None:
"""Traite un batch de requêtes"""
# Construire le payload batch
# Note: HolySheep supporte les batches via l'endpoint dédié
payload = {
"model": self.model,
"requests": [
{
"custom_id": item.id,
"messages": item.messages
}
for item in batch.items
]
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
import aiohttp
async with aiohttp.ClientSession() as session:
# Endpoint pour batch processing
endpoint = f"{self.base_url}/batch"
async with session.post(
endpoint,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=300)
) as response:
if response.status == 200:
result = await response.json()
# Stocker les résultats
for item in batch.items:
item_result = result.get('results', {}).get(item.id)
self.results[item.id] = {
'result': item_result,
'metadata': item.metadata
}
print(f"✓ Batch traité: {len(batch.items)} items, "
f"{batch.total_tokens} tokens")
else:
error = await response.text()
print(f"✗ Erreur batch: {response.status} - {error}")
except Exception as e:
print(f"✗ Échec traitement batch: {e}")
# Remettre les items dans la file d'attente
for item in batch.items:
self.batcher.pending_items.append(item)
def add_item(self, messages: List[Dict], metadata: Dict = None) -> str:
"""Ajoute un item à traiter"""
return self.batcher.add_item(messages, metadata)
def get_stats(self) -> Dict[str, Any]:
"""Retourne les statistiques de traitement"""
return {
"pending_items": self.batcher.get_pending_count(),
"pending_batches": self.batcher.get_batch_count(),
"processed_items": len(self.results),
"model": self.model
}
Exemple d'utilisation complète
async def example_usage():
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
processor = BatchProcessor(
api_key=API_KEY,
model="deepseek-v3.2", # Modèle économique à $0.42/1M tokens
max_concurrent_batches=3
)
# Ajouter des items à traiter
test_items = [
[
{"role": "user", "content": f"Requête #{i}: Analyse ce texte..."}
]
for i in range(100)
]
print("Ajout des items au batcher...")
for i, messages in enumerate(test_items):
processor.add_item(messages, metadata={"index": i})
if i % 20 == 0:
print(f" Ajouté {i}/{len(test_items)} items")
print(f"\nStatistiques avant traitement: {processor.get_stats()}")
# Traitement
results = await processor.process_stream()
print(f"\nStatistiques après traitement: {processor.get_stats()}")
print(f"Résultats obtenus: {len(results)}")
if __name__ == "__main__":
asyncio.run(example_usage())
Stratégie #3 : Retry Logic Avancée avec Circuit Breaker
Le pattern Circuit Breaker est indispensable pour les applications de production. Il prevent le cascading failure et protège votre système des pannes en cascade. J'ai implémenté ce système après avoir vécu un incident où une défaillance d'API a paralysé mon service pendant 4 heures.
/**
* Client API HolySheep avec Circuit Breaker Pattern
* TypeScript/Node.js - Production Ready
*/
interface CircuitBreakerConfig {
failureThreshold: number; // Nombre d'échecs avant ouverture
successThreshold: number; // Succès nécessaires pour fermeture
timeout: number; // Timeout en ms avant fallback
resetTimeout: number; // Temps avant nouvelle tentative
}
interface RateLimitConfig {
maxRequestsPerMinute: number;
maxTokensPerMinute: number;
maxConcurrentRequests: number;
}
interface HolySheepRequest {
model: string;
messages: Array<{role: string; content: string}>;
temperature?: number;
max_tokens?: number;
}
enum CircuitState {
CLOSED = 'CLOSED', // Fonctionnement normal
OPEN = 'OPEN', // Circuit ouvert - fallback immédiat
HALF_OPEN = 'HALF_OPEN' // Test de récupération
}
class CircuitBreaker {
private state: CircuitState = CircuitState.CLOSED;
private failureCount = 0;
private successCount = 0;
private lastFailureTime = 0;
constructor(private config: Required) {}
async execute(
fn: () => Promise,
fallback: () => Promise
): Promise {
if (this.state === CircuitState.OPEN) {
if (this.shouldAttemptReset()) {
this.state = CircuitState.HALF_OPEN;
} else {
console.log('🔴 Circuit OPEN - Fallback activé');
return fallback();
}
}
try {
const result = await fn();
this.onSuccess();
return result;
} catch (error) {
this.onFailure();
console.log('⚠️ Échec requête - Circuit mis à jour');
return fallback();
}
}
private shouldAttemptReset(): boolean {
return Date.now() - this.lastFailureTime >= this.config.resetTimeout;
}
private onSuccess(): void {
this.failureCount = 0;
this.successCount++;
if (this.state === CircuitState.HALF_OPEN &&
this.successCount >= this.config.successThreshold) {
this.state = CircuitState.CLOSED;
this.successCount = 0;
console.log('🟢 Circuit CLOSED - Service récupéré');
}
}
private onFailure(): void {
this.failureCount++;
this.lastFailureTime = Date.now();
this.successCount = 0;
if (this.failureCount >= this.config.failureThreshold) {
this.state = CircuitState.OPEN;
console.log('🔴 Circuit OPEN - Seuil d\'échec atteint');
}
}
getState(): CircuitState {
return this.state;
}
}
class TokenBucket {
private tokens: number;
private lastRefill: number;
constructor(
private capacity: number,
private refillRate: number // tokens par seconde
) {
this.tokens = capacity;
this.lastRefill = Date.now();
}
async acquire(tokens: number = 1): Promise {
while (!this.tryAcquire(tokens)) {
await this.sleep(10);
}
}
tryAcquire(tokens: number = 1): boolean {
this.refill();
if (this.tokens >= tokens) {
this.tokens -= tokens;
return true;
}
return false;
}
private refill(): void {
const now = Date.now();
const elapsed = (now - this.lastRefill) / 1000;
const newTokens = elapsed * this.refillRate;
this.tokens = Math.min(this.capacity, this.tokens + newTokens);
this.lastRefill = now;
}
private sleep(ms: number): Promise {
return new Promise(resolve => setTimeout(resolve, ms));
}
}
class HolySheepAPIClient {
private readonly baseURL = 'https://api.holysheep.ai/v1';
private circuitBreaker: CircuitBreaker;
private requestBucket: TokenBucket;
private tokenBucket: TokenBucket;
private requestQueue: Array<() => Promise> = [];
private processingQueue = false;
constructor(
private apiKey: string,
rateLimitConfig: RateLimitConfig
) {
// Configuration Circuit Breaker
this.circuitBreaker = new CircuitBreaker({
failureThreshold: 5,
successThreshold: 3,
timeout: 30000,
resetTimeout: 60000
});
// Configuration Token Buckets
this.requestBucket = new TokenBucket(
rateLimitConfig.maxRequestsPerMinute,
rateLimitConfig.maxRequestsPerMinute / 60
);
this.tokenBucket = new TokenBucket(
rateLimitConfig.maxTokensPerMinute,
rateLimitConfig.maxTokensPerMinute / 60
);
}
async chatCompletion(request: HolySheepRequest): Promise {
return this.circuitBreaker.execute(
async () => {
// Acquisition des tokens du bucket
const estimatedTokens = this.estimateTokens(request);
await Promise.all([
this.requestBucket.acquire(1),
this.tokenBucket.acquire(estimatedTokens)
]);
return this.executeRequest(request);
},
async () => this.fallbackResponse(request)
);
}
private estimateTokens(request: HolySheepRequest): number {
// Estimation conservative basée sur les caractères
const text = JSON.stringify(request.messages);
return Math.ceil(text.length / 4) + (request.max_tokens || 2048);
}
private async executeRequest(request: HolySheepRequest): Promise {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 120000);
try {
const response = await fetch(${this.baseURL}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: request.model,
messages: request.messages,
temperature: request.temperature ?? 0.7,
max_tokens: request.max_tokens ?? 4096
}),
signal: controller.signal
});
clearTimeout(timeout);
if (response.status === 429) {
throw new Error('RATE_LIMITED');
}
if (!response.ok) {
throw new Error(API_ERROR: ${response.status});
}
return await response.json();
} catch (error: any) {
clearTimeout(timeout);
if (error.name === 'AbortError') {
throw new Error('TIMEOUT');
}
throw error;
}
}
private async fallbackResponse(request: HolySheepRequest): Promise {
console.log('⚡ Utilisation du fallback intelligent');
// Stratégies de fallback par priorité:
// 1. Réponse cacheée si disponible
const cached = this.getFromCache(request);
if (cached) {
return {
...cached,
cached: true,
fallback: 'cache'
};
}
// 2. Modèle économique si disponible
if (request.model.includes('gpt-4')) {
return this.chatCompletion({
...request,
model: 'deepseek-v3.2' // $0.42/1M vs $8/1M
});
}
// 3. Réponse de base
return {
id: 'fallback-' + Date.now(),
model: 'fallback',
choices: [{
message: {
role: 'assistant',
content: 'Je suis temporairement indisponible. Veuillez réessayer dans quelques instants.'
},
finish_reason: 'fallback'
}],
fallback: true,
fallback_reason: 'circuit_open'
};
}
private cache = new Map();
private readonly CACHE_TTL = 3600000; // 1 heure
private getFromCache(request: HolySheepRequest): any | null {
const key = this.getCacheKey(request);
const cached = this.cache.get(key);
if (cached && Date.now() - cached.timestamp < this.CACHE_TTL) {
return cached.response;
}
return null;
}
private getCacheKey(request: HolySheepRequest): string {
return JSON.stringify({
model: request.model,
messages: request.messages.slice(-2) // Derniers 2 messages
});
}
async queueRequest(request: HolySheepRequest): Promise {
return new Promise((resolve, reject) => {
this.requestQueue.push(async () => {
try {
const result = await this.chatCompletion(request);
resolve(result);
} catch (error) {
reject(error);
}
});
this.processQueue();
});
}
private async processQueue():