LLM Observability with OpenTelemetry: Tracing LLM Calls in Production
How to add OpenTelemetry tracing to LLM applications in production — instrumenting Anthropic SDK calls, tracking token usage and latency, connecting LLM traces to your existing observability stack.
Running LLMs in production without observability is flying blind. You do not know which prompts are slow, which users are hitting errors, how much each feature costs in tokens, or whether model responses are degrading.
OpenTelemetry is the right tool for this — it is the same instrumentation standard you are already using for your services, and it integrates with the same backends (Grafana Tempo, Jaeger, Datadog, Honeycomb) you already run.
What LLM Observability Needs
Traditional observability for HTTP services tracks: latency, error rate, request rate. For LLM applications you also need:
- Token usage per request (input + output, tied to cost)
- Model and parameters (which model, temperature, max_tokens)
- Prompt templates (which template triggered this call)
- Latency broken down: time to first token vs total generation time
- Stop reason (end_turn, max_tokens, tool_use — max_tokens is often a problem)
- Tool calls (which tools were invoked, in what order, how long each took)
- User/session context (which user triggered this, for debugging)
Setup
pip install anthropic opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-apiStep 1: Instrumentation Wrapper
import anthropic
import time
import functools
from typing import Optional, Any
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
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
def setup_tracing(
service_name: str,
otlp_endpoint: str = "http://localhost:4317"
) -> trace.Tracer:
"""Initialize OpenTelemetry with OTLP exporter."""
resource = Resource.create({"service.name": service_name})
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
return trace.get_tracer(service_name)
tracer = setup_tracing("my-llm-service")
class InstrumentedAnthropic:
"""
Wrapper around the Anthropic client that adds OpenTelemetry tracing
to every messages.create call.
"""
def __init__(
self,
api_key: str,
default_model: str = "claude-sonnet-5",
prompt_template: Optional[str] = None,
user_id_getter=None
):
self._client = anthropic.Anthropic(api_key=api_key)
self.default_model = default_model
self.prompt_template = prompt_template
self.user_id_getter = user_id_getter # Callable returning current user ID
def create(
self,
messages: list[dict],
model: Optional[str] = None,
max_tokens: int = 2000,
system: str = "",
tools: Optional[list] = None,
span_name: str = "llm.call",
feature: str = "unknown",
**kwargs
) -> anthropic.types.Message:
"""
Create a message with full OpenTelemetry instrumentation.
"""
model = model or self.default_model
user_id = self.user_id_getter() if self.user_id_getter else "unknown"
input_tokens_estimate = sum(len(m.get("content", "")) // 4 for m in messages)
with tracer.start_as_current_span(
span_name,
kind=SpanKind.CLIENT
) as span:
# Set standard LLM semantic convention attributes
span.set_attribute("llm.vendor", "Anthropic")
span.set_attribute("llm.model", model)
span.set_attribute("llm.max_tokens", max_tokens)
span.set_attribute("llm.feature", feature)
span.set_attribute("llm.message_count", len(messages))
span.set_attribute("llm.has_system_prompt", bool(system))
span.set_attribute("llm.has_tools", bool(tools))
span.set_attribute("user.id", user_id)
if self.prompt_template:
span.set_attribute("llm.prompt_template", self.prompt_template)
start_time = time.time()
try:
response = self._client.messages.create(
model=model,
max_tokens=max_tokens,
system=system,
messages=messages,
tools=tools or anthropic.NOT_GIVEN,
**kwargs
)
elapsed = time.time() - start_time
# Record usage metrics
span.set_attribute("llm.usage.input_tokens", response.usage.input_tokens)
span.set_attribute("llm.usage.output_tokens", response.usage.output_tokens)
span.set_attribute("llm.usage.total_tokens",
response.usage.input_tokens + response.usage.output_tokens)
span.set_attribute("llm.stop_reason", response.stop_reason)
span.set_attribute("llm.latency_ms", int(elapsed * 1000))
# Detect tool use
tool_names = [
block.name for block in response.content
if hasattr(block, "type") and block.type == "tool_use"
]
if tool_names:
span.set_attribute("llm.tool_calls", ",".join(tool_names))
span.set_attribute("llm.tool_call_count", len(tool_names))
# Warning if hitting token limits
if response.stop_reason == "max_tokens":
span.set_attribute("llm.truncated", True)
span.add_event("response_truncated", {
"max_tokens": max_tokens,
"output_tokens": response.usage.output_tokens
})
# Calculate cost (approximate)
cost_usd = (
response.usage.input_tokens * 0.000003 + # $3/M input tokens (claude-sonnet-5)
response.usage.output_tokens * 0.000015 # $15/M output tokens
)
span.set_attribute("llm.cost_usd", round(cost_usd, 6))
span.set_status(StatusCode.OK)
return response
except anthropic.RateLimitError as e:
span.set_status(StatusCode.ERROR, "Rate limit exceeded")
span.set_attribute("llm.error.type", "rate_limit")
span.record_exception(e)
raise
except anthropic.APIStatusError as e:
span.set_status(StatusCode.ERROR, f"API error: {e.status_code}")
span.set_attribute("llm.error.type", "api_error")
span.set_attribute("llm.error.status_code", e.status_code)
span.record_exception(e)
raise
except Exception as e:
span.set_status(StatusCode.ERROR, str(e))
span.record_exception(e)
raiseStep 2: Tracing Multi-Turn Conversations
For multi-turn conversations and agentic loops, you want a parent span for the entire conversation with child spans for each LLM call:
def run_traced_conversation(
user_message: str,
conversation_id: str,
user_id: str
) -> str:
"""Run a multi-turn conversation with nested span tracing."""
llm = InstrumentedAnthropic(
api_key=os.getenv("ANTHROPIC_API_KEY"),
prompt_template="chat_assistant_v2"
)
messages = []
# Parent span for the entire conversation session
with tracer.start_as_current_span(
"conversation.session",
attributes={
"conversation.id": conversation_id,
"user.id": user_id,
"conversation.initial_message_length": len(user_message)
}
) as parent_span:
messages.append({"role": "user", "content": user_message})
# First turn
response = llm.create(
messages=messages,
system="You are a helpful DevOps assistant.",
span_name="llm.call.turn_1",
feature="chat"
)
reply = response.content[0].text
messages.append({"role": "assistant", "content": reply})
parent_span.set_attribute("conversation.turns_completed", 1)
parent_span.set_attribute("conversation.total_tokens",
response.usage.input_tokens + response.usage.output_tokens)
return replyStep 3: Prometheus Metrics from OTel Data
Most OTel backends can also export metrics. For a Prometheus-friendly approach, add span metrics to track:
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.exporter.prometheus import PrometheusMetricReader
from prometheus_client import start_http_server
# Start Prometheus metrics server
start_http_server(8001)
meter_provider = MeterProvider(metric_readers=[PrometheusMetricReader()])
meter = meter_provider.get_meter("llm-metrics")
# Create counters and histograms
token_counter = meter.create_counter(
"llm_tokens_total",
description="Total LLM tokens used",
unit="tokens"
)
latency_histogram = meter.create_histogram(
"llm_request_duration_seconds",
description="LLM request latency",
unit="seconds"
)
cost_counter = meter.create_counter(
"llm_cost_usd_total",
description="Total LLM cost in USD",
unit="USD"
)
def record_llm_metrics(response: anthropic.types.Message, elapsed: float, feature: str):
"""Record metrics after each LLM call."""
labels = {"model": response.model, "feature": feature, "stop_reason": response.stop_reason}
token_counter.add(
response.usage.input_tokens + response.usage.output_tokens,
attributes={**labels, "token_type": "total"}
)
latency_histogram.record(elapsed, attributes=labels)Grafana Dashboard Queries
Once spans flow into Grafana Tempo or a similar backend:
# P99 LLM latency by feature
histogram_quantile(0.99, rate(llm_request_duration_seconds_bucket[5m])) by (feature)
# Token usage rate
rate(llm_tokens_total[1h]) by (model, feature)
# Estimated daily cost
sum(increase(llm_cost_usd_total[24h]))
# Truncated responses (max_tokens hit)
rate(llm_requests_total{stop_reason="max_tokens"}[1h]) / rate(llm_requests_total[1h])What This Catches in Production
Token waste: A feature hitting max_tokens 30% of the time means either your prompts are too long or max_tokens is set too low. The trace shows you exactly which feature.
Slow feature diagnosis: P99 latency spike in "chat" but not "code-review" tells you the chat prompt got longer or is generating more tokens.
Cost attribution: Token counters per feature show you which feature consumes most of your API budget — useful for optimization prioritization.
Tool use patterns: Tool call traces show which tools are called in which order, helping you optimize agentic loops.
More LLMOps observability? Read our LLM evaluation with LLM-as-Judge and OpenTelemetry complete guide.
Today I Fixed
Short real fixes from production — posted daily
Stay ahead of the curve
Get the latest DevOps, Kubernetes, AWS, and AI/ML guides delivered straight to your inbox. No spam — just practical engineering content.
Related Articles
LLM Context Window Management in Production: What Nobody Tells You
How to manage LLM context windows in production systems — token budgeting, conversation compression, RAG vs context stuffing, and real strategies for keeping your LLM application fast and cheap at scale.
LLM Evaluation with LLM-as-Judge: How to Measure AI Quality in Production
How to evaluate LLM output quality in production using LLM-as-Judge — building automated evaluation pipelines, scoring rubrics, and golden dataset testing with Claude API. With real code examples.
LLM Evaluation in Production — Using Claude API as Judge to Detect Quality Regressions
LLMs degrade silently. Here's how to build a weekly eval pipeline using Claude Haiku as an LLM judge, SQLite for score tracking, and GitHub Actions to alert on quality drops.