Implementing LLM Observability: Tracing, Logging, and Monitoring Requests in Production

You've shipped your LLM-powered application. Users are hitting it. And then—something goes wrong. A customer complains that the chatbot gave them confidently wrong information. Your latency spikes to 12 seconds for no obvious reason. Token costs double overnight. You dig through your logs and find... nothing useful. Just raw HTTP responses and a timestamp.

This is the moment every team building with LLMs eventually faces: you've instrumented your traditional application stack just fine, but LLMs introduce a completely different class of failure modes. A function either returns a value or throws an exception—that's easy to observe. But an LLM can return a response that is technically successful (200 OK, tokens generated) while being factually wrong, off-brand, dangerously hallucinated, or economically catastrophic. Standard APM tools weren't built for this.

By the end of this lesson, you'll have a working observability stack built around real LLM workloads. You'll be able to trace individual requests through your entire pipeline, capture the data you actually need for debugging and cost control, set up meaningful alerts, and build dashboards that surface problems before your customers do.

What you'll learn:

• How to structure LLM traces to capture prompt context, chain steps, and model responses end-to-end
• What to log beyond the obvious (and why most teams log the wrong things)
• How to track token usage, latency distributions, and quality signals at scale
• How to integrate OpenTelemetry with LLM-specific tooling like LangSmith and Langfuse
• How to build alerts for the failure modes unique to LLM applications

---

Prerequisites

You should be comfortable with Python and have worked with the OpenAI API or a similar LLM provider before. You should understand what a RAG pipeline is at a conceptual level, and have passing familiarity with concepts like spans, traces, and structured logging. Basic experience with Docker is helpful for the Langfuse section.

---

Why LLM Observability Is Different

Before we write any code, it's worth being precise about what makes LLM observability hard—because if you don't understand the problem space, you'll build the wrong solution.

Traditional observability is about deterministic systems. You instrument a database query, a cache hit/miss, an HTTP call. The inputs and outputs are structured, the latency profile is relatively stable, and a failure is clearly a failure. Your existing tools—Datadog, New Relic, Prometheus—are excellent at this.

LLMs break all three of those assumptions. The inputs (prompts) are unstructured natural language that can vary enormously in length and complexity. The outputs are also unstructured, making automated quality assessment hard. Latency scales with output length in ways that are difficult to predict. And a "successful" response might be completely wrong.

This means LLM observability needs to capture things traditional tools ignore:

• Full prompt content, not just a request hash
• Token-level economics (input tokens vs. output tokens have different costs)
• Chain step attribution in multi-step pipelines
• Semantic quality signals (did the model actually answer the question?)
• Model and version metadata (gpt-4o vs. gpt-4o-mini behave differently)
• User feedback correlation (thumbs up/down tied back to specific traces)

You need all of this for debugging, cost control, and continuous improvement. Let's build it.

---

Setting Up Your Observability Infrastructure

We'll use a layered approach. At the foundation, we'll use structured logging to capture everything we need in a queryable format. On top of that, we'll add distributed tracing using OpenTelemetry to connect spans across complex pipelines. And we'll use Langfuse as our LLM-native observability platform, which handles the domain-specific concerns that general APM tools don't.

First, install the dependencies:

pip install openai langfuse opentelemetry-sdk opentelemetry-exporter-otlp \
    opentelemetry-instrumentation-requests structlog python-dotenv

Set up your environment variables:

# .env
OPENAI_API_KEY=sk-...
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.com  # or your self-hosted URL

Now let's build a foundational logging setup that will carry us through the rest of the lesson:

# observability/logging_config.py
import structlog
import logging
import sys
from datetime import datetime, timezone

def configure_logging(service_name: str, environment: str = "production"):
    """
    Configure structured logging with consistent fields for LLM workloads.
    structlog gives us JSON output that's queryable in any log aggregation system.
    """
    
    # Add standard processors to every log event
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.stdlib.add_log_level,
            structlog.stdlib.add_logger_name,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(sys.stdout)
    )
    
    # Bind global context that appears in every log event
    structlog.contextvars.bind_contextvars(
        service=service_name,
        environment=environment,
    )
    
    return structlog.get_logger()

logger = configure_logging("customer-support-llm", "production")

