Módulo 2: Chunking Strategies

Proyecto: Chunking Strategy Optimizer

Descripción del proyecto

Este proyecto integra todo el módulo: implementar las 4 chunking strategies, comparar con benchmarks cuantitativos, seleccionar strategy óptima para tu caso de uso, e integrar en baseline RAG del Módulo 1.

Objetivo: Mejorar precision del baseline +10-15% con chunking optimizado, documentar decisión con datos, y tener código reutilizable para production.

🎯 Objetivos del Proyecto

✅ Implementar 4 chunking strategies (fixed, recursive, semantic, structural)
✅ Comparar con benchmarks (precision, recall, latency, coherence)
✅ Seleccionar strategy óptima según document type
✅ Integrar en baseline RAG (reemplazar fixed-size)
✅ Documentar mejora (+10-15% precision esperado)

📁 Estructura del Proyecto

rag_baseline_project/  (del Módulo 1)
├── src/
│   ├── chunking/                    # NUEVO
│   │   ├── __init__.py
│   │   ├── base_chunker.py          # Interface base
│   │   ├── fixed_size_chunker.py    # Strategy 1
│   │   ├── recursive_chunker.py     # Strategy 2
│   │   ├── semantic_chunker.py      # Strategy 3
│   │   ├── structural_chunker.py    # Strategy 4
│   │   └── chunking_comparator.py   # Benchmark tool
│   └── indexing.py                  # Actualizar con chunking optimizado
└── notebooks/
    └── chunking_comparison.ipynb    # Demo interactivo

💻 Implementación

Paso 1: Base Chunker Interface

# src/chunking/base_chunker.py
from abc import ABC, abstractmethod
from typing import List

class BaseChunker(ABC):
    """Interface base para chunking strategies"""
    
    @abstractmethod
    def chunk(self, text: str) -> List[str]:
        """Dividir texto en chunks"""
        pass
    
    @abstractmethod
    def name(self) -> str:
        """Nombre de la strategy"""
        pass

Paso 2: Implementar Strategies

# src/chunking/recursive_chunker.py
from langchain.text_splitters import RecursiveCharacterTextSplitter
from .base_chunker import BaseChunker

class RecursiveChunker(BaseChunker):
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )
    
    def chunk(self, text: str) -> list[str]:
        return self.splitter.split_text(text)
    
    def name(self) -> str:
        return "recursive"

(Implementar similar para fixed_size, semantic, structural)

Paso 3: Comparator

# src/chunking/chunking_comparator.py
import time
import statistics
from typing import Dict

class ChunkingComparator:
    """Comparar chunking strategies con benchmarks"""
    
    def __init__(self, strategies: list):
        self.strategies = strategies
    
    def compare(self, document: str) -> Dict:
        """Comparar todas las strategies"""
        results = {}
        
        for strategy in self.strategies:
            # Medir latency
            start = time.time()
            chunks = strategy.chunk(document)
            latency = (time.time() - start) * 1000  # ms
            
            # Calcular métricas
            results[strategy.name()] = {
                "num_chunks": len(chunks),
                "avg_chunk_size": statistics.mean([len(c) for c in chunks]),
                "latency_ms": latency,
                "coherence": self._measure_coherence(chunks)
            }
        
        return results
    
    def _measure_coherence(self, chunks: list[str]) -> float:
        """Proxy: % de chunks que terminan con puntuación"""
        ends_with_punct = sum(1 for c in chunks if c.rstrip()[-1] in '.!?')
        return ends_with_punct / len(chunks) if chunks else 0

Paso 4: Ejecutar Benchmark

# benchmark_chunking.py
from src.chunking.fixed_size_chunker import FixedSizeChunker
from src.chunking.recursive_chunker import RecursiveChunker
from src.chunking.semantic_chunker import SemanticChunker
from src.chunking.chunking_comparator import ChunkingComparator

# Cargar documento
with open("data/fastapi_docs.txt", "r") as f:
    document = f.read()

# Configurar strategies
strategies = [
    FixedSizeChunker(chunk_size=500),
    RecursiveChunker(chunk_size=500, chunk_overlap=50),
    SemanticChunker(),  # Requiere OpenAI API key
]

# Comparar
comparator = ChunkingComparator(strategies)
results = comparator.compare(document)

# Mostrar resultados
for strategy_name, metrics in results.items():
    print(f"\n{strategy_name.upper()}:")
    print(f"  Chunks: {metrics['num_chunks']}")
    print(f"  Avg size: {metrics['avg_chunk_size']:.0f} chars")
    print(f"  Latency: {metrics['latency_ms']:.2f}ms")
    print(f"  Coherence: {metrics['coherence']:.2%}")

Output esperado:

FIXED_SIZE:
  Chunks: 250
  Avg size: 500 chars
  Latency: 0.08ms
  Coherence: 45%

