Checklist de Preparación de Datos para IA

Overview

Este checklist cubre la preparación de datos para sistemas LLM y RAG. Recorre cada sección antes de indexar datos en tu vector store o pipeline de fine-tuning. La mala calidad de los datos es la causa más común de fallos en RAG.

Cuándo Usar Este Checklist

• Estás construyendo un pipeline RAG nuevo y necesitas validar la calidad de los datos antes de indexar.
• Estás migrando un corpus de un vector store a otro y quieres re-validar la fragmentación y los embeddings.
• La calidad de recuperación bajó después de una actualización del corpus y necesitas auditar qué paso se degradó.
• Estás incorporando una fuente de datos nueva y necesitas confirmar que cumple el mismo estándar de calidad que las fuentes existentes.
• Estás preparando un dataset de fine-tuning y necesitas deduplicar, redactar PII y normalizar texto antes del entrenamiento.

---

1. Recolección de Datos

1.1 Inventario de Fuentes

• Todas las fuentes de datos identificadas y documentadas
• Formato de la fuente documentado (PDF, HTML, Markdown, JSON, CSV, base de datos)
• Acceso a la fuente verificado (claves de API, credenciales de base de datos, permisos de archivos)
• Cadencia de actualización de la fuente definida (tiempo real, por hora, diaria, semanal)
• Propiedad de la fuente documentada (quién mantiene cada fuente)

Source              | Format    | Size     | Refresh   | Owner
────────────────────┼───────────┼──────────┼───────────┼────────────
Product docs        | Markdown  | 50 MB    | Weekly    | Docs team
API reference       | OpenAPI   | 5 MB     | On change | Eng team
Support tickets     | JSON      | 200 MB   | Daily     | Support team
Internal wiki       | HTML      | 500 MB   | Daily     | Knowledge mgmt
Code repository     | Markdown  | 30 MB    | On commit | Eng team
Customer FAQs       | CSV       | 2 MB     | Monthly   | Support team

1.2 Legal y Cumplimiento

• Derechos de uso de datos verificados (¿puedes usar estos datos en un sistema de IA?)
• Términos de licencia revisados para cada fuente
• Datos personales identificados y documentados
• Estado de consentimiento verificado para contenido generado por usuarios
• Política de retención de datos aplicada
• Cumplimiento de GDPR/CCPA verificado para datos personales

---

2. Limpieza de Datos

2.1 Calidad del Texto

import re
from typing import List

def clean_text(text: str) -> str:
    # Elimina etiquetas HTML
    text = re.sub(r'<[^>]+>', '', text)
    
    # Elimina espacios en blanco excesivos
    text = re.sub(r'\s+', ' ', text)
    
    # Elimina caracteres no imprimibles
    text = re.sub(r'[^\x20-\x7E\n\r\t]', '', text)
    
    # Corrige problemas de codificación
    text = text.encode('utf-8', errors='ignore').decode('utf-8')
    
    # Elimina espacios al inicio y al final
    text = text.strip()
    
    return text

def remove_boilerplate(text: str, min_length: int = 100) -> str:
    # Elimina menús de navegación, pies de página, encabezados
    lines = text.split('\n')
    meaningful = [l for l in lines if len(l.strip()) > min_length]
    return '\n'.join(meaningful)

2.2 Checklist de Limpieza

• Etiquetas HTML eliminadas del contenido extraído de la web
• Formato Markdown normalizado
• Problemas de codificación corregidos (UTF-8 en todo el pipeline)
• Espacios en blanco excesivos colapsados
• Contenido repetitivo eliminado (navegación, pies de página, avisos de cookies)
• Documentos vacíos o casi vacíos eliminados (< 50 caracteres)
• Documentos duplicados identificados y gestionados
• Archivos corruptos identificados y eliminados
• Contenido no textual (imágenes, videos) extraído o referenciado por separado

---

3. Eliminación de PII y Datos Sensibles

3.1 Detección de PII

import re

PII_PATTERNS = {
    "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
    "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
    "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    "credit_card": r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
    "ip_address": r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
}

def detect_pii(text: str) -> dict:
    found = {}
    for pii_type, pattern in PII_PATTERNS.items():
        matches = re.findall(pattern, text)
        if matches:
            found[pii_type] = len(matches)
    return found

