Back to Blog

Observability in AI: Tracing, Debugging, and Monitoring LLM Applications

8 min read

Monitoring traditional web applications revolves around measuring response codes (200\text{ vs }500), system CPU/memory load, and database query latencies. However, in LLM-powered applications, traditional metrics leave engineers blind.

An agentic pipeline is not a single, linear function. A single user query can trigger multiple vector searches, prompt formatting steps, parallel tool invocations, and multi-turn loops. When a user complains that an agent gave an incorrect response or hung indefinitely, you need tracing to reconstruct the exact flow of execution.

This article details the core metrics of LLM observability and explains how to implement structured tracing.


The Core Pillars of LLM Observability

Observability in AI: Tracing, Debugging, and Monitoring: Observability dashboard visualization showing nested execution spans and trace hierarchies for LLM agentsObservability in AI: Tracing, Debugging, and Monitoring: Observability dashboard visualization showing nested execution spans and trace hierarchies for LLM agents

1. Traces and Spans

Like microservice tracing (using standards like OpenTelemetry), an LLM execution is modeled as a Trace.

  • Trace: The overall execution context of a single user request.
  • Span: A discrete unit of work within that trace. Spans can be nested. For example, a root span might be "Query Processing," containing sub-spans for "Semantic Search," "Prompt Compilation," and "LLM Completion."

Each span should record:

  • Inputs: The exact raw text, prompt variables, and system instructions.
  • Outputs: The model's raw text generation or JSON tool call details.
  • Metadata: The model name, temperature, and API version.

2. Latency Metrics

In generative systems, latency has two components:

  • Time-to-First-Token (TTFT): The duration between the user sending the request and the model generating the first token. This is critical for perceived speed in user interfaces.
  • Inter-Token Latency: The average time required to generate each subsequent token.

3. Token Tracking and Cost Allocation

Because LLMs charge by the token, you must log input and output tokens per request. In multi-tenant systems, tracking these figures allows you to calculate the cost per user, per feature, or per agent run.

4. Semantic Quality (LLM-as-a-Judge)

Unlike traditional assertions, language model outputs are non-deterministic. To monitor accuracy in production, engineers run evaluation models (often smaller, cheaper models) in the background to score responses on metrics such as:

  • Faithfulness: Is the answer derived only from the retrieved context (no hallucinations)?
  • Answer Relevance: Did the model actually answer the user's question?

Standardizing with OpenTelemetry and OpenLLMetry

To avoid vendor lock-in, the community leverages extensions of the OpenTelemetry standard, such as OpenLLMetry (developed by Traceloop). OpenLLMetry automatically hooks into popular libraries (OpenAI, Anthropic, LangChain, Chroma, Pinecone) and exports tracing data to destinations like Langfuse, Arize Phoenix, Dynatrace, or Datadog using standard OpenTelemetry protocols (OTLP).


Python Implementation: A Custom Tracing Framework

Below is a complete Python script demonstrating how to build a custom tracing system using decorators. This script logs nested spans, tracks execution latency, records token consumption, and builds a hierarchical trace graph of an agentic workflow.

import time
import uuid
import functools