RECURSIVE:
  Chunks: 268
  Avg size: 467 chars
  Latency: 1.25ms
  Coherence: 89%

SEMANTIC:
  Chunks: 185
  Avg size: 676 chars
  Latency: 18500ms
  Coherence: 96%

Paso 5: Integrar en Baseline RAG

# src/indexing.py (actualizado)
from src.chunking.recursive_chunker import RecursiveChunker  # NUEVO

class OptimizedIndexingPipeline:
    def __init__(self, chunking_strategy="recursive"):
        # Seleccionar chunking strategy
        if chunking_strategy == "recursive":
            self.chunker = RecursiveChunker(chunk_size=500, chunk_overlap=50)
        elif chunking_strategy == "semantic":
            self.chunker = SemanticChunker()
        # ... otros
        
        self.openai_client = OpenAI()
        self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
    
    def chunk_document(self, document: str) -> list[str]:
        """Usar chunking strategy seleccionada (no fixed-size)"""
        return self.chunker.chunk(document)
    
    # ... resto del código igual

📊 Medir Mejora

Antes (Módulo 1 baseline):

# Fixed-size chunking
baseline_results = {
    "precision@5": 0.68,
    "recall@50": 0.52,
    "coherence": 0.62
}

Después (Módulo 2 optimizado):

# Recursive chunking
optimized_results = {
    "precision@5": 0.78,  # +10%
    "recall@50": 0.57,    # +5%
    "coherence": 0.89     # +27%
}

Mejora conseguida: ✅ +10% precision, +5% recall (target alcanzado)

📝 Documentar Decisión

README.md actualizado:

# RAG System - Módulo 2: Chunking Optimizado

## Chunking Strategy Seleccionada: Recursive

### Justificación:
- **Mejora:** +10% precision, +5% recall vs baseline fixed-size
- **Costo:** Cero (no API calls adicionales)
- **Latency:** +1.2ms (negligible)
- **Coherence:** 89% vs 45% baseline (+44%)

### Alternativas Consideradas:
- **Semantic:** +15% precision pero $12/10K docs y +16s latency → Descartada por budget
- **Structural:** Solo aplicable para código → No aplica para dataset narrativo
- **Fixed-size:** Baseline simple pero -10% precision → Rechazada

### Configuración:
```python
RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,  # 10% overlap
    separators=["\n\n", "\n", ". ", " ", ""]
)

Performance Comparison (Módulo 1 vs Módulo 2)

Métrica	Módulo 1 (Fixed)	Módulo 2 (Recursive)	Delta
Precision@5	68%	78%	+10%
Recall@50	52%	57%	+5%
Coherence	62%	89%	+27%


---

## 🎯 Criterios de Éxito

✅ **Proyecto completo si:**

1. 4 strategies implementadas con código funcional
2. Benchmark comparativo ejecutado con resultados documentados
3. Strategy seleccionada con justificación basada en datos
4. Integración en baseline RAG completa
5. Mejora +10-15% precision vs Módulo 1 alcanzada

---

## 🚀 Bonus (Opcional)

- Notebook interactivo con visualizaciones de comparación
- Tests unitarios para cada chunking strategy
- CLI para probar diferentes strategies interactivamente
- Hybrid chunking (structural para código, recursive para texto)

---

## 📚 Entregables

1. **Código:** src/chunking/ folder completo con 4 strategies
2. **Benchmark results:** Tabla comparativa documentada
3. **Decision doc:** Justificación de strategy seleccionada
4. **Integration:** indexing.py actualizado con chunking optimizado
5. **Performance report:** Mejora vs baseline cuantificada

---

## 🎯 Resumen

**Proyecto Chunking Optimizer:**

- ✅ Implementar 4 strategies (fixed, recursive, semantic, structural)
- ✅ Comparar con benchmarks cuantitativos
- ✅ Seleccionar strategy óptima (típicamente: recursive)
- ✅ Mejorar baseline +10-15% precision
- ✅ Documentar decisión con datos

**Mejora esperada:** Precision 68% → 78% (+10%)

**Próximo módulo:** Módulo 3 optimiza query processing (+15-25% recall con query expansion, rewriting, HyDE).

---

## 📚 Recursos Adicionales

1. **[LangChain Text Splitters Cookbook](https://python.langchain.com/docs/modules/data_connection/document_transformers/)** - Ejemplos de código
2. **[Chunking Strategies Evaluation](https://arxiv.org/abs/2307.03172)** - Research paper
3. **[RAG Chunking Best Practices](https://www.pinecone.io/learn/chunking-strategies/)** - Pinecone guide
4. **[LlamaIndex Chunking](https://docs.llamaindex.ai/en/stable/module_guides/loading/node_parsers/)** - Alternative implementations

---

**Creado:** Febrero 6, 2026  
**Versión:** 1.0

← AnteriorCápsula 07: Decision framework — qué chunking strategy elegir Siguiente →Introducción a Query Optimization