Semantic Search

Crear un sistema RAG completo con LCEL

La construcción de un sistema RAG (Retrieval-Augmented Generation) representa la culminación de todos los conceptos que hemos explorado hasta ahora. Un sistema RAG combina la capacidad de recuperar información relevante de una base de conocimientos con la generación de respuestas contextualizadas mediante un modelo de lenguaje.

LCEL (LangChain Expression Language) nos permite crear este pipeline de forma elegante y eficiente, conectando cada componente del sistema mediante el operador | para formar una cadena de procesamiento fluida.

Arquitectura del sistema RAG

Un sistema RAG completo sigue un flujo de datos específico que transforma una pregunta del usuario en una respuesta fundamentada en documentos relevantes:

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

# Configuración del modelo y embeddings
model = ChatOpenAI(model="gpt-4o", temperature=0)
embeddings = OpenAIEmbeddings()

El pipeline RAG procesa la información siguiendo estos pasos fundamentales: recibe una consulta del usuario, recupera documentos relevantes del vector store, combina la consulta con el contexto recuperado en un prompt, genera una respuesta usando el modelo de lenguaje y devuelve la respuesta final formateada.

Preparación de la base de conocimientos

Antes de construir la cadena RAG, necesitamos preparar nuestra base de conocimientos con documentos procesados y almacenados en un vector store:

# Cargar y procesar documentos
loader = TextLoader("documentos/conocimiento_base.txt")
documents = loader.load()

# Dividir en chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# Crear vector store
vectorstore = FAISS.from_documents(chunks, embeddings)

Esta preparación establece la fuente de conocimiento que nuestro sistema RAG consultará para responder preguntas. El vector store actúa como una base de datos semántica que permite encontrar información relevante basándose en la similitud de significado.

Construcción del prompt template

El prompt template es crucial para el funcionamiento del sistema RAG, ya que debe instruir al modelo sobre cómo usar el contexto recuperado:

# Template para el sistema RAG
rag_prompt = ChatPromptTemplate.from_template("""
Eres un asistente útil que responde preguntas basándose únicamente en el contexto proporcionado.

Contexto:
{context}

Pregunta: {question}

Instrucciones:
- Responde únicamente basándote en la información del contexto
- Si la información no está en el contexto, indica que no tienes suficiente información
- Sé preciso y conciso en tu respuesta

Respuesta:
""")

Este template define la estructura de interacción entre el contexto recuperado y la pregunta del usuario, estableciendo reglas claras sobre cómo el modelo debe comportarse.

Implementación de la cadena RAG con LCEL

La cadena RAG completa se construye conectando el retriever, el prompt, el modelo y el parser de salida usando LCEL:

# Crear retriever desde el vector store
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# Función para formatear documentos
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

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

Esta implementación demuestra la elegancia de LCEL para crear pipelines complejos. El diccionario inicial define cómo se procesan los inputs: context se obtiene pasando la pregunta al retriever y formateando los documentos, mientras que question se pasa directamente usando RunnablePassthrough().

Ejemplo en funcionamiento:

Ejecución y uso del sistema

Una vez construida la cadena, el uso del sistema RAG es directo y eficiente:

# Hacer una pregunta al sistema RAG
pregunta = "¿Cuáles son los beneficios principales del aprendizaje automático?"
respuesta = rag_chain.invoke(pregunta)
print(respuesta)

El sistema procesa automáticamente toda la secuencia RAG: busca documentos relevantes, los incluye como contexto en el prompt, genera una respuesta fundamentada y devuelve el resultado final.

Ejemplo sin cadena LCEL equivalente:

Sistema RAG con manejo de contexto avanzado

Para casos más complejos, podemos implementar un sistema RAG mejorado que incluye validación y manejo de errores:

from langchain_core.runnables import RunnableLambda

def validate_and_format_context(docs):
    """Valida y formatea el contexto recuperado"""
    if not docs:
        return "No se encontró información relevante en la base de conocimientos."
    
    # Filtrar documentos por relevancia mínima si es necesario
    formatted_context = []
    for doc in docs:
        if len(doc.page_content.strip()) > 50:  # Filtro básico de calidad
            formatted_context.append(doc.page_content)
    
    return "\n\n".join(formatted_context) if formatted_context else "Contexto insuficiente."

# Cadena RAG mejorada
enhanced_rag_chain = (
    {
        "context": retriever | RunnableLambda(validate_and_format_context),
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | model
    | StrOutputParser()
)

Esta versión mejorada incluye validación del contexto recuperado, asegurando que el sistema maneje adecuadamente casos donde no se encuentra información relevante o la calidad del contexto es insuficiente.

Integración con streaming

LCEL permite implementar respuestas en streaming para mejorar la experiencia del usuario en aplicaciones interactivas:

# Usar el sistema RAG con streaming
for chunk in rag_chain.stream("¿Qué es la inteligencia artificial?"):
    print(chunk, end="", flush=True)

El streaming es especialmente útil en aplicaciones web o chatbots donde queremos mostrar la respuesta mientras se genera, proporcionando una experiencia más fluida y responsiva.

Este sistema RAG completo representa la base fundamental sobre la cual construiremos técnicas más avanzadas en las siguientes lecciones, incluyendo mejoras en la recuperación, filtrado de contexto y optimización de respuestas.

Retrievers: as_retriever() y configuración

Los retrievers actúan como la interfaz estandarizada entre los vector stores y el resto del sistema RAG. El método as_retriever() transforma cualquier vector store en un componente compatible con LCEL, proporcionando una API uniforme para la recuperación de documentos independientemente del tipo de almacén vectorial utilizado.

La conversión a retriever es fundamental porque permite que diferentes tipos de vector stores (FAISS, Chroma, Pinecone, etc.) se comporten de manera consistente dentro de las cadenas LCEL, facilitando el intercambio entre tecnologías sin modificar el resto del pipeline.

Configuración básica del retriever

El método as_retriever() acepta varios parámetros de configuración que determinan cómo se realiza la búsqueda de documentos relevantes:

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# Vector store previamente creado
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)

