EngineersofAI LogoCoursesBlogResearch LabAI LettersThe LabCode BankInteractive 3DKodrEarnestJobs
Skip to main content

:::tip 🎮 Interactive Playground Visualize this concept: Try the LLM Observability demo on the EngineersOfAI Playground - no code required. :::

OpenTelemetry for AI Systems

The Vendor Lock-In Crisis​

Your team spent six months building a comprehensive LLM observability stack around LangSmith. The tracing pipeline, the dataset curation, the evaluation workflows - all deeply integrated. The team knows the UI, the Python SDK, the concept of "runs" and "projects." Then the engineering organization makes an acquisition, and the new engineering standards require all observability to flow through Datadog. The migration cost: rewrite every @traceable decorator, the dataset management infrastructure, the CI eval pipelines. Estimated timeline: three months of engineering time, most of it undifferentiated plumbing with no product value.

Or consider the alternative path, where the original team made a different architectural decision. They built on OpenTelemetry from the start. Their instrumentation code uses standard OTel APIs - the same ones used for their existing microservices. The same TracerProvider and SpanProcessor that routes service traces to Datadog can be configured to route LLM spans there too. The migration is a configuration change in the OTel Collector YAML. One afternoon, not three months.

This is the strategic argument for OpenTelemetry: it converts the "which observability backend" decision from an irreversible architectural commitment into a runtime configuration choice.

What OpenTelemetry Is​

OpenTelemetry (OTel) is the CNCF-graduated standard for distributed observability. It covers three signals:

  • Traces: Distributed request flows - spans, trace context propagation, parent-child relationships
  • Metrics: Numeric time-series - counters, histograms, gauges
  • Logs: Structured log records with trace context correlation

For AI applications, OTel's trace API is the most relevant signal. It gives you a standard way to create spans for LLM calls, retrieval steps, and tool executions, and export those spans to any compatible backend: Jaeger, Grafana Tempo, Honeycomb, Datadog, Dynatrace, LangSmith's OTel endpoint, Langfuse's OTel endpoint, Arize Phoenix.

The key property: instrument once, export anywhere. Your instrumentation code does not care where spans end up. That decision is made in the OTel Collector or in your TracerProvider configuration.

The OTel Stack for AI​

Core OTel Concepts​

Spans​

A Span represents a unit of work: an LLM call, a retrieval operation, a tool execution. Each span contains:

  • trace_id: identifies the root request - shared across all spans in one user request
  • span_id: unique identifier for this specific operation
  • parent_span_id: links child to parent, creating the trace tree structure
  • name: what operation this span represents (e.g., "anthropic chat", "vector-retrieval")
  • start_time / end_time: when the operation began and ended
  • attributes: key-value pairs with operation metadata (the GenAI attributes go here)
  • status: OK, ERROR, or UNSET
  • events: timestamped events within a span (e.g., "first_token_received" for streaming)

The GenAI Semantic Conventions​

The OTel community defines semantic conventions - standard attribute names so all tools interpret spans consistently. The GenAI conventions, now part of the core OTel specification:

# Standard OTel GenAI semantic convention attribute names

# Provider and operation
gen_ai.system              # "anthropic" | "openai" | "google" | "aws_bedrock"
gen_ai.operation.name      # "chat" | "text_completion" | "embeddings"

# Request parameters
gen_ai.request.model       # "claude-opus-4-6"
gen_ai.request.max_tokens  # 1024
gen_ai.request.temperature # 0.7
gen_ai.request.top_p       # 0.95

# Response metadata
gen_ai.response.model      # actual model used (may differ from requested)
gen_ai.finish_reason       # "end_turn" | "max_tokens" | "stop_sequence"

# Token usage (critical for cost tracking)
gen_ai.usage.input_tokens  # tokens in the prompt
gen_ai.usage.output_tokens # tokens in the completion
gen_ai.usage.total_tokens  # sum (some backends set this directly)

# Content (optional - may be disabled for PII-sensitive apps)
gen_ai.prompt              # the actual prompt text (often sampled, not always captured)
gen_ai.completion          # the model's response text

OpenInference (used by Phoenix and now becoming an OTel standard) extends these with RAG-specific attributes:

# OpenInference extensions for RAG and embedding operations
retrieval.query            # the retrieval query string
retrieval.documents        # JSON array of retrieved documents with content and metadata
embedding.model_name       # which embedding model was used
embedding.embeddings       # the embedding vectors (for UMAP visualization in Phoenix)
llm.input_messages         # full message array for chat models
llm.output_messages        # the model's response messages

Why conventions matter: when all tools agree on attribute names, pre-built dashboards, alert rules, and queries work across backends. A Phoenix UMAP visualization can read retrieval.documents from your LangChain retriever span because LangChain's OpenInference instrumentor uses the same attribute name that Phoenix's evaluation engine expects.

