Evaluation Framework: Midiendo Performance de RAG

Descripción

En esta cápsula implementarás un Evaluation Framework completo para medir objetivamente la calidad de tu sistema RAG. Sin métricas rigurosas, no puedes saber si tus decisiones (chunking strategy, embedding model, retrieval parameters) mejoran o empeoran el sistema.

Implementarás las métricas clave de information retrieval: Recall@K, Precision@K, MRR, y nDCG@K. También crearás un framework de A/B testing para comparar estrategias y calcular significancia estadística.

Al final tendrás un evaluator production-ready que te permite iterar basándote en datos, no intuición.

Duración estimada: 40-50 minutos

---

Objetivos

Al completar esta cápsula, serás capaz de:

• ✅ Implementar métricas de retrieval (Recall@K, Precision@K, MRR, nDCG@K)
• ✅ Crear evaluation datasets (queries + relevance judgments)
• ✅ Evaluar sistema RAG con métricas objetivas
• ✅ Comparar estrategias con A/B testing
• ✅ Calcular significancia estadística (t-test)
• ✅ Generar reportes de evaluación

---

Métricas clave de Retrieval

1. Recall@K

Definición: De todos los documentos relevantes, ¿qué % está en top-K?

Recall@K = (Docs relevantes en top-K) / (Total docs relevantes)

Ejemplo:

# Query: "Python installation"
relevant_docs = ['doc_3', 'doc_7', 'doc_12']  # 3 relevantes totales
retrieved_top5 = ['doc_1', 'doc_3', 'doc_5', 'doc_7', 'doc_9']

# En top-5 hay 2 relevantes: doc_3, doc_7
Recall@5 = 2 / 3 = 0.67 (67%)

---

2. Precision@K

Definición: De los top-K resultados, ¿qué % es relevante?

Precision@K = (Docs relevantes en top-K) / K

Ejemplo:

# Top-5: ['doc_1', 'doc_3', 'doc_5', 'doc_7', 'doc_9']
# Relevantes: doc_3, doc_7 (2 de 5)

Precision@5 = 2 / 5 = 0.40 (40%)

---

3. MRR (Mean Reciprocal Rank)

Definición: Posición del PRIMER documento relevante

RR = 1 / posición_primer_relevante
MRR = promedio de RR para múltiples queries

Ejemplo:

# Top-5: ['doc_1', 'doc_3', ...]
# Primer relevante: doc_3 en posición 2

RR = 1 / 2 = 0.50

---

4. nDCG@K (Normalized Discounted Cumulative Gain)

Definición: Métrica que premia resultados relevantes en posiciones altas

DCG@K = Σ (rel_i / log2(i + 1))  # Para i en [1..K]
nDCG@K = DCG@K / IDCG@K  # Normalizado por ideal

Ejemplo:

# Top-3: ['doc_3', 'doc_1', 'doc_7']
# Relevance: [1, 0, 1]  # doc_3 y doc_7 son relevantes

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%)

---

Paso 1: Implementar RetrievalEvaluator

1.1: Clase con todas las métricas

Crear src/evaluation/retrieval_evaluator.py:

"""
Retrieval Evaluator - RAG System
Métricas de evaluación para sistemas de retrieval
"""

import numpy as np
from typing import List, Dict, Set
from collections import defaultdict
import logging