def redact_pii(text: str) -> str:
    for pii_type, pattern in PII_PATTERNS.items():
        text = re.sub(pattern, f'[REDACTED_{pii_type.upper()}]', text)
    return text

3.2 Checklist de PII

• Correos electrónicos redactados o eliminados
• Números de teléfono redactados
• Números de seguro social redactados
• Números de tarjetas de crédito redactados
• Direcciones IP redactadas (si son sensibles)
• Direcciones físicas redactadas
• Nombres de clientes/usuarios evaluados (puede ser necesario conservarlos para contexto)
• Claves de API y secretos eliminados
• URLs y nombres de host internos evaluados

---

4. Deduplicación

4.1 Deduplicación Exacta

import hashlib

def find_exact_duplicates(documents: list) -> dict:
    hashes = {}
    duplicates = {}
    
    for doc in documents:
        content_hash = hashlib.sha256(doc["content"].encode()).hexdigest()
        if content_hash in hashes:
            duplicates[doc["id"]] = hashes[content_hash]
        else:
            hashes[content_hash] = doc["id"]
    
    return duplicates

4.2 Detección de Duplicados Aproximados

Para la detección de duplicados aproximados, TfidfVectorizer de scikit-learn combinado con similitud coseno es un punto de partida común. Para corpus más grandes, considera enfoques basados en MinHash o LSH que escalan mejor que la comparación pareada O(n²).

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def find_near_duplicates(documents: list, threshold: float = 0.95) -> list:
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform([d["content"] for d in documents])
    
    similarity_matrix = cosine_similarity(tfidf_matrix)
    np.fill_diagonal(similarity_matrix, 0)
    
    duplicates = []
    for i in range(len(documents)):
        for j in range(i + 1, len(documents)):
            if similarity_matrix[i][j] > threshold:
                duplicates.append({
                    "doc_a": documents[i]["id"],
                    "doc_b": documents[j]["id"],
                    "similarity": similarity_matrix[i][j],
                })
    
    return duplicates

4.3 Checklist de Deduplicación

• Duplicados exactos eliminados
• Duplicados aproximados identificados (similitud coseno > 0.95)
• Duplicados aproximados revisados — conserva la versión más completa
• Duplicados entre fuentes identificados (mismo contenido de fuentes diferentes)
• Conflictos de versión resueltos (conserva la última versión de documentos actualizados)

---

5. Estrategia de Fragmentación

5.1 Selección de Tamaño de Fragmento

Content type          | Recommended chunk size | Overlap
──────────────────────┼───────────────────────┼─────────
Technical docs        | 512-1024 tokens        | 50-100
API reference         | 256-512 tokens         | 0-50
Support tickets       | 256-512 tokens         | 0
Code documentation    | 512-1024 tokens        | 50-100
Legal documents       | 1024-2048 tokens       | 100-200
FAQ entries           | One Q&A per chunk      | 0
Tables                | One table per chunk    | 0

5.2 Implementación de Fragmentación

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(documents: list, chunk_size: int = 512, overlap: int = 50) -> list:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    
    chunks = []
    for doc in documents:
        doc_chunks = splitter.split_text(doc["content"])
        for i, chunk in enumerate(doc_chunks):
            chunks.append({
                "id": f"{doc['id']}_chunk_{i}",
                "content": chunk,
                "metadata": {
                    "source": doc["source"],
                    "title": doc["title"],
                    "chunk_index": i,
                    "total_chunks": len(doc_chunks),
                },
            })
    
    return chunks

5.3 Checklist de Fragmentación

• Tamaño de fragmento apropiado para el tipo de contenido
• Solapamiento de fragmentos preserva el contexto entre límites
• Estructura del documento respetada (sin fragmentos que crucen límites de sección)
• Tablas conservadas intactas (sin dividir entre fragmentos)
• Bloques de código conservados intactos
• Metadatos adjuntos a cada fragmento (fuente, página, sección, título)
• Distribución de longitud de fragmentos verificada (sin valores atípicos extremos)

---

6. Generación de Embeddings

6.1 Selección de Modelo