Installation​

# Core OTel SDK and exporters
pip install opentelemetry-sdk
pip install opentelemetry-exporter-otlp-proto-grpc   # for gRPC export (most backends)
pip install opentelemetry-exporter-otlp-proto-http   # for HTTP export (some backends)

# Auto-instrumentation for AI SDKs (OpenInference project)
pip install openinference-instrumentation-anthropic
pip install openinference-instrumentation-openai
pip install openinference-instrumentation-langchain
pip install openinference-instrumentation-llama-index

# HTTP client instrumentation (for context propagation across services)
pip install opentelemetry-instrumentation-httpx
pip install opentelemetry-instrumentation-requests
pip install opentelemetry-instrumentation-fastapi

Setting Up OTel for AI Applications​

Basic Setup: Local Development​

# otel_setup.py - Basic OTel setup for local development
import os
import anthropic
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from openinference.instrumentation.anthropic import AnthropicInstrumentor


def configure_otel_for_development(
    service_name: str = "ai-app-dev",
    phoenix_endpoint: str = "http://localhost:6006/v1/traces",
    print_to_console: bool = False,
) -> trace.Tracer:
    """
    Configure OTel for local development with Phoenix as the backend.
    Auto-instruments the Anthropic SDK so all calls are traced automatically.
    """
    # Resource identifies your service in traces
    resource = Resource.create({
        "service.name":           service_name,
        "service.version":        os.getenv("APP_VERSION", "dev"),
        "deployment.environment": "development",
    })

    provider = TracerProvider(resource=resource)

    # Export to Phoenix for local analysis
    otlp_exporter = OTLPSpanExporter(endpoint=phoenix_endpoint)
    provider.add_span_processor(
        BatchSpanProcessor(
            otlp_exporter,
            max_queue_size=1024,
            max_export_batch_size=256,
            export_timeout_millis=10_000,
        )
    )

    if print_to_console:
        # Also print spans to console for debugging
        provider.add_span_processor(
            BatchSpanProcessor(ConsoleSpanExporter())
        )

    trace.set_tracer_provider(provider)

    # CRITICAL: instrument BEFORE creating any Anthropic client
    AnthropicInstrumentor().instrument()

    return trace.get_tracer(service_name)


# Usage
tracer = configure_otel_for_development()
client = anthropic.Anthropic()  # now auto-traced

# Every call below generates a span in Phoenix automatically
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain RAG in 3 sentences."}]
)
print(response.content[0].text)
# Open http://localhost:6006 to see the trace

Production Setup: Multi-Backend Routing​

Production systems typically need spans to flow to multiple backends simultaneously - your enterprise APM for operational monitoring and an AI-specific backend for quality analysis:

# otel_production.py - Production OTel setup with multiple backends
import os
import anthropic
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from openinference.instrumentation.anthropic import AnthropicInstrumentor


def configure_otel_for_production(
    service_name: str,
    otel_collector_endpoint: str = "http://otel-collector:4317",
) -> trace.Tracer:
    """
    Production OTel setup.

    Routes ALL spans to the OTel Collector, which then fans out to:
    - Datadog (100% of traces, for operational monitoring)
    - LangSmith (100% of AI-tagged traces, for quality observability)
    - Phoenix (10% sampled, for ML analysis)

    The collector handles routing - your app only talks to one endpoint.
    """
    resource = Resource.create({
        "service.name":           service_name,
        "service.version":        os.environ.get("APP_VERSION", "unknown"),
        "deployment.environment": os.environ.get("ENVIRONMENT", "production"),
        "team":                   os.environ.get("TEAM_NAME", "unknown"),
    })

    provider = TracerProvider(resource=resource)

    # Single export destination: the OTel Collector
    # The collector handles all routing, sampling, and fan-out
    collector_exporter = OTLPSpanExporter(
        endpoint=otel_collector_endpoint,
        insecure=True,  # use TLS in production
    )
    provider.add_span_processor(
        BatchSpanProcessor(
            collector_exporter,
            max_queue_size=4096,          # buffer up to 4096 spans in memory
            max_export_batch_size=512,    # export in batches of 512
            export_timeout_millis=30_000, # 30-second export timeout
            schedule_delay_millis=1000,   # flush every 1 second if batch not full
        )
    )

    trace.set_tracer_provider(provider)

    # Instrument all AI SDKs before creating any clients
    AnthropicInstrumentor().instrument()
    # LangChainInstrumentor().instrument()  # if using LangChain
    # OpenAIInstrumentor().instrument()     # if using OpenAI SDK

    return trace.get_tracer(service_name)

Manual Instrumentation for Custom Pipeline Steps​

