AI Observability: How to Monitor LLMs, AI Agents, and RAG Systems in Production

You’ve deployed an LLM-powered feature. Users are complaining it gives wrong answers sometimes. Your dashboards show p99 latency is 12 seconds. You have no idea which part of the pipeline is slow — is it the embedding model? The vector database? The LLM call itself? The prompt template you changed last Tuesday?

Welcome to the observability gap in AI systems.

Traditional observability was designed for deterministic software: a request comes in, code executes, a response goes out. LLMs are non-deterministic, stateful, expensive, and their failures look nothing like a 500 error. A model that confidently returns wrong information doesn’t throw an exception. A RAG pipeline that retrieves irrelevant documents produces a plausible-sounding but useless answer. You won’t catch these with uptime checks.

This article covers what you actually need to monitor LLMs, AI agents, and RAG systems in production.

Why Standard APM Falls Short

Take a standard web request trace. You see the endpoint, the database queries, the downstream HTTP calls, the response time. Everything is structured, typed, and predictable.

Now take an LLM pipeline. A single user request might involve:

1. An embedding call to transform the query into a vector
2. A similarity search against a vector database returning 5 document chunks
3. A prompt assembly step that injects those chunks into a template
4. An LLM inference call with 1,200 tokens of input
5. A response parsing step
6. A second LLM call for fact-checking or routing
7. A tool call (search, calculator, database lookup)
8. A final synthesis call

Each step has its own latency profile, failure mode, cost, and quality dimension. Standard APM treats this as one opaque HTTP call to an API endpoint. That’s not useful.

The Four Dimensions of AI Observability

Monitoring AI systems requires four dimensions that don’t exist in traditional observability:

1. Cost

Every LLM call has a token cost. At scale, this becomes a significant operational expense. You need to track:

• Input tokens per request and per feature
• Output tokens per request and per feature
• Cost per model (GPT-4o vs GPT-4o-mini vs Claude vs Llama)
• Cost per user / tenant for billing and abuse detection
• Cost trends over time (are costs growing faster than usage?)

2. Latency at the Component Level

A 10-second response is bad. A 10-second response where 8 seconds is LLM inference and 2 seconds is retrieval is actionable. You need latency broken down by:

• Embedding generation time
• Vector search time
• Prompt assembly time
• LLM first token latency (TTFT)
• LLM completion latency
• Tool execution time

3. Quality Signals

This is the hard one. LLM quality doesn’t produce exceptions — it produces subtle degradation that only shows up as user dissatisfaction.

Signals you can measure:

• Retrieval relevance: are the retrieved chunks actually related to the query?
• Hallucination rate: does the response cite sources that don’t exist in the context?
• Answer completeness: does the response address what the user asked?
• Refusal rate: how often does the model refuse to answer?
• Regeneration rate: how often do users click “regenerate” or “try again”?
• Downstream user actions: do users follow up with clarification requests?

4. Prompt and Context Tracing

You need to capture the exact inputs that produced each output — the prompt template, the injected context, the model parameters. Without this, you can’t reproduce failures or evaluate the effect of prompt changes.

Instrumenting with OpenTelemetry Semantic Conventions for AI

The OpenTelemetry community has defined semantic conventions for AI/LLM systems (under active development as gen_ai.*). These give you a standard schema for spans and attributes.

Here’s what a properly instrumented LLM call span looks like:

from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("ai-pipeline")

def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    with tracer.start_as_current_span(
        "gen_ai.chat",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.request.max_tokens", 1024)
        span.set_attribute("gen_ai.request.temperature", 0.7)
        
        # Capture the prompt (be careful about PII)
        span.set_attribute("gen_ai.prompt", prompt[:2000])  # truncate for safety
        
        response = openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=1024,
        )
        
        # Capture response metadata
        span.set_attribute("gen_ai.response.model", response.model)
        span.set_attribute("gen_ai.usage.input_tokens", response.usage.prompt_tokens)
        span.set_attribute("gen_ai.usage.output_tokens", response.usage.completion_tokens)
        span.set_attribute("gen_ai.response.finish_reason", response.choices[0].finish_reason)
        
        # Calculate cost (example rates, adjust per model pricing)
        cost = (response.usage.prompt_tokens * 0.000005) + (response.usage.completion_tokens * 0.000015)
        span.set_attribute("gen_ai.usage.cost_usd", round(cost, 6))
        
        return response.choices[0].message.content

The key attributes from the gen_ai.* namespace:

Tracing a Complete RAG Pipeline

A RAG pipeline is a chain of distinct operations. Each deserves its own span with appropriate attributes.

from opentelemetry import trace
from opentelemetry.trace import SpanKind
import time

tracer = trace.get_tracer("rag-pipeline")

