Observability, Drift, Feedback Loops, and LLM Evals

Shipping a model is the beginning, not the end. Production ML systems degrade silently: data pipelines change upstream schemas, user behavior shifts, business definitions evolve, and the model's predictions slowly become less reliable. Without observability, you discover these failures from a business metric regression weeks after the problem started.

This module covers the monitoring patterns that catch model quality issues early, and the evaluation infrastructure needed to measure LLM system quality over time.

The Three Layers of ML Observability

1. Infrastructure layer: latency, error rates, throughput - same as standard web services
2. Data layer: feature distribution shift, missing values, schema changes
3. Model layer: prediction distribution shift, label drift, performance degradation

Most teams instrument layer 1 well and layers 2-3 poorly. Layers 2-3 are where ML-specific failures live.

Feature Distribution Monitoring

Statistical tests detect when production inputs drift from training distribution:

pythonCopy
import numpy as np
from scipy import stats
from dataclasses import dataclass

@dataclass
class DriftReport:
    feature: str
    statistic: float
    p_value: float
    is_drift: bool
    severity: str

def ks_drift_test(
    training_values: np.ndarray,
    production_values: np.ndarray,
    feature_name: str,
    alpha: float = 0.05,
) -> DriftReport:
    """Kolmogorov-Smirnov test for distribution shift in continuous features."""
    stat, p_value = stats.ks_2samp(training_values, production_values)
    is_drift = p_value < alpha
    severity = "high" if stat > 0.2 else ("medium" if stat > 0.1 else "low")
    return DriftReport(feature=feature_name, statistic=stat, p_value=p_value,
                       is_drift=is_drift, severity=severity)

def psi_score(expected: np.ndarray, actual: np.ndarray, n_bins: int = 10) -> float:
    """Population Stability Index: PSI > 0.2 indicates significant shift."""
    breakpoints = np.percentile(expected, np.linspace(0, 100, n_bins + 1))
    expected_pct = np.histogram(expected, breakpoints)[0] / len(expected)
    actual_pct = np.histogram(actual, breakpoints)[0] / len(actual)

    # Avoid division by zero
    expected_pct = np.where(expected_pct == 0, 0.0001, expected_pct)
    actual_pct = np.where(actual_pct == 0, 0.0001, actual_pct)

    return np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))

def monitor_features(training_df, production_df, feature_cols):
    reports = []
    for col in feature_cols:
        report = ks_drift_test(training_df[col].values, production_df[col].values, col)
        psi = psi_score(training_df[col].values, production_df[col].values)
        reports.append({**report.__dict__, "psi": psi})
    return reports

Thresholds: PSI < 0.1 = no shift; 0.1–0.2 = minor (monitor); > 0.2 = significant (investigate).

Prediction Distribution Monitoring

Without ground-truth labels (common in production), monitor the prediction distribution itself:

pythonCopy
import pandas as pd
from collections import Counter

def monitor_prediction_distribution(
    baseline_predictions: list[int],
    current_predictions: list[int],
    classes: list[str],
) -> dict:
    baseline_dist = Counter(baseline_predictions)
    current_dist = Counter(current_predictions)

    baseline_pct = {c: baseline_dist.get(i, 0) / len(baseline_predictions)
                    for i, c in enumerate(classes)}
    current_pct = {c: current_dist.get(i, 0) / len(current_predictions)
                   for i, c in enumerate(classes)}

    shifts = {c: current_pct[c] - baseline_pct[c] for c in classes}
    return {
        "baseline": baseline_pct,
        "current": current_pct,
        "shift": shifts,
        "max_shift": max(abs(v) for v in shifts.values()),
    }

# Alert if positive-class rate shifts more than 5 percentage points
report = monitor_prediction_distribution(baseline_preds, today_preds, ["negative", "positive"])
if report["max_shift"] > 0.05:
    alert(f"Prediction distribution shifted: {report['shift']}")

LLM Evaluation: Measuring Quality Without Ground Truth

For LLM features (summarization, extraction, generation), traditional metrics often do not apply. The dominant patterns:

LLM-as-judge: use a stronger model to evaluate outputs against defined criteria:

pythonCopy
from openai import OpenAI

client = OpenAI()

