LLM Observability: Logging, Tracing, and Eval in Production

Why LLM Observability Is Different

Traditional application observability focuses on: did the request succeed? How long did it take? LLM observability adds a layer: was the response any good?

A 200 OK from an LLM API tells you the request succeeded. It tells you nothing about whether the response was correct, appropriate, grounded in the context, or aligned with what the user needed. You need a separate quality signal.

The Three Pillars

Logging: Record every LLM interaction - inputs, outputs, latency, token counts, cost, model version. This is the raw material for all other observability.

Tracing: For multi-step flows (RAG pipelines, agents), trace the entire chain - each retrieval, each LLM call, each tool call - as a connected unit. Tracing lets you diagnose which step in a pipeline caused a bad outcome.

Evaluation: Continuously score a sample of production responses on quality dimensions. Alerts you to regressions before users do.

Step 1: Structured Logging

pythonCopy
import json
import time
import uuid
from datetime import datetime
import anthropic

client = anthropic.Anthropic()

def logged_llm_call(
    messages: list[dict],
    system: str = None,
    model: str = "claude-haiku-4-5-20251001",
    **kwargs
) -> dict:
    request_id = str(uuid.uuid4())
    start_time = time.time()

    try:
        response = client.messages.create(
            model=model,
            messages=messages,
            system=system,
            max_tokens=kwargs.get("max_tokens", 1024),
            **{k: v for k, v in kwargs.items() if k != "max_tokens"}
        )
        latency_ms = (time.time() - start_time) * 1000

        log_entry = {
            "request_id": request_id,
            "timestamp": datetime.utcnow().isoformat(),
            "model": model,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "latency_ms": round(latency_ms, 2),
            "cost_usd": estimate_cost(model, response.usage.input_tokens, response.usage.output_tokens),
            "response_text": response.content[0].text,
            "stop_reason": response.stop_reason,
        }
        write_log(log_entry)
        return {"success": True, "response": response, "request_id": request_id}

    except Exception as e:
        latency_ms = (time.time() - start_time) * 1000
        log_entry = {"request_id": request_id, "timestamp": datetime.utcnow().isoformat(),
                     "error": str(e), "latency_ms": round(latency_ms, 2)}
        write_log(log_entry)
        return {"success": False, "error": str(e), "request_id": request_id}

def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    # Update with current pricing
    pricing = {
        "claude-haiku-4-5-20251001": {"input": 0.00025, "output": 0.00125},
        "claude-sonnet-4-6": {"input": 0.003, "output": 0.015},
    }
    p = pricing.get(model, {"input": 0.003, "output": 0.015})
    return round((input_tokens / 1000) * p["input"] + (output_tokens / 1000) * p["output"], 6)

def write_log(entry: dict):
    # In production: write to a structured logging service (Datadog, CloudWatch, etc.)
    print(json.dumps(entry))

Step 2: Distributed Tracing for Multi-Step Flows

pythonCopy
class Trace:
    def __init__(self, trace_id: str = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.spans = []
        self.start_time = time.time()

    def add_span(self, name: str, input_data: dict, output_data: dict, latency_ms: float, metadata: dict = None):
        self.spans.append({
            "name": name,
            "input": input_data,
            "output": output_data,
            "latency_ms": latency_ms,
            "metadata": metadata or {},
        })

    def finish(self) -> dict:
        return {
            "trace_id": self.trace_id,
            "total_latency_ms": round((time.time() - self.start_time) * 1000, 2),
            "spans": self.spans,
        }

def rag_with_tracing(query: str) -> dict:
    trace = Trace()

    # Span 1: Retrieval
    t0 = time.time()
    candidates = retrieve(query, top_k=20)
    trace.add_span("retrieval", {"query": query, "top_k": 20},
                   {"num_results": len(candidates)}, (time.time() - t0) * 1000)

    # Span 2: Reranking
    t0 = time.time()
    top_chunks = rerank(query, candidates, top_k=5)
    trace.add_span("reranking", {"num_candidates": len(candidates)},
                   {"num_results": len(top_chunks)}, (time.time() - t0) * 1000)

    # Span 3: Generation
    t0 = time.time()
    result = generate_response(query, top_chunks)
    trace.add_span("generation", {"query": query, "context_chunks": len(top_chunks)},
                   {"answer_length": len(result["answer"])}, (time.time() - t0) * 1000)

    trace_data = trace.finish()
    write_log({"type": "rag_trace", **trace_data})
    return result

Step 3: Continuous Quality Sampling

pythonCopy
import random

def production_quality_check(request_id: str, query: str, response: str, sample_rate: float = 0.05):
    """Evaluate a random sample of production responses for quality."""
    if random.random() > sample_rate:
        return  # Skip this request

    scores = llm_judge(query, response)
    quality_log = {
        "type": "quality_check",
        "request_id": request_id,
        "timestamp": datetime.utcnow().isoformat(),
        **scores
    }
    write_log(quality_log)

    # Alert if quality drops below threshold
    avg_score = sum(v for k, v in scores.items() if isinstance(v, (int, float))) / len(scores)
    if avg_score < 3.0:
        send_alert(f"Quality below threshold on request {request_id}: {scores}")

Dashboards and Alerts

Log all observability data to a time-series database or logging service. Track:

• P50/P95/P99 latency - alert on P99 spikes
• Daily cost - alert on unexpected increases
• Error rate - alert if > 1%
• Average quality score (from sampled eval) - alert on downward trend over 7 days

Tools that work well: Datadog, Grafana + Prometheus, LangSmith (purpose-built for LLM observability), Weights & Biases (for development and eval tracking).

The Feedback Loop

The best observability creates a feedback loop:

1. Log all interactions
2. Sample and evaluate a fraction
3. Log evaluation scores with the interaction
4. Identify systematic failures (low-scoring query types, input patterns)
5. Add those patterns to the eval dataset
6. Fix the underlying issue (prompt, retrieval, or fine-tuning)
7. Verify the fix improves eval scores

This loop, running continuously, is how production LLM systems improve over time.

What to Practice Next

• Instrument a FastAPI LLM endpoint with OpenTelemetry: emit a span per request that includes token counts, model name, and latency; verify the spans appear in a local Jaeger or Zipkin instance.
• Set up a simple LLM quality monitor: log prompt, response, and a heuristic quality signal (e.g., response length, refusal detection) to a database, then write a query that surfaces the bottom 5% of responses by quality signal.
• Design a sampling strategy for a high-traffic LLM endpoint - decide what percentage of requests to log in full, what metadata to log for all requests, and how you would trigger full logging for anomalous requests.