LLM Observability in Production — Tracing Every Token From Request to Response

Introduction

When your LLM application hits production, visibility becomes survival. A single slow model call cascades into timeouts, an unexpected cost spike destroys margins, and tokens disappear into a black box with no way to correlate which user triggered which LLM call. Production-grade LLM observability requires structured tracing, cost attribution, and anomaly detection — not just error logs.

What to Measure in an LLM System

Before picking tools, define the four dimensions of LLM observability:

Latency: Time to first token (TTFT), total response time, p50/p95/p99 per endpoint
Tokens: Input and output counts per request, per user, per feature
Cost: USD per request, per user, per day — attributed to specific features
Quality: Response relevance, hallucination rate, user satisfaction proxy signals

Most teams instrument latency well and completely miss cost attribution and quality scoring.

OpenTelemetry Spans for LLM Calls

Wrap every LLM API call in an OpenTelemetry span with semantic attributes. This enables correlation with traces from databases, queues, and downstream services:

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
import anthropic
import time
 
tracer = trace.get_tracer("llm-service")
 
def call_llm_with_tracing(prompt: str, user_id: str, model: str = "claude-3-5-sonnet-20241022") -> dict:
    with tracer.start_as_current_span("llm.completion") as span:
        span.set_attributes({
            "llm.model": model,
            "llm.user_id": user_id,
            "llm.prompt_length": len(prompt),
        })
        start = time.time()
        try:
            client = anthropic.Anthropic()
            response = client.messages.create(
                model=model,
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            )
            latency_ms = int((time.time() - start) * 1000)
            input_tokens  = response.usage.input_tokens
            output_tokens = response.usage.output_tokens
            # Claude Sonnet 3.5 pricing
            cost_usd = (input_tokens * 0.003 + output_tokens * 0.015) / 1000
 
            span.set_attributes({
                "llm.input_tokens":  input_tokens,
                "llm.output_tokens": output_tokens,
                "llm.total_tokens":  input_tokens + output_tokens,
                "llm.cost_usd":      cost_usd,
                "llm.latency_ms":    latency_ms,
                "llm.finish_reason": response.stop_reason,
            })
            span.set_status(Status(StatusCode.OK))
            return {
                "text": response.content[0].text,
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "cost_usd": cost_usd,
                "latency_ms": latency_ms,
            }
        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

Export spans to Jaeger, Grafana Tempo, Datadog, or Honeycomb using the OTLP exporter.

Trace Correlation Across Microservices

The power of observability appears when you correlate a user session with every LLM call, tool execution, and database query it triggered. Propagate a request ID through all service calls:

import uuid
from opentelemetry import trace, context
 
def create_request_context(user_id: str, session_id: str) -> dict:
    return {
        "request_id": str(uuid.uuid4()),
        "user_id": user_id,
        "session_id": session_id,
    }
 
def process_user_request(ctx: dict, query: str) -> str:
    with tracer.start_as_current_span("user_request") as span:
        span.set_attributes({
            "request.id":  ctx["request_id"],
            "user.id":     ctx["user_id"],
            "session.id":  ctx["session_id"],
        })
        # LLM call inherits the span context automatically
        llm_result = call_llm_with_tracing(query, ctx["user_id"])
 
        # Tool calls, database queries — all appear in the same trace
        tool_result = call_tool_with_context(ctx, llm_result["text"])
        span.add_event("tools_executed")
        return tool_result
 
def call_tool_with_context(ctx: dict, instruction: str) -> str:
    with tracer.start_as_current_span("tool_execution") as span:
        span.set_attributes({"request.id": ctx["request_id"]})
        # All tool spans share the same trace — visible in waterfall view
        return f"Tool result for: {instruction[:50]}"

When debugging a user complaint, search by request.id in your tracing backend to see every span — LLM, tool, database — in chronological order.

Adaptive Sampling for High-Volume Systems

At 1,000 requests/second, tracing everything costs $50k+/month in ingestion fees. Sample intelligently: always capture errors and high-cost requests; sample normal traffic at 5–10%:

import random
 
class AdaptiveLLMSampler:
    def __init__(
        self,
        default_rate: float = 0.05,      # 5% of normal requests
        error_rate: float = 1.0,          # 100% of errors
        slow_threshold_ms: int = 5000,    # Always trace slow calls
        slow_rate: float = 1.0,
        cost_threshold_usd: float = 0.10, # Always trace expensive calls
    ):
        self.default_rate      = default_rate
        self.error_rate        = error_rate
        self.slow_threshold_ms = slow_threshold_ms
        self.slow_rate         = slow_rate
        self.cost_threshold    = cost_threshold_usd
 
    def should_sample(self, attributes: dict) -> bool:
        # Always sample errors
        if attributes.get("error"):
            return random.random() < self.error_rate
 
        # Always sample slow requests
        latency = attributes.get("llm.latency_ms", 0)
        if latency > self.slow_threshold_ms:
            return random.random() < self.slow_rate
 
        # Always sample expensive requests
        cost = attributes.get("llm.cost_usd", 0)
        if cost > self.cost_threshold:
            return True
 
        # Default sampling
        return random.random() < self.default_rate

