Monitoring RAG Retrieval Quality Over Time

Monitoring RAG Retrieval Quality Over Time

Retrieval-Augmented Generation (RAG) systems are only as good as the documents they retrieve. When you first ship a RAG pipeline, your retrieval quality may look excellent on a held-out test set. But over weeks and months, content drifts, queries shift, embeddings get stale, and chunking strategies that once worked start to fail. Monitoring retrieval quality over time is the practice of continuously measuring how well your retriever surfaces the right context, so you can catch degradation before users notice.

What Is RAG Retrieval Quality Monitoring?

Retrieval quality monitoring is the systematic, ongoing measurement of how relevant and complete the retrieved context is for real user queries. Unlike one-off evaluation, it tracks metrics across time windows — daily, weekly, or per release — so you can detect regressions, seasonal patterns, and slow drift.

The core dimensions you typically track include:

• Context relevance: Are the retrieved chunks actually related to the query?
• Context coverage: Did retrieval find all the information needed to answer?
• Ranking quality: Are the most useful chunks ranked near the top?
• Retrieval latency: How long does retrieval take, and is it trending up?
• Failure rate: How often does retrieval return empty or low-confidence results?

Why It Matters

RAG systems degrade silently. A vector index that performed well at launch can quietly lose accuracy as new documents are added, as the underlying content changes, or as user queries evolve toward topics your embeddings represent poorly. Without monitoring, you only learn about these problems through support tickets and churn.

Key reasons to monitor continuously:

• Content drift: Knowledge bases grow and change; old chunks may no longer match new queries.
• Query drift: Users ask different questions over time, exposing gaps in your index.
• Embedding model upgrades: Swapping embedding models can silently break semantic matching.
• Chunking changes: Re-chunking documents changes what gets retrieved even for identical queries.
• Release safety: Monitoring gives you a guardrail before deploying retriever changes to production.

How to Use It: A Practical Implementation

The most effective approach combines offline evaluation on a golden dataset with online monitoring of real production queries. Below is a complete, runnable example using Python that logs retrieval events, computes relevance scores with an LLM-as-a-judge, and tracks metrics over time.

1. Define a Golden Evaluation Set

Start with a curated set of queries paired with the documents that should be retrieved. This is your regression baseline.

golden_set = [
    {
        "query": "How do I reset my password?",
        "expected_doc_ids": ["auth-12", "auth-15"],
        "expected_keywords": ["reset", "password", "email"]
    },
    {
        "query": "What is the refund policy for annual plans?",
        "expected_doc_ids": ["billing-04"],
        "expected_keywords": ["refund", "annual", "prorate"]
    },
    {
        "query": "How do I invite team members to my workspace?",
        "expected_doc_ids": ["teams-02", "teams-07"],
        "expected_keywords": ["invite", "workspace", "member"]
    }
]

2. Wrap Your Retriever With Logging

Instrument your retriever so every call records the query, retrieved chunks, scores, and latency. This becomes your online monitoring data.

import time
import json
from datetime import datetime, timezone
from pathlib import Path

LOG_FILE = Path("retrieval_events.jsonl")

def retrieve_with_logging(query, retriever, top_k=5):
    start = time.perf_counter()
    results = retriever.search(query, top_k=top_k)
    latency_ms = (time.perf_counter() - start) * 1000

    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "query": query,
        "retrieved": [
            {"doc_id": r.doc_id, "score": float(r.score), "text": r.text[:200]}
            for r in results
        ],
        "latency_ms": latency_ms,
        "top_k": top_k
    }

    with LOG_FILE.open("a") as f:
        f.write(json.dumps(event) + "\n")

    return results

3. Score Relevance With an LLM-as-a-Judge

For each retrieved chunk, ask an LLM to rate relevance on a 0–3 scale. This gives you a continuous signal even for queries not in your golden set.

import openai

JUDGE_PROMPT = """You are evaluating retrieval relevance for a RAG system.

Query: {query}
Retrieved chunk: {chunk}

Rate the relevance of the chunk to the query on a scale of 0-3:
0 = completely irrelevant
1 = marginally relevant
2 = relevant but incomplete
3 = highly relevant

Respond with only a single integer."""

def score_relevance(query, chunk_text, model="gpt-4o-mini"):
    resp = openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": JUDGE_PROMPT.format(
            query=query, chunk=chunk_text
        )}],
        temperature=0
    )
    try:
        return int(resp.choices[0].message.content.strip())
    except ValueError:
        return 0

4. Compute Metrics Over a Time Window

Aggregate logged events into daily or weekly metrics so you can plot trends and set alerts.

