OpenTelemetry for AI Systems: LLM and Agent Observability (2026)

# OpenAI and Anthropic
pip install opentelemetry-instrumentation-openai
pip install opentelemetry-instrumentation-anthropic

# LangChain and LlamaIndex
pip install opentelemetry-instrumentation-langchain
pip install opentelemetry-instrumentation-llamaindex

from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from opentelemetry.instrumentation.langchain import LangChainInstrumentor

# Call once at startup, before creating any clients
OpenAIInstrumentor().instrument()
AnthropicInstrumentor().instrument()
LangChainInstrumentor().instrument()

# All subsequent API calls are traced automatically
import openai
client = openai.OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this document..."}],
)
# → span: gen_ai.usage.input_tokens=245, gen_ai.usage.output_tokens=89, duration=1.2s

OpenAIInstrumentor().instrument(capture_message_content=True)

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

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

def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    with tracer.start_as_current_span("gen_ai.chat") as span:
        span.set_attributes({
            "gen_ai.system": "openai",
            "gen_ai.operation.name": "chat",
            "gen_ai.request.model": model,
            "gen_ai.request.temperature": 0.7,
        })

        # Record prompt as event (not attribute) to control size and privacy
        span.add_event("gen_ai.user.message", {
            "gen_ai.prompt.role": "user",
            "gen_ai.prompt.content": prompt[:500],  # cap length
        })

        try:
            response = openai_client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
            )
            completion = response.choices[0].message.content

            span.set_attributes({
                "gen_ai.response.model": response.model,
                "gen_ai.usage.input_tokens": response.usage.prompt_tokens,
                "gen_ai.usage.output_tokens": response.usage.completion_tokens,
                "gen_ai.response.finish_reasons": [response.choices[0].finish_reason],
            })

            span.add_event("gen_ai.assistant.message", {
                "gen_ai.completion.role": "assistant",
                "gen_ai.completion.content": completion[:500],
            })

            return completion

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

def retrieve_context(query: str, top_k: int = 5) -> list[str]:
    with tracer.start_as_current_span("gen_ai.embeddings") as embed_span:
        embed_span.set_attributes({
            "gen_ai.system": "openai",
            "gen_ai.operation.name": "embeddings",
            "gen_ai.request.model": "text-embedding-3-small",
        })
        embedding = openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=query,
        )
        embed_span.set_attribute(
            "gen_ai.usage.input_tokens",
            embedding.usage.prompt_tokens,
        )
        vector = embedding.data[0].embedding

    with tracer.start_as_current_span("db.vector_search") as search_span:
        search_span.set_attributes({
            "db.system": "chromadb",
            "db.operation.name": "query",
            "db.vector.top_k": top_k,
        })
        results = vector_db.query(query_embeddings=[vector], n_results=top_k)
        search_span.set_attribute(
            "db.vector.results_count",
            len(results["documents"][0]),
        )
        return results["documents"][0]

def run_agent(task: str) -> str:
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attributes({
            "gen_ai.system": "openai",
            "agent.name": "research-agent",
            "agent.task": task[:200],
        })

        context_docs = retrieve_context(task)  # creates its own child spans

        messages = [
            {"role": "system", "content": "You are a research assistant."},
            {"role": "user", "content": task},
        ]

        tool_calls_count = 0

        while True:
            response = openai_client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=TOOLS,
            )
            choice = response.choices[0]

            if choice.finish_reason == "tool_calls":
                for tool_call in choice.message.tool_calls:
                    with tracer.start_as_current_span("agent.tool_call") as tool_span:
                        tool_span.set_attributes({
                            "agent.tool.name": tool_call.function.name,
                            "agent.tool.call_id": tool_call.id,
                        })
                        result = execute_tool(tool_call)
                        tool_span.set_attribute(
                            "agent.tool.result_length",
                            len(str(result)),
                        )
                    messages.append({
                        "role": "tool",
                        "content": str(result),
                        "tool_call_id": tool_call.id,
                    })
                    tool_calls_count += 1
            else:
                span.set_attribute("agent.tool_calls_total", tool_calls_count)
                return choice.message.content

agent.run (1850ms)
├── gen_ai.embeddings (38ms)             ← text-embedding-3-small
├── db.vector_search (45ms)              ← ChromaDB query
├── gen_ai.chat (920ms)                  ← gpt-4o, decides to use tools
│   ├── agent.tool_call: web_search (310ms)
│   └── agent.tool_call: calculator (8ms)
└── gen_ai.chat (530ms)                  ← gpt-4o, final answer

from opentelemetry import metrics

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

token_counter = meter.create_counter(
    "gen_ai.client.token.usage",
    unit="token",
    description="Token usage by model and token type",
)

duration_histogram = meter.create_histogram(
    "gen_ai.client.operation.duration",
    unit="s",
    description="LLM call duration",
)

def tracked_llm_call(prompt: str, model: str) -> str:
    import time
    attrs = {"gen_ai.system": "openai", "gen_ai.request.model": model}
    start = time.time()

    try:
        response = openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
        token_counter.add(
            response.usage.prompt_tokens,
            {**attrs, "gen_ai.token.type": "input"},
        )
        token_counter.add(
            response.usage.completion_tokens,
            {**attrs, "gen_ai.token.type": "output"},
        )
        return response.choices[0].message.content
    finally:
        duration_histogram.record(time.time() - start, attrs)

processors:
  transform:
    trace_statements:
      - context: spanevent
        statements:
          # Remove prompt and completion content from all events
          - delete_matching_keys(attributes, "gen_ai.prompt.content")
          - delete_matching_keys(attributes, "gen_ai.completion.content")

  # Hash for correlation without exposure
  attributes:
    actions:
      - key: gen_ai.prompt.content
        action: hash
        hash_salt: ${env:HASH_SALT}

from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased

# Sample 10% of root spans; child spans inherit the decision
sampler = ParentBased(root=TraceIdRatioBased(0.10))

provider = TracerProvider(sampler=sampler, resource=resource)