Drift

A hard AI Coding interview practice problem on DataDriven.

Domain
AI Coding
Difficulty
hard
Seniority
L6
Asked in
OpenAI

Problem

Our RAG pipeline calls a hosted embedding endpoint. Last week the vendor silently upgraded the model under the same endpoint URL; cosine distances on a held-out probe set shifted by 18% mean, and product saw a precision drop on a fixed eval set. We have no alert for this. Run tests/test_drift_alarm.py to reproduce. Build a drift-detection check in monitoring/drift.py that runs daily and fires when the embedding distribution shifts beyond a defensible threshold. Write the threshold logic so an interviewer can defend the false-positive rate.

Nothing changed. Everything moved.

Worked solution and explanation

This is statistical process control dressed up as an embedding incident. The question underneath is how far today's probe summary can sit from its own recent history before you call it a different model, and how often that answer is wrong. Most candidates write a check that catches the 18% jump. The difference between candidates is whether the check stays quiet on normal days and still sees a re-normalized vendor swap. Diff against yesterday with a bare 0.05 and the alarm fires on ordinary noise. Lean on stats.zscore for mean_norm and the helper returns 0 whenever the baseline is flat, so a model that rescales its outputs never pages anyone. The first failure trips auto-revert. The second is the incident you were hired to prevent.

The shape of the check

Today is scored against a rolling window. It is never compared to a single prior day.
  • pinned probes (source)
  • embed_probes (transform)
  • 14-day history (storage)
  • probe count gate (quality gate)
  • z-score per metric (transform)
  • alarm + reason (consumer)
  • pinned probes -> embed_probes
  • embed_probes -> probe count gate
  • 14-day history -> z-score per metric
  • probe count gate -> z-score per metric
  • z-score per metric -> alarm + reason

Step 1: Slice the window, then guard its size

history[-BASELINE_DAYS:] takes up to 14 days and never raises an error when fewer exist. A guard clause refuses to score when fewer than MIN_BASELINE_POINTS remain, because a spread computed from two points is noise that looks like a number.

Step 2: Check n before you compare anything

When the embed job drops probes, the means describe a different population. Compare today['n'] to len(PROBES) and decline to score on a mismatch. A broken probe run is a pipeline bug. It should not trigger a model revert.

Step 3: Score both contract metrics

Loop over DRIFT_METRICS, build each baseline with one flat comprehension, and compare abs(z) to Z_THRESHOLD. The threshold is expressed in the baseline's own sigma, so the same rule holds on a quiet metric and a noisy one.

Step 4: Make the reason pageable

The on-call engineer reads the reason at 3am. It should name the metric, the z value, the window length and the baseline mean, so they can judge severity without opening a notebook.

monitoring/drift.py
import math
from typing import Any, Dict, List

from monitoring.embed import embed_probes
from monitoring.history import append_today, load_history
from monitoring.probes import PROBES
from monitoring.stats import mean, stdev

# Mirrors monitoring/constraints.sql (embedding_baseline_window).
BASELINE_DAYS = 14
Z_THRESHOLD = 4.0
MIN_BASELINE_POINTS = 3
DRIFT_METRICS = ['cosine_to_anchor_mean', 'mean_norm']


def detect_drift() -> Dict[str, Any]:
    history = load_history()
    today = embed_probes()
    append_today(today)

    baseline = history[-BASELINE_DAYS:]
    if len(baseline) < MIN_BASELINE_POINTS:
        return {'alarmed': False,
                'reason': f'baseline has {len(baseline)} days; not scored'}

    if today['n'] != len(PROBES):
        return {'alarmed': False,
                'reason': f"n={today['n']}, expected {len(PROBES)}: run not comparable"}

    breaches = []
    for metric in DRIFT_METRICS:
        values = [day[metric] for day in baseline]
        z = _z_score(today[metric], values)
        if abs(z) >= Z_THRESHOLD:
            breaches.append(
                f'{metric} z={z:+.1f} vs {len(values)}-day baseline mean {mean(values):.4f}'
            )

    if breaches:
        return {'alarmed': True,
                'reason': '; '.join(breaches) + f' (threshold {Z_THRESHOLD} sigma)'}
    return {'alarmed': False,
            'reason': f'all metrics within {Z_THRESHOLD} sigma of {len(baseline)}-day baseline'}


def _z_score(value: float, baseline_values: List[float]) -> float:
    spread = stdev(baseline_values)
    if spread == 0:
        # A flat baseline that moves at all is a regime change, not zero signal.
        return 0.0 if value == mean(baseline_values) else math.inf
    return (value - mean(baseline_values)) / spread

On the seeded sudden shift, cosine_to_anchor_mean scores about z=+41. The stable day scores about -1.

Diff against yesterday
abs(today - history[-1]) > 0.05 uses a one-day window and a magic constant. It fires on ordinary jitter, stays silent on a metric whose natural spread is smaller than 0.05, and gives you no FPR figure you can defend.
Rolling z-score
(value - mean) / stdev over history[-14:] measures the threshold in units of the metric's own noise. That lets you state the false-positive rate as a number and defend it in the review.

Common follow-up questions