Embedding Drift Alerts: Catch Provider Vector Changes Before RAG Dies

# drift/canary.py
from dataclasses import dataclass
import numpy as np

@dataclass
class Canary:
    query: str
    expected_ids: list[str]  # gold chunk ids
    emb_model: str

def recall_at_k(neighbors: list[str], expected: list[str], k: int = 5) -> float:
    top = set(neighbors[:k])
    return len(top & set(expected)) / max(1, len(expected))

def centroid_shift(old: np.ndarray, new: np.ndarray) -> float:
    # mean L2 between paired canary embeddings
    return float(np.linalg.norm(old - new, axis=1).mean())

def assert_healthy(recalls: list[float], shift: float):
    if np.median(recalls) < 0.8:
        raise SystemExit(f"recall_collapse:{np.median(recalls):.3f}")
    if shift > 0.35:  # tune per model family
        raise SystemExit(f"embedding_space_shift:{shift:.3f}")

// Always pin model IDs in config — never "latest"
export const EMBEDDING = {
  modelId: "amazon.titan-embed-text-v2:0",
  dim: 1024,
  indexAlias: "code-rag-v2",
};

// On deploy / cron:
async function probeProvider(modelId: string) {
  const meta = await bedrock.getFoundationModel({ modelIdentifier: modelId });
  // Persist content hash / ARN; alarm on unexpected change
  await putMetric("EmbeddingModelFingerprint", hash(meta));
}

def upsert(chunk_id: str, vector, model_id: str, index_model_id: str):
    if model_id != index_model_id:
        raise ValueError(f"model_mismatch:{model_id}!={index_model_id}")
    client.index(id=chunk_id, body={"vector": vector, "model_id": model_id})

# illustrative
aws opensearch update-alias --name code-rag --add code-rag-v3 --remove code-rag-v2