Elige un modelo de embeddings según tus requisitos de calidad, coste y latencia. La guía de embeddings de OpenAI cubre las diferencias entre text-embedding-3-small y text-embedding-3-large. Para opciones autoalojadas, la documentación de embeddings de LangChain lista proveedores locales como Sentence-Transformers y modelos de HuggingFace.

Model                      | Dimensions | Cost / 1K tokens | Quality
───────────────────────────┼────────────┼─────────────────┼────────
text-embedding-3-small     | 1536       | $0.00002         | Good
text-embedding-3-large     | 3072       | $0.00013         | Best
text-embedding-ada-002     | 1536       | $0.00010         | Good
Cohere embed-english-v3    | 1024       | $0.00010         | Good
Sentence-Transformers      | 384-768    | Free (self-host) | Fair

6.2 Generación de Embeddings

from openai import OpenAI
import time

client = OpenAI()

def generate_embeddings(chunks: list, model: str = "text-embedding-3-small", batch_size: int = 100) -> list:
    embeddings = []
    
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        texts = [c["content"] for c in batch]
        
        response = client.embeddings.create(
            model=model,
            input=texts,
        )
        
        for j, embedding in enumerate(response.data):
            embeddings.append({
                "chunk_id": batch[j]["id"],
                "embedding": embedding.embedding,
                "content": batch[j]["content"],
                "metadata": batch[j]["metadata"],
            })
        
        # Límite de tasa
        time.sleep(0.5)
    
    return embeddings

6.3 Checklist de Embeddings

• Modelo de embeddings seleccionado según requisitos de calidad y coste
• Todos los fragmentos embebidos con el mismo modelo
• Dimensiones de embeddings consistentes en todos los fragmentos
• Tamaño de lote optimizado para throughput sin superar límites de tasa
• Embeddings fallidos reintentados
• Coste de embeddings seguido y dentro del presupuesto
• Embeddings almacenados en base de datos vectorial con metadatos

---

7. Configuración del Índice Vectorial

7.1 Selección de Índice

La documentación de Pinecone y la guía de pgvector cubren la configuración de índices en profundidad. La tabla siguiente resume las opciones principales.

Vector store    | Index type | Best for                    | Max vectors
────────────────┼────────────┼─────────────────────────────┼────────────
Pinecone        | HNSW       | Managed, production         | 1B+
Weaviate        | HNSW       | Self-hosted, flexible       | 1B+
pgvector        | IVFFlat    | PostgreSQL integration      | 100M
Qdrant          | HNSW       | Self-hosted, Rust-based     | 1B+
Milvus          | IVF/HNSW   | Large-scale, self-hosted    | 10B+
Chroma          | HNSW       | Local development           | 1M

7.2 Parámetros del Índice

# Ejemplo de Pinecone
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")
index = pc.create_index(
    name="knowledge-base",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1",
    ),
)

# Parámetros HNSW (para self-hosted)
HNSW_CONFIG = {
    "m": 16,              # Conexiones máximas por nodo
    "ef_construction": 200,  # Ancho de búsqueda en construcción
    "ef_search": 50,      # Ancho de búsqueda en consulta
}

7.3 Checklist de Índice

• Vector store seleccionado según requisitos de escala y latencia
• Tipo de índice apropiado para el tamaño del corpus
• Métrica de distancia seleccionada (coseno para embeddings normalizados)
• Parámetros del índice ajustados para el balance recall vs latencia
• Filtros de metadatos configurados (fuente, fecha, categoría)
• Índice construido y verificado con consultas de prueba
• Latencia de consulta medida (p95 < 200ms objetivo)

---

8. Validación de Calidad

8.1 Métricas de Calidad de Datos

Metric                  | Target  | Measurement
────────────────────────┼─────────┼──────────────────────
Average chunk length    | 400-800 | tokens per chunk
Chunk length variance   | Low     | No chunks < 50 o > 2000 tokens
Duplicate chunk rate    | < 2%    | % de near-duplicate chunks
PII detection rate      | 0%      | After redaction
Empty chunk rate        | 0%      | No empty chunks indexed
Metadata completeness   | 100%    | Every chunk has source + title
Embedding success rate  | > 99%   | % of chunks successfully embedded

8.2 Prueba de Calidad de Recuperación