This gives you structured, queryable JSON logs from the start. Every event will include service name, environment, and a timestamp automatically—no discipline required from individual developers.

---

Building a Trace-Aware LLM Client

The most important architectural decision you'll make is wrapping your LLM calls in a thin client that handles observability concerns uniformly. If you scatter raw openai.chat.completions.create() calls throughout your codebase, you will miss events, have inconsistent logging, and spend days debugging production issues.

Here's a production-grade wrapper that handles tracing, logging, and error capture in one place:

# observability/llm_client.py
import time
import uuid
from typing import Optional, Any
from dataclasses import dataclass, field
from openai import OpenAI
from langfuse import Langfuse
import structlog

logger = structlog.get_logger()

@dataclass
class LLMRequestContext:
    """Carries trace context through a request lifecycle."""
    trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    session_id: Optional[str] = None
    user_id: Optional[str] = None
    feature: Optional[str] = None  # e.g., "support_chat", "doc_summarizer"
    metadata: dict = field(default_factory=dict)


@dataclass  
class LLMResponse:
    """Structured response that includes observability data."""
    content: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    latency_ms: float
    trace_id: str
    finish_reason: str
    cost_usd: Optional[float] = None


# Simple cost table — update this as pricing changes
MODEL_COSTS_PER_1K = {
    "gpt-4o": {"input": 0.0025, "output": 0.010},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
    "gpt-4-turbo": {"input": 0.010, "output": 0.030},
}


def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> Optional[float]:
    if model not in MODEL_COSTS_PER_1K:
        return None
    rates = MODEL_COSTS_PER_1K[model]
    return (input_tokens / 1000 * rates["input"]) + (output_tokens / 1000 * rates["output"])


class ObservableLLMClient:
    """
    A thin wrapper around the OpenAI client that handles tracing and logging
    consistently across all LLM calls in your application.
    """
    
    def __init__(self, openai_client: OpenAI, langfuse_client: Langfuse):
        self.openai = openai_client
        self.langfuse = langfuse_client
    
    def complete(
        self,
        messages: list[dict],
        model: str = "gpt-4o-mini",
        context: Optional[LLMRequestContext] = None,
        span_name: str = "llm_completion",
        **kwargs
    ) -> LLMResponse:
        """
        Make a chat completion with full observability.
        The `context` carries trace IDs and user info through the call.
        """
        
        if context is None:
            context = LLMRequestContext()
        
        # Bind trace context to all log events for this request
        structlog.contextvars.bind_contextvars(
            trace_id=context.trace_id,
            user_id=context.user_id,
            session_id=context.session_id,
            feature=context.feature,
        )
        
        # Start a Langfuse generation span
        generation = self.langfuse.generation(
            name=span_name,
            model=model,
            input=messages,
            metadata={
                "feature": context.feature,
                "user_id": context.user_id,
                **context.metadata
            },
            trace_id=context.trace_id,
            session_id=context.session_id,
            user_id=context.user_id,
        )
        
        logger.info(
            "llm_request_started",
            model=model,
            message_count=len(messages),
            prompt_preview=messages[-1]["content"][:100] if messages else "",
        )
        
        start_time = time.perf_counter()
        
        try:
            response = self.openai.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs
            )
            
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            output_text = response.choices[0].message.content
            usage = response.usage
            finish_reason = response.choices[0].finish_reason
            cost = calculate_cost(model, usage.prompt_tokens, usage.completion_tokens)
            
            # Update the Langfuse span with results
            generation.end(
                output=output_text,
                usage={
                    "input": usage.prompt_tokens,
                    "output": usage.completion_tokens,
                    "total": usage.total_tokens,
                    "unit": "TOKENS"
                },
                metadata={"finish_reason": finish_reason, "cost_usd": cost}
            )
            
            result = LLMResponse(
                content=output_text,
                model=model,
                input_tokens=usage.prompt_tokens,
                output_tokens=usage.completion_tokens,
                total_tokens=usage.total_tokens,
                latency_ms=latency_ms,
                trace_id=context.trace_id,
                finish_reason=finish_reason,
                cost_usd=cost,
            )
            
            logger.info(
                "llm_request_completed",
                model=model,
                input_tokens=usage.prompt_tokens,
                output_tokens=usage.completion_tokens,
                latency_ms=round(latency_ms, 2),
                cost_usd=cost,
                finish_reason=finish_reason,
            )
            
            return result
            
        except Exception as e:
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            generation.end(
                metadata={"error": str(e), "error_type": type(e).__name__}
            )
            
            logger.error(
                "llm_request_failed",
                error=str(e),
                error_type=type(e).__name__,
                latency_ms=round(latency_ms, 2),
                model=model,
            )
            raise
        
        finally:
            # Always clear context variables after the request
            structlog.contextvars.unbind_contextvars(
                "trace_id", "user_id", "session_id", "feature"
            )

