Monitoring & Observability

◆AI◆Advanced

Introduction

Monitoring and observability are critical for production LLM systems. Unlike traditional software, LLMs produce non-deterministic outputs, have complex failure modes (hallucination, bias, jailbreaking), and incur variable costs based on token usage. Without robust monitoring, you are flying blind.

Effective LLM observability encompasses: request tracing (end-to-end visibility into each LLM call), latency and throughput tracking, token usage and cost accounting, output quality monitoring, drift detection (model behavior changes over time), user feedback collection, and alerting on anomalies.

This guide covers how to set up comprehensive monitoring for LLM applications using purpose-built tools (LangFuse, LangSmith, Weights & Biases, MLflow) and custom instrumentation.

LLM Observability Fundamentals

Observability for LLM systems requires tracking rich, structured data that goes beyond traditional APM metrics. Every LLM call should be traced with: the full input prompt, the generated output, token counts (prompt + completion), latency breakdown, model version, temperature and other parameters, user ID or session, and any guardrail triggers.

Metric	What It Measures	Why It Matters
TTFT	Time to first token	User-perceived latency for streaming
TPOT	Time per output token	Generation throughput
Token/s	Tokens generated per second	Serving efficiency
Cost/Request	Monetary cost per call	Budget tracking and optimization
Error Rate	Percentage of failed calls	API reliability, model degradation
Hallucination Rate	Rate of factually incorrect outputs	Output quality and trustworthiness

ℹ

info

Start by instrumenting every LLM call with request IDs and structured logging. You can add dedicated observability tools later. The key is to capture all inputs, outputs, and metadata from day one — you cannot retroactively capture data you didn't log.

Tracing with LangFuse

LangFuse provides open-source observability for LLM applications with tracing, evaluation, and prompt management. It creates detailed traces of every LLM call with nested spans for complex chains.

langfuse-tracing.py

Python

1	from langfuse import Langfuse
2	from langfuse.decorators import observe
3
4	langfuse = Langfuse(
5	secret_key="sk-lf-...",
6	public_key="pk-lf-...",
7	host="https://cloud.langfuse.com"
8	)
9
10	@observe()
11	def generate_response(prompt: str, user_id: str) -> str:
12	# Trace will automatically capture inputs, outputs, and timing
13	response = client.chat.completions.create(
14	model="gpt-4o",
15	messages=[{"role": "user", "content": prompt}],
16	temperature=0.7,
17	max_tokens=1024,
18	user=user_id,
19	)
20	return response.choices[0].message.content
21
22	# Score generation quality
23	@observe()
24	def evaluate_response(prompt: str, response: str) -> float:
25	# Add custom scores to traces
26	score = llm_judge(prompt, response)
27	langfuse.score(
28	trace_id=langfuse.get_current_trace_id(),
29	name="response_quality",
30	value=score
31	)
32	return score
33
34	response = generate_response("What is RAG?", user_id="user_123")
35	quality = evaluate_response("What is RAG?", response)

Custom Instrumentation

custom-tracer.py

Python

1	import time
2	import uuid
3	from dataclasses import dataclass, field
4	from typing import Optional
5
6	@dataclass
7	class LLMCallTrace:
8	call_id: str = field(default_factory=lambda: str(uuid.uuid4()))
9	model: str = ""
10	prompt: str = ""
11	response: str = ""
12	prompt_tokens: int = 0
13	completion_tokens: int = 0
14	total_tokens: int = 0
15	latency_ms: float = 0.0
16	temperature: float = 0.0
17	status: str = "pending"
18
19	class LLMTracer:
20	def __init__(self):
21	self.traces: list = []
22
23	def trace_call(self, fn, args, *kwargs):
24	trace = LLMCallTrace()
25	start = time.time()
26	try:
27	result = fn(args, *kwargs)
28	trace.status = "success"
29	trace.response = result["content"]
30	trace.model = result.get("model", "unknown")
31	trace.prompt_tokens = result.get("prompt_tokens", 0)
32	trace.completion_tokens = result.get("completion_tokens", 0)
33	trace.total_tokens = trace.prompt_tokens + trace.completion_tokens
34	except Exception as e:
35	trace.status = f"error: {str(e)}"
36	result = None
37	finally:
38	trace.latency_ms = (time.time() - start) * 1000
39	self.traces.append(trace)
40	return result
41
42	def get_stats(self) -> dict:
43	if not self.traces:
44	return {}
45	latencies = [t.latency_ms for t in self.traces if t.status == "success"]
46	tokens = [t.total_tokens for t in self.traces if t.status == "success"]
47	return {
48	"total_calls": len(self.traces),
49	"error_rate": sum(1 for t in self.traces if t.status != "success") / len(self.traces),
50	"avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
51	"avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
52	"total_tokens": sum(tokens),
53	}