class ActiveTrace:
    """Manages the current active trace and its hierarchical spans."""
    def __init__(self):
        self.trace_id = str(uuid.uuid4())
        self.spans = []

    def start_span(self, name: str, parent_id: str = None) -> dict:
        span_id = str(uuid.uuid4())
        span = {
            "span_id": span_id,
            "parent_id": parent_id,
            "name": name,
            "start_time": time.time(),
            "end_time": None,
            "inputs": None,
            "outputs": None,
            "tokens": {"input": 0, "output": 0}
        }
        self.spans.append(span)
        return span

    def end_span(self, span_id: str, inputs=None, outputs=None, tokens=None):
        for span in self.spans:
            if span["span_id"] == span_id:
                span["end_time"] = time.time()
                span["inputs"] = inputs
                span["outputs"] = outputs
                if tokens:
                    span["tokens"].update(tokens)
                break

    def print_trace_tree(self):
        print(f"\n================ Trace ID: {self.trace_id} ================")
        # Map parent-child relationships
        by_parent = {}
        root_spans = []
        for s in self.spans:
            p_id = s["parent_id"]
            if p_id is None:
                root_spans.append(s)
            else:
                by_parent.setdefault(p_id, []).append(s)

        def _print_span(span, indent=0):
            duration = (span["end_time"] - span["start_time"]) * 1000
            tok = span["tokens"]
            token_str = f" | Tokens: {tok['input']} in, {tok['output']} out" if tok["input"] > 0 else ""
            print(f"{'  ' * indent}- [{span['name']}] {duration:.1f}ms{token_str}")
            print(f"{'  ' * (indent+1)}Input: {span['inputs']}")
            print(f"{'  ' * (indent+1)}Output: {span['outputs']}")
            
            # Print children recursively
            children = by_parent.get(span["span_id"], [])
            for child in children:
                _print_span(child, indent + 1)

        for root in root_spans:
            _print_span(root)
        print("========================================================\n")

# Global active trace container
current_trace = ActiveTrace()
current_parent_span_id = None

def trace_span(name: str):
    """Decorator to trace function executions as spans."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            global current_parent_span_id
            previous_parent = current_parent_span_id
            
            # Start new span
            span = current_trace.start_span(name, parent_id=previous_parent)
            current_parent_span_id = span["span_id"]
            
            # Capture arguments as input representation
            inputs = {"args": args, "kwargs": kwargs}
            
            # Execute function
            start_time = time.time()
            result = func(*args, **kwargs)
            
            # Extract mock token consumption if returned as tuple, else set to none
            outputs = result
            tokens = None
            if isinstance(result, tuple) and "token_usage" in str(result):
                outputs, tokens = result
            
            # Update parent back to previous level
            current_parent_span_id = previous_parent
            current_trace.end_span(span["span_id"], inputs=inputs, outputs=outputs, tokens=tokens)
            return result
        return wrapper
    return decorator

# --- Demo Agentic Workflow ---

@trace_span("Vector DB Retrieval")
def retrieve_documents(query: str):
    time.sleep(0.1)  # Simulate DB latency
    return "Context: Server database connection error occurred on node-42."

@trace_span("LLM Chat Completion")
def call_llm(system_prompt: str, context: str, user_query: str):
    time.sleep(0.4)  # Simulate API latency
    generated_text = "The system is experiencing an outage due to database failure on node-42."
    # Return output and mock token usage statistics
    token_usage = {"input": 120, "output": 25}
    return generated_text, token_usage

@trace_span("Agent Orchestration (Root)")
def run_diagnostic_agent(user_query: str):
    # Step 1: Retrieve details
    context = retrieve_documents(user_query)
    
    # Step 2: Query LLM
    sys_prompt = "You are a system admin helper. Summarize the diagnostic information."
    llm_output = call_llm(sys_prompt, context, user_query)
    
    return llm_output

# Run the trace demo
if __name__ == "__main__":
    final_output = run_diagnostic_agent("Why is the server down?")
    
    # Print the resulting hierarchy
    current_trace.print_trace_tree()

Tooling Landscape

Depending on your hosting requirements and infrastructure scale, choose from the following industry tools:

| Platform | Type | Best For | Deployment | | :--- | :--- | :--- | :--- | | Langfuse | Open-source / SaaS | Direct trace inspection, cost estimation | Self-hosted or Cloud | | Arize Phoenix| Open-source / SaaS | Evaluation datasets, hallucination tracking | Notebook or Cloud | | LangSmith | SaaS (Proprietary) | Native integration with LangChain/LangGraph | Cloud Only | | Datadog LLM | Enterprise APM | Integrating AI metrics alongside standard servers| Enterprise Cloud |

Structured tracing allows you to transition your AI systems from black-box experiments into predictable, auditable software infrastructure.