Auto-instrumentation covers Anthropic and other AI SDK calls, but you must manually instrument retrieval steps, tool calls, API calls to external services, and any custom processing:

# pipeline/manual_instrumentation.py
import anthropic
import time
import json
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("rag-pipeline", "1.0.0")
client = anthropic.Anthropic()


def retrieve_documents(
    query: str,
    k: int = 5,
    index_name: str = "knowledge-base-v3",
) -> list[dict]:
    """Retrieval with manual OTel instrumentation."""
    with tracer.start_as_current_span("vector-retrieval") as span:
        # Standard semantic attributes
        span.set_attribute("retrieval.query",  query[:500])
        span.set_attribute("retrieval.k",      k)
        span.set_attribute("db.system",        "pinecone")
        span.set_attribute("db.operation",     "similarity_search")
        span.set_attribute("db.collection",    index_name)

        start = time.monotonic()

        try:
            # Simulate vector DB call (replace with real implementation)
            results = [
                {"content": f"Document {i} about {query[:20]}...",
                 "source": f"doc-{i}", "score": 0.90 - i * 0.05}
                for i in range(k)
            ]

            duration_ms = (time.monotonic() - start) * 1000

            # Record output attributes
            span.set_attribute("retrieval.num_results",     len(results))
            span.set_attribute("retrieval.latency_ms",      round(duration_ms, 2))
            span.set_attribute("retrieval.avg_score",
                round(sum(r["score"] for r in results) / len(results), 4) if results else 0)

            # Log top 3 results for debugging
            for i, doc in enumerate(results[:3]):
                span.set_attribute(f"retrieval.result.{i}.content", doc["content"][:300])
                span.set_attribute(f"retrieval.result.{i}.source",  doc["source"])
                span.set_attribute(f"retrieval.result.{i}.score",   doc["score"])

            span.set_status(Status(StatusCode.OK))
            return results

        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise


def execute_tool(tool_name: str, parameters: dict) -> dict:
    """Tool call with OTel instrumentation and event logging."""
    with tracer.start_as_current_span(f"tool-{tool_name}") as span:
        span.set_attribute("tool.name",            tool_name)
        span.set_attribute("tool.parameters",      json.dumps(parameters)[:500])
        span.set_attribute("tool.parameter_count", len(parameters))

        # Add a timestamped event when the tool starts executing
        span.add_event("tool_invoked", {
            "tool.name":    tool_name,
            "param_keys":   str(list(parameters.keys())),
        })

        start = time.monotonic()
        result = {"status": "success", "data": f"Result from {tool_name}"}
        duration_ms = (time.monotonic() - start) * 1000

        span.add_event("tool_completed", {
            "duration_ms":  round(duration_ms, 2),
            "result_keys":  str(list(result.keys())),
        })
        span.set_attribute("tool.result_summary", str(result)[:300])
        span.set_attribute("tool.latency_ms",     round(duration_ms, 2))
        span.set_status(Status(StatusCode.OK))

        return result


def rag_pipeline_with_tools(
    user_id:    str,
    session_id: str,
    query:      str,
) -> dict:
    """
    Full RAG pipeline with tool calls.
    Span hierarchy:
    rag-request (root)
    ├── vector-retrieval (manual)
    ├── tool-calculator (manual)
    └── anthropic chat (auto-instrumented)
    """
    with tracer.start_as_current_span("rag-request") as root:
        root.set_attribute("user.id",     user_id)
        root.set_attribute("session.id",  session_id)
        root.set_attribute("input.query", query[:500])
        root.set_attribute("pipeline.version", "v2.1")

        try:
            # Step 1: Retrieve (manual span)
            docs = retrieve_documents(query, k=5)
            context_text = "\n\n".join(d["content"] for d in docs)

            # Step 2: Optionally call a tool
            if "calculate" in query.lower() or "compute" in query.lower():
                tool_result = execute_tool("calculator", {"expression": query})
                context_text += f"\n\nCalculation result: {tool_result['data']}"

            # Step 3: Generate - auto-instrumented by AnthropicInstrumentor
            response = client.messages.create(
                model="claude-opus-4-6",
                max_tokens=2048,
                temperature=0.0,
                system=(
                    "Answer questions using only the provided context. "
                    "Be precise and cite specific passages."
                ),
                messages=[{
                    "role": "user",
                    "content": f"Context:\n{context_text}\n\nQuestion: {query}"
                }]
            )

            answer = response.content[0].text

            root.set_attribute("output.answer_length",  len(answer))
            root.set_attribute("output.docs_retrieved", len(docs))
            root.set_status(Status(StatusCode.OK))

            trace_id = format(root.get_span_context().trace_id, "032x")

            return {
                "answer":   answer,
                "trace_id": trace_id,
                "docs_used": len(docs),
            }

        except Exception as e:
            root.set_status(Status(StatusCode.ERROR, str(e)))
            root.record_exception(e)
            raise

