Module 8: Final Capstone Project - Complete RAG System
Evaluation Framework: Measuring RAG Performance
Description
In this capsule you'll implement a complete Evaluation Framework to objectively measure the quality of your RAG system. Without rigorous metrics, you can't know whether your decisions (chunking strategy, embedding model, retrieval parameters) improve or hurt the system.
You'll implement the key information retrieval metrics: Recall@K, Precision@K, MRR, and nDCG@K. You'll also build an A/B testing framework to compare strategies and compute statistical significance.
By the end you'll have a production-ready evaluator that lets you iterate based on data, not intuition.
Estimated duration: 40-50 minutes
Objectives
By completing this capsule, you'll be able to:
- ✅ Implement retrieval metrics (Recall@K, Precision@K, MRR, nDCG@K)
- ✅ Create evaluation datasets (queries + relevance judgments)
- ✅ Evaluate a RAG system with objective metrics
- ✅ Compare strategies with A/B testing
- ✅ Compute statistical significance (t-test)
- ✅ Generate evaluation reports
Key Retrieval metrics
1. Recall@K
Definition: Of all the relevant documents, what % is in the top-K?
Recall@K = (Relevant docs in top-K) / (Total relevant docs)Example:
# Query: "Python installation"
relevant_docs = ['doc_3', 'doc_7', 'doc_12'] # 3 total relevant
retrieved_top5 = ['doc_1', 'doc_3', 'doc_5', 'doc_7', 'doc_9']
# In the top-5 there are 2 relevant: doc_3, doc_7
Recall@5 = 2 / 3 = 0.67 (67%)2. Precision@K
Definition: Of the top-K results, what % is relevant?
Precision@K = (Relevant docs in top-K) / KExample:
# Top-5: ['doc_1', 'doc_3', 'doc_5', 'doc_7', 'doc_9']
# Relevant: doc_3, doc_7 (2 of 5)
Precision@5 = 2 / 5 = 0.40 (40%)3. MRR (Mean Reciprocal Rank)
Definition: Position of the FIRST relevant document
RR = 1 / position_of_first_relevant
MRR = average of RR over multiple queriesExample:
# Top-5: ['doc_1', 'doc_3', ...]
# First relevant: doc_3 at position 2
RR = 1 / 2 = 0.504. nDCG@K (Normalized Discounted Cumulative Gain)
Definition: A metric that rewards relevant results in high positions
DCG@K = Σ (rel_i / log2(i + 1)) # For i in [1..K]
nDCG@K = DCG@K / IDCG@K # Normalized by the idealExample:
# Top-3: ['doc_3', 'doc_1', 'doc_7']
# Relevance: [1, 0, 1] # doc_3 and doc_7 are relevant
DCG@3 = 1/log2(2) + 0/log2(3) + 1/log2(4)
= 1.0 + 0.0 + 0.5
= 1.5
IDCG@3 = 1/log2(2) + 1/log2(3) + 0/log2(4) # Ideal order
= 1.0 + 0.63 + 0.0
= 1.63
nDCG@3 = 1.5 / 1.63 = 0.92 (92%)Step 1: Implement RetrievalEvaluator
1.1: Class with all the metrics
Create src/evaluation/retrieval_evaluator.py:
"""
Retrieval Evaluator - RAG System
Evaluation metrics for retrieval systems
"""
import numpy as np
from typing import List, Dict, Set
from collections import defaultdict
import logging
class RetrievalEvaluator:
"""
Evaluator for retrieval systems
Metrics:
- Recall@K: % of relevant docs retrieved
- Precision@K: % of retrieved docs that are relevant
- MRR: Position of the first relevant document
- nDCG@K: Relevance weighted by position
Example:
evaluator = RetrievalEvaluator()
metrics = evaluator.evaluate_query(
retrieved=['doc_1', 'doc_3', 'doc_5'],
relevant={'doc_3', 'doc_7'},
k=5
)
"""
def __init__(self):
self.logger = logging.getLogger(__name__)
def recall_at_k(
self,
retrieved: List[str],
relevant: Set[str],
k: int
) -> float:
"""
Compute Recall@K
Args:
retrieved: List of retrieved IDs (sorted by score)
relevant: Set of relevant IDs (ground truth)
k: Number of results to consider
Returns:
Recall score [0, 1]
"""
if not relevant:
return 0.0
# Top-K retrieved
retrieved_k = set(retrieved[:k])
# How many relevant are in the top-K
relevant_retrieved = len(retrieved_k & relevant)
# Recall = relevant_retrieved / total_relevant
return relevant_retrieved / len(relevant)
def precision_at_k(
self,
retrieved: List[str],
relevant: Set[str],
k: int
) -> float:
"""
Compute Precision@K
Args:
retrieved: List of retrieved IDs
relevant: Set of relevant IDs
k: Number of results to consider
Returns:
Precision score [0, 1]
"""
if k == 0:
return 0.0
retrieved_k = set(retrieved[:k])
relevant_retrieved = len(retrieved_k & relevant)
# Precision = relevant_retrieved / K
return relevant_retrieved / k
def f1_at_k(
self,
retrieved: List[str],
relevant: Set[str],
k: int
) -> float:
"""
Compute F1@K (harmonic mean of Precision and Recall)
Args:
retrieved: List of retrieved IDs
relevant: Set of relevant IDs
k: Number of results
Returns:
F1 score [0, 1]
"""
precision = self.precision_at_k(retrieved, relevant, k)
recall = self.recall_at_k(retrieved, relevant, k)
if precision + recall == 0:
return 0.0
return 2 * (precision * recall) / (precision + recall)
def mrr(
self,
retrieved: List[str],
relevant: Set[str]
) -> float:
"""
Compute MRR (Mean Reciprocal Rank)
Args:
retrieved: List of retrieved IDs
relevant: Set of relevant IDs
Returns:
RR score [0, 1]
"""
for i, doc_id in enumerate(retrieved, 1):
if doc_id in relevant:
return 1.0 / i
return 0.0 # No relevant found
def ndcg_at_k(
self,
retrieved: List[str],
relevant: Set[str],
k: int
) -> float:
"""
Compute nDCG@K (Normalized Discounted Cumulative Gain)
Args:
retrieved: List of retrieved IDs
relevant: Set of relevant IDs
k: Number of results
Returns:
nDCG score [0, 1]
"""
# DCG: Sum of rel_i / log2(i + 1)
dcg = 0.0
for i, doc_id in enumerate(retrieved[:k], 1):
if doc_id in relevant:
dcg += 1.0 / np.log2(i + 1)
# IDCG: ideal DCG (all relevant first)
num_relevant = min(len(relevant), k)
idcg = sum(1.0 / np.log2(i + 1) for i in range(1, num_relevant + 1))
if idcg == 0:
return 0.0
return dcg / idcg
def evaluate_query(
self,
retrieved: List[str],
relevant: Set[str],
k: int = 10
) -> Dict[str, float]:
"""
Evaluate a query with all the metrics
Args:
retrieved: List of retrieved IDs
relevant: Set of relevant IDs
k: Number of results
Returns:
Dict with all the metrics
"""
return {
'recall@k': self.recall_at_k(retrieved, relevant, k),
'precision@k': self.precision_at_k(retrieved, relevant, k),
'f1@k': self.f1_at_k(retrieved, relevant, k),
'mrr': self.mrr(retrieved, relevant),
'ndcg@k': self.ndcg_at_k(retrieved, relevant, k)
}
def evaluate_dataset(
self,
results: List[Dict],
k: int = 10
) -> Dict[str, float]:
"""
Evaluate a complete dataset (multiple queries)
Args:
results: List of dicts with 'retrieved' and 'relevant' per query
k: Number of results
Returns:
Dict with averaged metrics
"""
all_metrics = defaultdict(list)
for result in results:
retrieved = result['retrieved']
relevant = result['relevant']
metrics = self.evaluate_query(retrieved, relevant, k)
for metric, value in metrics.items():
all_metrics[metric].append(value)
# Average all the metrics
avg_metrics = {
metric: np.mean(values)
for metric, values in all_metrics.items()
}
# Add metadata
avg_metrics['num_queries'] = len(results)
return avg_metrics
def print_evaluation(self, metrics: Dict[str, float]):
"""
Print formatted metrics
Args:
metrics: Dict with metrics
"""
print("\n" + "=" * 60)
print("📊 EVALUATION METRICS")
print("=" * 60)
if 'num_queries' in metrics:
print(f"\nDataset: {metrics['num_queries']} queries")
print("\nRetrieval Performance:")
print(f" Recall@K: {metrics.get('recall@k', 0):.3f}")
print(f" Precision@K: {metrics.get('precision@k', 0):.3f}")
print(f" F1@K: {metrics.get('f1@k', 0):.3f}")
print(f" MRR: {metrics.get('mrr', 0):.3f}")
print(f" nDCG@K: {metrics.get('ndcg@k', 0):.3f}")
print("\n" + "=" * 60)
# Demo
if __name__ == "__main__":
evaluator = RetrievalEvaluator()
# Example query
retrieved = ['doc_1', 'doc_3', 'doc_5', 'doc_7', 'doc_9']
relevant = {'doc_3', 'doc_7', 'doc_12'}
metrics = evaluator.evaluate_query(retrieved, relevant, k=5)
evaluator.print_evaluation(metrics)Step 2: Create an Evaluation Dataset
2.1: Evaluation dataset format
"""
Evaluation dataset format:
[
{
'query': "How to install Python?",
'relevant_docs': ['doc_3', 'doc_7', 'doc_12']
},
{
'query': "What is FastAPI?",
'relevant_docs': ['doc_15', 'doc_22']
},
...
]
"""
# Save to JSON
import json
eval_dataset = [
{
'query': "How to install Python on macOS?",
'relevant_docs': ['chunk_doc1_3', 'chunk_doc1_7']
},
{
'query': "FastAPI vs Flask comparison",
'relevant_docs': ['chunk_doc5_2', 'chunk_doc5_8', 'chunk_doc7_1']
},
# ... more queries
]
with open('eval_dataset.json', 'w') as f:
json.dump(eval_dataset, f, indent=2)Step 3: A/B Testing Framework
3.1: Compare strategies
from scipy.stats import ttest_rel
def ab_test_chunking_strategies():
"""
A/B test: Fixed-size vs Semantic chunking
"""
# Strategy A: Fixed-size (500 tokens)
pipeline_a = RAGPipeline(chunk_size=500, overlap=50)
chunks_a, embeddings_a = pipeline_a.process_directory("./data")
index_a = FAISSIndex(dim=1536)
index_a.add(embeddings_a, chunks_a)
# Strategy B: Semantic chunking (paragraphs)
pipeline_b = RAGPipeline(chunk_size=300, overlap=30) # Smaller
chunks_b, embeddings_b = pipeline_b.process_directory("./data")
index_b = FAISSIndex(dim=1536)
index_b.add(embeddings_b, chunks_b)
# Evaluate both on the same dataset
evaluator = RetrievalEvaluator()
scores_a = []
scores_b = []
for item in eval_dataset:
query = item['query']
relevant = set(item['relevant_docs'])
# Evaluate strategy A
query_emb = pipeline_a.embedder.embed(query)
results_a = index_a.search(query_emb, k=10)
retrieved_a = [r['chunk'].id for r in results_a]
ndcg_a = evaluator.ndcg_at_k(retrieved_a, relevant, k=10)
scores_a.append(ndcg_a)
# Evaluate strategy B
query_emb = pipeline_b.embedder.embed(query)
results_b = index_b.search(query_emb, k=10)
retrieved_b = [r['chunk'].id for r in results_b]
ndcg_b = evaluator.ndcg_at_k(retrieved_b, relevant, k=10)
scores_b.append(ndcg_b)
# Averages
avg_a = np.mean(scores_a)
avg_b = np.mean(scores_b)
print(f"\n📊 A/B Test Results:")
print(f"Strategy A (Fixed-500): nDCG@10 = {avg_a:.3f}")
print(f"Strategy B (Semantic): nDCG@10 = {avg_b:.3f}")
print(f"Improvement: {((avg_b - avg_a) / avg_a * 100):.1f}%")
# Statistical significance (paired t-test)
t_stat, p_value = ttest_rel(scores_a, scores_b)
print(f"\nStatistical Significance:")
print(f" t-statistic: {t_stat:.3f}")
print(f" p-value: {p_value:.4f}")
if p_value < 0.05:
print(f" ✅ Significant difference (p < 0.05)")
else:
print(f" ⚠️ Not significant (p >= 0.05)")Troubleshooting
Problem 1: nDCG always 0
Cause: The retrieved IDs don't match the relevant ones
Solution:
# Check the ID format
print(f"Retrieved IDs: {retrieved[:3]}")
print(f"Relevant IDs: {list(relevant)[:3]}")
# Ensure consistency
retrieved = [chunk.id for chunk in chunks] # Use .id consistentlyProblem 2: Very low recall across all queries
Cause: The evaluation dataset doesn't reflect the indexed documents
Solution:
# Validate that relevant_docs exist in the index
for item in eval_dataset:
relevant = item['relevant_docs']
for doc_id in relevant:
if doc_id not in [c.id for c in chunks]:
print(f"⚠️ Doc {doc_id} not in index!")Summary
In this capsule you implemented:
- ✅
RetrievalEvaluatorwith 5 metrics (Recall, Precision, F1, MRR, nDCG) - ✅ Evaluation of individual queries
- ✅ Evaluation of complete datasets
- ✅ A/B testing framework
- ✅ Statistical significance testing (t-test)
- ✅ Evaluation dataset format
Next capsule: Query Expansion & Reranking - Two-stage retrieval to improve precision.
Additional Resources
- Information Retrieval Metrics - Stanford IR textbook
- nDCG Explained - Wikipedia guide
- RAG Evaluation Guide - LlamaIndex blog
- BEIR Benchmark - Standard IR benchmark
- scipy.stats - Statistical tests
- A/B Testing Guide - Microsoft research
- RAG Evaluation Metrics - DeepLearning.AI course
Module 8 - Capsule 05