def test_retrieval_quality(test_queries: list, vector_store, expected_sources: dict):
    results = []
    for query in test_queries:
        retrieved = vector_store.search(query, top_k=5)
        retrieved_sources = [r["metadata"]["source"] for r in retrieved]
        expected = expected_sources.get(query, [])
        
        # Verifica si la fuente esperada está en el top-k
        hit = any(src in retrieved_sources for src in expected)
        
        results.append({
            "query": query,
            "hit": hit,
            "retrieved_sources": retrieved_sources,
            "expected_sources": expected,
        })
    
    hit_rate = sum(1 for r in results if r["hit"]) / len(results)
    return {"hit_rate": hit_rate, "results": results}

# Ejecuta con 20+ consultas de prueba
# Objetivo: hit_rate > 0.85

8.3 Checklist de Calidad

• Longitud promedio de fragmentos dentro del rango objetivo
• Sin fragmentos vacíos o casi vacíos
• Sin PII detectada después de la redacción
• Tasa de duplicados por debajo del umbral
• Todos los fragmentos tienen metadatos completos
• Tasa de éxito de embeddings > 99%
• Tasa de acierto de recuperación > 0.85 en consultas de prueba
• Latencia de consulta p95 < 200ms

---

9. Checklist de Cierre

• Todas las fuentes de datos recolectadas y documentadas
• Cumplimiento legal verificado
• Texto limpio y normalizado
• PII redactada
• Duplicados eliminados
• Estrategia de fragmentación aplicada y validada
• Embeddings generados para todos los fragmentos
• Índice vectorial construido y configurado
• Métricas de calidad cumplen los objetivos
• Consultas de prueba de recuperación pasan
• Monitoreo y alertas configurados
• Pipeline de actualización de datos documentado

Errores Comunes

Omitir la auditoría de PII en documentos internos

Los wikis internos y los tickets de soporte suelen contener correos de clientes, números de teléfono y claves de API. Aunque los datos nunca salgan de la empresa, indexar PII en un vector store crea una vía de recuperación que elude los controles de acceso normales. Ejecuta detect_pii en cada fuente, incluso en las que confías, y registra lo que se redactó.

Usar un único tamaño de fragmento para todo

Un tamaño único de 512 tokens funciona para texto genérico pero falla en FAQs (demasiado grande, mezcla preguntas), documentos legales (demasiado pequeño, divide cláusulas) y tablas (separa filas de encabezados). Ajusta el tamaño al tipo de contenido usando la tabla de la sección 5.1, y mide la calidad de recuperación por tipo de contenido, no solo global.

Embeber antes de deduplicar

Si embebes duplicados aproximados, el vector store devuelve varias copias de la misma respuesta, desperdiciando tokens del contexto y confundiendo el ranking de recuperación. Ejecuta siempre la detección de duplicados exactos y aproximados antes de fragmentar y embeber. El umbral de similitud coseno de 0.95 detecta la mayoría de los duplicados parafraseados; bájalo a 0.90 si ves demasiados resultados redundantes.

Tratar el conjunto de prueba como el de entrenamiento

Si ajustas el tamaño de fragmento, el modelo de embeddings o los parámetros del índice con las mismas consultas que usas para validar la calidad de recuperación, estás sobreajustando a esas consultas. Reserva un conjunto separado de 20+ consultas que nunca uses para ajustes, y reporta la tasa de acierto solo en ese conjunto. Vuelve a ejecutar el conjunto reservado después de cada cambio de parámetros.

Reconstruir todo el índice en cada actualización

Para corpus que cambian a diario (tickets de soporte, ediciones de wiki), las reconstrucciones completas del índice desperdician cómputo y crean ventanas de indisponibilidad. La mayoría de los vector stores (Pinecone, Weaviate, pgvector, Qdrant) admiten upserts incrementales. Úsalos para actualizaciones diarias, y reserva las reconstrucciones completas para mantenimiento trimestral o cuando cambies el modelo de embeddings.

See Also

• Complete Guide to LLM Prompt Engineering
• AI RAG Evaluation Checklist
• Complete Guide to RAG in Production
• Build a RAG Pipeline with LangChain and Vector Databases
• Implement Semantic Search with Embeddings