Bienvenue dans ce tutoriel pratique. Aujourd'hui, je vais vous expliquer comment tester rigoureusement vos produits Agent IA avant leur mise en production. Si vous lancez une application qui utilise l'intelligence artificielle en Production, vous devez anticiper les pics de trafic, les erreurs temporaires et les limitations de débit. Sans ces protections, votre application risque de planter publiquement — un cauchemar pour n'importe quel développeur.
Personnellement, j'ai vécu ce problème lors du lancement d'un chatbot client pour une PME e-commerce. Notre système tombait en carafe dès que plus de 50 utilisateurs simultanés tentaient d'interagir. Après des nuits blanches à corriger en urgence, j'ai compris l'importance vitale des tests de charge préventifs. Aujourd'hui, je partage avec vous ma méthodologie complète.
Pour qui / Pour qui ce n'est pas fait
| ✅ Ce tutoriel est fait pour vous si : | ❌ Ce tutoriel n'est PAS pour vous si : |
|---|---|
|
|
Comprendre les 3 Piliers de la Résilience IA
Avant de coder, graspons ensemble les trois mécanismes essentiels qui protégeront votre application des pannes :
1. Le Rate Limiting (Limitation de Débit)
Imaginez un restaurant avec 10 places. Si 100 personnes arrivent en même temps, le système doit gérer la file d'attente intelligemment. Le rate limiting fait exactement cela : il limite le nombre de requêtes par seconde envoyées à l'API IA.
Sans rate limiting : Votre système envoie 1000 requêtes en 1 seconde, l'API vous bloque, vos utilisateurs reçoivent des erreurs 429. Horrible pour l'expérience.
Avec rate limiting : Votre système envoie maximum 100 requêtes par seconde, les autres patientent dans une queue, tous vos utilisateurs sont servis progressivement.
2. Le Retry (Nouvelle Tentative)
L'IA peut avoir des micro-coupures de 200 millisecondes. Au lieu de montrer une erreur à l'utilisateur, le retry automatique réessaie silencieusement. C'est commeredémarrer une machine à café quand elle bug — une seconde tentative résout 90% des problèmes.
3. Le Fallback (Solution de Secours)
Si votre API IA principale (ex: GPT-4.1) est saturée, le fallback bascule vers une alternative (ex: DeepSeek V3.2). L'utilisateur ne remarque rien, votre service continue de fonctionner. C'est le plan B intégré.
Architecture de Notre Solution de Test
Voici le schéma conceptuel de notre système de test haute concurrence :
┌─────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE DE TEST │
│ │
│ ┌─────────┐ ┌─────────────┐ ┌──────────────────────────┐ │
│ │ Load │───▶│ Queue │───▶│ Rate Limiter │ │
│ │ Tester │ │ (Buffer) │ │ (Token Bucket/Sliding) │ │
│ └─────────┘ └─────────────┘ └──────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ API HolySheep AI │ │
│ │ base_url: https://api.holysheep.ai/v1 │ │
│ │ Latence < 50ms │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │ │ │
│ ┌──────────┴───┐ ┌─────┴────┐ ┌────┴───────┐ │
│ │ GPT-4.1 │ │ Claude │ │ DeepSeek │ │
│ │ $8/MTok │ │ Sonnet │ │ V3.2 │ │
│ │ (Primary) │ │ $15/MTok │ │ $0.42/MTok │ │
│ └──────────────┘ └──────────┘ └────────────┘ │
│ │ │ │ │
│ └───────────┴───────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Circuit Breaker │ │
│ │ + Fallback Logic │ │
│ └─────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Metrics Dashboard │ │
│ │ Latence / Errors │ │
│ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Installation de l'Environnement de Test
Ouvrez votre terminal et exécutez les commandes suivantes pour installer les dépendances nécessaires :
# Installation des dépendances Python pour les tests de charge
pip install aiohttp asyncio-rate-limitor httpx prometheus-client psutil
Vérification de l'installation
python --version
Devrait afficher Python 3.9+
Le Code Complet du Système de Test
Module 1 : Configuration et Client HolySheep
# config.py - Configuration centralisée du système de test
import os
from typing import Dict, List
class HolySheepConfig:
"""Configuration pour l'API HolySheep AI"""
# ✅ URLs officielles HolySheep - JAMAIS api.openai.com ou api.anthropic.com
BASE_URL = "https://api.holysheep.ai/v1"
# ⚠️ Remplacez par votre vraie clé depuis https://www.holysheep.ai/register
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Modèles disponibles avec leurs prix 2026
MODELS = {
"gpt-4.1": {
"name": "GPT-4.1",
"provider": "OpenAI via HolySheep",
"price_per_mtok": 8.00, # USD par million de tokens
"latency_estimate": "35-45ms",
"context_window": 128000,
"use_case": "Réponses complexes, coding advanced"
},
"claude-sonnet-4.5": {
"name": "Claude Sonnet 4.5",
"provider": "Anthropic via HolySheep",
"price_per_mtok": 15.00,
"latency_estimate": "40-50ms",
"context_window": 200000,
"use_case": "Analyse nuancée, rédaction longue"
},
"gemini-2.5-flash": {
"name": "Gemini 2.5 Flash",
"provider": "Google via HolySheep",
"price_per_mtok": 2.50,
"latency_estimate": "25-35ms",
"context_window": 1000000,
"use_case": "Réponses rapides, volume élevé"
},
"deepseek-v3.2": {
"name": "DeepSeek V3.2",
"provider": "DeepSeek via HolySheep",
"price_per_mtok": 0.42, # ✅ Économie 85%+ vs GPT-4.1
"latency_estimate": "30-40ms",
"context_window": 64000,
"use_case": "Fallback économique, haute volumétrie"
}
}
# Configuration du rate limiting
RATE_LIMITS = {
"requests_per_second": 50, # 50 requêtes/sec max
"tokens_per_minute": 50000, # 50K tokens/minute
"concurrent_connections": 100, # 100 connexions simultanées
"retry_attempts": 3, # 3 tentatives en cas d'erreur
"retry_delay_base": 1.0, # Délai initial 1 seconde
"timeout_seconds": 30 # Timeout global 30 sec
}
# Configuration du test de charge
LOAD_TEST_CONFIG = {
"duration_seconds": 60, # Test de 60 secondes
"virtual_users": 200, # 200 utilisateurs virtuels
"ramp_up_seconds": 10, # Montée en charge sur 10 sec
"peak_requests_per_second": 500 # Pic à 500 req/sec
}
@classmethod
def get_fallback_chain(cls) -> List[str]:
"""Retourne la chaîne de fallback par priorité"""
return ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
@classmethod
def calculate_cost(cls, model: str, tokens: int) -> float:
"""Calcule le coût en USD pour un nombre de tokens"""
price = cls.MODELS.get(model, {}).get("price_per_mtok", 0)
return (tokens / 1_000_000) * price
@classmethod
def calculate_savings(cls, tokens: int) -> Dict[str, float]:
"""Calcule les économies potentielles avec HolySheep"""
gpt_cost = cls.calculate_cost("gpt-4.1", tokens)
deepseek_cost = cls.calculate_cost("deepseek-v3.2", tokens)
return {
"gpt_cost_usd": gpt_cost,
"deepseek_cost_usd": deepseek_cost,
"savings_usd": gpt_cost - deepseek_cost,
"savings_percentage": ((gpt_cost - deepseek_cost) / gpt_cost * 100) if gpt_cost > 0 else 0
}Module 2 : Rate Limiter Avancé avec Token Bucket
# rate_limiter.py - Implémentation du rate limiting avancé
import time
import asyncio
from collections import deque
from typing import Optional
from threading import Lock
class TokenBucketRateLimiter:
"""
Rate Limiter basé sur l'algorithme Token Bucket.
Permet de lisser les pics de trafic tout en maximisant le throughput.
"""
def __init__(self, rate: int, 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 = Lock()
self.total_requests = 0
self.total_wait_time = 0.0
self.rejected_requests = 0
def _refill(self):
"""Rajoute des tokens selon le temps écoulé"""
now = time.time()
elapsed = now - self.last_update
tokens_to_add = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_update = now
def acquire(self, tokens: int = 1, blocking: bool = True) -> bool:
"""
Acquiert des tokens pour traiter une requête.
Args:
tokens: Nombre de tokens nécessaires
blocking: Si True, attend si nécessaire ; si False, échoue immédiatement
Returns:
True si les tokens ont été acquis
"""
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
self.total_requests += 1
return True
if not blocking:
self.rejected_requests += 1
return False
# Calcul du temps d'attente
tokens_needed = tokens - self.tokens
wait_time = tokens_needed / self.rate
self.total_wait_time += wait_time
# Simulation du wait (en réalité, on libère le lock)
# Attente passive
time.sleep(wait_time)
with self.lock:
self.tokens -= tokens
self.total_requests += 1
return True
async def acquire_async(self, tokens: int = 1) -> bool:
"""Version async pour les environnements asynchrones"""
await asyncio.sleep(0) # Yield to event loop
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
self.total_requests += 1
return True
# Wait and retry
await asyncio.sleep(0.1)
return await self.acquire_async(tokens)
def get_stats(self) -> dict:
"""Retourne les statistiques du rate limiter"""
return {
"total_requests": self.total_requests,
"rejected_requests": self.rejected_requests,
"total_wait_time": self.total_wait_time,
"avg_wait_time": self.total_wait_time / self.total_requests if self.total_requests > 0 else 0,
"current_tokens": self.tokens
}
class SlidingWindowRateLimiter:
"""
Rate Limiter basé sur une fenêtre glissante.
Plus précis que Token Bucket pour respecter des limites strictes.
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self.lock = Lock()
def _clean_old_requests(self):
"""Supprime les requêtes hors de la fenêtre"""
now = time.time()
cutoff = now - self.window_seconds
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
def is_allowed(self) -> bool:
"""Vérifie si une nouvelle requête est autorisée"""
with self.lock:
self._clean_old_requests()
if len(self.requests) < self.max_requests:
self.requests.append(time.time())
return True
return False
def time_until_allowed(self) -> float:
"""Retourne le temps d'attente avant la prochaine requête"""
with self.lock:
self._clean_old_requests()
if len(self.requests) < self.max_requests:
return 0.0
oldest = self.requests[0]
return max(0.0, oldest + self.window_seconds - time.time())
Instanciation des rate limiters selon la config
def create_rate_limiters(config):
"""Factory pour créer les rate limiters selon la configuration"""
return {
"requests_per_second": TokenBucketRateLimiter(
rate=config.RATE_LIMITS["requests_per_second"],
capacity=config.RATE_LIMITS["concurrent_connections"]
),
"tokens_per_minute": TokenBucketRateLimiter(
rate=config.RATE_LIMITS["tokens_per_minute"] / 60,
capacity=config.RATE_LIMITS["tokens_per_minute"] / 2
),
"concurrent": SlidingWindowRateLimiter(
max_requests=config.RATE_LIMITS["concurrent_connections"],
window_seconds=1
)
}Module 3 : Client IA avec Retry et Fallback Intelligent
# holy_sheep_client.py - Client IA complet avec retry et fallback
import httpx
import asyncio
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal, requests pass through
OPEN = "open" # Bloqué, on bypass vers fallback
HALF_OPEN = "half_open" # Test de récupération
@dataclass
class RequestMetrics:
latency_ms: float
status_code: int
model_used: str
success: bool
error_type: Optional[str] = None
tokens_used: int = 0
class CircuitBreaker:
"""
Circuit Breaker pattern pour protéger contre les cascades d'erreurs.
Surveille les échecs et ouvre le circuit quand le seuil est dépassé.
"""
def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 30):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = CircuitState.CLOSED
def record_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def can_execute(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.timeout_seconds:
self.state = CircuitState.HALF_OPEN
return True
return False
# HALF_OPEN : on autorise quelques requêtes test
return True
class HolySheepAIClient:
"""
Client complet pour l'API HolySheep AI avec :
- Rate Limiting automatique
- Retry exponentiel
- Fallback multi-modèle
- Circuit Breaker
"""
def __init__(self, api_key: str, config):
self.api_key = api_key
self.config = config
self.base_url = config.BASE_URL
self.rate_limiters = create_rate_limiters(config)
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
self.metrics: List[RequestMetrics] = []
# Initialiser un circuit breaker par modèle
for model_id in config.MODELS.keys():
self.circuit_breakers[model_id] = CircuitBreaker()
def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
Envoie une requête de chat completion avec gestion complète des erreurs.
"""
# Étape 1 : Vérifier le circuit breaker
cb = self.circuit_breakers.get(model)
if cb and not cb.can_execute():
return await self._fallback_to_alternative(model, messages, temperature, max_tokens)
# Étape 2 : Acquérir le rate limiter
await self.rate_limiters["requests_per_second"].acquire_async()
# Étape 3 : Envoyer la requête avec retry
return await self._request_with_retry(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
async def _request_with_retry(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float,
max_tokens: int,
attempt: int = 0
) -> Dict[str, Any]:
"""Requête avec retry exponentiel"""
max_attempts = self.config.RATE_LIMITS["retry_attempts"]
base_delay = self.config.RATE_LIMITS["retry_delay_base"]
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=self.config.RATE_LIMITS["timeout_seconds"]) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=self._get_headers(),
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
latency = (time.time() - start_time) * 1000
if response.status_code == 200:
# Succès !
cb = self.circuit_breakers.get(model)
if cb:
cb.record_success()
data = response.json()
tokens_used = data.get("usage", {}).get("total_tokens", 0)
self.metrics.append(RequestMetrics(
latency_ms=latency,
status_code=200,
model_used=model,
success=True,
tokens_used=tokens_used
))
return {"success": True, "data": data, "model": model, "latency_ms": latency}
elif response.status_code == 429:
# Rate Limited par HolySheep - on retry
self.metrics.append(RequestMetrics(
latency_ms=latency,
status_code=429,
model_used=model,
success=False,
error_type="rate_limited"
))
raise httpx.HTTPStatusError("Rate limited", request=response.request, response=response)
elif response.status_code >= 500:
# Erreur serveur - on retry
self.metrics.append(RequestMetrics(
latency_ms=latency,
status_code=response.status_code,
model_used=model,
success=False,
error_type="server_error"
))
raise httpx.HTTPStatusError("Server error", request=response.request, response=response)
else:
# Erreur client (4xx) - pas de retry
return {
"success": False,
"error": f"HTTP {response.status_code}",
"model": model
}
except (httpx.HTTPStatusError, httpx.RequestError, asyncio.TimeoutError) as e:
# Gérer le retry
if attempt < max_attempts - 1:
delay = base_delay * (2 ** attempt) # Exponentiel : 1s, 2s, 4s
await asyncio.sleep(delay)
return await self._request_with_retry(
model, messages, temperature, max_tokens, attempt + 1
)
# Plus de retry disponibles - fallback
cb = self.circuit_breakers.get(model)
if cb:
cb.record_failure()
return await self._fallback_to_alternative(model, messages, temperature, max_tokens)
async def _fallback_to_alternative(
self,
failed_model: str,
messages: List[Dict[str, str]],
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Bascule vers le modèle suivant dans la chaîne de fallback"""
fallback_chain = self.config.get_fallback_chain()
failed_index = fallback_chain.index(failed_model) if failed_model in fallback_chain else 0
for i in range(failed_index + 1, len(fallback_chain)):
alternative_model = fallback_chain[i]
cb = self.circuit_breakers.get(alternative_model)
# Ne pas tester un modèle dont le circuit est ouvert
if cb and not cb.can_execute():
continue
try:
result = await self._request_with_retry(
alternative_model, messages, temperature, max_tokens
)
if result.get("success"):
result["fallback_from"] = failed_model
result["fallback_to"] = alternative_model
return result
except Exception:
continue
# Aucun modèle disponible
return {
"success": False,
"error": "All models unavailable",
"fallback_from": failed_model
}
def get_metrics_summary(self) -> Dict[str, Any]:
"""Retourne un résumé des métriques de performance"""
if not self.metrics:
return {"error": "No metrics available"}
successful = [m for m in self.metrics if m.success]
failed = [m for m in self.metrics if not m.success]
latencies = [m.latency_ms for m in successful]
return {
"total_requests": len(self.metrics),
"successful_requests": len(successful),
"failed_requests": len(failed),
"success_rate": len(successful) / len(self.metrics) * 100,
"avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
"p50_latency_ms": sorted(latencies)[len(latencies) // 2] if latencies else 0,
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0,
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0,
"total_tokens": sum(m.tokens_used for m in successful),
"fallback_count": len([m for m in self.metrics if hasattr(m, 'error_type')])
}Module 4 : Simulateur de Charge Haute Concurrence
# load_tester.py - Simulateur de charge pour tester la résilience
import asyncio
import random
import time
from typing import List, Dict, Any
from datetime import datetime
from collections import defaultdict
class LoadSimulator:
"""
Simule une charge réaliste sur votre système Agent IA.
Crée des utilisateurs virtuels qui génèrent des requêtes simultanées.
"""
def __init__(self, client, config):
self.client = client
self.config = config
self.results: List[Dict[str, Any]] = []
self.errors: List[Dict[str, Any]] = []
self.start_time = None
self.end_time = None
def _generate_realistic_prompt(self) -> List[Dict[str, str]]:
"""Génère des prompts réalistes pour le test"""
prompts = [
[{"role": "user", "content": "Explique-moi la différence entre l'apprentissage supervisé et non supervisé en 3 phrases."}],
[{"role": "user", "content": "Écris un email professionnel pour demander un report de délai de livraison."}],
[{"role": "user", "content": "Donne-moi 5 idées de contenu pour Instagram sur le thème de la cuisine healthy."}],
[{"role": "user", "content": "Rédige un résumé des dernières actualités en intelligence artificielle."}],
[{"role": "user", "content": "Comment optimiser mes chances de recrutement sur LinkedIn ?"}],
]
return random.choice(prompts)
async def _virtual_user_simulation(self, user_id: int, duration: int):
"""Simule le comportement d'un utilisateur virtuel"""
end_time = time.time() + duration
request_count = 0
while time.time() < end_time:
try:
# Générer une requête
messages = self._generate_realistic_prompt()
model = random.choice(["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"])
start = time.time()
result = await self.client.chat_completion(
messages=messages,
model=model,
temperature=0.7,
max_tokens=500
)
latency = (time.time() - start) * 1000
self.results.append({
"user_id": user_id,
"timestamp": datetime.now().isoformat(),
"model": result.get("model", model),
"latency_ms": latency,
"success": result.get("success", False),
"fallback_from": result.get("fallback_from"),
"error": result.get("error")
})
request_count += 1
# Pause aléatoire entre requêtes (simule le temps de réflexion humain)
await asyncio.sleep(random.uniform(0.5, 2.0))
except Exception as e:
self.errors.append({
"user_id": user_id,
"timestamp": datetime.now().isoformat(),
"error_type": type(e).__name__,
"error_message": str(e)
})
async def run_load_test(
self,
virtual_users: int = 100,
duration_seconds: int = 60,
ramp_up_seconds: int = 10
):
"""
Lance le test de charge complet.
Args:
virtual_users: Nombre d'utilisateurs virtuels simultanés
duration_seconds: Durée totale du test
ramp_up_seconds: Temps de montée en charge progressive
"""
print(f"🚀 Démarrage du test de charge...")
print(f" Utilisateurs virtuels : {virtual_users}")
print(f" Durée : {duration_seconds} secondes")
print(f" Rampe-up : {ramp_up_seconds} secondes")
print("-" * 50)
self.start_time = time.time()
self.results = []
self.errors = []
# Créer les tâches utilisateurs avec ramp-up progressif
tasks = []
for i in range(virtual_users):
# Delay progressif pour le ramp-up
delay = (i / virtual_users) * ramp_up_seconds if ramp_up_seconds > 0 else 0
task = asyncio.create_task(self._delayed_user_start(i, delay, duration_seconds))
tasks.append(task)
# Attendre la fin de tous les utilisateurs
await asyncio.gather(*tasks)
self.end_time = time.time()
return self.generate_report()
async def _delayed_user_start(self, user_id: int, delay: float, duration: int):
"""Démarre un utilisateur avec un délai"""
if delay > 0:
await asyncio.sleep(delay)
await self._virtual_user_simulation(user_id, duration)
def generate_report(self) -> Dict[str, Any]:
"""Génère un rapport détaillé des résultats du test"""
if not self.results:
return {"error": "No results available"}
total_duration = self.end_time - self.start_time if self.end_time else 0
# Calculer les statistiques
successful = [r for r in self.results if r["success"]]
failed = [r for r in self.results if not r["success"]]
latencies = [r["latency_ms"] for r in successful]
# Grouper par modèle
by_model = defaultdict(list)
for r in successful:
by_model[r["model"]].append(r["latency_ms"])
# Grouper par minute
by_minute = defaultdict(int)
for r in self.results:
minute = r["timestamp"][:16] # YYYY-MM-DDTHH:MM
by_minute[minute] += 1
report = {
"test_info": {
"start_time": datetime.fromtimestamp(self.start_time).isoformat() if self.start_time else None,
"end_time": datetime.fromtimestamp(self.end_time).isoformat() if self.end_time else None,
"duration_seconds": total_duration,
"total_requests": len(self.results),
"virtual_users": len(set(r["user_id"] for r in self.results))
},
"success_metrics": {
"successful": len(successful),
"failed": len(failed),
"success_rate": len(successful) / len(self.results) * 100 if self.results else 0,
"requests_per_second": len(self.results) / total_duration if total_duration > 0 else 0
},
"latency_metrics": {
"avg_ms": sum(latencies) / len(latencies) if latencies else 0,
"min_ms": min(latencies) if latencies else 0,
"max_ms": max(latencies) if latencies else 0,
"p50_ms": sorted(latencies)[len(latencies) // 2] if latencies else 0,
"p95_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0,
"p99_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0
},
"model_breakdown": {
model: {
"requests": len(latencies_list),
"avg_latency_ms": sum(latencies_list) / len(latencies_list) if latencies_list else 0,
"percentage": len(latencies_list) / len(successful) * 100 if successful else 0
}
for model, latencies_list in by_model.items()
},
"fallback_stats": {
"total_fallbacks": len([r for r in self.results if r.get("fallback_from")]),
"fallback_details": [
{"from": r.get("fallback_from"), "to": r.get("model")}
for r in self.results if r.get("fallback_from")
]
},
"error_summary": {
"total_errors": len(self.errors),
"by_type": {
error_type: len([e for e in self.errors if e["error_type"] == error_type])
for error_type in set(e["error_type"] for e in self.errors)
}
}
}
return report
Exemple d'utilisation complète
async def main():
"""Exemple d'exécution du test de charge"""
# Initialisation
config = HolySheepConfig()
client = HolySheepAIClient(
api_key=config.API_KEY,
config=config
)
# Créer et lancer le test
simulator = LoadSimulator(client, config)
report = await simulator.run_load_test(
virtual_users=100,
duration_seconds=60,
ramp_up_seconds=10
)
# Afficher le rapport
print("\n" + "=" * 60)
print("📊 RAPPORT DE TEST DE CHARGE")
print("=" * 60)
print(f"\n✅ Requêtes réussies : {report['success_metrics']['successful']}")
print(f"❌ Requêtes échouées : {