Context Propagation Across Service Boundaries​

OTel's most important capability for distributed AI systems: trace context propagates across service boundaries automatically when you use instrumented HTTP clients.

# distributed/context_propagation.py
import asyncio
import anthropic
import httpx
from opentelemetry import trace
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from fastapi import FastAPI, Request

# Instrument HTTPX - context propagates automatically in all HTTPX calls
HTTPXClientInstrumentor().instrument()

tracer = trace.get_tracer("ai-orchestrator")
client = anthropic.Anthropic()


async def orchestrate_rag_pipeline(query: str, user_id: str) -> str:
    """
    Orchestrator that calls downstream microservices.
    OTel context propagates via W3C traceparent headers automatically.
    """
    with tracer.start_as_current_span("orchestrate") as span:
        span.set_attribute("user.id", user_id)
        span.set_attribute("input.query", query)

        async with httpx.AsyncClient() as http:
            # HTTPXClientInstrumentor auto-injects traceparent header
            # The retrieval service picks it up and creates child spans
            retrieval_response = await http.post(
                "http://retrieval-service:8001/retrieve",
                json={"query": query, "k": 5},
                timeout=10.0,
            )
            retrieval_response.raise_for_status()
            docs = retrieval_response.json()["documents"]

            # Same for the embedding service
            embedding_response = await http.post(
                "http://embedding-service:8002/embed",
                json={"text": query},
                timeout=5.0,
            )
            embedding_response.raise_for_status()

        # The Anthropic call is auto-instrumented and appears as a child span
        context_text = "\n\n".join(d["content"] for d in docs)
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"Context:\n{context_text}\n\nQ: {query}"
            }]
        )

        return response.content[0].text


# The retrieval service receives the trace context automatically:
retrieval_app = FastAPI()
FastAPIInstrumentor.instrument_app(retrieval_app)  # auto-extracts traceparent

@retrieval_app.post("/retrieve")
async def retrieve(request: Request):
    """
    This span automatically becomes a child of the orchestrator's span
    because FastAPIInstrumentor extracts the traceparent header.
    No explicit context passing needed.
    """
    body = await request.json()
    query = body["query"]

    with tracer.start_as_current_span("retrieval-service-search") as span:
        span.set_attribute("retrieval.query", query)
        # ... perform retrieval
        return {"documents": [{"content": f"Doc about {query}...", "score": 0.88}]}

Sampling Strategies​

Tracing 100% of production traffic is expensive. OTel supports several sampling strategies at different levels:

# sampling/otel_sampling.py
from opentelemetry.sdk.trace.sampling import (
    TraceIdRatioBased,
    ParentBased,
    ALWAYS_ON,
    ALWAYS_OFF,
    Sampler,
    SamplingResult,
    Decision,
)
from opentelemetry.sdk.trace import TracerProvider
import random


# ── Head Sampling (decision made at trace start) ─────────────────────────────

# Sample 10% of all traces randomly
ratio_sampler = TraceIdRatioBased(0.10)
provider = TracerProvider(sampler=ratio_sampler)

# ParentBased: respect upstream decisions, apply ratio to roots
# If the HTTP gateway already decided to sample this request, continue sampling
parent_based = ParentBased(root=TraceIdRatioBased(0.10))
provider = TracerProvider(sampler=parent_based)


# ── Custom Sampler: Always sample errors and enterprise users ─────────────────

class IntelligentSampler(Sampler):
    """
    Always sample: errors, enterprise users, slow requests.
    Sample at base_rate for everything else.

    This is the recommended approach for AI applications:
    - Errors always need investigation
    - Enterprise users expect full observability
    - Slow requests indicate bottlenecks worth investigating
    """

    def __init__(
        self,
        base_rate:         float = 0.05,  # 5% of normal traffic
        enterprise_rate:   float = 1.0,   # 100% of enterprise traffic
        slow_threshold_ms: float = 5000,  # always trace if > 5s
    ):
        self.base_rate         = base_rate
        self.enterprise_rate   = enterprise_rate
        self.slow_threshold_ms = slow_threshold_ms
        self._ratio            = TraceIdRatioBased(base_rate)

    def should_sample(
        self, parent_context, trace_id, name, kind, attributes, links
    ):
        attrs = dict(attributes or {})

        # Always sample errors
        if attrs.get("error") or "error" in str(name).lower():
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        # Always sample enterprise tier
        if attrs.get("user.tier") == "enterprise":
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        # Always sample slow requests (detected at request start via expected duration hint)
        if attrs.get("http.expected_duration_ms", 0) > self.slow_threshold_ms:
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        # Always sample when the upstream decided to sample (ParentBased behavior)
        if parent_context and parent_context.is_valid and parent_context.trace_flags.sampled:
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        # Base rate sampling for everything else
        return self._ratio.should_sample(
            parent_context, trace_id, name, kind, attributes, links
        )

    def get_description(self) -> str:
        return f"IntelligentSampler(base_rate={self.base_rate})"


