Testing

Production Monitoring for RAG: Detecting Retrieval Drift and Quality Degradation

HelpMeTest

HelpMeTest

— 9 min read
Production Monitoring for RAG: Detecting Retrieval Drift and Quality Degradation

Your RAG system passes all tests in staging. You deploy to production. Three weeks later, users start complaining that answers are wrong. You check the code — nothing changed. You look at the LLM — same model, same prompts. But something degraded.

What actually happened: the knowledge base grew, old documents became stale, the query distribution shifted as new users arrived, and an embedding model update silently changed similarity score distributions. None of these triggered an alert. None showed up in your staging tests. They accumulated slowly, and by the time you noticed, you had three weeks of degraded answers in production.

This is the RAG monitoring problem. It is fundamentally different from monitoring traditional software because the failure mode is gradual quality degradation, not crashes or errors. This guide covers the tooling, metrics, and patterns needed to catch it early.

What Degrades in Production RAG Systems

Before building a monitoring system, understand what you are actually watching for:

Retrieval drift: The relationship between queries and stored embeddings shifts. Causes include new documents added to the knowledge base that introduce vocabulary the embedding model maps differently, corpus expansion that dilutes the signal, and query pattern shifts as new user segments arrive.

Knowledge base staleness: Stored facts become outdated. A product pricing page updated last month still has the old price in your vector store. Users get confidently wrong answers.

Embedding model drift: If you use a hosted embedding API, the provider may update the underlying model. OpenAI has done this with text-embedding-ada-002. The new model may have different similarity distributions, making your calibrated thresholds wrong.

LLM generation drift: The generative model's behavior changes — either through fine-tuning, RLHF updates, or temperature changes — affecting faithfulness and answer quality even with identical retrieval.

Query distribution shift: Your initial users asked one type of question. As the product scales, new use cases emerge that your knowledge base does not cover well.

Core Monitoring Metrics

Every production RAG system should track these five metrics continuously:

1. Faithfulness Score

The fraction of claims in generated answers that are supported by retrieved context. Use an LLM judge on a sample of production traffic:

import asyncio
from openai import AsyncOpenAI
import json

client = AsyncOpenAI()

async def score_faithfulness_async(context: str, answer: str) -> float:
    prompt = f"""Does every factual claim in the answer appear in the context?
    
Context: {context}

Answer: {answer}

Return JSON: {{"faithfulness": float (0-1), "unsupported_claims": [...]}}"""
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)["faithfulness"]

async def score_production_sample(interactions: list[dict], sample_size: int = 50) -> dict:
    """Score a random sample of production interactions."""
    import random
    sample = random.sample(interactions, min(sample_size, len(interactions)))
    
    tasks = [
        score_faithfulness_async(i["context"], i["answer"])
        for i in sample
    ]
    scores = await asyncio.gather(*tasks)
    
    return {
        "mean_faithfulness": sum(scores) / len(scores),
        "p10_faithfulness": sorted(scores)[int(len(scores) * 0.1)],
        "low_quality_count": sum(1 for s in scores if s < 0.6),
        "sample_size": len(scores)
    }

2. Retrieval Relevancy

The average similarity score between the query and the top retrieved chunk. A sustained drop signals retrieval drift:

def compute_retrieval_relevancy_metrics(recent_queries: list[dict]) -> dict:
    """
    recent_queries: [{"query": str, "top_chunk_similarity": float, "retrieved_k": int}]
    """
    similarities = [q["top_chunk_similarity"] for q in recent_queries]
    
    return {
        "mean_top_similarity": sum(similarities) / len(similarities),
        "p25_similarity": sorted(similarities)[int(len(similarities) * 0.25)],
        "low_confidence_rate": sum(1 for s in similarities if s < 0.60) / len(similarities),
        "no_result_rate": sum(1 for q in recent_queries if q["retrieved_k"] == 0) / len(recent_queries),
    }

3. Answer Latency

Track p50, p95, p99 latency for the full RAG pipeline. Sudden latency increases can signal retrieval index problems (fragmentation, need for reindexing) before quality degrades:

import time
from functools import wraps
from dataclasses import dataclass, field
from collections import deque
import statistics

