En tant qu'ingénieur qui a passé trois ans à intégrer des APIs d'intelligence artificielle dans des systèmes de production, je peux vous assurer d'une chose : le jour où vous négligez la supervision de vos APIs IA est exactement le jour où votre application cessera de fonctionner à 23h45 un vendredi soir. J'ai vécu cette situation plusieurs fois avant de comprendre l'importance capitale d'un mécanisme de health check robuste.
Aujourd'hui, je vais vous guider pas à pas dans la conception d'un système de vérification de santé pour vos APIs IA. Que vous soyez débutant complet ou développeur intermédiaire, cet article vous fournira toutes les bases nécessaires pour implémenter une surveillance efficace de vos services IA.
Qu'est-ce qu'un Health Check et Pourquoi Est-Ce Essentiel ?
Un health check (ou vérification de santé) est un mécanisme qui permet de confirmer qu'une API fonctionne correctement et qu'elle est capable de traiter des requêtes. Pour les APIs IA comme celles proposées par HolySheep AI, cette vérification devient critique car les réponses peuvent prendre du temps et les ressources sont souvent coûteuses.
Concrètement, un health check répond à trois questions fondamentales : L'API est-elle joignable ? Répond-elle dans un délai acceptable ? Les données retournées sont-elles cohérentes ?
Architecture de Base d'un Health Check
Avant de coder, comprenons l'architecture que nous allons implémenter. Un système de health check efficace se compose de trois niveaux : la vérification de connectivité de base, la vérification de la latence, et la validation de la réponse.
1. Vérification de Connectivité Simple
Commençons par le niveau le plus élémentaire : s'assurer que l'API répond. Voici comment implémenter cela avec Python en utilisant l'API HolySheep :
import requests
import time
Configuration HolySheep API
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def check_api_connectivity():
"""
Vérifie si l'API HolySheep est joignable.
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
try:
# Test de connexion basique
response = requests.get(
f"{BASE_URL}/models",
headers=headers,
timeout=5
)
if response.status_code == 200:
print("✓ API joignable - Connexion établie")
return True
else:
print(f"✗ Code de réponse inattendu: {response.status_code}")
return False
except requests.exceptions.Timeout:
print("✗ Délai d'attente dépassé (timeout)")
return False
except requests.exceptions.ConnectionError:
print("✗ Impossible de se connecter à l'API")
return False
except Exception as e:
print(f"✗ Erreur inattendue: {str(e)}")
return False
Exécution du test
if __name__ == "__main__":
check_api_connectivity()
Résultat attendu : Le programme affiche si la connexion à l'API est établie ou échoue, avec gestion des erreurs courantes.
2. Vérification de Latence et Performance
La latence est cruciale pour les APIs IA. HolySheep AI propose une latence inférieure à 50ms, ce qui représente un avantage compétitif majeur. Mesurons cela précisément :
import requests
import time
from datetime import datetime
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class HealthCheckMonitor:
"""
Moniteur de santé pour l'API HolySheep avec mesures de latence.
"""
def __init__(self):
self.headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
self.results = []
def measure_latency(self, test_name="ping"):
"""
Mesure la latence de l'API en millisecondes.
"""
start_time = time.time()
try:
if test_name == "models_list":
response = requests.get(
f"{BASE_URL}/models",
headers=self.headers,
timeout=10
)
else:
response = requests.head(
f"{BASE_URL}/models",
headers=self.headers,
timeout=10
)
end_time = time.time()
latency_ms = (end_time - start_time) * 1000
result = {
"test": test_name,
"latency_ms": round(latency_ms, 2),
"status_code": response.status_code,
"timestamp": datetime.now().isoformat(),
"success": response.status_code == 200
}
self.results.append(result)
return result
except Exception as e:
result = {
"test": test_name,
"latency_ms": -1,
"error": str(e),
"timestamp": datetime.now().isoformat(),
"success": False
}
self.results.append(result)
return result
def get_health_status(self):
"""
Retourne le statut de santé global basé sur les résultats.
"""
if not self.results:
return {"status": "unknown", "message": "Aucun test effectué"}
recent_results = self.results[-5:] # 5 derniers tests
success_rate = sum(1 for r in recent_results if r["success"]) / len(recent_results)
avg_latency = sum(r["latency_ms"] for r in recent_results if r["latency_ms"] > 0) / len(recent_results)
if success_rate == 1.0 and avg_latency < 100:
status = "healthy"
message = "API fonctionnelle - Performance optimale"
elif success_rate >= 0.8:
status = "degraded"
message = "API fonctionnelle - Performance réduite"
else:
status = "unhealthy"
message = "API nécessite attention"
return {
"status": status,
"success_rate": f"{success_rate * 100:.0f}%",
"avg_latency_ms": round(avg_latency, 2),
"message": message
}
def print_report(self):
"""
Affiche un rapport complet du monitoring.
"""
print("=" * 50)
print("RAPPORT DE SANTÉ API HOLYSHEEP")
print("=" * 50)
health = self.get_health_status()
print(f"\nStatut Global: {health['status'].upper()}")
print(f"Message: {health['message']}")
print(f"Taux de succès: {health['success_rate']}")
print(f"Latence moyenne: {health['avg_latency_ms']} ms")
print("\nHistorique des tests:")
for r in self.results[-5:]:
status_icon = "✓" if r["success"] else "✗"
latency_str = f"{r['latency_ms']} ms" if r["latency_ms"] > 0 else "ÉCHEC"
print(f" {status_icon} {r['test']}: {latency_str}")
Exécution
monitor = HealthCheckMonitor()
monitor.measure_latency("ping")
monitor.measure_latency("models_list")
monitor.measure_latency("ping")
monitor.print_report()
3. Vérification Complète avec Test de Génération
Pour une validation vraiment complète, nous devons tester une véritable génération de texte pour nous assurer que le modèle IA fonctionne correctement :
import requests
import json
import time
from dataclasses import dataclass
from typing import Optional
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class HealthCheckResult:
"""
Structure de données pour les résultats de health check.
"""
endpoint: str
status: str
latency_ms: float
response_valid: bool
error_message: Optional[str] = None
details: Optional[dict] = None
class HolySheepHealthChecker:
"""
Vérificateur de santé complet pour l'API HolySheep AI.
Inclut tests de connectivité, latence et génération.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def check_models_endpoint(self) -> HealthCheckResult:
"""
Vérifie la disponibilité de la liste des modèles.
"""
start = time.time()
try:
response = requests.get(
f"{BASE_URL}/models",
headers=self.headers,
timeout=10
)
latency = (time.time() - start) * 1000
if response.status_code == 200:
models = response.json().get("data", [])
return HealthCheckResult(
endpoint="/models",
status="healthy",
latency_ms=round(latency, 2),
response_valid=True,
details={"models_count": len(models)}
)
else:
return HealthCheckResult(
endpoint="/models",
status="unhealthy",
latency_ms=round(latency, 2),
response_valid=False,
error_message=f"Code {response.status_code}"
)
except Exception as e:
return HealthCheckResult(
endpoint="/models",
status="unhealthy",
latency_ms=(time.time() - start) * 1000,
response_valid=False,
error_message=str(e)
)
def check_chat_completion(self) -> HealthCheckResult:
"""
Vérifie le bon fonctionnement de la génération de chat.
"""
start = time.time()
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": "Répondez simplement 'OK' en un mot."}
],
"max_tokens": 10,
"temperature": 0
}
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
latency = (time.time() - start) * 1000
if response.status_code == 200:
data = response.json()
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
return HealthCheckResult(
endpoint="/chat/completions",
status="healthy",
latency_ms=round(latency, 2),
response_valid=True,
details={
"model": data.get("model"),
"response_length": len(content),
"usage": data.get("usage", {})
}
)
else:
return HealthCheckResult(
endpoint="/chat/completions",
status="unhealthy",
latency_ms=round(latency, 2),
response_valid=False,
error_message=f"Code {response.status_code}: {response.text[:100]}"
)
except Exception as e:
return HealthCheckResult(
endpoint="/chat/completions",
status="unhealthy",
latency_ms=(time.time() - start) * 1000,
response_valid=False,
error_message=str(e)
)
def run_full_check(self) -> dict:
"""
Exécute tous les tests de santé et retourne un rapport complet.
"""
print("Démarrage du health check complet...\n")
models_check = self.check_models_endpoint()
chat_check = self.check_chat_completion()
all_healthy = all(r.status == "healthy" for r in [models_check, chat_check])
report = {
"overall_status": "healthy" if all_healthy else "degraded",
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"checks": {
"models_endpoint": models_check.__dict__,
"chat_completion": chat_check.__dict__
}
}
# Affichage du rapport
print("=" * 60)
print("RAPPORT DE SANTÉ HOLYSHEEP AI")
print("=" * 60)
print(f"\nStatut Global: {'✓ FONCTIONNEL' if all_healthy else '⚠ DÉGRADÉ'}\n")
for check_name, check in [("Modèles", models_check), ("Chat", chat_check)]:
icon = "✓" if check.status == "healthy" else "✗"
print(f"{icon} {check_name}")
print(f" Latence: {check.latency_ms} ms")
if check.error_message:
print(f" Erreur: {check.error_message}")
if check.details:
print(f" Détails: {json.dumps(check.details, indent=4)}")
print()
return report
Exécution
if __name__ == "__main__":
checker = HolySheepHealthChecker(API_KEY)
report = checker.run_full_check()
Implémentation d'un Monitor Continu
Pour une surveillance en temps réel, créons un monitor qui vérifie périodiquement la santé de l'API :
import time
import threading
import queue
from datetime import datetime
from typing import Callable, List
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class ContinuousHealthMonitor:
"""
Moniteur continu qui vérifie la santé de l'API à intervalles réguliers.
Idéal pour les environnements de production.
"""
def __init__(self, check_interval_seconds: int = 60):
self.check_interval = check_interval_seconds
self.is_running = False
self.health_history: List[dict] = []
self.alert_queue = queue.Queue()
self._thread = None
def _perform_check(self) -> dict:
"""
Effectue une vérification rapide de santé.
"""
import requests
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
start = time.time()
try:
response = requests.get(
f"{BASE_URL}/models",
headers=headers,
timeout=5
)
latency = (time.time() - start) * 1000
result = {
"timestamp": datetime.now().isoformat(),
"success": response.status_code == 200,
"latency_ms": round(latency, 2),
"status_code": response.status_code
}
# Alerte si latence anormale ou échec
if not result["success"] or result["latency_ms"] > 200:
self.alert_queue.put({
"type": "warning" if result["success"] else "critical",
"message": f"Problème détecté: latence {result['latency_ms']}ms",
"time": result["timestamp"]
})
return result
except Exception as e:
error_result = {
"timestamp": datetime.now().isoformat(),
"success": False,
"latency_ms": -1,
"error": str(e)
}
self.alert_queue.put({
"type": "critical",
"message": f"Échec de connexion: {str(e)}",
"time": error_result["timestamp"]
})
return error_result
def _monitor_loop(self):
"""
Boucle principale du monitor.
"""
while self.is_running:
result = self._perform_check()
self.health_history.append(result)
# Garder seulement les 100 derniers résultats
if len(self.health_history) > 100:
self.health_history = self.health_history[-100:]
time.sleep(self.check_interval)
def start(self):
"""
Démarre le monitoring en arrière-plan.
"""
if not self.is_running:
self.is_running = True
self._thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._thread.start()
print(f"Monitor démarré - vérification toutes les {self.check_interval}s")
def stop(self):
"""
Arrête le monitoring.
"""
self.is_running = False
if self._thread:
self._thread.join(timeout=5)
print("Monitor arrêté")
def get_status(self) -> dict:
"""
Retourne le statut actuel basé sur l'historique.
"""
if not self.health_history:
return {"status": "unknown", "message": "Aucune donnée"}
recent = self.health_history[-10:]
success_count = sum(1 for r in recent if r["success"])
success_rate = success_count / len(recent)
avg_latency = sum(r["latency_ms"] for r in recent if r["latency_ms"] > 0) / len(recent)
if success_rate == 1.0:
status = "optimal"
elif success_rate >= 0.8:
status = "acceptable"
else:
status = "problematic"
return {
"status": status,
"success_rate": f"{success_rate * 100:.0f}%",
"avg_latency_ms": round(avg_latency, 2),
"checks_count": len(self.health_history)
}
def get_alerts(self) -> List[dict]:
"""
Récupère les alertes en attente.
"""
alerts = []
while not self.alert_queue.empty():
try:
alerts.append(self.alert_queue.get_nowait())
except queue.Empty:
break
return alerts
Démonstration
if __name__ == "__main__":
monitor = ContinuousHealthMonitor(check_interval_seconds=10)
monitor.start()
# Surveillance pendant 60 secondes
print("\nSurveillance pendant 60 secondes...\n")
for i in range(6):
time.sleep(10)
status = monitor.get_status()
alerts = monitor.get_alerts()
print(f"[{i+1}/6] Statut: {status['status']} | "
f"Succès: {status['success_rate']} | "
f"Latence: {status['avg_latency_ms']}ms")
for alert in alerts:
print(f" ⚠ ALERTE: {alert['message']}")
monitor.stop()
Intégration avec les Tarifs HolySheep
Lors de la conception de votre système de monitoring, gardez à l'esprit les coûts associés. HolySheep AI propose des tarifs compétitifs avec un taux de change avantageux : 1 yuan = 1 dollar, soit une économie de plus de 85% par rapport aux tarifs occidentaux. Les prix 2026 pour les principaux modèles sont : GPT-4.1 à 8$/million de tokens, Claude Sonnet 4.5 à 15$/million de tokens, Gemini 2.5 Flash à 2,50$/million de tokens, et DeepSeek V3.2 à seulement 0,42$/million de tokens.
Votre système de health check devrait intégrer un suivi des coûts pour éviter les surprises. Voici un exemple simple de tracker de consommation :
import time
from datetime import datetime
from typing import Dict
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
Tarifs HolySheep 2026 (USD par million de tokens)
HOLYSHEEP_PRICES = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.5,
"deepseek-v3.2": 0.42
}
class UsageTracker:
"""
Tracker de consommation et de coûts pour HolySheep API.
"""
def __init__(self):
self.total_tokens = 0
self.total_cost = 0.0
self.requests_count = 0
self.model_usage: Dict[str, int] = {}
def record_usage(self, model: str, usage: dict):
"""
Enregistre l'utilisation et calcule le coût.
"""
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
self.total_tokens += total_tokens
self.requests_count += 1
if model not in self.model_usage:
self.model_usage[model] = 0
self.model_usage[model] += total_tokens
# Calcul du coût
price_per_million = HOLYSHEEP_PRICES.get(model, 1.0)
cost = (total_tokens / 1_000_000) * price_per_million
self.total_cost += cost
def get_summary(self) -> dict:
"""
Retourne un résumé de l'utilisation.
"""
return {
"requests_count": self.requests_count,
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost, 4),
"total_cost_cny": round(self.total_cost, 2), # 1:1 avec HolySheep
"model_breakdown": {
model: {
"tokens": tokens,
"cost_usd": round((tokens / 1_000_000) * HOLYSHEEP_PRICES.get(model, 1.0), 4)
}
for model, tokens in self.model_usage.items()
}
}
def print_report(self):
"""
Affiche un rapport détaillé.
"""
summary = self.get_summary()
print("=" * 50)
print("RAPPORT D'UTILISATION HOLYSHEEP")
print("=" * 50)
print(f"Requêtes: {summary['requests_count']}")
print(f"Tokens totaux: {summary['total_tokens']:,}")
print(f"Coût total: ${summary['total_cost_usd']} USD")
print(f" ¥{summary['total_cost_cny']} CNY")
print("\nRépartition par modèle:")
for model, data in summary["model_breakdown"].items():
print(f" {model}:")
print(f" Tokens: {data['tokens']:,}")
print(f" Coût: ${data['cost_usd']}")
Utilisation avec health check
if __name__ == "__main__":
import requests
tracker = UsageTracker()
# Simulation de plusieurs appels API
test_usage = [
{"model": "gpt-4.1", "usage": {"prompt_tokens": 100, "completion_tokens": 50}},
{"model": "deepseek-v3.2", "usage": {"prompt_tokens": 200, "completion_tokens": 100}},
{"model": "gemini-2.5-flash", "usage": {"prompt_tokens": 150, "completion_tokens": 75}},
]
for usage_data in test_usage:
tracker.record_usage(usage_data["model"], usage_data["usage"])
tracker.print_report()
Meilleures Pratiques et Recommandations
Après des mois d'utilisation intensive, voici mes recommandations personnelles pour un système de health check efficace :
- Fréquence adaptée : Vérifiez toutes les 30 à 60 secondes en production. Une vérification trop fréquente génère des coûts inutiles, trop редкая risque de ne pas détecter les problèmes à temps.
- Seuils de latence : Définissez des seuils d'alerte. Pour HolySheep avec sa latence inférieure à 50ms, un dépassement de 200ms devrait déclencher une alerte.
- Multi-étapes : Combinez vérification de connectivité, mesure de latence, et test de génération réelle pour une couverture complète.
- Récupération automatique : Implémentez des mécanismes de reconnexion automatique en cas d'échec temporaire.
- Historique et alertes : Gardez un historique des vérifications pour identifier les tendances et configurez des alertes intelligentes.
Erreurs Courantes et Solutions
1. Erreur : "Connection timeout exceeded"
Symptôme : Votre health check échoue avec un timeout après plusieurs secondes d'attente.
Causes possibles : Le réseau bloque les requêtes, l'API est temporairement indisponible, ou le serveur proxy ralentit les connexions.
# Solution : Implémenter des retries avec backoff exponentiel
import time
import requests
def robust_health_check(max_retries=3):
"""
Health check avec retries et gestion des timeouts.
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
for attempt in range(max_retries):
try:
response = requests.get(
f"{BASE_URL}/models",
headers=headers,
timeout=10 # Timeout de 10 secondes
)
if response.status_code == 200:
return {"success": True, "attempts": attempt + 1}
# Retry sur erreur 5xx
if 500 <= response.status_code < 600:
wait_time = 2 ** attempt # Backoff exponentiel
print(f"Tentative {attempt + 1} échouée, retry dans {wait_time}s")
time.sleep(wait_time)
continue
return {"success": False, "error": f"HTTP {response.status_code}"}
except requests.exceptions.Timeout:
print(f"Tentative {attempt + 1}: Timeout")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
except requests.exceptions.ConnectionError as e:
print(f"Tentative {attempt + 1}: Erreur de connexion - {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
except Exception as e:
return {"success": False, "error": str(e)}
return {"success": False, "error": "Max retries dépassé"}
2. Erreur : "401 Unauthorized" ou clé API invalide
Symptôme : Toutes vos requêtes retournent 401 Unauthorized.
Causes possibles : Clé API incorrecte, expiration du quota, format d'autorisation incorrect.
# Solution : Validation de la clé API et gestion du format
import os
def validate_api_key(api_key: str) -> dict:
"""
Valide la clé API avant toute utilisation.
"""
if not api_key:
return {"valid": False, "error": "Clé API vide"}
if api_key == "YOUR_HOLYSHEEP_API_KEY":
return {"valid": False, "error": "Placeholder detected - utilisez votre vraie clé"}
if len(api_key) < 20:
return {"valid": False, "error": "Clé API trop courte"}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
try:
response = requests.get(
f"{BASE_URL}/models",
headers=headers,
timeout=5
)
if response.status_code == 401:
return {
"valid": False,
"error": "Clé API invalide ou expirée. Vérifiez votre dashboard HolySheep."
}
if response.status_code == 200:
return {"valid": True, "message": "Clé API valide"}
return {
"valid": False,
"error": f"Réponse inattendue: {response.status_code}"
}
except Exception as e:
return {"valid": False, "error": str(e)}
Vérification au démarrage
api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
validation = validate_api_key(api_key)
print(f"Validation API Key: {validation}")
3. Erreur : "Response validation failed" malgré code 200
Symptôme : La requête retourne 200 mais les données sont vides ou incorrectes.
Causes possibles : Changement de format de réponse API, problème de sérialisation, données corrompues.
# Solution : Validation stricte des réponses
import json
def validate_response(response: requests.Response, expected_fields: list) -> dict:
"""
Valide la structure et le contenu de la réponse API.
"""
try:
data = response.json()
except json.JSONDecodeError:
return {
"valid": False,
"error": "Réponse non-JSON valide",
"raw_text": response.text[:200]
}
# Vérification des champs attendus
missing_fields = [f for f in expected_fields if f not in data]
if missing_fields:
return {
"valid": False,
"error": f"Champs manquants: {missing_fields}",
"received_fields": list(data.keys())
}
# Validation spécifique selon le type de réponse
if "data" in data and isinstance(data["data"], list):
if len(data["data"]) == 0:
return {
"valid": False,
"error": "Liste de modèles vide - vérifier l'endpoint"
}
# Vérifier la structure des premiers éléments
for i, item in enumerate(data["data"][:3]):
required_item_fields = ["id", "object", "created"]
item_missing = [f for f in required_item_fields if f not in item]
if item_missing:
return {
"valid": False,
"error": f"Modèle {i}: champs manquants {item_missing}"
}
return {
"valid": True,
"data_preview": {
"models_count": len(data.get("data", [])),
"first_model": data.get("data", [{}])[0].get("id") if data.get("data") else None
}
}
Test de validation
def test_response_validation():
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
try:
response = requests.get(f"{BASE_URL}/models", headers=headers, timeout=10)
if response.status_code == 200:
result = validate_response(response, ["object", "data"])
print(f"Validation: {'✓' if result['valid'] else '✗'}")
if not result["valid"]:
print(f"Erreur: {result['error']}")
else:
print(f"Données: {result.get('data_preview')}")
else:
print(f"Erreur HTTP: {response.status_code}")
except Exception as e:
print(f"Exception: {e}")
test_response_validation()
4. Erreur : Latence anormalement élevée
Symptôme : Les réponses fonctionnent mais la latence dépasse régulièrement 200ms.
Causes possibles : Surcharge du serveur, problème réseau géographique, trop de requêtes simultanées.
# Solution : Rate limiting intelligent et fallback
import time
import threading
from collections import deque
class RateLimitedClient:
"""
Client HTTP avec limitation de débit et détection de latence anormale.
"""
def __init__(self, max_requests_per_second=5, latency_threshold_ms=150):
self.max_rps = max_requests_per_second
self.latency_threshold = latency_threshold_ms
self.request_times = deque(maxlen=100)
self.lock = threading.Lock()
self.degraded_mode = False
def _wait_for_rate_limit(self):
"""
Assure le respect du rate limiting.
"""
with self.lock:
now = time.time()
# Supprimer les requêtes anciennes
while self.request_times and now - self.request_times[0] > 1:
self.request_times.popleft()
# Attendre si trop de requêtes récentes
if len(self.request_times) >= self.max_rps:
sleep_time = 1 - (now - self.request_times[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.request_times.append(time.time())
def _check_latency(self, latency_ms: float):
"""
Met à jour le mode dégradé basé sur la latence.
"""
if latency_ms > self.latency_threshold:
if not self.degraded_mode:
print(f"⚠ Latence élevée détectée: {latency_ms}ms")
print("⚠ Activation du mode dégradé")
self.degraded_mode = True
else:
if self.degraded_mode:
print(f"✓ Latence恢复正常: {latency_ms}ms")
self.degraded_mode = False
def request(self, method: str, url: str, **kwargs) -> requests.Response:
"""
Effectue une requête avec rate limiting et monitoring de latence.
"""
self._wait_for_rate_limit()
start = time.time()
response = requests.request(method, url, **kwargs)
latency_ms = (time.time() - start) * 1000
self._check_latency(latency_ms)
return response, latency_ms
Démonstration
if __name__ == "__main__":
client = RateLimitedClient(max_requests_per_second=3, latency_threshold_ms=100)
headers = {"Authorization": f"Bearer {API_KEY}"}
print("Test de requêtes avec rate limiting:\n")
for i in range(6):
response, latency = client.request("GET", f"{BASE_URL}/models", headers=headers)
print(f"Requête {i+1}: {response.status_code} - Latence: {latency:.1f}ms")
Conclusion
La conception d'un mécanisme de health check robuste est fondamentale pour toute application utilisant des APIs IA en production. Dans cet article, nous avons couvert les bases de la vérification de connectivité, le monitoring de latence, les tests de génération, et les stratégies de dépannage des erreurs courantes.
Mon expérience