Vous venez de découvrir les APIs d'intelligence artificielle et vous souhaitez monitorer leurs performances ? Vous n'êtes pas seul. En tant qu'auteur technique qui accompagne des centaines de débutants chaque mois, je vais vous expliquer pas à pas comment maîtriser la surveillance de vos APIs AI, même sans aucune expérience préalable.
Ce que vous allez apprendre :
- Comprendre les 3 métriques essentielles d'une API performante
- Implémenter un système de monitoring complet en Python
- Visualiser vos données avec des graphiques professionnels
- Diagnostiquer et résoudre les problèmes courants
Comprendre les Fondamentaux de la Surveillance API
Avant de vous lancer dans le code, Laissez-moi vous expliquer pourquoi la surveillance API est cruciale pour vos applications. Une API d'intelligence artificielle comme celle proposée par HolySheep AI doit être surveillée car vous dépensez de l'argent à chaque requête, et chaque latence impacte l'expérience utilisateur.
Les 3 piliers de la performance API :
1. Le Temps de Réponse (Response Time)
C'est le temps entre l'envoi de votre requête et la réception de la réponse. Avec HolySheep AI, la latence moyenne est inférieure à 50 millisecondes, ce qui rend vos applications ultra-réactives.
2. Le Débit (Throughput)
Le nombre de requêtes que votre API peut traiter par seconde. Plus ce chiffre est élevé, plus votre système est scalable.
3. Le Taux d'Erreur (Error Rate)
Le pourcentage de requêtes qui échouent. Un bon taux d'erreur devrait être inférieur à 1% pour une API fiable.
Mise en Place de l'Environnement de Surveillance
Vous n'avez besoin que de Python installé sur votre machine. Voici comment configurer votre environnement en moins de 5 minutes.
Installation des Bibliothèques Nécessaires
Ouvrez votre terminal et exécutez la commande suivante pour installer les packages requis :
# Installation des dépendances pour le monitoring API
pip install requests pandas matplotlib python-dotenv psutil
Vérification de l'installation
python -c "import requests, pandas, matplotlib; print('Tous les packages sont installés !')"Ces bibliothèques vous permettront respectivement de faire des appels HTTP, de manipuler les données, et de créer des visualisations professionnelles.
Implémentation du Système de Monitoring
Configuration de la Connexion API
Créez un fichier nommé config.py qui contiendra vos paramètres de connexion. Utilisez toujours des variables d'environnement pour sécuriser vos clés API.
import os
from dotenv import load_dotenv
Charge les variables d'environnement depuis le fichier .env
load_dotenv()
Configuration HolySheep AI
CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
"model": "gpt-4.1", # Prix: $8/MTok, excellent rapport qualité-prix
"timeout": 30,
"max_retries": 3
}
Tarification HolySheep AI 2026 (prix en dollars par million de tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00, "currency": "USD"},
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00, "currency": "USD"},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50, "currency": "USD"},
"deepseek-v3.2": {"input": 0.42, "output": 0.42, "currency": "USD"},
}
def verify_connection():
"""Vérifie que la connexion à l'API fonctionne"""
import requests
headers = {
"Authorization": f"Bearer {CONFIG['api_key']}",
"Content-Type": "application/json"
}
try:
response = requests.get(
f"{CONFIG['base_url']}/models",
headers=headers,
timeout=5
)
return response.status_code == 200
except Exception as e:
print(f"Erreur de connexion: {e}")
return False
Exemple d'utilisation
if __name__ == "__main__":
print(f"URL de base: {CONFIG['base_url']}")
print(f"Modèle par défaut: {CONFIG['model']}")
print(f"Prix {CONFIG['model']}: ${HOLYSHEEP_PRICING[CONFIG['model']]['input']}/MTok")
print(f"Connexion: {'✓ OK' if verify_connection() else '✗ ÉCHEC'}")Note pour les débutants : Le fichier .env doit contenir HOLYSHEEP_API_KEY=votre_cle_api. HolySheep AI offre un taux de change avantageux avec ¥1=$1, soit une économie de plus de 85% par rapport aux tarifs officiels.
Classe de Monitoring Complète
Voici le code complet de surveillance qui capture toutes les métriques importantes. Ce script est directement utilisable dans vos projets.
import time
import requests
import pandas as pd
from datetime import datetime
from typing import Dict, List, Optional
import statistics
class APIMonitor:
"""
Moniteur de performance pour API AI.
Capture le temps de réponse, le throughput et le taux d'erreur.
"""
def __init__(self, base_url: str, api_key: str, model: str = "gpt-4.1"):
self.base_url = base_url
self.api_key = api_key
self.model = model
self.history = []
def call_api(self, prompt: str, max_tokens: int = 100) -> Dict:
"""
Appelle l'API et mesure les performances.
Retourne un dictionnaire avec la réponse et les métriques.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens
}
metrics = {
"timestamp": datetime.now(),
"model": self.model,
"success": False,
"response_time_ms": 0,
"error": None,
"tokens_used": 0
}
start_time = time.perf_counter()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
end_time = time.perf_counter()
metrics["response_time_ms"] = (end_time - start_time) * 1000
if response.status_code == 200:
data = response.json()
metrics["success"] = True
metrics["tokens_used"] = (
data.get("usage", {}).get("total_tokens", 0)
)
metrics["response"] = data["choices"][0]["message"]["content"]
else:
metrics["error"] = f"HTTP {response.status_code}: {response.text}"
except requests.exceptions.Timeout:
metrics["error"] = "Timeout - L'API n'a pas répondu à temps"
except requests.exceptions.ConnectionError:
metrics["error"] = "Erreur de connexion - Vérifiez votre internet"
except Exception as e:
metrics["error"] = f"Erreur inattendue: {str(e)}"
self.history.append(metrics)
return metrics
def get_statistics(self) -> Dict:
"""Calcule les statistiques de performance"""
if not self.history:
return {"error": "Aucune donnée disponible"}
response_times = [m["response_time_ms"] for m in self.history if m["success"]]
successful = sum(1 for m in self.history if m["success"])
total = len(self.history)
stats = {
"total_requests": total,
"successful_requests": successful,
"failed_requests": total - successful,
"error_rate_percent": round((total - successful) / total * 100, 2),
"avg_response_time_ms": round(statistics.mean(response_times), 2) if response_times else 0,
"min_response_time_ms": round(min(response_times), 2) if response_times else 0,
"max_response_time_ms": round(max(response_times), 2) if response_times else 0,
"median_response_time_ms": round(statistics.median(response_times), 2) if response_times else 0,
"requests_per_second": round(total / (time.time() - self.history[0]["timestamp"].timestamp()) if self.history else 0, 2)
}
return stats
def generate_report(self) -> pd.DataFrame:
"""Génère un rapport complet des performances"""
df = pd.DataFrame(self.history)
return df
Exemple d'utilisation complète
if __name__ == "__main__":
# Initialisation du moniteur avec HolySheep AI
monitor = APIMonitor(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1"
)
# Simulation de 10 requêtes pour tester le monitoring
print("=" * 60)
print("TEST DE MONITORING API - HolySheep AI")
print("=" * 60)
for i in range(10):
result = monitor.call_api(f"Explique-moi le concept {i} en une phrase")
status = "✓" if result["success"] else "✗"
print(f"Requête {i+1}: {status} - {result['response_time_ms']:.2f}ms")
# Affichage des statistiques
print("\n" + "=" * 60)
print("STATISTIQUES DE PERFORMANCE")
print("=" * 60)
stats = monitor.get_statistics()
for key, value in stats.items():
print(f"{key}: {value}")Visualisation des Métriques de Performance
Les chiffres bruts sont difficiles à interpréter. Créons des graphiques professionnels qui vous permettront de comprendre instantanément la santé de votre API.
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime, timedelta
import numpy as np
class PerformanceVisualizer:
"""
Génère des visualisations professionnelles des métriques API.
"""
def __init__(self, monitor: 'APIMonitor'):
self.monitor = monitor
self.set_style()
def set_style(self):
"""Configure le style des graphiques"""
plt.style.use('seaborn-v0_8-whitegrid')
plt.rcParams.update({
'figure.figsize': (14, 10),
'font.size': 11,
'axes.titlesize': 14,
'axes.labelsize': 12
})
def plot_response_time_history(self, save_path: str = None):
"""Affiche l'historique des temps de réponse"""
history = self.monitor.history
if not history:
print("Aucune donnée à afficher")
return
df = pd.DataFrame(history)
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# Graphique 1: Timeline des temps de réponse
ax1 = axes[0, 0]
successful = df[df['success'] == True]
failed = df[df['success'] == False]
ax1.plot(range(len(successful)), successful['response_time_ms'],
'b-o', label='Succès', markersize=6)
if len(failed) > 0:
ax1.scatter(range(len(df))[len(successful):],
[np.nan] * len(failed),
color='red', marker='x', s=100, label='Échec')
ax1.axhline(y=successful['response_time_ms'].mean(),
color='green', linestyle='--', label=f'Moyenne: {successful["response_time_ms"].mean():.1f}ms')
ax1.fill_between(range(len(successful)),
0, successful['response_time_ms'],
alpha=0.3)
ax1.set_xlabel('Numéro de requête')
ax1.set_ylabel('Temps de réponse (ms)')
ax1.set_title('Historique des Temps de Réponse - HolySheep AI')
ax1.legend()
ax1.grid(True, alpha=0.3)
# Graphique 2: Distribution des temps de réponse
ax2 = axes[0, 1]
ax2.hist(successful['response_time_ms'], bins=15,
color='#3498db', edgecolor='white', alpha=0.8)
ax2.axvline(successful['response_time_ms'].mean(),
color='red', linestyle='--', linewidth=2,
label=f'Moyenne: {successful["response_time_ms"].mean():.1f}ms')
ax2.axvline(successful['response_time_ms'].median(),
color='orange', linestyle='--', linewidth=2,
label=f'Médiane: {successful["response_time_ms"].median():.1f}ms')
ax2.set_xlabel('Temps de réponse (ms)')
ax2.set_ylabel('Fréquence')
ax2.set_title('Distribution des Temps de Réponse')
ax2.legend()
# Graphique 3: Taux d'erreur cumulatif
ax3 = axes[1, 0]
df['is_error'] = ~df['success']
df['cumulative_errors'] = df['is_error'].cumsum()
df['cumulative_rate'] = df['cumulative_errors'] / (df.index + 1) * 100
ax3.plot(range(len(df)), df['cumulative_rate'],
'r-', linewidth=2, marker='o')
ax3.axhline(y=1, color='green', linestyle='--',
label='Seuil cible: 1%')
ax3.set_xlabel('Nombre de requêtes')
ax3.set_ylabel('Taux d\'erreur (%)')
ax3.set_title('Évolution du Taux d\'Erreur')
ax3.legend()
ax3.set_ylim(0, max(10, df['cumulative_rate'].max() * 1.2))
# Graphique 4: Statistiques résumées
ax4 = axes[1, 1]
ax4.axis('off')
stats = self.monitor.get_statistics()
stats_text = f"""
╔══════════════════════════════════════════════════╗
║ RAPPORT DE PERFORMANCE ║
╠══════════════════════════════════════════════════╣
║ ║
║ 📊 Requêtes totales: {stats['total_requests']:>6} ║
║ ✅ Succès: {stats['successful_requests']:>6} ║
║ ❌ Échecs: {stats['failed_requests']:>6} ║
║ ⚠️ Taux d'erreur: {stats['error_rate_percent']:>5.2f}% ║
║ ║
║ ⏱️ Temps de réponse moyen: {stats['avg_response_time_ms']:>6.1f} ms ║
║ ⚡ Temps min: {stats['min_response_time_ms']:>6.1f} ms ║
║ 🐢 Temps max: {stats['max_response_time_ms']:>6.1f} ms ║
║ 📈 Médiane: {stats['median_response_time_ms']:>6.1f} ms ║
║ ║
║ 💨 Requêtes/seconde: {stats['requests_per_second']:>6.2f} ║
║ ║
╚══════════════════════════════════════════════════╝
"""
ax4.text(0.1, 0.5, stats_text, fontsize=11,
family='monospace', verticalalignment='center',
bbox=dict(boxstyle='round', facecolor='#ecf0f1', alpha=0.8))
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
print(f"Graphique sauvegardé: {save_path}")
plt.show()
Exécution du script de visualisation
if __name__ == "__main__":
from api_monitor import APIMonitor
# Création du moniteur
monitor = APIMonitor(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2" # Modèle économique: $0.42/MTok
)
# Génération de données de test
print("Génération des données de test...")
prompts = [
"Bonjour, comment allez-vous ?",
"Expliquez-moi l'intelligence artificielle",
"Qu'est-ce qu'une API REST ?",
"Comment réduire les coûts Cloud ?",
"Décrivez la photosynthesis",
]
for prompt in prompts:
monitor.call_api(prompt, max_tokens=50)
# Création des visualisations
visualizer = PerformanceVisualizer(monitor)
visualizer.plot_response_time_history("rapport_performance.png")
print("\nRapport généré avec succès !")Dashboard Complet en Temps Réel
Pour une surveillance continue, je vous recommande ce dashboard léger qui affiche les métriques en temps réel dans votre terminal.
import os
import sys
import time
from datetime import datetime
class RealtimeDashboard:
"""
Dashboard temps réel pour terminal.
Affiche les métriques en continu sans navigateur.
"""
def __init__(self, monitor: 'APIMonitor'):
self.monitor = monitor
self.refresh_interval = 2 # Secondes entre chaque rafraîchissement
def clear_screen(self):
"""Efface le terminal de manière cross-platform"""
os.system('cls' if os.name == 'nt' else 'clear')
def format_time(self, ms: float) -> str:
"""Formate le temps de réponse avec couleur ANSI"""
if ms < 50:
return f"\033[92m{ms:.1f}ms\033[0m" # Vert
elif ms < 100:
return f"\033[93m{ms:.1f}ms\033[0m" # Jaune
else:
return f"\033[91m{ms:.1f}ms\033[0m" # Rouge
def display_header(self):
"""Affiche l'en-tête du dashboard"""
print("╔" + "═" * 78 + "╗")
print("║" + " HOLYSHEEP AI - MONITEUR DE PERFORMANCE EN TEMPS RÉEL ".center(78) + "║")
print("║" + f" Modèle: {self.monitor.model} | URL: {self.monitor.base_url} ".center(78) + "║")
print("╠" + "═" * 78 + "╣")
def display_metrics_bar(self):
"""Affiche la barre de métriques principales"""
stats = self.monitor.get_statistics()
error_indicator = "✓ OK" if stats['error_rate_percent'] < 1 else "⚠ ALERTE"
error_color = "\033[92m" if stats['error_rate_percent'] < 1 else "\033[91m"
print(f"║ 📊 Requêtes: {stats['total_requests']:>4} | "
f"⏱️ Avg: {self.format_time(stats['avg_response_time_ms'])} | "
f"⚠️ Erreurs: {error_color}{stats['error_rate_percent']:.1f}%\033[0m | "
f"💨 RPS: {stats['requests_per_second']:.2f}".ljust(60) + "║")
def display_recent_requests(self, count: int = 8):
"""Affiche les dernières requêtes"""
history = self.monitor.history[-count:] if len(self.monitor.history) >= count else self.monitor.history
print("╠" + "═" * 78 + "╣")
print("║ DERNIÈRES REQUÊTES".ljust(78) + "║")
print("║" + "─" * 78 + "║")
for i, req in enumerate(history):
status = "\033[92m✓\033[0m" if req['success'] else "\033[91m✗\033[0m"
time_str = self.format_time(req['response_time_ms'])
timestamp = req['timestamp'].strftime("%H:%M:%S")
error_note = f" - {req['error'][:30]}..." if req['error'] else ""
line = f" {status} [{timestamp}] {time_str}{error_note}"
print(f"║{line.ljust(78)}║")
def display_footer(self):
"""Affiche le pied de page"""
print("╠" + "═" * 78 + "╣")
print("║ Appuyez sur Ctrl+C pour arrêter | Rafraîchissement: toutes les 2s".center(78) + "║")
print("╚" + "═" * 78 + "╝")
def run(self):
"""Lance le dashboard en boucle"""
try:
while True:
self.clear_screen()
self.display_header()
self.display_metrics_bar()
self.display_recent_requests()
self.display_footer()
time.sleep(self.refresh_interval)
except KeyboardInterrupt:
print("\n\nDashboard arrêté. Au revoir !")
stats = self.monitor.get_statistics()
print(f"\nRécapitulatif final: {stats['total_requests']} requêtes traitées, "
f"taux d'erreur: {stats['error_rate_percent']}%")
Script principal pour lancer le dashboard
if __name__ == "__main__":
from api_monitor import APIMonitor
# Initialisation
monitor = APIMonitor(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gemini-2.5-flash" # Prix: $2.50/MTok - excellent rapport qualité/vitesse
)
# Test initial
print("Test de connexion à HolySheep AI...")
test = monitor.call_api("Dis 'OK' si tu me lis")
print(f"Résultat du test: {'Connexion réussie ✓' if test['success'] else 'Échec ✗'}")
if test['success']:
dashboard = RealtimeDashboard(monitor)
print("\nDémarrage du dashboard en temps réel...\n")
time.sleep(1)
dashboard.run()
else:
print(f"Erreur: {test['error']}")Erreurs Courantes et Solutions
Après des centaines d'heures de debugging pour mes lecteurs, j'ai compilé les 3 erreurs les plus fréquentes que vous rencontrerez certainement.
Erreur 1 : Erreur 401 - Clé API Invalide ou Mancuante
# ❌ ERREUR FRÉQUENTE #1: 401 Unauthorized
Symptôme: {"error": {"message": "Incorrect API key provided", "type": "invalid_request_error"}}
Solutions à appliquer:
Solution A: Vérifiez votre variable d'environnement
import os
print(f" HOLYSHEEP_API_KEY = {os.getenv('HOLYSHEEP_API_KEY', 'NON DÉFINIE')}")
Solution B: Chargez explicitement le fichier .env
from dotenv import load_dotenv
load_dotenv() # Ajoute ce ligne au début de votre script
Solution C: Vérifiez le format de la clé
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
if API_KEY and len(API_KEY) < 20:
print("⚠️ Votre clé API semble trop courte. Vérifiez sur https://www.holysheep.ai/dashboard")
else:
print("✓ Format de clé OK")
Solution D: Test de connexion complet
import requests
def diagnose_connection():
"""Diagnostique complet de la connexion"""
api_key = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
base_url = "https://api.holysheep.ai/v1"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
print("\n🔍 DIAGNOSTIC DE CONNEXION")
print("-" * 50)
# Test 1: Ping simple
try:
response = requests.get(f"{base_url}/models", headers=headers, timeout=5)
print(f"1. GET /models: {response.status_code}")
if response.status_code == 200:
print(" ✓ Connexion établie")
models = response.json().get("data", [])
print(f" 📦 {len(models)} modèles disponibles")
elif response.status_code == 401:
print(" ✗ Clé API invalide")
print(" → Inscrivez-vous sur https://www.holysheep.ai/register")
except Exception as e:
print(f" ✗ Erreur: {e}")
diagnose_connection()Erreur 2 : Timeout - L'API ne Répond Pas
# ❌ ERREUR FRÉQUENTE #2: Timeout
Symptôme: requests.exceptions.ReadTimeout ou timeout de 30s dépassé
Solutions à appliquer:
Solution A: Augmentez le timeout
import requests
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Une question complexe"}],
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {os.getenv('HOLYSHEEP_API_KEY')}",
"Content-Type": "application/json"
}
Timeout adaptatif selon la complexité
try:
# Pour une requête simple: 10 secondes suffisent
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=10 # Augmenté de 5 à 10 secondes
)
print(f"✓ Réponse reçue en {response.elapsed.total_seconds():.2f}s")
except requests.exceptions.Timeout:
print("⚠️ Timeout - La requête est trop longue")
print("→ Solutions: réduire max_tokens ou utiliser un modèle plus rapide")
print("→ HolySheep propose Gemini 2.5 Flash ($2.50/MTok) optimisé pour la vitesse")
Solution B: Implémentez des retries automatiques
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_retries():
"""Crée une session avec retry automatique"""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
Utilisation
session = create_session_with_retries()
print("✓ Session avec retry automatique créée")
Solution C: Gérez le timeout de manière élégante
import socket
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def call_with_fallback_timeout(api_key, prompt, timeout_primary=5, timeout_secondary=15):
"""
Appelle l'API avec timeout progressif
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 100
}
# Tentative 1: Timeout court pour requêtes simples
try:
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=timeout_primary
)
return response.json()
except requests.exceptions.Timeout:
print(f"⏱️ Timeout primaire ({timeout_primary}s) dépassé")
# Tentative 2: Timeout long avec message d'avertissement
print(" → Nouvelle tentative avec timeout étendu...")
try:
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=timeout_secondary
)
return response.json()
except requests.exceptions.Timeout:
return {"error": "Échec après 2 tentatives - serveur peut-être surchargé"}Erreur 3 : Erreur de Limite de Quota (429 Too Many Requests)
# ❌ ERREUR FRÉQUENTE #3: 429 Rate Limit Exceeded
Symptôme: {"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
Solutions à appliquer:
Solution A: Implémentez un rate limiter
import time
from collections import deque
from threading import Lock
class RateLimiter:
"""
Limiteur de taux de requêtes pour éviter les erreurs 429.
"""
def __init__(self, max_requests_per_minute: int = 60):
self.max_requests = max_requests_per_minute
self.requests = deque()
self.lock = Lock()
def wait_if_needed(self):
"""Attend si nécessaire pour respecter le rate limit"""
with self.lock:
now = time.time()
# Supprime les requêtes anciennes (plus de 60 secondes)
while self.requests and self.requests[0] < now - 60:
self.requests.popleft()
# Si on a atteint la limite, attend
if len(self.requests) >= self.max_requests:
sleep_time = 60 - (now - self.requests[0])
print(f"⏳ Rate limit atteint. Attente de {sleep_time:.1f}s...")
time.sleep(sleep_time)
# Nettoie après l'attente
while self.requests and self.requests[0] < time.time() - 60:
self.requests.popleft()
# Enregistre cette requête
self.requests.append(time.time())
def get_remaining(self) -> int:
"""Retourne le nombre de requêtes restantes"""
with self.lock:
now = time.time()
while self.requests and self.requests[0] < now - 60:
self.requests.popleft()
return self.max_requests - len(self.requests)
Utilisation du rate limiter
rate_limiter = RateLimiter(max_requests_per_minute=30) # 30 req/min pour être sûr
print("Test du rate limiter:")
for i in range(35):
rate_limiter.wait_if_needed()
remaining = rate_limiter.get_remaining()
print(f"Requête {i+1}: {remaining} requêtes restantes")
# Simule un appel API
time.sleep(0.5)
Solution B: Utilisation avec le monitoring
class MonitoredAPIClient:
"""
Client API avec rate limiting intégré et monitoring.
"""
def __init__(self, api_key: str, model: str = "gpt-4.1"):
self.api_key = api_key
self.model = model
self.rate_limiter = RateLimiter(max_requests_per_minute=60)
self.stats = {"requests": 0, "errors_429": 0, "success": 0}
def call(self, prompt: str) -> dict:
"""Appelle l'API avec rate limiting automatique"""
self.rate_limiter.wait_if_needed()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}]
}
try:
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=30
)
self.stats["requests"] += 1
if response.status_code == 429:
self.stats["errors_429"] += 1
wait_time = int(response.headers.get("Retry-After", 5))
print(f"⚠️ Rate limit. Attente de {wait_time}s...")
time.sleep(wait_time)
return self.call(prompt) # Retry
elif response.status_code == 200:
self.stats["success"] += 1
return response.json()
else:
return {"error": f"HTTP {response.status_code}"}
except Exception as e:
return {"error": str(e)}
def print_stats(self):
"""Affiche les statistiques d'utilisation"""
total = self.stats["requests"]
success_rate = (self.stats["success"] / total * 100) if total > 0