When a Python microservice misbehaves in production, you reach for logs and distributed traces. When an LLM-powered application misbehaves, most teams are flying blind. Tokens leak, latency spikes, costs explode — and all you have is a user complaint. That gap is what AI observability is designed to close.
What you actually need to trace
Classic application observability covers latency, error rate, and throughput. LLM applications need more:
- Token usage (prompt tokens, completion tokens, total) — directly tied to cost
- Model and version — a provider rollout can silently degrade quality
- Latency breakdown — time-to-first-token vs. total generation time
- Prompt fingerprint — which template version triggered this call
- Output quality signals — length, structured output validity, guard failures
Omit any of these and you will debug production issues by reading tea leaves.
Building a minimal LLM tracer in Python
You don't need an observability vendor on day one. A decorator-based tracer that writes structured JSON gives you enough to start.
import time
import json
import hashlib
import logging
from functools import wraps
from dataclasses import dataclass, asdict
from typing import Optional
logger = logging.getLogger("llm.trace")
@dataclass
class LLMSpan:
trace_id: str
model: str
prompt_tokens: int
completion_tokens: int
latency_ms: float
cost_usd: float
error: Optional[str] = None
COST_PER_1K = {
"gpt-4o": {"input": 0.005, "output": 0.015},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}
def compute_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
prices = COST_PER_1K.get(model, {"input": 0.001, "output": 0.003})
return (prompt_tokens / 1000 * prices["input"]) + \
(completion_tokens / 1000 * prices["output"])
def trace_llm(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
error_msg = None
result = None
try:
result = func(*args, **kwargs)
return result
except Exception as e:
error_msg = str(e)
raise
finally:
elapsed = (time.perf_counter() - start) * 1000
usage = getattr(result, "usage", None) if result else None
prompt_tokens = getattr(usage, "prompt_tokens", 0) if usage else 0
completion_tokens = getattr(usage, "completion_tokens", 0) if usage else 0
model = kwargs.get("model", "unknown")
span = LLMSpan(
trace_id=hashlib.md5(str(start).encode()).hexdigest()[:12],
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
latency_ms=round(elapsed, 2),
cost_usd=round(compute_cost(model, prompt_tokens, completion_tokens), 6),
error=error_msg,
)
logger.info(json.dumps(asdict(span)))
return wrapper
Wrap your API call:
@trace_llm
def call_llm(client, messages: list, model: str = "gpt-4o-mini"):
return client.chat.completions.create(
model=model,
messages=messages,
)
Every call now emits a structured JSON line. Feed that to jq, Loki, or any log aggregation tool you already have. No new infrastructure required.
Connecting to OpenTelemetry
Once your team outgrows flat logs, OpenTelemetry spans let you attach LLM traces to the parent HTTP request that triggered them.
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
provider = TracerProvider()
provider.add_span_processor(
BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otelcol:4317"))
)
trace.set_tracer_provider(provider)
otel_tracer = trace.get_tracer("llm-service")
def call_llm_otel(client, messages: list, model: str = "gpt-4o-mini"):
with otel_tracer.start_as_current_span("llm.completion") as span:
start = time.perf_counter()
response = client.chat.completions.create(model=model, messages=messages)
elapsed_ms = (time.perf_counter() - start) * 1000
usage = response.usage
span.set_attribute("llm.model", model)
span.set_attribute("llm.prompt_tokens", usage.prompt_tokens)
span.set_attribute("llm.completion_tokens", usage.completion_tokens)
span.set_attribute("llm.latency_ms", round(elapsed_ms, 2))
span.set_attribute(
"llm.cost_usd",
round(compute_cost(model, usage.prompt_tokens, usage.completion_tokens), 6)
)
return response
Your Grafana or Jaeger instance will now show LLM latency nested inside the request trace. When a specific user flow slows down, you can see exactly which LLM call is responsible — and whether it was a slow model or a bloated prompt.
Aggregating and alerting on LLM metrics
Raw spans are useful for debugging individual requests. Aggregated metrics catch systemic problems before users notice.
A small script that reads your structured logs and summarizes by model:
import json
import sys
from collections import defaultdict
def summarize_llm_logs(log_lines: list[str]) -> dict:
totals = defaultdict(lambda: {
"calls": 0, "errors": 0,
"tokens": 0, "cost_usd": 0.0, "latency_ms": []
})
for line in log_lines:
try:
span = json.loads(line)
except json.JSONDecodeError:
continue
m = span.get("model", "unknown")
totals[m]["calls"] += 1
if span.get("error"):
totals[m]["errors"] += 1
totals[m]["tokens"] += span.get("prompt_tokens", 0) + span.get("completion_tokens", 0)
totals[m]["cost_usd"] += span.get("cost_usd", 0)
totals[m]["latency_ms"].append(span.get("latency_ms", 0))
result = {}
for model, data in totals.items():
lats = sorted(data["latency_ms"])
result[model] = {
"calls": data["calls"],
"error_rate": round(data["errors"] / max(data["calls"], 1), 4),
"total_tokens": data["tokens"],
"total_cost_usd": round(data["cost_usd"], 4),
"p50_ms": lats[len(lats) // 2] if lats else 0,
"p99_ms": lats[int(len(lats) * 0.99)] if lats else 0,
}
return result
if __name__ == "__main__":
lines = sys.stdin.readlines()
summary = summarize_llm_logs(lines)
print(json.dumps(summary, indent=2))
Run it with:
grep '"model"' /var/log/app/llm.log | python3 summarize.py
Set a threshold: if p99_ms exceeds 10,000 ms or error_rate exceeds 0.02, fire an alert. These numbers beat a vague "the AI is slow today" complaint.
What the data reveals in practice
Once you have a week of traces, patterns emerge quickly:
-
Cost spikes usually trace to one prompt template that grew during a refactor. A
prompt_tokenshistogram shows this immediately. - Latency regression often coincides with a provider's infrastructure event. Check your p99 against the provider's status page — if they correlate, the bug is not yours.
- Error clusters reveal retry storms: the tracer will show 10x expected calls in a 30-second window, which points to a misconfigured backoff policy.
None of this requires a paid APM product. The security hardening checklists at AYI NEDJIMI include an LLM application security checklist that covers observability requirements alongside prompt injection and output validation controls — worth running through before you go to production.
The takeaway
AI observability is not a nice-to-have — it is the prerequisite for cost control, SLA enforcement, and meaningful debugging. Start with the structured JSON decorator above: it costs nothing, adds under 5 ms of overhead, and gives you actionable data within minutes. Layer in OpenTelemetry when you need to correlate LLM calls with the rest of your service mesh.
The hardest part is not the implementation. It is the habit: every language model call in your codebase should emit a trace span before it reaches production.
I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.
Top comments (0)