Introduction
When your LLM application hits production, visibility becomes survival. A single slow model call cascades into timeouts, an unexpected cost spike destroys margins, and tokens disappear into a black box with no way to correlate which user triggered which LLM call. Production-grade LLM observability requires structured tracing, cost attribution, and anomaly detection — not just error logs.
What to Measure in an LLM System
Before picking tools, define the four dimensions of LLM observability:
- Latency: Time to first token (TTFT), total response time, p50/p95/p99 per endpoint
- Tokens: Input and output counts per request, per user, per feature
- Cost: USD per request, per user, per day — attributed to specific features
- Quality: Response relevance, hallucination rate, user satisfaction proxy signals
Most teams instrument latency well and completely miss cost attribution and quality scoring.
OpenTelemetry Spans for LLM Calls
Wrap every LLM API call in an OpenTelemetry span with semantic attributes. This enables correlation with traces from databases, queues, and downstream services:
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
tracer = trace.get_tracer("llm-service")
def call_llm_with_tracing(prompt: str, user_id: str, model: str = "claude-3-5-sonnet-20241022") -> dict:
with tracer.start_as_current_span("llm.completion") as span:
span.set_attributes({
"llm.model": model,
"llm.user_id": user_id,
"llm.prompt_length": len(prompt),
})
start = time.time()
try:
client = anthropic.Anthropic()
response = client.messages.create(
model=model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
latency_ms = int((time.time() - start) * 1000)
input_tokens = response.usage.input_tokens
output_tokens = response.usage.output_tokens
# Claude Sonnet 3.5 pricing
cost_usd = (input_tokens * 0.003 + output_tokens * 0.015) / 1000
span.set_attributes({
"llm.input_tokens": input_tokens,
"llm.output_tokens": output_tokens,
"llm.total_tokens": input_tokens + output_tokens,
"llm.cost_usd": cost_usd,
"llm.latency_ms": latency_ms,
"llm.finish_reason": response.stop_reason,
})
span.set_status(Status(StatusCode.OK))
return {
"text": response.content[0].text,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": cost_usd,
"latency_ms": latency_ms,
}
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
raiseExport spans to Jaeger, Grafana Tempo, Datadog, or Honeycomb using the OTLP exporter.
Trace Correlation Across Microservices
The power of observability appears when you correlate a user session with every LLM call, tool execution, and database query it triggered. Propagate a request ID through all service calls:
from opentelemetry import trace, context
def create_request_context(user_id: str, session_id: str) -> dict:
return {
"request_id": str(uuid.uuid4()),
"user_id": user_id,
"session_id": session_id,
}
def process_user_request(ctx: dict, query: str) -> str:
with tracer.start_as_current_span("user_request") as span:
span.set_attributes({
"request.id": ctx["request_id"],
"user.id": ctx["user_id"],
"session.id": ctx["session_id"],
})
# LLM call inherits the span context automatically
llm_result = call_llm_with_tracing(query, ctx["user_id"])
# Tool calls, database queries — all appear in the same trace
tool_result = call_tool_with_context(ctx, llm_result["text"])
span.add_event("tools_executed")
return tool_result
def call_tool_with_context(ctx: dict, instruction: str) -> str:
with tracer.start_as_current_span("tool_execution") as span:
span.set_attributes({"request.id": ctx["request_id"]})
# All tool spans share the same trace — visible in waterfall view
return f"Tool result for: {instruction[:50]}"When debugging a user complaint, search by request.id in your tracing backend to see every span — LLM, tool, database — in chronological order.
Adaptive Sampling for High-Volume Systems
At 1,000 requests/second, tracing everything costs $50k+/month in ingestion fees. Sample intelligently: always capture errors and high-cost requests; sample normal traffic at 5–10%:
class AdaptiveLLMSampler:
def __init__(
self,
default_rate: float = 0.05, # 5% of normal requests
error_rate: float = 1.0, # 100% of errors
slow_threshold_ms: int = 5000, # Always trace slow calls
slow_rate: float = 1.0,
cost_threshold_usd: float = 0.10, # Always trace expensive calls
):
self.default_rate = default_rate
self.error_rate = error_rate
self.slow_threshold_ms = slow_threshold_ms
self.slow_rate = slow_rate
self.cost_threshold = cost_threshold_usd
def should_sample(self, attributes: dict) -> bool:
# Always sample errors
if attributes.get("error"):
return random.random() < self.error_rate
# Always sample slow requests
latency = attributes.get("llm.latency_ms", 0)
if latency > self.slow_threshold_ms:
return random.random() < self.slow_rate
# Always sample expensive requests
cost = attributes.get("llm.cost_usd", 0)
if cost > self.cost_threshold:
return True
# Default sampling
return random.random() < self.default_rateCost Attribution Per User and Feature
Knowing total LLM spend is insufficient — you need to know which users and features drive cost:
from collections import defaultdict
from datetime import datetime
class CostAttributor:
def __init__(self):
self._records: list[dict] = []
# Pricing per 1K tokens
PRICING = {
"claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
"claude-3-haiku-20240307": {"input": 0.00025, "output": 0.00125},
}
def record(self, user_id: str, feature_id: str, model: str, input_tok: int, output_tok: int):
rates = self.PRICING.get(model, {"input": 0, "output": 0})
cost = (input_tok * rates["input"] + output_tok * rates["output"]) / 1000
self._records.append({
"user_id": user_id,
"feature_id": feature_id,
"model": model,
"cost_usd": cost,
"ts": datetime.utcnow(),
})
def top_users_by_cost(self, limit: int = 10, days: int = 30) -> list[dict]:
cutoff = datetime.utcnow().timestamp() - days * 86400
by_user: dict[str, float] = defaultdict(float)
for r in self._records:
if r["ts"].timestamp() > cutoff:
by_user[r["user_id"]] += r["cost_usd"]
ranked = sorted(by_user.items(), key=lambda x: x[1], reverse=True)
return [{"user_id": u, "cost_usd": round(c, 4)} for u, c in ranked[:limit]]
def cost_by_feature(self, days: int = 30) -> dict[str, float]:
cutoff = datetime.utcnow().timestamp() - days * 86400
by_feature: dict[str, float] = defaultdict(float)
for r in self._records:
if r["ts"].timestamp() > cutoff:
by_feature[r["feature_id"]] += r["cost_usd"]
return {k: round(v, 4) for k, v in by_feature.items()}Anomaly Detection on Latency and Cost
Automate alerting when metrics deviate from baseline. Use z-score detection on a rolling window:
from collections import deque
class LLMAnomalyDetector:
def __init__(self, window_size: int = 100, z_threshold: float = 3.0):
self.window_size = window_size
self.z_threshold = z_threshold
self._latencies = deque(maxlen=window_size)
self._costs = deque(maxlen=window_size)
def _z_score(self, values: deque, new_value: float) -> float:
if len(values) < 10:
return 0.0
mean = sum(values) / len(values)
variance = sum((v - mean) ** 2 for v in values) / len(values)
stddev = math.sqrt(variance) if variance > 0 else 0.0
if stddev == 0:
return 0.0
return abs(new_value - mean) / stddev
def check_latency(self, latency_ms: float) -> bool:
"""Returns True if latency is anomalous."""
is_anomaly = self._z_score(self._latencies, latency_ms) > self.z_threshold
self._latencies.append(latency_ms)
return is_anomaly
def check_cost(self, cost_usd: float) -> bool:
"""Returns True if cost is anomalous (e.g., prompt injection causing huge output)."""
is_anomaly = self._z_score(self._costs, cost_usd) > self.z_threshold
self._costs.append(cost_usd)
return is_anomaly
detector = LLMAnomalyDetector()
def handle_llm_response(result: dict) -> None:
if detector.check_latency(result["latency_ms"]):
print(f"[ALERT] Latency anomaly: {result['latency_ms']}ms")
if detector.check_cost(result["cost_usd"]):
print(f"[ALERT] Cost anomaly: ${result['cost_usd']:.4f}")Key Takeaways
- Wrap every LLM API call in an OpenTelemetry span with
llm.model,llm.input_tokens,llm.output_tokens,llm.cost_usd, andllm.latency_msattributes. - Propagate a
request_idthrough all services so a single user request is traceable across LLM, tool, and database spans. - Adaptive sampling at 5% default, 100% on errors and slow requests, reduces ingestion cost by 90% while preserving signal.
- Cost attribution by
user_idandfeature_idis essential for identifying which features are unprofitable. - Z-score anomaly detection on a 100-request rolling window flags latency and cost spikes within seconds.
- Instrument before you need it — adding observability during an incident is too late.
- Alert on p95 latency regression (more than 20% above baseline) and cost-per-request spikes (more than 3x baseline).
- Export metrics to Grafana dashboards and set up PagerDuty/Opsgenie escalation for critical anomalies.