Quality Monitoring

Monitoring LLM output quality in production requires automated evaluation that runs on every response or a sampled subset. Key quality dimensions include: factual accuracy, instruction following, safety compliance, formatting adherence, and response coherence.

quality-monitor.py

Python

1	import random
2
3	class QualityMonitor:
4	def __init__(self, sample_rate: float = 0.1):
5	self.sample_rate = sample_rate
6	self.scores = []
7
8	def should_evaluate(self) -> bool:
9	return random.random() < self.sample_rate
10
11	def evaluate_output(self, prompt: str, response: str, expected: str = None) -> dict:
12	eval_result = llm_judge(
13	system="Evaluate the quality of this response on a scale of 1-5.
14	Assess: helpfulness, accuracy, safety, and clarity.",
15	prompt=f"Prompt: {prompt}
16	Response: {response}",
17	temperature=0.1
18	)
19	score = eval_result["score"]
20	self.scores.append(score)
21	return {
22	"score": score,
23	"details": eval_result["details"],
24	"requires_review": score < 3.0
25	}
26
27	def get_quality_report(self) -> dict:
28	if not self.scores:
29	return {"status": "no_data"}
30	recent = self.scores[-100:]
31	return {
32	"average_score": sum(recent) / len(recent),
33	"min_score": min(recent),
34	"max_score": max(recent),
35	"samples_evaluated": len(self.scores),
36	"pct_below_threshold": sum(1 for s in recent if s < 3.0) / len(recent) * 100
37	}

Drift Detection

Model drift occurs when the distribution of inputs, outputs, or model behavior changes over time. This can happen due to API model updates (the provider changes the underlying model), data distribution shifts (users change how they interact), or model degradation (performance decay).

drift-detection.py

Python

1	import numpy as np
2	from scipy import stats
3
4	class DriftDetector:
5	def __init__(self, reference_window: int = 1000):
6	self.reference_window = reference_window
7	self.reference_scores = []
8	self.current_scores = []
9
10	def add_reference(self, scores: list):
11	self.reference_scores = scores[-self.reference_window:]
12
13	def add_sample(self, score: float):
14	self.current_scores.append(score)
15	if len(self.current_scores) > self.reference_window:
16	self.current_scores.pop(0)
17
18	def detect_drift(self, threshold: float = 0.05) -> dict:
19	if len(self.reference_scores) < 30 or len(self.current_scores) < 30:
20	return {"drift_detected": False, "reason": "insufficient_data"}
21
22	# Kolmogorov-Smirnov test
23	stat, p_value = stats.ks_2samp(
24	self.reference_scores,
25	self.current_scores
26	)
27
28	# Population stability index
29	ps = self._psi(self.reference_scores, self.current_scores)
30
31	drift_detected = p_value < threshold or ps > 0.2
32	return {
33	"drift_detected": drift_detected,
34	"ks_statistic": float(stat),
35	"ks_p_value": float(p_value),
36	"psi": float(ps),
37	"reference_mean": float(np.mean(self.reference_scores)),
38	"current_mean": float(np.mean(self.current_scores)),
39	}
40
41	def _psi(self, expected, actual, bins=10):
42	expected_hist, _ = np.histogram(expected, bins=bins, range=(0, 5))
43	actual_hist, _ = np.histogram(actual, bins=bins, range=(0, 5))
44	expected_pct = expected_hist / len(expected) + 1e-10
45	actual_pct = actual_hist / len(actual) + 1e-10
46	return np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))

Alerting

Effective alerting notifies the right people about the right problems at the right time. Alert fatigue from noisy alerts leads to ignored warnings. Define clear thresholds and escalation paths for different severity levels.

Severity	Condition	Response
Critical	Error rate > 10% or latency > 30s	Page on-call immediately
Warning	Error rate > 5% or drift detected	Notify team in Slack within 5 min
Info	Quality score drop > 10%	Daily digest report

✓

best practice

Set up a monitoring dashboard that shows: real-time request volume, error rate, latency percentiles (p50, p95, p99), token usage, cost accumulation, quality score trend, and recent flagged outputs. Review this dashboard daily during the first month of a new deployment, then weekly once stable.

Tools Comparison

Tool	Type	Key Features	Pricing
LangFuse	Open source	Tracing, eval, prompt mgmt, playground	Free tier + cloud or self-hosted
LangSmith	SaaS	LangChain integration, datasets, feedback	Free tier + pay-as-you-go
W&B	SaaS	Experiment tracking, model registry, eval	Free for personal + team plans
MLflow	Open source	Experiment tracking, model registry, serving	Free (self-hosted)

$Blueprint — Engineering Documentation·Section ID: AI-MON-01·Revision: 1.0

X GitHub YouTube Instagram Facebook Buy me a coffee Sponsor