Quality Monitoring › Code Implementation
Language AI HandbookAll chapters Quality Monitoring › Code ImplementationOne section · Sections in this chapter7 / 10 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
This section implements a practical quality monitoring system covering metric computation, drift detection, and alert generation. We use synthetic data to simulate a production deployment where quality degrades gradually over time, matching the pattern you would see after an input distribution shift or a slow knowledge staleness effect.
Setup and Imports
import numpy as np
# Set reproducible seed
rng = np.random.default_rng(42)Simulating a Production Quality Stream
We simulate 500 time steps representing a deployment that runs normally for the first 200 steps, then gradually degrades. The degradation follows a linear ramp, mimicking the kind of slow input distribution shift that is common in production but hard to notice without monitoring.
def simulate_quality_stream(n_steps=500, drift_start=200, rng=None):
"""
Simulate a stream of quality metric values with a gradual drift.
Returns bleu_scores and bertscore_values as numpy arrays.
"""
if rng is None:
rng = np.random.default_rng(0)
bleu_scores = []
bertscore_values = []
for t in range(n_steps):
if t < drift_start:
# Stable production quality
bleu = rng.normal(loc=0.42, scale=0.05)
bert = rng.normal(loc=0.85, scale=0.03)
else:
# Gradual degradation after drift_start
drift_magnitude = (t - drift_start) / (n_steps - drift_start)
bleu = rng.normal(loc=0.42 - 0.15 * drift_magnitude, scale=0.06)
bert = rng.normal(loc=0.85 - 0.08 * drift_magnitude, scale=0.04)
bleu_scores.append(max(0.0, min(1.0, bleu)))
bertscore_values.append(max(0.0, min(1.0, bert)))
return np.array(bleu_scores), np.array(bertscore_values)
bleu_scores, bertscore_values = simulate_quality_stream(
n_steps=500, drift_start=200, rng=rng
)Total time steps: 500 Pre-drift BLEU : mean=0.4211, std=0.0493 Post-drift BLEU : mean=0.3425, std=0.0721 Mean BLEU drop : 0.0786
The simulated stream shows a clear drop in mean BLEU after step 200, while the standard deviation remains similar. This is a classic location shift: the model's outputs are less similar to references on average, but the output-to-output variability is unchanged. A monitoring system that tracks only the variance of outputs would miss this entirely, which is why mean-sensitive detectors like CUSUM are important alongside variance-sensitive ones.
Visualizing the Quality Stream
The raw metric values are noisy enough that step-by-step comparisons would be misleading: many individual post-drift values are higher than many individual pre-drift values simply due to sampling variability. The rolling mean makes the trend clearly visible. In a production monitoring dashboard, the rolling mean is the signal you monitor; the raw values are useful for diagnostics but not for threshold-based alerting.
KS Test for Distribution Shift
We implement a sliding-window KS test that compares a reference window to the most recent window of the same size. When the KS statistic exceeds a threshold and the p-value falls below a significance level, the test reports drift. The reference window is fixed at the initial 100 observations; the current window slides forward one step at a time.
def sliding_window_ks_test(scores, ref_size=100, window_size=50, alpha=0.05):
"""
Apply a two-sample KS test between a fixed reference window and
a sliding current window. Returns detected drift points.
"""
reference = scores[:ref_size]
drift_points = []
ks_stats = []
p_values = []
for start in range(ref_size, len(scores) - window_size + 1):
current = scores[start : start + window_size]
ks_stat, p_val = stats.ks_2samp(reference, current)
ks_stats.append(ks_stat)
p_values.append(p_val)
if p_val < alpha:
drift_points.append(start + window_size // 2)
return np.array(ks_stats), np.array(p_values), drift_points
ks_stats, p_values, bleu_drift_points = sliding_window_ks_test(
bleu_scores, ref_size=100, window_size=50, alpha=0.05
)KS test results (BLEU, alpha=0.05): Reference window size : 100 steps Current window size : 50 steps First drift detected : step 253 Total drift detections: 223 Max KS statistic : 0.8500 Min p-value observed : 3.50e-25
The KS test detects drift shortly after step 200, where the true degradation begins. The first detection slightly lags the actual drift start because the test requires observing a full current window before accumulating sufficient evidence. The lag is approximately window_size / 2 steps in expectation, which for our 50-step window means a typical lag of about 25 steps.
CUSUM Sequential Detector
The CUSUM detector is more sensitive to gradual drift than the windowed KS test because it processes each observation immediately rather than waiting for a full window. Every new metric value is incorporated into the running statistic, so the detector can alarm within a handful of steps of the drift starting rather than waiting for a full window to fill.
def cusum_detector(scores, target_mean, k=0.01, h=5.0):
"""
Cumulative sum control chart for detecting downward drift.
k: allowance (sensitivity), h: threshold (specificity).
Returns alarm time indices and the CUSUM statistic series.
"""
s_minus = 0.0 # cumulative sum for downward drift
alarms = []
s_values = []
for t, x in enumerate(scores):
s_minus = max(0.0, s_minus + (target_mean - x - k))
s_values.append(s_minus)
if s_minus > h:
alarms.append(t)
# Reset after alarm to detect subsequent drift episodes
s_minus = 0.0
return alarms, np.array(s_values)
# Use pre-drift mean as the target
target_mean = bleu_scores[:200].mean()
cusum_alarms, cusum_values = cusum_detector(
bleu_scores, target_mean=target_mean, k=0.01, h=5.0
)CUSUM detector results (BLEU): Target mean : 0.4211 Allowance k : 0.01 Threshold h : 5.0 First alarm : step 351 Total alarms raised : 4
Visualizing Drift Detection
The CUSUM plot reveals an important behavioral property: after the alarm fires and the statistic resets to zero, it climbs again quickly because the degradation is ongoing. This produces a series of alarms rather than a single alarm, which in practice would drive escalating alert severity or automated rollback. The KS test, by contrast, saturates at very low p-values once drift is fully established; the p-value provides a measure of evidence strength rather than a count of events.
Population Stability Index
We compute PSI to quantify how much the distribution has shifted between the reference period and the current period, binning BLEU scores into 10 buckets. PSI gives a scalar summary that is easy to threshold for alerting: below 0.1 is stable, 0.1 to 0.25 warrants investigation, above 0.25 requires action.
def compute_psi(reference, current, n_bins=10, epsilon=1e-6):
"""
Population Stability Index between two distributions.
Bins are defined by quantiles of the reference distribution.
"""
display_edges = np.quantile(reference, np.linspace(0, 1, n_bins + 1))
bin_edges = display_edges.copy()
# Capture values beyond the reference range in the two tail bins.
bin_edges[0] = -np.inf
bin_edges[-1] = np.inf
ref_counts, _ = np.histogram(reference, bins=bin_edges)
curr_counts, _ = np.histogram(current, bins=bin_edges)
ref_pct = ref_counts / ref_counts.sum()
curr_pct = curr_counts / curr_counts.sum()
# Replace zeros to avoid log(0)
ref_pct = np.where(ref_pct == 0, epsilon, ref_pct)
curr_pct = np.where(curr_pct == 0, epsilon, curr_pct)
psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
return psi, ref_pct, curr_pct, display_edges
reference_window = bleu_scores[:200]
current_window = bleu_scores[350:] # Late-stage degraded period
psi_value, ref_pct, curr_pct, bin_edges = compute_psi(
reference_window, current_window
)PSI Results (BLEU: reference vs. current window): PSI value : 4.7114 Interpretation: Major shift -- action required
The grouped bar chart makes the shift concrete. The reference bars sit near 0.10 because the ten bins were defined from reference quantiles, so each contains roughly one tenth of the reference observations. The current window piles probability into the lowest bins and leaves much less in the higher bins. This is exactly the signature of quality degradation: the model still produces some good outputs, but the proportion of low-quality outputs has increased substantially.
Alert Engine
We build a multi-metric alert engine that consumes metric streams and fires alerts when the rolling mean deviates from the baseline by more than a specified number of standard deviations. The engine supports initializing any number of named metrics, updating them one observation at a time, and maintaining a history of all alerts for post-hoc analysis.
class QualityAlertEngine:
"""
Multi-metric alert engine with rolling statistics and threshold levels.
"""
def __init__(self, window_size=50, warn_sigma=2.0, crit_sigma=3.0):
self.window_size = window_size
self.warn_sigma = warn_sigma
self.crit_sigma = crit_sigma
self.history = {}
self.alerts = []
def initialize_metric(self, name, baseline_values):
self.history[name] = {
"baseline_mean": float(np.mean(baseline_values)),
"baseline_std": float(np.std(baseline_values)),
"window": deque(maxlen=self.window_size),
}
def update(self, t, metrics: dict):
fired = []
for name, value in metrics.items():
if name not in self.history:
continue
info = self.history[name]
info["window"].append(value)
if len(info["window"]) < self.window_size:
continue
current_mean = np.mean(info["window"])
deviation = (info["baseline_mean"] - current_mean) / (
info["baseline_std"] + 1e-9
)
if deviation >= self.crit_sigma:
level = "CRITICAL"
elif deviation >= self.warn_sigma:
level = "WARNING"
else:
level = None
if level:
fired.append(
{
"t": t,
"metric": name,
"level": level,
"current_mean": round(current_mean, 4),
"baseline_mean": round(info["baseline_mean"], 4),
"deviation_sigma": round(deviation, 2),
}
)
self.alerts.extend(fired)
return fired
engine = QualityAlertEngine(window_size=50, warn_sigma=2.0, crit_sigma=3.0)
baseline = bleu_scores[:200]
engine.initialize_metric("bleu", baseline)
engine.initialize_metric("bertscore", bertscore_values[:200])
for t in range(200, len(bleu_scores)):
engine.update(t, {"bleu": bleu_scores[t], "bertscore": bertscore_values[t]})Alert Engine Summary:
Total alerts : 136
WARNING alerts : 136
CRITICAL alerts : 0
First WARNING (t=418):
Metric : bleu
Current mean : 0.3218
Baseline mean : 0.4211
Deviation : 2.01 sigma
The alert timeline shows the intended separation between severity levels. Warning alerts appear once the rolling mean dips into the 2-to-3-sigma band. The BLEU signal does not cross the 3-sigma threshold in this run, so no critical BLEU alerts fire. That is useful behavior rather than a missing result: the on-call team receives early notice without the system overstating the severity of the observed degradation.
Regression Detection with A/B Test
When deploying a new model version, we need to determine whether any observed quality difference is statistically significant rather than random noise. We implement a one-sided Mann-Whitney U test, which asks whether the treatment (new model) is stochastically less than the control (current model). The Mann-Whitney test is preferred over a t-test for quality metrics because it makes no distributional assumptions.
def ab_regression_test(control_scores, treatment_scores, alpha=0.05):
"""
Test whether treatment (new model) regresses on a quality metric
compared to control (current model).
Returns a summary dict with test statistics and verdict.
"""
# One-sided Mann-Whitney U test: is treatment stochastically less than control?
stat, p_value = stats.mannwhitneyu(
treatment_scores, control_scores, alternative="less"
)
regressed = p_value < alpha
effect_size = treatment_scores.mean() - control_scores.mean()
return {
"control_mean": float(control_scores.mean()),
"treatment_mean": float(treatment_scores.mean()),
"effect_size": float(effect_size),
"u_statistic": float(stat),
"p_value": float(p_value),
"alpha": alpha,
"regression_detected": bool(regressed),
}
# Scenario A: new model version with genuine regression
control = rng.normal(loc=0.42, scale=0.05, size=500)
regressed_treatment = rng.normal(loc=0.38, scale=0.05, size=500)
# Scenario B: new model version without regression (noise only)
stable_treatment = rng.normal(loc=0.42, scale=0.05, size=500)
result_regressed = ab_regression_test(control, regressed_treatment)
result_stable = ab_regression_test(control, stable_treatment)Scenario A (genuine regression): Control mean : 0.4198 Treatment mean : 0.3721 Effect size : -0.0477 p-value : 0.0000 Verdict : REGRESSION DETECTED Scenario B (no regression): Control mean : 0.4198 Treatment mean : 0.4199 Effect size : +0.0001 p-value : 0.5062 Verdict : No significant regression
The test correctly identifies Scenario A as a regression and Scenario B as acceptable noise. The 500 observations per arm provide ample separation for the simulated 0.04-point effect. In practice, the required sample size depends on the metric variance, target effect size, significance level, and desired statistical power, so it should be determined with a power analysis rather than treated as a fixed minimum.
Key Parameters
The key parameters for a quality monitoring system are:
- reference window size: The period used to establish baseline statistics. Longer windows produce more stable baselines but are slower to adapt to deliberate model updates. Typical values range from two to four weeks of production data.
- detection window size: The rolling window over which current statistics are computed and compared to the baseline. Smaller windows detect drift sooner but are noisier; larger windows are more stable but introduce detection lag.
- KS test alpha: The significance threshold for rejecting the null hypothesis of equal distributions. Common choices are 0.01 or 0.05. Lower alpha reduces false alarms at the cost of slower detection.
- CUSUM k (allowance): Controls sensitivity to small shifts. Set to half the minimum shift magnitude you want to detect.
- CUSUM h (threshold): Controls the false alarm rate. Higher values mean fewer false alarms but slower detection. Tune by simulation against historical data.
- alert sigma thresholds: The number of standard deviations from the baseline mean that trigger warning (typically 2 sigma) and critical (typically 3 sigma) alerts.
- PSI bin count: The number of bins used to compute PSI. More bins capture finer distributional differences but require larger sample sizes for stable estimates. Ten to twenty bins is typical.