# Apply the custom sampler
intelligent_sampler = IntelligentSampler(base_rate=0.05)
production_provider = TracerProvider(sampler=intelligent_sampler)

OTel Collector Configuration​

The OTel Collector is a standalone process that receives spans from your application, applies processing (sampling, enrichment, batching), and routes to multiple backends. It decouples your application from your observability backends:

# otel-collector-config.yaml - Production multi-backend routing
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  # Batch spans for efficient export
  batch:
    timeout: 1s
    send_batch_size: 1024
    send_batch_max_size: 2048

  # Add environment metadata to all spans
  resource:
    attributes:
      - action: insert
        key: environment
        value: production
      - action: insert
        key: region
        value: us-east-1

  # Tail-based sampling: sample after seeing the complete trace
  # Requires buffering - use for precise control over what gets exported
  tail_sampling:
    decision_wait: 10s    # wait 10s for all spans in a trace to arrive
    num_traces: 50000     # buffer up to 50K traces
    expected_new_traces_per_sec: 200
    policies:
      # Always keep error traces
      - name: errors
        type: status_code
        status_code: {status_codes: [ERROR]}

      # Always keep slow traces (> 5 seconds)
      - name: slow-traces
        type: latency
        latency: {threshold_ms: 5000}

      # Always keep traces from enterprise users
      - name: enterprise-users
        type: string_attribute
        string_attribute:
          key: user.tier
          values: ["enterprise"]

      # Sample 5% of everything else
      - name: base-rate
        type: probabilistic
        probabilistic: {sampling_percentage: 5}

exporters:
  # Datadog: 100% of traces via tail sampling
  datadog:
    api:
      key: ${DATADOG_API_KEY}
      site: datadoghq.com
    traces:
      span_name_as_resource_name: true

  # LangSmith: AI-specific traces
  otlphttp/langsmith:
    endpoint: https://api.smith.langchain.com/otel/v1/traces
    headers:
      x-api-key: ${LANGSMITH_API_KEY}

  # Langfuse: AI-specific traces (alternative to LangSmith)
  otlphttp/langfuse:
    endpoint: ${LANGFUSE_HOST}/api/public/otel/v1/traces
    auth:
      authenticator: basicauth
    headers:
      authorization: Basic ${LANGFUSE_BASIC_AUTH_B64}

  # Phoenix: sampled for ML analysis
  otlphttp/phoenix:
    endpoint: http://phoenix-server:6006/v1/traces

  # Grafana Tempo: open-source option
  otlphttp/tempo:
    endpoint: http://tempo:4318

service:
  pipelines:
    # Full pipeline with tail sampling for production
    traces/production:
      receivers:  [otlp]
      processors: [resource, tail_sampling, batch]
      exporters:  [datadog, otlphttp/langsmith]

    # Sampled pipeline for ML analysis backends
    traces/analysis:
      receivers:  [otlp]
      processors: [resource, batch]
      exporters:  [otlphttp/phoenix]

  # Enable Collector health checks
  extensions: [health_check, pprof, zpages]

Batch Processing with OTel​

For batch LLM jobs (document classification, embedding generation, bulk summarization), OTel context must be manually propagated to worker threads:

# batch/traced_batch_processor.py
import anthropic
import time
import uuid
import asyncio
from concurrent.futures import ThreadPoolExecutor
from opentelemetry import trace, context as otel_context
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("batch-processor")
client = anthropic.Anthropic()


def process_single_document(
    document: dict,
    parent_context,  # OTel context from the parent span
) -> dict:
    """
    Process one document in a batch job.
    Must receive parent_context explicitly because thread pool
    workers don't inherit context automatically.
    """
    # Restore the OTel context from the parent thread
    token = otel_context.attach(parent_context)

    try:
        with tracer.start_as_current_span("process-document") as span:
            span.set_attribute("document.id",   document["id"])
            span.set_attribute("document.type", document.get("type", "unknown"))
            span.set_attribute("document.length", len(document.get("text", "")))

            start = time.monotonic()

            response = client.messages.create(
                model="claude-haiku-4-5-20251001",  # cheap model for bulk processing
                max_tokens=256,
                temperature=0.0,
                messages=[{
                    "role": "user",
                    "content": f"Classify this document's topic in one word:\n\n{document['text'][:2000]}"
                }]
            )

            duration_ms = (time.monotonic() - start) * 1000

            classification = response.content[0].text.strip()

            span.set_attribute("document.classification", classification)
            span.set_attribute("processing.latency_ms",   round(duration_ms, 2))
            span.set_attribute("gen_ai.usage.input_tokens",  response.usage.input_tokens)
            span.set_attribute("gen_ai.usage.output_tokens", response.usage.output_tokens)
            span.set_status(Status(StatusCode.OK))

            return {
                "id":             document["id"],
                "classification": classification,
                "input_tokens":   response.usage.input_tokens,
                "output_tokens":  response.usage.output_tokens,
                "latency_ms":     round(duration_ms, 2),
            }

    except Exception as e:
        with tracer.start_as_current_span("process-document-error") as span:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
        return {"id": document["id"], "error": str(e)}
    finally:
        otel_context.detach(token)  # ALWAYS detach to avoid context leaks


