Detecting Embedding Drift: A Statistical Gate Before Sovereign RAG

Embedding drift is the quiet killer of retrieval-augmented generation. You deploy a RAG pipeline, embeddings look fine, queries return plausible chunks, and then one day the answers start feeling wrong. Not catastrophically wrong — just subtly off. The root cause is usually not the LLM. It is the vector space shifting underneath you: documents get re-embedded with a newer model, source data changes, or the corpus grows in a way that distorts local neighborhoods. By the time you notice, your retrieval has been serving stale or mismatched embeddings for weeks.

The standard answer is to re-embed everything on a schedule, but that is expensive and often unnecessary. A better approach is to detect drift before it affects inference. This article describes a statistical gate that runs between your ingestion pipeline and your inference service, flagging when embeddings have drifted enough to warrant re-embedding or retraining. The gate is lightweight, uses only your existing stack (Postgres, Qdrant, and a bit of Python), and can be the foundation of a self-healing system.

What Is Embedding Drift?

Embedding drift is the gradual change in the distribution of vectors in your vector database. It can happen for several reasons:

• Model updates: You switch from one embedding model to another (or a newer version of the same model). Even a minor version bump can change the geometric relationships between texts.
• Data evolution: Your corpus changes. Documents are added, removed, or edited. New topics appear, old ones fade. The embeddings for existing documents do not change, but the overall distribution does.
• Domain shift: The language used in new documents drifts from the language used when you first embedded your corpus. This is common in fast-moving domains like technology or finance.

Drift matters because RAG retrieval relies on semantic similarity. If the embedding space has shifted, the nearest neighbors for a query may no longer be the most relevant documents. The LLM then generates answers from the wrong context, and you get hallucinations or incomplete responses.

Why a Statistical Gate?

A statistical gate is a check that runs before inference (or before a batch of inferences) and decides whether the current embedding space is trustworthy. If the gate passes, you proceed with retrieval as usual. If it fails, you trigger a re-embedding job or retrain a projection layer.

The gate is not about measuring absolute quality — that requires ground truth labels and expensive evaluation. Instead, it measures the statistical health of your vector space using lightweight, unsupervised signals. These signals are cheap to compute and can be run on every query or on a sliding window of recent queries.

A typical gate has three components:

1. Baseline statistics: Computed after a known-good embedding run.
2. Live statistics: Computed on a rolling basis from recent queries and their retrieved neighbors.
3. A decision rule: Compares live statistics to baseline and flags if the deviation exceeds a threshold.

The gate should be fast — it adds latency to your inference path, so it must not become a bottleneck. In practice, you can compute the statistics on a sample of queries (e.g., 1%) and cache the result for a few minutes.

Building the Gate: A Concrete Example

Let's walk through a concrete implementation using the RiNET stack: Postgres for metadata, Qdrant for vectors, and Python for the gate logic. The gate runs as a small service that is queried before each inference request (or in a background loop).

Step 1: Define Baseline Statistics

After a fresh embedding run (e.g., after re-embedding your entire corpus), you compute a set of statistics over the vectors in Qdrant. The most useful statistics are:

• Mean vector (centroid) and its norm.
• Average pairwise cosine similarity between a sample of vectors (or the average norm of the vectors).
• The distribution of distances from the centroid to each vector (e.g., mean and standard deviation).
• The effective dimensionality (participation ratio) of the vector set.

These statistics capture the global shape of your embedding space. If the space drifts, these numbers will change.

In Python, you might compute them like this:

import numpy as np
from qdrant_client import QdrantClient

client = QdrantClient(url="http://localhost:6333")