Notice a few deliberate design choices here. We log a preview of the prompt (first 100 characters) rather than the full prompt at the INFO level—full prompts belong in your tracing system, not your log aggregator, because they can be enormous. We calculate cost immediately and attach it to the log event so cost queries are fast. And we always clear context variables in a finally block to prevent context leakage between requests in a long-running server.

---

Tracing Multi-Step Pipelines

A single LLM call is easy to observe. A RAG pipeline that retrieves documents, re-ranks them, summarizes each one, and then synthesizes a final answer is where observability gets genuinely hard—and genuinely valuable.

Let's build a realistic RAG pipeline with proper span nesting. The key insight is that each step should be its own span, nested under a parent trace, so you can see exactly where time and cost went.

# pipelines/rag_pipeline.py
import time
from typing import Optional
from langfuse import Langfuse
from observability.llm_client import ObservableLLMClient, LLMRequestContext
import structlog

logger = structlog.get_logger()


class ObservableRAGPipeline:
    """
    A retrieval-augmented generation pipeline with full step-level observability.
    Each stage creates its own span so you can identify bottlenecks precisely.
    """
    
    def __init__(
        self,
        llm_client: ObservableLLMClient,
        langfuse: Langfuse,
        vector_store,  # your actual vector store client
        reranker=None,
    ):
        self.llm = llm_client
        self.langfuse = langfuse
        self.vector_store = vector_store
        self.reranker = reranker
    
    def answer_question(
        self,
        question: str,
        context: Optional[LLMRequestContext] = None,
        top_k: int = 5,
    ) -> dict:
        """
        Full RAG pipeline with observability at each step.
        Returns the answer plus complete trace metadata.
        """
        
        if context is None:
            context = LLMRequestContext()
        
        # Create the root trace for this entire pipeline run
        trace = self.langfuse.trace(
            name="rag_question_answering",
            input={"question": question, "top_k": top_k},
            user_id=context.user_id,
            session_id=context.session_id,
            trace_id=context.trace_id,
            metadata={"feature": context.feature},
        )
        
        pipeline_start = time.perf_counter()
        
        try:
            # Step 1: Retrieve candidate documents
            retrieved_docs = self._retrieve(
                question=question,
                top_k=top_k,
                trace=trace,
            )
            
            # Step 2: Rerank (if available)
            if self.reranker:
                ranked_docs = self._rerank(
                    question=question,
                    documents=retrieved_docs,
                    trace=trace,
                )
            else:
                ranked_docs = retrieved_docs
            
            # Step 3: Build context and generate answer
            answer_response = self._generate_answer(
                question=question,
                documents=ranked_docs[:3],  # Use top 3 after reranking
                context=context,
                trace=trace,
            )
            
            total_latency_ms = (time.perf_counter() - pipeline_start) * 1000
            
            result = {
                "answer": answer_response.content,
                "trace_id": context.trace_id,
                "sources": [doc["source"] for doc in ranked_docs[:3]],
                "total_latency_ms": round(total_latency_ms, 2),
                "cost_usd": answer_response.cost_usd,
            }
            
            # Update the root trace with the final output
            trace.update(
                output={"answer": answer_response.content, "source_count": len(ranked_docs[:3])},
                metadata={"total_latency_ms": total_latency_ms, "cost_usd": answer_response.cost_usd}
            )
            
            logger.info(
                "rag_pipeline_completed",
                trace_id=context.trace_id,
                total_latency_ms=round(total_latency_ms, 2),
                retrieved_count=len(retrieved_docs),
                final_source_count=len(ranked_docs[:3]),
                cost_usd=answer_response.cost_usd,
            )
            
            return result
            
        except Exception as e:
            trace.update(metadata={"error": str(e), "error_type": type(e).__name__})
            logger.error("rag_pipeline_failed", error=str(e), trace_id=context.trace_id)
            raise
    
    def _retrieve(self, question: str, top_k: int, trace) -> list[dict]:
        """Retrieval step with its own span."""
        span = trace.span(name="vector_retrieval", input={"query": question, "top_k": top_k})
        start = time.perf_counter()
        
        try:
            docs = self.vector_store.similarity_search(question, k=top_k)
            latency_ms = (time.perf_counter() - start) * 1000
            
            span.end(
                output={"retrieved_count": len(docs)},
                metadata={"latency_ms": latency_ms}
            )
            
            logger.info(
                "retrieval_completed",
                retrieved_count=len(docs),
                latency_ms=round(latency_ms, 2),
            )
            
            return docs
            
        except Exception as e:
            span.end(metadata={"error": str(e)})
            raise
    
    def _rerank(self, question: str, documents: list[dict], trace) -> list[dict]:
        """Reranking step with its own span."""
        span = trace.span(
            name="document_reranking",
            input={"query": question, "candidate_count": len(documents)}
        )
        start = time.perf_counter()
        
        try:
            reranked = self.reranker.rerank(question, documents)
            latency_ms = (time.perf_counter() - start) * 1000
            span.end(output={"reranked_count": len(reranked)}, metadata={"latency_ms": latency_ms})
            return reranked
        except Exception as e:
            span.end(metadata={"error": str(e)})
            raise
    
    def _generate_answer(self, question: str, documents: list[dict], context, trace) -> object:
        """Answer generation step — this calls the LLM."""
        context_text = "\n\n".join([
            f"Source: {doc['source']}\n{doc['content']}"
            for doc in documents
        ])
        
        messages = [
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant. Answer the user's question using only "
                    "the provided context. If the context doesn't contain enough information "
                    "to answer confidently, say so explicitly."
                )
            },
            {
                "role": "user",
                "content": f"Context:\n{context_text}\n\nQuestion: {question}"
            }
        ]
        
        return self.llm.complete(
            messages=messages,
            model="gpt-4o-mini",
            context=context,
            span_name="answer_generation",
        )

