Imaginez le scénario : vous lancez votre assistant IA pour un site e-commerce pendant les soldes du Black Friday. En quelques heures, vous recevez des milliers de requêtes. Subitement, vous constatez des temps de réponse dégradés et des réponses incohérentes. Comment identifier rapidement le problème ? Sans un système d'observabilité robuste, vous êtes aveugle face à ces incidents.
Dans cet article, nous allons construire pas à pas une architecture complète de logging et d'observabilité pour vos intégrations d'API IA, en utilisant HolySheep AI comme fournisseur. Vous apprendrez à capturer, analyser et visualiser vos requêtes pour maintenir des performances optimales.
Pourquoi l'observabilité est essentielle pour les APIs IA
Les APIs d'intelligence artificielle présentent des défis uniques par rapport aux APIs traditionnelles. La latence varie considérablement selon la complexité des prompts, les coûts s'accumulent rapidement avec chaque token généré, et les réponses peuvent être imprévisibles sans监控 approprié.
Un système d'observabilité bien conçu vous permet de :
- Détecter les dégradations de performance en temps réel
- Identifier les prompts problématiques ou trop coûteux
- Optimiser vos dépenses en analysant les patterns d'utilisation
- Déboguer rapidement les échecs de requêtes
Architecture du système d'observabilité
Notre architecture comprendra trois composantes principales : un logger centralisé pour capturer toutes les requêtes, un système de métriques pour les KPIs critiques, et un tableau de bord pour la visualisation. Commençons par implémenter le logger.
Implémentation du Logger de Requêtes
Créons d'abord un logger structuré qui capturera toutes les interactions avec l'API. Ce module enregistrera les timestamps, les prompts, les réponses, les latences et les codes d'erreur.
import json
import time
import logging
from datetime import datetime
from typing import Optional, Dict, Any
from enum import Enum
class LogLevel(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
class AIRequestLogger:
"""Logger structuré pour les requêtes API IA"""
def __init__(self, log_file: str = "ai_requests.log"):
self.log_file = log_file
self.logger = logging.getLogger("AI_Observer")
self.logger.setLevel(logging.DEBUG)
# Configuration du handler fichier
handler = logging.FileHandler(log_file, encoding='utf-8')
handler.setLevel(logging.DEBUG)
# Formatage structuré JSON
formatter = logging.Formatter('%(message)s')
handler.setFormatter(formatter)
if not self.logger.handlers:
self.logger.addHandler(handler)
def log_request(
self,
model: str,
prompt: str,
response: Optional[str],
latency_ms: float,
tokens_used: Optional[int] = None,
error: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None
):
"""Enregistre une requête API avec tous les détails"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": "api_request",
"model": model,
"prompt_length": len(prompt),
"response_length": len(response) if response else 0,
"latency_ms": round(latency_ms, 2),
"tokens_used": tokens_used,
"success": error is None,
"error": error,
"metadata": metadata or {}
}
if error:
self.logger.error(json.dumps(log_entry, ensure_ascii=False))
else:
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
return log_entry
Instance globale du logger
logger = AIRequestLogger()Client API HolySheep avec Intégration d'Observabilité
Maintenant, créons un client enrichi qui interagira avec l'API HolySheep tout en automatiquement journalisant chaque interaction. Le client supportera plusieurs modèles et inclura la gestion automatique des retries.
import requests
import time
from typing import Optional, List, Dict, Any
class HolySheepAIClient:
"""Client API HolySheep avec logging intégré"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, logger: AIRequestLogger):
self.api_key = api_key
self.logger = logger
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000,
metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Envoie une requête de chat completion avec logging"""
start_time = time.time()
prompt = self._format_messages(messages)
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
assistant_message = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
self.logger.log_request(
model=model,
prompt=prompt,
response=assistant_message,
latency_ms=latency_ms,
tokens_used=usage.get("total_tokens"),
metadata=metadata
)
return {
"success": True,
"response": assistant_message,
"usage": usage,
"latency_ms": latency_ms
}
else:
error_msg = f"HTTP {response.status_code}: {response.text}"
self.logger.log_request(
model=model,
prompt=prompt,
response=None,
latency_ms=latency_ms,
error=error_msg,
metadata=metadata
)
return {"success": False, "error": error_msg}
except requests.exceptions.Timeout:
latency_ms = (time.time() - start_time) * 1000
self.logger.log_request(
model=model, prompt=prompt, response=None,
latency_ms=latency_ms, error="Request timeout"
)
return {"success": False, "error": "Request timeout"}
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
self.logger.log_request(
model=model, prompt=prompt, response=None,
latency_ms=latency_ms, error=str(e)
)
return {"success": False, "error": str(e)}
def _format_messages(self, messages: List[Dict[str, str]]) -> str:
"""Formate les messages pour le logging"""
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
Utilisation
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
logger=logger
)Système de Métriques et Alerting
Au-delà du logging, vous avez besoin de métriques agrégées pour comprendre les tendances. Implémentons un système de métriques qui calculera des statistiques en temps réel.
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List
import time
@dataclass
class MetricsSnapshot:
"""Snapshot des métriques à un instant T"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
total_tokens: int = 0
total_cost_usd: float = 0.0
model_usage: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
class MetricsCollector:
"""Collecteur de métriques pour l'observabilité"""
# Tarification HolySheep (USD par million de tokens)
PRICING = {
"gpt-4.1": 8.0, # $8/M tokens
"claude-sonnet-4.5": 15.0, # $15/M tokens
"gemini-2.5-flash": 2.50, # $2.50/M tokens
"deepseek-v3.2": 0.42 # $0.42/M tokens
}
def __init__(self):
self._lock = threading.Lock()
self._snapshot = MetricsSnapshot()
self._latencies: List[float] = []
self._window_seconds = 300 # Fenêtre glissante de 5 minutes
def record_request(
self,
model: str,
latency_ms: float,
tokens_used: int,
success: bool
):
"""Enregistre une requête dans les métriques"""
with self._lock:
self._snapshot.total_requests += 1
self._snapshot.total_latency_ms += latency_ms
self._snapshot.total_tokens += tokens_used
self._snapshot.model_usage[model] += 1
if success:
self._snapshot.successful_requests += 1
else:
self._snapshot.failed_requests += 1
# Calcul du coût
cost = (tokens_used / 1_000_000) * self.PRICING.get(model, 8.0)
self._snapshot.total_cost_usd += cost
# Ajout à la fenêtre de latence
self._latencies.append((time.time(), latency_ms))
self._cleanup_latencies()
def _cleanup_latencies(self):
"""Supprime les données hors fenêtre"""
cutoff = time.time() - self._window_seconds
self._latencies = [(t, l) for t, l in self._latencies if t > cutoff]
def get_stats(self) -> Dict:
"""Retourne les statistiques actuelles"""
with self._lock:
if self._snapshot.total_requests == 0:
return {"error": "Aucune donnée disponible"}
avg_latency = self._snapshot.total_latency_ms / self._snapshot.total_requests
success_rate = (self._snapshot.successful_requests / self._snapshot.total_requests) * 100
# Calcul P95 depuis la fenêtre
if self._latencies:
sorted_latencies = sorted(l for _, l in self._latencies)
p95_idx = int(len(sorted_latencies) * 0.95)
p95_latency = sorted_latencies[p95_idx] if p95_idx < len(sorted_latencies) else avg_latency
else:
p95_latency = avg_latency
return {
"total_requests": self._snapshot.total_requests,
"successful_requests": self._snapshot.successful_requests,
"failed_requests": self._snapshot.failed_requests,
"success_rate_percent": round(success_rate, 2),
"avg_latency_ms": round(avg_latency, 2),
"p95_latency_ms": round(p95_latency, 2),
"total_tokens": self._snapshot.total_tokens,
"total_cost_usd": round(self._snapshot.total_cost_usd, 4),
"model_usage": dict(self._snapshot.model_usage)
}
Instance globale
metrics = MetricsCollector()Tableau de Bord d'Observabilité
Pour visualiser vos métriques, créons un tableau de bord léger utilisant FastAPI et un template HTML. Ce dashboard affichera en temps réel les statistiques de vos APIs IA.
from fastapi import FastAPI, Response
from fastapi.responses import HTMLResponse
import uvicorn
app = FastAPI(title="AI Observability Dashboard")
@app.get("/", response_class=HTMLResponse)
async def dashboard():
"""Sert le tableau de bord HTML"""
html = """
<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8">
<title>AI Observability Dashboard - HolySheep</title>
<style>
body { font-family: -apple-system, sans-serif; padding: 20px; background: #f5f5f5; }
.metric-card { background: white; padding: 20px; border-radius: 8px; margin: 10px; display: inline-block; min-width: 200px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
.metric-value { font-size: 32px; font-weight: bold; color: #2563eb; }
.metric-label { color: #666; margin-top: 5px; }
.success-rate { color: #10b981; }
.error-rate { color: #ef4444; }
.cost { color: #f59e0b; }
h1 { color: #1e293b; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; background: white; }
th, td { padding: 12px; text-align: left; border-bottom: 1px solid #e5e7eb; }
th { background: #f8fafc; font-weight: 600; }
</style>
</head>
<body>
<h1>🔍 AI Observability Dashboard</h1>
<p>Fournisseur: <strong>HolySheep AI</strong> | Latence moyenne: <strong><50ms</strong></p>
<div id="metrics">
<div class="metric-card">
<div class="metric-value" id="total-requests">0</div>
<div class="metric-label">Total Requêtes</div>
</div>
<div class="metric-card">
<div class="metric-value success-rate" id="success-rate">0%</div>
<div class="metric-label">Taux de Succès</div>
</div>
<div class="metric-card">
<div class="metric-value" id="avg-latency">0ms</div>
<div class="metric-label">Latence Moyenne</div>
</div>
<div class="metric-card">
<div class="metric-value" id="p95-latency">0ms</div>
<div class="metric-label">P95 Latence</div>
</div>
<div class="metric-card">
<div class="metric-value cost" id="total-cost">$0.00</div>
<div class="metric-label">Coût Total (USD)</div>
</div>
<div class="metric-card">
<div class="metric-value" id="total-tokens">0</div>
<div class="metric-label">Tokens Utilisés</div>
</div>
</div>
<h2>Utilisation par Modèle</h2>
<table id="model-table">
<thead>
<tr><th>Modèle</th><th>Requêtes</th><th>Prix/M Tokens</th></tr>
</thead>
<tbody id="model-body"></tbody>
</table>
<script>
async function fetchMetrics() {
const resp = await fetch('/api/metrics');
const data = await resp.json();
document.getElement
Ressources connexes
Articles connexes