Módulo 3: Query Optimization

Proyecto: Query Optimizer System

Siguiente →

Descripción del proyecto

Proyecto integrador del Módulo 3: implementar 4 técnicas de query optimization (Expansion, Rewriting, Decomposition, HyDE), comparar con benchmarks, seleccionar técnica óptima, e integrar en baseline RAG.

Objetivo: Mejorar recall +15-25% vs queries directas, documentar trade-offs (latency, cost), y tener código production-ready.

🎯 Objetivos del Proyecto

  1. ✅ Implementar 4 técnicas de query optimization
  2. ✅ Comparar con benchmarks (recall, precision, latency, cost)
  3. ✅ Seleccionar técnica óptima según requirements
  4. ✅ Integrar en baseline RAG (reemplazar queries directas)
  5. ✅ Documentar mejora (+15-25% recall esperado)

📁 Estructura del Proyecto

rag_baseline_project/  (del Módulo 1-2)
├── src/
│   ├── query_optimization/          # NUEVO
│   │   ├── __init__.py
│   │   ├── base_optimizer.py        # Interface base
│   │   ├── query_expansion.py       # Técnica 1
│   │   ├── query_rewriting.py       # Técnica 2
│   │   ├── query_decomposition.py   # Técnica 3
│   │   ├── hyde.py                  # Técnica 4
│   │   └── optimizer_comparator.py  # Benchmark tool
│   ├── retrieval.py                 # Actualizar con optimization
│   └── ...
└── notebooks/
    └── query_optimization_demo.ipynb

💻 Implementación

Paso 1: Base Optimizer Interface

# src/query_optimization/base_optimizer.py
from abc import ABC, abstractmethod

class BaseQueryOptimizer(ABC):
    """Interface base para query optimization"""
    
    @abstractmethod
    def optimize(self, query: str) -> str | list[str]:
        """Optimizar query"""
        pass
    
    @abstractmethod
    def name(self) -> str:
        """Nombre de la técnica"""
        pass

Paso 2: Implementar Técnicas

# src/query_optimization/query_expansion.py
from openai import OpenAI
from .base_optimizer import BaseQueryOptimizer

class QueryExpansion(BaseQueryOptimizer):
    def __init__(self, num_expansions: int = 5):
        self.num_expansions = num_expansions
        self.client = OpenAI()
    
    def optimize(self, query: str) -> list[str]:
        """Generar múltiples queries relacionadas"""
        
        prompt = f"""Generate {self.num_expansions} diverse search queries related to: "{query}"

Output format (one query per line):
1. [query]
2. [query]
..."""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "You are a search query expert."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7
        )
        
        # Parsear queries
        lines = response.choices[0].message.content.strip().split('\n')
        queries = [line.split('. ', 1)[1] if '. ' in line else line for line in lines if line]
        
        return queries[:self.num_expansions]
    
    def name(self) -> str:
        return "query_expansion"

(Implementar similar para query_rewriting.py, query_decomposition.py, hyde.py)

Paso 3: Query Optimizer Comparator

# src/query_optimization/optimizer_comparator.py
import time
import statistics
from typing import Dict

