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 recuperacion. 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.

Aqui es donde RAG pasa de teoria a practica.

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: Disenar la Plantilla de Prompt

La plantilla de prompt es el puente entre la recuperacion y la generacion. Un prompt bien disenado 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 Diseno de Prompts

Fundamenta el modelo. La instruccion "Responde SOLO basandote en el contexto proporcionado" es la linea mas importante. Sin ella, el modelo mezclara libremente su conocimiento parametrico con el contexto recuperado, lo cual derrota el proposito de RAG.

Maneja lo desconocido con gracia. La instruccion "No tengo suficiente informacion" previene que el modelo alucine cuando el contexto no contiene la respuesta. Esto es una caracteristica, no una limitacion.

Solicita citas. Pedir referencias de fuentes hace la respuesta verificable. El usuario puede revisar el documento original si necesita mas detalle o quiere confirmar la afirmacion.

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

Aqui esta el pipeline entero en un unico 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 abstraccion de nivel mas 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 mas opinionado y requiere menos codigo para pipelines estandar. LangChain te da mas control sobre cada componente. Elige segun las preferencias de tu equipo y cuanta personalizacion necesitas.

Rastreo de Fuentes y Citas

El rastreo de fuentes no es opcional en RAG de produccion. Los usuarios necesitan verificar respuestas y entender de donde proviene la informacion.

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 Limite

Cuando el Contexto No Tiene Respuesta

Si el retriever retorna fragmentos que no son relevantes a la pregunta, el LLM deberia reconocerlo. Prueba tu sistema con preguntas que estan claramente fuera de la base de conocimiento para asegurarte de que el comportamiento "No lo se" funciona.

Cuando Multiples Fragmentos se Contradicen

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

Contextos Largos

Si recuperas muchos fragmentos, el contexto total podria exceder la atencion efectiva del LLM. Estrategias para manejar esto: reducir el numero de fragmentos recuperados, usar compresion contextual, o usar un modelo con ventana de contexto mas larga.

Consejos para Produccion

  • 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 interaccion. Estos datos son invaluables para depuracion y evaluacion.
  • Establece temperature en 0. Para RAG factual, quieres respuestas deterministas y fundamentadas. Temperature 0 reduce la variacion creativa.
  • Prueba con preguntas adversarias. Haz preguntas que estan ligeramente fuera de tema, usan diferente fraseado, o referencian cosas que no estan en tu base de conocimiento. Estas pruebas revelan donde se rompe tu pipeline.

Ahora tienes un pipeline RAG funcional. En la siguiente leccion, aprenderas patrones avanzados que van mas alla del basico recuperar-y-generar.