Tip: The nesting of spans matters. When you open Langfuse and look at a trace, you'll see rag_question_answering as the root, with vector_retrieval, document_reranking, and answer_generation as children. You can immediately see that retrieval took 340ms, reranking took 85ms, and generation took 1,200ms—instead of staring at a single 1,625ms duration wondering where it went.

---

What to Actually Log (And What Not To)

Most teams get logging wrong in one of two directions: they log too little (a timestamp and a request ID) or they log everything (entire prompts and responses in every log line). Both create problems.

Here's a practical framework for what belongs where:

In your structured logs (Datadog, CloudWatch, etc.):

• Request ID / trace ID
• Model name and version
• Token counts (input, output, total)
• Latency in milliseconds
• Cost in USD
• Finish reason
• Feature/endpoint name
• User ID (hashed or tokenized—never raw PII)
• Error type and message (not stack traces)
• First 100 characters of the prompt (for quick debugging)

In your tracing system (Langfuse, LangSmith):

• Full prompt content
• Full response content
• All intermediate chain steps
• Retrieved document content and scores
• Model parameters (temperature, max_tokens)
• User feedback when collected

In your error tracker (Sentry, Bugsnag):

• Full stack traces
• Exception context
• The request that caused the failure

The reason to separate these isn't pedantry—it's economics and compliance. Log aggregation systems charge per GB ingested. Storing full prompts and responses in CloudWatch for a high-volume application will cost a fortune and slow down your log queries. Your tracing system is built for storing large payloads with fast lookup by trace ID. Keep the cheap data cheap.

# Example: what a good LLM log event looks like in production
{
    "timestamp": "2024-11-15T14:23:07.412Z",
    "level": "info",
    "event": "llm_request_completed",
    "service": "customer-support-llm",
    "environment": "production",
    "trace_id": "a3f2c891-1234-4abc-9def-567890123456",
    "user_id": "usr_hashed_abc123",  # hashed, not raw email
    "session_id": "sess_xyz789",
    "feature": "ticket_classification",
    "model": "gpt-4o-mini",
    "input_tokens": 847,
    "output_tokens": 43,
    "latency_ms": 1243.5,
    "cost_usd": 0.000153,
    "finish_reason": "stop",
    "prompt_preview": "Classify the following support ticket into one of..."
}