class QueryOptimizerComparator:
    """Comparar query optimization techniques con benchmarks"""
    
    def __init__(self, optimizers: list, rag_system):
        self.optimizers = optimizers
        self.rag_system = rag_system
    
    def compare(
        self,
        test_queries: list[str],
        ground_truth: dict
    ) -> Dict:
        """Comparar todas las técnicas"""
        results = {}
        
        for optimizer in self.optimizers:
            print(f"\nBenchmarking {optimizer.name()}...")
            
            recalls = []
            precisions = []
            latencies = []
            
            for query in test_queries:
                # Medir latency
                start = time.time()
                
                # Optimizar query
                optimized = optimizer.optimize(query)
                
                # Buscar (manejar single query o list)
                if isinstance(optimized, list):
                    # Multiple queries (expansion, decomposition)
                    all_results = []
                    for q in optimized:
                        res = self.rag_system.retrieve(q, top_k=10)
                        all_results.append(res)
                    # Merge con RRF
                    final_results = self._rrf_merge(all_results)[:5]
                else:
                    # Single query (rewriting, hyde)
                    final_results = self.rag_system.retrieve(optimized, top_k=5)
                
                latency = (time.time() - start) * 1000  # ms
                latencies.append(latency)
                
                # Calcular precision/recall
                retrieved = set(final_results['ids'])
                relevant = set(ground_truth[query]['relevant_docs'])
                
                relevant_retrieved = retrieved & relevant
                precision = len(relevant_retrieved) / len(retrieved) if retrieved else 0
                recall = len(relevant_retrieved) / len(relevant) if relevant else 0
                
                precisions.append(precision)
                recalls.append(recall)
            
            # Agregados
            results[optimizer.name()] = {
                "recall_avg": statistics.mean(recalls),
                "precision_avg": statistics.mean(precisions),
                "latency_p50": statistics.median(latencies),
                "latency_p95": statistics.quantiles(latencies, n=20)[18] if len(latencies) >= 20 else max(latencies)
            }
        
        return results
    
    def _rrf_merge(self, results_list: list, k: int = 60) -> list:
        """Reciprocal Rank Fusion para merge"""
        scores = {}
        for results in results_list:
            for rank, doc_id in enumerate(results['ids'], 1):
                if doc_id not in scores:
                    scores[doc_id] = 0
                scores[doc_id] += 1 / (k + rank)
        
        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return {"ids": [doc_id for doc_id, score in ranked]}

Paso 4: Ejecutar Benchmark

# benchmark_query_optimization.py
from src.query_optimization.query_expansion import QueryExpansion
from src.query_optimization.query_rewriting import QueryRewriting
from src.query_optimization.query_decomposition import QueryDecomposition
from src.query_optimization.hyde import HyDE
from src.query_optimization.optimizer_comparator import QueryOptimizerComparator
from src.rag_system import BaselineRAGSystem

# Configurar optimizers
optimizers = [
    QueryExpansion(num_expansions=5),
    QueryRewriting(),
    QueryDecomposition(),
    HyDE()
]

# RAG system
rag_system = BaselineRAGSystem()

# Test queries con ground truth
test_queries = [
    "How to implement authentication in FastAPI?",
    "What are FastAPI performance optimizations?",
    "Compare FastAPI and Flask",
    # ... 20 más
]

ground_truth = {
    "How to implement authentication in FastAPI?": {
        "relevant_docs": ["doc_123", "doc_456", "doc_789"]
    },
    # ... para todas las queries
}

# Comparar
comparator = QueryOptimizerComparator(optimizers, rag_system)
results = comparator.compare(test_queries, ground_truth)

# Mostrar resultados
print("\n" + "="*60)
print("QUERY OPTIMIZATION BENCHMARK RESULTS")
print("="*60)

for optimizer_name, metrics in results.items():
    print(f"\n{optimizer_name.upper()}:")
    print(f"  Recall: {metrics['recall_avg']:.2%}")
    print(f"  Precision: {metrics['precision_avg']:.2%}")
    print(f"  Latency P50: {metrics['latency_p50']:.0f}ms")
    print(f"  Latency P95: {metrics['latency_p95']:.0f}ms")

Output esperado:

============================================================
QUERY OPTIMIZATION BENCHMARK RESULTS
============================================================

QUERY_EXPANSION:
  Recall: 72%
  Precision: 80%
  Latency P50: 820ms
  Latency P95: 1,050ms

QUERY_REWRITING:
  Recall: 60%
  Precision: 88%
  Latency P50: 680ms
  Latency P95: 850ms

QUERY_DECOMPOSITION:
  Recall: 65%
  Precision: 85%
  Latency P50: 1,180ms
  Latency P95: 1,450ms

