Vous avez déployé votre système d'IA en production. Tout fonctionne parfaitement... jusqu'au moment où un client signale des réponses incohérentes. Vous ouvrez vos logs et découvrez un mur de texte illisible : timestamps incohérents, métadonnées manquantes, erreurs cryptiques. Combien de temps allez-vous perdre à reproduire le problème ?
Dans ce tutoriel, nous allons explorer comment structurer efficacement vos logs de requêtes API IA pour achieve une observabilité maximale. Nous utiliserons HolySheep AI comme fournisseur d'API, qui offre des tarifs compétitifs avec un taux de ¥1=$1 permettant des économies de plus de 85% par rapport aux alternatives, ainsi qu'une latence inférieure à 50ms garantissant des performances optimales en production.
Cas Concret : Le Pic de Trafic du Black Friday
Imaginez une plateforme e-commerce française qui utilise l'IA pour son service client. Pendant le Black Friday, le trafic explode : 10 000 requêtes par minute au lieu des 500 habituelles. Soudain, certains clients reçoivent des recommandations de produits complètement hors contexte. Sans logs structurés, diagnostiquer le problème relève du cauchemar.
Avec une architecture d'observabilité appropriée, vous identifiez en 5 minutes que le problème vient d'un timeout mal configuré provoquant des fallsbacks vers un modèle différent. C'est exactement ce que nous allons construire ensemble.
Architecture de Logging Structuré
Le Format JSON Standardisé
Pour une analyse efficace, chaque log de requête doit contenir des champs standardisés. Voici notre structure de base :
{
"request_id": "req_a1b2c3d4e5f6",
"timestamp": "2026-03-15T14:32:00.123Z",
"service": "ecommerce-recommendations",
"provider": "holysheep",
"model": "gpt-4.1",
"request": {
"messages": [...],
"temperature": 0.7,
"max_tokens": 500
},
"response": {
"status": "success",
"content": "...",
"usage": {
"prompt_tokens": 150,
"completion_tokens": 320,
"total_tokens": 470
}
},
"performance": {
"latency_ms": 245,
"time_to_first_token": 89
},
"metadata": {
"user_id": "usr_12345",
"session_id": "sess_abc123",
"feature_flag": "v2_recommendations"
}
}Implémentation du Logger en Python
import json
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from enum import Enum
class LogLevel(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
class StructuredAIPILogger:
"""
Logger structuré pour les API d'IA avec support HolySheep AI.
Inclut la capture automatique des métriques de performance et d'usage.
"""
def __init__(self, service_name: str, log_file: str = "ai_api_logs.jsonl"):
self.service_name = service_name
self.log_file = log_file
self.logger = logging.getLogger(f"ai_logger_{service_name}")
self.logger.setLevel(logging.DEBUG)
# Configuration du handler fichier JSONL
handler = logging.FileHandler(log_file, encoding='utf-8')
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
# Configuration console pour développement
console = logging.StreamHandler()
console.setLevel(logging.INFO)
self.logger.addHandler(console)
def _generate_request_id(self) -> str:
"""Génère un ID unique pour chaque requête."""
import uuid
return f"req_{uuid.uuid4().hex[:12]}"
def log_request(
self,
model: str,
messages: list,
user_id: Optional[str] = None,
session_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs
) -> str:
"""
Log une requête API avant son exécution.
Retourne l'ID de requête pour correlation ultérieure.
"""
request_id = self._generate_request_id()
log_entry = {
"event_type": "request_started",
"request_id": request_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"service": self.service_name,
"provider": "holysheep",
"model": model,
"request": {
"messages": messages,
**kwargs
},
"metadata": {
"user_id": user_id,
"session_id": session_id,
**(metadata or {})
}
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
return request_id
def log_response(
self,
request_id: str,
status: str,
response_content: Optional[str] = None,
usage: Optional[Dict[str, int]] = None,
latency_ms: Optional[float] = None,
error: Optional[str] = None
):
"""
Log la réponse de l'API avec les métriques de performance.
"""
log_entry = {
"event_type": "request_completed" if status == "success" else "request_failed",
"request_id": request_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"response": {
"status": status,
"content": response_content,
"usage": usage,
"error": error
},
"performance": {
"latency_ms": latency_ms
}
}
if status == "success":
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
else:
self.logger.error(json.dumps(log_entry, ensure_ascii=False))
def log_cost_analysis(
self,
request_id: str,
model: str,
prompt_tokens: int,
completion_tokens: int
):
"""
Analyse et log le coût de la requête.
Prix HolySheep 2026 (USD par million de tokens) :
- GPT-4.1: $8
- Claude Sonnet 4.5: $15
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42
"""
prices = {
"gpt-4.1": {"prompt": 8, "completion": 8},
"claude-sonnet-4.5": {"prompt": 15, "completion": 15},
"gemini-2.5-flash": {"prompt": 2.50, "completion": 2.50},
"deepseek-v3.2": {"prompt": 0.42, "completion": 0.42}
}
model_key = model.lower().replace(" ", "-")
price = prices.get(model_key, {"prompt": 8, "completion": 8})
prompt_cost = (prompt_tokens / 1_000_000) * price["prompt"]
completion_cost = (completion_tokens / 1_000_000) * price["completion"]
total_cost = prompt_cost + completion_cost
log_entry = {
"event_type": "cost_analysis",
"request_id": request_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"cost": {
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"prompt_cost_usd": round(prompt_cost, 6),
"completion_cost_usd": round(completion_cost, 6),
"total_cost_usd": round(total_cost, 6)
}
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
return total_costIntégration avec l'API HolySheep
L'API HolySheep offre une compatibilité OpenAI-style, ce qui facilite l'intégration. Voici comment combiner notre logger avec les appels API réels :
import requests
from openai import OpenAI
class HolySheepAIClient:
"""
Client API HolySheep avec logging automatique et observabilité intégrée.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, service_name: str):
self.client = OpenAI(
api_key=api_key,
base_url=self.BASE_URL
)
self.logger = StructuredAIPILogger(service_name)
def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
user_id: Optional[str] = None,
session_id: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 1000,
metadata: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Effectue un appel de chat completion avec logging complet.
"""
# Log de la requête initiale
request_id = self.logger.log_request(
model=model,
messages=messages,
user_id=user_id,
session_id=session_id,
metadata=metadata,
temperature=temperature,
max_tokens=max_tokens
)
start_time = time.time()
try:
# Appel API
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
latency_ms = (time.time() - start_time) * 1000
# Extraction des données de réponse
response_content = response.choices[0].message.content
usage = {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
# Log de la réponse réussie
self.logger.log_response(
request_id=request_id,
status="success",
response_content=response_content,
usage=usage,
latency_ms=latency_ms
)
# Analyse des coûts
self.logger.log_cost_analysis(
request_id=request_id,
model=model,
prompt_tokens=usage["prompt_tokens"],
completion_tokens=usage["completion_tokens"]
)
return {
"request_id": request_id,
"content": response_content,
"usage": usage,
"latency_ms": latency_ms,
"model": model
}
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
# Log de l'erreur
self.logger.log_response(
request_id=request_id,
status="failed",
latency_ms=latency_ms,
error=str(e)
)
raise
Exemple d'utilisation
if __name__ == "__main__":
# Initialisation du client avec la clé API
ai_client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
service_name="ecommerce-chatbot"
)
# Exemple de requête de service client e-commerce
response = ai_client.chat_completion(
messages=[
{"role": "system", "content": "Vous êtes un assistant client e-commerce français."},
{"role": "user", "content": "Je cherche un cadeau pour mon mari, budget 100€"}
],
model="gpt-4.1",
user_id="usr_78234",
session_id="sess_blackfriday_001",
temperature=0.8,
metadata={"page": "/produits/cadeaux", "referrer": "email_campaign"}
)
print(f"Request ID: {response['request_id']}")
print(f"Latence: {response['latency_ms']:.2f}ms")
print(f"Tokens utilisés: {response['usage']['total_tokens']}")Tableaux de Bord d'Observabilité
Pour transformer vos logs en insights actionnables, configurez un pipeline de visualisation. Voici un exemple de métriques essentielles à surveiller :
- Taux de succès des requêtes : Pourcentage d'appels réussis vs échecs
- Latence P50/P95/P99 : Distribution des temps de réponse
- Coût par heure/requête : Suivi budgétaire en temps réel
- Utilisation des tokens : Ratio prompt/completion par modèle
- Taux d'erreurs par type : Classification des échecs (timeout, rate limit, invalid request)
- Distribution par modèle : Quel modèle utilise quelle proportion du budget
Alertes Automatisées
import asyncio
from dataclasses import dataclass
from typing import Callable
@dataclass
class AlertConfig:
"""Configuration des alertes d'observabilité."""
error_rate_threshold: float = 0.05 # 5% d'erreurs max
latency_p95_threshold_ms: float = 2000 # 2s max pour P95
cost_per_hour_threshold_usd: float = 100 # $100/heure max
class AIAlertManager:
"""
Gestionnaire d'alertes pour la surveillance proactive.
"""
def __init__(self, config: AlertConfig):
self.config = config
self.alerts = []
def check_error_rate(self, logs: list) -> Optional[dict]:
"""Vérifie si le taux d'erreur dépasse le seuil."""
total = len(logs)
errors = sum(1 for log in logs if log.get("response", {}).get("status") == "failed")
if total > 0:
error_rate = errors / total
if error_rate > self.config.error_rate_threshold:
return {
"alert": "HIGH_ERROR_RATE",
"error_rate": f"{error_rate:.2%}",
"threshold": f"{self.config.error_rate_threshold:.2%}",
"action": "Vérifier la connectivité API et les credentials"
}
return None
def check_latency(self, recent_logs: list) -> Optional[dict]:
"""Vérifie si la latence P95 dépasse le seuil."""
latencies = [
log.get("performance", {}).get("latency_ms")
for log in recent_logs
if log.get("performance", {}).get("latency_ms")
]
if len(latencies) >= 20:
latencies.sort()
p95_index = int(len(latencies) * 0.95)
p95_latency = latencies[p95_index]
if p95_latency > self.config.latency_p95_threshold_ms:
return {
"alert": "HIGH_LATENCY_P95",
"p95_latency_ms": f"{p95_latency:.0f}",
"threshold_ms": f"{self.config.latency_p95_threshold_ms:.0f}",
"action": "Envisager un scale-up ou optimisation du prompt"
}
return None
def check_cost(self, hourly_costs: list) -> Optional[dict]:
"""Vérifie si le coût horaire dépasse le budget."""
if hourly_costs:
current_hour_cost = sum(hourly_costs)
if current_hour_cost > self.config.cost_per_hour_threshold_usd:
return {
"alert": "BUDGET_OVERRUN",
"current_cost_usd": f"{current_hour_cost:.2f}",
"threshold_usd": f"{self.config.cost_per_hour_threshold_usd:.2f}",
"action": "Réviser les limites de taux et quotas utilisateur"
}
return None
def run_monitoring_cycle(self, logs: list, hourly_costs: list):
"""Exécute un cycle complet de surveillance."""
alerts = []
# Vérifications
if error_alert := self.check_error_rate(logs):
alerts.append(error_alert)