def process_batch(
    documents: list[dict],
    max_workers: int = 8,
    batch_id: str = None,
) -> list[dict]:
    """
    Process a batch of documents with full OTel tracing.
    All document spans appear as children of the batch span.
    """
    if batch_id is None:
        batch_id = str(uuid.uuid4())

    with tracer.start_as_current_span("batch-processing") as batch_span:
        batch_span.set_attribute("batch.id",       batch_id)
        batch_span.set_attribute("batch.size",     len(documents))
        batch_span.set_attribute("batch.workers",  max_workers)
        batch_span.set_attribute("batch.model",    "claude-haiku-4-5-20251001")

        # Capture current OTel context to propagate to worker threads
        # This is the critical step - without it, worker threads have no trace context
        current_context = otel_context.get_current()

        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [
                executor.submit(process_single_document, doc, current_context)
                for doc in documents
            ]
            for future in futures:
                try:
                    result = future.result(timeout=60)
                    results.append(result)
                except Exception as e:
                    results.append({"error": str(e)})

        # Aggregate batch statistics
        successful   = [r for r in results if "error" not in r]
        failed       = [r for r in results if "error"     in r]
        total_input  = sum(r.get("input_tokens",  0) for r in successful)
        total_output = sum(r.get("output_tokens", 0) for r in successful)
        avg_latency  = (
            sum(r.get("latency_ms", 0) for r in successful) / len(successful)
            if successful else 0
        )

        batch_span.set_attribute("batch.success_count",    len(successful))
        batch_span.set_attribute("batch.error_count",      len(failed))
        batch_span.set_attribute("batch.total_input_tokens",  total_input)
        batch_span.set_attribute("batch.total_output_tokens", total_output)
        batch_span.set_attribute("batch.avg_latency_ms",      round(avg_latency, 2))

        if failed:
            batch_span.add_event("batch_errors", {
                "error_count": len(failed),
                "sample_error": str(failed[0].get("error", ""))[:200],
            })

        print(f"Batch {batch_id}: {len(successful)} success, {len(failed)} errors")
        print(f"  Tokens: {total_input} input, {total_output} output")
        print(f"  Avg latency: {avg_latency:.0f}ms per document")

        return results

Context Propagation Through Celery​

Celery tasks are the most common gap in distributed AI pipeline tracing - they use message queues, not HTTP, so OTel context does not propagate automatically:

# workers/celery_otel.py
import celery
import anthropic
from opentelemetry import trace, context as otel_context
from opentelemetry.propagate import inject, extract
from opentelemetry.trace import Status, StatusCode

celery_app = celery.Celery("ai-tasks")
tracer = trace.get_tracer("ai-task-worker")
client = anthropic.Anthropic()


# ── In the dispatcher (your API/web service) ──────────────────────────────────

def dispatch_rag_task(query: str, user_id: str) -> str:
    """
    Dispatch a Celery task and propagate the current OTel trace context.
    Returns the Celery task ID for status tracking.
    """
    with tracer.start_as_current_span("dispatch-rag-task") as span:
        span.set_attribute("user.id", user_id)
        span.set_attribute("task.query", query[:200])

        # Serialize the current OTel context into a dict
        # inject() writes traceparent and tracestate into the dict
        otel_headers = {}
        inject(otel_headers)

        # Pass context through task payload
        result = process_rag_task.delay(
            query=query,
            user_id=user_id,
            otel_context=otel_headers,  # OTel context travels with the task
        )

        span.set_attribute("task.id", result.id)
        return result.id


# ── In the Celery worker (different process, possibly different machine) ───────