class RetrievalEvaluator:
    """
    Evaluator para sistemas de retrieval
    
    Metrics:
    - Recall@K: % de docs relevantes recuperados
    - Precision@K: % de docs recuperados que son relevantes
    - MRR: Posición del primer documento relevante
    - nDCG@K: Relevancia ponderada por posición
    
    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:
        """
        Calcular Recall@K
        
        Args:
            retrieved: Lista de IDs recuperados (ordenados por score)
            relevant: Set de IDs relevantes (ground truth)
            k: Cantidad de resultados a considerar
        
        Returns:
            Recall score [0, 1]
        """
        if not relevant:
            return 0.0
        
        # Top-K recuperados
        retrieved_k = set(retrieved[:k])
        
        # Cuántos relevantes están en top-K
        relevant_retrieved = len(retrieved_k & relevant)
        
        # Recall = relevantes_recuperados / total_relevantes
        return relevant_retrieved / len(relevant)
    
    def precision_at_k(
        self,
        retrieved: List[str],
        relevant: Set[str],
        k: int
    ) -> float:
        """
        Calcular Precision@K
        
        Args:
            retrieved: Lista de IDs recuperados
            relevant: Set de IDs relevantes
            k: Cantidad de resultados a considerar
        
        Returns:
            Precision score [0, 1]
        """
        if k == 0:
            return 0.0
        
        retrieved_k = set(retrieved[:k])
        relevant_retrieved = len(retrieved_k & relevant)
        
        # Precision = relevantes_recuperados / K
        return relevant_retrieved / k
    
    def f1_at_k(
        self,
        retrieved: List[str],
        relevant: Set[str],
        k: int
    ) -> float:
        """
        Calcular F1@K (media armónica de Precision y Recall)
        
        Args:
            retrieved: Lista de IDs recuperados
            relevant: Set de IDs relevantes
            k: Cantidad de resultados
        
        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:
        """
        Calcular MRR (Mean Reciprocal Rank)
        
        Args:
            retrieved: Lista de IDs recuperados
            relevant: Set de IDs relevantes
        
        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  # Ningún relevante encontrado
    
    def ndcg_at_k(
        self,
        retrieved: List[str],
        relevant: Set[str],
        k: int
    ) -> float:
        """
        Calcular nDCG@K (Normalized Discounted Cumulative Gain)
        
        Args:
            retrieved: Lista de IDs recuperados
            relevant: Set de IDs relevantes
            k: Cantidad de resultados
        
        Returns:
            nDCG score [0, 1]
        """
        # DCG: Suma de 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: DCG ideal (todos los relevantes primero)
        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]:
        """
        Evaluar query con todas las métricas
        
        Args:
            retrieved: Lista de IDs recuperados
            relevant: Set de IDs relevantes
            k: Cantidad de resultados
        
        Returns:
            Dict con todas las métricas
        """
        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]:
        """
        Evaluar dataset completo (múltiples queries)
        
        Args:
            results: Lista de dicts con 'retrieved' y 'relevant' por query
            k: Cantidad de resultados
        
        Returns:
            Dict con métricas promediadas
        """
        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)
        
        # Promediar todas las métricas
        avg_metrics = {
            metric: np.mean(values)
            for metric, values in all_metrics.items()
        }
        
        # Agregar metadata
        avg_metrics['num_queries'] = len(results)
        
        return avg_metrics
    
    def print_evaluation(self, metrics: Dict[str, float]):
        """
        Imprimir métricas formateadas
        
        Args:
            metrics: Dict con métricas
        """
        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()
    
    # Query de ejemplo
    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)

---

Paso 2: Crear Evaluation Dataset

2.1: Formato de evaluation dataset

"""
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']
    },
    ...
]
"""

# Guardar en 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']
    },
    # ... más queries
]

with open('eval_dataset.json', 'w') as f:
    json.dump(eval_dataset, f, indent=2)

---

Paso 3: A/B Testing Framework

3.1: Comparar estrategias

from scipy.stats import ttest_rel

def ab_test_chunking_strategies():
    """
    A/B test: Fixed-size vs Semantic chunking
    """
    # Estrategia 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)
    
    # Estrategia B: Semantic chunking (paragraphs)
    pipeline_b = RAGPipeline(chunk_size=300, overlap=30)  # Más pequeño
    chunks_b, embeddings_b = pipeline_b.process_directory("./data")
    index_b = FAISSIndex(dim=1536)
    index_b.add(embeddings_b, chunks_b)
    
    # Evaluar ambas en mismo dataset
    evaluator = RetrievalEvaluator()
    
    scores_a = []
    scores_b = []
    
    for item in eval_dataset:
        query = item['query']
        relevant = set(item['relevant_docs'])
        
        # Evaluar estrategia 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)
        
        # Evaluar estrategia 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)
    
    # Promedios
    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

Problema 1: nDCG siempre 0

Causa: IDs de retrieved no coinciden con relevant

Solución:

# Verificar formato de IDs
print(f"Retrieved IDs: {retrieved[:3]}")
print(f"Relevant IDs: {list(relevant)[:3]}")

# Asegurar consistencia
retrieved = [chunk.id for chunk in chunks]  # Usar .id consistente

---

Problema 2: Recall muy bajo en todas las queries

Causa: Evaluation dataset no refleja documentos indexados

Solución:

# Validar que relevant_docs existen en el índice
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!")

---

Resumen

En esta cápsula implementaste:

• ✅ RetrievalEvaluator con 5 métricas (Recall, Precision, F1, MRR, nDCG)
• ✅ Evaluación de queries individuales
• ✅ Evaluación de datasets completos
• ✅ A/B testing framework
• ✅ Statistical significance testing (t-test)
• ✅ Formato de evaluation datasets

Próxima cápsula: Query Expansion & Reranking - Two-stage retrieval para mejorar precision.

---

Recursos Adicionales

1. Information Retrieval Metrics - Stanford IR textbook
2. nDCG Explained - Wikipedia guide
3. RAG Evaluation Guide - LlamaIndex blog
4. BEIR Benchmark - Standard IR benchmark
5. scipy.stats - Statistical tests
6. A/B Testing Guide - Microsoft research
7. RAG Evaluation Metrics - DeepLearning.AI course

---

Módulo 8 - Cápsula 05