Tracing a RAG pipeline end to end with OpenTelemetry

Tracing RAG with OpenTelemetry gives you a single timeline of every retrieval, prompt assembly, and model call in your pipeline. Without it, you’re guessing which chunk blew up the context window or why a response cited the wrong document.

Prerequisites

Python 3.10 or newer
opentelemetry-api and opentelemetry-sdk for instrumentation
openai if you plan to hit a real model (we’ll mock it for a runnable demo)
Optional: opentelemetry-exporter-otlp-proto-grpc and Docker for Jaeger

Install the basics:

pip install opentelemetry-api opentelemetry-sdk openai

Configure the tracer

You need a TracerProvider and at least one span processor. For local development, ConsoleSpanExporter prints spans as they flush.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("rag.tutorial")

This is the foundation for tracing RAG with OpenTelemetry: every step below becomes a child span of a root pipeline span.

Step 1: Load and chunk documents

Keep the span small and attribute it with counts. Never put full document text in attributes—that floods your trace backend.

def load_and_chunk():
    with tracer.start_as_current_span("load_documents") as span:
        docs = [
            "The capital of France is Paris.",
            "LLM observability needs distributed traces.",
        ]
        chunks = docs  # already sentence-sized
        span.set_attribute("chunk.count", len(chunks))
        return chunks

Step 2: Embed and retrieve

A real system uses a vector DB. For the tutorial, a fake embedding keeps it runnable without network calls.

import hashlib
import numpy as np

def fake_embed(text):
    h = hashlib.sha256(text.encode()).digest()
    return np.frombuffer(h[:8], dtype=np.uint8).astype(float)

def retrieve(query, chunks, k=1):
    with tracer.start_as_current_span("retrieve") as span:
        q_emb = fake_embed(query)
        scored = []
        for c in chunks:
            c_emb = fake_embed(c)
            sim = float(np.dot(q_emb, c_emb) /
                        (np.linalg.norm(q_emb) * np.linalg.norm(c_emb) + 1e-9))
            scored.append((sim, c))
        scored.sort(reverse=True)
        top = [c for _, c in scored[:k]]
        span.set_attribute("retrieve.k", k)
        span.set_attribute("retrieve.top_chunk", top[0] if top else "")
        span.set_attribute("retrieve.score", scored[0][0] if scored else 0.0)
        return top

The retrieve span now exposes which chunk won and its similarity score. When tracing RAG with OpenTelemetry at scale, these attributes are how you spot drift in embedding quality.

Step 3: Generate the answer

Wrap the LLM call. If you use a gateway such as n4n.ai, point the OpenAI client at its OpenAI-compatible endpoint; it fronts 240+ models, applies automatic fallback on provider degradation, and forwards cache-control hints so your span can note cache hits via response headers.

def generate(query, context):
    with tracer.start_as_current_span("generate") as span:
        # from openai import OpenAI
        # client = OpenAI(base_url="https://api.n4n.ai/v1")
        # resp = client.chat.completions.create(
        #     model="gpt-4o-mini",
        #     messages=[{"role": "user", "content": f"Query: {query}\nContext: {context}"}],
        # )
        prompt = f"Query: {query}\nContext: {context}"
        span.set_attribute("prompt.length", len(prompt))
        # mock completion
        answer = "Paris" if "France" in context else "traces"
        span.set_attribute("completion.tokens", len(answer.split()))
        return answer

Step 4: Compose the pipeline

def run_rag(query):
    with tracer.start_as_current_span("rag_pipeline") as root:
        root.set_attribute("query", query)
        chunks = load_and_chunk()
        context = retrieve(query, chunks)
        answer = generate(query, " ".join(context))
        root.set_attribute("answer", answer)
        return answer

if __name__ == "__main__":
    print(run_rag("What is the capital of France?"))

Run the script. The console exporter emits each span as JSON. Expected checkpoint output (abridged):

{
  "name": "rag_pipeline",
  "attributes": {"query": "What is the capital of France?", "answer": "Paris"},
  "events": [],
  "children": [
    {"name": "load_documents", "attributes": {"chunk.count": 2}},
    {"name": "retrieve", "attributes": {"retrieve.k": 1, "retrieve.top_chunk": "The capital of France is Paris.", "retrieve.score": 0.91}},
    {"name": "generate", "attributes": {"prompt.length": 54, "completion.tokens": 1}}
  ]
}

You now have an end-to-end trace. The hierarchy shows exactly where latency accumulated.

Step 5: Export to Jaeger via OTLP

Console is fine for a laptop. For a team, ship spans to a collector.

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

Start Jaeger:

docker run -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one:latest

Open http://localhost:16686, search for service rag.tutorial, and you’ll see the waterfall.

Propagate context to remote services

If your retriever is a separate service, inject the trace context into outbound requests:

from opentelemetry.propagate import inject
import requests

def remote_retrieve(query):
    with tracer.start_as_current_span("retrieve_remote") as span:
        headers = {}
        inject(headers)  # adds traceparent
        # requests.get("http://retriever:8000/search", params={"q": query}, headers=headers)

The remote service should extract the context so the span tree stays connected across process boundaries.

Error handling and status

A RAG pipeline fails in many ways: empty retrieval, malformed prompt, model timeout. Record exceptions on the span:

def generate(query, context):
    with tracer.start_as_current_span("generate") as span:
        try:
            # ... LLM call ...
            if not context:
                raise ValueError("empty context")
        except Exception as e:
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            raise

Marking status lets you build dashboards that alert on error rates per pipeline stage.

Attributes that actually pay off

When tracing RAG with OpenTelemetry in production, instrument these minimally:

retrieve.top_chunk_id (not full text)
retrieve.score
prompt.token_count (use tokenizer, not len())
completion.token_count
model.id

If you route through an OpenAI-compatible gateway that provides per-token usage metering, copy usage.total_tokens into the generate span. That turns a latency trace into a cost trace.

Wrap-up

You built a traced RAG pipeline from scratch: document load, retrieval, and generation each emit spans with actionable attributes. Swap the fake embedder for a real model and the console exporter for OTLP, and you have production-grade observability. The pattern stays identical as you add rerankers, query rewriting, or multi-hop retrieval—wrap each step in a span, tag it with IDs and scores, and let the trace tell you where the system breaks.