Logging et observabilité
Logging et observabilité
En production, vous devez savoir exactement ce qui se passe avec vos appels API : latence, tokens consommés, erreurs, coûts. Le logging structuré et l’observabilité vous donnent cette visibilité.
Logging structuré de base
import logging
import json
import time
from openai import OpenAI
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger("openai_api")
client = OpenAI()
def appel_avec_logging(prompt: str, model: str = "gpt-5.3") -> str:
"""Appel API avec logging structuré complet."""
start = time.monotonic()
request_id = None
try:
response = client.responses.create(
model=model,
input=prompt
)
duree = time.monotonic() - start
logger.info(json.dumps({
"event": "api_call",
"status": "success",
"model": response.model,
"response_id": response.id,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"total_tokens": response.usage.total_tokens,
"duration_ms": round(duree * 1000),
"prompt_length": len(prompt),
}))
return response.output_text
except Exception as e:
duree = time.monotonic() - start
logger.error(json.dumps({
"event": "api_call",
"status": "error",
"error_type": type(e).__name__,
"error_message": str(e),
"model": model,
"duration_ms": round(duree * 1000),
"prompt_length": len(prompt),
}))
raise
resultat = appel_avec_logging("Bonjour !")Middleware de logging
Pour appliquer le logging automatiquement à tous les appels :
from dataclasses import dataclass, field
from typing import Optional
import time
import json
@dataclass
class APIMetrics:
"""Métriques collectées pour chaque appel."""
model: str = ""
input_tokens: int = 0
output_tokens: int = 0
total_tokens: int = 0
duration_ms: int = 0
status: str = "unknown"
error: Optional[str] = None
response_id: Optional[str] = None
class ObservableClient:
"""Client OpenAI avec observabilité intégrée."""
def __init__(self):
self.client = OpenAI()
self.metrics_history: list[APIMetrics] = []
def create(self, model: str, input: str, **kwargs) -> object:
start = time.monotonic()
metrics = APIMetrics(model=model)
try:
response = self.client.responses.create(
model=model,
input=input,
**kwargs
)
metrics.status = "success"
metrics.response_id = response.id
metrics.input_tokens = response.usage.input_tokens
metrics.output_tokens = response.usage.output_tokens
metrics.total_tokens = response.usage.total_tokens
metrics.duration_ms = round((time.monotonic() - start) * 1000)
self.metrics_history.append(metrics)
return response
except Exception as e:
metrics.status = "error"
metrics.error = str(e)
metrics.duration_ms = round((time.monotonic() - start) * 1000)
self.metrics_history.append(metrics)
raise
def rapport(self) -> dict:
"""Génère un rapport des métriques collectées."""
if not self.metrics_history:
return {"message": "Aucun appel enregistré"}
total_calls = len(self.metrics_history)
erreurs = sum(1 for m in self.metrics_history if m.status == "error")
total_tokens = sum(m.total_tokens for m in self.metrics_history)
durees = [m.duration_ms for m in self.metrics_history]
return {
"total_appels": total_calls,
"erreurs": erreurs,
"taux_succes": f"{((total_calls - erreurs) / total_calls) * 100:.1f}%",
"total_tokens": total_tokens,
"latence_moyenne_ms": round(sum(durees) / len(durees)),
"latence_p95_ms": sorted(durees)[int(len(durees) * 0.95)],
}
# Utilisation
obs_client = ObservableClient()
obs_client.create("gpt-5.3", "Bonjour !")
obs_client.create("gpt-5.3", "Comment allez-vous ?")
obs_client.create("gpt-5.3", "Merci !")
print(json.dumps(obs_client.rapport(), indent=2))
# Résultat :
# {
# "total_appels": 3,
# "erreurs": 0,
# "taux_succes": "100.0%",
# "total_tokens": 95,
# "latence_moyenne_ms": 450,
# "latence_p95_ms": 520
# }Tracer les appels avec un contexte
import uuid
class TracingClient:
"""Client avec tracing pour corréler les requêtes."""
def __init__(self):
self.client = OpenAI()
def create_with_trace(self, prompt: str, trace_id: str = None,
**kwargs) -> tuple:
"""Appel avec trace_id pour corrélation."""
if trace_id is None:
trace_id = str(uuid.uuid4())[:8]
start = time.monotonic()
logger.info(json.dumps({
"event": "api_request_start",
"trace_id": trace_id,
"prompt_preview": prompt[:100],
}))
response = self.client.responses.create(
model=kwargs.get("model", "gpt-5.3"),
input=prompt,
**{k: v for k, v in kwargs.items() if k != "model"}
)
duration = time.monotonic() - start
logger.info(json.dumps({
"event": "api_request_complete",
"trace_id": trace_id,
"response_id": response.id,
"tokens": response.usage.total_tokens,
"duration_ms": round(duration * 1000),
}))
return response, trace_id
# Utilisation avec trace_id pour corréler dans les logs
tracing = TracingClient()
response, trace = tracing.create_with_trace(
"Résumez les bonnes pratiques de logging.",
trace_id="user-123-req-456"
)
print(f"Trace: {trace} | Réponse: {response.output_text[:80]}...")Exporter vers un système de monitoring
# Pattern pour exporter vers Prometheus, Datadog, etc.
class MetricsExporter:
"""Exporteur de métriques compatible avec les systèmes de monitoring."""
def __init__(self):
self.counters = {
"api_calls_total": 0,
"api_errors_total": 0,
"tokens_consumed_total": 0,
}
self.histograms = {
"api_latency_ms": [],
}
def record_success(self, tokens: int, latency_ms: int):
self.counters["api_calls_total"] += 1
self.counters["tokens_consumed_total"] += tokens
self.histograms["api_latency_ms"].append(latency_ms)
def record_error(self):
self.counters["api_calls_total"] += 1
self.counters["api_errors_total"] += 1
def get_metrics(self) -> dict:
latencies = self.histograms["api_latency_ms"]
return {
**self.counters,
"latency_avg_ms": round(sum(latencies) / max(len(latencies), 1)),
"error_rate": self.counters["api_errors_total"] /
max(self.counters["api_calls_total"], 1),
}
exporter = MetricsExporter()
# Appelez exporter.record_success() ou exporter.record_error()
# après chaque appel APIPoints clés à retenir
- Loguez chaque appel avec : modèle, tokens, latence, status, response_id
- Utilisez le JSON structuré pour des logs exploitables par vos outils
- Collectez les métriques clés : taux de succès, latence p95, consommation tokens
- Ajoutez un
trace_idpour corréler les requêtes dans les systèmes distribués - Exportez les métriques vers votre système de monitoring (Prometheus, Datadog, etc.)
- Ne loguez jamais le contenu des prompts en production (données sensibles)