LLM-Specific Monitoring & Drift Detection

1. Types of Drift in LLM Systems

Drift in LLM applications occurs when the behavior of any component changes over time without intentional modification. Unlike classical ML systems where data drift and concept drift are the primary concerns, LLM systems face several additional drift categories that are unique to the LLM technology stack.

2. Prompt Drift Detection

Prompt drift occurs when the system prompt or prompt templates gradually change over time through incremental edits, accumulating special-case instructions, or growing context. Each individual change may seem harmless, but the cumulative effect can significantly alter model behavior. Without version control and monitoring, these changes are invisible.

import hashlib
import json
from datetime import datetime, timezone
from difflib import unified_diff

class PromptDriftMonitor:
    """Monitor prompt templates for unauthorized or untracked changes."""

    def __init__(self):
        self.known_hashes: dict[str, str] = {}  # template_name -> hash
        self.change_log: list[dict] = []

    def register_template(self, name: str, template: str):
        """Register a prompt template and its hash."""
        h = hashlib.sha256(template.encode()).hexdigest()[:16]
        self.known_hashes[name] = h
        return h

    def check_template(self, name: str, current_template: str) -> dict:
        """Check if a template has drifted from its registered version."""
        current_hash = hashlib.sha256(current_template.encode()).hexdigest()[:16]
        registered_hash = self.known_hashes.get(name)

        if registered_hash is None:
            return {"status": "unregistered", "name": name}

        drifted = current_hash != registered_hash
        result = {
            "name": name,
            "drifted": drifted,
            "registered_hash": registered_hash,
            "current_hash": current_hash,
            "checked_at": datetime.now(timezone.utc).isoformat(),
        }

        if drifted:
            self.change_log.append(result)

        return result

    def compute_template_metrics(self, template: str) -> dict:
        """Compute structural metrics for a prompt template."""
        return {
            "char_count": len(template),
            "word_count": len(template.split()),
            "line_count": template.count("\n") + 1,
            "variable_count": template.count("{"),
            "instruction_density": round(
                sum(1 for w in template.lower().split()
                    if w in {"must", "always", "never", "should", "ensure"})
                / max(len(template.split()), 1), 4
            ),
        }

3. Embedding Drift in RAG Systems

Embedding drift is particularly insidious in RAG systems because it can happen without any code changes. The most common cause is updating the embedding model (or the provider silently updating it), which changes the vector space. Documents embedded with the old model and queries embedded with the new model will have mismatched representations, degrading retrieval quality.

import numpy as np
from typing import Callable

class EmbeddingDriftDetector:
    """Detect drift in embedding model behavior using reference queries."""

    def __init__(self, embed_fn: Callable, reference_queries: list[str]):
        self.embed_fn = embed_fn
        self.reference_queries = reference_queries
        # Store baseline embeddings
        self.baseline_embeddings = self._compute_embeddings(reference_queries)
        self.baseline_pairwise = self._pairwise_similarities(self.baseline_embeddings)

    def _compute_embeddings(self, texts: list[str]) -> np.ndarray:
        return np.array([self.embed_fn(t) for t in texts])

    def _pairwise_similarities(self, embeddings: np.ndarray) -> np.ndarray:
        # Cosine similarity matrix
        norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
        normalized = embeddings / norms
        return normalized @ normalized.T

    def check_drift(self, threshold: float = 0.05) -> dict:
        """Re-embed reference queries and compare to baseline.

        Drift is detected when the pairwise similarity structure changes
        beyond the threshold, indicating the embedding model has changed.
        """
        current_embeddings = self._compute_embeddings(self.reference_queries)
        current_pairwise = self._pairwise_similarities(current_embeddings)

        # Compare pairwise similarity matrices
        diff = np.abs(self.baseline_pairwise - current_pairwise)
        mean_diff = float(np.mean(diff))
        max_diff = float(np.max(diff))

        # Also check direct cosine similarity of same-query embeddings
        direct_sims = []
        for b, c in zip(self.baseline_embeddings, current_embeddings):
            sim = np.dot(b, c) / (np.linalg.norm(b) * np.linalg.norm(c))
            direct_sims.append(float(sim))

        return {
            "drift_detected": mean_diff > threshold,
            "mean_pairwise_diff": round(mean_diff, 6),
            "max_pairwise_diff": round(max_diff, 6),
            "mean_direct_similarity": round(np.mean(direct_sims), 4),
            "min_direct_similarity": round(min(direct_sims), 4),
            "recommendation": (
                "Re-index all documents with new embedding model"
                if mean_diff > threshold
                else "No action needed"
            ),
        }

