rag-evaluation

RAG Evaluation

Measure, monitor, and improve RAG system performance with comprehensive metrics.

When to Use

• Setting up RAG evaluation pipelines
• Comparing retrieval strategies
• Measuring generation quality
• Building regression tests for RAG
• Debugging poor RAG performance

Evaluation Framework

┌─────────────────────────────────────────────────────────┐
│                  RAG Evaluation                          │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │  Retrieval  │  │ Generation  │  │  End-to-End │     │
│  │  Metrics    │  │  Metrics    │  │   Metrics   │     │
│  ├─────────────┤  ├─────────────┤  ├─────────────┤     │
│  │ • MRR       │  │ • Faithful- │  │ • Answer    │     │
│  │ • Recall@k  │  │   ness      │  │   Correct-  │     │
│  │ • Precision │  │ • Relevance │  │   ness      │     │
│  │ • NDCG      │  │ • Coherence │  │ • Latency   │     │
│  │ • Hit Rate  │  │ • Toxicity  │  │ • Cost      │     │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
│                                                          │
└─────────────────────────────────────────────────────────┘

Retrieval Metrics

Implementation

import numpy as np
from typing import List, Dict

def mean_reciprocal_rank(results: List[List[str]], relevant: List[List[str]]) -> float:
    """
    Calculate MRR across queries.
    results: List of ranked document IDs per query
    relevant: List of relevant document IDs per query
    """
    mrr_sum = 0.0
    for res, rel in zip(results, relevant):
        rel_set = set(rel)
        for rank, doc_id in enumerate(res, 1):
            if doc_id in rel_set:
                mrr_sum += 1.0 / rank
                break
    return mrr_sum / len(results)

def recall_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
    """Calculate Recall@k."""
    recall_sum = 0.0
    for res, rel in zip(results, relevant):
        retrieved_k = set(res[:k])
        relevant_set = set(rel)
        if relevant_set:
            recall_sum += len(retrieved_k & relevant_set) / len(relevant_set)
    return recall_sum / len(results)

def precision_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
    """Calculate Precision@k."""
    precision_sum = 0.0
    for res, rel in zip(results, relevant):
        retrieved_k = set(res[:k])
        relevant_set = set(rel)
        precision_sum += len(retrieved_k & relevant_set) / k
    return precision_sum / len(results)

def ndcg_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
    """Calculate NDCG@k."""
    def dcg(scores):
        return sum(s / np.log2(i + 2) for i, s in enumerate(scores))

    ndcg_sum = 0.0
    for res, rel in zip(results, relevant):
        rel_set = set(rel)
        gains = [1 if doc in rel_set else 0 for doc in res[:k]]
        ideal_gains = sorted(gains, reverse=True)

        dcg_val = dcg(gains)
        idcg_val = dcg(ideal_gains)

        ndcg_sum += dcg_val / idcg_val if idcg_val > 0 else 0

    return ndcg_sum / len(results)

def hit_rate(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
    """Calculate Hit Rate (any relevant doc in top-k)."""
    hits = 0
    for res, rel in zip(results, relevant):
        if set(res[:k]) & set(rel):
            hits += 1
    return hits / len(results)