@celery_app.task(name="process-rag-task", bind=True)
def process_rag_task(self, query: str, user_id: str, otel_context: dict = None):
    """
    Celery worker that continues the trace from the originating request.
    The task span appears as a child of the dispatcher's span in your trace view.
    """
    # Restore OTel context from the task payload
    # extract() parses traceparent and tracestate
    ctx   = extract(otel_context or {})
    token = otel_context_.attach(ctx)

    try:
        with tracer.start_as_current_span("process-rag-task") as span:
            span.set_attribute("task.id",    self.request.id)
            span.set_attribute("user.id",    user_id)
            span.set_attribute("task.query", query[:200])
            span.set_attribute("worker.hostname", self.request.hostname or "unknown")

            # The Anthropic call below appears as a grandchild span:
            # dispatch-rag-task → process-rag-task → anthropic chat
            response = client.messages.create(
                model="claude-opus-4-6",
                max_tokens=1024,
                messages=[{"role": "user", "content": query}],
            )

            answer = response.content[0].text
            span.set_attribute("output.answer_length",  len(answer))
            span.set_attribute("gen_ai.usage.input_tokens",  response.usage.input_tokens)
            span.set_attribute("gen_ai.usage.output_tokens", response.usage.output_tokens)
            span.set_status(Status(StatusCode.OK))

            return {"answer": answer, "user_id": user_id}

    except Exception as e:
        raise self.retry(exc=e, max_retries=3, countdown=5)
    finally:
        otel_context_.detach(token)  # ALWAYS detach to prevent context leaks

Common Mistakes​

:::danger Always flush before process exit in serverless functions OTel's BatchSpanProcessor buffers spans and exports them asynchronously. In AWS Lambda, Cloud Run, or one-shot scripts, the process may terminate before the buffer flushes - silently dropping all traces.

# At the end of every Lambda handler or short-lived script:
provider = trace.get_tracer_provider()
if hasattr(provider, "force_flush"):
    provider.force_flush(timeout_millis=5000)
if hasattr(provider, "shutdown"):
    provider.shutdown()

:::

:::warning Do not log full prompt text at high volume without sampling Storing full prompts in span attributes increases your observability bill (Datadog charges by span attribute volume) and creates PII risks at scale. Use sampling for full content capture:

import random

def set_content_attributes(span, query: str, answer: str, sample_rate: float = 0.10):
    """Only log full content for a sample of spans."""
    if random.random() < sample_rate:
        span.set_attribute("gen_ai.prompt",     query[:4000])
        span.set_attribute("gen_ai.completion", answer[:4000])
    else:
        # Always log metadata even without full content
        span.set_attribute("input.query_length",  len(query))
        span.set_attribute("output.answer_length", len(answer))

:::

:::danger Context propagation breaks without instrumented HTTP clients If your AI orchestrator calls downstream services using plain requests or httpx without OTel instrumentation, trace context will NOT propagate. Child spans in the downstream service will appear as disconnected root spans - not as children of your orchestrator span.

Always instrument HTTP clients: HTTPXClientInstrumentor().instrument() or RequestsInstrumentor().instrument(). Call these before creating any HTTP client instances. :::

:::warning Use the OTel Collector in production - never send directly to multiple backends Sending spans directly from your application to three different backends adds latency (three network calls per batch), complexity (three separate SDK configurations), and tight coupling. Use the OTel Collector as the single export destination. It handles routing, sampling, retries, and fan-out. Your application talks to one endpoint at localhost:4317. :::

:::warning Thread pool workers do not inherit OTel context When using ThreadPoolExecutor, each worker thread starts with an empty OTel context. You must explicitly capture the context in the main thread and pass it to each worker: current_ctx = otel_context.get_current(), then token = otel_context.attach(current_ctx) in the worker before creating spans. Always detach(token) in a finally block to prevent context leaks between worker tasks. :::

Interview Q&A​

Q1: Why use OpenTelemetry for LLM observability instead of a framework-specific SDK like LangSmith?

The strategic argument for OTel is vendor neutrality and architectural flexibility. When you instrument with OTel, you use a CNCF-graduated standard supported by every major observability vendor. If your company changes observability platforms, you change configuration - not application code. If a better AI observability tool emerges, you route to it immediately without re-instrumentation.

Beyond vendor neutrality, OTel enables unified observability: your LLM call spans live in the same trace as your database calls, external API calls, and cache lookups. When a user reports a slow response, you don't correlate between LangSmith (for the LLM span) and Datadog (for the rest) - it's one trace tree showing the complete picture.

The tradeoff: OTel has more setup than framework-specific SDKs. @traceable in LangSmith is one decorator. OTel requires understanding TracerProvider, SpanProcessor, and exporters - roughly 30 lines of setup code. The OpenInference auto-instrumentors reduce this significantly, but OTel still has higher initial friction. For small teams starting out, LangSmith/Langfuse is faster to ship. For platform engineering teams building AI infrastructure that must outlive current tool choices, OTel is the more defensible choice.

Q2: Explain the OpenInference semantic conventions for LLM spans. Why do they matter for the ecosystem?

