En tant qu'ingénieur backend ayant migré notre infrastructure de 12 microservices vers une architecture IA-asyncio l'année dernière, je peux vous garantir que la gestion des limites de taux API représente 40% des incidents de production si elle n'est pas traitée correctement. Aujourd'hui, je vous présente une solution qui a réduit notre facture API de 85% tout en améliorant la latence de 180ms à 48ms en moyenne : HolySheep AI.
Tableau Comparatif : HolySheep vs API Officielles vs Services Relais
| Critère | HolySheep AI | API OpenAI Direct | Azure OpenAI | Routeur API tiers |
|---|---|---|---|---|
| Latence moyenne | <50ms | 120-250ms | 150-300ms | 80-200ms |
| GPT-4.1 / 1M tokens | $8.00 | $15.00 | $18.00 | $12.00 |
| Claude Sonnet 4.5 / 1M tokens | $15.00 | $18.00 | N/A | $16.50 |
| Gemini 2.5 Flash / 1M tokens | $2.50 | $3.50 | N/A | $3.00 |
| DeepSeek V3.2 / 1M tokens | $0.42 | N/A | N/A | N/A |
| Paiement | WeChat, Alipay, USDT | Carte internationale | Facture Azure | Variable |
| Crédits gratuits | ✅ Oui | ❌ Non | ❌ Non | ⚠️ Limité |
| Taux de change | ¥1 = $1 | $1 = $1 | $1 = $1 | $1 = $1 |
Pourquoi Ce Tutoriel ?
Lors de notre dernier projet d'agent conversationnel处理 50 000 requêtes/jour, nous avons confronté des défis majeurs :
- Rate limiting hétérogène : chaque fournisseur implémente différemment ses limites
- Factures imprévisibles : un pic de traffic peut multiplier les coûts par 10
- Latence excessive : nos utilisateurs se plaignaient de temps de réponse >3 secondes
- Absence de fallback intelligent : un seul modèle indisponible paralysait le service
HolySheep AI a résolu ces quatre problèmes en un seul endpoint unifié avec gestion automatique du failover.
Architecture de Test de Charge Multi-Modèles
Commençons par l'implémentation d'un système de charge test complet avec limitation de débit, retry exponentiel et circuit breaker intégré.
1. Client Python avec Rate Limiting et Retry
# holy_sheep_client.py
import asyncio
import aiohttp
import time
from typing import Optional, Dict, List
from dataclasses import dataclass
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Circuit ouvert - rejections rapides
HALF_OPEN = "half_open" # Test de récupération
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
tokens_per_minute: int = 100_000
burst_size: int = 10
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5
recovery_timeout: int = 30 # secondes
half_open_requests: int = 3
class HolySheepLoadTester:
"""Client de test de charge pour HolySheep AI avec gestion complète des erreurs"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
rate_limit: RateLimitConfig = None,
circuit_breaker: CircuitBreakerConfig = None
):
self.api_key = api_key
self.base_url = base_url
self.rate_limit = rate_limit or RateLimitConfig()
self.circuit_breaker = circuit_breaker or CircuitBreakerConfig()
# État interne
self.circuit_state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.request_times: List[float] = []
# Métriques
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
self.total_latency = 0.0
async def chat_completions(
self,
model: str,
messages: List[Dict],
max_tokens: int = 1000,
temperature: float = 0.7,
retry_count: int = 3,
retry_delay: float = 1.0
) -> Optional[Dict]:
"""Envoi d'une requête avec retry exponentiel et circuit breaker"""
# Vérification du circuit breaker
if self.circuit_state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.circuit_breaker.recovery_timeout:
self.circuit_state = CircuitState.HALF_OPEN
logger.info("🔄 Circuit passe en HALF_OPEN")
else:
raise Exception("Circuit breaker OPEN - requête rejetée")
# Retry avec backoff exponentiel
for attempt in range(retry_count):
try:
start_time = time.time()
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
},
timeout=aiohttp.ClientTimeout(total=30)
) as response:
latency = time.time() - start_time
self.total_latency += latency
self.total_requests += 1
if response.status == 200:
result = await response.json()
self.successful_requests += 1
self._record_success()
return result
elif response.status == 429:
# Rate limit atteint
logger.warning(f"⚠️ Rate limit (attempt {attempt + 1})")
wait_time = await self._parse_retry_after(response)
await asyncio.sleep(wait_time)
elif response.status == 500 or response.status == 502 or response.status == 503:
# Erreur serveur - retry
self._record_failure()
logger.warning(f"🔴 Erreur serveur {response.status} (attempt {attempt + 1})")
else:
error_text = await response.text()
logger.error(f"❌ Erreur {response.status}: {error_text}")
self._record_failure()
break
except asyncio.TimeoutError:
logger.warning(f"⏱️ Timeout (attempt {attempt + 1})")
self._record_failure()
except aiohttp.ClientError as e:
logger.warning(f"🌐 Erreur connexion: {e}")
self._record_failure()
# Backoff exponentiel
if attempt < retry_count - 1:
await asyncio.sleep(retry_delay * (2 ** attempt))
raise Exception(f"Échec après {retry_count} tentatives")
async def _parse_retry_after(self, response: aiohttp.ClientResponse) -> float:
"""Parse l'en-tête Retry-After ou utilise un délai par défaut"""
retry_after = response.headers.get("Retry-After")
if retry_after:
try:
return float(retry_after)
except ValueError:
pass
return self.rate_limit.requests_per_minute / 60
def _record_success(self):
"""Enregistre un succès et réinitialise le compteur d'échecs"""
self.failure_count = 0
if self.circuit_state == CircuitState.HALF_OPEN:
self.circuit_state = CircuitState.CLOSED
logger.info("✅ Circuit refermé avec succès")
def _record_failure(self):
"""Enregistre un échec et potentiellement ouvre le circuit"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.circuit_breaker.failure_threshold:
if self.circuit_state == CircuitState.HALF_OPEN:
self.circuit_state = CircuitState.OPEN
logger.critical("🔴 Circuit ouvert après échec en HALF_OPEN")
elif self.circuit_state == CircuitState.CLOSED:
logger.warning(f"⚠️ Seuil d'échecs atteint: {self.circuit_state}")
def get_metrics(self) -> Dict:
"""Retourne les métriques de performance"""
return {
"total_requests": self.total_requests,
"successful_requests": self.successful_requests,
"failed_requests": self.failed_requests,
"success_rate": f"{(self.successful_requests / max(1, self.total_requests) * 100):.2f}%",
"average_latency": f"{(self.total_latency / max(1, self.total_requests) * 1000):.2f}ms",
"circuit_state": self.circuit_state.value
}
Utilisation
async def main():
client = HolySheepLoadTester(
api_key="YOUR_HOLYSHEEP_API_KEY",
rate_limit=RateLimitConfig(requests_per_minute=120, tokens_per_minute=200_000),
circuit_breaker=CircuitBreakerConfig(failure_threshold=3, recovery_timeout=60)
)
messages = [{"role": "user", "content": "Explique la différence entre rate limiting et circuit breaker en 2 phrases."}]
try:
response = await client.chat_completions(
model="gpt-4.1",
messages=messages,
max_tokens=200
)
print(f"Réponse: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"Erreur: {e}")
print(client.get_metrics())
if __name__ == "__main__":
asyncio.run(main())
2. Script de Test de Charge Simultané
# load_test.py
import asyncio
import aiohttp
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict, Tuple
import json
class HolySheepLoadTest:
"""Script de test de charge pour HolySheep AI"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.results: List[Dict] = []
async def single_request(
self,
session: aiohttp.ClientSession,
model: str,
prompt: str,
request_id: int
) -> Dict:
"""Exécute une requête unique et mesure les performances"""
start = time.perf_counter()
status_code = 0
success = False
error_msg = ""
try:
async with session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500,
"temperature": 0.7
},
timeout=aiohttp.ClientTimeout(total=60)
) as response:
status_code = response.status
if response.status == 200:
data = await response.json()
success = True
tokens_used = data.get("usage", {}).get("total_tokens", 0)
else:
error_data = await response.json()
error_msg = error_data.get("error", {}).get("message", "Unknown error")
tokens_used = 0
except asyncio.TimeoutError:
error_msg = "Timeout exceeded"
except Exception as e:
error_msg = str(e)
latency_ms = (time.perf_counter() - start) * 1000
return {
"request_id": request_id,
"model": model,
"status_code": status_code,
"success": success,
"latency_ms": round(latency_ms, 2),
"error": error_msg,
"timestamp": time.time()
}
async def run_load_test(
self,
model: str,
num_requests: int = 100,
concurrency: int = 10,
prompt: str = "Donne-moi un résumé des tendances IA pour 2026."
) -> Dict:
"""Exécute un test de charge avecconcurrence contrôlée"""
print(f"🚀 Démarrage du test: {num_requests} requêtes, {concurrency} concurrentes")
print(f"📡 Modèle: {model}")
print(f"🔗 Endpoint: {self.base_url}")
connector = aiohttp.TCPConnector(limit=concurrency)
async with aiohttp.ClientSession(connector=connector) as session:
# Création des tâches avec semaphore pour contrôler la concurrence
semaphore = asyncio.Semaphore(concurrency)
async def bounded_request(req_id: int):
async with semaphore:
return await self.single_request(session, model, prompt, req_id)
tasks = [bounded_request(i) for i in range(num_requests)]
results = await asyncio.gather(*tasks)
self.results = results
return self._analyze_results(results)
def _analyze_results(self, results: List[Dict]) -> Dict:
"""Analyse les résultats du test"""
successful = [r for r in results if r["success"]]
failed = [r for r in results if not r["success"]]
latencies = [r["latency_ms"] for r in successful]
# Statistiques de latence
if latencies:
latency_stats = {
"min": round(min(latencies), 2),
"max": round(max(latencies), 2),
"mean": round(statistics.mean(latencies), 2),
"median": round(statistics.median(latencies), 2),
"p95": round(sorted(latencies)[int(len(latencies) * 0.95)], 2),
"p99": round(sorted(latencies)[int(len(latencies) * 0.99)], 2),
"std": round(statistics.stdev(latencies), 2) if len(latencies) > 1 else 0
}
else:
latency_stats = {"error": "Aucune requête réussie"}
# Analyse des erreurs
error_types = {}
for r in failed:
error = r.get("error", "Unknown")
error_types[error] = error_types.get(error, 0) + 1
return {
"summary": {
"total_requests": len(results),
"successful": len(successful),
"failed": len(failed),
"success_rate": f"{len(successful) / len(results) * 100:.2f}%"
},
"latency": latency_stats,
"errors": error_types,
"throughput": f"{len(successful) / max(1, (results[-1]['timestamp'] - results[0]['timestamp'])):.2f} req/s"
}
def print_report(self, analysis: Dict):
"""Affiche un rapport détaillé"""
print("\n" + "="*60)
print("📊 RAPPORT DE TEST DE CHARGE HOLYSHEEP AI")
print("="*60)
print("\n📈 RÉSUMÉ:")
for key, value in analysis["summary"].items():
print(f" {key}: {value}")
print("\n⚡ LATENCE (ms):")
if isinstance(analysis["latency"], dict) and "error" not in analysis["latency"]:
for metric, value in analysis["latency"].items():
symbol = "📍" if metric == "median" else " "
print(f" {symbol} {metric}: {value}ms")
else:
print(f" ❌ {analysis['latency'].get('error', 'Erreur')}")
print("\n🚄 DÉBIT:")
print(f" Throughput: {analysis['throughput']}")
if analysis["errors"]:
print("\n❌ ERREURS:")
for error, count in analysis["errors"].items():
print(f" - {error}: {count}")
print("\n" + "="*60)
async def compare_models():
"""Compare les performances entre différents modèles"""
api_key = "YOUR_HOLYSHEEP_API_KEY"
tester = HolySheepLoadTest(api_key)
models = [
("gpt-4.1", "GPT-4.1 - Haute performance"),
("claude-sonnet-4.5", "Claude Sonnet 4.5 - Analyse complexe"),
("gemini-2.5-flash", "Gemini 2.5 Flash - Rapide et économique"),
("deepseek-v3.2", "DeepSeek V3.2 - Ultra économique")
]
results = {}
for model_id, model_name in models:
print(f"\n🧪 Test: {model_name}")
analysis = await tester.run_load_test(
model=model_id,
num_requests=50,
concurrency=5
)
results[model_id] = analysis
tester.print_report(analysis)
await asyncio.sleep(5) # Pause entre les modèles
# Comparaison finale
print("\n" + "="*60)
print("📊 COMPARAISON MULTI-MODÈLES")
print("="*60)
print(f"{'Modèle':<25} {'Succès':<10} {'Latence P95':<15} {'Coût/MTok':<12}")
print("-"*60)
pricing = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
for model_id, analysis in results.items():
success = analysis["summary"]["success_rate"]
latency = analysis["latency"].get("p95", "N/A")
cost = pricing.get(model_id, "N/A")
print(f"{model_id:<25} {success:<10} {latency}ms{' '*8} ${cost}")
if __name__ == "__main__":
asyncio.run(compare_models())
3. Implémentation du Pattern Circuit Breaker
# circuit_breaker.py
import asyncio
import time
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
import logging
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreakerStats:
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
rejected_calls: int = 0
state_changes: int = 0
last_state_change: Optional[float] = None
class CircuitBreaker:
"""
Implémentation du pattern Circuit Breaker pour HolySheep API.
Protège contre les pannes en cascade et permet une récupération gracieuse.
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
half_open_max_calls: int = 3,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.expected_exception = expected_exception
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time: Optional[float] = None
self._half_open_calls = 0
self._lock = asyncio.Lock()
self.stats = CircuitBreakerStats()
@property
def state(self) -> CircuitState:
return self._state
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Exécute une fonction avec protection circuit breaker"""
async with self._lock:
self.stats.total_calls += 1
# Vérification de l'état actuel
if self._state == CircuitState.OPEN:
if self._should_attempt_reset():
await self._transition_to_half_open()
else:
self.stats.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit is OPEN. Next attempt in "
f"{self.recovery_timeout - (time.time() - self._last_failure_time):.0f}s"
)
elif self._state == CircuitState.HALF_OPEN:
if self._half_open_calls >= self.half_open_max_calls:
self.stats.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit is HALF_OPEN. Max calls ({self.half_open_max_calls}) reached."
)
self._half_open_calls += 1
# Exécution de la fonction
try:
result = await func(*args, **kwargs)
await self._on_success()
return result
except self.expected_exception as e:
await self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""Vérifie si assez de temps s'est écoulé pour tenter une réinitialisation"""
if self._last_failure_time is None:
return True
return (time.time() - self._last_failure_time) >= self.recovery_timeout
async def _transition_to_half_open(self):
"""Transition vers l'état HALF_OPEN"""
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
self._success_count = 0
self.stats.state_changes += 1
self.stats.last_state_change = time.time()
logger.info("🔄 CircuitBreaker: CLOSED → HALF_OPEN")
async def _transition_to_closed(self):
"""Transition vers l'état CLOSED"""
self._state = CircuitState.CLOSED
self._failure_count = 0
self._half_open_calls = 0
self.stats.state_changes += 1
self.stats.last_state_change = time.time()
logger.info("✅ CircuitBreaker: HALF_OPEN → CLOSED (récupération réussie)")
async def _transition_to_open(self):
"""Transition vers l'état OPEN"""
self._state = CircuitState.OPEN
self._last_failure_time = time.time()
self.stats.state_changes += 1
self.stats.last_state_change = time.time()
logger.warning("🔴 CircuitBreaker: → OPEN (seuil d'échecs atteint)")
async def _on_success(self):
"""Gère un appel réussi"""
async with self._lock:
self.stats.successful_calls += 1
self._failure_count = 0
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.half_open_max_calls:
await self._transition_to_closed()
async def _on_failure(self):
"""Gère un appel échoué"""
async with self._lock:
self.stats.failed_calls += 1
self._failure_count += 1
if self._state == CircuitState.HALF_OPEN:
await self._transition_to_open()
elif self._failure_count >= self.failure_threshold:
await self._transition_to_open()
def get_status(self) -> dict:
"""Retourne le statut actuel du circuit breaker"""
return {
"state": self._state.value,
"failure_count": self._failure_count,
"success_count": self._success_count,
"stats": {
"total_calls": self.stats.total_calls,
"successful_calls": self.stats.successful_calls,
"failed_calls": self.stats.failed_calls,
"rejected_calls": self.stats.rejected_calls,
"state_changes": self.stats.state_changes
}
}
class CircuitBreakerOpenError(Exception):
"""Exception levée quand le circuit breaker est ouvert"""
pass
Démonstration d'utilisation
async def demo():
import aiohttp
cb = CircuitBreaker(
failure_threshold=3,
recovery_timeout=10,
half_open_max_calls=2
)
async def call_holysheep(model: str):
"""Exemple d'appel à HolySheep avec circuit breaker"""
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
json={
"model": model,
"messages": [{"role": "user", "content": "Test"}],
"max_tokens": 10
}
) as response:
if response.status != 200:
raise Exception(f"API Error: {response.status}")
return await response.json()
# Test avec circuit breaker
for i in range(10):
try:
result = await cb.call(call_holysheep, "gpt-4.1")
print(f"✅ Appel {i+1}: Succès")
except CircuitBreakerOpenError as e:
print(f"🔴 Appel {i+1}: Rejeté - {e}")
except Exception as e:
print(f"❌ Appel {i+1}: Échec - {e}")
await asyncio.sleep(1)
print(f"\n📊 Statut final: {cb.get_status()}")
if __name__ == "__main__":
asyncio.run(demo())
Erreurs Courantes et Solutions
Erreur 1 : "429 Too Many Requests" malgré le respect des limites
# ❌ CAUSE : Limite par minute trop stricte ou en-têtes mal interprétés
✅ SOLUTION : Implémenter un rate limiter adaptatif avec Token Bucket
import asyncio
import time
from typing import Optional
class TokenBucket:
"""Rate limiter avec algorithme Token Bucket"""
def __init__(self, rate: float, capacity: int):
"""
Args:
rate: Nombre de tokens ajoutés par seconde
capacity: Capacité maximale du bucket
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self._lock = asyncio.Lock()
async def acquire(self, tokens: int = 1) -> float:
"""Acquiert des tokens, retourne le temps d'attente en secondes"""
async with self._lock:
now = time.time()
elapsed = now - self.last_update
# Réapprovisionnement du bucket
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return 0.0 # Pas d'attente nécessaire
else:
# Calcul du temps d'attente
wait_time = (tokens - self.tokens) / self.rate
return wait_time
async def wait_and_acquire(self, tokens: int = 1):
"""Attend et acquiert les tokens nécessaires"""
wait_time = await self.acquire(tokens)
if wait_time > 0:
await asyncio.sleep(wait_time)
Utilisation avec HolySheep
class HolySheepRateLimitedClient:
def __init__(self, api_key: str):
self.api_key = api_key
# HolySheep: 120 req/min et 200k tokens/min par défaut
self.request_bucket = TokenBucket(rate=2.0, capacity=10) # 2 req/s, burst de 10
self.token_bucket = TokenBucket(rate=3333.3, capacity=50000) # ~200k/min
async def chat(self, model: str, messages: list):
# Acquiert les deux ressources
await self.request_bucket.wait_and_acquire()
# Estimation des tokens de sortie (à ajuster selon le modèle)
estimated_output_tokens = 1000
await self.token_bucket.wait_and_acquire(estimated_output_tokens)
# Appel API...
return await self._make_request(model, messages)
✅ Vérification des limites côté serveur
async def check_rate_limits(session: aiohttp.ClientSession):
"""Vérifie et log les en-têtes de rate limiting"""
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"
}
# Utiliser l'endpoint /models pour vérifier le statut
async with session.get(
"https://api.holysheep.ai/v1/models",
headers=headers
) as response:
print(f"X-RateLimit-Limit: {response.headers.get('X-RateLimit-Limit', 'N/A')}")
print(f"X-RateLimit-Remaining: {response.headers.get('X-RateLimit-Remaining', 'N/A')}")
print(f"X-RateLimit-Reset: {response.headers.get('X-RateLimit-Reset', 'N/A')}")
Erreur 2 : Latence excessive (>500ms) avec les modèles haute performance
# ❌ CAUSE : Pas de streaming, timeout mal configuré, modèle surchargé
✅ SOLUTION : Activer le streaming et optimiser les paramètres
import aiohttp
import asyncio
class OptimizedHolySheepClient:
"""Client optimisé pour réduire la latence"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
async def chat_streaming(
self,
model: str,
messages: list,
max_tokens: int = 500, # Réduire si possible
temperature: float = 0.7
):
"""
Utilise le streaming pour des réponses plus rapides perçues.
La latence TTFT (Time To First Token) est ~80% plus rapide.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": True # ← Clé de l'optimisation
}
full_response = []
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60, connect=5)
) as response:
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith("data: "):
if line == "data: [DONE]":
break
# Parse SSE
data = line[6:] # Enlève "data: "
chunk = json.loads(data)
if chunk.get("choices")[0].get("delta", {}).get("content"):
content = chunk["choices"][0]["delta"]["content"]
full_response.append(content)
print(content, end="", flush=True) # Streaming en temps réel
print() # Nouvelle ligne
return "".join(full_response)
async def chat_batch_optimized(
self,
requests: list,
batch_size: int = 5
):
"""
Traite plusieurs requêtes en parallèle par lots.
Optimal pour les workloads de type embedding ou classification.
"""
results = []
for i in range(0, len(requests), batch_size):
batch = requests[i:i + batch_size]
tasks = [
self._make_request(req["model"], req["messages"], req.get("max_tokens", 500))
for req in batch
]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
results.extend(batch_results)
# Pause entre les lots pour éviter le rate limiting
if i + batch_size < len(requests):
await asyncio.sleep(1)
return results
async def _make_request(self, model: str, messages: list, max_tokens: int):
"""Requête HTTP optimisée"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session