def compute_baseline(collection_name: str, sample_size: int = 1000):
    # Fetch a sample of vectors
    vectors = []
    offset = None
    while len(vectors) < sample_size:
        result = client.scroll(
            collection_name=collection_name,
            limit=min(100, sample_size - len(vectors)),
            with_vectors=True,
            offset=offset,
        )
        points = result[0]
        offset = result[1]
        if not points:
            break
        vectors.extend([p.vector for p in points])
    
    arr = np.array(vectors)
    centroid = arr.mean(axis=0)
    norms = np.linalg.norm(arr, axis=1)
    centered = arr - centroid
    # Participation ratio: (sum of eigenvalues)^2 / sum of eigenvalues^2
    cov = np.cov(centered.T)
    eigenvalues = np.linalg.eigvalsh(cov)
    participation_ratio = (eigenvalues.sum() ** 2) / (eigenvalues ** 2).sum()
    
    baseline = {
        "centroid": centroid.tolist(),
        "mean_norm": float(norms.mean()),
        "std_norm": float(norms.std()),
        "participation_ratio": float(participation_ratio),
    }
    return baseline

Store this baseline in Postgres or as a JSON file. You will refer to it later.

Step 2: Compute Live Statistics from Query Traffic

Instead of scanning the whole collection, you can compute live statistics from the vectors that are actually retrieved for real queries. This is more efficient and reflects the part of the space that matters.

For each query, you get a set of retrieved vectors (the top-k neighbors). Over a time window (e.g., the last hour), you aggregate these vectors and compute statistics similar to the baseline. The difference is that you are sampling from the query distribution, not the corpus distribution.

from collections import deque
import time

class DriftDetector:
    def __init__(self, baseline, window_size=1000, threshold=0.05):
        self.baseline = baseline
        self.window = deque(maxlen=window_size)
        self.threshold = threshold
    
    def add_observation(self, vectors):
        # vectors: list of vectors retrieved for a query
        self.window.extend(vectors)
    
    def compute_live_stats(self):
        if len(self.window) < 100:
            return None
        arr = np.array(self.window)
        centroid = arr.mean(axis=0)
        norms = np.linalg.norm(arr, axis=1)
        centered = arr - centroid
        cov = np.cov(centered.T)
        eigenvalues = np.linalg.eigvalsh(cov)
        participation_ratio = (eigenvalues.sum() ** 2) / (eigenvalues ** 2).sum()
        return {
            "centroid": centroid.tolist(),
            "mean_norm": float(norms.mean()),
            "std_norm": float(norms.std()),
            "participation_ratio": float(participation_ratio),
        }
    
    def check_drift(self, live_stats):
        if live_stats is None:
            return False
        # Compare centroid shift (cosine distance)
        centroid_shift = 1 - np.dot(self.baseline["centroid"], live_stats["centroid"]) / (
            np.linalg.norm(self.baseline["centroid"]) * np.linalg.norm(live_stats["centroid"])
        )
        # Compare mean norm difference (relative)
        norm_diff = abs(live_stats["mean_norm"] - self.baseline["mean_norm"]) / self.baseline["mean_norm"]
        # Compare participation ratio difference (relative)
        pr_diff = abs(live_stats["participation_ratio"] - self.baseline["participation_ratio"]) / self.baseline["participation_ratio"]
        # Combine or use individual thresholds
        if centroid_shift > self.threshold or norm_diff > self.threshold or pr_diff > self.threshold:
            return True
        return False

Step 3: Integrate with Inference

In a typical RAG service, you have a query endpoint that looks like:

@app.post("/query")
def query(request: QueryRequest):
    # Check drift gate
    if detector.check_drift(detector.compute_live_stats()):
        trigger_reembedding()
        # Optionally, fall back to a stale index or return a warning
    # Proceed with normal retrieval
    query_vector = embed(request.text)
    results = qdrant.search(...)
    context = format_context(results)
    response = llm.generate(context, request.text)
    # Add observations to detector
    detector.add_observation([r.vector for r in results])
    return response

This is a simplified version. In practice, you would run the gate asynchronously to avoid adding latency to every request. A common pattern is to have a background worker that periodically samples recent queries, computes live stats, and updates a flag in Redis. The inference service then checks the flag before processing a batch.

Choosing the Right Statistics

The statistics above are not the only options. The choice depends on your embedding model and data characteristics. Some alternatives:

• Centroid shift: A simple cosine distance between baseline and live centroids. Sensitive to global shifts.
• Norm distribution: If your embedding model produces vectors with varying norms, changes in the mean norm can indicate drift.
• Participation ratio: Measures the effective dimensionality. A drop could mean that the vectors are collapsing into a lower-dimensional subspace, which might indicate that the model is no longer distinguishing between concepts.
• Neighborhood overlap: For a fixed set of queries, compare the overlap of retrieved document IDs between baseline and live. If the overlap drops significantly, retrieval is changing.

Neighborhood overlap is perhaps the most direct measure of retrieval quality, but it requires a fixed query set and ground truth. The statistical measures are cheaper and can be computed without labels.

Setting Thresholds

Thresholds are tricky. If they are too loose, you will miss drift. If they are too tight, you will trigger unnecessary re-embeddings. A common approach is to compute the statistics on a validation set and observe their natural variability over time. Then set the threshold at, say, three standard deviations from the mean of that variability.

Another approach is to use a control chart (like a CUSUM or EWMA) that detects small shifts over time. This is more sensitive than a simple threshold and can catch gradual drift.

Handling Drift When Detected

Once the gate flags drift, you have several options:

1. Re-embed the entire corpus with the current model. This is the most thorough but can be expensive.
2. Re-embed only the affected subset if you can identify which documents are contributing to the drift. This is harder but more efficient.
3. Retrain a projection layer if you are using a model that supports adapters (e.g., LoRA). You could train a small linear layer that maps old embeddings to the new space, avoiding a full re-embed.
4. Fallback to a previous index if you maintain multiple versions.

In a self-healing system, you might automate the re-embedding process, but you should always have a human in the loop for major changes.

Integration with Your Sovereign Stack

In the RiNET stack, we run three Hetzner servers connected via WireGuard. On these, we host Qwen models on vLLM, a BGE-M3 embedding model, and a combination of Postgres, Qdrant, and Neo4j for storage. The drift gate can be deployed as a separate microservice that talks to Qdrant and Postgres.

For example, you could store the baseline statistics in Postgres, and have a cron job that runs every hour to compute live stats from recent queries logged in a separate table. The job writes a decision to a status table, which the inference service reads.

Here is a sketch of the schema:

CREATE TABLE embedding_baseline (
    model_version TEXT PRIMARY KEY,
    stats JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE drift_status (
    model_version TEXT PRIMARY KEY,
    is_drifted BOOLEAN NOT NULL,
    last_checked TIMESTAMPTZ NOT NULL,
    details JSONB
);

Then, the drift detector can be a simple Python script that runs on a schedule and updates the status table.

Beyond Detection: Self-Healing

A drift gate is only the first step. The ultimate goal is a self-healing system that not only detects drift but also responds automatically. In a sovereign setup, you control the entire pipeline, so you can implement automated responses without external dependencies.

For instance, you could set up a workflow that:

1. Detects drift using the gate.
2. Triggers a re-embedding job that runs on one of your servers.
3. Validates the new index by running a set of golden queries and comparing retrieval quality.
4. Switches the inference service to the new index only if validation passes.

This requires careful orchestration, but it is feasible with tools like Airflow or even a simple Makefile with cron.

Practical Considerations

• Sampling: Do not compute statistics on every vector in the collection. Use a random sample or a sample from recent queries. The sample size should be large enough to be statistically significant but small enough to be fast.
• Latency: The gate should not add noticeable latency to your inference path. Compute live stats asynchronously and cache the result.
• False positives: Be prepared for false alarms. The gate might trigger due to a temporary spike in unusual queries. Use a hysteresis mechanism: require the drift to persist for a certain number of checks before acting.
• Model versioning: Always record which embedding model version was used to create the vectors. This is critical for comparing statistics.

Conclusion

Embedding drift is a real problem in RAG systems, but it is detectable with a few statistical measures. By building a gate that runs before inference, you can catch drift early and avoid serving bad answers. The gate is cheap to implement and uses only the tools you already have. With a bit of automation, you can turn it into a self-healing mechanism that keeps your sovereign RAG system reliable over time.

Start by computing a baseline after your next embedding run. Then instrument your query path to log retrieved vectors. Within a day, you will have enough data to see if your gate works. The sooner you start, the sooner you will catch the next drift.