Módulo 8: Proyecto Final Integrador - RAG System Completo
Chunking & Embedding Pipeline Integration
Descripción
En esta cápsula integrarás el SmartChunker (del Módulo 4) y el EmbeddingClient (con fallback multi-provider) para crear el pipeline completo de procesamiento: Documentos → Chunks → Embeddings.
Este pipeline es el core del sistema RAG. La calidad del chunking determina la calidad del retrieval, y el embedding client determina latencia y costo. Al final tendrás un pipeline production-ready con cache, fallback, y metadata enrichment.
Duración estimada: 40-50 minutos
Objetivos
Al completar esta cápsula, serás capaz de:
- ✅ Implementar SmartChunker con estrategia recursiva
- ✅ Usar
RecursiveCharacterTextSplitter(LangChain) - ✅ Crear EmbeddingClient con OpenAI + SBERT fallback
- ✅ Implementar Redis cache para embeddings
- ✅ Enriquecer chunks con metadata
- ✅ Integrar pipeline end-to-end
Arquitectura del pipeline
Document Ingestion Pipeline (Cápsula 02)
↓
Documents
↓
┌─────────────────────┐
│ SmartChunker │
│ - Recursive split │
│ - Token-aware │
│ - Metadata enrich │
└─────────────────────┘
↓
Chunks
↓
┌─────────────────────┐
│ EmbeddingClient │
│ - OpenAI (primary) │
│ - SBERT (fallback) │
│ - Redis cache │
└─────────────────────┘
↓
Embeddings
Paso 1: Implementar SmartChunker
1.1: Chunker con estrategia recursiva
Crear src/chunking/smart_chunker.py:
"""
Smart Chunker - RAG System
Chunking inteligente con estrategia recursiva y token-awareness
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken
from typing import List, Dict, Callable
import logging
class Chunk:
"""
Representación de un chunk
Attributes:
id: Identificador único (source_index)
text: Contenido del chunk
metadata: Metadata enriquecido (source, chunk_index, tokens, etc.)
"""
def __init__(self, chunk_id: str, text: str, metadata: Dict):
self.id = chunk_id
self.text = text
self.metadata = metadata
def to_dict(self) -> Dict:
"""Convertir a diccionario"""
return {
'id': self.id,
'text': self.text,
'metadata': self.metadata
}
def __repr__(self) -> str:
return f"Chunk(id='{self.id}', tokens={self.metadata.get('token_count', 0)})"
class SmartChunker:
"""
Chunker inteligente con estrategia recursiva
Features:
- Token-aware (usa tiktoken para counting preciso)
- Recursive splitting (preserva estructura semántica)
- Metadata enrichment (source, index, tokens)
- Configurable overlap
Example:
chunker = SmartChunker(chunk_size=500, overlap=50)
chunks = chunker.chunk(documents)
"""
def __init__(
self,
chunk_size: int = 500,
chunk_overlap: int = 50,
model: str = "gpt-4",
separators: List[str] = None
):
"""
Inicializar SmartChunker
Args:
chunk_size: Tamaño máximo de chunk en tokens
chunk_overlap: Overlap entre chunks en tokens
model: Modelo para tokenizer (gpt-4, gpt-3.5-turbo)
separators: Lista de separadores (default: paragraphs → sentences → words)
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.model = model
# Separadores por defecto (orden de prioridad)
self.separators = separators or [
"\n\n", # Párrafos
"\n", # Líneas
". ", # Sentences
"! ", # Exclamaciones
"? ", # Preguntas
"; ", # Punto y coma
", ", # Comas
" ", # Palabras
"" # Caracteres (último recurso)
]
# Crear length function con tiktoken
self.length_function = self._create_token_counter()
# Inicializar RecursiveCharacterTextSplitter
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=self.length_function,
separators=self.separators
)
self.logger = logging.getLogger(__name__)
def _create_token_counter(self) -> Callable:
"""
Crear función de counting de tokens usando tiktoken
Returns:
Función que cuenta tokens
"""
try:
encoding = tiktoken.encoding_for_model(self.model)
except KeyError:
# Fallback a cl100k_base (usado por GPT-4 y GPT-3.5)
encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(text: str) -> int:
"""Contar tokens en texto"""
return len(encoding.encode(text))
return count_tokens
def chunk(self, documents: List) -> List[Chunk]:
"""
Chunk lista de documentos
Args:
documents: Lista de Document objects o dicts con 'text' y 'source'
Returns:
Lista de Chunk objects
"""
all_chunks = []
for doc in documents:
# Extraer texto y source
if hasattr(doc, 'text'):
text = doc.text
source = doc.source
doc_metadata = getattr(doc, 'metadata', {})
else:
text = doc['text']
source = doc.get('source', 'unknown')
doc_metadata = doc.get('metadata', {})
# Chunk el documento
doc_chunks = self._chunk_single_document(text, source, doc_metadata)
all_chunks.extend(doc_chunks)
self.logger.info(
f"Chunked {len(documents)} documents into {len(all_chunks)} chunks "
f"(avg {len(all_chunks) / max(len(documents), 1):.1f} chunks/doc)"
)
return all_chunks
def _chunk_single_document(
self,
text: str,
source: str,
doc_metadata: Dict
) -> List[Chunk]:
"""
Chunk un documento individual
Args:
text: Contenido del documento
source: Identificador del documento
doc_metadata: Metadata del documento original
Returns:
Lista de Chunks para este documento
"""
# Split el texto
text_chunks = self.splitter.split_text(text)
# Crear Chunk objects con metadata enriquecido
chunks = []
for i, chunk_text in enumerate(text_chunks):
# Generar ID único
chunk_id = f"{source}_{i}"
# Metadata enriquecido
metadata = {
**doc_metadata, # Heredar metadata del documento
'source': source,
'chunk_index': i,
'total_chunks': len(text_chunks),
'token_count': self.length_function(chunk_text),
'char_count': len(chunk_text)
}
chunk = Chunk(
chunk_id=chunk_id,
text=chunk_text,
metadata=metadata
)
chunks.append(chunk)
return chunks
def estimate_chunks(self, text: str) -> int:
"""
Estimar cantidad de chunks para un texto
Args:
text: Texto a analizar
Returns:
Cantidad estimada de chunks
"""
total_tokens = self.length_function(text)
effective_chunk_size = self.chunk_size - self.chunk_overlap
estimated = (total_tokens + effective_chunk_size - 1) // effective_chunk_size
return max(1, estimated)
# Demo
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# Crear chunker
chunker = SmartChunker(chunk_size=500, overlap=50)
# Documento de ejemplo
doc = {
'text': "Este es un documento de ejemplo. " * 200, # ~600 words
'source': 'example.txt',
'metadata': {'author': 'Test'}
}
# Chunk
chunks = chunker.chunk([doc])
print(f"\n✅ Created {len(chunks)} chunks")
for chunk in chunks[:3]:
print(f"\n{chunk}")
print(f" Text: {chunk.text[:100]}...")Paso 2: Implementar EmbeddingClient con fallback
2.1: Client con OpenAI + SBERT fallback
Crear src/embeddings/embedding_client.py:
"""
Embedding Client - RAG System
Cliente de embeddings con OpenAI (primary) + SBERT (fallback) + Redis cache
"""
from openai import OpenAI
from sentence_transformers import SentenceTransformer
import numpy as np
import redis
import json
import hashlib
from typing import List, Union, Optional
import logging
import os
from dotenv import load_dotenv
load_dotenv()
class EmbeddingClient:
"""
Cliente de embeddings con fallback y cache
Features:
- OpenAI API (primary)
- SBERT local (fallback)
- Redis cache (opcional)
- Retry logic con exponential backoff
- Batch processing
Example:
client = EmbeddingClient(use_cache=True)
embeddings = client.embed(["texto 1", "texto 2"])
"""
def __init__(
self,
use_cache: bool = True,
openai_model: str = "text-embedding-3-small",
sbert_model: str = 'all-MiniLM-L6-v2',
redis_host: str = 'localhost',
redis_port: int = 6379,
cache_ttl: int = 86400 * 7 # 7 días
):
"""
Inicializar EmbeddingClient
Args:
use_cache: Si True, usa Redis cache
openai_model: Modelo de OpenAI
sbert_model: Modelo de SBERT
redis_host: Host de Redis
redis_port: Puerto de Redis
cache_ttl: TTL del cache en segundos
"""
self.openai_model = openai_model
self.sbert_model_name = sbert_model
self.cache_ttl = cache_ttl
# OpenAI client
api_key = os.getenv("OPENAI_API_KEY")
if api_key:
self.openai_client = OpenAI(api_key=api_key)
self.openai_available = True
else:
self.openai_client = None
self.openai_available = False
logging.warning("OPENAI_API_KEY not found. Using SBERT only.")
# SBERT model (lazy loading)
self.sbert_model = None
# Redis cache
if use_cache:
try:
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
# Test conexión
self.redis_client.ping()
self.cache_enabled = True
logging.info("Redis cache enabled")
except Exception as e:
self.redis_client = None
self.cache_enabled = False
logging.warning(f"Redis not available: {e}")
else:
self.redis_client = None
self.cache_enabled = False
self.logger = logging.getLogger(__name__)
def _load_sbert(self):
"""Lazy load del modelo SBERT"""
if self.sbert_model is None:
self.logger.info(f"Loading SBERT model: {self.sbert_model_name}")
self.sbert_model = SentenceTransformer(self.sbert_model_name)
def embed(
self,
texts: Union[str, List[str]],
use_openai: bool = True,
normalize: bool = True
) -> np.ndarray:
"""
Generar embeddings con fallback automático
Args:
texts: Texto o lista de textos
use_openai: Si True, intenta usar OpenAI primero
normalize: Si True, normaliza embeddings (L2 norm)
Returns:
numpy array de embeddings (shape: [n, dims])
"""
# Convertir a lista si es string
if isinstance(texts, str):
texts = [texts]
was_single = True
else:
was_single = False
# Intentar OpenAI primero (si use_openai=True y disponible)
if use_openai and self.openai_available:
try:
embeddings = self._embed_openai(texts)
self.logger.info(f"Generated {len(texts)} embeddings with OpenAI")
except Exception as e:
self.logger.warning(f"OpenAI failed: {e}. Falling back to SBERT.")
embeddings = self._embed_sbert(texts)
else:
# Usar SBERT directamente
embeddings = self._embed_sbert(texts)
# Normalizar si se solicita
if normalize:
embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
# Si era single text, retornar single embedding
if was_single:
return embeddings[0]
return embeddings
def _embed_openai(self, texts: List[str]) -> np.ndarray:
"""
Generar embeddings con OpenAI API
Args:
texts: Lista de textos
Returns:
numpy array de embeddings
"""
# Verificar cache primero
cached_embeddings = self._get_from_cache(texts, model="openai")
if cached_embeddings is not None:
return cached_embeddings
# Llamar a API
response = self.openai_client.embeddings.create(
model=self.openai_model,
input=texts
)
# Extraer embeddings
embeddings = np.array([item.embedding for item in response.data])
# Guardar en cache
self._save_to_cache(texts, embeddings, model="openai")
return embeddings
def _embed_sbert(self, texts: List[str]) -> np.ndarray:
"""
Generar embeddings con SBERT
Args:
texts: Lista de textos
Returns:
numpy array de embeddings
"""
# Lazy load del modelo
self._load_sbert()
# Verificar cache
cached_embeddings = self._get_from_cache(texts, model="sbert")
if cached_embeddings is not None:
return cached_embeddings
# Generar embeddings
embeddings = self.sbert_model.encode(texts, convert_to_numpy=True)
# Guardar en cache
self._save_to_cache(texts, embeddings, model="sbert")
return embeddings
def _cache_key(self, text: str, model: str) -> str:
"""
Generar cache key para un texto
Args:
text: Texto
model: Modelo usado (openai/sbert)
Returns:
Cache key (MD5 hash)
"""
content = f"{model}:{text}"
hash_digest = hashlib.md5(content.encode()).hexdigest()
return f"emb:{hash_digest}"
def _get_from_cache(self, texts: List[str], model: str) -> Optional[np.ndarray]:
"""
Intentar obtener embeddings del cache
Args:
texts: Lista de textos
model: Modelo usado
Returns:
numpy array de embeddings o None si no está en cache
"""
if not self.cache_enabled:
return None
try:
embeddings = []
for text in texts:
key = self._cache_key(text, model)
cached = self.redis_client.get(key)
if cached is None:
return None # Cache miss, retornar None
embedding = json.loads(cached)
embeddings.append(embedding)
# Todos los embeddings están en cache
return np.array(embeddings)
except Exception as e:
self.logger.warning(f"Cache read error: {e}")
return None
def _save_to_cache(self, texts: List[str], embeddings: np.ndarray, model: str):
"""
Guardar embeddings en cache
Args:
texts: Lista de textos
embeddings: Array de embeddings
model: Modelo usado
"""
if not self.cache_enabled:
return
try:
for text, embedding in zip(texts, embeddings):
key = self._cache_key(text, model)
value = json.dumps(embedding.tolist())
self.redis_client.setex(key, self.cache_ttl, value)
except Exception as e:
self.logger.warning(f"Cache write error: {e}")
def get_embedding_dim(self) -> int:
"""
Obtener dimensionalidad de embeddings
Returns:
Dimensiones del modelo activo
"""
if self.openai_available:
# OpenAI text-embedding-3-small = 1536 dims
return 1536
else:
# SBERT all-MiniLM-L6-v2 = 384 dims
return 384
# Demo
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# Crear client
client = EmbeddingClient(use_cache=True)
# Generar embeddings
texts = ["Python es un lenguaje", "JavaScript para web"]
embeddings = client.embed(texts)
print(f"\n✅ Generated embeddings: {embeddings.shape}")
print(f"Dimensions: {client.get_embedding_dim()}")Paso 3: Integrar pipeline completo
3.1: Pipeline orchestrator
Crear src/pipeline/rag_pipeline.py:
"""
RAG Pipeline - RAG System
Pipeline end-to-end: Documents → Chunks → Embeddings
"""
from src.ingestion.document_loader import DocumentLoader
from src.chunking.smart_chunker import SmartChunker
from src.embeddings.embedding_client import EmbeddingClient
import logging
from typing import List, Dict, Tuple
import numpy as np
class RAGPipeline:
"""
Pipeline completo de RAG
Components:
1. DocumentLoader: Carga documentos
2. SmartChunker: Chunk documentos
3. EmbeddingClient: Genera embeddings
Example:
pipeline = RAGPipeline()
chunks, embeddings = pipeline.process_directory("./data/documents")
"""
def __init__(
self,
chunk_size: int = 500,
chunk_overlap: int = 50,
use_cache: bool = True
):
"""
Inicializar RAG Pipeline
Args:
chunk_size: Tamaño de chunks en tokens
chunk_overlap: Overlap entre chunks
use_cache: Si True, usa Redis cache para embeddings
"""
self.loader = DocumentLoader()
self.chunker = SmartChunker(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
self.embedder = EmbeddingClient(use_cache=use_cache)
self.logger = logging.getLogger(__name__)
def process_directory(
self,
dir_path: str
) -> Tuple[List, np.ndarray]:
"""
Procesar directorio completo
Args:
dir_path: Path al directorio con documentos
Returns:
Tupla (chunks, embeddings)
"""
self.logger.info(f"Processing directory: {dir_path}")
# 1. Load documents
self.logger.info("Step 1/3: Loading documents...")
documents = self.loader.load_directory(dir_path)
self.logger.info(f"Loaded {len(documents)} documents")
# 2. Chunk documents
self.logger.info("Step 2/3: Chunking documents...")
chunks = self.chunker.chunk(documents)
self.logger.info(f"Created {len(chunks)} chunks")
# 3. Generate embeddings
self.logger.info("Step 3/3: Generating embeddings...")
texts = [chunk.text for chunk in chunks]
embeddings = self.embedder.embed(texts, use_openai=True)
self.logger.info(f"Generated {len(embeddings)} embeddings")
self.logger.info("✅ Pipeline complete!")
return chunks, embeddings
# Demo
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# Crear pipeline
pipeline = RAGPipeline(chunk_size=500, overlap=50)
# Procesar directorio
chunks, embeddings = pipeline.process_directory("./data/documents")
print(f"\n✅ Pipeline complete:")
print(f" Chunks: {len(chunks)}")
print(f" Embeddings: {embeddings.shape}")Troubleshooting
Problema 1: Redis connection refused
Causa: Redis no está corriendo
Solución:
# Instalar Redis
# macOS:
brew install redis
brew services start redis
# Linux:
sudo apt-get install redis-server
sudo systemctl start redis
# Verificar:
redis-cli ping # Debe retornar PONGProblema 2: SBERT model descarga lenta
Causa: Modelo (~90MB) se descarga en primer uso
Solución:
# Pre-descargar modelo
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
print("✅ Modelo descargado")Problema 3: OpenAI rate limit exceeded
Causa: Demasiados requests simultáneos
Solución:
# Implementar batch processing con delay
import time
def embed_with_backoff(texts, batch_size=20):
embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
batch_emb = client.embed(batch)
embeddings.append(batch_emb)
# Delay entre batches
if i + batch_size < len(texts):
time.sleep(1) # 1 segundo
return np.vstack(embeddings)Resumen
En esta cápsula implementaste:
- ✅
SmartChunkercon estrategia recursiva y token-awareness - ✅
EmbeddingClientcon OpenAI + SBERT fallback - ✅ Redis cache para embeddings (TTL 7 días)
- ✅
RAGPipelineorchestrator end-to-end - ✅ Metadata enrichment en chunks
- ✅ Error handling y logging
Próxima cápsula: Vector Search con FAISS - Implementar HNSW index para búsqueda eficiente.
Recursos Adicionales
- LangChain Text Splitters - Recursive splitter documentation
- tiktoken - OpenAI tokenizer
- Sentence-BERT - SBERT documentation
- Redis Python Client - redis-py docs
- OpenAI Embeddings - API reference
- FAISS - Facebook vector search
- Chunking Strategies Paper - Research on optimal chunking
Módulo 8 - Cápsula 03