Crear una tubería ETL de producción para RAG utilizando LangChain y FAISS.
Diseñar la división en bloques, el incrustado y el indexado vectorial para 40,000 PDFs de cumplimiento, con intentos repetidos, colas de correos no entregados, métricas de Prometheus y pruebas de rendimiento medidas.
Fuentes: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
Imagínese como el ingeniero de GenAI en una empresa fintech de tamaño medio. El equipo de cumplimiento posee más de 40,000 PDFs que incluyen documentos regulatorios, circulares del banco central, directrices de las autoridades de valores y políticas internas, distribuidos en SharePoint, S3 y unidades locales. Los analistas pasan de cuatro a seis horas al día buscando entre ellos manualmente. La solicitud del CTO es un chatbot que responda a cualquier pregunta relacionada con el cumplimiento de esa biblioteca en menos de tres segundos.
Obviamente se trata de un trabajo de generación reforzada por recuperación de información. Lo que es fácil pasar por alto es que la calidad de las respuestas se determina mucho antes de que llegue una pregunta, en el pipeline ETL que prepara la base de conocimientos. Un fragmentado inadecuado rompe el contexto, los embeddings ingenuos permiten que el significado se desvíe y una indexación deficiente ralentiza la recuperación de información. Una regla aproximada basada en la experiencia indica que alrededor del 80% de los problemas de calidad en los sistemas RAG son, en realidad, problemas del pipeline ETL. Por lo general, el modelo no es el cuello de botella, sino el pipeline.
La tecnología utilizada aquí es Python, LangChain, FAISS, Pinecone y embeddings de OpenAI.
La escala involucrada
- Aproximadamente 40,000 documentos en formatos mixtos: PDF, DOCX, HTML y texto plano.
- Aproximadamente 120 fragmentos por documento tras la división recursiva en bloques de 512 tokens.
- Un índice vectorial con 4.8 millones de embeddings en Pinecone, con 1536 dimensiones.
Arquitectura de la pipeline
La estructura es la clásica de extraer, transformar y cargar. Cada etapa oculta complejidades específicas del dominio que son fáciles de subestimar.
┌────────────────────────────────────────────────────────────────────────────┐
│ RAG ETL Pipeline │
│ │
│ [EXTRACT] [TRANSFORM 1] [TRANSFORM 2] [LOAD] │
│ Doc Loader → Chunker → Embedder → VectorDB → RAG API │
│ PDF·DOCX·HTML Semantic splits Batch encode Pinecone/FAISS │
└────────────────────────────────────────────────────────────────────────────┘
Cada etapa se ejecuta como un proceso idempotente independiente y escribe en un almacén temporal antes de que comience la siguiente. Esto proporciona tolerancia a fallos. Si el proceso de incrustación falla en el documento 30,000, nada obliga a extraer nuevamente los 40,000 documentos.
El orquestador es un ejecutor de etapas controlado por una dataclass:
# pipeline/orchestrator.py
# Top-level pipeline orchestration using a dataclass-driven stage runner
from dataclasses import dataclass, field
from typing import List, Optional
import logginglogger = logging.getLogger(__name__)@dataclass
class PipelineConfig:
source_bucket: str
chunk_size: int = 512 # tokens, not characters
chunk_overlap: int = 64 # token overlap between chunks
embedding_model: str = "text-embedding-3-small"
embedding_batch_size: int = 256
vector_store: str = "pinecone"
index_name: str = "compliance-docs"
namespace: str = "v2"
max_workers: int = 8 # parallel PDF extraction workersclass ETLPipeline:
def __init__(self, config: PipelineConfig):
self.config = config
self.loader = DocumentLoader(config)
self.chunker = AdaptiveChunker(config)
self.embedder = BatchEmbedder(config)
self.indexer = VectorIndexer(config) def run(self, doc_ids: Optional[List[str]] = None) -> dict:
"""Full ETL run. Pass doc_ids for incremental updates."""
stats = {} # Stage 1: Extract (parallelised — 2-5s per PDF, 8 workers)
raw_docs = self.loader.load(doc_ids=doc_ids)
stats["extracted"] = len(raw_docs)
logger.info(f"Extracted {len(raw_docs)} documents") # Stage 2a: Chunk
chunks = self.chunker.chunk_batch(raw_docs)
stats["chunks"] = len(chunks) # Stage 2b: Embed
embedded = self.embedder.embed_batch(chunks) # Stage 3: Index
upserted = self.indexer.upsert(embedded)
stats["upserted"] = upserted
return stats
El diagrama de flujo como fuente de verdad
Un diagrama de flujo en ASCII representa cada punto de decisión, desde la recepción de documentos en bruto, pasando por las reintentos y el agrupamiento, hasta la escritura final en el almacén vectorial.
┌──────────────────────┐
│ START: Doc Queue │
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ Detect Doc Type │ ← PDF · DOCX · HTML · MD · TXT
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ Parseable? (OCR) │
└────┬──────────┬──────┘
YES │ │ NO
│ ▼
│ ┌──────────────────┐
│ │ Dead-Letter Queue │ ← PagerDuty alert
│ └──────────────────┘
┌────────────▼────────────────────┐
EXTRACT │ Text Extraction + Cleaning │
│ PyMuPDF · python-docx · BS4 │
│ 2–5s / PDF · Parallel workers │
└────────────┬────────────────────┘
│
┌────────────▼────────────────────┐
│ Select Chunking Strategy │
└──────┬────────────┬─────────────┘
│ │ │
Code/Tables Default (prose) Long-form
│ │ │
┌────────▼──┐ ┌──────▼──────┐ ┌───▼─────────┐
TRANSFORM │ Fixed-Size│ │ Recursive │ │ Semantic │
(CHUNK) │ char split│ │ Char Split │ │ (NLTK+emb) │
└────────┬──┘ └──────┬──────┘ └───┬─────────┘
└────────────┼─────────────┘
│ (converge)
┌───────────▼─────────────┐
│ Attach Metadata Per Chunk│
│ doc_id·page·source·auth │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
│ Token Length Validation │
│ Assert chunk ≤ 8191 tok │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
│ Group Chunks → Batches │
│ batch_size=256 │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
TRANSFORM │ Call Embedding API │ ◄─────────┐
(EMBED) │ async · retry · rate │ │
└───────────┬─────────────┘ │
│ │
┌───────────▼─────────────┐ ┌─────┴────── ┐
│ API Call Success? │ NO │ Exp Backoff│
└──────┬──────────┬────────┘ │ 2^n+jitter │
YES │ └────────────►└─────────────┘
│
┌──────▼──────────────────┐
│ Normalize Embeddings │
│ L2 norm · float32 │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Write to Redis Cache │
│ hash(text)→vec · TTL7d │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
LOAD │ Upsert to Vector Store │
(INDEX) │ Pinecone · batch=100 │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Index Updated? │ NO ──► Retry DLQ (max 5)
└──────┬──────────────────┘
YES │
┌──────▼──────────────────┐
│ Emit Pipeline Metrics │
│ Prometheus · Grafana │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Pipeline Complete ✓ │
└─────────────────────────┘
Cada diamante en ese diagrama corresponde a una rama real if/else del código. Dibujarlo primero convierte el diagrama de flujo en la referencia de la arquitectura antes de que exista cualquier implementación, y no solo en documentación.
División en bloques
La división en bloques podría ser la decisión más influyente en todo el proceso. Un bloque es la unidad de recuperación. Si se hace demasiado grande, se desperdicia el espacio de contexto en texto irrelevante; si es demasiado pequeño, el sentido se desintegra. No existe un tamaño de bloque que sea universalmente adecuado.
División de tamaño fijo
El enfoque más simple consiste en dividir cada N caracteres o tokens, opcionalmente con solapamiento. Funciona bien para datos uniformes como registros de logs y tablas, pero arruina los límites entre oraciones y párrafos en textos narrativos.
# chunking/fixed_size.py
from langchain.text_splitter import CharacterTextSplitter
import tiktokenclass FixedSizeChunker:
"""
Use for: Code files, CSV-like data, log files.
Avoid for: Prose documents, legal text, RBI circulars.
"""
def __init__(self, chunk_size: int = 512, overlap: int = 50):
self.enc = tiktoken.get_encoding("cl100k_base")
self.splitter = CharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=overlap,
separator="", # hard cut — no respect for sentences
length_function=lambda t: len(self.enc.encode(t)),
) def chunk(self, text: str, metadata: dict) -> list[dict]:
chunks = self.splitter.split_text(text)
return [
{
"text": c,
"token_count": len(self.enc.encode(c)),
"chunk_idx": i,
**metadata,
}
for i, c in enumerate(chunks)
]
División recursiva de caracteres (valor predeterminado)
Este es el método principal. Intenta primero el separador más significativo (línea en blanco, luego salto de línea, luego punto, luego espacio, luego un único carácter) y solo recurre a una división más estricta cuando es necesario. RecursiveCharacterTextSplitter de LangChain lo implementa. La versión adaptativa que se muestra a continuación cuenta los tokens con tiktoken, de modo que ningún grupo supere el límite de 8192 tokens del modelo de embedding.
# chunking/recursive_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken
from typing import Listclass AdaptiveChunker:
"""
Production-grade chunker for compliance PDFs.
Uses tiktoken to count tokens (not characters), ensuring
we never exceed the 8192 embedding model context limit.
""" # Separator hierarchy: try broader splits first
SEPARATORS = [
"\n\n\n", # Section breaks
"\n\n", # Paragraph breaks
".\n", # Sentence at line end
". ", # Sentence in paragraph
"\n", # Line break
" ", # Word boundary (last resort)
] def __init__(self, config):
self.enc = tiktoken.encoding_for_model("text-embedding-3-small")
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=config.chunk_size, # 512 tokens
chunk_overlap=config.chunk_overlap, # 64 tokens
separators=self.SEPARATORS,
length_function=self._token_len,
is_separator_regex=False,
keep_separator=True, # preserve paragraph structure
) def _token_len(self, text: str) -> int:
return len(self.enc.encode(text, disallowed_special=())) def chunk_document(self, raw_doc: dict) -> List[dict]:
text = raw_doc["text"]
splits = self.splitter.split_text(text) chunks = []
for idx, chunk_text in enumerate(splits):
token_count = self._token_len(chunk_text) # Safety guard — should not happen, but log if it does
if token_count > 8191:
raise ValueError(
f"Chunk {idx} has {token_count} tokens — exceeds model limit. "
f"doc_id={raw_doc['doc_id']}"
) chunks.append({
"chunk_id": f"{raw_doc['doc_id']}_chunk_{idx}",
"text": chunk_text,
"token_count": token_count,
"chunk_index": idx,
"total_chunks": len(splits), # for reconstruction context
# Metadata propagated to vector store
"doc_id": raw_doc["doc_id"],
"source": raw_doc["source"],
"page_number": raw_doc.get("page_number", 0),
"doc_type": raw_doc["doc_type"],
"created_date": raw_doc["created_date"],
"authority": raw_doc.get("authority", "unknown"), # RBI, SEBI, etc.
}) return chunks def chunk_batch(self, raw_docs: List[dict]) -> List[dict]:
all_chunks = []
for doc in raw_docs:
try:
all_chunks.extend(self.chunk_document(doc))
except Exception as e:
logger.error(f"Chunking failed for {doc['doc_id']}: {e}")
return all_chunks
Fragmentación semántica
Aquí cada oración se incrusta, y el texto se divide donde la similitud coseno entre elementos adyacentes disminuye drásticamente. Los fragmentos resultantes son coherentes, pero el proceso es de 10 a 15 veces más lento y requiere llamadas de incrustación únicamente para la fragmentación, lo que duplica efectivamente el uso de la API. Solo tiene sentido para documentos narrativos de alto valor, como extensos dictámenes legales, informes anuales y documentos de política.
# chunking/semantic_chunker.py
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddingsclass SemanticChunkerWrapper:
"""
Use ONLY for: Long-form legal opinions, annual reports,
policy white papers where paragraph coherence is critical. Cost warning: This calls the embedding API TWICE per document
(once for chunking, once for indexing). Cache aggressively.
"""
def __init__(self, breakpoint_threshold: float = 0.85):
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.splitter = SemanticChunker(
embeddings=embeddings,
# Splits when cosine similarity between sentences drops below 0.85
breakpoint_threshold_type="cosine",
breakpoint_threshold_amount=breakpoint_threshold,
buffer_size=1, # include 1 sentence of context across boundaries
) def chunk(self, text: str) -> list[str]:
return self.splitter.split_text(text)
Cómo se comparan:
- Tamaño fijo: ideal para código, registros y datos tabulares. Rápido y económico, pero con baja coherencia.
- Recursivo (por defecto): mejor para PDF, DOCX y HTML en la mayoría de los casos. Rápido y económico, con coherencia media.
- Semántico: óptimo para documentos legales y de políticas extensos. Alta coherencia, pero 10 a 15 veces más lento y con aproximadamente el doble de llamadas a la API.
Generación de embeddings
Los embeddings convierten un fragmento en un vector denso que captura su significado. Las cuestiones de diseño son qué modelo utilizar, cómo procesar en lotes, si cachear los resultados y cómo manejar los límites de velocidad de forma asíncrona.
La aritmética relacionada con los límites de velocidad es preocupante. El modelo text-embedding-3-small de OpenAI permite un millón de tokens por minuto en el nivel 2. Con 4,8 millones de fragmentos que promedian 300 tokens cada uno, la cantidad total asciende a 1,44 mil millones de tokens. El procesamiento secuencial dentro de ese límite tomaría alrededor de 24 horas, mientras que el procesamiento por lotes asíncrono con trabajadores en paralelo lo reduce a unas dos horas y media.
# embedding/batch_embedder.py
import asyncio
import hashlib
import json
import time
from typing import List
import openai
import redis
import numpy as np
from tenacity import retry, stop_after_attempt, wait_exponentialclass BatchEmbedder:
"""
Async batch embedder with:
- Redis-backed cache (avoid re-embedding unchanged chunks)
- Exponential backoff on rate limit errors
- Token budget management to avoid 429s
"""
TOKEN_BUDGET_PER_MINUTE = 900_000 # Stay under 1M limit with headroom def __init__(self, config):
self.client = openai.AsyncOpenAI()
self.model = config.embedding_model
self.batch_size = config.embedding_batch_size # 256
self.redis = redis.Redis(host="localhost", decode_responses=True)
self.tokens_this_minute = 0
self.minute_start = time.monotonic() def _cache_key(self, text: str) -> str:
return f"emb:{hashlib.sha256(text.encode()).hexdigest()[:16]}" def _from_cache(self, text: str):
val = self.redis.get(self._cache_key(text))
return json.loads(val) if val else None def _to_cache(self, text: str, vector: list, ttl: int = 604800): # 7 days
self.redis.setex(self._cache_key(text), ttl, json.dumps(vector)) @retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=60),
)
async def _call_api(self, texts: List[str]) -> List[List[float]]:
response = await self.client.embeddings.create(
model=self.model,
input=texts,
encoding_format="float",
)
return [item.embedding for item in response.data] async def _embed_batch(self, chunks: List[dict]) -> List[dict]:
texts = [c["text"] for c in chunks]
cache_hits, uncached_texts, uncached_idxs = [], [], [] # Separate cache hits from misses
for i, text in enumerate(texts):
cached = self._from_cache(text)
if cached:
cache_hits.append((i, cached))
else:
uncached_texts.append(text)
uncached_idxs.append(i) # Token budget enforcement
elapsed = time.monotonic() - self.minute_start
if elapsed > 60:
self.tokens_this_minute = 0
self.minute_start = time.monotonic() estimated_tokens = sum(len(t.split()) * 1.3 for t in uncached_texts)
if self.tokens_this_minute + estimated_tokens > self.TOKEN_BUDGET_PER_MINUTE:
sleep_time = 60 - elapsed + 2
logger.info(f"Token budget hit, sleeping {sleep_time:.1f}s")
await asyncio.sleep(sleep_time)
self.tokens_this_minute = 0
self.minute_start = time.monotonic() # Embed uncached chunks
new_vectors = []
if uncached_texts:
new_vectors = await self._call_api(uncached_texts)
self.tokens_this_minute += estimated_tokens
for text, vec in zip(uncached_texts, new_vectors):
self._to_cache(text, vec) # Assemble results in original order
all_vectors = [None] * len(chunks)
for i, vec in cache_hits:
all_vectors[i] = vec
for i, vec in zip(uncached_idxs, new_vectors):
all_vectors[i] = vec # L2 normalize for cosine similarity in FAISS / Pinecone
for chunk, vector in zip(chunks, all_vectors):
arr = np.array(vector, dtype=np.float32)
arr /= np.linalg.norm(arr) + 1e-9
chunk["embedding"] = arr.tolist() return chunks async def embed_batch_async(self, all_chunks: List[dict]) -> List[dict]:
batches = [
all_chunks[i : i + self.batch_size]
for i in range(0, len(all_chunks), self.batch_size)
]
tasks = [self._embed_batch(batch) for batch in batches]
results = await asyncio.gather(*tasks, return_exceptions=True) embedded_chunks = []
for r in results:
if isinstance(r, Exception):
logger.error(f"Batch embedding failed: {r}")
else:
embedded_chunks.extend(r)
return embedded_chunks def embed_batch(self, chunks: List[dict]) -> List[dict]:
return asyncio.run(self.embed_batch_async(chunks))
La regla a seguir es nunca incrustar el mismo fragmento dos veces. Un caché respaldado por Redis reduce en un 70 a 90 por ciento las ejecuciones incrementales en colecciones que rara vez cambian.
Indexación
La indexación es donde los fragmentos incrustados terminan en un almacenamiento buscable. Dos decisiones son clave: el esquema de metadatos (para búsquedas filtradas) y el tipo de índice (Flat, HNSW o IVF).
Los metadatos diferencian a un sistema RAG básico de uno para producción. Tomemos como ejemplo la pregunta sobre lo que dijo la autoridad reguladora de valores sobre el trading interno en 2023. El sistema debe filtrar previamente con authority=SEBI y year=2023 antes de realizar cualquier búsqueda de vecino más cercano, lo que reduce el espacio de búsqueda y mejora la precisión.
# indexing/vector_indexer.py
from pinecone import Pinecone, ServerlessSpec
from typing import List
import uuid# ── Pinecone Metadata Schema ──────────────────────────────────────────
# These fields are indexed for filtered retrieval.
# Only add fields you will actually filter on — excess metadata costs money.
METADATA_SCHEMA = {
"doc_id": str, # for document-level retrieval
"chunk_index": int, # for ordered reconstruction
"total_chunks": int, # for context window sizing
"source": str, # S3 path or SharePoint URL
"doc_type": str, # "circular", "guideline", "policy"
"authority": str, # "RBI", "SEBI", "IRDAI", "internal"
"year": int, # publication year (for temporal filters)
"page_number": int, # for citation in UI
"token_count": int, # for context window management at query time
}class VectorIndexer:
def __init__(self, config):
self.pc = Pinecone(api_key=config.pinecone_api_key)
self.index_name = config.index_name
self.namespace = config.namespace
self._ensure_index() def _ensure_index(self):
existing = [i.name for i in self.pc.list_indexes()]
if self.index_name not in existing:
self.pc.create_index(
name=self.index_name,
dimension=1536, # text-embedding-3-small output dim
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
logger.info(f"Created Pinecone index: {self.index_name}")
self.index = self.pc.Index(self.index_name) def _build_vectors(self, embedded_chunks: List[dict]) -> list:
"""
Convert embedded chunk dicts to Pinecone upsert format.
Each vector = (id, embedding, metadata).
"""
vectors = []
for chunk in embedded_chunks:
vector_id = chunk.get("chunk_id") or str(uuid.uuid4())
metadata = {
k: chunk[k]
for k in METADATA_SCHEMA.keys()
if k in chunk
}
# Store a snippet for quick UI display without fetching full doc
metadata["text_preview"] = chunk["text"][:200] vectors.append({
"id": vector_id,
"values": chunk["embedding"],
"metadata": metadata,
})
return vectors def upsert(self, embedded_chunks: List[dict]) -> int:
vectors = self._build_vectors(embedded_chunks)
total_upserted = 0 # Pinecone upsert in batches of 100
for i in range(0, len(vectors), 100):
batch = vectors[i : i + 100]
self.index.upsert(vectors=batch, namespace=self.namespace)
total_upserted += len(batch) logger.info(f"Upserted {total_upserted} vectors to '{self.namespace}'")
return total_upserted def query(
self,
query_vector: list,
top_k: int = 10,
filters: dict = None
) -> list:
"""
Filtered ANN search. Example:
filters = {"authority": {"$in": ["SEBI", "RBI"]}, "year": {"$gte": 2022}}
"""
result = self.index.query(
vector=query_vector,
top_k=top_k,
namespace=self.namespace,
filter=filters,
include_metadata=True,
include_values=False, # we don't need the vectors back
)
return result.matches
FAISS para implementaciones locales
Cuando los datos no pueden ser enviados a una base de datos vectorial en la nube, lo cual es común en banca, finanzas y gobierno, FAISS es la solución estándar. Elegir el tipo de índice es muy importante.
# indexing/faiss_indexer.py
import faiss
import numpy as npclass FAISSIndexer:
def __init__(self, dim: int = 1536, n_vectors: int = 5_000_000):
self.dim = dim
# Index selection logic based on corpus size:
# < 1M vectors → IndexFlatIP (exact, fast enough)
# 1M–10M vectors → IndexHNSWFlat (M=32, efConstruction=64)
# > 10M vectors → IndexIVFPQ (nlist=4096, m=32)
if n_vectors < 1_000_000:
self.index = faiss.IndexFlatIP(dim)
elif n_vectors < 10_000_000:
self.index = faiss.IndexHNSWFlat(dim, 32) # M=32 edges per node
self.index.hnsw.efConstruction = 64
self.index.hnsw.efSearch = 200 # higher = more accurate, slower
else:
quantizer = faiss.IndexFlatIP(dim)
self.index = faiss.IndexIVFPQ(quantizer, dim, 4096, 32, 8) # GPU offload for faster build (comment out if no GPU)
res = faiss.StandardGpuResources()
self.index = faiss.index_cpu_to_gpu(res, 0, self.index) def add(self, vectors: np.ndarray):
# Vectors must be float32, L2 normalized for cosine similarity
faiss.normalize_L2(vectors)
self.index.add(vectors) def search(self, query: np.ndarray, k: int = 10):
faiss.normalize_L2(query)
distances, indices = self.index.search(query, k)
return distances, indices def save(self, path: str):
cpu_index = faiss.index_gpu_to_cpu(self.index)
faiss.write_index(cpu_index, path) def load(self, path: str):
self.index = faiss.read_index(path)
IndexFlatIPes adecuado para corpus con menos de 1 millón de vectores, ofrece una recuperación exacta del 100%, funciona a velocidad media y consume mucha memoria.IndexHNSWFlates adecuado para entre 1 y 10 millones de vectores, alcanza una recuperación del aproximado 98.5% en 10 consultas, es rápido en las búsquedas y utiliza una cantidad media de memoria.IndexIVFPQes adecuado para más de 10 millones de vectores, logra una recuperación del 94 al 96%, es muy rápido y consume poca memoria gracias a la compresión de 4 a 8 bits.
Gestión de fallos y reintentos
Cada etapa de un pipeline de producción fallará en ocasiones. Lo importante es hacerlo de manera ordenada: guardar los documentos defectuosos sin interrumpir el proceso y dejar un registro claro para su revisión manual. Una cola de correos no entregados cumple esta función, etiquetando cada fallo con la etapa en la que ocurrió.
# pipeline/fault_handler.py
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetimeclass FailureStage(Enum):
EXTRACTION = "extraction"
CHUNKING = "chunking"
EMBEDDING = "embedding"
INDEXING = "indexing"@dataclass
class PipelineFailure:
doc_id: str
stage: FailureStage
error_type: str
error_message: str
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
retry_count: int = 0
is_retriable: bool = Trueclass DeadLetterQueue:
"""Stores failed items for later inspection and retry."""
def __init__(self, redis_client, key_prefix="dlq"):
self.r = redis_client
self.prefix = key_prefix def push(self, failure: PipelineFailure):
key = f"{self.prefix}:{failure.stage.value}"
self.r.lpush(key, json.dumps(failure.__dict__))
logger.warning(
f"DLQ push: doc={failure.doc_id} stage={failure.stage.value} "
f"err={failure.error_type}: {failure.error_message[:80]}"
) def pop_batch(self, stage: FailureStage, n: int = 50) -> list:
key = f"{self.prefix}:{stage.value}"
items = self.r.lrange(key, 0, n - 1)
self.r.ltrim(key, n, -1)
return [json.loads(i) for i in items]# ── Categorize errors for smart retry decisions ────────────────────
RETRIABLE_ERRORS = {
"openai.RateLimitError",
"openai.APITimeoutError",
"pinecone.exceptions.PineconeApiException",
"ConnectionError",
}NON_RETRIABLE_ERRORS = {
"UnicodeDecodeError", # Corrupted doc — send to DLQ permanently
"ValueError", # Bad chunking logic — needs code fix
"pdfminer.pdfpage.PDFTextExtractionNotAllowed", # Encrypted PDF
}def is_retriable(exc: Exception) -> bool:
return type(exc).__name__ in RETRIABLE_ERRORS
Monitoreo
Un pipeline sin métricas es una caja negra. Haga un seguimiento al menos del ancho de banda, la latencia, la tasa de errores y el costo por documento, los cuales se exponen a Prometheus y se representan gráficamente en Grafana.
# monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server# ── Define metrics ─────────────────────────────────────────────────
docs_processed = Counter(
"etl_docs_processed_total",
"Total documents processed",
["status"], # labels: "success" or "failed"
)
chunks_created = Counter(
"etl_chunks_created_total",
"Total chunks created"
)
embedding_latency = Histogram(
"etl_embedding_seconds",
"Embedding API call latency",
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)
index_queue_depth = Gauge(
"etl_index_queue_depth",
"Vectors waiting to be indexed"
)
embedding_cache_hit_rate = Gauge(
"etl_cache_hit_rate",
"Redis cache hit rate for embeddings"
)
dlq_depth = Gauge(
"etl_dlq_depth",
"Dead-letter queue depth",
["stage"]
)# Usage in BatchEmbedder:
with embedding_latency.time(): # measures duration automatically
vectors = await self._call_api(uncached_texts)# Start Prometheus metrics endpoint on port 8001
start_http_server(8001)
Paneles útiles del panel de control:
- Documentos por minuto a lo largo del tiempo.
- Un histograma de la latencia de incrustación P95.
- Tasa de acierto de la caché Redis, con un objetivo superior al 80 %.
- Profundidad de la cola de mensajes no procesados por etapa, con alertas si permanece distinta de cero durante más de cinco minutos.
- Costo por documento, es decir, tokens multiplicados por el precio por token.
- Latencia de las consultas en FAISS o Pinecone en el momento de la recuperación (P50, P95, P99).
Resultados medidos
Estas cifras provienen del sistema de cumplimiento tras realizar ajustes, en una instancia AWS c5.4xlarge (16 vCPU, 32 GB de RAM) con trabajadores de incrustación asíncronos.
Extracción de PDF
PyMuPDF sin OCR tarda de 2 a 5 segundos por documento, dependiendo del número de páginas y el diseño, y el proceso se paraleliza de manera eficiente. Con 8 trabajadores concurrentes, extraer 40,000 PDFs toma aproximadamente:
40,000 docs × avg 3.5s / 8 workers = ~4.9 hours wall-clock
(scales linearly — 16 workers → ~2.4 hours)
Por tipo de archivo: un folleto circular sencillo de 1 a 5 páginas tarda de 2 a 3 segundos, ya que la capa de texto es sencilla. Una guía de 20 a 50 páginas con varias columnas y tablas incrustadas tarda de 3 a 4 segundos. Un informe anual de más de 100 páginas tarda de 4 a 5 segundos, ya que es necesario omitir diseños densos y muchas imágenes. Los PDF encriptados o escaneados van a la cola de correos no entregados, ya que necesitan OCR.
OCR es un proceso independiente. Los archivos escaneados que requieren Tesseract o AWS Textract tardan de 15 a 45 segundos cada uno y deben estar en un grupo de trabajadores aislado, para que no obstaculicen el proceso rápido.
Paso a paso: enfoque simple versus optimizado
- Extracción de PDF: De 2 a 5 segundos por documento en ambos casos, pero con 8 trabajadores en paralelo en un
ThreadPoolExecutoren lugar de un bucle secuencial. El trabajo está limitado por las esperas en el disco y la red, no por la CPU.
Extracción paralela
El cargador distribuye los documentos entre un grupo de hilos:
# loader/parallel_pdf_loader.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional
import fitz # PyMuPDFclass DocumentLoader:
"""
Parallel PDF extraction using ThreadPoolExecutor.
PDF parsing is I/O-bound (disk reads + text decode),
so threads outperform processes here. Benchmarks (c5.4xlarge, non-OCR PDFs):
- 1 worker: ~2–5s per doc, sequential
- 8 workers: ~2–5s per doc, 8x throughput
- 16 workers: diminishing returns past disk I/O saturation
""" def __init__(self, config):
self.max_workers = config.max_workers # default: 8 def _extract_single(self, doc_path: str) -> dict:
"""Extract text from a single non-OCR PDF via PyMuPDF."""
doc = fitz.open(doc_path)
pages_text = []
for page in doc:
pages_text.append(page.get_text("text"))
doc.close() full_text = "\n\n".join(pages_text)
return {
"doc_id": doc_path,
"text": full_text,
"page_count": len(pages_text),
"source": doc_path,
"doc_type": "pdf",
} def load(
self,
doc_paths: List[str],
doc_ids: Optional[List[str]] = None
) -> List[dict]:
raw_docs, failed = [], [] with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_path = {
executor.submit(self._extract_single, path): path
for path in doc_paths
}
for future in as_completed(future_to_path):
path = future_to_path[future]
try:
raw_docs.append(future.result())
except Exception as e:
logger.error(f"Extraction failed: {path} → {e}")
failed.append(path) logger.info(
f"Extracted {len(raw_docs)} docs, "
f"{len(failed)} failures routed to DLQ"
)
return raw_docs
Lecciones de su implementación a gran escala
La creación de un RAG en entorno de producción consiste principalmente en ingeniería ETL, y hay algunas lecciones destacables.
- La estrategia de fragmentación influye más en la calidad de la recuperación que la elección del modelo. Estudie los tipos de documentos antes de seleccionar un divisor.
- Mida la longitud en tokens, no en caracteres. Utilice
tiktoken. Los recuentos de caracteres no significan nada para un modelo de embedding. - Diseñe el esquema de metadatos antes de escribir cualquier código ETL. No se pueden añadir filtros a un corpus ya indexado sin volver a procesarlo.
- Almacene en caché los embeddings de forma intensiva. Con Redis, las ejecuciones incrementales en documentos actualizados cuestan del 10 al 15 por ciento de la primera ejecución completa.
- Utilice una cola de correos no entregados. Un PDF defectuoso no debería interrumpir la ejecución. Aíslelo, regístrelo, envíe una alerta y véalo más tarde.
IndexFlatIP a IndexHNSWFlat implica una reconstrucción completa, por lo que decida con antelación según el tamaño estimado del corpus.