Build OpenTelemetry instrumentation for multi-step AI pipelines

Introduction

When you debug a multi-step AI pipeline with generic HTTP tracing, every failure looks the same: a 200 OK with a few hundred milliseconds of latency and no clue whether retrieval returned junk chunks, the reranker dropped the right answer, or the model hallucinated despite correct context. Teams that ship AI features without pipeline-aware instrumentation routinely burn hours per incident reading raw logs to reconstruct what a single span tree should have told them in seconds — and silently regress on quality, cost, and safety between deploys. By the end of this lesson you'll be able to build a custom OpenTelemetry span processor that stamps every pipeline stage with gen_ai.* semantic-convention attributes (model, token counts, finish reason, guardrail verdict, retrieval metadata) and wire it into a multi-step RAG pipeline so each failure is one trace query away from a root cause.

Key Terminology

• Span processor: An OpenTelemetry hook (here, AISpanProcessor) that stamps every pipeline-stage span with gen_ai.* attributes before it is exported.
• GenAI semantic conventions: The standardized gen_ai.* attribute vocabulary (model, token counts, finish reason, guardrail verdict) that makes telemetry portable across backends like Jaeger, Honeycomb, and Datadog.
• Cardinality control: The discipline of recording bounded, aggregatable attributes (e.g., query length, not raw query text) so spans do not explode the metrics backend's time-series count.

Concepts

Why Generic Instrumentation Fails for AI Pipelines

Standard distributed tracing captures HTTP request/response pairs and database queries. For an AI pipeline, this produces spans like POST /v1/completions with a duration and status code. That is almost useless when debugging why your RAG pipeline returned a hallucinated answer. You need spans that capture:

• Token counts (prompt tokens, completion tokens) to correlate cost with quality
• Model parameters (temperature, top_p, model version) to detect configuration drift
• Retrieval metadata (chunk count, similarity scores, source documents) to diagnose context quality
• Guardrail verdicts (pass/fail, violation categories, confidence scores) to trace safety regressions
• Pipeline stage ordering with parent-child relationships to reconstruct the full execution graph

OpenTelemetry's semantic conventions for GenAI (currently in experimental status under the gen_ai namespace) define standardized attribute names for these concerns. Building on these conventions ensures your telemetry is portable across backends — whether you ship traces to Jaeger, Honeycomb, Datadog, or Grafana Tempo.

The OTel GenAI Semantic Convention Namespace

Before writing any code, you must understand the attribute naming contract. The emerging gen_ai.* semantic conventions define a structured vocabulary that backends can index, query, and alert on without custom parsing:

• gen_ai.system: The AI system provider (e.g., openai, anthropic, bedrock)
• gen_ai.request.model: The model identifier requested (e.g., gpt-4o, claude-sonnet-4-20250514)
• gen_ai.request.temperature: The sampling temperature parameter
• gen_ai.request.max_tokens: The maximum token budget for the completion
• gen_ai.response.model: The actual model that served the request (may differ from requested)
• gen_ai.usage.prompt_tokens: Number of tokens in the prompt
• gen_ai.usage.completion_tokens: Number of tokens in the completion
• gen_ai.response.finish_reason: Why generation stopped (stop, length, content_filter)

For pipeline-level concerns beyond a single model call, you extend this namespace with custom attributes following the same dot-notation pattern: gen_ai.pipeline.stage, gen_ai.retrieval.chunk_count, gen_ai.guardrail.verdict. This consistency is critical — inconsistent attribute naming creates cardinality explosions in your metrics backend and makes cross-signal correlation impossible.

Attribute Hygiene and Cardinality Control

Every attribute you add to a span becomes a queryable dimension in your observability backend — and a potential cardinality bomb. Follow these rules to keep your observability cost under control:

• Never record raw user input as a span attribute. Use gen_ai.request.user_query_length (an integer) instead of gen_ai.request.user_query (an unbounded string). If you need the raw query for debugging, emit it as a span event with sampling, not as an indexed attribute.
• Use bounded enumerations for categorical attributes. gen_ai.guardrail.verdict should be pass or fail, not a free-text explanation. gen_ai.response.finish_reason has a fixed set of values defined by the model provider.
• Derive aggregatable attributes at write time. Computing gen_ai.usage.total_tokens and gen_ai.cost.estimated_usd in the span processor means your backend can aggregate these directly without post-processing.
• Prefix all custom attributes with your namespace. Attributes like gen_ai.retrieval.chunk_count are clearly AI-pipeline attributes. Avoid generic names like count or result that collide across instrumentation libraries.

From Instrumentation to MTTD/MTTR Reduction

Raw traces are necessary but not sufficient for fast incident response. The instrumentation built in this section feeds directly into the cross-signal correlation architecture covered later in this chapter. Specifically:

1. MTTD improvement: By recording gen_ai.guardrail.verdict, gen_ai.response.finish_reason, and gen_ai.usage.total_tokens as span attributes, you enable real-time alerting on anomalies — a spike in finish_reason=length across traces signals a prompt template regression before users report truncated answers.
2. MTTR improvement: When an alert fires, the trace tree gives you the complete execution path. You filter by gen_ai.pipeline.stage=retrieval to see if chunk counts dropped, then pivot to gen_ai.pipeline.stage=inference to check if the model changed. Each span's attributes are the diagnostic evidence that eliminates guesswork.
3. Cost attribution: The gen_ai.cost.estimated_usd attribute on every inference span lets you build per-pipeline, per-model, and per-user cost dashboards directly from trace data — no separate billing pipeline required.

