Logging et observabilité
Mis à jour le 29 juillet 2026
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é. La différence se mesure le jour où un utilisateur signale que « l’assistant est lent depuis ce matin » : sans instrumentation, vous en êtes réduit aux hypothèses ; avec elle, vous constatez en trente secondes que la latence p95 a doublé depuis un changement de prompt qui a triplé la taille du contexte.
Logging structuré de base
Le réflexe à prendre est de mesurer autour de l’appel, pas à l’intérieur. On relève l’horloge avant, on la relit après, et on émet une ligne unique décrivant l’événement complet. Le format JSON n’est pas une coquetterie : il permet à votre agrégateur de filtrer sur status, de tracer duration_ms dans le temps ou de retrouver un response_id précis, ce qu’une phrase rédigée à la main ne permettra jamais. Le bloc except produit exactement la même structure, avec le type et le message de l’exception, puis relance : loguer n’est pas gérer.
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.6-terra") -> 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 !")Remarquez qu’on logue prompt_length et non le prompt lui-même. C’est une décision de conception, pas un oubli : la longueur suffit à corréler avec la latence, sans exporter de données utilisateur vers votre système de logs.
Middleware de logging
Répéter ce bloc dans chaque fonction qui appelle l’API garantit qu’on l’oubliera quelque part. Mieux vaut envelopper le client une fois pour toutes. Le dataclass APIMetrics fixe le contrat de ce qu’on mesure, et ObservableClient remplit cet enregistrement puis l’archive, que l’appel réussisse ou échoue. La méthode rapport transforme ensuite l’historique en indicateurs directement exploitables — taux de succès, latence moyenne, latence p95.
C’est ce dernier chiffre qui compte le plus. Une moyenne de 450 ms peut cacher un utilisateur sur vingt qui attend trois secondes ; le p95 le révèle. En triant les durées et en prenant la valeur au 95e centile, vous décrivez l’expérience réelle du plus mal servi de vos utilisateurs courants.
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.6-terra", "Bonjour !")
obs_client.create("gpt-5.6-terra", "Comment allez-vous ?")
obs_client.create("gpt-5.6-terra", "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
Dans une architecture distribuée, un appel API n’existe jamais seul : il est déclenché par une requête HTTP, elle-même issue d’une action utilisateur, et parfois suivi de deux autres appels. Le trace_id est le fil qui relie ces événements. En le propageant depuis votre couche web — ici user-123-req-456 — vous pouvez reconstituer une session entière dans vos logs à partir d’une seule réclamation client. Deux lignes sont émises, une au départ et une à l’arrivée, ce qui permet de repérer aussi les appels qui n’ont jamais rendu la main.
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.6-terra"),
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]}...")Attention toutefois au prompt_preview : ces cent premiers caractères partent tels quels dans vos logs. Acceptable en développement, à retirer dès que des données réelles circulent.
Exporter vers un système de monitoring
Un historique en mémoire disparaît au redémarrage du processus. Pour surveiller un service dans la durée, il faut exposer les mesures à un outil externe. Le pattern universel repose sur deux familles : les compteurs, qui ne font que croître et se dérivent en taux, et les histogrammes, qui conservent une distribution pour en tirer des centiles. Cette classe est volontairement neutre — il vous suffit de brancher get_metrics sur un endpoint pour que Prometheus vienne le scruter, ou de pousser le dictionnaire vers Datadog.
# 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)