LLM-Specific Monitoring & Drift Detection

# Define PromptDriftMonitor; implement __init__, register_template, check_template
# Key operations: prompt construction
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
 ),
 }

# Define EmbeddingDriftDetector; implement __init__, _compute_embeddings, _pairwise_similarities
# Key operations: embedding lookup
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"
 ),
 }

# Define QualityWindow; implement add_score, check_degradation
# Key operations: forward pass computation, RAG pipeline, evaluation logic
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())

# Define InterventionAction, DriftReport; implement recommend_action
# Key operations: training loop, embedding lookup, results display
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}")