Introduction : Le Cas de Madame Lefort et son Catastrophe de Budget
Imaginez la scène : nous sommes en mars 2026, et Madame Lefort, CTO d'une jeune startup e-commerce bordelaise, découvre avec horreur que sa facture mensuelle d'IA a atteint 12 800 € en une seule nuit. Comment ? Un script défectueux qui générait des descriptions produits en boucle pendant le week-end.
Ce cauchemar, je l'ai vécu de l'intérieur — en tant qu'ingénieur solutions chez un intégrateur IA, j'ai rescue plus de 15 projets de ce genre. La leçon est claire : sans monitoring des tokens et sans alertes de budget, vous naviguez à l'aveugle dans un océan de coûts qui peuvent exploser en quelques heures.
Dans ce tutoriel, je vais vous montrer comment implémenter un système complet de surveillance et d'alertes pour vos modèles IA, en utilisant
HolySheep AI comme plateforme de référence. Préparez votre éditeur de code : nous allons construire cela ensemble.
Comprendre les Tokens et leur Impact Financier
Qu'est-ce qu'un Token ?
Un token représente environ 4 caractères de texte ou une fraction de mot. Quand vous envoyez "Bonjour, comment allez-vous ?" à un modèle, cela représente environ 7 tokens. Chaque requête que vous envoyez (prompt) et chaque réponse générée (completion) consomme des tokens, et donc de l'argent.
Tableau Comparatif des Coûts 2026
| Modèle | Prix par Million de Tokens (Input) | Prix par Million de Tokens (Output) | Latence Moyenne |
|--------|-----------------------------------|-------------------------------------|-----------------|
| GPT-4.1 | $8,00 | $24,00 | ~850ms |
| Claude Sonnet 4.5 | $15,00 | $75,00 | ~920ms |
| Gemini 2.5 Flash | $2,50 | $10,00 | ~380ms |
| DeepSeek V3.2 | $0,42 | $1,90 | ~210ms |
Comme vous pouvez le voir, le choix du modèle a un impact dramatique sur vos coûts. HolySheep AI offre un taux de change avantageux avec ¥1 = $1, ce qui représente une économie de plus de 85% pour les utilisateurs chinois et une réduction significative des frais de change pour tous.
Architecture du Système de Monitoring
Notre solution repose sur trois piliers fondamentaux : la collecte en temps réel des métriques, le stockage structuré des données, et le système d'alertes intelligent.
Composants Nécessaires
Pour implémenter ce système, vous aurez besoin de :
- Un projet Python 3.10+ avec les dépendances appropriées
- Une instance Redis pour le cache temps réel
- Une base PostgreSQL pour l'historique
- L'API HolySheep pour les requêtes IA
- Un service d'alertes (email, webhook, ou SMS)
Implémentation : Le Client Python de Monitoring
Commençons par créer le client principal qui encapsulera tous nos appels API tout en Trackant automatiquement la consommation.
# monitor_client.py
import time
import hashlib
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
import threading
@dataclass
class TokenUsage:
"""Structure de données pour une utilisation de tokens"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
request_id: str
cost_usd: float
cost_cny: float
@dataclass
class BudgetAlert:
"""Configuration d'une alerte de budget"""
threshold_usd: float
current_spend: float = 0.0
enabled: bool = True
webhook_url: Optional[str] = None
email: Optional[str] = None
class HolySheepMonitor:
"""
Client monitoré pour HolySheep AI avec tracking automatique des tokens.
Avantages HolySheep :
- Taux de change ¥1 = $1 (économie 85%+ vs fournisseurs occidentaux)
- Latence moyenne <50ms pour les modèles optimisés
- Support WeChat/Alipay pour les paiements
- Crédits gratuits pour les nouveaux inscrits
"""
# Tarifs 2026 (en USD par million de tokens)
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 24.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 75.0},
"gemini-2.5-flash": {"input": 2.5, "output": 10.0},
"deepseek-v3.2": {"input": 0.42, "output": 1.90}
}
# Taux de change HolySheep
CNY_EXCHANGE_RATE = 1.0 # ¥1 = $1
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self._usage_history: List[TokenUsage] = []
self._daily_usage: Dict[str, float] = defaultdict(float)
self._monthly_usage: Dict[str, float] = defaultdict(float)
self._alerts: Dict[str, BudgetAlert] = {}
self._lock = threading.Lock()
def _generate_request_id(self, prompt: str) -> str:
"""Génère un ID unique pour la requête"""
timestamp = str(time.time())
return hashlib.sha256(f"{prompt}{timestamp}".encode()).hexdigest()[:16]
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> Tuple[float, float]:
"""Calcule le coût en USD et CNY"""
pricing = self.PRICING.get(model, {"input": 10.0, "output": 30.0})
cost_usd = (input_tokens / 1_000_000 * pricing["input"] +
output_tokens / 1_000_000 * pricing["output"])
cost_cny = cost_usd * self.CNY_EXCHANGE_RATE
return cost_usd, cost_cny
def _check_alerts(self, usage: TokenUsage):
"""Vérifie si une alerte doit être déclenchée"""
for alert_name, alert in self._alerts.items():
if not alert.enabled:
continue
if usage.cost_usd + alert.current_spend >= alert.threshold_usd:
self._trigger_alert(alert_name, alert, usage)
def _trigger_alert(self, alert_name: str, alert: BudgetAlert, usage: TokenUsage):
"""Déclenche une alerte (implémentation basique)"""
print(f"🚨 ALERTE BUDGET [{alert_name}]: {alert.current_spend:.2f}$ / {alert.threshold_usd}$")
print(f" Nouvelle requête: {usage.cost_usd:.4f}$ - Modèle: {usage.model}")
def chat_completion(
self,
model: str,
messages: List[Dict],
max_tokens: int = 1000,
temperature: float = 0.7,
track_usage: bool = True
) -> Dict:
"""
Envoie une requête de chat completion avec monitoring automatique.
Args:
model: Nom du modèle (ex: "deepseek-v3.2", "gemini-2.5-flash")
messages: Liste des messages au format OpenAI
max_tokens: Limite de tokens pour la réponse
temperature: Température de génération
track_usage: Si True, enregistre l'utilisation des tokens
Returns:
Réponse de l'API avec métadonnées de coût ajoutées
"""
import requests
request_id = self._generate_request_id(str(messages))
# Estimation initiale pour le budget
estimated_input_tokens = sum(
len(str(m)) // 4 for m in messages
)
estimated_output_tokens = max_tokens
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
start_time = time.time()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start_time) * 1000
if track_usage:
# Extraction des tokens depuis la réponse
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", estimated_input_tokens)
output_tokens = usage.get("completion_tokens", 0)
cost_usd, cost_cny = self._calculate_cost(model, input_tokens, output_tokens)
token_usage = TokenUsage(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
request_id=request_id,
cost_usd=cost_usd,
cost_cny=cost_cny
)
with self._lock:
self._usage_history.append(token_usage)
self._daily_usage[datetime.now().date().isoformat()] += cost_usd
self._monthly_usage[datetime.now().strftime("%Y-%m")] += cost_usd
self._check_alerts(token_usage)
# Ajout des métadonnées à la réponse
data["usage_metadata"] = {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
"cost_usd": round(cost_usd, 6),
"cost_cny": round(cost_cny, 6),
"latency_ms": round(latency_ms, 2),
"request_id": request_id
}
return data
except requests.exceptions.RequestException as e:
raise Exception(f"Erreur API HolySheep: {str(e)}")
def set_budget_alert(
self,
name: str,
threshold_usd: float,
webhook_url: Optional[str] = None,
email: Optional[str] = None
):
"""Configure une alerte de budget"""
self._alerts[name] = BudgetAlert(
threshold_usd=threshold_usd,
webhook_url=webhook_url,
email=email
)
def get_usage_report(self, days: int = 7) -> Dict:
"""Génère un rapport d'utilisation sur N jours"""
cutoff = datetime.now() - timedelta(days=days)
with self._lock:
recent_usage = [u for u in self._usage_history if u.timestamp >= cutoff]
if not recent_usage:
return {"error": "Aucune donnée disponible"}
total_cost = sum(u.cost_usd for u in recent_usage)
total_input = sum(u.input_tokens for u in recent_usage)
total_output = sum(u.output_tokens for u in recent_usage)
by_model = defaultdict(lambda: {"requests": 0, "cost": 0, "tokens": 0})
for usage in recent_usage:
by_model[usage.model]["requests"] += 1
by_model[usage.model]["cost"] += usage.cost_usd
by_model[usage.model]["tokens"] += usage.input_tokens + usage.output_tokens
return {
"period_days": days,
"total_requests": len(recent_usage),
"total_cost_usd": round(total_cost, 4),
"total_cost_cny": round(total_cost * self.CNY_EXCHANGE_RATE, 4),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"avg_cost_per_request": round(total_cost / len(recent_usage), 6),
"by_model": dict(by_model)
}
Exemple d'utilisation
if __name__ == "__main__":
monitor = HolySheepMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
# Configuration des alertes
monitor.set_budget_alert(
name="daily_limit",
threshold_usd=100.0,
email="[email protected]"
)
monitor.set_budget_alert(
name="monthly_limit",
threshold_usd=2000.0,
webhook_url="https://hooks.votredomaine.com/alertes"
)
# Première requête monitorée
response = monitor.chat_completion(
model="deepseek-v3.2", # Modèle économique à $0.42/M tokens input
messages=[
{"role": "system", "content": "Tu es un assistant commercial expert."},
{"role": "user", "content": "Explique les avantages de HolySheep AI"}
],
max_tokens=500
)
print(f"Coût de la requête: {response['usage_metadata']['cost_usd']:.6f}$")
print(f"Latence: {response['usage_metadata']['latency_ms']:.2f}ms")
Système d'Alertes Avancé avec Webhooks
Maintenant, créons le module d'alertes sophistiqué qui enverra des notifications intelligentes quand vos seuils seront atteints.
# alert_system.py
import json
import smtplib
import asyncio
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
import aiohttp
from collections import deque
class AlertSeverity(Enum):
"""Niveaux de sévérité des alertes"""
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
EMERGENCY = "emergency"
@dataclass
class Alert:
"""Structure d'une alerte"""
id: str
timestamp: datetime
severity: AlertSeverity
title: str
message: str
metric_name: str
current_value: float
threshold_value: float
percentage: float
metadata: Dict = None
class AlertManager:
"""
Gestionnaire d'alertes multiples pour le monitoring des coûts IA.
Fonctionnalités :
- Alertes temps réel avec seuils personnalisables
- Notifications multi-canal (email, webhook, SMS)
- Cooldowns pour éviter la fatigue d'alertes
- Historique et statistiques
"""
# Seuils par défaut (en USD)
DEFAULT_THRESHOLDS = {
"per_request": {"warning": 0.50, "critical": 1.00, "emergency": 5.00},
"hourly": {"warning": 10.0, "critical": 50.0, "emergency": 100.0},
"daily": {"warning": 100.0, "critical": 500.0, "emergency": 1000.0},
"monthly": {"warning": 1000.0, "critical": 5000.0, "emergency": 10000.0}
}
def __init__(
self,
webhook_url: Optional[str] = None,
email_config: Optional[Dict] = None,
cooldown_minutes: int = 15
):
self.webhook_url = webhook_url
self.email_config = email_config
self.cooldown_minutes = cooldown_minutes
self._thresholds = self.DEFAULT_THRESHOLDS.copy()
self._last_alerts: deque = deque(maxlen=1000)
self._cooldowns: Dict[str, datetime] = {}
self._custom_handlers: List[Callable] = []
def set_threshold(
self,
metric: str,
severity: str,
value: float
):
"""Configure un seuil personnalisé"""
if metric not in self._thresholds:
self._thresholds[metric] = {}
self._thresholds[metric][severity] = value
def check_cost(
self,
cost_usd: float,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
request_id: str
) -> List[Alert]:
"""
Vérifie les coûts et génère les alertes si nécessaire.
Returns:
Liste des alertes déclenchées
"""
alerts = []
now = datetime.now()
# Vérification par requête
per_request_severity = self._get_severity("per_request", cost_usd)
if per_request_severity:
alert = self._create_alert(
severity=per_request_severity,
title=f"Coût par requête élevé: {model}",
message=f"Une requête au modèle {model} a coûté ${cost_usd:.4f}",
metric_name="per_request",
current_value=cost_usd,
threshold_value=self._thresholds["per_request"][per_request_severity.value],
metadata={
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"latency_ms": latency_ms,
"request_id": request_id
}
)
alerts.append(alert)
# Vérification des alertes configurées
for alert_name, threshold in self._thresholds.items():
if alert_name == "per_request":
continue
for severity, value in threshold.items():
if cost_usd >= value:
cooldown_key = f"{alert_name}_{severity}"
if self._is_in_cooldown(cooldown_key):
continue
alert = self._create_alert(
severity=AlertSeverity(severity),
title=f"Seuil {severity.upper()} atteint",
message=f"Coût actuel: ${cost_usd:.2f} / Seuil: ${value:.2f}",
metric_name=alert_name,
current_value=cost_usd,
threshold_value=value
)
alerts.append(alert)
self._set_cooldown(cooldown_key)
# Dispatch des alertes
for alert in alerts:
self._dispatch_alert(alert)
return alerts
def _get_severity(self, metric: str, value: float) -> Optional[AlertSeverity]:
"""Détermine la sévérité basée sur les seuils"""
thresholds = self._thresholds.get(metric, {})
if value >= thresholds.get("emergency", float('inf')):
return AlertSeverity.EMERGENCY
elif value >= thresholds.get("critical", float('inf')):
return AlertSeverity.CRITICAL
elif value >= thresholds.get("warning", float('inf')):
return AlertSeverity.WARNING
return None
def _create_alert(
self,
severity: AlertSeverity,
title: str,
message: str,
metric_name: str,
current_value: float,
threshold_value: float,
metadata: Optional[Dict] = None
) -> Alert:
"""Crée une nouvelle alerte"""
return Alert(
id=f"{metric_name}_{severity.value}_{datetime.now().timestamp()}",
timestamp=datetime.now(),
severity=severity,
title=title,
message=message,
metric_name=metric_name,
current_value=current_value,
threshold_value=threshold_value,
percentage=round((current_value / threshold_value) * 100, 1),
metadata=metadata or {}
)
def _is_in_cooldown(self, key: str) -> bool:
"""Vérifie si l'alerte est en période de cooldown"""
if key not in self._cooldowns:
return False
elapsed = (datetime.now() - self._cooldowns[key]).total_seconds() / 60
return elapsed < self.cooldown_minutes
def _set_cooldown(self, key: str):
"""Active le cooldown pour une alerte"""
self._cooldowns[key] = datetime.now()
def _dispatch_alert(self, alert: Alert):
"""Distribue l'alerte vers tous les canaux configurés"""
self._last_alerts.append(alert)
# Handlers personnalisés
for handler in self._custom_handlers:
try:
handler(alert)
except Exception as e:
print(f"Erreur handler personnalisé: {e}")
# Webhook async
if self.webhook_url:
asyncio.create_task(self._send_webhook(alert))
# Email
if self.email_config:
self._send_email(alert)
async def _send_webhook(self, alert: Alert):
"""Envoie l'alerte via webhook"""
try:
payload = {
"alert_id": alert.id,
"timestamp": alert.timestamp.isoformat(),
"severity": alert.severity.value,
"title": alert.title,
"message": alert.message,
"metric": alert.metric_name,
"current_value": alert.current_value,
"threshold": alert.threshold_value,
"percentage": alert.percentage,
"metadata": alert.metadata
}
async with aiohttp.ClientSession() as session:
async with session.post(
self.webhook_url,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status != 200:
print(f"Webhook error: {response.status}")
except Exception as e:
print(f"Erreur webhook: {e}")
def _send_email(self, alert: Alert):
"""Envoie l'alerte par email"""
if not self.email_config:
return
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[{alert.severity.value.upper()}] {alert.title}"
msg["From"] = self.email_config["from"]
msg["To"] = self.email_config["to"]
html = f"""
<html>
<body style="font-family: Arial, sans-serif;">
<div style="background-color: {'#ffcccc' if alert.severity == AlertSeverity.CRITICAL else '#fff3cd'};
padding: 20px; border-radius: 10px;">
<h2 style="color: #333;">🚨 {alert.title}</h2>
<p><strong>Sévérité :</strong> {alert.severity.value.upper()}</p>
<p><strong>Message :</strong> {alert.message}</p>
<p><strong>Valeur actuelle :</strong> ${alert.current_value:.4f}</p>
<p><strong>Seuil :</strong> ${alert.threshold_value:.2f}</p>
<p><strong>Progression :</strong> {alert.percentage}%</p>
<hr>
<p style="color: #666; font-size: 12px;">
Alerte générée le {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')}
</p>
</div>
</body>
</html>
"""
msg.attach(MIMEText(html, "html"))
with smtplib.SMTP(
self.email_config["smtp_host"],
self.email_config["smtp_port"]
) as server:
server.starttls()
server.login(
self.email_config["smtp_user"],
self.email_config["smtp_password"]
)
server.send_message(msg)
except Exception as e:
print(f"Erreur envoi email: {e}")
def add_handler(self, handler: Callable[[Alert], None]):
"""Ajoute un handler personnalisé pour les alertes"""
self._custom_handlers.append(handler)
def get_alert_history(
self,
hours: int = 24,
severity: Optional[AlertSeverity] = None
) -> List[Dict]:
"""Récupère l'historique des alertes"""
cutoff = datetime.now().timestamp() - (hours * 3600)
alerts = [
asdict(a) for a in self._last_alerts
if a.timestamp.timestamp() >= cutoff
]
if severity:
alerts = [a for a in alerts if a["severity"] == severity.value]
return alerts
def get_statistics(self) -> Dict:
"""Génère des statistiques sur les alertes"""
if not self._last_alerts:
return {"total": 0, "by_severity": {}}
by_severity = {}
for alert in self._last_alerts:
sev = alert.severity.value
by_severity[sev] = by_severity.get(sev, 0) + 1
return {
"total": len(self._last_alerts),
"by_severity": by_severity,
"last_24h": len([
a for a in self._last_alerts
if (datetime.now() - a.timestamp).total_seconds() < 86400
])
}
Démonstration
if __name__ == "__main__":
# Configuration du gestionnaire d'alertes
alert_manager = AlertManager(
webhook_url="https://api.votredomaine.com/alertes",
email_config={
"from": "[email protected]",
"to": "[email protected]",
"smtp_host": "smtp.gmail.com",
"smtp_port": 587,
"smtp_user": "[email protected]",
"smtp_password": "votre_mot_de_passe"
},
cooldown_minutes=5
)
# Configuration de seuils personnalisés
alert_manager.set_threshold("per_request", "warning", 0.10)
alert_manager.set_threshold("per_request", "critical", 0.50)
alert_manager.set_threshold("per_request", "emergency", 2.00)
# Handler personnalisé pour slack ou autre
def slack_notification(alert: Alert):
print(f"[SLACK] #{alert.severity.value}: {alert.title}")
print(f" → {alert.message}")
alert_manager.add_handler(slack_notification)
# Test de détection d'alertes
test_scenarios = [
{"cost": 0.05, "model": "deepseek-v3.2"},
{"cost": 0.15, "model": "gemini-2.5-flash"},
{"cost": 0.75, "model": "claude-sonnet-4.5"},
{"cost": 5.50, "model": "gpt-4.1"}
]
print("=" * 60)
print("SIMULATION DE DÉTECTION D'ALERTES")
print("=" * 60)
for scenario in test_scenarios:
print(f"\n📊 Test: Coût = ${scenario['cost']:.2f} ({scenario['model']})")
alerts = alert_manager.check_cost(
cost_usd=scenario["cost"],
model=scenario["model"],
input_tokens=1500,
output_tokens=500,
latency_ms=250.0,
request_id="test_req_001"
)
if alerts:
for alert in alerts:
print(f" ✅ Alerte déclenchée: [{alert.severity.value.upper()}] {alert.title}")
else:
print(" ✅ Aucune alerte (coût acceptable)")
print("\n" + "=" * 60)
print("STATISTIQUES")
print("=" * 60)
stats = alert_manager.get_statistics()
print(f"Total des alertes: {stats['total']}")
print(f"Par sévérité: {stats['by_severity']}")
Dashboard Temps Réel avec Streamlit
Pour visualiser vos données en temps réel, créons un dashboard interactif avec Streamlit.
# dashboard.py
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
import time
from monitor_client import HolySheepMonitor, TokenUsage
def initialize_session_state():
"""Initialise les états de session Streamlit"""
if 'monitor' not in st.session_state:
st.session_state.monitor = HolySheepMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
if 'usage_data' not in st.session_state:
st.session_state.usage_data = []
if 'alerts' not in st.session_state:
st.session_state.alerts = []
def render_metrics_cards(total_cost: float, total_requests: int, avg_cost: float):
"""Affiche les métriques principales"""
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric(
label="💰 Coût Total (USD)",
value=f"${total_cost:.2f}",
delta=f"${total_cost * 7.2:.2f} CNY" # Estimation approximative
)
with col2:
st.metric(
label="📊 Requêtes Totales",
value=total_requests,
delta="Aujourd'hui"
)
with col3:
st.metric(
label="💵 Coût Moyen/Requête",
value=f"${avg_cost:.4f}",
delta="vs $0.10 target"
)
with col4:
# Comparaison avecGPT-4.1
gpt_cost = total_cost * (8.0 / 0.42) if total_cost > 0 else 0
st.metric(
label="💡 Économie vs GPT-4.1",
value=f"${gpt_cost - total_cost:.2f}",
delta="18x moins cher avec DeepSeek"
)
def render_cost_chart(usage_data: list):
"""Affiche le graphique d'évolution des coûts"""
if not usage_data:
st.info("Aucune donnée disponible. Lancez des requêtes pour voir les statistiques.")
return
df = pd.DataFrame([
{
"timestamp": u.timestamp,
"cost_usd": u.cost_usd,
"input_tokens": u.input_tokens,
"output_tokens": u.output_tokens,
"model": u.model,
"latency_ms": 250 # Simulation
}
for u in usage_data
])
# Graphique 1: Coût cumulé
fig1 = px.line(
df,
x="timestamp",
y="cost_usd",
color="model",
title="📈 Évolution des Coûts par Modèle",
labels={"cost_usd": "Coût (USD)", "timestamp": "Temps"}
)
fig1.add_trace(go.Scatter(
x=df["timestamp"],
y=df["cost_usd"].cumsum(),
name="Cumulé",
line=dict(dash="dash", color="red")
))
st.plotly_chart(fig1, use_container_width=True)
# Graphique 2: Tokens par modèle
fig2 = px.bar(
df.groupby("model").agg({
"input_tokens": "sum",
"output_tokens": "sum"
}).reset_index(),
x="model",
y=["input_tokens", "output_tokens"],
title="🔢 Distribution des Tokens par Modèle",
barmode="group"
)
st.plotly_chart(fig2, use_container_width=True)
def render_alert_panel():
"""Panneau de configuration des alertes"""
st.sidebar.subheader("⚙️ Configuration des Alertes")
alert_name = st.sidebar.text_input("Nom de l'alerte", value="production_limit")
threshold = st.sidebar.number_input(
"Seuil (USD)",
min_value=1.0,
max_value=10000.0,
value=100.0,
step=10.0
)
if st.sidebar.button("➕ Ajouter Alerte"):
st.session_state.monitor.set_budget_alert(
name=alert_name,
threshold_usd=threshold
)
st.sidebar.success(f"Alerte '{alert_name}' ajoutée: ${threshold}")
# Affichage des alertes actives
st.sidebar.subheader("📋 Alertes Actives")
st.sidebar.info("Visualisez les alertes dans la console du serveur")
def render_demo_requests():
"""Interface de test avec requêtes démo"""
st.subheader("🧪 Test des Requêtes Monitorées")
col1, col2 = st.columns([3, 1])
with col1:
model = st.selectbox(
"Modèle",
["deepseek-v3.2", "gemini-2.5-flash", "claude-sonnet-4.5", "gpt-4.1"],
format_func=lambda x: f"{x} (${HolySheepMonitor.PRICING.get(x, {}).get('input', 0):.2f}/M tok)"
)
prompt = st.text_area(
"Votre prompt",
value="Explique en 3 phrases pourquoi le monitoring des tokens est important.",
height=100
)
with col2:
st.write("**Paramètres**")
max_tokens = st.slider("Max tokens", 100, 2000, 500)
temperature = st.slider("Temperature", 0.0, 1.0, 0.7)
if st.button("🚀 Exécuter avec Monitoring", type="primary"):
if not prompt.strip():
st.warning("Veuillez entrer un prompt")
else:
with st.spinner("Envoi vers HolySheep AI..."):
try:
response = st.session_state.monitor.chat_completion(
model=model,
messages=[
{"role": "system", "content": "Tu es un assistant concis et informatif."},
{"role": "user", "content": prompt}
],
max_tokens=max_tokens,
temperature=temperature
)
# Affichage de la réponse
st.success("✅ Requête réussie!")
# Métriques de la requête
meta = response.get("usage_metadata", {})
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Input Tokens", meta.get("input_tokens", 0))
with col2:
st.metric("Output Tokens", meta.get("output_tokens", 0))
with col3:
st.metric("Coût USD", f"${meta.get('cost_usd', 0):.6f}")
with col4:
st.metric("Latence",
Ressources connexes
Articles connexes