En tant qu'architecte infrastructure qui a supervisé la migration de trois plateformes de production vers des API d'IA générative, je mesure chaque semaine l'impact financier des appels non gouvernés. En 2026, avec des modèles comme Claude Sonnet 4.5 facturé à $15 par million de tokens, une simple boucle infinie peut engendrer des milliers de dollars en quelques heures. Cet article détaille mon framework complet de cost governance pour l'API HolySheep AI, incluant le découpage granularisé des dépenses, les alertes budgetaires et l'automatisation complète des rapports mensuels.
Architecture de Tracking Multi-Dimensionnel
La foundation d'une gouvernance efficace repose sur un système de tracking capable d'attribuer chaque token consommé à un caller précis, un modèle spécifique et une fenêtre temporelle arbitraire. HolySheep AI expose nativement des endpoints de métriques qui, combinés à notre middleware maison, permettent un découpage avec une granularité au niveau du millisecond.
Middleware de Instrumentation Python
"""
HolySheep AI Cost Governance Middleware
Version: 2.1.0
Compatible Python 3.10+
"""
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from enum import Enum
from typing import Optional
import httpx
Configuration — À REMPLACER PAR VOS VALEURS
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
CALLER_HEADER = "X-Caller-ID"
REQUEST_ID_HEADER = "X-Request-ID"
class CostAlertLevel(Enum):
"""Niveaux d'alerte selon le budget consommé."""
OK = "ok"
WARNING_50 = "warning_50" # 50% du budget atteint
WARNING_75 = "warning_75" # 75% du budget atteint
CRITICAL_90 = "critical_90" # 90% du budget atteint
EXCEEDED = "exceeded" # Budget dépassé
@dataclass
class CostMetrics:
"""Métriques de coût agrégées."""
total_input_tokens: int = 0
total_output_tokens: int = 0
total_cost_usd: float = 0.0
request_count: int = 0
avg_latency_ms: float = 0.0
by_caller: dict = field(default_factory=dict)
by_model: dict = field(default_factory=dict)
by_hour: dict = field(default_factory=dict)
@dataclass
class BudgetConfig:
"""Configuration du budget avec alertes."""
monthly_limit_usd: float
caller_limits: dict[str, float] = field(default_factory=dict)
model_limits: dict[str, float] = field(default_factory=dict)
alert_thresholds: list[float] = field(default_factory=lambda: [0.50, 0.75, 0.90, 1.0])
class HolySheepCostTracker:
"""
Tracker de coûts HolySheep AI avec découpage multi-dimensionnel.
Fonctionnalités:
- Tracking par caller (identifiant client/application)
- Tracking par modèle (gpt-4.1, claude-sonnet-4.5, etc.)
- Tracking par fenêtre temporelle (horaire, quotidien, mensuel)
- Alertes budgetaires configurables
- Export JSON pour dashboards Grafana/Prometheus
"""
# Tarification HolySheep AI 2026 (USD par million de tokens)
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
"default": {"input": 5.0, "output": 5.0},
}
def __init__(
self,
budget_config: Optional[BudgetConfig] = None,
storage_adapter: Optional["StorageAdapter"] = None
):
self.budget = budget_config
self.metrics = CostMetrics()
self.storage = storage_adapter
self._lock = asyncio.Lock()
self._alert_callbacks: list[callable] = []
async def track_request(
self,
caller_id: str,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
request_id: Optional[str] = None
) -> CostMetrics:
"""Enregistre une requête et met à jour les métriques."""
# Calcul du coût
pricing = self.PRICING.get(model, self.PRICING["default"])
cost_usd = (
(input_tokens / 1_000_000) * pricing["input"] +
(output_tokens / 1_000_000) * pricing["output"]
)
timestamp = datetime.now(timezone.utc)
hour_key = timestamp.strftime("%Y-%m-%dT%H:00:00Z")
async with self._lock:
# Métriques globales
self.metrics.total_input_tokens += input_tokens
self.metrics.total_output_tokens += output_tokens
self.metrics.total_cost_usd += cost_usd
self.metrics.request_count += 1
# Métriques par caller
if caller_id not in self.metrics.by_caller:
self.metrics.by_caller[caller_id] = CostMetrics()
caller_metrics = self.metrics.by_caller[caller_id]
caller_metrics.total_cost_usd += cost_usd
caller_metrics.request_count += 1
# Métriques par modèle
if model not in self.metrics.by_model:
self.metrics.by_model[model] = CostMetrics()
model_metrics = self.metrics.by_model[model]
model_metrics.total_cost_usd += cost_usd
model_metrics.total_input_tokens += input_tokens
model_metrics.total_output_tokens += output_tokens
# Métriques par heure
if hour_key not in self.metrics.by_hour:
self.metrics.by_hour[hour_key] = CostMetrics()
hour_metrics = self.metrics.by_hour[hour_key]
hour_metrics.total_cost_usd += cost_usd
hour_metrics.request_count += 1
# Persistance asynchrone
if self.storage:
await self.storage.persist_metric(
caller_id=caller_id,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost_usd,
latency_ms=latency_ms,
timestamp=timestamp,
request_id=request_id or self._generate_request_id()
)
# Vérification des alertes
await self._check_alerts(caller_id, model)
return self.metrics
async def _check_alerts(self, caller_id: str, model: str):
"""Vérifie si les seuils d'alerte sont atteints."""
if not self.budget:
return
total_cost = self.metrics.total_cost_usd
alert_level = self._calculate_alert_level(total_cost, self.budget.monthly_limit_usd)
if alert_level != CostAlertLevel.OK:
for callback in self._alert_callbacks:
await callback(alert_level, total_cost, self.budget.monthly_limit_usd)
# Alert caller-specific si configuré
if caller_id in self.budget.caller_limits:
caller_cost = self.metrics.by_caller[caller_id].total_cost_usd
caller_alert = self._calculate_alert_level(
caller_cost, self.budget.caller_limits[caller_id]
)
if caller_alert != CostAlertLevel.OK:
await self._trigger_caller_alert(caller_id, caller_alert)
def _calculate_alert_level(self, spent: float, limit: float) -> CostAlertLevel:
"""Calcule le niveau d'alerte selon le ratio dépenses/limite."""
if limit <= 0:
return CostAlertLevel.EXCEEDED
ratio = spent / limit
if ratio >= 1.0:
return CostAlertLevel.EXCEEDED
elif ratio >= 0.90:
return CostAlertLevel.CRITICAL_90
elif ratio >= 0.75:
return CostAlertLevel.WARNING_75
elif ratio >= 0.50:
return CostAlertLevel.WARNING_50
else:
return CostAlertLevel.OK
async def _trigger_caller_alert(self, caller_id: str, alert: CostAlertLevel):
"""Envoie une alerte spécifique au caller."""
print(f"[ALERTE] Caller '{caller_id}' — Niveau: {alert.value}")
def _generate_request_id(self) -> str:
"""Génère un ID de requête unique."""
timestamp = str(time.time_ns())
return hashlib.sha256(timestamp.encode()).hexdigest()[:16]
def get_cost_report(self) -> dict:
"""Génère un rapport complet des coûts."""
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"summary": {
"total_cost_usd": round(self.metrics.total_cost_usd, 4),
"total_input_tokens": self.metrics.total_input_tokens,
"total_output_tokens": self.metrics.total_output_tokens,
"request_count": self.metrics.request_count,
"avg_cost_per_request": (
round(self.metrics.total_cost_usd / self.metrics.request_count, 6)
if self.metrics.request_count > 0 else 0
),
},
"by_caller": {
caller: {
"cost_usd": round(data.total_cost_usd, 4),
"requests": data.request_count
}
for caller, data in self.metrics.by_caller.items()
},
"by_model": {
model: {
"cost_usd": round(data.total_cost_usd, 4),
"input_tokens": data.total_input_tokens,
"output_tokens": data.total_output_tokens
}
for model, data in self.metrics.by_model.items()
},
"by_hour": dict(self.metrics.by_hour),
"budget_status": self._get_budget_status()
}
def _get_budget_status(self) -> dict:
"""Retourne le statut du budget."""
if not self.budget:
return {"configured": False}
total = self.metrics.total_cost_usd
limit = self.budget.monthly_limit_usd
return {
"configured": True,
"limit_usd": limit,
"spent_usd": round(total, 4),
"remaining_usd": round(max(0, limit - total), 4),
"utilization_percent": round((total / limit) * 100, 2) if limit > 0 else 0,
"alert_level": self._calculate_alert_level(total, limit).value
}
Exemple d'utilisation
async def main():
# Configuration du budget
budget = BudgetConfig(
monthly_limit_usd=500.0,
caller_limits={
"app-mobile": 100.0,
"app-web": 200.0,
"batch-processor": 150.0
},
model_limits={
"claude-sonnet-4.5": 200.0,
"gpt-4.1": 150.0
}
)
tracker = HolySheepCostTracker(budget_config=budget)
# Simulation de requêtes
test_calls = [
("app-web", "gpt-4.1", 15000, 3200, 145.3),
("app-mobile", "claude-sonnet-4.5", 8000, 1800, 230.5),
("batch-processor", "deepseek-v3.2", 45000, 12000, 42.1),
("app-web", "gemini-2.5-flash", 5000, 950, 38.7),
]
for caller, model, input_tok, output_tok, latency in test_calls:
await tracker.track_request(caller, model, input_tok, output_tok, latency)
# Affichage du rapport
import json
report = tracker.get_cost_report()
print(json.dumps(report, indent=2, default=str))
if __name__ == "__main__":
asyncio.run(main())Client HTTP Asynchrone avec Rate Limiting Intelligent
"""
HolySheep AI Async Client avec Cost Tracking Intégré
Version: 2.1.0 — Production Ready
"""
import asyncio
import json
from typing import Any, AsyncIterator
from dataclasses import dataclass
import httpx
from holySheep_cost_tracker import HolySheepCostTracker, BudgetConfig
@dataclass
class HolySheepRequest:
"""Configuration d'une requête HolySheep."""
model: str
messages: list[dict]
temperature: float = 0.7
max_tokens: int = 2048
caller_id: str = "default"
stream: bool = False
extra_params: dict = None
class HolySheepAIClient:
"""
Client asynchrone HolySheep AI avec:
- Rate limiting adaptatif
- Retry exponentiel avec jitter
- Cost tracking en temps réel
- Cache des réponses fréquentes
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_concurrent: int = 10,
cost_tracker: HolySheepCostTracker = None
):
self.api_key = api_key
self.base_url = base_url
self.max_concurrent = max_concurrent
self.cost_tracker = cost_tracker or HolySheepCostTracker()
# Client HTTP avec pooling
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0, connect=10.0),
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-SDK": "holySheep-python/2.1.0"
}
)
# Rate limiter sémaphore
self._semaphore = asyncio.Semaphore(max_concurrent)
# Cache LRU simple
self._response_cache: dict[str, Any] = {}
self._cache_hits = 0
self._cache_misses = 0
async def chat_completions(
self,
request: HolySheepRequest
) -> dict[str, Any]:
"""
Envoie une requête de chat completion avec tracking complet.
Returns:
dict avec 'content', 'usage', 'latency_ms', 'cost_usd'
"""
async with self._semaphore:
start_time = asyncio.get_event_loop().time()
# Vérification cache
cache_key = self._generate_cache_key(request)
if cache_key in self._response_cache:
self._cache_hits += 1
return self._response_cache[cache_key]
try:
response = await self._make_request(request)
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
# Extraction des métriques d'usage
usage = response.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# Tracking des coûts
await self.cost_tracker.track_request(
caller_id=request.caller_id,
model=request.model,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms
)
# Construction de la réponse
result = {
"content": response.get("choices", [{}])[0].get("message", {}).get("content", ""),
"usage": {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens
},
"latency_ms": round(latency_ms, 2),
"model": request.model,
"finish_reason": response.get("choices", [{}])[0].get("finish_reason")
}
# Mise en cache
if request.caller_id != "batch-processor":
self._response_cache[cache_key] = result
self._cache_misses += 1
return result
except httpx.HTTPStatusError as e:
await self._handle_http_error(e, request)
raise
async def _make_request(self, request: HolySheepRequest) -> dict:
"""Effectue la requête HTTP avec retry."""
payload = {
"model": request.model,
"messages": request.messages,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
"stream": request.stream
}
if request.extra_params:
payload.update(request.extra_params)
url = f"{self.base_url}/chat/completions"
# Retry avec backoff exponentiel
for attempt in range(3):
try:
response = await self._client.post(
url,
json=payload,
headers={"X-Caller-ID": request.caller_id}
)
response.raise_for_status()
return response.json()
except (httpx.ConnectError, httpx.TimeoutException) as e:
if attempt == 2:
raise RuntimeError(f"Échec après 3 tentatives: {e}")
await asyncio.sleep(2 ** attempt + asyncio.get_event_loop().time() % 1)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Rate limited — attente plus longue
retry_after = int(e.response.headers.get("Retry-After", 60))
await asyncio.sleep(retry_after)
elif e.response.status_code >= 500:
await asyncio.sleep(2 ** attempt)
else:
raise
async def _handle_http_error(self, error: httpx.HTTPStatusError, request: HolySheepRequest):
"""Gestion des erreurs HTTP avec logging."""
status = error.response.status_code
body = error.response.text[:500]
error_types = {
401: "Clé API invalide ou expirée",
403: "Accès interdit — vérifiez les permissions",
429: "Rate limit atteint — implémentez du backoff",
500: "Erreur serveur HolySheep — réessayez",
503: "Service temporairement indisponible"
}
message = error_types.get(status, f"Erreur HTTP {status}")
raise RuntimeError(f"{message}: {body}")
def _generate_cache_key(self, request: HolySheepRequest) -> str:
"""Génère une clé de cache pour éviter les doublons."""
import hashlib
content = json.dumps({
"model": request.model,
"messages": request.messages,
"temperature": request.temperature
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
async def stream_chat_completions(
self,
request: HolySheepRequest
) -> AsyncIterator[str]:
"""Streaming avec tracking différé des coûts."""
request.stream = True
url = f"{self.base_url}/chat/completions"
payload = {
"model": request.model,
"messages": request.messages,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
"stream": True
}
async with self._semaphore:
async with self._client.stream(
"POST",
url,
json=payload,
headers={"X-Caller-ID": request.caller_id}
) as response:
response.raise_for_status()
full_content = ""
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
chunk = json.loads(data)
delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
if delta:
full_content += delta
yield delta
# Estimation des coûts après streaming
# Note: En production, utilisez les vraies métriques d'usage
estimated_tokens = len(full_content.split()) * 1.3
estimated_cost = (estimated_tokens / 1_000_000) * \
self.cost_tracker.PRICING.get(request.model, {}).get("output", 5.0)
await self.cost_tracker.track_request(
caller_id=request.caller_id,
model=request.model,
input_tokens=request.max_tokens // 2,
output_tokens=int(estimated_tokens),
latency_ms=0
)
async def close(self):
"""Ferme le client proprement."""
await self._client.aclose()
print(f"Cache stats — Hits: {self._cache_hits}, Misses: {self._cache_misses}")
Script de test avec benchmark
async def benchmark():
"""Benchmark de performance du client HolySheep."""
import time
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=5
)
test_request = HolySheepRequest(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": "Tu es un assistant technique concis."},
{"role": "user", "content": "Explique la différence entre REST et GraphQL en 3 lignes."}
],
caller_id="benchmark-test",
max_tokens=150
)
# Test de latence
latencies = []
for i in range(10):
start = time.perf_counter()
result = await client.chat_completions(test_request)
latency = (time.perf_counter() - start) * 1000
latencies.append(latency)
print(f"Requête {i+1}: {latency:.1f}ms, Coût: ${result.get('usage', {}).get('total_tokens', 0) / 1_000_000 * 0.42:.6f}")
avg_latency = sum(latencies) / len(latencies)
p95_latency = sorted(latencies)[int(len(latencies) * 0.95)]
print(f"\n--- Benchmark Results ---")
print(f"Latence moyenne: {avg_latency:.1f}ms")
print(f"Latence P95: {p95_latency:.1f}ms")
await client.close()
if __name__ == "__main__":
asyncio.run(benchmark())Découpage des Coûts par Dimension
La puissance de HolySheep AI réside dans sa capacité à fournir des métriques exploitables. En structurant votre système de tracking autour de trois axes — caller, modèle et période — vous obtainez une visibilité complète sur vos patterns de consommation.
Tableau Comparatif des Modèles HolySheep 2026
| Modèle | Input ($/M tokens) | Output ($/M tokens) | Latence Typique | Use Case Optimal | Ratio Coût/Efficacité |
|---|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $0.42 | <50ms | Batch processing, volume élevé | ⭐⭐⭐⭐⭐ Excellent |
| Gemini 2.5 Flash | $2.50 | $2.50 | <50ms | Applications temps réel | ⭐⭐⭐⭐ Très bon |
| GPT-4.1 | $8.00 | $8.00 | <100ms | Tâches complexes, coding | ⭐⭐⭐ Bon |
| Claude Sonnet 4.5 | $15.00 | $15.00 | <120ms | Analyse, reasoning avancé | ⭐⭐ Investissement premium |
Système d'Alertes Budgetaires
Un budget sans alertes est une bombe à retardement. Mon implémentation inclut un système de seuils progressifs qui vous notifie avant d'atteindre des limites critiques. Avec HolySheep AI offrant un taux de change ¥1=$1 et des crédits gratuits pour les nouveaux utilisateurs, vous pouvez expérimenter sans risque avant de vous engager.
"""
HolySheep AI Budget Alert Manager
Version: 2.1.0 — Production Ready
"""
import asyncio
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Protocol
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class AlertChannel(Enum):
"""Canaux de notification disponibles."""
EMAIL = "email"
WEBHOOK = "webhook"
SLACK = "slack"
PAGERDUTY = "pagerduty"
CONSOLE = "console"
@dataclass
class AlertConfig:
"""Configuration d'une alerte."""
channel: AlertChannel
threshold_percent: float
recipient: str # email, webhook URL, slack channel, etc.
enabled: bool = True
cooldown_seconds: int = 3600 # Évite le spam d'alertes
@dataclass
class BudgetAlert:
"""Représentation d'une alerte déclenchée."""
timestamp: datetime
level: str
current_spend: float
budget_limit: float
utilization_percent: float
caller_id: str = "global"
model: str = "all"
class AlertManager:
"""
Gestionnaire d'alertes budgetaires multi-canal.
Fonctionnalités:
- Seuils configurables (50%, 75%, 90%, 100%)
- Cooldown pour éviter le spam
- Multi-canal (email, webhook, Slack, PagerDuty)
- Persistance des alertes envoyées
"""
def __init__(self):
self.alerts: list[AlertConfig] = []
self._sent_alerts: dict[str, datetime] = {} # Clé -> Dernière envoi
self._lock = asyncio.Lock()
def add_alert(self, config: AlertConfig):
"""Ajoute une configuration d'alerte."""
self.alerts.append(config)
def setup_default_alerts(self):
"""Configure les alertes par défaut recommandées."""
self.alerts = [
# Alertes globales
AlertConfig(
channel=AlertChannel.EMAIL,
threshold_percent=50.0,
recipient="[email protected]",
cooldown_seconds=86400 # 1 alert/jour max
),
AlertConfig(
channel=AlertChannel.WEBHOOK,
threshold_percent=75.0,
recipient="https://hooks.slack.com/services/XXX",
cooldown_seconds=43200
),
AlertConfig(
channel=AlertChannel.SLACK,
threshold_percent=90.0,
recipient="#ops-alerts",
cooldown_seconds=7200
),
# Alertes critiques
AlertConfig(
channel=AlertChannel.PAGERDUTY,
threshold_percent=100.0,
recipient="BUDGET_EXCEEDED",
cooldown_seconds=3600
),
]
async def check_and_trigger(
self,
current_spend: float,
budget_limit: float,
caller_id: str = "global",
model: str = "all"
) -> list[BudgetAlert]:
"""Vérifie les seuils et déclenche les alertes nécessaires."""
if budget_limit <= 0:
return []
utilization = (current_spend / budget_limit) * 100
alerts_triggered = []
async with self._lock:
for config in self.alerts:
if not config.enabled:
continue
if utilization >= config.threshold_percent:
alert = BudgetAlert(
timestamp=datetime.now(timezone.utc),
level=f"alert_{int(config.threshold_percent)}",
current_spend=round(current_spend, 2),
budget_limit=budget_limit,
utilization_percent=round(utilization, 2),
caller_id=caller_id,
model=model
)
# Vérifie le cooldown
cooldown_key = self._get_cooldown_key(config, caller_id)
if self._can_send_alert(cooldown_key, config.cooldown_seconds):
await self._send_alert(config, alert)
self._sent_alerts[cooldown_key] = datetime.now(timezone.utc)
alerts_triggered.append(alert)
return alerts_triggered
def _get_cooldown_key(self, config: AlertConfig, caller_id: str) -> str:
"""Génère la clé de cooldown unique."""
return f"{config.channel.value}:{config.recipient}:{caller_id}"
def _can_send_alert(self, cooldown_key: str, cooldown_seconds: int) -> bool:
"""Vérifie si l'alerte peut être envoyée (hors cooldown)."""
if cooldown_key not in self._sent_alerts:
return True
last_sent = self._sent_alerts[cooldown_key]
elapsed = (datetime.now(timezone.utc) - last_sent).total_seconds()
return elapsed >= cooldown_seconds
async def _send_alert(self, config: AlertConfig, alert: BudgetAlert):
"""Envoie l'alerte via le canal configuré."""
if config.channel == AlertChannel.EMAIL:
await self._send_email(config.recipient, alert)
elif config.channel == AlertChannel.WEBHOOK:
await self._send_webhook(config.recipient, alert)
elif config.channel == AlertChannel.SLACK:
await self._send_slack(config.recipient, alert)
elif config.channel == AlertChannel.CONSOLE:
self._send_console(alert)
async def _send_email(self, recipient: str, alert: BudgetAlert):
"""Envoie une alerte par email."""
msg = MIMEMultipart()
msg['From'] = '[email protected]'
msg['To'] = recipient
msg['Subject'] = f"⚠️ Alerte Budget HolySheep — {alert.utilization_percent}% atteint"
body = f"""
🔴 Alerte Budget HolySheep AI
Niveau: {alert.level} Dépense actuelle: ${alert.current_spend:.2f} Limite budgétaire: ${alert.budget_limit:.2f} Utilisation: {alert.utilization_percent}% Caller: {alert.caller_id} Modèle: {alert.model} Timestamp: {alert.timestamp.isoformat()}
"""
msg.attach(MIMEText(body, 'html'))
# En production, utilisez un vrai serveur SMTP
print(f"[EMAIL] Alerte envoyée à {recipient}: {alert.level}")
async def _send_webhook(self, url: str, alert: BudgetAlert):
"""Envoie une alerte via webhook."""
payload = {
"event": "budget_alert",
"level": alert.level,
"current_spend_usd": alert.current_spend,
"budget_limit_usd": alert.budget_limit,
"utilization_percent": alert.utilization_percent,
"caller_id": alert.caller_id,
"model": alert.model,
"timestamp": alert.timestamp.isoformat()
}
async with httpx.AsyncClient() as client:
try:
response = await client.post(
url,
json=payload,
timeout=10.0
)
print(f"[WEBHOOK] Alerte envoyée: {response.status_code}")
except Exception as e:
print(f"[WEBHOOK] Erreur: {e}")
async def _send_slack(self, channel: str, alert: BudgetAlert):
"""Envoie une alerte Slack formatée."""
emoji = "🔴" if alert.utilization_percent >= 90 else "🟠"
payload = {
"channel": channel,
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji} Alerte Budget HolySheep AI"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Niveau:*\n{alert.level}"},
{"type": "mrkdwn", "text": f"*Utilisation:*\n{alert.utilization_percent}%"},
{"type": "mrkdwn", "text": f"*Dép