Saltar al contenido
Lección 7 de 12

Construyendo un Pipeline RAG

7 min read

De Componentes a Sistema

En las cinco lecciones anteriores, aprendiste cada componente del stack RAG de manera aislada: embeddings, bases de datos vectoriales, procesamiento de documentos, chunking y recuperación. Ahora es momento de ensamblarlos en un pipeline completo y funcional que tome la pregunta de un usuario, recupere documentos relevantes y genere una respuesta fundamentada con citas de fuentes.

Aquí es donde RAG pasa de teoría a práctica.

El Pipeline de Principio a Fin

Todo sistema RAG sigue el mismo flujo:

  1. Cargar -- ingestar documentos desde sus formatos de origen
  2. Fragmentar -- dividir documentos en segmentos recuperables
  3. Embeber -- convertir fragmentos en vectores
  4. Almacenar -- guardar vectores en una base de datos vectorial
  5. Recuperar -- encontrar fragmentos relevantes para una consulta
  6. Generar -- producir una respuesta fundamentada en el contexto recuperado

Construyamos cada paso.

Paso 1: Cargar y Fragmentar Documentos

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_documents(data_dir: str) -> list:
    """Cargar documentos de multiples formatos."""
    all_docs = []

    # Cargar PDFs
    pdf_loader = DirectoryLoader(
        data_dir, glob="**/*.pdf", loader_cls=PyPDFLoader
    )
    all_docs.extend(pdf_loader.load())

    # Cargar archivos Markdown
    md_loader = DirectoryLoader(
        data_dir, glob="**/*.md", loader_cls=TextLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    all_docs.extend(md_loader.load())

    return all_docs

def chunk_documents(docs: list, chunk_size: int = 1000, overlap: int = 200) -> list:
    """Dividir documentos en fragmentos."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = splitter.split_documents(docs)
    print(f"Divididos {len(docs)} documentos en {len(chunks)} fragmentos")
    return chunks

# Ejecutar
docs = load_documents("./data")
chunks = chunk_documents(docs)

Paso 2: Embeber y Almacenar

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

def create_vectorstore(chunks: list, persist_dir: str = "./chroma_db") -> Chroma:
    """Embeber fragmentos y almacenar en ChromaDB."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_metadata={"hnsw:space": "cosine"},
    )

    print(f"Almacenados {len(chunks)} fragmentos en la base de datos vectorial")
    return vectorstore

vectorstore = create_vectorstore(chunks)

Paso 3: Construir el Retriever

def create_retriever(vectorstore, k: int = 5):
    """Crear un retriever con MMR para diversidad."""
    return vectorstore.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": k,
            "fetch_k": 20,
            "lambda_mult": 0.7,
        },
    )

retriever = create_retriever(vectorstore)

Paso 4: Diseñar la Plantilla de Prompt

La plantilla de prompt es el puente entre la recuperación y la generación. Un prompt bien diseñado instruye al LLM para responder desde el contexto proporcionado, admitir cuando no sabe y citar sus fuentes.

from langchain_core.prompts import ChatPromptTemplate

RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """Eres un asistente util que responde preguntas basandose en
el contexto proporcionado. Sigue estas reglas estrictamente:

1. Responde SOLO basandote en el contexto proporcionado.
2. Si el contexto no contiene suficiente informacion para responder,
   di "No tengo suficiente informacion para responder esta pregunta."
3. Cita el documento fuente para cada afirmacion usando [Fuente: archivo].
4. Se conciso y directo. No agregues informacion mas alla de lo que
   el contexto proporciona.
5. Si multiples fuentes proporcionan informacion relevante, sintetizalas
   en una respuesta coherente.

Contexto:
{context}"""),
    ("human", "{question}"),
])

Principios Clave de Diseño de Prompts

Fundamenta el modelo. La instrucción "Responde SOLO basándote en el contexto proporcionado" es la línea más importante. Sin ella, el modelo mezclará libremente su conocimiento paramétrico con el contexto recuperado, lo cual derrota el propósito de RAG.

Maneja lo desconocido con gracia. La instrucción "No tengo suficiente información" previene que el modelo alucine cuando el contexto no contiene la respuesta. Esto es una característica, no una limitación.

Solicita citas. Pedir referencias de fuentes hace la respuesta verificable. El usuario puede revisar el documento original si necesita más detalle o quiere confirmar la afirmación.

Paso 5: Ensamblar la Cadena

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

def format_docs(docs):
    """Formatear documentos recuperados para el prompt."""
    formatted = []
    for doc in docs:
        source = doc.metadata.get("source", "Desconocido")
        formatted.append(f"[Fuente: {source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(formatted)

# Inicializar el LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Construir la cadena RAG
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | RAG_PROMPT
    | llm
    | StrOutputParser()
)

# Hacer una pregunta
answer = rag_chain.invoke("Cual es la politica de trabajo remoto de la empresa?")
print(answer)

Ejemplo Completo Funcional

