Observability, Monitoring, Drift, and LLM Evals

Deploying a model is not the end of the work - it is the beginning of the operational work. Models degrade when the world changes. LLMs hallucinate and drift in subtle ways that only evaluations can detect. This module covers how to detect degradation before users notice, what metrics to monitor, and how to build eval harnesses for LLM-powered systems.

Why Models Degrade in Production

Three causes of production model degradation:

1. Data drift (covariate shift): the distribution of input features changes. A model trained on pre-pandemic purchase data performs worse post-pandemic because user behavior changed.
2. Concept drift: the relationship between features and labels changes. A fraud detection model trained before a new fraud technique was invented will miss the new pattern entirely.
3. Data quality degradation: an upstream data pipeline changes and features are now computed differently (or missing) - the model sees inputs it has never seen.

What to Monitor

Data health metrics:

pythonCopy
def compute_distribution_stats(df: pd.DataFrame) -> dict:
    stats = {}
    for col in df.select_dtypes(include='number').columns:
        stats[col] = {
            'mean': df[col].mean(),
            'std': df[col].std(),
            'p50': df[col].quantile(0.5),
            'p95': df[col].quantile(0.95),
            'null_rate': df[col].isnull().mean(),
        }
    return stats

# Compare current stats against baseline
baseline_stats = load_json('baseline_stats.json')
current_stats = compute_distribution_stats(today_features)

for col, curr in current_stats.items():
    if col not in baseline_stats:
        continue
    base = baseline_stats[col]
    # Z-score based drift detection for mean shift
    z_score = abs(curr['mean'] - base['mean']) / (base['std'] + 1e-8)
    if z_score > 3.0:
        alert(f"Drift detected in {col}: z={z_score:.2f}")
    # Null rate spike
    if curr['null_rate'] > base['null_rate'] + 0.05:
        alert(f"Null rate spike in {col}: {curr['null_rate']:.3f} vs baseline {base['null_rate']:.3f}")

Population Stability Index (PSI): a standard metric in financial ML for measuring how much a feature distribution has shifted.

pythonCopy
import numpy as np

def population_stability_index(baseline: np.ndarray, current: np.ndarray, bins: int = 10) -> float:
    """PSI < 0.1: stable; 0.1–0.2: moderate shift; > 0.2: significant drift."""
    baseline_counts, bin_edges = np.histogram(baseline, bins=bins)
    current_counts, _ = np.histogram(current, bins=bin_edges)

    baseline_pct = (baseline_counts / len(baseline)) + 1e-8
    current_pct = (current_counts / len(current)) + 1e-8

    psi = np.sum((current_pct - baseline_pct) * np.log(current_pct / baseline_pct))
    return float(psi)

# Monitor PSI daily for key features
for feature in ['purchase_amount', 'session_duration', 'days_since_last_login']:
    psi = population_stability_index(baseline_df[feature], today_df[feature])
    log_metric(f"psi_{feature}", psi)
    if psi > 0.2:
        alert(f"High PSI for {feature}: {psi:.3f}")

Model output monitoring: track the distribution of prediction scores over time. A shift in the mean predicted probability often signals upstream data changes before you have labels to confirm performance degradation.

pythonCopy
from scipy import stats

def score_distribution_alert(baseline_scores: np.ndarray, current_scores: np.ndarray, p_threshold=0.01):
    """Kolmogorov-Smirnov test for distribution shift in model scores."""
    ks_stat, p_value = stats.ks_2samp(baseline_scores, current_scores)
    if p_value < p_threshold:
        alert(f"Score distribution shift detected: KS={ks_stat:.3f}, p={p_value:.4f}")

Outcome metrics with delay: when ground truth labels arrive (e.g., did the user churn?), compare model predictions against actuals and track AUC, precision, and recall over time. This is the most reliable signal but has inherent lag.

LLM Evaluation Harnesses

LLMs require a different approach to evaluation because outputs are natural language - there is no single "correct" answer.

Reference-free metrics: evaluate properties of the output without ground truth.

pythonCopy
def evaluate_llm_response(response: str, context: str, question: str) -> dict:
    """Use an LLM judge to evaluate response quality."""
    judge_prompt = f"""Evaluate this AI response on three criteria.
Question: {question}
Retrieved Context: {context}
Response: {response}

Rate each criterion from 1-5:
1. Faithfulness: Does the response stay within the provided context without adding unsupported claims?
2. Relevance: Does the response directly address the question?
3. Completeness: Does the response cover all key points in the context relevant to the question?

Respond with JSON: {{"faithfulness": N, "relevance": N, "completeness": N, "reasoning": "..."}}"""

    judge_response = call_llm(judge_prompt, temperature=0)
    return parse_json_safely(judge_response)

# Run evals on a test set
eval_results = []
for item in eval_dataset:
    result = evaluate_llm_response(
        response=rag_system.answer(item['question']),
        context='\n'.join(item['retrieved_chunks']),
        question=item['question']
    )
    eval_results.append(result)

mean_faithfulness = np.mean([r['faithfulness'] for r in eval_results])
print(f"Mean faithfulness: {mean_faithfulness:.2f}/5.0")

Regression detection: track whether specific known-good test cases continue to pass after model or prompt changes.

pythonCopy
# Critical test cases that must always pass
golden_tests = [
    {
        "question": "What is the return policy?",
        "expected_contains": "30 days",
        "expected_not_contains": ["competitor", "other companies"],
    },
    {
        "question": "How do I contact support?",
        "expected_contains": "[email protected]",
    }
]

def run_regression_suite(rag_system, tests: list[dict]) -> bool:
    passed = 0
    for test in tests:
        response = rag_system.answer(test['question'])
        if test.get('expected_contains') not in response:
            print(f"FAIL: Missing '{test['expected_contains']}' in response")
            continue
        if any(bad in response for bad in test.get('expected_not_contains', [])):
            print(f"FAIL: Found forbidden content in response")
            continue
        passed += 1
    print(f"Passed {passed}/{len(tests)} regression tests")
    return passed == len(tests)

Alerting and On-Call

Define alert thresholds before deploying, not after your first incident:

pythonCopy
# Structured alert definition
monitoring_rules = [
    {"metric": "psi_purchase_amount", "threshold": 0.2, "severity": "warning"},
    {"metric": "val_auc_rolling_7d", "threshold": 0.03, "type": "degradation", "severity": "critical"},
    {"metric": "null_rate_user_id", "threshold": 0.01, "severity": "critical"},
    {"metric": "prediction_mean", "change_threshold": 0.1, "window": "1d", "severity": "warning"},
]

Route critical alerts to on-call immediately. Route warnings to a monitoring dashboard for daily review.

Common Mistakes and Bad Instincts

Monitoring only when you have labels. Labels often arrive days or weeks later. Monitor feature distributions and prediction score distributions, which are available immediately.

Using accuracy as your monitoring metric on imbalanced data. A churn model with 5% positive rate can have "stable" 95% accuracy while recall drops from 0.80 to 0.40 - the model has stopped detecting churners.

Not running evals before and after prompt changes. A prompt change to an LLM feature that looks better on manual inspection might regress on the specific test cases that matter. Always run the eval suite on a set of representative inputs before deploying.

Ignoring the "silent" failures. The worst production ML failures are the ones where the model produces outputs, the API returns 200, but the outputs are wrong. Add business metric tracking (conversion rate, support ticket rate, etc.) alongside model metrics so you catch silent failures.

Where to Go Next

• Module 26 (MLOps CI/CD) covers the automated testing that runs before deployment to prevent many of the issues this module detects after deployment.
• Module 28 (AI System Design) covers incorporating observability requirements into system architecture from the start.