from collections import defaultdict
from statistics import mean

def compute_daily_metrics(log_path, judge_fn=None, sample_rate=1.0):
    by_day = defaultdict(list)
    with open(log_path) as f:
        for line in f:
            event = json.loads(line)
            day = event["timestamp"][:10]
            by_day[day].append(event)

    report = {}
    for day, events in sorted(by_day.items()):
        latencies = [e["latency_ms"] for e in events]
        empty_count = sum(1 for e in events if len(e["retrieved"]) == 0)

        relevance_scores = []
        if judge_fn:
            for e in events[::int(1/sample_rate)]:
                for r in e["retrieved"][:3]:
                    relevance_scores.append(judge_fn(e["query"], r["text"]))

        report[day] = {
            "query_count": len(events),
            "avg_latency_ms": round(mean(latencies), 2),
            "p95_latency_ms": round(sorted(latencies)[int(len(latencies)*0.95)-1], 2),
            "empty_result_rate": round(empty_count / len(events), 4),
            "avg_relevance": round(mean(relevance_scores), 3) if relevance_scores else None
        }
    return report

5. Run the Golden Set as a Regression Check

On every deploy or nightly, run your golden queries and compare against expected documents. This catches regressions before they reach users.

def run_golden_regression(retriever, golden_set, top_k=5):
    results = []
    for item in golden_set:
        retrieved_ids = [r.doc_id for r in retriever.search(item["query"], top_k)]
        expected = set(item["expected_doc_ids"])
        hit = expected & set(retrieved_ids)
        recall = len(hit) / len(expected) if expected else 1.0
        mrr = 0.0
        for rank, doc_id in enumerate(retrieved_ids, 1):
            if doc_id in expected:
                mrr = 1.0 / rank
                break
        results.append({
            "query": item["query"],
            "recall@5": round(recall, 3),
            "mrr": round(mrr, 3),
            "retrieved_ids": retrieved_ids
        })
    avg_recall = mean(r["recall@5"] for r in results)
    avg_mrr = mean(r["mrr"] for r in results)
    return {"per_query": results, "avg_recall@5": avg_recall, "avg_mrr": avg_mrr}

6. Alert on Degradation

Define thresholds and trigger alerts when metrics cross them. Simple rules work well to start.

def check_alerts(daily_report, thresholds):
    alerts = []
    latest_day = max(daily_report.keys())
    metrics = daily_report[latest_day]

    if metrics["empty_result_rate"] > thresholds["max_empty_rate"]:
        alerts.append(f"Empty result rate {metrics['empty_result_rate']} exceeds threshold")
    if metrics["p95_latency_ms"] > thresholds["max_p95_latency"]:
        alerts.append(f"P95 latency {metrics['p95_latency_ms']}ms exceeds threshold")
    if metrics["avg_relevance"] is not None and metrics["avg_relevance"] < thresholds["min_relevance"]:
        alerts.append(f"Avg relevance {metrics['avg_relevance']} below threshold")

    return alerts

thresholds = {
    "max_empty_rate": 0.05,
    "max_p95_latency": 800,
    "min_relevance": 2.0
}

Best Practices

• Keep your golden set fresh. Add new queries as you discover failure modes in production. A stale golden set gives false confidence.
• Sample for LLM judging. Scoring every retrieved chunk with an LLM is expensive. Sample 5–10% of traffic and weight toward low-confidence retrievals.
• Track per-topic metrics. Aggregate metrics can hide localized failures. Break down by query category, language, or source document set.
• Version your retriever. Log the embedding model, chunking config, and index version with each event so you can attribute regressions to specific changes.
• Monitor user feedback signals. Thumbs-down ratings, follow-up rephrased queries, and low citation clicks are strong proxies for poor retrieval.
• Compare against a shadow index. When testing a new embedding model or chunking strategy, run it in parallel on production queries and compare retrieval overlap.
• Watch for index bloat. As your corpus grows, recall can drop because relevant chunks get buried. Track corpus size alongside quality metrics.
• Set SLOs, not just alerts. Define service-level objectives like "recall@5 above 0.85 for 99% of daily golden queries" and track them over rolling windows.

Conclusion

Monitoring RAG retrieval quality over time turns a fragile system into a maintainable one. By combining a curated golden set for regression testing, LLM-judged relevance scoring on production traffic, and clear alerting thresholds, you create a feedback loop that catches drift before users do. The implementation above gives you a starting point — instrument your retriever today, build a small golden set, and begin logging. The metrics you collect this week become the baseline that tells you whether next month's changes made things better or worse.