RAG: Pipeline de Indexación con Chunking, Embeddings y FAISS
- Dacadev
- Ia generativa
- 30 de julio de 2026
Tabla de Contenido
En el artículo anterior de esta serie vimos que el indexing pipeline se compone de cuatro etapas: carga, fragmentación, conversión y almacenamiento. Hoy vamos a construir cada una de esas etapas con código real, usando como hilo conductor el ejemplo del primer artículo: un LLM que no sabe qué equipos jugarán la semifinal del Mundial 2026 porque su conocimiento no llega hasta esa fecha.
Note
Este es el tercer artículo de la serie RAGs. Vamos a implementar el indexing pipeline completo en Python: carga de datos, chunking, embeddings y una base de datos vectorial con FAISS. El código se irá acumulando en cada sección hasta terminar en un script único, listo para usar. 🚀
Carga de datos
El primer paso es traer la información desde su fuente original, sin importar si esta es una página web, un PDF o una API. Aunque suena trivial, esta etapa suele resolver varios problemas antes de entregar texto limpio: conexión y autenticación con la fuente, límites de rate limit, paginación y el formato en el que llegan los datos.
flowchart LR
S["📂 Fuente"] --> C["🔌 Conector<br/>conexión y auth"]
C --> E["📑 Extractor<br/>parseo de texto"]
E --> M["🏷️ Metadatos<br/>añade/actualiza"]
M --> T["🧹 Transformador<br/>limpieza y formato"]
T --> OUT["📄 Texto plano"]
Para nuestro ejemplo usaremos la página de Wikipedia de la Copa Mundial de Fútbol de 2026. Con LangChain podemos descargar el HTML de forma asíncrona y transformarlo a texto plano en pocas líneas.
import asyncio
from typing import Final
from langchain_community.document_loaders import AsyncHtmlLoader
from langchain_community.document_loaders.async_html import default_header_template
from langchain_community.document_transformers import Html2TextTransformer
URL_OBJETIVO: Final[str] = (
"https://es.wikipedia.org/wiki/Copa_Mundial_de_F%C3%BAtbol_de_2026"
)
async def cargar_y_limpiar_documento(url: str = URL_OBJETIVO) -> str:
"""
Descarga el contenido HTML de una URL de forma asíncrona
y lo transforma a texto plano.
"""
# Wikipedia exige identificar el User-Agent en las peticiones (https://w.wiki/4wJS)
encabezados = dict(default_header_template)
encabezados["User-Agent"] = "DacaDev/0.0 (mi_correo@gmail.com)"
cargador_html = AsyncHtmlLoader(url, header_template=encabezados)
documentos_html = cargador_html.load()
transformador_html_a_texto = Html2TextTransformer()
documentos_texto_plano = transformador_html_a_texto.transform_documents(
documentos_html
)
return documentos_texto_plano[0].page_content
Info
Dependencias usadas en esta sección: langchain, langchain-community, bs4 y html2text.
Segmentación en chunks
Con el texto limpio en mano, el siguiente problema es que no podemos meter un documento entero en cada consulta al LLM. Dividirlo en fragmentos —chunks— ataca tres limitaciones que ya vimos en el primer artículo:
- Context window limitado: si llenamos el contexto con un documento completo para usar solo una parte, desperdiciamos capacidad del modelo.
- Lost in the middle: en documentos largos, el modelo tiende a perder información ubicada en la mitad del texto.
- Precisión de la búsqueda semántica: chunks pequeños y enfocados producen coincidencias más relevantes que un documento gigante.
El proceso de chunking sigue siempre la misma lógica, sin importar la herramienta que uses:
flowchart LR
D["📄 Documento<br/>cargado"] --> P1["1️⃣ Dividir en<br/>unidades pequeñas"]
P1 --> P2["2️⃣ Fusionar en<br/>chunks de tamaño objetivo"]
P2 --> P3["3️⃣ Agregar overlap<br/>entre chunks"]
P3 --> CH[("🧩 Chunks")]
El overlap (solapamiento) repite un pequeño trozo del chunk anterior al inicio del siguiente, para no cortar una idea justo en la frontera entre dos fragmentos.
Métodos de chunking
No todos los métodos dividen el texto de la misma forma. Estos son los más comunes:
- Ancho fijo
- Recursivo
- Por tokens
- Estructural
Corta por un carácter específico (\n, por ejemplo) y mide el tamaño en caracteres. Es el método más simple y predecible.
from langchain_text_splitters import CharacterTextSplitter
separador = CharacterTextSplitter(
separator="\n",
chunk_size=1000,
chunk_overlap=200,
)
fragmentos = separador.create_documents([texto])
Prueba una lista de separadores en orden de jerarquía (párrafos, líneas, oraciones, palabras) hasta lograr chunks del tamaño deseado. Es el método recomendado para texto genérico.
from langchain_text_splitters import RecursiveCharacterTextSplitter
separador = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", ".", " ", ""],
chunk_size=1000,
chunk_overlap=200,
)
fragmentos = separador.create_documents([texto])
Igual que el ancho fijo o el recursivo, pero mide chunk_size y chunk_overlap en tokens en vez de caracteres, usando el encoder de un tokenizer.
separador = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
encoding_name="cl100k_base",
chunk_size=100,
chunk_overlap=10,
)
Divide según la estructura del documento (encabezados HTML, tablas) en vez de un tamaño fijo. Útil cuando el formato original aporta información.
from langchain_text_splitters import HTMLSectionSplitter
secciones = [("h1", "Encabezado 1"), ("h2", "Encabezado 2"), ("table", "Tabla")]
segmentador = HTMLSectionSplitter(secciones)
fragmentos = segmentador.split_text(contenido_html)
Tip
No existe un método “correcto” universal: depende del tipo de contenido, del tamaño esperado de las consultas y del modelo de embeddings que uses. Para nuestro caso —texto genérico proveniente de Wikipedia— el splitter recursivo es la mejor opción por defecto.
Sumamos esta etapa a nuestro flujo:
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
def dividir_texto_en_fragmentos(texto: str) -> list[Document]:
"""
Divide un texto intentando mantener oraciones y párrafos completos.
"""
separador_recursivo = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", ".", " ", ""],
chunk_size=1000,
chunk_overlap=200,
)
return separador_recursivo.create_documents([texto])
Dependencia añadida:
langchain-text-splitters.
Conversión a embeddings
Ya tenemos los chunks, pero un motor de búsqueda no entiende texto: entiende números. Los embeddings son la solución — representaciones vectoriales donde textos con significado similar quedan ubicados cerca entre sí en el espacio vectorial.
Con una sola dimensión apenas podemos ordenar palabras en una línea, lo que deja fuera relaciones de significado más ricas:
Al aumentar las dimensiones ganamos flexibilidad para ubicar cada palabra según su relación real con las demás:
Un modelo de embeddings generaliza esta idea a n dimensiones: convierte cada texto en un vector numérico que captura su significado.
Estos modelos se entrenan sobre grandes corpus de texto usando arquitecturas como Word2Vec, GloVe o BERT. Algunos de los modelos preentrenados más usados hoy:
| Proveedor | Modelo | Dimensiones |
|---|---|---|
| OpenAI | text-embedding-3-small/large | 1536 / 3072 |
text-embedding-004 | hasta 768 | |
| Voyage AI | voyage-large-2 | 1536 |
| Cohere | embed-multilingual-v3.0 | 1024 |
| Open source (HF) | sentence-transformers | varía (ej. 768) |
Cómo se mide la similitud
Una vez que dos textos son vectores, ¿cómo sabemos si son “parecidos”? La métrica más usada es la similitud coseno: el coseno del ángulo entre dos vectores. Cuanto más cercano a 1, más similares son; cercano a 0 implica que no están relacionados, y cercano a -1 indica significados opuestos.
Para nuestro pipeline usaremos un modelo open source de sentence-transformers vía LangChain, que genera embeddings de 768 dimensiones sin depender de una API externa:
from langchain_huggingface import HuggingFaceEmbeddings
def crear_modelo_embeddings() -> HuggingFaceEmbeddings:
"""
Instancia el modelo de embeddings que convertirá cada chunk en un vector.
"""
return HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")
Info
También podrías usar modelos privados como los de OpenAI (OpenAIEmbeddings) o Voyage AI simplemente cambiando esta función; el resto del pipeline no cambia.
Dependencias añadidas:
langchain-huggingface,sentence-transformers.
Almacenamiento en una base de datos vectorial
Ya tenemos vectores; ahora necesitamos guardarlos en algún lugar donde se puedan buscar por similitud de forma rápida. Ahí entran las bases de datos vectoriales, optimizadas para indexar y consultar vectores de alta dimensión.
| Categoría | Ejemplos |
|---|---|
| Índices vectoriales | FAISS, ANNOY, ScaNN |
| Bases especializadas | Pinecone, Chroma, Qdrant, Milvus, Weaviate |
| Plataformas de búsqueda | Elasticsearch, OpenSearch |
| SQL con soporte vectorial | PostgreSQL (pgvector), Azure SQL |
| NoSQL con soporte vectorial | MongoDB |
Para este ejemplo usaremos FAISS (Facebook AI Similarity Search): una librería ligera, local y suficiente para un prototipo o un proyecto pequeño.
from langchain_community.vectorstores import FAISS
def construir_base_vectorial(
fragmentos: list[Document], modelo_embeddings: HuggingFaceEmbeddings
) -> FAISS:
"""
Genera el embedding de cada chunk y los indexa en una base vectorial FAISS.
"""
return FAISS.from_documents(fragmentos, modelo_embeddings)
Tip
FAISS se puede persistir en disco con base_vectorial.save_local(ruta) y recuperar después con FAISS.load_local(ruta, modelo_embeddings), sin necesidad de recalcular los embeddings cada vez que arranca la aplicación.
Al elegir una base de datos vectorial para producción, más allá de FAISS, vale la pena evaluar:
| Factor | Qué implica |
|---|---|
| Precisión vs. velocidad | Índices más precisos suelen ser más lentos de consultar. |
| Local vs. cloud | Acceso rápido y control total vs. escalabilidad y redundancia administradas. |
| Costo | Autohospedado con costo de infraestructura vs. servicio administrado con costo por uso. |
Pipeline completo
Con las cuatro etapas listas, así queda el script completo: descarga la página, la limpia, la fragmenta, genera los embeddings y los indexa en FAISS.
import asyncio
from typing import Final
from langchain_community.document_loaders import AsyncHtmlLoader
from langchain_community.document_loaders.async_html import default_header_template
from langchain_community.document_transformers import Html2TextTransformer
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
URL_OBJETIVO: Final[str] = (
"https://es.wikipedia.org/wiki/Copa_Mundial_de_F%C3%BAtbol_de_2026"
)
RUTA_INDICE: Final[str] = "indice_mundial_2026"
async def cargar_y_limpiar_documento(url: str = URL_OBJETIVO) -> str:
"""
Descarga el contenido HTML de una URL de forma asíncrona
y lo transforma a texto plano.
"""
encabezados = dict(default_header_template)
encabezados["User-Agent"] = "DacaDev/0.0 (mi_correo@gmail.com)"
cargador_html = AsyncHtmlLoader(url, header_template=encabezados)
documentos_html = cargador_html.load()
transformador_html_a_texto = Html2TextTransformer()
documentos_texto_plano = transformador_html_a_texto.transform_documents(
documentos_html
)
return documentos_texto_plano[0].page_content
def dividir_texto_en_fragmentos(texto: str) -> list[Document]:
"""
Divide un texto intentando mantener oraciones y párrafos completos.
"""
separador_recursivo = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", ".", " ", ""],
chunk_size=1000,
chunk_overlap=200,
)
return separador_recursivo.create_documents([texto])
def crear_modelo_embeddings() -> HuggingFaceEmbeddings:
"""
Instancia el modelo de embeddings que convertirá cada chunk en un vector.
"""
return HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")
def construir_base_vectorial(
fragmentos: list[Document], modelo_embeddings: HuggingFaceEmbeddings
) -> FAISS:
"""
Genera el embedding de cada chunk y los indexa en una base vectorial FAISS.
"""
return FAISS.from_documents(fragmentos, modelo_embeddings)
async def ejecutar_flujo_indexacion() -> None:
"""
Flujo principal: carga, fragmenta, convierte a embeddings e indexa.
"""
print("Iniciando la descarga y procesamiento del documento...")
contenido_limpio = await cargar_y_limpiar_documento()
fragmentos = dividir_texto_en_fragmentos(contenido_limpio)
print(f"Se crearon {len(fragmentos)} fragmentos")
modelo_embeddings = crear_modelo_embeddings()
base_vectorial = construir_base_vectorial(fragmentos, modelo_embeddings)
base_vectorial.save_local(RUTA_INDICE)
print(f"Se indexaron {base_vectorial.index.ntotal} fragmentos en '{RUTA_INDICE}'")
if __name__ == "__main__":
asyncio.run(ejecutar_flujo_indexacion())
Verificación rápida
Con la base vectorial ya construida, podemos comprobar que la indexación funcionó haciendo una búsqueda por similitud directamente sobre FAISS, sin pasar todavía por un LLM:
base_vectorial = FAISS.load_local(
RUTA_INDICE, crear_modelo_embeddings(), allow_dangerous_deserialization=True
)
resultados = base_vectorial.similarity_search(
"¿Qué equipos jugarán la semifinal de la Copa del Mundo 2026?"
)
print(resultados[0].page_content)
Si todo salió bien, el chunk más relevante debería contener información sobre las semifinales del torneo — justo lo que el LLM no podía responder en el primer artículo de la serie.
Conclusión y próximos pasos
Ya tenemos un indexing pipeline funcional de punta a punta: carga los datos, los fragmenta en chunks manejables, los convierte en embeddings y los almacena en una base vectorial lista para consultarse. Esa base es exactamente lo que el retriever necesita para hacer su trabajo.
En el siguiente artículo construiremos el retrieval pipeline: cómo el retriever consulta esta base vectorial, arma el contexto y lo combina con el prompt del usuario antes de llegar al LLM.
