Tracing LLM calls with OpenTelemetry

import json
import os
import time

import openai
from opentelemetry import metrics, trace
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.metrics.view import ExplicitBucketHistogramAggregation, View
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import SpanKind
from fake_openai import fake_client

spans = InMemorySpanExporter()
tp = TracerProvider()
tp.add_span_processor(SimpleSpanProcessor(spans))
reader = InMemoryMetricReader()
mp = MeterProvider(metric_readers=[reader], views=[
    View(instrument_name="gen_ai.client.token.usage", aggregation=ExplicitBucketHistogramAggregation(
        [1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864])),
    View(instrument_name="gen_ai.client.operation.duration", aggregation=ExplicitBucketHistogramAggregation(
        [0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92]))])
tracer, meter = tp.get_tracer("genai.demo"), mp.get_meter("genai.demo")
token_usage = meter.create_histogram("gen_ai.client.token.usage", unit="{token}")
duration = meter.create_histogram("gen_ai.client.operation.duration", unit="s")

CAPTURE = os.environ.get("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "").lower() in (
    "true", "span_only", "span_and_event")                                      # opt-in, off by default

def traced_response(client, model, input, instructions=None, **kwargs):
    attrs = {"gen_ai.operation.name": "chat", "gen_ai.provider.name": "openai", "gen_ai.request.model": model}
    start = time.perf_counter()
    with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT, attributes=attrs) as span:
        if CAPTURE:
            span.set_attribute("gen_ai.input.messages", json.dumps([{"role": "user", "content": input}]))
        try:
            r = client.responses.create(model=model, input=input, instructions=instructions, **kwargs)
        except openai.APIError as e:
            error = type(e).__name__
            span.set_attribute("error.type", error)                   # the exception itself is recorded
            duration.record(time.perf_counter() - start, {**attrs, "error.type": error})
            raise
        finish = "stop" if r.status == "completed" else "length"
        span.set_attributes({"gen_ai.response.model": r.model, "gen_ai.response.id": r.id,
                             "gen_ai.usage.input_tokens": r.usage.input_tokens,
                             "gen_ai.usage.output_tokens": r.usage.output_tokens,
                             "gen_ai.response.finish_reasons": [finish]})
        labels = {**attrs, "gen_ai.response.model": r.model}
        token_usage.record(r.usage.input_tokens, {**labels, "gen_ai.token.type": "input"})
        token_usage.record(r.usage.output_tokens, {**labels, "gen_ai.token.type": "output"})
        duration.record(time.perf_counter() - start, labels)
        return r

client = fake_client(script=["warm-up", {"text": "Use keyset pagination.", "usage": (1310, 42)},
                             {"status": 429}], max_retries=0)
client.responses.create(model="gpt-5.6-luna", input="hi")   # first call pays one-time client setup
traced_response(client, "gpt-5.6-luna", "Why is OFFSET slow for deep pages?")
try:
    traced_response(client, "gpt-5.6-luna", "Second question")
except openai.RateLimitError:
    pass

for s in spans.get_finished_spans():
    print(f"{s.name} [{s.kind.name}] status={s.status.status_code.name}")
    for k, v in s.attributes.items():
        print(f"    {k} = {v}")
for m in reader.get_metrics_data().resource_metrics[0].scope_metrics[0].metrics:
    for p in m.data.data_points:
        tag = p.attributes.get("gen_ai.token.type") or p.attributes.get("error.type") or "ok"
        print(f"{m.name} ({tag}): count={p.count} sum={p.sum:.3g}")

import re

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import SpanKind, Status, StatusCode
from fake_openai import fake_client
from mini_rag import HybridIndex
from site_docs import load_sections

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer("docs_copilot")
index = HybridIndex(load_sections())
client = fake_client(script=["warm-up", "Keep the last key you showed and ask for rows after it [1]."])
client.responses.create(model="gpt-5.6-luna", input="hi")   # keep one-time client setup out of the trace
PRICE_IN, PRICE_OUT = 0.20 / 1e6, 1.20 / 1e6

def ask(question):
    with tracer.start_as_current_span("POST /ask", kind=SpanKind.SERVER) as root:
        root.set_attributes({"app.prompt.version": "answer-v3", "app.index.version": "2026-09-19"})
        with tracer.start_as_current_span("retrieval site-lessons", attributes={
                "gen_ai.operation.name": "retrieval", "gen_ai.data_source.id": "site-lessons"}) as r:
            hits = index.search(question, k=3)
            r.set_attributes({"app.retrieval.urls": [h.doc["url"] for h in hits], "app.retrieval.k": 3})
        with tracer.start_as_current_span("chat gpt-5.6-luna", kind=SpanKind.CLIENT, attributes={
                "gen_ai.operation.name": "chat", "gen_ai.provider.name": "openai",
                "gen_ai.request.model": "gpt-5.6-luna"}) as llm:
            sources = "\n".join(f"[{i}] {h.doc['text'][:600]}" for i, h in enumerate(hits, 1))
            resp = client.responses.create(model="gpt-5.6-luna", input=f"{sources}\nQuestion: {question}")
            u = resp.usage
            llm.set_attributes({"gen_ai.usage.input_tokens": u.input_tokens,
                                "gen_ai.usage.output_tokens": u.output_tokens,
                                "app.cost_usd": round(u.input_tokens * PRICE_IN + u.output_tokens * PRICE_OUT, 7)})
        with tracer.start_as_current_span("verify_citations") as v:
            cited = [int(n) for n in re.findall(r"\[(\d+)\]", resp.output_text)]
            grounded = bool(cited) and all(1 <= n <= len(hits) for n in cited)
            v.set_attribute("app.grounded", grounded)
            if not grounded:
                v.set_status(Status(StatusCode.ERROR, "ungrounded answer"))
        root.set_attribute("app.grounded", grounded)
        return resp.output_text

ask("Why is OFFSET slow for deep pages?")
spans = sorted(exporter.get_finished_spans(), key=lambda s: s.start_time)
root = next(s for s in spans if s.parent is None)
for s in spans:
    depth = 0 if s.parent is None else 1
    ms = (s.end_time - s.start_time) / 1e6
    offset = (s.start_time - root.start_time) / 1e6
    shown = {k: v for k, v in s.attributes.items() if not k.startswith("gen_ai.provider")}
    print(f"{'   ' * depth}{s.name:<26} +{offset:5.0f} ms {ms:6.1f} ms  {shown}")

from opentelemetry.instrumentation.openai_v2 import OpenAIInstrumentor

OpenAIInstrumentor().instrument()        # after setting up the tracer and meter providers (lesson 16)
# every client.chat.completions.create / embeddings.create call now emits GenAI spans and metrics

# prompts and completions are NOT captured unless you opt in, e.g.:
#   OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental
#   OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only