Skip to main content
EN · FR Start for free
Menu
Lesson 15 of 22 · Responses API: OpenAI Fundamentals

Logging and Observability

Lesson 15 of 22 Updated July 29, 2026 Developer
On this page
  1. Logging and Observability
  2. Basic Structured Logging
  3. Logging Middleware
  4. Tracing Calls with Context
  5. Exporting to a Monitoring System
  6. Key Takeaways

Logging and Observability

In production, you need to know exactly what’s happening with your API calls: latency, tokens consumed, errors, and costs. Structured logging and observability give you that visibility. The difference becomes clear the moment a user reports, “The assistant has been slow since this morning.” Without instrumentation, you’re stuck guessing. With it, you can see in thirty seconds that p95 latency doubled after a prompt change that tripled the context size.

Basic Structured Logging

Make it a habit to measure around the call, not inside it. Record the clock before, read it again after, and emit a single line describing the entire event. JSON isn’t just for show: it lets your aggregator filter on status, track duration_ms over time, or look up a specific response_id—something free-form text can never do. The except block produces the exact same structure, including the exception type and message, then re-raises. Logging isn’t handling.

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:
    """API call with full structured logging."""
    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("Hello!")

Notice we log prompt_length, not the prompt itself. That’s a design choice, not an oversight: length is enough to correlate with latency, without exporting user data to your logging system.

Logging Middleware

Repeating this block in every function that calls the API guarantees you’ll forget it somewhere. Better to wrap the client once and for all. The APIMetrics dataclass defines the contract for what we measure, and ObservableClient populates that record and archives it, whether the call succeeds or fails. The rapport method then turns the history into directly actionable metrics—success rate, average latency, p95 latency.

That last number matters most. An average of 450 ms can hide the one in twenty users waiting three seconds; p95 reveals it. By sorting durations and taking the value at the 95th percentile, you describe the real experience of your worst-served active users.

from dataclasses import dataclass, field
from typing import Optional
import time
import json

@dataclass
class APIMetrics:
    """Metrics collected for each call."""
    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:
    """OpenAI client with built-in observability."""

    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:
        """Generate a report of collected metrics."""
        if not self.metrics_history:
            return {"message": "No calls recorded"}

        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)],
        }

# Usage
obs_client = ObservableClient()

obs_client.create("gpt-5.6-terra", "Hello!")
obs_client.create("gpt-5.6-terra", "How are you?")
obs_client.create("gpt-5.6-terra", "Thanks!")

print(json.dumps(obs_client.rapport(), indent=2))
# Result:
# {
#   "total_appels": 3,
#   "erreurs": 0,
#   "taux_succes": "100.0%",
#   "total_tokens": 95,
#   "latence_moyenne_ms": 450,
#   "latence_p95_ms": 520
# }

Tracing Calls with Context

In a distributed architecture, an API call never exists in isolation: it’s triggered by an HTTP request, itself stemming from a user action, and sometimes followed by two more calls. The trace_id is the thread that ties these events together. By propagating it from your web layer—here user-123-req-456—you can reconstruct an entire session in your logs from a single customer complaint. Two lines are emitted, one at the start and one at completion, which also helps spot calls that never returned.

import uuid

class TracingClient:
    """Client with tracing to correlate requests."""

    def __init__(self):
        self.client = OpenAI()

    def create_with_trace(self, prompt: str, trace_id: str = None,
                          **kwargs) -> tuple:
        """Call with trace_id for correlation."""
        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

# Usage with trace_id to correlate in logs
tracing = TracingClient()
response, trace = tracing.create_with_trace(
    "Summarize logging best practices.",
    trace_id="user-123-req-456"
)
print(f"Trace: {trace} | Response: {response.output_text[:80]}...")

Be careful with prompt_preview: those first hundred characters go straight into your logs as-is. Acceptable in development, remove it as soon as real data is flowing.

Exporting to a Monitoring System

An in-memory history disappears when the process restarts. To monitor a service over time, you need to expose metrics to an external tool. The universal pattern relies on two families: counters, which only increase and are derived into rates, and histograms, which retain a distribution to extract percentiles. This class is intentionally neutral—you just need to hook get_metrics up to an endpoint for Prometheus to scrape, or push the dictionary to Datadog.

# Pattern for exporting to Prometheus, Datadog, etc.
class MetricsExporter:
    """Metrics exporter compatible with monitoring systems."""

    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()
# Call exporter.record_success() or exporter.record_error()
# after each API call

Key Takeaways

  • Log every call with: model, tokens, latency, status, response_id
  • Use structured JSON for logs your tools can parse
  • Collect key metrics: success rate, p95 latency, token consumption
  • Add a trace_id to correlate requests in distributed systems
  • Export metrics to your monitoring system (Prometheus, Datadog, etc.)
  • Never log prompt content in production (sensitive data)
← Previous lessonRetry, Exponential Backoff, and Circuit BreakerNext lesson →Monitoring costs and usage