@dataclass
class LatencyTracker:
    window_size: int = 1000
    measurements: deque = field(default_factory=lambda: deque(maxlen=1000))
    
    def record(self, latency_ms: float):
        self.measurements.append(latency_ms)
    
    def percentile(self, p: float) -> float:
        if not self.measurements:
            return 0.0
        sorted_vals = sorted(self.measurements)
        idx = int(len(sorted_vals) * p / 100)
        return sorted_vals[min(idx, len(sorted_vals) - 1)]
    
    def report(self) -> dict:
        if not self.measurements:
            return {}
        vals = list(self.measurements)
        return {
            "p50_ms": self.percentile(50),
            "p95_ms": self.percentile(95),
            "p99_ms": self.percentile(99),
            "mean_ms": statistics.mean(vals),
            "sample_count": len(vals)
        }

latency_tracker = LatencyTracker()

def track_rag_latency(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.monotonic()
        result = await func(*args, **kwargs)
        elapsed_ms = (time.monotonic() - start) * 1000
        latency_tracker.record(elapsed_ms)
        return result
    return wrapper

4. Coverage Rate

The fraction of queries where the system returned a confident answer vs. said "I don't know." Track both directions — a rising "I don't know" rate can mean knowledge base gaps, while a falling rate can mean the system is hallucinating where it previously admitted uncertainty:

def compute_coverage_metrics(interactions: list[dict]) -> dict:
    total = len(interactions)
    confident = sum(1 for i in interactions if i.get("confidence_score", 0) >= 0.6)
    uncertain = sum(1 for i in interactions if i.get("is_uncertain", False))
    
    return {
        "coverage_rate": confident / total,
        "uncertainty_rate": uncertain / total,
        "total_queries": total
    }

5. Feedback Signal (If Available)

If users can thumbs-up/down answers, this is your highest-signal metric. Even a 2% sample of explicit feedback is invaluable for calibrating your automated scorers:

def compute_user_satisfaction(feedback_events: list[dict]) -> dict:
    if not feedback_events:
        return {"no_feedback_data": True}
    
    positive = sum(1 for f in feedback_events if f["rating"] == "positive")
    negative = sum(1 for f in feedback_events if f["rating"] == "negative")
    total = len(feedback_events)
    
    return {
        "satisfaction_rate": positive / total,
        "dissatisfaction_rate": negative / total,
        "feedback_count": total
    }

Building the Monitoring Pipeline

Architecture

Production RAG App
    │
    ├── Every query: log {query, context, answer, similarity_scores, latency}
    │
    ▼
Event Log / Message Queue (e.g., Kafka, SQS, or simple DB table)
    │
    ├── Real-time: latency alerts, no-result rate
    │
    └── Async (every 15 min): LLM judge sampling for faithfulness/relevancy
            │
            ▼
        Metrics Store (Prometheus, InfluxDB, or your APM)
            │
            ▼
        Dashboard + Alerts

Logging Every Interaction

Instrument your RAG pipeline to emit structured logs:

import uuid
import json
import logging
from datetime import datetime, timezone

logger = logging.getLogger("rag.interactions")

class InstrumentedRAGPipeline:
    def __init__(self, retriever, generator, interaction_store):
        self.retriever = retriever
        self.generator = generator
        self.store = interaction_store
    
    async def query(self, user_query: str, user_id: str = None) -> dict:
        interaction_id = str(uuid.uuid4())
        start_time = time.monotonic()
        
        # Retrieval
        retrieval_start = time.monotonic()
        retrieved_chunks = await self.retriever.search(user_query, k=5)
        retrieval_latency_ms = (time.monotonic() - retrieval_start) * 1000
        
        top_similarity = max((c.similarity for c in retrieved_chunks), default=0.0)
        
        # Generation
        generation_start = time.monotonic()
        context = "\n\n".join(c.text for c in retrieved_chunks)
        answer = await self.generator.generate(user_query, context)
        generation_latency_ms = (time.monotonic() - generation_start) * 1000
        
        total_latency_ms = (time.monotonic() - start_time) * 1000
        
        # Log interaction
        interaction = {
            "id": interaction_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "query": user_query,
            "answer": answer,
            "context": context,
            "retrieved_chunk_ids": [c.id for c in retrieved_chunks],
            "top_similarity": top_similarity,
            "retrieved_count": len(retrieved_chunks),
            "retrieval_latency_ms": retrieval_latency_ms,
            "generation_latency_ms": generation_latency_ms,
            "total_latency_ms": total_latency_ms,
            "user_id": user_id,
            # Faithfulness scored asynchronously, not blocking the response
            "faithfulness_score": None,  # Filled in by background scorer
        }
        
        await self.store.save(interaction)
        
        return {"answer": answer, "interaction_id": interaction_id}

Async Background Scorer

Score faithfulness asynchronously so it does not add latency to user responses:

import asyncio
from datetime import datetime, timezone, timedelta

async def background_quality_scorer(interaction_store, sample_rate: float = 0.10):
    """
    Runs continuously, scoring a random sample of recent unscored interactions.
    """
    while True:
        try:
            # Get recent unscored interactions
            cutoff = datetime.now(timezone.utc) - timedelta(hours=1)
            unscored = await interaction_store.get_unscored(
                since=cutoff,
                limit=100
            )
            
            if not unscored:
                await asyncio.sleep(60)
                continue
            
            # Sample at configured rate
            import random
            sample = random.sample(unscored, max(1, int(len(unscored) * sample_rate)))
            
            for interaction in sample:
                score = await score_faithfulness_async(
                    interaction["context"],
                    interaction["answer"]
                )
                await interaction_store.update_score(
                    interaction["id"],
                    faithfulness_score=score
                )
            
            await asyncio.sleep(15 * 60)  # Run every 15 minutes
            
        except Exception as e:
            logger.error(f"Background scorer error: {e}")
            await asyncio.sleep(60)

Detecting Retrieval Drift

Retrieval drift is insidious because it shows up as a gradual decline in similarity scores and answer quality, not as an error. Use statistical process control to detect it:

EWMA-Based Drift Detection

Exponentially Weighted Moving Average (EWMA) control charts detect small sustained shifts better than threshold-based alerts:

import numpy as np

class EWMADriftDetector:
    """
    Detects drift in a streaming metric using EWMA control charts.
    Alerts when the EWMA exceeds baseline_mean ± k * baseline_std.
    """
    
    def __init__(self, baseline_mean: float, baseline_std: float, 
                 alpha: float = 0.1, k: float = 3.0):
        self.baseline_mean = baseline_mean
        self.baseline_std = baseline_std
        self.alpha = alpha  # Smoothing factor (lower = slower response, less noise)
        self.k = k          # Control limit multiplier (3.0 = ~0.3% false positive rate)
        self.ewma = baseline_mean
        
        self.upper_limit = baseline_mean + k * baseline_std
        self.lower_limit = baseline_mean - k * baseline_std
    
    def update(self, new_value: float) -> dict:
        self.ewma = self.alpha * new_value + (1 - self.alpha) * self.ewma
        
        drift_detected = (
            self.ewma > self.upper_limit or 
            self.ewma < self.lower_limit
        )
        
        return {
            "ewma": self.ewma,
            "drift_detected": drift_detected,
            "direction": "up" if self.ewma > self.upper_limit else "down" if self.ewma < self.lower_limit else "stable",
            "deviation_sigmas": (self.ewma - self.baseline_mean) / self.baseline_std
        }

# Initialize with your baseline metrics
faithfulness_detector = EWMADriftDetector(
    baseline_mean=0.87,  # Your production baseline
    baseline_std=0.05,
    alpha=0.1,
    k=2.5  # More sensitive than 3.0
)

similarity_detector = EWMADriftDetector(
    baseline_mean=0.73,
    baseline_std=0.08,
    alpha=0.1,
    k=2.5
)

# Update detectors as new scores come in
def process_new_interactions(interaction_batch: list[dict]):
    for interaction in interaction_batch:
        if interaction.get("faithfulness_score") is not None:
            result = faithfulness_detector.update(interaction["faithfulness_score"])
            if result["drift_detected"]:
                alert_on_drift("faithfulness", result)
        
        similarity_result = similarity_detector.update(interaction["top_similarity"])
        if similarity_result["drift_detected"]:
            alert_on_drift("retrieval_similarity", similarity_result)

Knowledge Base Staleness Detection

Detect when stored documents are becoming outdated by tracking a "freshness score":

from datetime import datetime, timezone, timedelta

async def compute_knowledge_freshness(vector_store, source_system_client) -> dict:
    """
    Compare timestamps of stored chunks against source system.
    Flag chunks where the source has been updated since last indexing.
    """
    stale_chunks = []
    total_checked = 0
    
    # Sample stored chunks
    stored_chunks = await vector_store.sample_with_metadata(n=500)
    
    for chunk in stored_chunks:
        total_checked += 1
        source_id = chunk.metadata.get("source_doc_id")
        indexed_at = chunk.metadata.get("indexed_at")
        
        if not source_id or not indexed_at:
            continue
        
        # Check if source document has been updated since indexing
        source_updated_at = await source_system_client.get_last_modified(source_id)
        
        if source_updated_at and source_updated_at > datetime.fromisoformat(indexed_at):
            stale_chunks.append({
                "chunk_id": chunk.id,
                "source_doc_id": source_id,
                "indexed_at": indexed_at,
                "source_updated_at": source_updated_at.isoformat(),
                "stale_days": (datetime.now(timezone.utc) - source_updated_at).days
            })
    
    staleness_rate = len(stale_chunks) / total_checked if total_checked > 0 else 0
    
    return {
        "staleness_rate": staleness_rate,
        "stale_chunk_count": len(stale_chunks),
        "total_checked": total_checked,
        "stale_chunks": sorted(stale_chunks, key=lambda x: x["stale_days"], reverse=True)[:10]
    }

Alerting on Quality Degradation

Define clear alert thresholds for each metric tier:

ALERT_THRESHOLDS = {
    # Page immediately
    "critical": {
        "faithfulness_score": {"lt": 0.65},
        "top_similarity_p25": {"lt": 0.45},
        "no_result_rate": {"gt": 0.15},
        "p99_latency_ms": {"gt": 8000},
    },
    # Alert during business hours
    "warning": {
        "faithfulness_score": {"lt": 0.75},
        "top_similarity_p25": {"lt": 0.55},
        "no_result_rate": {"gt": 0.08},
        "staleness_rate": {"gt": 0.20},
    },
    # Track, review weekly
    "watch": {
        "faithfulness_score": {"lt": 0.82},
        "user_satisfaction_rate": {"lt": 0.80},
    }
}

def evaluate_alerts(metrics: dict) -> list[dict]:
    triggered = []
    
    for severity, thresholds in ALERT_THRESHOLDS.items():
        for metric_name, condition in thresholds.items():
            if metric_name not in metrics:
                continue
            
            value = metrics[metric_name]
            
            if "lt" in condition and value < condition["lt"]:
                triggered.append({
                    "severity": severity,
                    "metric": metric_name,
                    "current_value": value,
                    "threshold": condition["lt"],
                    "message": f"{metric_name} = {value:.3f} (below threshold {condition['lt']})"
                })
            elif "gt" in condition and value > condition["gt"]:
                triggered.append({
                    "severity": severity,
                    "metric": metric_name,
                    "current_value": value,
                    "threshold": condition["gt"],
                    "message": f"{metric_name} = {value:.3f} (above threshold {condition['gt']})"
                })
    
    return sorted(triggered, key=lambda a: ["critical", "warning", "watch"].index(a["severity"]))

A/B Testing Retrieval Strategies

Production monitoring becomes most powerful when combined with A/B testing. Before committing to a new chunking strategy, embedding model, or retrieval parameter, test it against a fraction of live traffic:

import hashlib
from enum import Enum

class RetrievalVariant(Enum):
    CONTROL = "control"        # Current production config
    TREATMENT = "treatment"    # New config being tested

def assign_variant(user_id: str, experiment_id: str, treatment_fraction: float = 0.10) -> RetrievalVariant:
    """Consistent assignment — same user always gets same variant."""
    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
    bucket = (hash_value % 100) / 100.0
    
    return RetrievalVariant.TREATMENT if bucket < treatment_fraction else RetrievalVariant.CONTROL

class ABTestingRAGPipeline:
    def __init__(self, control_retriever, treatment_retriever, experiment_id: str):
        self.retrievers = {
            RetrievalVariant.CONTROL: control_retriever,
            RetrievalVariant.TREATMENT: treatment_retriever,
        }
        self.experiment_id = experiment_id
    
    async def query(self, user_query: str, user_id: str) -> dict:
        variant = assign_variant(user_id, self.experiment_id)
        retriever = self.retrievers[variant]
        
        result = await retriever.search(user_query, k=5)
        
        # Log variant for analysis
        return {
            "results": result,
            "variant": variant.value,
            "experiment_id": self.experiment_id
        }

To analyze results, compare all five monitoring metrics between variants:

def analyze_ab_test(experiment_id: str, interaction_store) -> dict:
    control = interaction_store.get_by_variant(experiment_id, "control")
    treatment = interaction_store.get_by_variant(experiment_id, "treatment")
    
    def compute_metrics(interactions):
        scored = [i for i in interactions if i.get("faithfulness_score")]
        return {
            "mean_faithfulness": sum(i["faithfulness_score"] for i in scored) / len(scored) if scored else None,
            "mean_top_similarity": sum(i["top_similarity"] for i in interactions) / len(interactions),
            "no_result_rate": sum(1 for i in interactions if i["retrieved_count"] == 0) / len(interactions),
            "p95_latency_ms": sorted(i["total_latency_ms"] for i in interactions)[int(len(interactions) * 0.95)],
            "sample_size": len(interactions)
        }
    
    control_metrics = compute_metrics(control)
    treatment_metrics = compute_metrics(treatment)
    
    return {
        "experiment_id": experiment_id,
        "control": control_metrics,
        "treatment": treatment_metrics,
        "faithfulness_delta": (
            treatment_metrics["mean_faithfulness"] - control_metrics["mean_faithfulness"]
            if control_metrics["mean_faithfulness"] and treatment_metrics["mean_faithfulness"]
            else None
        ),
        "similarity_delta": treatment_metrics["mean_top_similarity"] - control_metrics["mean_top_similarity"],
    }

A treatment variant needs to show a statistically significant improvement in faithfulness or retrieval similarity before you promote it to 100% of traffic. Use a minimum sample size of 200 interactions per variant before drawing conclusions.

Prometheus Integration

Export your RAG metrics to Prometheus for standard dashboarding and alerting:

from prometheus_client import Gauge, Histogram, Counter, start_http_server

rag_faithfulness_gauge = Gauge("rag_faithfulness_score", "Mean faithfulness score (EWMA)")
rag_similarity_gauge = Gauge("rag_top_similarity_p25", "P25 top chunk similarity")
rag_no_result_counter = Counter("rag_no_result_queries_total", "Queries with no retrieved results")
rag_latency_histogram = Histogram(
    "rag_query_latency_ms",
    "Full RAG pipeline latency in milliseconds",
    buckets=[100, 300, 500, 1000, 2000, 5000, 10000]
)

def export_metrics_to_prometheus(metrics: dict):
    if "mean_faithfulness" in metrics:
        rag_faithfulness_gauge.set(metrics["mean_faithfulness"])
    if "top_similarity_p25" in metrics:
        rag_similarity_gauge.set(metrics["top_similarity_p25"])

# In your Grafana dashboard, alert on:
# rag_faithfulness_score < 0.75 for 10 minutes
# rag_top_similarity_p25 < 0.50 for 5 minutes

HelpMeTest can complement your production monitoring by running scheduled end-to-end tests against your RAG application — verifying that specific high-stakes user queries still return correct answers, running on a configurable schedule (hourly, daily) and alerting your team the moment a known-good answer breaks.

Summary

Production RAG monitoring requires four components working together:

  1. Structured interaction logging — every query, context, answer, similarity scores, and latency
  2. Async quality scoring — faithfulness and relevancy computed on sampled traffic without blocking responses
  3. Drift detection — EWMA control charts on key metrics, knowledge base staleness tracking
  4. A/B testing infrastructure — consistent user assignment, per-variant metric comparison before promoting changes

Set baselines during your first stable production week. Then alert when any metric deviates beyond 2-3 standard deviations from baseline. For most production RAG systems, a sustained faithfulness drop of more than 5 points or a retrieval similarity drop of more than 8 points represents a meaningful quality regression that warrants investigation.

The goal is to move from reactive debugging ("users told us answers are wrong") to proactive quality management ("the drift detector caught it before any user noticed").

Read more

Start now free