---

Capturing Quality Signals

Latency and cost are easy to measure. Quality is hard—and it's the thing that matters most for user outcomes. You need a strategy for surfacing quality problems automatically, because you cannot read every LLM response manually.

We'll use three layers of quality monitoring:

Layer 1: Automated heuristics — cheap, run on every response

# observability/quality_checks.py
import re
from dataclasses import dataclass

@dataclass
class QualitySignals:
    refusal_detected: bool
    length_anomaly: bool      # suspiciously short or long
    repetition_detected: bool
    uncertainty_expressed: bool
    json_parse_failed: bool   # for structured output tasks

def check_response_quality(
    response_text: str,
    expected_format: str = "text",  # or "json"
    min_expected_chars: int = 50,
    max_expected_chars: int = 4000,
) -> QualitySignals:
    """
    Run cheap heuristic checks on every LLM response.
    These aren't perfect, but they catch the obvious failures at scale.
    """
    
    refusal_phrases = [
        "i cannot", "i can't", "i'm unable to", "i am unable to",
        "i won't", "i will not", "as an ai, i", "i don't have the ability"
    ]
    refusal_detected = any(
        phrase in response_text.lower() for phrase in refusal_phrases
    )
    
    length_anomaly = (
        len(response_text) < min_expected_chars or
        len(response_text) > max_expected_chars
    )
    
    # Detect obvious repetition loops (a common LLM failure mode)
    words = response_text.split()
    if len(words) > 20:
        first_quarter = " ".join(words[:len(words)//4])
        second_quarter = " ".join(words[len(words)//4:len(words)//2])
        repetition_detected = first_quarter == second_quarter
    else:
        repetition_detected = False
    
    uncertainty_phrases = [
        "i'm not sure", "i don't know", "i cannot confirm",
        "you may want to verify", "i'm uncertain"
    ]
    uncertainty_expressed = any(
        phrase in response_text.lower() for phrase in uncertainty_phrases
    )
    
    json_parse_failed = False
    if expected_format == "json":
        try:
            import json
            json.loads(response_text)
        except (json.JSONDecodeError, ValueError):
            json_parse_failed = True
    
    return QualitySignals(
        refusal_detected=refusal_detected,
        length_anomaly=length_anomaly,
        repetition_detected=repetition_detected,
        uncertainty_expressed=uncertainty_expressed,
        json_parse_failed=json_parse_failed,
    )

Layer 2: User feedback collection — attach thumbs up/down to traces

# observability/feedback.py
from langfuse import Langfuse

def record_user_feedback(
    langfuse: Langfuse,
    trace_id: str,
    rating: int,  # 1 for positive, 0 for negative
    comment: Optional[str] = None,
):
    """
    Record explicit user feedback tied to a specific trace.
    This is gold for identifying systematic quality problems.
    """
    langfuse.score(
        trace_id=trace_id,
        name="user_feedback",
        value=rating,
        comment=comment,
    )

Layer 3: LLM-as-judge — run periodically on a sample of responses

# observability/llm_judge.py

JUDGE_PROMPT = """You are evaluating the quality of an AI assistant's response.

Original Question: {question}
AI Response: {response}

Rate the response on three dimensions (each 1-5):
- Accuracy: Does it appear factually correct and free of obvious errors?
- Helpfulness: Does it actually address what the user asked?
- Safety: Does it avoid harmful, biased, or inappropriate content?

Return a JSON object with keys: accuracy, helpfulness, safety, reasoning
"""

def judge_response_quality(
    llm_client: ObservableLLMClient,
    question: str,
    response: str,
    trace_id: str,
    langfuse: Langfuse,
):
    """
    Use a separate LLM call to evaluate response quality.
    Run this on a 5-10% sample in production to avoid doubling your costs.
    """
    import json, random
    
    if random.random() > 0.05:  # Sample 5%
        return None
    
    judge_messages = [
        {"role": "user", "content": JUDGE_PROMPT.format(
            question=question, response=response
        )}
    ]
    
    judge_context = LLMRequestContext(feature="quality_evaluation")
    
    judge_response = llm_client.complete(
        messages=judge_messages,
        model="gpt-4o-mini",
        context=judge_context,
        span_name="llm_judge",
        response_format={"type": "json_object"}
    )
    
    try:
        scores = json.loads(judge_response.content)
        
        for dimension, value in scores.items():
            if dimension in ("accuracy", "helpfulness", "safety"):
                langfuse.score(
                    trace_id=trace_id,
                    name=f"judge_{dimension}",
                    value=float(value),
                )
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        logger.warning("judge_parse_failed", error=str(e))

Warning: LLM-as-judge is powerful but expensive and slightly circular (you're using an LLM to evaluate an LLM). Keep it to a sample. Use it to identify trends and systematic failures, not to make per-request decisions.

---

Setting Up Metrics and Alerts

Logs and traces are great for debugging individual incidents. Metrics are how you catch systemic problems before they become incidents. Here's how to instrument the metrics that actually matter for LLM applications.

# observability/metrics.py
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
import threading
import time

@dataclass
class LLMMetricsCollector:
    """
    A simple in-process metrics collector.
    In production, replace the `_emit` calls with your actual metrics system
    (Prometheus, StatsD, CloudWatch metrics, etc.)
    """
    
    def record_request(
        self,
        model: str,
        feature: str,
        latency_ms: float,
        input_tokens: int,
        output_tokens: int,
        cost_usd: Optional[float],
        success: bool,
        finish_reason: str,
    ):
        tags = {"model": model, "feature": feature}
        
        # Latency histogram
        self._emit_histogram("llm.request.latency_ms", latency_ms, tags)
        
        # Token counters
        self._emit_counter("llm.tokens.input", input_tokens, tags)
        self._emit_counter("llm.tokens.output", output_tokens, tags)
        
        # Cost counter (watch this one closely)
        if cost_usd:
            self._emit_counter("llm.cost.usd", cost_usd, tags)
        
        # Success rate
        self._emit_counter(
            "llm.requests.total",
            1,
            {**tags, "success": str(success), "finish_reason": finish_reason}
        )
    
    def _emit_histogram(self, name: str, value: float, tags: dict):
        # Replace with: statsd.histogram(name, value, tags=tags)
        # Or: prometheus_histogram.labels(**tags).observe(value)
        pass
    
    def _emit_counter(self, name: str, value: float, tags: dict):
        # Replace with: statsd.increment(name, value, tags=tags)
        pass

For alerting, define thresholds that make sense for your specific application. Here are the alerts worth setting up for most LLM applications:

# Example alert configurations — adapt these thresholds to your baseline

ALERT_CONFIGS = {
    "high_error_rate": {
        "metric": "llm.requests.total{success:false}",
        "condition": "rate > 5% over 5 minutes",
        "severity": "critical",
        "notify": ["oncall-slack", "pagerduty"],
        "message": "LLM error rate exceeded 5%. Check for API outages or quota issues."
    },
    
    "p95_latency_spike": {
        "metric": "llm.request.latency_ms p95",
        "condition": "p95 > 8000ms over 10 minutes",
        "severity": "warning",
        "notify": ["engineering-slack"],
        "message": "P95 LLM latency above 8s. Check for output token length increases or API degradation."
    },
    
    "cost_spike": {
        "metric": "llm.cost.usd",
        "condition": "hourly_rate > 3x rolling_7day_average",
        "severity": "critical",
        "notify": ["oncall-slack", "engineering-manager"],
        "message": "LLM cost spiking. Possible prompt injection, runaway loop, or traffic anomaly."
    },
    
    "finish_reason_length": {
        "metric": "llm.requests.total{finish_reason:length}",
        "condition": "rate > 10% over 15 minutes",
        "severity": "warning",
        "notify": ["engineering-slack"],
        "message": "High rate of responses cut off by max_token limit. Check if max_tokens needs adjustment."
    },
    
    "refusal_rate": {
        "metric": "llm.quality.refusals",
        "condition": "rate > 15% over 30 minutes",
        "severity": "warning",
        "notify": ["product-slack"],
        "message": "High refusal rate. Check if recent prompt changes are triggering safety filters."
    }
}

The cost spike alert deserves special attention. Cost doubling overnight is a real failure mode—it can happen because a bug introduced an infinite loop, because a prompt change dramatically increased output lengths, or because of a prompt injection attack that's causing the model to generate enormous responses. An alert that fires at 3x your rolling average will catch all of these.

---

Hands-On Exercise

Now let's put it all together. Your task is to build an observable customer support ticket classifier that you can deploy and actually monitor.

The scenario: You're building a ticket classification system that categorizes incoming support tickets into: billing, technical_support, account_management, feature_request, or complaint. Each ticket needs to be classified with a confidence level, and you need to be able to monitor classification quality over time.

Step 1: Set up Langfuse

Sign up at cloud.langfuse.com and create a project. Grab your public and secret keys. For self-hosted, run:

docker run -d \
  -e DATABASE_URL=postgresql://langfuse:password@postgres:5432/langfuse \
  -p 3000:3000 \
  langfuse/langfuse:latest

Step 2: Build the classifier

# ticket_classifier.py
import json
from openai import OpenAI
from langfuse import Langfuse
from observability.llm_client import ObservableLLMClient, LLMRequestContext
from observability.quality_checks import check_response_quality

openai_client = OpenAI()
langfuse = Langfuse()
llm = ObservableLLMClient(openai_client, langfuse)

CLASSIFIER_SYSTEM_PROMPT = """You are a customer support ticket classifier for a SaaS product.

Classify the ticket into exactly one of these categories:
- billing: payment issues, invoices, subscription changes, refunds
- technical_support: bugs, errors, performance issues, integration problems  
- account_management: login issues, password resets, user permissions, account settings
- feature_request: requests for new features or enhancements
- complaint: general dissatisfaction, negative feedback about product or service

Return a JSON object with:
- category: one of the five categories above
- confidence: a float between 0.0 and 1.0
- reasoning: one sentence explaining your classification

Return only valid JSON, no other text."""


def classify_ticket(
    ticket_text: str,
    ticket_id: str,
    user_id: Optional[str] = None,
) -> dict:
    
    context = LLMRequestContext(
        user_id=user_id,
        feature="ticket_classification",
        metadata={"ticket_id": ticket_id}
    )
    
    messages = [
        {"role": "system", "content": CLASSIFIER_SYSTEM_PROMPT},
        {"role": "user", "content": f"Ticket:\n{ticket_text}"}
    ]
    
    response = llm.complete(
        messages=messages,
        model="gpt-4o-mini",
        context=context,
        span_name="ticket_classification",
        response_format={"type": "json_object"},
        temperature=0.1,  # Low temperature for consistent classification
    )
    
    quality = check_response_quality(response.content, expected_format="json")
    
    if quality.json_parse_failed:
        logger.error(
            "classification_parse_failed",
            trace_id=context.trace_id,
            ticket_id=ticket_id,
            response_preview=response.content[:200],
        )
        raise ValueError(f"Classifier returned invalid JSON for ticket {ticket_id}")
    
    result = json.loads(response.content)
    result["trace_id"] = context.trace_id
    result["ticket_id"] = ticket_id
    
    # Log a quality signal if confidence is low
    if result.get("confidence", 1.0) < 0.6:
        langfuse.score(
            trace_id=context.trace_id,
            name="low_confidence",
            value=result["confidence"],
            comment=f"Classifier uncertain about ticket: {ticket_text[:100]}"
        )
        logger.warning(
            "low_confidence_classification",
            trace_id=context.trace_id,
            ticket_id=ticket_id,
            confidence=result["confidence"],
            category=result.get("category"),
        )
    
    return result


# Test it
if __name__ == "__main__":
    test_tickets = [
        ("TKT-001", "I was charged twice for my Pro plan this month. Please help."),
        ("TKT-002", "The API returns a 500 error when I try to upload files larger than 10MB."),
        ("TKT-003", "Would love to see dark mode added to the dashboard."),
        ("TKT-004", "I can't log in — it says my password is wrong but I just reset it."),
    ]
    
    for ticket_id, ticket_text in test_tickets:
        result = classify_ticket(ticket_text, ticket_id, user_id="test_user_001")
        print(f"\n{ticket_id}: {result['category']} (confidence: {result['confidence']})")
        print(f"Reasoning: {result['reasoning']}")
        print(f"Trace ID: {result['trace_id']}")

Step 3: Open Langfuse and explore the traces

After running the test, navigate to your Langfuse project and look at the Traces view. Click into one trace. You'll see the generation span with the full prompt, the response, token counts, latency, and any scores you attached. Try filtering traces by feature=ticket_classification and sorting by latency descending.

Step 4: Add a dashboard query

In your log aggregation system, build a query for classification distribution:

-- If you're using CloudWatch Logs Insights:
fields @timestamp, feature, cost_usd, input_tokens, output_tokens, latency_ms
| filter feature = "ticket_classification"
| stats 
    count() as request_count,
    avg(latency_ms) as avg_latency,
    sum(cost_usd) as total_cost,
    avg(input_tokens) as avg_input_tokens
  by bin(1h)
| sort @timestamp desc

---

Common Mistakes & Troubleshooting

Mistake 1: Logging PII in plain text

Prompts often contain user data—names, emails, account details. If you're logging full prompts to CloudWatch or Datadog, you might be writing PII to systems that don't have appropriate retention or access controls. Hash or tokenize user identifiers before logging. For Langfuse, configure it as a data processor under your privacy policies.

Mistake 2: Missing the finish_reason field

A finish_reason of length means the model ran out of tokens and stopped mid-response. This is a real quality failure that's invisible if you only look at the response text—the content looks fine, it's just truncated. Always log and alert on finish_reason.

Mistake 3: Not propagating trace context across async boundaries

In async Python applications or when using task queues (Celery, etc.), you need to explicitly pass your trace_id across boundaries. Python's contextvars are not automatically propagated to worker processes. Pass the trace ID as an explicit parameter to any background task that should be part of the same trace.

Mistake 4: Using wall-clock time for latency comparisons

LLM latency scales with output length. A request that generated 50 tokens and took 800ms is very different from one that generated 800 tokens and took 800ms. Track both raw latency and tokens-per-second when investigating performance:

tokens_per_second = (output_tokens / latency_ms) * 1000
logger.info("llm_throughput", tokens_per_second=round(tokens_per_second, 2))

Mistake 5: Alerting on averages instead of percentiles

Average LLM latency is a useless metric because the distribution is heavily right-skewed. A single 30-second request drags the average up while P50 looks fine. Always alert on P95 or P99 latency. If your metrics system doesn't support percentile computation, use a histogram metric type.

Mistake 6: Forgetting to flush before process exit

Langfuse (and OpenTelemetry exporters) batch their data before sending. If your process exits before the batch flushes, you'll lose the last few traces. Always call langfuse.flush() at the end of a script or in your application shutdown handler:

import atexit
atexit.register(langfuse.flush)

---

Summary & Next Steps

You now have a complete observability stack for LLM applications: structured logging that captures cost, latency, and quality signals on every request; distributed tracing that connects spans across complex multi-step pipelines; quality monitoring through heuristics, user feedback, and LLM-as-judge sampling; and a set of meaningful alerts for the failure modes that actually matter.

The architecture you've built here—a thin wrapper around your LLM client, consistent trace context propagation, and separation of logs from traces—will scale from a single-developer prototype to a production system handling millions of requests. The investment in observability pays for itself the first time you debug a production issue in 20 minutes instead of 3 days.

Where to go next:

• Prompt version management: Use Langfuse's prompt management features to version your prompts and A/B test them, correlating prompt versions with quality scores in your traces.
• Cost optimization: Use your token-level data to identify features where you're over-engineering prompts. Often you can cut context length significantly with no quality impact.
• Evaluation datasets: Export traces where users gave negative feedback and build an offline evaluation dataset. Run this against new model versions before deploying.
• OpenTelemetry integration: If your LLM application is part of a larger microservices system, connect your LLM traces to your existing OpenTelemetry pipeline using the W3C Trace Context headers for full end-to-end visibility.
• Anomaly detection: Once you have 2-4 weeks of baseline metrics, implement statistical anomaly detection on your cost and quality metrics rather than fixed thresholds.