Anomaly Detection in LLM Outputs: Alert Systems
Anomaly detection automatically flags unexpected behavior in production LLM systems. A sudden spike in output latency, a drop in coherence scores, an unusual distribution of refusals—these are anomalies that warrant investigation. This article covers statistical and ML approaches to detecting anomalies in time-series data, setting alert thresholds that balance sensitivity with false-alarm rates, and integrating alerts into on-call workflows.
Why anomalies matter for LLM observability
LLM systems are complex: they involve models, APIs, infrastructure, and prompts. Any component can degrade silently. A model provider's API might experience latency increase; a prompt change might degrade quality; a user behavior shift might expose an untrained domain. Without anomaly detection, these issues go unnoticed until user complaints arrive or dashboards are manually reviewed.
Anomalies in LLM systems typically manifest as deviations from baseline in:
- Latency: p95 response time increases from 200ms to 800ms.
- Quality scores: Mean coherence drops from 0.85 to 0.72.
- Error rates: Exceptions increase from 0.1% to 2%.
- Refusal rates: Model refuses 50% of requests instead of 5%.
- Distribution shifts: Output length distribution changes dramatically.
Detecting these automatically and alerting quickly is the difference between a 5-minute incident and a 1-hour user outage.
Statistical anomaly detection methods
1. Z-score method (univariate)
Compute the z-score of each observation: z = (x - mean) / stddev. Flag as anomaly if |z| > threshold (typically 2-3). Simple, interpretable, and works for Gaussian data.
# Z-score anomaly detection
import numpy as np
from collections import deque
class ZScoreAnomalyDetector:
def __init__(self, window_size=100, threshold_z=2.5):
self.window_size = window_size
self.threshold_z = threshold_z
self.window = deque(maxlen=window_size)
def detect(self, value):
"""
Detect anomaly in incoming value.
Returns (is_anomaly: bool, z_score: float, msg: str)
"""
self.window.append(value)
if len(self.window) < 2:
return False, 0.0, "Not enough data"
mean = np.mean(self.window)
stddev = np.std(self.window)
if stddev == 0:
return False, 0.0, "Zero variance"
z_score = (value - mean) / stddev
is_anomaly = abs(z_score) > self.threshold_z
msg = f"z={z_score:.2f}, threshold={self.threshold_z}, {'ANOMALY' if is_anomaly else 'OK'}"
return is_anomaly, z_score, msg
# Example: monitoring latency
detector = ZScoreAnomalyDetector(window_size=100, threshold_z=2.5)
latencies = [200, 210, 195, 205, 200, 800, 210, 205] # Spike at index 5
for i, latency in enumerate(latencies):
is_anomaly, z, msg = detector.detect(latency)
print(f"Latency {i}: {latency}ms, {msg}")
2. IQR method (robust to outliers)
Use interquartile range (IQR) instead of stddev. Flag as anomaly if value < Q1 - 1.5*IQR or value > Q3 + 1.5*IQR. More robust to extreme outliers than z-score.
# IQR anomaly detection
class IQRAnomalyDetector:
def __init__(self, window_size=100, multiplier=1.5):
self.window_size = window_size
self.multiplier = multiplier
self.window = deque(maxlen=window_size)
def detect(self, value):
"""
IQR-based anomaly detection.
"""
self.window.append(value)
if len(self.window) < 4:
return False, None, "Not enough data"
q1 = np.percentile(self.window, 25)
q3 = np.percentile(self.window, 75)
iqr = q3 - q1
lower_bound = q1 - self.multiplier * iqr
upper_bound = q3 + self.multiplier * iqr
is_anomaly = value < lower_bound or value > upper_bound
msg = f"Value {value:.1f}, bounds [{lower_bound:.1f}, {upper_bound:.1f}], {'ANOMALY' if is_anomaly else 'OK'}"
return is_anomaly, (lower_bound, upper_bound), msg
# Example
detector = IQRAnomalyDetector(window_size=50, multiplier=1.5)
for latency in latencies:
is_anomaly, bounds, msg = detector.detect(latency)
print(f"Latency: {latency}ms, {msg}")
3. Isolation Forest (multivariate)
For detecting anomalies in high-dimensional data (e.g., multiple metrics together), use Isolation Forest. It isolates outliers in a decision-tree ensemble; anomalies are isolated more quickly than normal points.
# Multivariate anomaly detection with Isolation Forest
from sklearn.ensemble import IsolationForest
import numpy as np
class IsolationForestDetector:
def __init__(self, contamination=0.05):
"""
contamination: expected fraction of anomalies in training data (0-1).
"""
self.contamination = contamination
self.model = IsolationForest(contamination=contamination, random_state=42)
self.training_data = []
def fit(self, data):
"""
Train on baseline data.
data: numpy array of shape (n_samples, n_features)
"""
self.training_data = data
self.model.fit(data)
def detect(self, values):
"""
Detect anomaly in new observation.
values: numpy array of shape (n_features,)
Returns (is_anomaly: bool, anomaly_score: float)
"""
values = np.array(values).reshape(1, -1)
predictions = self.model.predict(values) # -1 for anomaly, 1 for normal
scores = self.model.score_samples(values)
is_anomaly = predictions[0] == -1
return is_anomaly, scores[0]
# Example: monitoring multiple metrics together
baseline_data = np.array([
[200, 0.85, 0.02], # [latency_ms, coherence, error_rate]
[210, 0.84, 0.01],
[195, 0.86, 0.02],
# ... more baseline samples
])
detector = IsolationForestDetector(contamination=0.05)
detector.fit(baseline_data)
# Check new observation
new_observation = [800, 0.5, 0.15] # High latency, low coherence, high error rate
is_anomaly, score = detector.detect(new_observation)
print(f"Anomaly: {is_anomaly}, Score: {score:.3f}")
4. EWMA and CUSUM (change detection)
For detecting gradual or sudden changes in a time series, use Exponentially Weighted Moving Average (EWMA) or Cumulative Sum Control Chart (CUSUM).
# EWMA for detecting gradual shifts
class EWMADetector:
def __init__(self, alpha=0.2, threshold=3.0):
"""
alpha: smoothing factor (lower = more memory).
threshold: number of standard deviations to trigger alert.
"""
self.alpha = alpha
self.threshold = threshold
self.ewma = None
self.ewma_std = None
def detect(self, value):
"""
Detect anomaly using EWMA.
"""
if self.ewma is None:
self.ewma = value
self.ewma_std = 0.0
return False, 0.0, "Initializing"
# Update EWMA
self.ewma = self.alpha * value + (1 - self.alpha) * self.ewma
# Update variance
residual = value - self.ewma
self.ewma_std = self.alpha * abs(residual) + (1 - self.alpha) * self.ewma_std
# Detect anomaly
if self.ewma_std > 0:
z_score = residual / self.ewma_std
else:
z_score = 0.0
is_anomaly = abs(z_score) > self.threshold
return is_anomaly, z_score, f"EWMA z={z_score:.2f}"
# Example
detector = EWMADetector(alpha=0.1, threshold=2.5)
for latency in latencies:
is_anomaly, z, msg = detector.detect(latency)
print(f"Latency {latency}ms: {msg}")
Configuring alert thresholds
Thresholds must balance sensitivity (catching real issues) and specificity (avoiding false alarms). A few strategies:
Absolute threshold: Alert if latency > 1000ms. Use when you have a clear SLA.
Relative threshold: Alert if latency increases by >50% vs. rolling baseline. Adapts to seasonal patterns.
Statistical threshold: Alert if z-score > 2.5 (97.5% confidence interval). More principled but assumes Gaussian data.
Composite threshold: Require multiple anomalies (e.g., latency spike AND coherence drop) to alert. Reduces false positives.
# Composite alert rules
class AlertRuleEngine:
def __init__(self):
self.detectors = {
"latency": ZScoreAnomalyDetector(window_size=100, threshold_z=2.5),
"coherence": ZScoreAnomalyDetector(window_size=100, threshold_z=2.0),
"error_rate": ZScoreAnomalyDetector(window_size=100, threshold_z=3.0),
}
def check_alerts(self, metrics):
"""
metrics: dict of {metric_name: value}
Returns list of triggered alerts.
"""
alerts = []
anomalies = {}
# Check individual metrics
for name, value in metrics.items():
if name in self.detectors:
is_anomaly, z, msg = self.detectors[name].detect(value)
anomalies[name] = is_anomaly
if is_anomaly:
alerts.append(f"{name}: {msg}")
# Composite rules: require multiple anomalies
if anomalies.get("latency") and anomalies.get("error_rate"):
alerts.append("CRITICAL: Latency spike + high error rate detected")
if anomalies.get("coherence") and anomalies.get("latency"):
alerts.append("WARNING: Low coherence + latency increase (possible API degradation)")
return alerts
# Example
rule_engine = AlertRuleEngine()
current_metrics = {
"latency": 800,
"coherence": 0.5,
"error_rate": 0.15
}
alerts = rule_engine.check_alerts(current_metrics)
for alert in alerts:
print(f"ALERT: {alert}")
Escalation and routing
Different alerts require different responses:
# Alert escalation and routing
class AlertRouter:
def __init__(self, paging_service):
self.paging_service = paging_service
def route_alert(self, alert_message, severity):
"""
severity: "info", "warning", "critical"
"""
if severity == "critical":
# Page on-call engineer immediately
self.paging_service.page_oncall(
service="llm-production",
title="Critical LLM Anomaly Detected",
body=alert_message
)
elif severity == "warning":
# Create ticket for triage
# ticket_system.create_ticket(alert_message)
pass
elif severity == "info":
# Log to dashboard only
pass
# Example
router = AlertRouter(paging_service)
router.route_alert("Coherence dropped to 0.5", severity="critical")
Preventing alert fatigue
Alert fatigue occurs when too many false alerts desensitize engineers. Mitigate by:
- Aggregating alerts: Group related anomalies (e.g., all latency spikes in 5-minute window) into one alert.
- Suppressing duplicate alerts: Don't alert twice for the same issue.
- Tuning thresholds: Use your alert history to find thresholds that reduce false positives without missing real issues.
- Contextualizing alerts: Include baseline, current value, and trend in the alert message.
# Alert deduplication and context
class SmartAlertManager:
def __init__(self, dedup_window_minutes=5):
self.dedup_window_minutes = dedup_window_minutes
self.recent_alerts = {}
def should_alert(self, alert_type, current_value, baseline_value):
"""
Decide whether to send an alert, avoiding duplicates.
"""
now = datetime.utcnow()
# Check if we've alerted on this type recently
if alert_type in self.recent_alerts:
last_alert_time = self.recent_alerts[alert_type]
if (now - last_alert_time).total_seconds() < self.dedup_window_minutes * 60:
return False, "Duplicate suppressed"
# Record alert
self.recent_alerts[alert_type] = now
# Generate contextual message
pct_change = 100.0 * (current_value - baseline_value) / baseline_value if baseline_value != 0 else 0
msg = f"{alert_type} anomaly: {current_value:.2f} ({pct_change:+.1f}% from baseline {baseline_value:.2f})"
return True, msg
# Example
manager = SmartAlertManager(dedup_window_minutes=5)
should_alert, msg = manager.should_alert("latency", 800, 200)
if should_alert:
print(f"ALERT: {msg}")
Key Takeaways
- Anomaly detection flags unexpected behavior in production LLMs before users notice issues.
- Statistical methods (z-score, IQR, EWMA) are fast and interpretable; ML methods (Isolation Forest) handle multivariate data.
- Configure thresholds carefully: absolute for hard SLAs, statistical for adaptive detection, composite for reducing false positives.
- Route alerts by severity: critical issues page on-call, warnings create tickets, info logs to dashboard.
- Prevent alert fatigue by deduplicating, contextualizing, and tuning thresholds against your alert history.
Frequently Asked Questions
How many data points do I need to establish a baseline?
100-500 for simple methods (z-score, IQR), 500+ for ML methods (Isolation Forest). More is better; aim for 2-4 weeks of production data.
Should I alert on every metric deviation, or only aggregate them?
Composite alerts (require multiple anomalies) reduce false positives but may miss single-metric issues. Start with a few key metrics (latency, coherence, error rate), then expand as you gain confidence.
Can I use anomaly detection for detecting adversarial prompts?
Partially. Adversarial prompts may cause unusual output patterns, but they're hard to detect without semantic understanding. Combine anomaly detection with safety classifiers for better coverage.
How do I handle false positives without missing real issues?
Tune thresholds on your alert history: compute precision and recall for different thresholds, then pick one that achieves your acceptable false-positive rate (e.g., max 10% false positives).
Should I tune thresholds separately per user cohort?
Yes, if cohorts have different baselines (e.g., mobile users have higher latency). Segment your baseline and train detectors per segment.