def llm_judge(
    query: str,
    response: str,
    criteria: list[str],
) -> dict:
    criteria_text = "\n".join(f"{i+1}. {c}" for i, c in enumerate(criteria))
    prompt = f"""
Evaluate the response on each criterion. For each, give a score from 1-5 and a brief reason.

Query: {query}

Response: {response}

Criteria:
{criteria_text}

Return JSON: {{"scores": [{{"criterion": "...", "score": N, "reason": "..."}}]}}
"""
    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.0,
    )
    import json
    return json.loads(result.choices[0].message.content)

# Example: evaluate a RAG answer
eval_result = llm_judge(
    query="What is the refund policy?",
    response="You can request a refund within 30 days of purchase.",
    criteria=[
        "The response directly answers the user's question",
        "The response is factually accurate based on the provided context",
        "The response is appropriately concise (not too long, not too short)",
    ],
)

Reference-based metrics (when you have ground truth):

pythonCopy
from rouge_score import rouge_scorer
from bert_score import score as bert_score

# ROUGE for summarization
scorer = rouge_scorer.RougeScorer(["rouge1", "rouge2", "rougeL"], use_stemmer=True)
scores = scorer.score(reference_summary, generated_summary)
print(f"ROUGE-L: {scores['rougeL'].fmeasure:.3f}")

# BERTScore for semantic similarity (more robust than ROUGE)
P, R, F1 = bert_score([generated], [reference], lang="en", model_type="distilbert-base-uncased")
print(f"BERTScore F1: {F1.mean().item():.3f}")

Building an Eval Harness

Run systematic evals on every code change and daily in production:

pythonCopy
import json
import time
from pathlib import Path

class EvalHarness:
    def __init__(self, eval_set_path: str, judge_fn):
        with open(eval_set_path) as f:
            self.eval_set = json.load(f)
        self.judge_fn = judge_fn

    def run(self, system_fn, tag: str = "") -> dict:
        results = []
        for example in self.eval_set:
            start = time.perf_counter()
            response = system_fn(example["query"])
            latency_ms = (time.perf_counter() - start) * 1000

            judgment = self.judge_fn(example["query"], response, example.get("criteria", []))
            results.append({
                "query": example["query"],
                "response": response,
                "judgment": judgment,
                "latency_ms": latency_ms,
            })

        avg_score = sum(
            sum(s["score"] for s in r["judgment"]["scores"]) / len(r["judgment"]["scores"])
            for r in results
        ) / len(results)

        return {
            "tag": tag,
            "n_examples": len(results),
            "avg_score": avg_score,
            "avg_latency_ms": sum(r["latency_ms"] for r in results) / len(results),
            "results": results,
        }

Store eval results in your observability system (or a simple database). Plot average score over time. Alert if score drops more than 5% below the baseline.

Feedback Loops: Closing the Loop with Human Data

pythonCopy
# Capture implicit feedback from user behavior
def log_user_feedback(session_id: str, query: str, response: str, action: str):
    """
    action: 'thumbs_up', 'thumbs_down', 'copied', 'ignored', 'followed_up'
    """
    feedback_record = {
        "session_id": session_id,
        "query": query,
        "response": response,
        "action": action,
        "timestamp": time.time(),
    }
    # Store for later analysis and retraining
    store_feedback(feedback_record)

# Use feedback data to build better eval sets and fine-tuning datasets
# thumbs_down + follow-up = failure case to add to eval set
# thumbs_up + copied = high-quality example for fine-tuning

Production feedback is the most valuable data you have. Instrument your UI to capture it from day one, even before you have a plan to use it.

Common Mistakes and Bad Instincts

Monitoring only infrastructure metrics (latency, error rate) and not model quality. A model can be serving at low latency with zero errors while producing increasingly wrong predictions. Latency metrics do not tell you whether the model is useful.

Running evals only before deployment, not continuously. Prod data distribution shifts over days and weeks. A model that was great at launch may be mediocre 6 months later. Run evals daily on a sample of production traffic.

Using a single aggregate score to hide regressions. An average score of 3.8/5 might hide the fact that the model now scores 4.5 on easy queries but 2.0 on an important subset. Always slice eval results by query type, user segment, or difficulty category.

Where to Go Next

• ai-system-design-quality-cost-latency-and-safety-tradeoffs: decide when monitoring signals should trigger retraining vs. prompt updates vs. architectural changes
• fine-tuning-adaptation-and-when-not-to-fine-tune: use production feedback data to improve the model when observability shows degradation