Observability in AI: Tracing, Debugging, and Monitoring LLM Applications
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 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.