def rag_query(user_query: str) -> str:
    with tracer.start_as_current_span("rag.pipeline") as pipeline_span:
        pipeline_span.set_attribute("rag.query", user_query)
        pipeline_span.set_attribute("rag.pipeline.version", "v2.3")
        
        # Step 1: Embed the query
        with tracer.start_as_current_span("rag.embed_query", kind=SpanKind.CLIENT) as embed_span:
            embed_span.set_attribute("gen_ai.system", "openai")
            embed_span.set_attribute("gen_ai.request.model", "text-embedding-3-small")
            
            query_embedding = embed_client.embeddings.create(
                model="text-embedding-3-small",
                input=user_query
            )
            embed_span.set_attribute("gen_ai.usage.input_tokens", query_embedding.usage.total_tokens)
        
        # Step 2: Retrieve relevant chunks
        with tracer.start_as_current_span("rag.retrieve", kind=SpanKind.CLIENT) as retrieve_span:
            retrieve_span.set_attribute("db.system", "pinecone")
            retrieve_span.set_attribute("rag.retrieval.top_k", 5)
            retrieve_span.set_attribute("rag.retrieval.namespace", "product-docs")
            
            results = vector_db.query(
                vector=query_embedding.data[0].embedding,
                top_k=5,
                include_metadata=True
            )
            
            # Critical: capture what was actually retrieved
            retrieve_span.set_attribute("rag.retrieval.results_count", len(results.matches))
            retrieve_span.set_attribute("rag.retrieval.top_score", results.matches[0].score if results.matches else 0)
            retrieve_span.set_attribute("rag.retrieval.min_score", results.matches[-1].score if results.matches else 0)
            
            # Log retrieved document IDs for debugging
            doc_ids = [m.id for m in results.matches]
            retrieve_span.set_attribute("rag.retrieval.document_ids", str(doc_ids))
        
        # Step 3: Assemble the prompt
        with tracer.start_as_current_span("rag.assemble_prompt") as assemble_span:
            context_chunks = [m.metadata["text"] for m in results.matches]
            context_text = "\n\n---\n\n".join(context_chunks)
            
            prompt = f"""Use the following context to answer the question. 
If the answer is not in the context, say so explicitly.

Context:
{context_text}

Question: {user_query}

Answer:"""
            
            assemble_span.set_attribute("rag.prompt.template_version", "v1.2")
            assemble_span.set_attribute("rag.prompt.context_chunks", len(context_chunks))
            assemble_span.set_attribute("rag.prompt.total_chars", len(prompt))
        
        # Step 4: Generate the answer
        answer = call_llm(prompt, model="gpt-4o")
        
        pipeline_span.set_attribute("rag.answer_length", len(answer))
        
        return answer

This trace gives you a complete picture of what happened for every single user request: which documents were retrieved and with what relevance scores, how many tokens were consumed at each step, and exactly which prompt template produced the output.

Monitoring AI Agents

Agents are harder to trace because they’re dynamic — the number of steps isn’t known at request time. An agent might call one tool, or it might run 12 tool calls in a loop before reaching an answer (or hitting a loop detection limit).

The key pattern for agent observability is a hierarchical span structure:

def run_agent(user_goal: str) -> str:
    with tracer.start_as_current_span("agent.run") as agent_span:
        agent_span.set_attribute("agent.goal", user_goal)
        agent_span.set_attribute("agent.model", "gpt-4o")
        
        messages = [{"role": "user", "content": user_goal}]
        step = 0
        max_steps = 10
        
        while step < max_steps:
            step += 1
            
            with tracer.start_as_current_span(f"agent.step") as step_span:
                step_span.set_attribute("agent.step.number", step)
                
                # LLM decides what to do next
                response = call_llm_with_tools(messages, tools=AVAILABLE_TOOLS)
                
                if response.finish_reason == "stop":
                    # Agent decided to respond directly
                    step_span.set_attribute("agent.step.type", "final_answer")
                    agent_span.set_attribute("agent.total_steps", step)
                    return response.content
                
                elif response.finish_reason == "tool_calls":
                    # Agent decided to use a tool
                    for tool_call in response.tool_calls:
                        with tracer.start_as_current_span("agent.tool_call") as tool_span:
                            tool_span.set_attribute("agent.tool.name", tool_call.function.name)
                            tool_span.set_attribute("agent.tool.arguments", tool_call.function.arguments)
                            
                            # Execute the tool
                            t0 = time.time()
                            result = execute_tool(tool_call.function.name, tool_call.function.arguments)
                            tool_span.set_attribute("agent.tool.execution_ms", int((time.time() - t0) * 1000))
                            tool_span.set_attribute("agent.tool.result_length", len(str(result)))
                            
                            messages.append({"role": "tool", "content": str(result)})
        
        # Agent hit step limit
        agent_span.set_attribute("agent.hit_step_limit", True)
        agent_span.set_attribute("agent.total_steps", step)
        raise AgentStepLimitError(f"Agent exceeded {max_steps} steps without reaching a final answer")

Critical metrics to derive from agent traces:

• Steps per completion — how many LLM calls does it take to answer? High step counts indicate prompt or tool design issues.
• Tool call distribution — which tools get called most? Which are slowest?
• Step limit hit rate — if agents frequently hit your max steps cap, your system is often failing silently.
• Agent success rate — how often does the agent produce a final answer vs error out?

Metrics to Track in Production

Beyond traces, you need aggregated metrics for dashboards and alerts:

from opentelemetry import metrics

meter = metrics.get_meter("ai-pipeline")