HYDE:
  Recall: 77%
  Precision: 82%
  Latency P50: 880ms
  Latency P95: 1,120ms

Paso 5: Integrar en Baseline RAG

# src/retrieval.py (actualizado)
from src.query_optimization.query_expansion import QueryExpansion

class OptimizedRetrievalPipeline:
    def __init__(self, optimization_technique: str = "expansion"):
        # Seleccionar técnica
        if optimization_technique == "expansion":
            self.optimizer = QueryExpansion()
        elif optimization_technique == "rewriting":
            self.optimizer = QueryRewriting()
        elif optimization_technique == "hyde":
            self.optimizer = HyDE()
        else:
            self.optimizer = None  # Fallback a query directa
        
        # ... resto del setup
    
    def retrieve(self, query: str, top_k: int = 5):
        """Retrieval con query optimization"""
        
        if self.optimizer:
            # Optimizar query
            optimized = self.optimizer.optimize(query)
            
            if isinstance(optimized, list):
                # Multiple queries → RRF merge
                all_results = []
                for q in optimized:
                    results = self._search(q, top_k=10)
                    all_results.append(results)
                return self._rrf_merge(all_results)[:top_k]
            else:
                # Single query
                return self._search(optimized, top_k=top_k)
        else:
            # Sin optimization (baseline)
            return self._search(query, top_k=top_k)

📊 Medir Mejora

Antes (Módulo 1-2 baseline):

# Queries directas + chunking optimizado
baseline_results = {
    "recall@50": 0.57,
    "precision@5": 0.78
}

Después (Módulo 3 optimizado):

# Query expansion + chunking optimizado
optimized_results = {
    "recall@50": 0.72,  # +15%
    "precision@5": 0.80  # +2%
}

Mejora conseguida: ✅ +15% recall (target alcanzado)

📝 Documentar Decisión

README.md actualizado:

# RAG System - Módulo 3: Query Optimization

## Query Optimization Técnica Seleccionada: Query Expansion

### Justificación:
- **Mejora:** +15% recall, +2% precision vs queries directas
- **Costo:** +$0.001/query (aceptable)
- **Latency:** +650ms (dentro de budget 1s)
- **Rationale:** Mejor balance recall/cost/latency

### Alternativas Consideradas:
- **HyDE:** +20% recall pero +$0.0015/query → Descartada por costo
- **Rewriting:** +10% precision pero +3% recall → Insuficiente para recall target
- **Decomposition:** +20% precision para complejas pero +1s latency → Too slow

### Configuración:
```python
QueryExpansion(num_expansions=5)

Performance Acumulada (Módulos 1-3)

MétricaMódulo 1Módulo 2Módulo 3Total Improvement
Precision@568%78% (+10%)80% (+2%)+12%
Recall@5052%57% (+5%)72% (+15%)+20%

---

## 🎯 Criterios de Éxito

✅ **Proyecto completo si:**

1. 4 técnicas implementadas con código funcional
2. Benchmark comparativo ejecutado
3. Técnica seleccionada con justificación
4. Integración en baseline RAG completa
5. Mejora +15-25% recall vs Módulo 2 alcanzada

---

## 🎯 Resumen

**Proyecto Query Optimizer:**

- ✅ Implementar 4 técnicas (Expansion, Rewriting, Decomposition, HyDE)
- ✅ Comparar con benchmarks cuantitativos
- ✅ Seleccionar técnica óptima (típicamente: Expansion)
- ✅ Mejorar baseline +15-25% recall
- ✅ Documentar trade-offs (latency, cost)

**Mejora esperada:** Recall 57% → 72% (+15%)

**Próximo módulo:** Módulo 4 optimiza retrieval con re-ranking (+20-25% precision después de retrieval inicial).

---

**Creado:** Febrero 6, 2026  
**Versión:** 1.0
← AnteriorCápsula 07: Decision framework — qué técnica de query optimization elegirSiguiente →Módulo 4: Re-ranking — la segunda etapa que transforma retrieval mediocre en excelente