Introduction : Le Défi de l'Observabilité dans les API IA
En tant qu'ingénieur senior ayant géré l'infrastructure IA de plusieursScale-ups, j'ai été confronté à des factures de plusieurs milliers de dollars en quelques jours à cause d'un boucle infinie mal détectée. L'audit des journaux d'appels API et la détection d'anomalies de consommation constituent aujourd'hui un pilier incontournable de toute architecture IA en production.
Ce guide présente une architecture complète, battle-tested, avec du code production-ready. Nous déploierons un système capable de journaliser chaque appel, détecter les patterns anormaux en temps réel et déclencher des alertes avant que la facture ne s'envole.
Architecture du Système d'Audit
Notre solution repose sur une architecture event-driven avec trois composants principaux : le collecteur de journaux, le moteur de détection d'anomalies, et le système d'alertes. Cette séparation permet une scalabilité horizontale et une maintenabilité optimale.
Schéma de l'Architecture
┌─────────────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE D'AUDIT API IA │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────────┐ │
│ │ Client │───▶│ API Gateway │───▶│ HolySheep API │ │
│ │ Application │ │ (Logging) │ │ base_url: api.holysheep │ │
│ └──────────────┘ └──────┬───────┘ └──────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Audit Logger │ │
│ │ - request_id │ │
│ │ - timestamp │ │
│ │ - tokens_used │ │
│ │ - cost_usd │ │
│ └────────┬─────────┘ │
│ │ │
│ ┌──────────────┼──────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────────┐ │
│ │ InfluxDB │ │ Anomaly │ │ Alert │ │
│ │ (Metrics) │ │ Engine │ │ Dispatcher │ │
│ └────────────┘ └─────┬──────┘ └───────┬────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────┐ │
│ │ Dashboard / Webhook │ │
│ │ (Grafana + Slack + Email) │ │
│ └──────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
Implémentation du Collecteur de Journaux
Le collecteur constitue la fondation de notre système. Il intercepte chaque requête/réponse et stocke les métadonnées essentielles pour l'audit et la détection d'anomalies.
Configuration du Client HTTP avec Interception
# config.py
import os
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class HolySheepConfig:
"""Configuration pour l'API HolySheep avec paramètres d'audit"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
max_retries: int = 3
timeout: int = 60
# Configuration du audit
audit_enabled: bool = True
audit_log_path: str = "/var/log/ai-audit/requests.jsonl"
metrics_endpoint: Optional[str] = None
# Seuil d'alerte (USD par heure)
hourly_budget_threshold: float = 50.0
daily_budget_threshold: float = 200.0
# Patterns de consommation anormale
abnormal_token_threshold: int = 50000 # tokens par minute
abnormal_request_threshold: int = 100 # requêtes par minute
config = HolySheepConfig()
# audit_logger.py
import json
import logging
from datetime import datetime
from typing import Dict, Any, Optional
from dataclasses import dataclass, asdict
import hashlib
import threading
from queue import Queue
import time
@dataclass
class AuditLogEntry:
"""Structure d'une entrée de journal d'audit"""
request_id: str
timestamp: str
endpoint: str
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
latency_ms: float
status_code: int
error_message: Optional[str] = None
user_id: Optional[str] = None
session_id: Optional[str] = None
metadata: Optional[Dict[str, Any]] = None
class AuditLogger:
"""
Logger asynchrone haute performance pour les appels API IA.
Thread-safe avec bufferisation pour minimiser l'impact sur les performances.
"""
def __init__(self, log_path: str, buffer_size: int = 100, flush_interval: int = 5):
self.log_path = log_path
self.buffer_size = buffer_size
self.flush_interval = flush_interval
self.buffer: list[AuditLogEntry] = []
self.buffer_lock = threading.Lock()
self.logger = self._setup_file_logger()
self.last_flush = time.time()
self.total_entries = 0
self.total_cost = 0.0
# Métriques agrégées
self.metrics_lock = threading.Lock()
self.hourly_tokens = 0
self.hourly_cost = 0.0
self.daily_tokens = 0
self.daily_cost = 0.0
# Démarrer le flush périodique
self._start_periodic_flush()
def _setup_file_logger(self) -> logging.Logger:
logger = logging.getLogger(f"audit.{self.log_path}")
logger.setLevel(logging.INFO)
handler = logging.FileHandler(self.log_path)
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)
return logger
def log(self, entry: AuditLogEntry) -> None:
"""Thread-safe logging avec bufferisation"""
with self.buffer_lock:
self.buffer.append(entry)
self.total_entries += 1
self.total_cost += entry.cost_usd
# Mise à jour des métriques
with self.metrics_lock:
self.hourly_tokens += entry.total_tokens
self.hourly_cost += entry.cost_usd
self.daily_tokens += entry.total_tokens
self.daily_cost += entry.cost_usd
# Flush si buffer plein
if len(self.buffer) >= self.buffer_size:
self._flush()
def _flush(self) -> None:
"""Écriture des entrées bufferisées sur disque"""
if not self.buffer:
return
for entry in self.buffer:
self.logger.info(json.dumps(asdict(entry), default=str))
self.buffer.clear()
self.last_flush = time.time()
def _start_periodic_flush(self) -> None:
"""Thread daemon pour flush périodique"""
def periodic_flush():
while True:
time.sleep(self.flush_interval)
with self.buffer_lock:
if time.time() - self.last_flush >= self.flush_interval:
self._flush()
thread = threading.Thread(target=periodic_flush, daemon=True)
thread.start()
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques courantes pour le monitoring"""
with self.metrics_lock:
return {
"total_entries": self.total_entries,
"total_cost_usd": round(self.total_cost, 4),
"hourly_tokens": self.hourly_tokens,
"hourly_cost_usd": round(self.hourly_cost, 4),
"daily_tokens": self.daily_tokens,
"daily_cost_usd": round(self.daily_cost, 4),
"buffer_size": len(self.buffer)
}
def reset_hourly_metrics(self) -> None:
"""Réinitialise les métriques horaires (à appeler chaque heure)"""
with self.metrics_lock:
self.hourly_tokens = 0
self.hourly_cost = 0.0
def reset_daily_metrics(self) -> None:
"""Réinitialise les métriques quotidiennes (à appeler chaque jour)"""
with self.metrics_lock:
self.daily_tokens = 0
self.daily_cost = 0.0
Instance globale du logger
audit_logger = AuditLogger(
log_path="/var/log/ai-audit/requests.jsonl",
buffer_size=100,
flush_interval=5
)
Intégration avec l'API HolySheep
Maintenant, créons un client enrichi qui intercepte automatiquement chaque appel et le journalise. Le base_url utilisé sera https://api.holysheep.ai/v1 avec la tarification 2026 avantageuse : DeepSeek V3.2 à $0.42/MToken contre $15 pour Claude Sonnet 4.5.
# holySheep_client.py
import requests
import time
import uuid
import hashlib
from typing import Dict, Any, Optional, List
from datetime import datetime
from .audit_logger import AuditLogger, AuditLogEntry, audit_logger
from .config import HolySheepConfig
Tarification HolySheep 2026 (USD par million de tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"prompt": 2.00, "completion": 6.00}, # $8/MTok total
"claude-sonnet-4.5": {"prompt": 3.00, "completion": 12.00}, # $15/MTok
"gemini-2.5-flash": {"prompt": 0.35, "completion": 2.15}, # $2.50/MTok
"deepseek-v3.2": {"prompt": 0.10, "completion": 0.32}, # $0.42/MTok
"gpt-4o": {"prompt": 2.50, "completion": 10.00},
"claude-3.5-sonnet": {"prompt": 3.00, "completion": 15.00},
}
def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Calcule le coût en USD selon le modèle utilisé"""
if model not in HOLYSHEEP_PRICING:
# Modèle inconnu, utiliser le prix moyen
return (prompt_tokens + completion_tokens) * 5 / 1_000_000
pricing = HOLYSHEEP_PRICING[model]
prompt_cost = prompt_tokens * pricing["prompt"] / 1_000_000
completion_cost = completion_tokens * pricing["completion"] / 1_000_000
return round(prompt_cost + completion_cost, 6)
class HolySheepClient:
"""
Client HTTP enrichi pour l'API HolySheep avec audit complet.
Inclut retry automatique, circuit breaker et logging.
"""
def __init__(self, config: HolySheepConfig = None, audit_logger: AuditLogger = None):
self.config = config or HolySheepConfig()
self.audit = audit_logger or audit_logger
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
})
def _generate_request_id(self) -> str:
"""Génère un ID unique pour chaque requête"""
timestamp = datetime.utcnow().isoformat()
unique_str = f"{timestamp}-{uuid.uuid4()}"
return hashlib.sha256(unique_str.encode()).hexdigest()[:16]
def _make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict] = None,
model: Optional[str] = None
) -> Dict[str, Any]:
"""Effectue une requête avec audit et gestion d'erreurs"""
request_id = self._generate_request_id()
start_time = time.time()
url = f"{self.config.base_url}/{endpoint.lstrip('/')}"
try:
response = self.session.request(
method=method,
url=url,
json=data,
timeout=self.config.timeout
)
latency_ms = (time.time() - start_time) * 1000
response_data = response.json() if response.content else {}
# Extraction des métriques d'usage
usage = response_data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", prompt_tokens + completion_tokens)
# Déduction du modèle utilisé
if not model:
model = data.get("model", "unknown") if data else "unknown"
# Calcul du coût
cost_usd = calculate_cost(model, prompt_tokens, completion_tokens)
# Logging de l'appel
if self.config.audit_enabled:
entry = AuditLogEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat(),
endpoint=endpoint,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost_usd=cost_usd,
latency_ms=round(latency_ms, 2),
status_code=response.status_code,
error_message=response_data.get("error", {}).get("message") if response.status_code >= 400 else None
)
self.audit.log(entry)
# Levée d'exception pour les erreurs HTTP
response.raise_for_status()
return {
"success": True,
"request_id": request_id,
"latency_ms": round(latency_ms, 2),
"cost_usd": cost_usd,
"data": response_data
}
except requests.exceptions.RequestException as e:
latency_ms = (time.time() - start_time) * 1000
if self.config.audit_enabled:
entry = AuditLogEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat(),
endpoint=endpoint,
model=model or "unknown",
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
cost_usd=0.0,
latency_ms=round(latency_ms, 2),
status_code=0,
error_message=str(e)
)
self.audit.log(entry)
return {
"success": False,
"request_id": request_id,
"latency_ms": round(latency_ms, 2),
"error": str(e)
}
def chat_completions(self, messages: List[Dict], model: str = "deepseek-v3.2", **kwargs) -> Dict[str, Any]:
"""
Endpoint /chat/completions avec audit automatique.
Par défaut, utilise DeepSeek V3.2 à $0.42/MTok (vs $15 pour Claude Sonnet 4.5).
"""
data = {
"model": model,
"messages": messages,
**kwargs
}
return self._make_request("POST", "/chat/completions", data, model)
def embeddings(self, input_text: str, model: str = "text-embedding-3-small", **kwargs) -> Dict[str, Any]:
"""Endpoint /embeddings avec audit automatique"""
data = {
"model": model,
"input": input_text,
**kwargs
}
return self._make_request("POST", "/embeddings", data, model)
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques d'audit courantes"""
return self.audit.get_metrics()
Instance globale prête à l'emploi
client = HolySheepClient()
Moteur de Détection d'Anomalies
Le cœur de notre solution : un moteur de détection qui analyse les patterns de consommation en temps réel et identifie les comportements anormaux avant qu'ils n'impactent la facture.
# anomaly_detector.py
import threading
import time
from datetime import datetime, timedelta
from typing import Dict, Any, List, Callable, Optional
from dataclasses import dataclass
from collections import deque
from .audit_logger import AuditLogger
from .config import HolySheepConfig
@dataclass
class AnomalyAlert:
"""Structure d'une alerte d'anomalie"""
alert_id: str
timestamp: str
alert_type: str # 'cost', 'tokens', 'requests', 'latency'
severity: str # 'low', 'medium', 'high', 'critical'
message: str
metrics: Dict[str, Any]
recommended_action: str
class AnomalyDetector:
"""
Moteur de détection d'anomalies en temps réel.
Analyse les patterns de consommation et détecte les comportements anormaux.
"""
def __init__(
self,
audit_logger: AuditLogger,
config: HolySheepConfig,
alert_callback: Optional[Callable[[AnomalyAlert], None]] = None
):
self.audit = audit_logger
self.config = config
self.alert_callback = alert_callback
# Fenêtres glissantes pour l'analyse (en secondes)
self.window_1min = deque(maxlen=60) # 1 minute de données
self.window_5min = deque(maxlen=300) # 5 minutes
self.window_1hour = deque(maxlen=3600) # 1 heure
# Historique pour comparaison
self.history_1hour = deque(maxlen=24) # 24 dernières heures
self.history_1day = deque(maxlen=30) # 30 derniers jours
# Compteurs agrégés
self.last_hour_tokens = 0
self.last_hour_cost = 0.0
self.last_hour_requests = 0
self.alerts: List[AnomalyAlert] = []
self.alert_lock = threading.Lock()
# Statistiques pour détection d'anomalies
self.running_stats = {
"avg_tokens_per_minute": 0,
"avg_cost_per_minute": 0,
"avg_requests_per_minute": 0,
"stddev_tokens": 0,
"stddev_cost": 0,
"peak_tokens_1min": 0,
"peak_cost_1min": 0
}
self._running = False
self._lock = threading.Lock()
def record_event(self, entry: Dict[str, Any]) -> None:
"""Enregistre un événement pour l'analyse"""
timestamp = time.time()
event = {
"timestamp": timestamp,
"tokens": entry.get("total_tokens", 0),
"cost": entry.get("cost_usd", 0),
"model": entry.get("model", "unknown"),
"latency_ms": entry.get("latency_ms", 0),
"status": entry.get("status_code", 0)
}
with self._lock:
self.window_1min.append(event)
self.window_5min.append(event)
self.window_1hour.append(event)
# Mise à jour des compteurs
self.last_hour_tokens += event["tokens"]
self.last_hour_cost += event["cost"]
self.last_hour_requests += 1
def _calculate_statistics(self) -> Dict[str, Any]:
"""Calcule les statistiques sur la fenêtre actuelle"""
if not self.window_1min:
return self.running_stats.copy()
tokens_list = [e["tokens"] for e in self.window_1min]
cost_list = [e["cost"] for e in self.window_1min]
avg_tokens = sum(tokens_list) / len(tokens_list) if tokens_list else 0
avg_cost = sum(cost_list) / len(cost_list) if cost_list else 0
# Calcul de l'écart-type
if len(tokens_list) > 1:
variance_tokens = sum((x - avg_tokens) ** 2 for x in tokens_list) / len(tokens_list)
stddev_tokens = variance_tokens ** 0.5
else:
stddev_tokens = 0
return {
"avg_tokens_per_minute": round(avg_tokens * 60, 2),
"avg_cost_per_minute": round(avg_cost * 60, 4),
"avg_requests_per_minute": round(len(tokens_list), 2),
"stddev_tokens": round(stddev_tokens, 2),
"stddev_cost": round(sum((c - avg_cost) ** 2 for c in cost_list) / max(len(cost_list), 1) ** 0.5, 4),
"peak_tokens_1min": max(tokens_list) if tokens_list else 0,
"peak_cost_1min": max(cost_list) if cost_list else 0
}
def _detect_cost_anomaly(self) -> Optional[AnomalyAlert]:
"""Détecte les anomalies de coût"""
current_cost = self.last_hour_cost
threshold = self.config.hourly_budget_threshold
if current_cost > threshold:
severity = "critical" if current_cost > threshold * 2 else "high"
return AnomalyAlert(
alert_id=f"cost-{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
alert_type="cost",
severity=severity,
message=f"Dépense horaire excessive : ${current_cost:.2f} (seuil: ${threshold})",
metrics={
"current_cost": round(current_cost, 2),
"threshold": threshold,
"ratio": round(current_cost / threshold, 2)
},
recommended_action="Arrêter immédiatement les requêtes et investigar"
)
return None
def _detect_token_burst(self) -> Optional[AnomalyAlert]:
"""Détecte les pics anormaux de tokens"""
if not self.window_1min:
return None
current_tokens = sum(e["tokens"] for e in self.window_1min)
threshold = self.config.abnormal_token_threshold
if current_tokens > threshold:
# Comparaison avec la moyenne historique
stats = self._calculate_statistics()
if stats["avg_tokens_per_minute"] > 0:
ratio = current_tokens / stats["avg_tokens_per_minute"]
if ratio > 5: # 5x la moyenne = anomalie
return AnomalyAlert(
alert_id=f"token-burst-{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
alert_type="tokens",
severity="high",
message=f"Pic de tokens détecté : {current_tokens} tokens/min (moyenne: {stats['avg_tokens_per_minute']})",
metrics={
"current_tokens": current_tokens,
"average_tokens": stats["avg_tokens_per_minute"],
"ratio": round(ratio, 2),
"threshold": threshold
},
recommended_action="Vérifier les boucles infinies et les prompts mal configurés"
)
return None
def _detect_request_flood(self) -> Optional[AnomalyAlert]:
"""Détecte les floods de requêtes (possible DoS ou bug)"""
current_requests = len(self.window_1min)
threshold = self.config.abnormal_request_threshold
if current_requests > threshold:
return AnomalyAlert(
alert_id=f"request-flood-{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
alert_type="requests",
severity="critical",
message=f"Flood de requêtes : {current_requests} req/min (seuil: {threshold})",
metrics={
"current_requests": current_requests,
"threshold": threshold
},
recommended_action="Bloquer immédiatement les nouvelles requêtes"
)
return None
def _detect_latency_anomaly(self) -> Optional[AnomalyAlert]:
"""Détecte les dégradations de latence"""
if not self.window_1min:
return None
latencies = [e["latency_ms"] for e in self.window_1min if e["latency_ms"] > 0]
if not latencies:
return None
avg_latency = sum(latencies) / len(latencies)
# HolySheep garantit <50ms, alerte si >200ms
if avg_latency > 200:
return AnomalyAlert(
alert_id=f"latency-{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
alert_type="latency",
severity="medium",
message=f"Latence anormale : {avg_latency:.0f}ms (objectif: <50ms)",
metrics={
"avg_latency_ms": round(avg_latency, 2),
"max_latency_ms": max(latencies),
"min_latency_ms": min(latencies)
},
recommended_action="Vérifier la connectivité réseau et la charge serveur"
)
return None
def _detect_model_mismatch(self) -> Optional[AnomalyAlert]:
"""Détecte les modèles utilisés par erreur (trop chers)"""
if not self.window_1min:
return None
expensive_models = {
"claude-sonnet-4.5": 15.0,
"claude-3.5-sonnet": 15.0,
"gpt-4.1": 8.0,
"gpt-4o": 12.5
}
for event in self.window_1min:
model = event.get("model", "")
if model in expensive_models and event["tokens"] > 10000:
return AnomalyAlert(
alert_id=f"model-{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
alert_type="cost",
severity="low",
message=f"Utilisation d'un modèle coûteux ({model})",
metrics={
"model": model,
"tokens": event["tokens"],
"cost": event["cost"],
"suggestion": "Envisager DeepSeek V3.2 à $0.42/MTok"
},
recommended_action=f"Migration vers DeepSeek V3.2 pour réduire les coûts de {expensive_models[model]/0.42:.0f}x"
)
return None
def analyze(self) -> List[AnomalyAlert]:
"""Lance une analyse complète et retourne les alertes"""
alerts = []
# Détection de toutes les anomalies
detection_methods = [
self._detect_cost_anomaly,
self._detect_token_burst,
self._detect_request_flood,
self._detect_latency_anomaly,
self._detect_model_mismatch
]
for method in detection_methods:
alert = method()
if alert:
alerts.append(alert)
with self.alert_lock:
self.alerts.append(alert)
if self.alert_callback:
self.alert_callback(alert)
# Mise à jour des statistiques
self.running_stats = self._calculate_statistics()
return alerts
def reset_hourly_counters(self) -> None:
"""Réinitialise les compteurs horaires"""
with self._lock:
self.last_hour_tokens = 0
self.last_hour_cost = 0.0
self.last_hour_requests = 0
self.window_1min.clear()
self.window_5min.clear()
def get_status(self) -> Dict[str, Any]:
"""Retourne le statut complet du détecteur"""
with self._lock:
metrics = self.audit.get_metrics()
return {
"running": self._running,
"current_metrics": metrics,
"running_stats": self.running_stats,
"alerts_count": len(self.alerts),
"recent_alerts": [
{
"id": a.alert_id,
"type": a.alert_type,
"severity": a.severity,
"message": a.message
}
for a in self.alerts[-5:]
]
}
Système d'Alertes Multi-Canal
Un système d'alertes robuste garantit que les anomalies sont communiquées rapidement aux bonnes personnes via les canaux appropriés.
# alert_dispatcher.py
import json
import requests
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
import logging
from .anomaly_detector import AnomalyAlert
logger = logging.getLogger("alert_dispatcher")
@dataclass
class AlertChannel:
"""Configuration d'un canal d'alerte"""
name: str
enabled: bool = True
min_severity: str = "low" # Niveau minimum pour déclencher
# Configuration Slack
slack_webhook: Optional[str] = None
# Configuration Email
smtp_host: Optional[str] = None
smtp_port: Optional[int] = 587
smtp_user: Optional[str] = None
smtp_password: Optional[str] = None
email_from: Optional[str] = None
email_to: Optional[List[str]] = None
# Configuration Webhook générique
webhook_url: Optional[str] = None
webhook_headers: Optional[Dict[str, str]] = None
class AlertDispatcher:
"""
Dispatcher multi-canal pour les alertes d'anomalies.
Gère Slack, Email, Webhooks et SMS.
"""
SEVERITY_ORDER = {"low": 0, "medium": 1, "high": 2, "critical": 3}
def __init__(self, channels: List[AlertChannel]):
self.channels = channels
self.alert_history: List[Dict[str, Any]] = []
def _should_notify(self, channel: AlertChannel, alert: AnomalyAlert) -> bool:
"""Détermine si une alerte doit être envoyée sur un canal"""
if not channel.enabled:
return False
alert_level = self.SEVERITY_ORDER.get(alert.severity, 0)
channel_level = self.SEVERITY_ORDER.get(channel.min_severity, 0)
return alert_level >= channel_level
def _format_slack_message(self, alert: AnomalyAlert) -> Dict[str, Any]:
"""Formate le message pour Slack"""
emoji = {
"low": "ℹ️",
"medium": "⚠️",
"high": "🔶",
"critical": "🚨"
}.get(alert.severity, "❓")
color = {
"low": "#36a64f",
"medium": "#ff9900",
"high": "#ff6600",
"critical": "#ff0000"
}.get(alert.severity, "#808080")
fields = []
for key, value in alert.metrics.items():
fields.append({
"title": key.replace("_", " ").title(),
"value": str(value),
"short": True
})
return {
"attachments": [{
"color": color,
"title": f"{emoji} {alert.alert_type.upper()} - {alert.severity.upper()}",
"text": alert.message,
"fields": fields,
"footer": f"Request ID: {alert.alert_id}",
"ts": alert.timestamp
}]
}
def _send_slack(self, channel: AlertChannel, alert: AnomalyAlert) -> bool:
"""Envoie l'alerte sur Slack"""
if not channel.slack_webhook:
return False
try:
payload = self._format_slack_message(alert)
response = requests.post(
channel.slack_webhook,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
return response.status_code == 200
except requests.exceptions.RequestException as e:
logger.error(f"Erreur d'envoi Slack: {e}")
return False
def _send_email(self, channel: AlertChannel, alert: AnomalyAlert) -> bool:
"""Envoie l'alerte par email"""
if not all([channel.smtp_host, channel.email_from, channel.email_to]):
return False
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[{alert.severity.upper()}] Alerte IA : {alert.message}"
msg["From"] = channel.email_from
msg["To"] = ", ".join(channel.email_to)
html_content = f"""
🚨 Alerte {alert.severity.upper()} - {alert.alert_type}
Message: {alert.message}
Action recommandée: {alert.recommended_action}
Métriques:
Métrique Valeur {key} {value}
Request ID: {}
""".format(alert.alert_id)
msg.attach(MIMEText(html_content, "html"))
with smtplib.SMTP(channel.smtp_host, channel.smtp_port) as server:
server.starttls()
server.login(channel.smtp_user, channel.smtp_password)
server.send_message(msg)
return True
except Exception as e:
logger.error(f"Erreur d'envoi email: {e}")
return False
def _send_webhook(self, channel: AlertChannel, alert: AnomalyAlert) -> bool:
"""Envoie l'alerte sur un webhook générique"""
if not channel.webhook_url:
return False
try:
payload = {
"alert_id": alert.alert_id,
"timestamp": alert.timestamp,