Introduction
En tant qu'ingénieur qui a supervisé des centaines de millions d'appels API sur des systèmes de production critiques, je peux vous confirmer que la qualité des sorties de modèles IA n'est jamais acquise. Un modèle qui performe admirablement en phase de test peut dévier significativement en production — c'est ce qu'on appelle la « dérive de modèle » (model drift). Ce tutoriel détaille l'architecture complète d'un système de monitoring statistique que j'ai déployé sur plusieurs infrastuctures critiques, avec des mesures concrètes et du code production-ready.
La surveillance proactive des métriques de qualité vous permet d'intervenir avant que les dégradations n'impactent vos utilisateurs. Avec HolySheep AI, nous disposons d'une plateforme offrant moins de 50ms de latence et des coûts avantageux (DeepSeek V3.2 à $0.42 par million de tokens), rendant le monitoring rentable même à très haut volume.
Architecture du Système de Monitoring
Principes Fondamentaux
Un système de monitoring statistique robuste repose sur trois piliers : la collecte continue de métriques, l'analyse statistique en temps réel, et les alertes intelligentes. L'architecture que je recommande sépare clairement ces responsabilités pour éviter les goulots d'étranglement.
┌─────────────────────────────────────────────────────────────────┐
│ PIPELINE DE MONITORING │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────┐ │
│ │ Collecte │───▶│ Agrégation │───▶│ Analyse Statistique │ │
│ │ Temps │ │ Fenêtre │ │ Contrôle Qualité │ │
│ │ Réel │ │ Glissante │ │ │ │
│ └──────────┘ └──────────────┘ └─────────────────────┘ │
│ │ │
│ ┌──────────────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ TABLEAU DE BORD + ALERTES │ │
│ └────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Architecture Recommandée
Pour un système résilient, je préconise une architecture event-driven avec Kafka comme bus de messages. Les métriques sont agrégées par fenêtres glissantes de 5 minutes, permettant une détection rapide des anomalies tout en lissant les fluctuations aléatoires.
Implémentation du Collecteur de Métriques
Client Python Production-Ready
Le cœur du système réside dans un client capable de capturer chaque sortie avec ses métriques associées. Voici mon implémentation complète, battle-tested sur plusieurs projets :
import time
import hashlib
import asyncio
import aiohttp
import statistics
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class OutputMetrics:
"""Métriques associées à une sortie de modèle IA."""
request_id: str
model_id: str
timestamp: float
input_tokens: int
output_tokens: int
latency_ms: float
confidence_score: Optional[float] = None
error: Optional[str] = None
custom_metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class StatisticalSummary:
"""Résumé statistique d'une fenêtre de métriques."""
window_start: float
window_end: float
count: int
mean_latency: float
median_latency: float
p95_latency: float
p99_latency: float
std_latency: float
error_rate: float
avg_tokens_per_request: float
throughput_rps: float
class QualityMonitor:
"""
Moniteur de qualité statistique pour sorties de modèles IA.
Capture, agrège et analyse les métriques en temps réel.
"""
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
window_seconds: int = 300,
alert_threshold_p99: float = 2000.0,
alert_error_rate: float = 0.05
):
self.base_url = base_url
self.api_key = api_key
self.window_seconds = window_seconds
self.alert_threshold_p99 = alert_threshold_p99
self.alert_error_rate = alert_error_rate
# Buffer circulaire pour la fenêtre glissante
self.metrics_buffer: deque[OutputMetrics] = deque(maxlen=10000)
self.window_start = time.time()
# Cache pour analyse par modèle
self.model_metrics: Dict[str, deque] = {}
# Session HTTP persistente
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Obtient ou crée une session HTTP persistante."""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self._session
async def generate_request_id(self) -> str:
"""Génère un ID unique pour la requête."""
timestamp = str(time.time()).encode()
random_bytes = str(id(self)).encode()
return hashlib.sha256(timestamp + random_bytes).hexdigest()[:16]
async def call_model(
self,
prompt: str,
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048,
enable_monitoring: bool = True
) -> Dict[str, Any]:
"""
Appelle le modèle IA avec capture complète des métriques.
"""
session = await self._get_session()
request_id = await self.generate_request_id()
start_time = time.time()
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"max_tokens": max_tokens
}
metrics = OutputMetrics(
request_id=request_id,
model_id=model,
timestamp=start_time,
input_tokens=0,
output_tokens=0,
latency_ms=0
)
try:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload
) as response:
end_time = time.time()
latency_ms = (end_time - start_time) * 1000
if response.status == 200:
data = await response.json()
metrics = OutputMetrics(
request_id=request_id,
model_id=model,
timestamp=start_time,
input_tokens=data.get("usage", {}).get("prompt_tokens", 0),
output_tokens=data.get("usage", {}).get("completion_tokens", 0),
latency_ms=latency_ms
)
metrics.custom_metadata["finish_reason"] = data.get("choices", [{}])[0].get("finish_reason")
metrics.custom_metadata["response_id"] = data.get("id")
return {
"success": True,
"content": data.get("choices", [{}])[0].get("message", {}).get("content"),
"metrics": metrics
}
else:
error_text = await response.text()
metrics.error = f"HTTP {response.status}: {error_text}"
return {"success": False, "error": metrics.error, "metrics": metrics}
except Exception as e:
metrics.error = str(e)
metrics.latency_ms = (time.time() - start_time) * 1000
return {"success": False, "error": str(e), "metrics": metrics}
finally:
if enable_monitoring:
self.record_metrics(metrics)
def record_metrics(self, metrics: OutputMetrics) -> None:
"""Enregistre les métriques dans le buffer."""
self.metrics_buffer.append(metrics)
# Indexation par modèle pour analyses ciblées
if metrics.model_id not in self.model_metrics:
self.model_metrics[metrics.model_id] = deque(maxlen=5000)
self.model_metrics[metrics.model_id].append(metrics)
# Rotation de fenêtre si nécessaire
current_time = time.time()
if current_time - self.window_start >= self.window_seconds:
self._rotate_window()
def _rotate_window(self) -> None:
"""Effectue la rotation de la fenêtre de métriques."""
old_start = self.window_start
self.window_start = time.time()
logger.info(f"Fenêtre pivotée: {old_start} -> {self.window_start}")
def get_summary(self, model_id: Optional[str] = None) -> StatisticalSummary:
"""
Calcule le résumé statistique de la fenêtre actuelle.
Retourne des métriques complètes pour l'analyse de qualité.
"""
buffer = self.model_metrics.get(model_id, self.metrics_buffer) if model_id else self.metrics_buffer
if len(buffer) == 0:
return StatisticalSummary(
window_start=self.window_start,
window_end=time.time(),
count=0,
mean_latency=0,
median_latency=0,
p95_latency=0,
p99_latency=0,
std_latency=0,
error_rate=0,
avg_tokens_per_request=0,
throughput_rps=0
)
latencies = [m.latency_ms for m in buffer]
errors = sum(1 for m in buffer if m.error is not None)
total_tokens = sum(m.input_tokens + m.output_tokens for m in buffer)
sorted_latencies = sorted(latencies)
p95_idx = int(len(sorted_latencies) * 0.95)
p99_idx = int(len(sorted_latencies) * 0.99)
window_duration = time.time() - self.window_start
return StatisticalSummary(
window_start=self.window_start,
window_end=time.time(),
count=len(buffer),
mean_latency=statistics.mean(latencies),
median_latency=statistics.median(latencies),
p95_latency=sorted_latencies[p95_idx] if sorted_latencies else 0,
p99_latency=sorted_latencies[p99_idx] if sorted_latencies else 0,
std_latency=statistics.stdev(latencies) if len(latencies) > 1 else 0,
error_rate=errors / len(buffer),
avg_tokens_per_request=total_tokens / len(buffer),
throughput_rps=len(buffer) / window_duration if window_duration > 0 else 0
)
def check_alerts(self) -> List[Dict[str, Any]]:
"""Vérifie les conditions d'alerte et retourne les alertes actives."""
alerts = []
summary = self.get_summary()
if summary.p99_latency > self.alert_threshold_p99:
alerts.append({
"type": "HIGH_LATENCY",
"severity": "WARNING",
"message": f"P99 latence ({summary.p99_latency:.2f}ms) dépasse le seuil ({self.alert_threshold_p99}ms)",
"metric": summary.p99_latency,
"threshold": self.alert_threshold_p99
})
if summary.error_rate > self.alert_error_rate:
alerts.append({
"type": "HIGH_ERROR_RATE",
"severity": "CRITICAL",
"message": f"Taux d'erreur ({summary.error_rate*100:.2f}%) dépasse le seuil ({self.alert_error_rate*100}%)",
"metric": summary.error_rate,
"threshold": self.alert_error_rate
})
return alerts
async def close(self):
"""Ferme les ressources."""
if self._session and not self._session.closed:
await self._session.close()
Exemple d'utilisation
async def main():
monitor = QualityMonitor(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Simulation de requêtes avec monitoring
prompts = [
"Explique la différence entre RAID 5 et RAID 6",
"Comment implémenter un cache LRU en Python?",
"Décris l'algorithme de Dijkstra"
]
for prompt in prompts:
result = await monitor.call_model(prompt, model="deepseek-v3.2")
if result["success"]:
logger.info(f"✓ Requête réussie: {result['content'][:50]}...")
else:
logger.error(f"✗ Erreur: {result['error']}")
# Récupération des statistiques
summary = monitor.get_summary()
logger.info(f"""
╔══════════════════════════════════════════════╗
║ RAPPORT DE QUALITÉ ║
╠══════════════════════════════════════════════╣
║ Requêtes totales : {summary.count:>4} ║
║ Latence moyenne : {summary.mean_latency:>8.2f} ms ║
║ Latence P95 : {summary.p95_latency:>8.2f} ms ║
║ Latence P99 : {summary.p99_latency:>8.2f} ms ║
║ Taux d'erreur : {summary.error_rate*100:>8.2f} % ║
║ Débits : {summary.throughput_rps:>8.2f} req/s ║
╚══════════════════════════════════════════════╝
""")
# Vérification des alertes
alerts = monitor.check_alerts()
for alert in alerts:
logger.warning(f"⚠ {alert['type']}: {alert['message']}")
await monitor.close()
if __name__ == "__main__":
asyncio.run(main())
Optimisation des Performances et Contrôle de Concurrence
Gestion des Burst de Trafic
En production, le trafic suit rarement une courbe prévisible. Les pics soudains peuvent submerger votre système de monitoring si vous n'implémentez pas un contrôle de concurrence robuste. Mon implémentation utilise un sémaphore configurable permettant de limiter le nombre de requêtes concurrentes tout en optimisant le throughput global.
import asyncio
from typing import List, Callable, Any, Optional
from dataclasses import dataclass
import time
import logging
from collections import defaultdict
logger = logging.getLogger(__name__)
@dataclass
class ConcurrencyConfig:
"""Configuration du contrôle de concurrence."""
max_concurrent_requests: int = 50
max_queue_size: int = 1000
timeout_per_request: float = 30.0
retry_attempts: int = 3
retry_backoff_base: float = 1.5
circuit_breaker_threshold: int = 100
circuit_breaker_timeout: float = 60.0
class ConcurrencyController:
"""
Contrôleur de concurrence intelligent avec circuit breaker.
Gère les bursts de trafic tout en maintenant la stabilité.
"""
def __init__(self, config: Optional[ConcurrencyConfig] = None):
self.config = config or ConcurrencyConfig()
self.semaphore = asyncio.Semaphore(self.config.max_concurrent_requests)
self.queue: asyncio.Queue = asyncio.Queue(maxsize=self.config.max_queue_size)
# Métriques de monitoring
self.active_requests = 0
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
self.rejected_requests = 0
# Circuit breaker state
self.circuit_open = False
self.circuit_open_time: Optional[float] = None
self.failure_count = 0
self.success_count = 0
# Rate limiting
self.request_timestamps: List[float] = []
self.rate_limit_window = 1.0 # 1 seconde
# Lock pour thread-safety
self._lock = asyncio.Lock()
def _update_rate_limiter(self) -> bool:
"""Vérifie et met à jour le rate limiter. Retourne True si la requête est autorisée."""
current_time = time.time()
# Nettoyer les timestamps vieux
cutoff = current_time - self.rate_limit_window
self.request_timestamps = [ts for ts in self.request_timestamps if ts > cutoff]
# Vérifier la limite
if len(self.request_timestamps) >= self.config.max_concurrent_requests:
return False
self.request_timestamps.append(current_time)
return True
async def _check_circuit_breaker(self) -> bool:
"""Vérifie l'état du circuit breaker."""
async with self._lock:
if not self.circuit_open:
return True
# Vérifier si le timeout est écoulé
if (time.time() - self.circuit_open_time) >= self.config.circuit_breaker_timeout:
logger.info("Circuit breaker: passage à demi-ouvert")
self.circuit_open = False
self.failure_count = 0
return True
return False
def _record_success(self):
"""Enregistre un succès pour le circuit breaker."""
self.success_count += 1
if self.success_count >= 10: # 10 succès pour fermer le circuit
self.circuit_open = False
self.failure_count = 0
logger.info("Circuit breaker: fermé après succes")
def _record_failure(self):
"""Enregistre un échec pour le circuit breaker."""
self.failure_count += 1
if self.failure_count >= self.config.circuit_breaker_threshold:
self.circuit_open = True
self.circuit_open_time = time.time()
logger.warning(f"Circuit breaker: ouvert après {self.failure_count} échecs")
async def execute_with_control(
self,
coro: Callable,
*args,
**kwargs
) -> Any:
"""
Exécute une coroutine avec contrôle de concurrence complet.
Inclut retry, timeout et circuit breaker.
"""
if not await self._check_circuit_breaker():
raise CircuitBreakerOpenError("Circuit breaker est ouvert")
if not self._update_rate_limiter():
self.rejected_requests += 1
raise QueueFullError("File d'attente pleine")
async with self._lock:
self.active_requests += 1
self.total_requests += 1
last_exception = None
for attempt in range(self.config.retry_attempts):
try:
async with asyncio.timeout(self.config.timeout_per_request):
result = await coro(*args, **kwargs)
self._record_success()
async with self._lock:
self.active_requests -= 1
self.successful_requests += 1
return result
except asyncio.TimeoutError:
last_exception = TimeoutError(f"Timeout après {self.config.timeout_per_request}s")
logger.warning(f"Timeout attempt {attempt + 1}/{self.config.retry_attempts}")
except Exception as e:
last_exception = e
logger.warning(f"Erreur attempt {attempt + 1}/{self.config.retry_attempts}: {e}")
if attempt < self.config.retry_attempts - 1:
sleep_time = self.config.retry_backoff_base ** attempt
await asyncio.sleep(sleep_time)
self._record_failure()
async with self._lock:
self.active_requests -= 1
self.failed_requests += 1
raise MaxRetriesExceededError(f"Échec après {self.config.retry_attempts} tentatives: {last_exception}")
def get_stats(self) -> Dict[str, Any]:
"""Retourne les statistiques du contrôleur."""
return {
"active_requests": self.active_requests,
"total_requests": self.total_requests,
"successful_requests": self.successful_requests,
"failed_requests": self.failed_requests,
"rejected_requests": self.rejected_requests,
"success_rate": self.successful_requests / self.total_requests if self.total_requests > 0 else 0,
"circuit_breaker_open": self.circuit_open,
"queue_size": self.queue.qsize(),
"max_concurrent": self.config.max_concurrent_requests
}
class CircuitBreakerOpenError(Exception):
"""Le circuit breaker est ouvert."""
pass
class QueueFullError(Exception):
"""La file d'attente est pleine."""
pass
class MaxRetriesExceededError(Exception):
"""Nombre maximum de tentatives dépassé."""
pass
Démonstration avec monitoring de qualité
async def demo_monitoring_with_concurrency():
"""Démonstration complète du système de monitoring avec contrôle de concurrence."""
import aiohttp
monitor = QualityMonitor()
controller = ConcurrencyController(
ConcurrencyConfig(
max_concurrent_requests=20,
circuit_breaker_threshold=50,
retry_attempts=2
)
)
async def monitored_api_call(prompt: str, model: str) -> Dict:
"""Appel API monitoré avec contrôle de concurrence."""
return await controller.execute_with_control(
monitor.call_model,
prompt=prompt,
model=model
)
# Burst de test
test_prompts = [
f"Requête {i}: Analyse technique détaillée" for i in range(100)
]
start_time = time.time()
results = []
errors = []
# Exécution concurrente limitée
tasks = [
monitored_api_call(prompt, "deepseek-v3.2")
for prompt in test_prompts
]
for coro in asyncio.as_completed(tasks):
try:
result = await coro
results.append(result)
except Exception as e:
errors.append(str(e))
duration = time.time() - start_time
print(f"""
╔════════════════════════════════════════════════════════╗
║ RÉSULTATS DU TEST DE CONCURRENCE ║
╠════════════════════════════════════════════════════════╣
║ Durée totale : {duration:>10.2f} s ║
║ Requêtes totales : {len(results) + len(errors):>10} ║
║ Succès : {len(results):>10} ║
║ Erreurs : {len(errors):>10} ║
║ Throughput moyen : {(len(results) + len(errors)) / duration:>10.2f} req/s ║
╚════════════════════════════════════════════════════════╝
""")
# Statistiques du contrôleur
stats = controller.get_stats()
print(f"Circuit breaker ouvert: {stats['circuit_breaker_open']}")
print(f"Taux de succès: {stats['success_rate']*100:.2f}%")
await monitor.close()
if __name__ == "__main__":
asyncio.run(demo_monitoring_with_concurrency())
Optimisation des Coûts avec Analyse Statistique
Estimation et Contrôle des Coûts en Temps Réel
La facturation des API IA peut rapidement exploser si vous ne monitorer pas activement votre consommation. Avec les tarifs HolySheep AI (DeepSeek V3.2 à $0.42/M tokens, GPT-4.1 à $8/M tokens), chaque requête compte. Mon système calcule les coûts en temps réel avec alertes de budget.
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime, timedelta
import time
Tarifs HolySheep AI 2026 (en USD par million de tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 2.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 3.75, "output": 15.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.10, "output": 0.42}
}
@dataclass
class CostAlert:
"""Configuration d'alerte de coût."""
threshold_usd: float
window_hours: int
callback: Optional[Callable] = None
class CostAnalyzer:
"""
Analyseur de coûts pour API IA avec budgets et alertes.
Calcule les coûts en temps réel et prédit les dépenses.
"""
def __init__(
self,
pricing: Optional[Dict[str, Dict[str, float]]] = None,
budget_daily_usd: float = 100.0,
currency: str = "USD"
):
self.pricing = pricing or HOLYSHEEP_PRICING
self.budget_daily_usd = budget_daily_usd
self.currency = currency
# Historique des coûts
self.cost_history: Dict[str, List[Dict]] = defaultdict(list)
self.total_cost = 0.0
# Budgets par modèle
self.model_budgets: Dict[str, float] = {}
# Alertes configurées
self.alerts: List[CostAlert] = []
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict[str, float]:
"""Calcule le coût exact d'une requête."""
if model not in self.pricing:
raise ValueError(f"Modèle '{model}' non trouvé dans les tarifs")
rates = self.pricing[model]
input_cost = (input_tokens / 1_000_000) * rates["input"]
output_cost = (output_tokens / 1_000_000) * rates["output"]
total_cost = input_cost + output_cost
return {
"input_cost": round(input_cost, 6),
"output_cost": round(output_cost, 6),
"total_cost": round(total_cost, 6),
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens
}
def record_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
request_id: str,
timestamp: Optional[float] = None
) -> Dict[str, float]:
"""Enregistre l'utilisation et calcule le coût."""
cost_breakdown = self.calculate_cost(model, input_tokens, output_tokens)
cost_breakdown["model"] = model
cost_breakdown["request_id"] = request_id
cost_breakdown["timestamp"] = timestamp or time.time()
# Ajout à l'historique
self.cost_history[model].append(cost_breakdown)
self.total_cost += cost_breakdown["total_cost"]
# Vérification des alertes
self._check_cost_alerts(cost_breakdown)
return cost_breakdown
def _check_cost_alerts(self, cost_record: Dict):
"""Vérifie si les alertes de coût doivent être déclenchées."""
current_time = time.time()
for alert in self.alerts:
window_start = current_time - (alert.window_hours * 3600)
# Calcul du coût dans la fenêtre
window_cost = sum(
record["total_cost"]
for records in self.cost_history.values()
for record in records
if record["timestamp"] >= window_start
)
if window_cost >= alert.threshold_usd:
if alert.callback:
alert.callback(
current_cost=window_cost,
threshold=alert.threshold_usd,
window_hours=alert.window_hours
)
def get_cost_summary(
self,
model: Optional[str] = None,
hours: int = 24
) -> Dict[str, any]:
"""Génère un résumé des coûts sur une période."""
cutoff = time.time() - (hours * 3600)
if model:
records = [r for r in self.cost_history.get(model, []) if r["timestamp"] >= cutoff]
else:
records = [
r for records in self.cost_history.values()
for r in records if r["timestamp"] >= cutoff
]
if not records:
return {
"period_hours": hours,
"total_requests": 0,
"total_cost": 0,
"cost_per_request": 0,
"model_breakdown": {}
}
total_cost = sum(r["total_cost"] for r in records)
total_input = sum(r["input_tokens"] for r in records)
total_output = sum(r["output_tokens"] for r in records)
# Répartition par modèle
model_breakdown = {}
for record in records:
m = record["model"]
if m not in model_breakdown:
model_breakdown[m] = {"cost": 0, "requests": 0, "tokens": 0}
model_breakdown[m]["cost"] += record["total_cost"]
model_breakdown[m]["requests"] += 1
model_breakdown[m]["tokens"] += record["total_tokens"]
# Projection journalière
hourly_rate = total_cost / hours if hours > 0 else 0
projected_daily = hourly_rate * 24
budget_utilization = (projected_daily / self.budget_daily_usd) * 100
return {
"period_hours": hours,
"total_requests": len(records),
"total_cost": round(total_cost, 4),
"cost_per_request": round(total_cost / len(records), 6),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"projected_daily_cost": round(projected_daily, 2),
"budget_daily_usd": self.budget_daily_usd,
"budget_utilization_pct": round(budget_utilization, 2),
"model_breakdown": {
m: {
"cost": round(d["cost"], 4),
"requests": d["requests"],
"avg_cost_per_request": round(d["cost"] / d["requests"], 6)
}
for m, d in model_breakdown.items()
}
}
def predict_monthly_cost(self, hours: int = 24) -> Dict[str, float]:
"""Prédit le coût mensuel basé sur l'utilisation récente."""
summary = self.get_cost_summary(hours=hours)
hourly_rate = summary["total_cost"] / hours if hours > 0 else 0
monthly_cost = hourly_rate * 24 * 30
# Distribution par modèle
model_predictions = {}
for model, data in summary["model_breakdown"].items():
if model in self.pricing:
model_hourly = data["cost"] / hours if hours > 0 else 0
model_predictions[model] = round(model_hourly * 24 * 30, 2)
return {
"predicted_monthly_usd": round(monthly_cost, 2),
"model_predictions": model_predictions,
"confidence": "high" if hours >= 24 else "medium",
"sample_hours": hours
}
def add_budget_alert(
self,
threshold_usd: float,
window_hours: int,
callback: Optional[Callable] = None
):
"""Ajoute une alerte de budget."""
self.alerts.append(CostAlert(threshold_usd, window_hours, callback))
def alert_callback(current_cost: float, threshold: float, window_hours: int):
"""Callback exemple pour alerte de coût."""
print(f"⚠️ ALERTE: Coût de ${current_cost:.2f} atteint pour les {window_hours}h dernières")
Démonstration
async def demo_cost_monitoring():
"""Démonstration du système d'analyse des coûts."""
analyzer = CostAnalyzer(budget_daily_usd=50.0)
# Configuration d'alertes
analyzer.add_budget_alert(
threshold_usd=25.0,
window_hours=1,
callback=alert_callback
)
# Simulation d'utilisation réelle
import random
test_scenarios = [
{"model": "deepseek-v3.2", "input": 150, "output": 450}, # $0.000339
{"model": "gpt-4.1", "input": 200, "output": 600}, # $0.0052
{"model": "gemini-2.5-flash", "input": 100, "output": 300}, # $0.00087
{"model": "claude-sonnet-4.5", "input": 180, "output": 520} # $0.00891
]
print("╔══════════════════════════════════════════════════════════╗")
print("║ ANALYSE COMPARATIVE DES COÛTS HOLYSHEEP ║")
print("╠══════════════════════════════════════════════════════════╣")
for i in range(100):
scenario = random.choice(test_scenarios)
cost = analyzer.record_usage(
model=scenario["model"],
input_tokens=scenario["input"],
output_tokens=scenario["output"],
request_id=f"req_{i:06d}"
)
if i % 25 == 0:
print(f"║ Requête {i:3d}: {scenario['model']:<20} - ${cost['total_cost']:.6f} ║")
summary = analyzer.get_cost_summary(hours=1)
prediction = analyzer.predict_monthly_cost(hours=1)
print("╠══════════════════════════════════════════════════════════╣")
print(f"║ Total coûts (1h): ${summary['total_cost']:.4f} ║")
print(f"║ Requêtes: {summary['total_requests']:>4} ║")
print(f"║ Coût moyen/requête: ${summary['cost_per_request']:.6f} ║")
print(f"║ Projection mensuelle: ${prediction['predicted_monthly_usd']:.2f} ║")
print(f"║ Utilisation budget: {summary['budget_utilization_pct']:.1f}% ║")
print("╚══════════════════════════════════════════════════════════╝")
print("\n📊 Répartition par modèle:")
for model, data in summary['model_breakdown'].items():
print(f" {model}: ${data['cost']:.4f} ({data['requests']} requêtes)")
if __name__ == "__main__":
asyncio.run(demo_cost_monitoring())
Tableaux de Bord et Visualisation
Métriques Clés de Performance (KPIs)
Un système de
Ressources connexes
Articles connexes