# Configuración básica del retriever
retriever = vectorstore.as_retriever(
    search_type="similarity",  # Tipo de búsqueda
    search_kwargs={"k": 5}     # Número de documentos a recuperar
)

El parámetro k controla la cantidad de documentos que el retriever devuelve para cada consulta. Un valor más alto proporciona más contexto pero puede introducir ruido, mientras que un valor más bajo es más preciso pero puede omitir información relevante.

Tipos de búsqueda disponibles

Los retrievers soportan diferentes estrategias de búsqueda que se adaptan a distintos casos de uso:

Búsqueda por similitud estándar:

# Recupera los k documentos más similares
similarity_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

Búsqueda con umbral de similitud:

# Solo recupera documentos que superen un umbral de similitud
threshold_retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "score_threshold": 0.7,  # Umbral mínimo de similitud
        "k": 10                  # Máximo número de documentos
    }
)

Búsqueda MMR (Maximum Marginal Relevance):

# Balancea relevancia y diversidad en los resultados
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "fetch_k": 20,      # Documentos candidatos iniciales
        "lambda_mult": 0.7  # Balance relevancia/diversidad (0-1)
    }
)

La búsqueda MMR es especialmente útil cuando queremos evitar documentos muy similares entre sí, promoviendo la diversidad en el contexto recuperado.

Configuración avanzada de parámetros

Los retrievers permiten configuraciones específicas según el tipo de vector store y las necesidades del sistema:

# Configuración detallada para diferentes escenarios
production_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 4,                    # Documentos a recuperar
        "filter": {"source": "docs"},  # Filtros de metadatos (si están disponibles)
        "include_metadata": True   # Incluir metadatos en los resultados
    }
)

Los filtros de metadatos son especialmente valiosos en sistemas con múltiples fuentes de información, permitiendo restringir la búsqueda a documentos específicos basándose en sus metadatos.

Personalización del comportamiento del retriever

Podemos crear retrievers personalizados que implementen lógica específica de filtrado o procesamiento:

from langchain_core.runnables import RunnableLambda

def custom_retriever_logic(query):
    """Lógica personalizada para el retriever"""
    # Recuperar documentos base
    base_docs = vectorstore.similarity_search(query, k=8)
    
    # Aplicar filtros personalizados
    filtered_docs = []
    for doc in base_docs:
        # Filtrar por longitud mínima
        if len(doc.page_content) > 100:
            # Filtrar por palabras clave relevantes
            if any(keyword in doc.page_content.lower() 
                   for keyword in ["importante", "clave", "fundamental"]):
                filtered_docs.append(doc)
    
    # Limitar a los mejores resultados
    return filtered_docs[:4]

# Crear retriever personalizado
custom_retriever = RunnableLambda(custom_retriever_logic)

Optimización del rendimiento del retriever

La configuración óptima del retriever depende del tamaño del corpus y los requisitos de latencia:

# Configuración para corpus grandes
large_corpus_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 3,           # Menos documentos para reducir latencia
        "fetch_k": 15,    # Candidatos iniciales moderados
        "lambda_mult": 0.8  # Priorizar relevancia sobre diversidad
    }
)

# Configuración para respuestas detalladas
detailed_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 7,  # Más contexto para respuestas completas
    }
)

Validación y monitoreo del retriever

Es importante implementar validación para asegurar que el retriever funciona correctamente:

def validate_retriever_results(retriever, test_query):
    """Valida el funcionamiento del retriever"""
    try:
        docs = retriever.invoke(test_query)
        
        print(f"Documentos recuperados: {len(docs)}")
        for i, doc in enumerate(docs):
            print(f"Doc {i+1}: {doc.page_content[:100]}...")
            if hasattr(doc, 'metadata'):
                print(f"Metadatos: {doc.metadata}")
        
        return len(docs) > 0
    except Exception as e:
        print(f"Error en retriever: {e}")
        return False

# Validar configuración
is_working = validate_retriever_results(
    retriever, 
    "¿Cuáles son los conceptos fundamentales?"
)

Integración con diferentes vector stores

Los retrievers mantienen compatibilidad entre diferentes tecnologías de almacenamiento vectorial:

# Ejemplo con Chroma
from langchain_community.vectorstores import Chroma

chroma_store = Chroma.from_documents(chunks, embeddings)
chroma_retriever = chroma_store.as_retriever(
    search_kwargs={"k": 4}
)

# Ejemplo con Pinecone (configuración similar)
# pinecone_retriever = pinecone_store.as_retriever(
#     search_kwargs={"k": 4}
# )

# Ambos retrievers se usan de la misma manera en LCEL
rag_chain_chroma = (
    {"context": chroma_retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

Esta flexibilidad permite cambiar la tecnología de almacenamiento subyacente sin modificar el resto del sistema RAG, facilitando la migración y experimentación con diferentes soluciones de vector stores.