Production monitoring for AI features
Why this matters
LLM features degrade in ways that traditional software does not. A web API either returns a 200 or it does not. An LLM feature can return a 200 with a response that is confidently wrong, subtly less helpful than it was last week, or increasingly expensive to generate. None of those failure modes show up in your error rate.
The engineering problem: AI quality is gradual and non-deterministic. A prompt that worked well in March may work less well in June because the underlying model was updated, the distribution of user queries shifted, or the retrieved context became stale. Production monitoring is the discipline of detecting these drifts before they become user complaints.
Core concepts
The three layers of AI observability:
- System health: Are requests succeeding? Are latencies within SLO? This layer uses the same tooling as traditional API monitoring.
- Quality health: Are responses meeting quality thresholds? Are faithfulness scores declining? This layer requires AI-specific metrics computed from sampled responses.
- Business health: Are users getting value from the feature? Are follow-up questions decreasing (suggesting first-response resolution)?
Sampling strategy for quality monitoring. Scoring every production response with an LLM judge is expensive. A practical strategy:
- Score 100% of responses with fast deterministic checks (length, format validation, toxicity classifier)
- Score 5-10% of responses with an LLM judge for quality metrics
- Score 100% of flagged responses (user corrections, thumbs-down)
- Run batch eval on the golden dataset nightly to detect model drift
Key metrics to track:
| Metric | Alert condition | Frequency |
|---|---|---|
| Pass rate | < 85% rolling 24h | Hourly |
| Average faithfulness | < 0.75 (14-day rolling) | Daily |
| Toxicity rate | > 0.5% sampled | Real-time |
| P95 latency | > 2x SLO baseline | Real-time |
| Token cost per request | > 20% above 14-day avg | Daily |
| Provider error rate | > 2% over 5 minutes | Real-time |
Drift detection. Quality drift is a sustained change in a quality metric over time, distinct from a one-time spike. Two patterns:
- Statistical process control: alert when the current value is more than 2 standard deviations below the 30-day mean.
- Reference set drift: re-run your golden evaluation dataset weekly and compare to baseline. A decline of more than 5 percentage points is a signal worth investigating.
SLOs for AI features. Define both system SLOs (latency, availability, error rate) and quality SLOs (minimum pass rate, maximum toxicity rate) before launch. When quality SLOs are breached, the response is a product decision (degrade gracefully, show a fallback) not just an engineering escalation.
Working example
A monitoring pipeline that processes traces and emits quality metrics:
from dataclasses import dataclass
from collections import deque
from typing import Callable
import statistics
import time
@dataclass
class QualityEvent:
trace_id: str
feature_name: str
timestamp_ms: int
pass_rate: float
faithfulness: float | None
latency_ms: int
prompt_tokens: int
completion_tokens: int
class QualityMonitor:
def __init__(self, window_hours: int = 24, alert_fn: Callable | None = None):
self.window_ms = window_hours * 3600 * 1000
self.alert_fn = alert_fn or (lambda name, data: print(f'ALERT {name}: {data}'))
self._events: deque[QualityEvent] = deque()
self._baselines: dict[str, dict] = {}
def record(self, event: QualityEvent) -> None:
self._events.append(event)
self._evict_old()
self._check_alerts(event.feature_name)
def _evict_old(self) -> None:
cutoff = int(time.time() * 1000) - self.window_ms
while self._events and self._events[0].timestamp_ms < cutoff:
self._events.popleft()
def _check_alerts(self, feature_name: str) -> None:
events = [e for e in self._events if e.feature_name == feature_name]
if len(events) < 10:
return
avg_pass = statistics.mean(e.pass_rate for e in events)
baseline = self._baselines.get(feature_name, {})
if avg_pass < 0.80:
self.alert_fn('quality_degradation',
{'feature': feature_name, 'avg_pass_rate': avg_pass})
baseline_pass = baseline.get('pass_rate', avg_pass)
if avg_pass < baseline_pass - 0.05:
self.alert_fn('drift_detected', {
'feature': feature_name, 'current': avg_pass,
'baseline': baseline_pass, 'delta': avg_pass - baseline_pass,
})
def set_baseline(self, feature_name: str, metrics: dict) -> None:
self._baselines[feature_name] = metrics
def summary(self, feature_name: str) -> dict:
events = [e for e in self._events if e.feature_name == feature_name]
if not events:
return {}
latencies = sorted(e.latency_ms for e in events)
n = len(latencies)
return {
'feature': feature_name,
'sample_size': n,
'avg_pass_rate': statistics.mean(e.pass_rate for e in events),
'p95_latency_ms': latencies[int(n * 0.95)],
'total_cost_estimate': sum(
e.prompt_tokens * 0.000003 + e.completion_tokens * 0.000015 for e in events
),
}
Common mistakes
-
Monitoring only system metrics. Error rate and latency being green does not mean the feature is working well. Quality metrics require a separate instrument.
-
No baseline to compare against. An average faithfulness score of 0.78 is meaningless without knowing whether it was 0.85 last week. Always store a baseline before making changes.
-
Alerting on every sample. Individual quality scores are noisy. Alert on rolling averages over meaningful sample sizes (at least 50 events).
-
Cost monitoring as an afterthought. Teams regularly discover that a new prompt version tripled their token count after they see the cloud bill. Track cost per request in real time.
-
No on-call playbook for quality alerts. Document the investigation steps before you need them.
Try it yourself
Implement a DriftDetector class that maintains a 30-day rolling window of daily quality scores for a feature. It should compute a z-score for the most recent day's score relative to the 30-day distribution and return an is_drift: bool signal when the z-score exceeds -2.0. Test it with a simulated series of 30 stable days followed by 3 days of declining quality.