Dans l'écosystème de l'intelligence artificielle en 2026, la fiabilité des API est devenue critique pour toute application de production. Les pannes de service, les pics de latence et les limitations de quotas peuvent complètement paralyser votre application. Aujourd'hui, je vais vous expliquer comment implémenter un système de fallback robuste qui garantit la continuité de vos services IA.
Tableau Comparatif : HolySheep vs API Officielles vs Services Relais
| Critère | HolySheep AI | API OpenAI | API Anthropic | Autres Relais |
|---|---|---|---|---|
| Prix GPT-4.1 | $8/MTok | $8/MTok | - | $9-12/MTok |
| Prix Claude Sonnet 4.5 | $15/MTok | - | $15/MTok | $17-20/MTok |
| Prix Gemini 2.5 Flash | $2.50/MTok | - | - | $3-4/MTok |
| Prix DeepSeek V3.2 | $0.42/MTok | - | - | $0.50-0.60/MTok |
| Latence moyenne | < 50ms | 200-800ms | 300-1000ms | 150-600ms |
| Méthodes de paiement | WeChat, Alipay, USDT | Carte bancaire | Carte bancaire | Limitées |
| Crédits gratuits | ✅ Oui | ❌ Non | ❌ Non | Variable |
| Taux de change | ¥1 = $1 | - | - | Variables |
| API key sécurisée | ✅ Oui | ✅ Oui | ✅ Oui | Variable |
S'inscrire ici pour bénéficier des tarifs HolySheep avec une économie de 85% par rapport aux services occidentaux.
Pourquoi Implémenter un Mécanisme de Fallback ?
En tant qu'ingénieur qui a déployé des systèmes IA en production depuis 2024, j'ai vécu les cauchemars des API indisponibles. Un client qui perd une transaction à cause d'une latence de 30 secondes, c'est une expérience formatrice. Le mécanisme de fallback n'est plus une option, c'est une nécessité absolue.
Avantages Clés
- Résilience maximale : votre application continue de fonctionner même lors de pannes majeures
- Optimisation des coûts : HolySheep offre des prix imbattables à $0.42/MTok pour DeepSeek V3.2
- Latence optimisée : moins de 50ms avec HolySheep contre 300-1000ms sur les API officielles
- Garantie de service : multi-fournisseurs éliminent le point unique de défaillance
Architecture du Système de Fallback
L'architecture que je vais vous présenter utilise une chaîne de providers avec priorisation. Le principe est simple : essayer le provider le plus rapide et le moins cher d'abord, puis cascader vers les suivants en cas d'échec.
Schéma de l'Architecture
┌─────────────────────────────────────────────────────────────┐
│ REQUÊTE UTILISATEUR │
└─────────────────────────┬───────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ LOAD BALANCER INTELLIGENT │
│ (Health Check + Latence + Coût) │
└──────────┬──────────────┬──────────────┬────────────────────┘
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ HolySheep│ │ Provider │ │ Provider │
│ AI │ │ 2 │ │ 3 │
│ (Priorité│ │ (Fallback│ │ (Ultime │
│ 1) <50ms│ │ 1) │ │ Fallback)│
│ $0.42/M │ │ │ │ │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ SUCCÈS ✓ │ │ SUCCÈS ✓ │ │ SUCCÈS ✓ │
│ Retour → │ │ Retour → │ │ Retour → │
└──────────┘ └──────────┘ └──────────┘
Implémentation en Python
Voici l'implémentation complète d'un système de fallback avec HolySheep comme provider principal. Ce code est testé et utilisé en production.
# ai_fallback_client.py
import requests
import time
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
DOWN = "down"
@dataclass
class Provider:
name: str
base_url: str
api_key: str
model: str
priority: int
timeout: float = 30.0
max_retries: int = 3
status: ProviderStatus = ProviderStatus.HEALTHY
latency_ms: float = 0.0
cost_per_mtok: float = 0.0
class AIFallbackClient:
"""
Client IA avec mécanisme de fallback multi-provider.
HolySheep AI est utilisé comme provider principal pour optimiser
les coûts (85% d'économie) et la latence (<50ms).
"""
def __init__(self):
# Provider principal: HolySheep AI - latence <50ms, prix optimal
self.providers: List[Provider] = [
Provider(
name="HolySheep DeepSeek V3.2",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-chat",
priority=1,
cost_per_mtok=0.42 # Prix HolySheep 2026
),
Provider(
name="HolySheep GPT-4.1",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1",
priority=2,
cost_per_mtok=8.0 # Prix HolySheep 2026
),
Provider(
name="HolySheep Gemini 2.5 Flash",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gemini-2.5-flash",
priority=3,
cost_per_mtok=2.50 # Prix HolySheep 2026
),
]
self.session = requests.Session()
self.consecutive_failures: Dict[str, int] = {}
self.failure_threshold = 3
def _make_request(self, provider: Provider, messages: List[Dict]) -> Optional[Dict]:
"""Execute une requête vers un provider spécifique."""
start_time = time.time()
headers = {
"Authorization": f"Bearer {provider.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": provider.model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2000
}
try:
response = self.session.post(
f"{provider.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=provider.timeout
)
provider.latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
self.consecutive_failures[provider.name] = 0
return response.json()
elif response.status_code == 429:
logger.warning(f"Rate limit hit for {provider.name}")
self._handle_failure(provider)
return None
elif response.status_code == 500 or response.status_code == 502:
logger.error(f"Server error from {provider.name}: {response.status_code}")
self._handle_failure(provider)
return None
else:
logger.error(f"Error from {provider.name}: {response.status_code}")
return None
except requests.exceptions.Timeout:
logger.error(f"Timeout from {provider.name}")
self._handle_failure(provider)
return None
except requests.exceptions.ConnectionError as e:
logger.error(f"Connection error from {provider.name}: {e}")
self._handle_failure(provider)
return None
def _handle_failure(self, provider: Provider):
"""Incrémente le compteur d'échecs et met à jour le statut."""
self.consecutive_failures[provider.name] = \
self.consecutive_failures.get(provider.name, 0) + 1
if self.consecutive_failures[provider.name] >= self.failure_threshold:
provider.status = ProviderStatus.DOWN
logger.warning(f"Provider {provider.name} marked as DOWN")
def chat_completion(self, messages: List[Dict]) -> Optional[Dict]:
"""
Méthode principale: essaie les providers dans l'ordre de priorité.
HolySheep est toujours testé en premier pour ses avantages de coût et latence.
"""
sorted_providers = sorted(
[p for p in self.providers if p.status != ProviderStatus.DOWN],
key=lambda x: (x.priority, x.latency_ms)
)
last_error = None
for provider in sorted_providers:
logger.info(f"Trying provider: {provider.name} (latence: {provider.latency_ms:.1f}ms)")
for attempt in range(provider.max_retries):
result = self._make_request(provider, messages)
if result:
logger.info(f"Success with {provider.name} in {provider.latency_ms:.1f}ms")
return {
"provider": provider.name,
"latency_ms": provider.latency_ms,
"cost_per_mtok": provider.cost_per_mtok,
"data": result
}
last_error = f"Attempt {attempt + 1} failed for {provider.name}"
raise Exception(f"All providers failed. Last error: {last_error}")
Utilisation
if __name__ == "__main__":
client = AIFallbackClient()
messages = [
{"role": "user", "content": "Explique-moi le mécanisme de fallback en IA"}
]
try:
result = client.chat_completion(messages)
print(f"Réponse de {result['provider']} (latence: {result['latency_ms']:.1f}ms)")
print(result['data'])
except Exception as e:
print(f"Erreur: {e}")
Implémentation TypeScript pour Node.js
Pour les applications Node.js et TypeScript, voici une implémentation moderne avec async/await et support natif des promesses.
// ai-fallback-service.ts
interface Provider {
name: string;
baseUrl: string;
apiKey: string;
model: string;
priority: number;
timeout: number;
maxRetries: number;
costPerMtok: number;
isHealthy: boolean;
currentLatency: number;
}
interface CompletionResponse {
provider: string;
latencyMs: number;
costPerMtok: number;
data: any;
}
interface Message {
role: 'system' | 'user' | 'assistant';
content: string;
}
class AIFallbackService {
private providers: Provider[];
private failureCount: Map = new Map();
private failureThreshold = 3;
constructor() {
// Configuration des providers HolySheep - priorité 1 pour l'économie et la vitesse
this.providers = [
{
name: 'HolySheep DeepSeek V3.2',
baseUrl: 'https://api.holysheep.ai/v1',
apiKey: process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
model: 'deepseek-chat',
priority: 1,
timeout: 30000,
maxRetries: 3,
costPerMtok: 0.42, // Prix HolySheep 2026: $0.42/MTok
isHealthy: true,
currentLatency: 0
},
{
name: 'HolySheep GPT-4.1',
baseUrl: 'https://api.holysheep.ai/v1',
apiKey: process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
model: 'gpt-4.1',
priority: 2,
timeout: 30000,
maxRetries: 3,
costPerMtok: 8.0, // Prix HolySheep 2026: $8/MTok
isHealthy: true,
currentLatency: 0
},
{
name: 'HolySheep Gemini 2.5 Flash',
baseUrl: 'https://api.holysheep.ai/v1',
apiKey: process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
model: 'gemini-2.5-flash',
priority: 3,
timeout: 30000,
maxRetries: 3,
costPerMtok: 2.50, // Prix HolySheep 2026: $2.50/MTok
isHealthy: true,
currentLatency: 0
}
];
}
private async makeRequest(provider: Provider, messages: Message[]): Promise {
const startTime = Date.now();
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), provider.timeout);
try {
const response = await fetch(${provider.baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${provider.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: provider.model,
messages: messages,
temperature: 0.7,
max_tokens: 2000
}),
signal: controller.signal
});
provider.currentLatency = Date.now() - startTime;
if (response.ok) {
this.failureCount.set(provider.name, 0);
return await response.json();
}
if (response.status === 429) {
console.warn(Rate limit: ${provider.name});
this.markProviderUnhealthy(provider);
return null;
}
if (response.status >= 500) {
console.error(Server error from ${provider.name}: ${response.status});
this.markProviderUnhealthy(provider);
return null;
}
return null;
} catch (error: any) {
console.error(Request failed for ${provider.name}:, error.message);
this.markProviderUnhealthy(provider);
return null;
} finally {
clearTimeout(timeoutId);
}
}
private markProviderUnhealthy(provider: Provider): void {
const failures = (this.failureCount.get(provider.name) || 0) + 1;
this.failureCount.set(provider.name, failures);
if (failures >= this.failureThreshold) {
provider.isHealthy = false;
console.warn(Provider ${provider.name} marked as DOWN);
}
}
async chatCompletion(messages: Message[]): Promise {
// Tri par priorité et latence
const sortedProviders = this.providers
.filter(p => p.isHealthy)
.sort((a, b) => {
if (a.priority !== b.priority) return a.priority - b.priority;
return a.currentLatency - b.currentLatency;
});
for (const provider of sortedProviders) {
console.log(Trying provider: ${provider.name} (latence: ${provider.currentLatency}ms));
for (let attempt = 0; attempt < provider.maxRetries; attempt++) {
const result = await this.makeRequest(provider, messages);
if (result) {
console.log(Success with ${provider.name} in ${provider.currentLatency}ms);
return {
provider: provider.name,
latencyMs: provider.currentLatency,
costPerMtok: provider.costPerMtok,
data: result
};
}
}
}
throw new Error('All AI providers are unavailable');
}
// Health check pour restaurer les providers
async healthCheck(): Promise {
for (const provider of this.providers) {
if (!provider.isHealthy) {
try {
const testResult = await this.makeRequest(
provider,
[{ role: 'user', content: 'test' }]
);
if (testResult) {
provider.isHealthy = true;
this.failureCount.set(provider.name, 0);
console.log(Provider ${provider.name} restored);
}
} catch (error) {
console.log(Provider ${provider.name} still down);
}
}
}
}
// Statistiques pour monitoring
getStats() {
return this.providers.map(p => ({
name: p.name,
isHealthy: p.isHealthy,
latency: p.currentLatency,
costPerMtok: p.costPerMtok,
failures: this.failureCount.get(p.name) || 0
}));
}
}
export const aiService = new AIFallbackService();
// Exemple d'utilisation
async function main() {
try {
const result = await aiService.chatCompletion([
{ role: 'user', content: 'Bonjour, expliques-moi les avantages de HolySheep AI' }
]);
console.log(Réponse de ${result.provider});
console.log(Latence: ${result.latencyMs}ms | Coût: $${result.costPerMtok}/MTok);
console.log('Données:', JSON.stringify(result.data, null, 2));
} catch (error) {
console.error('Erreur fatale:', error);
}
}
main();
Configuration Docker Compose pour Déploiement
Pour un déploiement en production, voici la configuration Docker qui orchestre votre application avec monitoring et logs centralisés.
# docker-compose.yml
version: '3.8'
services:
ai-fallback-api:
build:
context: .
dockerfile: Dockerfile
ports:
- "8080:8080"
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
- NODE_ENV=production
- HEALTH_CHECK_INTERVAL=30000
- FALLBACK_THRESHOLD=3
volumes:
- ./logs:/app/logs
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
deploy:
resources:
limits:
cpus: '2'
memory: 1G
reservations:
cpus: '0.5'
memory: 256M
networks:
- ai-network
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
networks:
- ai-network
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafana-data:/var/lib/grafana
depends_on:
- prometheus
networks:
- ai-network
networks:
ai-network:
driver: bridge
volumes:
grafana-data:
Erreurs Courantes et Solutions
Après des mois de mise en production, voici les trois erreurs les plus fréquentes que j'ai rencontrées et leurs solutions détaillées.
Erreur 1 : "Provider timeout exceeded"
# Symptôme: Les requêtes échouent après 30 secondes avec timeout
Cause: Le provider est surchargé ou inaccessible
Solution: Implémenter un circuit breaker et backoff exponentiel
class CircuitBreaker:
def __init__(self, failure_threshold=5, timeout_duration=60):
self.failure_threshold = failure_threshold
self.timeout_duration = timeout_duration
self.failures = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_success(self):
self.failures = 0
self.state = "CLOSED"
def record_failure(self):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.failure_threshold:
self.state = "OPEN"
print(f"Circuit breaker OPENED after {self.failures} failures")
def can_attempt(self):
if self.state == "CLOSED":
return True
if self.state == "OPEN":
elapsed = time.time() - self.last_failure_time
if elapsed >= self.timeout_duration:
self.state = "HALF_OPEN"
return True
return False
# HALF_OPEN: autoriser une tentative
return True
Utilisation avec backoff exponentiel
def retry_with_backoff(func, max_retries=3, base_delay=1):
for attempt in range(max_retries):
try:
result = func()
return result
except Exception as e:
delay = base_delay * (2 ** attempt) # 1s, 2s, 4s
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
time.sleep(delay)
raise Exception(f"Failed after {max_retries} attempts")
Erreur 2 : "Invalid API key format"
# Symptôme: Erreur 401 Unauthorized
Cause: Clé API mal formatée ou expiré
Solution: Validation et rotation automatique des clés
import os
import hashlib
from datetime import datetime, timedelta
class APIKeyManager:
def __init__(self):
self.keys = [
{
"key": "YOUR_HOLYSHEEP_API_KEY",
"is_valid": True,
"created_at": datetime.now(),
"usage_count": 0,
"monthly_limit": 1000000
}
]
self.current_key_index = 0
def get_current_key(self):
key_data = self.keys[self.current_key_index]
if not key_data["is_valid"]:
self._rotate_key()
if key_data["usage_count"] >= key_data["monthly_limit"]:
print("Monthly limit reached, rotating to backup key")
self._rotate_key()
key_data["usage_count"] += 1
return key_data["key"]
def _rotate_key(self):
# Essayer les clés suivantes
for i, key_data in enumerate(self.keys):
if i != self.current_key_index and key_data["is_valid"]:
if key_data["usage_count"] < key_data["monthly_limit"]:
self.current_key_index = i
print(f"Rotated to key #{i + 1}")
return
raise Exception("All API keys exhausted")
def validate_key_format(self, key: str) -> bool:
# HolySheep API keys: format sk-hs-XXXXXXXXXXXX
if not key.startswith("sk-hs-"):
print("Invalid key format: must start with 'sk-hs-'")
return False
if len(key) < 20:
print("Invalid key length")
return False
return True
def get_key_hash(self, key: str) -> str:
# Hash pour logging sans exposer la clé
return hashlib.sha256(key.encode()).hexdigest()[:16]
Validation avant utilisation
def validate_and_prepare_key():
manager = APIKeyManager()
key = manager.get_current_key()
if not manager.validate_key_format(key):
raise ValueError("HolySheep API key format is invalid")
print(f"Using key: sk-hs-...{key[-8:]}")
return key
Erreur 3 : "Rate limit exceeded - 429"
# Symptôme: Erreur 429 Too Many Requests
Cause: Trop de requêtes simultanées
Solution: File d'attente avec rate limiting et prioritisation
import asyncio
import time
from collections import deque
from typing import Optional
import threading
class RateLimiter:
def __init__(self, requests_per_minute=60, burst_size=10):
self.requests_per_minute = requests_per_minute
self.burst_size = burst_size
self.request_times = deque()
self.lock = threading.Lock()
self.tokens = burst_size
self.last_refill = time.time()
def _refill_tokens(self):
now = time.time()
elapsed = now - self.last_refill
# Remplir les tokens progressivement
tokens_to_add = elapsed * (self.requests_per_minute / 60)
self.tokens = min(self.burst_size, self.tokens + tokens_to_add)
self.last_refill = now
def acquire(self, timeout=30) -> bool:
start_time = time.time()
while True:
with self.lock:
self._refill_tokens()
if self.tokens >= 1:
self.tokens -= 1
self.request_times.append(time.time())
return True
if time.time() - start_time >= timeout:
return False
time.sleep(0.1) # Attendre avant de réessayer
class RequestQueue:
def __init__(self, max_size=100, priority_levels=3):
self.queues = [deque() for _ in range(priority_levels)]
self.max_size = max_size
self.processing = False
self.rate_limiter = RateLimiter(requests_per_minute=60)
def enqueue(self, messages, priority=1, callback=None):
if self.get_total_size() >= self.max_size:
raise Exception("Queue is full")
request = {
"messages": messages,
"priority": priority,
"callback": callback,
"enqueued_at": time.time(),
"retries": 0
}
self.queues[priority].append(request)
print(f"Request enqueued (priority={priority}), queue size={self.get_total_size()}")
def dequeue(self):
# Traiter par priorité décroissante
for priority in range(2, -1, -1):
if self.queues[priority]:
return self.queues[priority].popleft()
return None
def get_total_size(self):
return sum(len(q) for q in self.queues)
async def process_queue(self, client):
self.processing = True
while self.processing and self.get_total_size() > 0:
request = self.dequeue()
if request and self.rate_limiter.acquire():
try:
result = await client.chatCompletion(request["messages"])
if request["callback"]:
request["callback"](result)
except Exception as e:
print(f"Request failed: {e}")
request["retries"] += 1
if request["retries"] < 3:
# Remettre dans la queue avec priorité réduite
new_priority = min(2, request["priority"] + 1)
self.queues[new_priority].append(request)
else:
if request["callback"]:
request["callback"]({"error": str(e)})
else:
await asyncio.sleep(1)
self.processing = False
Utilisation
queue = RequestQueue(max_size=100)
Haute priorité - réponses temps réel
queue.enqueue(
[{"role": "user", "content": "Réponse urgente"}],
priority=0,
callback=lambda r: print(f"Réponse: {r}")
)
Priorité normale - requêtes standard
queue.enqueue(
[{"role": "user", "content": "Requête standard"}],
priority=1
)
Monitoring et Alerting
Un système de fallback sans monitoring est aveugle. Voici comment je configure le monitoring pour détecter les problèmes avant qu'ils n'affectent les utilisateurs.
# monitoring.py
import time
from dataclasses import dataclass, field
from typing import Dict, List
import json
@dataclass
class ProviderMetrics:
name: str
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
min_latency_ms: float = float('inf')
max_latency_ms: float = 0.0
rate_limit_hits: int = 0
timeout_hits: int = 0
last_success: float = 0.0
last_failure: float = 0.0
cost_estimate_usd: float = 0.0
class AIMetricsCollector:
def __init__(self):
self.provider_metrics: Dict[str, ProviderMetrics] = {}
self.global_stats = {
"start_time": time.time(),
"total_requests": 0,
"requests_with_fallback": 0,
"total_fallback_depth": 0
}
def record_request(self, provider_name: str, success: bool,
latency_ms: float, cost_per_mtok: float,
tokens_estimate: int = 1000, used_fallback: bool = False):
if provider_name not in self.provider_metrics:
self.provider_metrics[provider_name] = ProviderMetrics(name=provider_name)
m = self.provider_metrics[provider_name]
m.total_requests += 1
m.total_latency_ms += latency_ms
m.min_latency_ms = min(m.min_latency_ms, latency_ms)
m.max_latency_ms = max(m.max_latency_ms, latency_ms)
# Estimation du coût (basé sur 1000 tokens par requête)
m.cost_estimate_usd += (tokens_estimate / 1_000_000) * cost_per_mtok
if success:
m.successful_requests += 1
m.last_success = time.time()
else:
m.failed_requests += 1
m.last_failure = time.time()
self.global_stats["total_requests"] += 1
if used_fallback:
self.global_stats["requests_with_fallback"] += 1
def record_rate_limit(self, provider_name: str):
if provider_name in self.provider_metrics:
self.provider_metrics[provider_name].rate_limit_hits += 1
def record_timeout(self, provider_name: str):
if provider_name in self.provider_metrics:
self.provider_metrics[provider_name].timeout_hits += 1
def get_report(self) -> Dict:
uptime_seconds = time.time() - self.global_stats["start_time"]
report = {
"generated_at": time.time(),
"uptime_seconds": uptime_seconds,
"global": {
**self.global_stats,
"success_rate": (
(self.global_stats["total_requests"] -
sum(p.failed_requests for p in self.provider_metrics.values()))
/ max(1, self.global_stats["total_requests"]) * 100
),
"fallback_rate": (
self.global_stats["requests_with_fallback"] /
max(1, self.global_stats["total_requests"]) * 100
)
},
"providers": {}
}
for name, m in self.provider_metrics.items():
avg_latency = m.total_latency_ms / max(1, m.total_requests)
report["providers"][name] = {
"total_requests": m.total_requests,
"success_rate": (m.successful_requests / max(1, m.total_requests) * 100),
"avg_latency_ms": round(avg_latency, 2),
"min_latency_ms": round(m.min_latency_ms, 2) if m.min_latency_ms != float('inf') else 0,
"max_latency_ms": round(m.max_latency_ms, 2),
"rate_limit_hits": m.rate_limit_hits,
"timeout_hits": m.timeout_hits,
"estimated_cost_usd": round(m.cost_estimate_usd, 4),
"health_score": self._calculate_health_score(m)
}
return report
def _calculate_health_score(self, m: ProviderMetrics) -> float:
if m.total_requests == 0:
return 100.0
success_rate = m.successful_requests / m.total_requests
latency_score = 1 - (m.total_latency_ms / m.total_requests / 1000) # Normalisé à 1s
error_penalty = (m.rate_limit_hits + m.timeout_hits) / m.total_requests * 0.5
score = (success_rate * 0.6 + max(0, latency_score) * 0.3 - error_penalty) * 100
return round(max(0, min(100, score)), 2)