Aquí está el pipeline entero en un único script ejecutable:

import os
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# --- Configuracion ---
DATA_DIR = "./data"
CHROMA_DIR = "./chroma_db"
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4o"
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200

# --- Paso 1: Cargar y fragmentar ---
loader = DirectoryLoader(DATA_DIR, glob="**/*.md", loader_cls=TextLoader,
                         loader_kwargs={"encoding": "utf-8"})
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP
)
chunks = splitter.split_documents(docs)
print(f"Cargados {len(docs)} docs -> {len(chunks)} fragmentos")

# --- Paso 2: Embeber y almacenar ---
embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory=CHROMA_DIR)

# --- Paso 3: Retriever ---
retriever = vectorstore.as_retriever(
    search_type="mmr", search_kwargs={"k": 5, "fetch_k": 20}
)

# --- Paso 4: Prompt ---
prompt = ChatPromptTemplate.from_messages([
    ("system", """Responde basandote SOLO en el contexto a continuacion.
Si no estas seguro, dilo. Cita fuentes como [Fuente: archivo].

Contexto:
{context}"""),
    ("human", "{question}"),
])

# --- Paso 5: Cadena ---
def format_docs(docs):
    return "\n\n---\n\n".join(
        f"[Fuente: {d.metadata.get('source', '?')}]\n{d.page_content}"
        for d in docs
    )

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model=LLM_MODEL, temperature=0)
    | StrOutputParser()
)

# --- Usarlo ---
response = chain.invoke("Cual es la politica de vacaciones para nuevos empleados?")
print(response)

Usando LlamaIndex en su Lugar

LlamaIndex proporciona una abstracción de nivel más alto para el mismo pipeline:

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# Configurar ajustes globales
Settings.llm = OpenAI(model="gpt-4o", temperature=0)
Settings.embed_model = OpenAIEmbedding(model_name="text-embedding-3-small")

# Cargar, fragmentar, embeber y almacenar en un solo paso
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)

# Consultar
query_engine = index.as_query_engine(similarity_top_k=5)
response = query_engine.query("Cual es la politica de vacaciones?")

print(response.response)
print("\nFuentes:")
for node in response.source_nodes:
    print(f"  - {node.metadata.get('file_name', 'Desconocido')} "
          f"(puntuacion: {node.score:.4f})")

LlamaIndex es más opinionado y requiere menos código para pipelines estándar. LangChain te da más control sobre cada componente. Elige según las preferencias de tu equipo y cuánta personalización necesitas.

Rastreo de Fuentes y Citas

El rastreo de fuentes no es opcional en RAG de producción. Los usuarios necesitan verificar respuestas y entender de dónde proviene la información.

from langchain_core.runnables import RunnableParallel

# Retornar tanto la respuesta como los documentos fuente
rag_with_sources = RunnableParallel(
    answer=rag_chain,
    sources=retriever,
)

result = rag_with_sources.invoke("Cual es la politica de PTO?")
print("Respuesta:", result["answer"])
print("\nFuentes:")
for doc in result["sources"]:
    print(f"  - {doc.metadata.get('source', 'Desconocido')}")
    print(f"    Vista previa: {doc.page_content[:100]}...")

Manejo de Casos Límite

Cuando el Contexto No Tiene Respuesta

Si el retriever retorna fragmentos que no son relevantes a la pregunta, el LLM debería reconocerlo. Prueba tu sistema con preguntas que están claramente fuera de la base de conocimiento para asegurarte de que el comportamiento "No lo sé" funciona.

Cuando Múltiples Fragmentos se Contradicen

Si diferentes documentos dicen cosas distintas, el LLM debería notar la discrepancia en lugar de elegir uno arbitrariamente. Agrega esto a tu prompt: "Si las fuentes se contradicen entre sí, menciona el desacuerdo."

Contextos Largos

Si recuperas muchos fragmentos, el contexto total podría exceder la atención efectiva del LLM. Estrategias para manejar esto: reducir el número de fragmentos recuperados, usar compresión contextual, o usar un modelo con ventana de contexto más larga.

Consejos para Producción

  • Usa streaming. Para interfaces de chat, transmite la respuesta token por token en lugar de esperar la respuesta completa. LangChain y LlamaIndex soportan streaming de forma nativa.
  • Registra todo. Registra la consulta, documentos recuperados y respuesta generada para cada interacción. Estos datos son invaluables para depuración y evaluación.
  • Establece temperature en 0. Para RAG factual, quieres respuestas deterministas y fundamentadas. Temperature 0 reduce la variación creativa.
  • Prueba con preguntas adversarias. Haz preguntas que están ligeramente fuera de tema, usan diferente fraseado, o referencian cosas que no están en tu base de conocimiento. Estas pruebas revelan dónde se rompe tu pipeline.

Ahora tienes un pipeline RAG funcional. En la siguiente lección, aprenderás patrones avanzados que van más allá del básico recuperar-y-generar.