The AISpanProcessor you built here is the foundation. In subsequent sections, you will layer AI-specific metrics collection (token histograms, cost counters, quality scores) on top of these spans, implement tail-based sampling to control observability cost without losing critical traces, and build cross-signal correlation queries that link trace anomalies to metric degradation and log errors. Every component depends on the semantic convention discipline established here — consistent attribute names are the contract that makes the entire observability stack composable.

Code Walkthrough

Building on the gen_ai.* semantic conventions above, you now turn that attribute vocabulary into a working span processor. The core abstraction is a custom AISpanProcessor that wraps each pipeline stage in a properly named span (ai.retrieval, ai.inference, ai.guardrail, …), stamps it with the standardized attribute names, and computes derived values like gen_ai.usage.total_tokens and gen_ai.cost.estimated_usd so downstream consumers query pre-computed numbers instead of reprocessing raw telemetry.

The create_pipeline_span context manager handles span lifecycle and stage tagging; enrich_model_span applies the model-call attributes and the cost/token math. Because a ReadableSpan is immutable once exported, the derived attributes are stamped on the live span during enrichment, while on_end remains the export-time hook.

Code snippetpython
1from contextlib import contextmanager
2from opentelemetry import trace
3from opentelemetry.sdk.trace import SpanProcessor, ReadableSpan
4
5MODEL_COST_PER_1K = {
6    "gpt-4o": {"prompt": 0.0025, "completion": 0.01},
7    "text-embedding-3-small": {"prompt": 0.00002, "completion": 0.0},
8}
9
10class AISpanProcessor(SpanProcessor):
11    def __init__(self, pipeline_name: str):
12        self._tracer = trace.get_tracer("ai.pipeline.instrumentation", "1.0.0")
13        self._pipeline_name = pipeline_name
14
15    @contextmanager
16    def create_pipeline_span(self, stage: str, attributes: dict | None = None):
17        attrs = {
18            "gen_ai.pipeline.name": self._pipeline_name,
19            "gen_ai.pipeline.stage": stage,
20        }
21        attrs.update(attributes or {})
22        with self._tracer.start_as_current_span(f"ai.{stage}", attributes=attrs) as span:
23            yield span
24
25    def enrich_model_span(self, span, model, prompt_tokens, completion_tokens, finish_reason):
26        rate = MODEL_COST_PER_1K.get(model, {"prompt": 0.0, "completion": 0.0})
27        cost = (prompt_tokens * rate["prompt"] + completion_tokens * rate["completion"]) / 1000
28        span.set_attribute("gen_ai.request.model", model)
29        span.set_attribute("gen_ai.usage.prompt_tokens", prompt_tokens)
30        span.set_attribute("gen_ai.usage.completion_tokens", completion_tokens)
31        span.set_attribute("gen_ai.usage.total_tokens", prompt_tokens + completion_tokens)
32        span.set_attribute("gen_ai.response.finish_reason", finish_reason)
33        span.set_attribute("gen_ai.cost.estimated_usd", round(cost, 6))
34
35    def on_start(self, span, parent_context=None): ...
36    def on_end(self, span: ReadableSpan) -> None: ...
37    def shutdown(self): ...
38    def force_flush(self, timeout_millis: int = 30000) -> bool:
39        return True
40
41# Wiring it into a stage
42processor = AISpanProcessor("rag_pipeline")
43with processor.create_pipeline_span("inference") as span:
44    processor.enrich_model_span(span, "gpt-4o", 2048, 512, "stop")

Register the processor on your TracerProvider, then run one query through the pipeline. You'll know it works when the exported ai.inference span carries gen_ai.request.model, gen_ai.usage.total_tokens (2560), and a non-zero gen_ai.cost.estimated_usd queryable by trace_id in your backend.

Do's and Don'ts

Do's

1. ✓Do stamp gen_ai.* attributes on the live span inside enrich_model_span — ReadableSpan is immutable once the span ends and enters on_end, so any set_attribute calls there silently no-op; enrichment must happen while the span is still active and mutable.
2. ✓Do name every pipeline span with the ai.<stage> convention and tag it with gen_ai.pipeline.stage via create_pipeline_span — without stage-tagged spans (ai.retrieval, ai.inference, ai.guardrail), all pipeline failures surface as a single undifferentiated trace node and root-cause isolation devolves back to reading raw logs.
3. ✓Do pre-compute gen_ai.usage.total_tokens and gen_ai.cost.estimated_usd inside enrich_model_span using MODEL_COST_PER_1K — attaching derived numbers to the live span means downstream consumers query one pre-computed field instead of re-deriving cost math from raw prompt_tokens and completion_tokens per query.

Don'ts

1. ✗Don't call span.set_attribute inside on_end — the ReadableSpan argument to on_end is the immutable export snapshot; mutations to it silently no-op, leaving gen_ai.request.model, gen_ai.response.finish_reason, and cost fields absent from every exported span.
2. ✗Don't collapse retrieval, inference, and guardrail stages into a single parent span — merging all pipeline steps under one trace node is exactly the generic HTTP tracing pattern the AISpanProcessor replaces; without per-stage spans you cannot distinguish a retrieval failure from a model hallucination in trace queries.
3. ✗Don't scatter per-model cost rates outside MODEL_COST_PER_1K — inlining rate math at individual enrich_model_span call sites means a model price change requires hunting across every instrumented stage, and risks gen_ai.cost.estimated_usd values diverging between ai.inference and ai.embedding spans.