4. Output Quality Monitoring

Output quality monitoring samples production responses and evaluates them against quality criteria. Because you cannot evaluate every response in real time (LLM-based evaluation is too slow and expensive), the typical approach is to sample a fraction of requests, evaluate them asynchronously, and track quality metrics over time on a dashboard.

import random
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class QualityWindow:
    """Sliding window for tracking quality metrics over time."""
    window_size: int = 100
    scores: deque = field(default_factory=lambda: deque(maxlen=100))
    alert_threshold: float = 0.7
    degradation_threshold: float = 0.05  # alert if mean drops by this much
    baseline_mean: Optional[float] = None

    def add_score(self, score: float):
        self.scores.append(score)
        if self.baseline_mean is None and len(self.scores) >= self.window_size:
            self.baseline_mean = sum(self.scores) / len(self.scores)

    def check_degradation(self) -> dict:
        if len(self.scores) < 10:
            return {"status": "insufficient_data"}

        current_mean = sum(self.scores) / len(self.scores)
        recent_mean = sum(list(self.scores)[-20:]) / min(20, len(self.scores))

        result = {
            "current_mean": round(current_mean, 4),
            "recent_mean": round(recent_mean, 4),
            "baseline_mean": round(self.baseline_mean, 4) if self.baseline_mean else None,
            "below_threshold": recent_mean < self.alert_threshold,
            "samples": len(self.scores),
        }

        if self.baseline_mean:
            drop = self.baseline_mean - recent_mean
            result["drop_from_baseline"] = round(drop, 4)
            result["degradation_alert"] = drop > self.degradation_threshold

        return result

# Usage: add scores from sampled production evaluations
quality_monitor = QualityWindow(window_size=200, alert_threshold=0.75)
# Simulate scored production responses
for score in [0.85, 0.90, 0.78, 0.82, 0.91, 0.73, 0.88, 0.79, 0.84, 0.86]:
    quality_monitor.add_score(score)
print(quality_monitor.check_degradation())

Types of Drift and Their Detection Strategies

5. Retraining and Intervention Triggers

from dataclasses import dataclass
from enum import Enum

class InterventionAction(Enum):
    NONE = "no_action"
    INVESTIGATE = "investigate"
    ROLLBACK_PROMPT = "rollback_prompt"
    PIN_MODEL_VERSION = "pin_model_version"
    REINDEX_EMBEDDINGS = "reindex_embeddings"
    RETRAIN_ADAPTER = "retrain_adapter"

@dataclass
class DriftReport:
    """Aggregated drift report with intervention recommendations."""
    prompt_drifted: bool = False
    provider_changed: bool = False
    embedding_drifted: bool = False
    quality_degraded: bool = False
    quality_drop: float = 0.0

    def recommend_action(self) -> InterventionAction:
        """Determine the appropriate intervention based on drift signals."""
        if self.embedding_drifted:
            return InterventionAction.REINDEX_EMBEDDINGS
        if self.prompt_drifted and self.quality_degraded:
            return InterventionAction.ROLLBACK_PROMPT
        if self.provider_changed and self.quality_degraded:
            return InterventionAction.PIN_MODEL_VERSION
        if self.quality_degraded and self.quality_drop > 0.10:
            return InterventionAction.RETRAIN_ADAPTER
        if self.quality_degraded:
            return InterventionAction.INVESTIGATE
        return InterventionAction.NONE

# Example: generate and act on a drift report
report = DriftReport(
    prompt_drifted=False,
    provider_changed=True,
    quality_degraded=True,
    quality_drop=0.08,
)
action = report.recommend_action()
print(f"Recommended action: {action.value}")