# Cost tracking
llm_cost_counter = meter.create_counter(
    "gen_ai.cost.usd",
    description="Total LLM cost in USD",
    unit="USD",
)

# Token usage
input_token_counter = meter.create_counter(
    "gen_ai.usage.input_tokens",
    description="Total input tokens consumed",
)
output_token_counter = meter.create_counter(
    "gen_ai.usage.output_tokens",
    description="Total output tokens generated",
)

# Latency histograms
llm_latency = meter.create_histogram(
    "gen_ai.request.duration",
    description="LLM request duration",
    unit="s",
)

# Quality signals
retrieval_score = meter.create_histogram(
    "rag.retrieval.top_score",
    description="Top similarity score from vector search",
)

refusal_counter = meter.create_counter(
    "gen_ai.response.refusals",
    description="Number of times the model refused to answer",
)

# Record these in your instrumentation code
def record_llm_call(model: str, feature: str, input_tokens: int, output_tokens: int, duration_s: float, cost: float):
    labels = {"gen_ai.request.model": model, "feature": feature}
    
    input_token_counter.add(input_tokens, labels)
    output_token_counter.add(output_tokens, labels)
    llm_cost_counter.add(cost, labels)
    llm_latency.record(duration_s, labels)

Key dashboards to build:

Cost Dashboard:

• Cost per hour by model
• Cost per feature (chat, search, summarization)
• Cost per user (for multi-tenant systems)
• Cost trend vs request volume (are you getting more efficient over time?)

Performance Dashboard:

• LLM latency p50/p90/p99 by model
• TTFT (time to first token) for streaming responses
• Retrieval latency by vector database
• End-to-end pipeline latency

Quality Dashboard:

• Refusal rate over time
• Retrieval top score distribution (are scores dropping? your data may be drifting)
• User regeneration rate (if you track it)
• Finish reason distribution (how often does generation stop due to length vs stop?)

Handling Sensitive Data in AI Traces

Prompts and responses often contain sensitive user data. Before capturing them in traces, establish clear policies:

What to capture:

• Token counts (always safe)
• Model name and parameters (always safe)
• Latency and cost (always safe)
• Document IDs retrieved, not document content (usually safe)
• Truncated prompts with PII removed

What to avoid:

• Raw user queries if they may contain health, financial, or personal data
• Full prompt text if it contains injected user content
• LLM responses if they may reflect sensitive input back

A practical approach is to hash user-identifying fields and capture only metadata about prompt content:

import hashlib

def safe_span_attributes(span, user_query: str, response: str):
    # Safe: hashed user identifier for correlation without exposing content
    span.set_attribute("user.query.hash", hashlib.sha256(user_query.encode()).hexdigest()[:16])
    
    # Safe: structural metadata about the prompt
    span.set_attribute("prompt.word_count", len(user_query.split()))
    span.set_attribute("prompt.has_code", "```" in user_query)
    
    # Safe: structural metadata about the response  
    span.set_attribute("response.word_count", len(response.split()))
    span.set_attribute("response.has_citations", "Source:" in response)
    
    # Conditional: capture content only in development/staging
    import os
    if os.getenv("CAPTURE_PROMPT_CONTENT", "false") == "true":
        span.set_attribute("prompt.content", user_query[:500])

Alerting on AI-Specific Failure Modes

Standard error-rate alerts don’t work for AI systems. Here are the alerts you actually need:

Using OpenLLMetry for Drop-in Instrumentation

If you don’t want to write all this instrumentation manually, OpenLLMetry (by Traceloop) provides OpenTelemetry-based auto-instrumentation for most LLM SDKs:

pip install opentelemetry-sdk traceloop-sdk

from traceloop.sdk import Traceloop

Traceloop.init(
    app_name="my-rag-service",
    api_endpoint="http://localhost:4318",  # Your OTLP collector endpoint
    disable_batch=False,
)

# From this point, all calls to openai, anthropic, langchain, llama-index,
# pinecone, chromadb, etc. are automatically traced with gen_ai.* attributes

OpenLLMetry instruments: OpenAI, Anthropic, Azure OpenAI, Cohere, Bedrock, LangChain, LlamaIndex, Pinecone, Chroma, Qdrant, Weaviate, and more. For most teams, this is the right starting point before building custom instrumentation.

The Production Readiness Checklist

Before shipping an AI feature to production, verify:

• Every LLM call is traced with gen_ai.* attributes (model, token counts, latency, cost)
• RAG retrieval spans capture document IDs and similarity scores
• Agent traces include step count and tool call details
• Cost is tracked per feature and per model, not just globally
• Prompts and responses are captured with PII protections in place
• Alerts exist for cost spikes, refusal rate, and step limit hits
• You can reproduce any failure by replaying the traced prompt and context
• Dashboards show quality signals, not just uptime and latency
• Log-trace correlation is in place (every log line carries the trace ID)
• You have a baseline for “normal” token usage to detect anomalies

Observability for AI systems is not a nice-to-have. When your LLM pipeline starts returning wrong answers in production, or when a cost anomaly hits your bill at the end of the month, your traces are the only path back to a root cause.

Build the instrumentation before you need it.