Il est 3h47 du matin quand mon téléphone vibre violemment. Je bondis du lit, les yeux encore embués de sommeil, pour découvrir un cauchemar financier : ma console d'administration affiche une consommation de 2 847 $ en seulement 6 heures. L'erreur était simple mais dévastatrice : une boucle infinie dans mon pipeline de traitement envoyait des requêtes continues à l'API sans le moindre mécanisme de limitation ou de surveillance. Ce n'était pas un hack, ni une fuite de credentials — c'était simplement l'absence d'un système d'alerte de、成本控制 adequado.
Cette expérience m'a coûté cher, mais elle m'a appris une leçon inestimable sur la conception de systèmes robustes pour les API de grands modèles de langage. Aujourd'hui, je partage avec vous l'architecture complète que j'ai développée pour éviter ce genre de catastrophe et optimiser mes dépenses Cloud.
Comprendre les Risques Financiers des API LLM
Les API de grands modèles de langage sont facturées au token — chaque requête génère des coûts qui s'accumulent silencieusement. Avec des prix pouvant atteindre 15 $ par million de tokens pour les modèles les plus performants comme Claude Sonnet 4.5, une simple erreur de programmation peut se transformer en facture de plusieurs milliers de dollars en quelques heures.
HolySheep AI offre des tarifs avantageux avec un taux de change de ¥1 pour 1 $, représentant une économie de plus de 85% par rapport aux providers traditionnels. Leurs modèles incluent DeepSeek V3.2 à seulement 0,42 $ par million de tokens, Gemini 2.5 Flash à 2,50 $, et pour les besoins avancés, GPT-4.1 à 8 $ et Claude Sonnet 4.5 à 15 $. La latence moyenne reste inférieure à 50 millisecondes, et le support pour WeChat et Alipay facilite les paiements pour les utilisateurs internationaux.
Architecture du Système d'Alerte
Le système que je vais vous présenter s'articule autour de quatre composants principaux : un proxy API intelligent, un module de surveillance en temps réel, un système d'alerte multicanal, et une couche de limitation automatique. Commençons par l'implémentation complète.
Implémentation du Proxy API avec Monitoring
#!/usr/bin/env python3
"""
Système d'Alerte de Surconsommation API LLM
Auteur: HolySheep AI Technical Blog
Version: 2.0
"""
import time
import json
import asyncio
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Dict, List, Callable
from collections import defaultdict
from threading import Lock
import hashlib
Configuration HolySheep API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class TokenUsage:
"""Suivi détaillé de l'utilisation des tokens"""
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost: float
timestamp: datetime = field(default_factory=datetime.now)
request_id: str = ""
@dataclass
class AlertThresholds:
"""Configuration des seuils d'alerte"""
hourly_cost: float = 50.0 # Alerte si > 50$/heure
daily_cost: float = 200.0 # Alerte si > 200$/jour
request_rate: int = 100 # Alerte si > 100 req/min
burst_requests: int = 20 # Alerte si > 20 req/seconde
error_rate: float = 0.1 # Alerte si > 10% d'erreurs
class CostTracker:
"""Tracker de coûts avec alertes intelligentes"""
# Prix par million de tokens (2026)
MODEL_PRICES = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
"holy-deepseek": 0.42,
"holy-gpt4": 8.0,
"holy-claude": 15.0
}
def __init__(self, alert_callback: Optional[Callable] = None):
self.usage_history: List[TokenUsage] = []
self.lock = Lock()
self.alert_callback = alert_callback
self.daily_cost = 0.0
self.hourly_cost = 0.0
self.alert_cooldown = 300 # 5 minutes entre alertes
# Compteurs par modèle
self.model_costs: Dict[str, float] = defaultdict(float)
self.model_requests: Dict[str, int] = defaultdict(int)
# Surveillance du taux de requêtes
self.request_timestamps: List[float] = []
self.last_alert_time = 0
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def calculate_cost(self, model: str, prompt_tokens: int,
completion_tokens: int) -> float:
"""Calcule le coût exact basé sur le modèle"""
price_per_million = self.MODEL_PRICES.get(model, 1.0)
total_tokens = prompt_tokens + completion_tokens
cost = (total_tokens / 1_000_000) * price_per_million
return round(cost, 6)
def record_usage(self, model: str, prompt_tokens: int,
completion_tokens: int, request_id: str = "") -> TokenUsage:
"""Enregistre l'utilisation et vérifie les seuils"""
cost = self.calculate_cost(model, prompt_tokens, completion_tokens)
usage = TokenUsage(
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
cost=cost,
request_id=request_id or self._generate_request_id()
)
with self.lock:
self.usage_history.append(usage)
self._update_costs()
self._update_rate_tracking()
# Vérification des alertes
self._check_thresholds()
self.logger.info(
f"Usage: {model} | Tokens: {usage.total_tokens} | "
f"Cost: ${cost:.6f} | Cumulative: ${self.daily_cost:.2f}"
)
return usage
def _generate_request_id(self) -> str:
"""Génère un ID unique pour traçabilité"""
timestamp = str(time.time()).encode()
return hashlib.md5(timestamp).hexdigest()[:12]
def _update_costs(self):
"""Met à jour les compteurs de coûts"""
now = datetime.now()
cutoff_hour = now - timedelta(hours=1)
cutoff_day = now - timedelta(days=1)
# Calcul du coût horaire
hourly_usages = [u for u in self.usage_history if u.timestamp > cutoff_hour]
self.hourly_cost = sum(u.cost for u in hourly_usages)
# Calcul du coût journalier
daily_usages = [u for u in self.usage_history if u.timestamp > cutoff_day]
self.daily_cost = sum(u.cost for u in daily_usages)
# Nettoyage des vieux enregistrements
self.usage_history = daily_usages.copy()
def _update_rate_tracking(self):
"""Surveille le taux de requêtes"""
now = time.time()
self.request_timestamps = [
t for t in self.request_timestamps if now - t < 60
]
self.request_timestamps.append(now)
def _check_thresholds(self):
"""Vérifie tous les seuils d'alerte"""
now = time.time()
if now - self.last_alert_time < self.alert_cooldown:
return
alerts_triggered = []
if self.hourly_cost > AlertThresholds.hourly_cost:
alerts_triggered.append(
f"🚨 ALERTE: Coût horaire ${self.hourly_cost:.2f} "
f"(seuil: ${AlertThresholds.hourly_cost})"
)
if self.daily_cost > AlertThresholds.daily_cost:
alerts_triggered.append(
f"🚨 ALERTE: Coût journalier ${self.daily_cost:.2f} "
f"(seuil: ${AlertThresholds.daily_cost})"
)
request_rate = len(self.request_timestamps)
if request_rate > AlertThresholds.request_rate:
alerts_triggered.append(
f"⚠️ ALERTE: Taux de requêtes {request_rate}/min "
f"(seuil: {AlertThresholds.request_rate})"
)
if alerts_triggered and self.alert_callback:
for alert in alerts_triggered:
self.logger.warning(alert)
self.alert_callback(alert)
self.last_alert_time = now
def get_cost_breakdown(self) -> Dict:
"""Retourne un rapport détaillé des coûts"""
with self.lock:
return {
"hourly_cost": round(self.hourly_cost, 4),
"daily_cost": round(self.daily_cost, 4),
"total_requests": len(self.usage_history),
"by_model": {
model: {
"cost": round(cost, 4),
"requests": self.model_requests[model]
}
for model, cost in self.model_costs.items()
},
"avg_cost_per_request": round(
self.daily_cost / len(self.usage_history)
if self.usage_history else 0, 6
)
}
def emergency_stop(self) -> bool:
"""Active le blocage d'urgence"""
self.logger.critical("🚨 ARRÊT D'URGENCE ACTIVÉ - Toutes les requêtes bloquées")
return True
def reset_limits(self):
"""Réinitialise les compteurs"""
with self.lock:
self.usage_history.clear()
self.model_costs.clear()
self.model_requests.clear()
self.hourly_cost = 0.0
self.daily_cost = 0.0
self.logger.info("Compteurs réinitialisés")
Démonstration
if __name__ == "__main__":
def handle_alert(message: str):
print(f"📱 ALERTE ENVOYÉE: {message}")
tracker = CostTracker(alert_callback=handle_alert)
# Simulation d'appels API
test_requests = [
("deepseek-v3.2", 1500, 300),
("gemini-2.5-flash", 2000, 450),
("deepseek-v3.2", 1800, 400),
("holy-gpt4", 2500, 600),
]
for model, prompt, completion in test_requests:
tracker.record_usage(model, prompt, completion)
print("\n📊 Rapport de coûts:")
print(json.dumps(tracker.get_cost_breakdown(), indent=2))Client API avec Rate Limiting Intelligent
#!/usr/bin/env python3
"""
Client API LLM avec Rate Limiting et Retry Intelligent
Compatible HolySheep AI
"""
import os
import time
import asyncio
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
import aiohttp
from abc import ABC, abstractmethod
@dataclass
class APIResponse:
"""Structure de réponse standardisée"""
success: bool
data: Optional[Dict] = None
error: Optional[str] = None
tokens_used: int = 0
cost: float = 0.0
latency_ms: float = 0.0
model: str = ""
class RateLimiter:
"""Rate limiter avec tokens bucket et burst control"""
def __init__(self, requests_per_minute: int = 60,
burst_limit: int = 10):
self.rpm = requests_per_minute
self.burst_limit = burst_limit
self.tokens = burst_limit
self.last_update = time.time()
self.retry_after: Optional[float] = None
self.request_times: List[float] = []
def acquire(self, blocking: bool = True) -> bool:
"""Acquiert un token, attend si nécessaire"""
now = time.time()
# Nettoyage des requêtes anciennes
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.rpm:
if not blocking:
return False
# Calcul du temps d'attente
oldest = min(self.request_times)
wait_time = 60 - (now - oldest)
if wait_time > 0:
time.sleep(wait_time)
self.request_times = [t for t in self.request_times if now - t < 60]
if self.tokens < 1:
if not blocking:
return False
# Attente pour un token
time.sleep(1.0 / self.rpm)
self.tokens = min(self.burst_limit, self.tokens + (time.time() - self.last_update) * (self.rpm / 60))
self.tokens -= 1
self.last_update = time.time()
self.request_times.append(now)
return True
def get_wait_time(self) -> float:
"""Retourne le temps d'attente estimé en secondes"""
now = time.time()
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) < self.rpm:
return 0.0
oldest = min(self.request_times)
return max(0.0, 60 - (now - oldest))
class LLMAPIClient:
"""Client complet pour HolySheep AI avec surveillance des coûts"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL,
rate_limiter: Optional[RateLimiter] = None,
cost_tracker: Optional[Any] = None,
max_retries: int = 3,
timeout: int = 60):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.rate_limiter = rate_limiter or RateLimiter()
self.cost_tracker = cost_tracker
self.max_retries = max_retries
self.timeout = timeout
self.emergency_stop = False
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO)
# Session aiohttp pour performances
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Obtient ou crée une session aiohttp"""
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=self.timeout)
self._session = aiohttp.ClientSession(timeout=timeout)
return self._session
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> APIResponse:
"""Appel principal avec gestion complète des erreurs"""
if self.emergency_stop:
return APIResponse(
success=False,
error="EMERGENCY_STOP: Toutes les requêtes sont bloquées"
)
# Acquisition du rate limit
if not self.rate_limiter.acquire(blocking=True):
return APIResponse(
success=False,
error="RATE_LIMIT: Impossible d'acquérir un token"
)
start_time = time.time()
last_error = None
for attempt in range(self.max_retries):
try:
session = await self._get_session()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
url = f"{self.base_url}/chat/completions"
async with session.post(url, json=payload,
headers=headers) as response:
latency = (time.time() - start_time) * 1000
if response.status == 200:
data = await response.json()
# Enregistrement du coût
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
if self.cost_tracker:
usage_record = self.cost_tracker.record_usage(
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens
)
cost = usage_record.cost
else:
cost = 0.0
self.logger.info(
f"✅ {model} | Latence: {latency:.0f}ms | "
f"Tokens: {prompt_tokens + completion_tokens} | "
f"Coût: ${cost:.6f}"
)
return APIResponse(
success=True,
data=data,
tokens_used=prompt_tokens + completion_tokens,
cost=cost,
latency_ms=latency,
model=model
)
elif response.status == 401:
return APIResponse(
success=False,
error="AUTH_ERROR: Clé API invalide ou expirée",
latency_ms=latency
)
elif response.status == 429:
retry_after = response.headers.get('Retry-After', '5')
wait_time = float(retry_after) * (2 ** attempt)
self.logger.warning(
f"Rate limit atteint, attente {wait_time:.1f}s"
)
await asyncio.sleep(min(wait_time, 30))
continue
elif response.status == 500:
last_error = f"SERVER_ERROR: {response.status}"
await asyncio.sleep(2 ** attempt)
continue
else:
error_text = await response.text()
last_error = f"HTTP_{response.status}: {error_text}"
break
except asyncio.TimeoutError:
last_error = f"TIMEOUT: Délai dépassé ({self.timeout}s)"
self.logger.error(f"Timeout lors de l'appel API (tentative {attempt + 1})")
except aiohttp.ClientError as e:
last_error = f"CONNECTION_ERROR: {str(e)}"
self.logger.error(f"Erreur de connexion: {e}")
except Exception as e:
last_error = f"UNEXPECTED_ERROR: {str(e)}"
self.logger.exception("Erreur inattendue")
break
return APIResponse(
success=False,
error=last_error or "Erreur inconnue",
latency_ms=(time.time() - start_time) * 1000
)
async def batch_completion(
self,
requests: List[Dict[str, Any]],
concurrency: int = 5
) -> List[APIResponse]:
"""Traitement par lots avec limitation de concurrence"""
semaphore = asyncio.Semaphore(concurrency)
async def process_single(req: Dict) -> APIResponse:
async with semaphore:
return await self.chat_completion(**req)
tasks = [process_single(req) for req in requests]
return await asyncio.gather(*tasks)
async def close(self):
"""Fermeture propre de la session"""
if self._session and not self._session.closed:
await self._session.close()
def enable_emergency_stop(self):
"""Active l'arrêt d'urgence"""
self.emergency_stop = True
self.logger.critical("🛑 Arrêt d'urgence activé")
def disable_emergency_stop(self):
"""Désactive l'arrêt d'urgence"""
self.emergency_stop = False
self.logger.info("✅ Arrêt d'urgence désactivé")
Exemple d'utilisation avec HolySheep
async def example_usage():
"""Exemple complet d'utilisation du client"""
api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
def alert_handler(message: str):
print(f"📱 NOTIFICATION: {message}")
cost_tracker = CostTracker(alert_callback=alert_handler)
rate_limiter = RateLimiter(requests_per_minute=30, burst_limit=5)
client = LLMAPIClient(
api_key=api_key,
rate_limiter=rate_limiter,
cost_tracker=cost_tracker
)
try:
messages = [
{"role": "system", "content": "Tu es un assistant expert en optimisation de coûts."},
{"role": "user", "content": "Explique comment réduire les coûts d'API LLM."}
]
response = await client.chat_completion(
model="deepseek-v3.2",
messages=messages,
temperature=0.7,
max_tokens=1000
)
if response.success:
print(f"✅ Réponse reçue en {response.latency_ms:.0f}ms")
print(f"💰 Coût: ${response.cost:.6f}")
print(f"📊 Contenu: {response.data['choices'][0]['message']['content'][:200]}...")
else:
print(f"❌ Erreur: {response.error}")
finally:
await client.close()
print("\n📊 Coûts totaux:", json.dumps(cost_tracker.get_cost_breakdown(), indent=2))
if __name__ == "__main__":
asyncio.run(example_usage())Système d'Alerte Multi-Canal
#!/usr/bin/env python3
"""
Système d'Alerte Multi-Canal avec Actions Automatisées
Support: Email, Webhook, SMS, Telegram, WeChat
"""
import os
import smtplib
import asyncio
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime
from abc import ABC, abstractmethod
import json
import logging
from enum import Enum
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
@dataclass
class Alert:
"""Structure d'une alerte"""
severity: AlertSeverity
title: str
message: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
cost_impact: float = 0.0
resolved: bool = False
class AlertChannel(ABC):
"""Classe de base pour les canaux d'alerte"""
@abstractmethod
async def send(self, alert: Alert) -> bool:
"""Envoie l'alerte"""
pass
@abstractmethod
def is_configured(self) -> bool:
"""Vérifie si le canal est configuré"""
pass
class EmailAlertChannel(AlertChannel):
"""Alertes par email avec templates HTML"""
def __init__(self, smtp_host: str, smtp_port: int,
username: str, password: str,
from_addr: str, to_addrs: List[str]):
self.smtp_host = smtp_host
self.smtp_port = smtp_port
self.username = username
self.password = password
self.from_addr = from_addr
self.to_addrs = to_addrs
def is_configured(self) -> bool:
return bool(self.smtp_host and self.username and self.password)
def _get_html_template(self, alert: Alert) -> str:
"""Template HTML pour l'email"""
severity_colors = {
AlertSeverity.INFO: "#17a2b8",
AlertSeverity.WARNING: "#ffc107",
AlertSeverity.ERROR: "#dc3545",
AlertSeverity.CRITICAL: "#8b0000"
}
color = severity_colors.get(alert.severity, "#6c757d")
return f"""
🚨 {alert.title}
Severité: {alert.severity.value.upper()}
Heure: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
{alert.message}
{f'💰 Impact financier: ${alert.cost_impact:.2f}' if alert.cost_impact > 0 else ''}
{f'{json.dumps(alert.metadata, indent=2)}' if alert.metadata else ''}
"""
async def send(self, alert: Alert) -> bool:
if not self.is_configured():
return False
try:
msg = MIMEMultipart('alternative')
msg['Subject'] = f"[{alert.severity.value.upper()}] {alert.title}"
msg['From'] = self.from_addr
msg['To'] = ', '.join(self.to_addrs)
html_content = self._get_html_template(alert)
msg.attach(MIMEText(html_content, 'html'))
# Envoi synchrone dans un executor
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, self._send_sync, msg)
logging.info(f"Email envoyé à {self.to_addrs}")
return True
except Exception as e:
logging.error(f"Erreur envoi email: {e}")
return False
def _send_sync(self, msg):
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
class WebhookAlertChannel(AlertChannel):
"""Alertes via webhook HTTP (Slack, Discord, custom)"""
def __init__(self, url: str, headers: Optional[Dict] = None,
timeout: int = 10):
self.url = url
self.headers = headers or {"Content-Type": "application/json"}
self.timeout = timeout
def is_configured(self) -> bool:
return bool(self.url)
def _format_slack_message(self, alert: Alert) -> Dict:
"""Format Slack Block Kit"""
severity_emoji = {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.ERROR: "❌",
AlertSeverity.CRITICAL: "🚨"
}
emoji = severity_emoji.get(alert.severity, "📢")
blocks = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji} {alert.title}"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Severity:*\n{alert.severity.value}"},
{"type": "mrkdwn", "text": f"*Time:*\n{alert.timestamp.isoformat()}"}
]
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": alert.message
}
}
]
if alert.cost_impact > 0:
blocks.append({
"type": "section",
"fields": [{
"type": "mrkdwn",
"text": f"*💰 Cost Impact:*\n${alert.cost_impact:.2f}"
}]
})
return {"blocks": blocks}
async def send(self, alert: Alert) -> bool:
if not self.is_configured():
return False
try:
import aiohttp
payload = self._format_slack_message(alert)
async with aiohttp.ClientSession() as session:
async with session.post(
self.url,
json=payload,
headers=self.headers,
timeout=aiohttp.ClientTimeout(total=self.timeout)
) as response:
if response.status < 300:
logging.info(f"Webhook envoyé: {self.url}")
return True
else:
logging.error(f"Webhook error: {response.status}")
return False
except Exception as e:
logging.error(f"Erreur webhook: {e}")
return False
class TelegramAlertChannel(AlertChannel):
"""Alertes via Telegram Bot"""
def __init__(self, bot_token: str, chat_id: str):
self.bot_token = bot_token
self.chat_id = chat_id
self.api_url = f"https://api.telegram.org/bot{bot_token}"
def is_configured(self) -> bool:
return bool(self.bot_token and self.chat_id)
async def send(self, alert: Alert) -> bool:
if not self.is_configured():
return False
try:
import aiohttp
emoji = "🔵" if alert.severity == AlertSeverity.INFO else \
"🟡" if alert.severity == AlertSeverity.WARNING else \
"🔴" if alert.severity == AlertSeverity.ERROR else \
"🚨"
message = f"""
{emoji} {alert.title}
Severity: {alert.severity.value.upper()}
Time: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')}
{alert.message}
{f"💰 Cost Impact: ${alert.cost_impact:.2f}" if alert.cost_impact > 0 else ""}
"""
url = f"{self.api_url}/sendMessage"
payload = {
"chat_id": self.chat_id,
"text": message,
"parse_mode": "HTML"
}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as response:
return response.status == 200
except Exception as e:
logging.error(f"Erreur Telegram: {e}")
return False
class AlertManager:
"""Gestionnaire centralisé des alertes"""
def __init__(self):
self.channels: List[AlertChannel] = []
self.alert_history: List[Alert] = []
self.auto_actions: Dict[AlertSeverity, List[Callable]] = {
severity: [] for severity in AlertSeverity
}
self.logger = logging.getLogger(__name__)
def add_channel(self, channel: AlertChannel):
"""Ajoute un canal d'alerte"""
if channel.is_configured():
self.channels.append(channel)
self.logger.info(f"Canal ajouté: {channel.__class__.__name__}")
def register_auto_action(self, severity: AlertSeverity,
action: Callable[[Alert], None]):
"""Enregistre une action automatique pour une sévérité"""
self.auto_actions[severity].append(action)
async def send_alert(self, severity: AlertSeverity, title: str,
message: str, metadata: Optional[Dict] = None,
cost_impact: float = 0.0):
"""Envoie une alerte sur tous les canaux"""
alert = Alert(
severity=severity,
title=title,
message=message,
metadata=metadata or {},
cost_impact=cost_impact
)
self.alert_history.append(alert)
# Exécution des actions automatiques
for action in self.auto_actions.get(severity, []):
try:
action(alert)
except Exception as e:
self.logger.error(f"Erreur action auto: {e}")
# Envoi sur tous les canaux
results = await asyncio.gather(
*[channel.send(alert) for channel in self.channels],
return_exceptions=True
)
success_count = sum(1 for r in results if r is True)
self.logger.info(
f"Alerte envoyée: {title} ({success_count}/{len(self.channels)} canaux)"
)
return alert
def get_alert_summary(self, hours: int = 24) -> Dict:
"""Résumé des alertes des dernières heures"""
cutoff = datetime.now() - timedelta(hours=hours)
recent = [a for a in self.alert_history if a.timestamp > cutoff]
return {
"total": len(recent),
"by_severity": {
s.value: len([a for a in recent if a.severity == s])
for s in AlertSeverity
},
"total_cost_impact": sum(a.cost_impact for a in recent)
}
Configuration exemple
async def setup_alert_system():
"""Configuration complète du système d'alertes"""
manager = AlertManager()
# Email
email_channel = EmailAlertChannel(
smtp_host=os.environ.get("SMTP_HOST", "smtp.gmail.com"),
smtp_port=int(os.environ.get("SMTP_PORT", "587")),
username=os.environ.get("SMTP_USER"),
password=os.environ.get("SMTP_PASS"),
from_addr="[email protected]",
to_addrs=["[email protected]", "[email protected]"]
)
manager.add_channel(email_channel)
# Slack/Discord Webhook
webhook_channel = WebhookAlertChannel(
url=os.environ.get("SLACK_WEBHOOK_URL", ""),
headers={"Content-Type": "application/json"}
)
manager.add_channel(webhook_channel)
# Telegram
telegram_channel = TelegramAlertChannel(
bot_token=os.environ.get("TELEGRAM_BOT_TOKEN", ""),
chat_id=os.environ.get("