Cost Attribution Per User and Feature

Knowing total LLM spend is insufficient — you need to know which users and features drive cost:

from collections import defaultdict
from datetime import datetime
 
class CostAttributor:
    def __init__(self):
        self._records: list[dict] = []
 
    # Pricing per 1K tokens
    PRICING = {
        "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
        "claude-3-haiku-20240307":    {"input": 0.00025, "output": 0.00125},
    }
 
    def record(self, user_id: str, feature_id: str, model: str, input_tok: int, output_tok: int):
        rates = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (input_tok * rates["input"] + output_tok * rates["output"]) / 1000
        self._records.append({
            "user_id": user_id,
            "feature_id": feature_id,
            "model": model,
            "cost_usd": cost,
            "ts": datetime.utcnow(),
        })
 
    def top_users_by_cost(self, limit: int = 10, days: int = 30) -> list[dict]:
        cutoff = datetime.utcnow().timestamp() - days * 86400
        by_user: dict[str, float] = defaultdict(float)
        for r in self._records:
            if r["ts"].timestamp() > cutoff:
                by_user[r["user_id"]] += r["cost_usd"]
        ranked = sorted(by_user.items(), key=lambda x: x[1], reverse=True)
        return [{"user_id": u, "cost_usd": round(c, 4)} for u, c in ranked[:limit]]
 
    def cost_by_feature(self, days: int = 30) -> dict[str, float]:
        cutoff = datetime.utcnow().timestamp() - days * 86400
        by_feature: dict[str, float] = defaultdict(float)
        for r in self._records:
            if r["ts"].timestamp() > cutoff:
                by_feature[r["feature_id"]] += r["cost_usd"]
        return {k: round(v, 4) for k, v in by_feature.items()}

Anomaly Detection on Latency and Cost

Automate alerting when metrics deviate from baseline. Use z-score detection on a rolling window:

import math
from collections import deque
 
class LLMAnomalyDetector:
    def __init__(self, window_size: int = 100, z_threshold: float = 3.0):
        self.window_size   = window_size
        self.z_threshold   = z_threshold
        self._latencies    = deque(maxlen=window_size)
        self._costs        = deque(maxlen=window_size)
 
    def _z_score(self, values: deque, new_value: float) -> float:
        if len(values) < 10:
            return 0.0
        mean = sum(values) / len(values)
        variance = sum((v - mean) ** 2 for v in values) / len(values)
        stddev = math.sqrt(variance) if variance > 0 else 0.0
        if stddev == 0:
            return 0.0
        return abs(new_value - mean) / stddev
 
    def check_latency(self, latency_ms: float) -> bool:
        """Returns True if latency is anomalous."""
        is_anomaly = self._z_score(self._latencies, latency_ms) > self.z_threshold
        self._latencies.append(latency_ms)
        return is_anomaly
 
    def check_cost(self, cost_usd: float) -> bool:
        """Returns True if cost is anomalous (e.g., prompt injection causing huge output)."""
        is_anomaly = self._z_score(self._costs, cost_usd) > self.z_threshold
        self._costs.append(cost_usd)
        return is_anomaly
 
detector = LLMAnomalyDetector()
 
def handle_llm_response(result: dict) -> None:
    if detector.check_latency(result["latency_ms"]):
        print(f"[ALERT] Latency anomaly: {result['latency_ms']}ms")
    if detector.check_cost(result["cost_usd"]):
        print(f"[ALERT] Cost anomaly: ${result['cost_usd']:.4f}")

Key Takeaways

Wrap every LLM API call in an OpenTelemetry span with llm.model, llm.input_tokens, llm.output_tokens, llm.cost_usd, and llm.latency_ms attributes.
Propagate a request_id through all services so a single user request is traceable across LLM, tool, and database spans.
Adaptive sampling at 5% default, 100% on errors and slow requests, reduces ingestion cost by 90% while preserving signal.
Cost attribution by user_id and feature_id is essential for identifying which features are unprofitable.
Z-score anomaly detection on a 100-request rolling window flags latency and cost spikes within seconds.
Instrument before you need it — adding observability during an incident is too late.
Alert on p95 latency regression (more than 20% above baseline) and cost-per-request spikes (more than 3x baseline).
Export metrics to Grafana dashboards and set up PagerDuty/Opsgenie escalation for critical anomalies.

Introduction#

What to Measure in an LLM System#

OpenTelemetry Spans for LLM Calls#

Trace Correlation Across Microservices#

Adaptive Sampling for High-Volume Systems#

Cost Attribution Per User and Feature#

Anomaly Detection on Latency and Cost#

Key Takeaways#

Related reading