OpenInference is a set of attribute naming conventions for LLM and AI spans, developed by Arize AI and now being merged into the core OTel GenAI specification. Core attributes:

  • gen_ai.system: which AI provider ("anthropic", "openai")
  • gen_ai.request.model: the model requested
  • gen_ai.usage.input_tokens: prompt tokens consumed
  • gen_ai.usage.output_tokens: completion tokens
  • retrieval.query: the retrieval query string
  • retrieval.documents: JSON array of retrieved documents
  • embedding.model_name: which embedding model was used

They matter for three reasons: (1) Shared tooling - when all tools agree on attribute names, pre-built dashboards work without customization. Phoenix's UMAP visualization reads retrieval.documents from any instrumentor that follows OpenInference, because the attribute name is standardized. (2) Interoperability - you can send spans from multiple different instrumentors (Anthropic, LangChain, LlamaIndex) to the same backend and they all appear with consistent attribute names. (3) Industry standardization - the GenAI attributes are now proposed for the core OTel specification, which means Datadog, Honeycomb, and other major backends will add native UI support for these attributes, turning them into first-class citizens in the broader observability ecosystem.

Q3: How do you implement distributed tracing across a multi-service AI pipeline?

A multi-service AI pipeline - frontend → orchestrator → retrieval service → embedding service → LLM gateway → response - needs all hops connected in one trace tree. OTel does this via W3C TraceContext propagation through HTTP headers (traceparent, tracestate).

When the orchestrator makes an HTTP call to the retrieval service, the instrumented HTTP client injects a traceparent header containing {version}-{trace_id}-{parent_span_id}-{flags}. The retrieval service's OTel middleware reads this header and creates its spans as children of the orchestrator's span. The trace ID is identical throughout - one unified tree in your observability backend.

Setup steps: (1) In each service, install HTTP server instrumentation: FastAPIInstrumentor.instrument_app(app). (2) In each service, install HTTP client instrumentation: HTTPXClientInstrumentor().instrument(). (3) That is all - propagation is automatic.

Edge cases that break propagation: message queues (Kafka, SQS, Celery) don't use HTTP headers. You must manually inject(headers) when producing messages and extract(headers) when consuming. gRPC services need opentelemetry-instrumentation-grpc. WebSocket connections need custom header injection at the handshake.

Q4: What is the difference between head sampling and tail sampling in OTel?

Head sampling makes the keep/drop decision at the beginning of a trace, before any work is done. The decision is based only on information available at request start (trace ID, initial attributes). The TraceIdRatioBased(0.10) sampler is head sampling - it makes a deterministic random decision based on the trace ID. Simple and low overhead, but you cannot base the decision on what happened in the request (you don't know yet).

Tail sampling makes the keep/drop decision after the trace is complete, once you know: did it error? Was it slow? Did it involve an enterprise user? This allows much more precise sampling strategies - "keep 100% of error traces, 100% of slow traces, 5% of everything else."

The cost: tail sampling requires buffering completed spans before the export decision. The OTel Collector's tail_sampling processor does this, buffering up to num_traces in memory and waiting decision_wait seconds for all spans to arrive before deciding.

For AI applications: always use tail sampling in production. You want 100% of error traces (LLM failures, content policy violations, connection timeouts), 100% of traces with enterprise users, and 5-10% of healthy traffic. Head sampling at 10% would drop 90% of errors - exactly the wrong trade-off.

Q5: How do you propagate OTel context through Celery or other async task queues?

Standard OTel context propagation works via HTTP headers. Celery tasks use a message broker (Redis, RabbitMQ) and don't pass HTTP headers between producer and consumer. You must explicitly serialize and deserialize the context.

Producing (in your API endpoint or service):

from opentelemetry.propagate import inject
headers = {}
inject(headers)  # writes traceparent and tracestate into the dict
task.delay(payload=data, otel_context=headers)

Consuming (in the Celery worker):

from opentelemetry import context as otel_context
from opentelemetry.propagate import extract

ctx   = extract(otel_context_headers)  # restores the trace context
token = otel_context.attach(ctx)
try:
    with tracer.start_as_current_span("worker-task") as span:
        # This span is now a child of the producing span
        pass
finally:
    otel_context.detach(token)  # ALWAYS detach in finally to prevent leaks

The detach(token) in a finally block is critical. If you forget it, subsequent tasks processed by the same worker thread inherit the wrong trace context - causing spans from different requests to appear in the same trace. This creates confusing, incorrect trace trees that are hard to debug.

For teams using Celery heavily, the opentelemetry-instrumentation-celery package provides automatic propagation without manual inject/extract, at the cost of a framework dependency.

© 2026 EngineersOfAI. All rights reserved.