Inicio / Inteligencia Artificial / AI Engineering Pro / RAG Avanzado

RAG Avanzado

Query transformation, HyDE, Contextual RAG, Self-RAG, CRAG y evaluación con RAGAS.

Avanzado
🔒 Solo lectura
📖

Estás en modo lectura

Puedes leer toda la lección, pero para marcar progreso, hacer ejercicios y ganar XP necesitas una cuenta Pro.

Desbloquear por $9/mes

RAG Avanzado: Reranking, Evaluación y Patrones de Producción

Más Allá del RAG Básico

El RAG "naïve" (chunk → embed → search → generate) funciona para demos, pero en producción necesitas patrones avanzados para manejar queries complejas, múltiples fuentes y calidad consistente.

Query Transformation

Query Expansion (Multi-Query)

Un solo query puede no capturar toda la intención del usuario. Genera múltiples variantes.

from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm,
)

# Query original: "¿Cómo escalar RAG?"
# Genera variantes:
#   1. "Estrategias de escalabilidad para sistemas RAG"
#   2. "Optimizar rendimiento de Retrieval-Augmented Generation"
#   3. "Arquitectura RAG para alto tráfico"
# Busca con todas y deduplica resultados

results = multi_query_retriever.invoke("¿Cómo escalar RAG?")

HyDE (Hypothetical Document Embeddings)

Genera un documento hipotético que respondería la pregunta, y usa ese embedding para buscar.

from langchain.chains import HypotheticalDocumentEmbedder

# HyDE: en lugar de buscar directamente con el embedding del query del usuario,
# primero genera un "documento hipotético" que respondería la pregunta usando el LLM,
# y luego busca con el embedding de ESE documento.
# Ventaja: alineación documento-documento (el embedding de un doc hipotético se parece
# más a los documentos reales que el embedding de una pregunta corta).
hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    base_embeddings=OpenAIEmbeddings(),
    # prompt_key: template para generar el doc hipotético.
    # Opciones: "web_search", "sci_fact", "fiqa", "trec_covid", etc.
    # "web_search" es el más general; cada uno ajusta el estilo del doc generado.
    prompt_key="web_search",
)

# En lugar de buscar con el embedding del query,
# genera un documento hipotético y busca con ese embedding
# → Mejor alineación documento-documento

Step-Back Prompting

Para queries muy específicas, primero genera una pregunta más general.

step_back_prompt = """
Dada esta pregunta específica, genera una pregunta más general 
que ayude a encontrar el contexto necesario.

Pregunta: {question}
Pregunta general:
"""

# "¿Cuál es el ef_construction óptimo para HNSW con 1M vectores?"
# Step-back: "¿Cómo configurar índices HNSW en bases de datos vectoriales?"

Contextual RAG (Anthropic)

Añade contexto a cada chunk antes de indexarlo para mejorar la búsqueda.

CONTEXT_PROMPT = """
Aquí está el documento completo:
<document>
{document}
</document>

Aquí está el chunk que vamos a indexar:
<chunk>
{chunk}
</chunk>

Genera un contexto breve (2-3 oraciones) que sitúe este chunk 
dentro del documento completo. Este contexto se prependerá al chunk
para mejorar la búsqueda.
"""

# Resultado: cada chunk indexado tiene contexto del documento padre
# "Este chunk pertenece a la sección de autenticación del manual de API.
#  Describe específicamente la configuración de JWT tokens."
# + chunk original

Self-RAG: Reflexión sobre Calidad

El modelo evalúa si necesita buscar y si los resultados son útiles.

class SelfRAG:
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever

    def answer(self, question: str) -> str:
        # 1. ¿Necesito buscar información?
        need_retrieval = self.llm.invoke(
            f"¿Necesitas información externa para responder: '{question}'? Sí/No"
        )

        if "sí" in need_retrieval.content.lower():
            # 2. Recuperar documentos
            docs = self.retriever.invoke(question)

            # 3. ¿Son relevantes?
            relevant_docs = []
            for doc in docs:
                is_relevant = self.llm.invoke(
                    f"¿Este documento es relevante para '{question}'?\n"
                    f"Documento: {doc.page_content[:500]}\nSí/No"
                )
                if "sí" in is_relevant.content.lower():
                    relevant_docs.append(doc)

            # 4. Generar respuesta con docs relevantes
            context = "\n".join(d.page_content for d in relevant_docs)
            response = self.llm.invoke(
                f"Contexto:\n{context}\n\nPregunta: {question}"
            )

            # 5. ¿La respuesta está soportada por el contexto?
            is_supported = self.llm.invoke(
                f"¿Esta respuesta está soportada por el contexto?\n"
                f"Respuesta: {response.content}\nContexto: {context}\nSí/No"
            )

            return response.content
        else:
            return self.llm.invoke(question).content

Corrective RAG (CRAG)

Evalúa la calidad de los documentos recuperados y toma acciones correctivas.

Query ──► Retriever ──► Evaluador de Relevancia
                              │
                    ┌─────────┼─────────┐
                    ▼         ▼         ▼
               CORRECTO   AMBIGUO   INCORRECTO
                    │         │         │
                    ▼         ▼         ▼
               Usar docs  Refinar   Web Search
                          query     como fallback
                    │         │         │
                    └─────────┼─────────┘
                              ▼
                          Generar respuesta

Evaluación Rigurosa de RAG

Framework RAGAS

from ragas import evaluate
from ragas.metrics import (
    faithfulness,           # ¿La respuesta es fiel al contexto?
    answer_relevancy,       # ¿La respuesta es relevante?
    context_precision,      # ¿Los docs recuperados son precisos?
    context_recall,         # ¿Se recuperaron todos los docs necesarios?
    context_entity_recall,  # ¿Se capturaron las entidades clave?
    answer_similarity,      # Similitud con respuesta de referencia
)

# Preparar dataset de evaluación
from datasets import Dataset

eval_data = {
    "question": [
        "¿Cómo configuro JWT en la API?",
        "¿Cuál es el rate limit?",
    ],
    "answer": [
        "Para configurar JWT, debes...",  # Respuesta generada por RAG
        "El rate limit es de 1000 req/min...",
    ],
    "contexts": [
        ["JWT se configura en config/auth.js...", "El middleware verifica..."],
        ["Rate limiting: 1000 requests por minuto..."],
    ],
    "ground_truth": [
        "JWT se configura editando config/auth.js...",  # Respuesta correcta
        "El rate limit es 1000 requests por minuto.",
    ],
}

dataset = Dataset.from_dict(eval_data)
result = evaluate(dataset=dataset, metrics=[
    faithfulness, answer_relevancy, context_precision, context_recall
])

print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 'context_precision': 0.85, ...}

Métricas Custom con LLM-as-Judge

JUDGE_PROMPT = """
Evalúa la calidad de esta respuesta RAG en una escala de 1-5:

Pregunta: {question}
Contexto recuperado: {context}
Respuesta generada: {answer}
Respuesta esperada: {expected}

Evalúa en estos criterios:
1. Precisión factual (1-5)
2. Completitud (1-5)
3. Relevancia (1-5)
4. Fidelidad al contexto (1-5)

Responde en JSON: {{"precision": X, "completitud": X, "relevancia": X, "fidelidad": X}}
"""

Patrones de RAG en Producción

Caching de Embeddings

import hashlib
import redis
import json

# ¿Por qué cachear embeddings? Cada llamada al modelo de embedding tiene:
# 1. Costo ($0.02-$0.13 por millón de tokens) — queries repetidos = dinero desperdiciado
# 2. Latencia (~100-300ms por llamada API) — el cache responde en <1ms desde Redis
class CachedEmbeddings:
    def __init__(self, embeddings, redis_client):
        self.embeddings = embeddings
        self.cache = redis_client

    def embed_query(self, text: str) -> list:
        # MD5 como cache key: rápido de calcular, longitud fija (32 chars),
        # suficiente para evitar colisiones en queries de texto (no es seguridad, es indexación)
        cache_key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"
        cached = self.cache.get(cache_key)

        if cached:
            return json.loads(cached)

        vector = self.embeddings.embed_query(text)
        # setex = SET + EXpire: guarda el valor Y establece TTL en una sola operación atómica
        # 3600 * 24 = 86400 segundos = 24 horas de TTL
        # 24h es un buen balance: los queries se repiten durante el día de trabajo,
        # pero no queremos cache stale si el modelo de embedding cambia
        self.cache.setex(cache_key, 3600 * 24, json.dumps(vector))  # 24h TTL
        return vector

Guardrails de Respuesta

# Guardrail = mecanismo de protección que previene que el sistema genere respuestas
# cuando no tiene contexto suficiente. Es mejor decir "no sé" que alucinar una respuesta
# incorrecta — especialmente crítico en aplicaciones médicas, legales o financieras.
class RAGWithGuardrails:
    def __init__(self, chain, min_context_score=0.7):
        self.chain = chain
        self.min_context_score = min_context_score

    def invoke(self, question: str) -> dict:
        docs_with_scores = self.vectorstore.similarity_search_with_score(
            question, k=5
        )

        # Guardrail: si ningún doc supera el threshold, no generar
        # Score es similitud coseno (0 a 1): 0 = irrelevante, 1 = idéntico
        top_score = max(score for _, score in docs_with_scores)
        if top_score < self.min_context_score:
            return {
                "answer": "No tengo información suficiente para responder.",
                "confidence": "low",
                "sources": [],
            }

        answer = self.chain.invoke(question)
        return {
            "answer": answer,
            # 0.85+ = "high": el contexto es muy relevante, alta confianza en la respuesta
            # 0.7-0.85 = "medium": contexto parcialmente relevante, respuesta probable pero no segura
            # <0.7 = rechazado arriba — mejor no responder que alucinar
            "confidence": "high" if top_score > 0.85 else "medium",
            "sources": [doc.metadata["source"] for doc, _ in docs_with_scores],
        }

Resumen

RAG avanzado introduce:

  1. Query transformation para mejorar la búsqueda
  2. Contextual RAG para chunks con contexto enriquecido
  3. Self-RAG/CRAG para auto-evaluación y corrección
  4. Evaluación con RAGAS y métricas específicas
  5. Patrones de producción: caching, guardrails, fallbacks

🧠 Preguntas de Repaso

1. ¿Qué es HyDE (Hypothetical Document Embeddings) y cuál es su ventaja principal?

  • A) Un modelo de embedding más rápido que text-embedding-3-large
  • B) Una técnica que genera un documento hipotético para luego buscar con su embedding, logrando alineación documento-documento
  • C) Un tipo de vector database optimizada para documentos largos
  • D) Un framework para evaluar la calidad de embeddings

Respuesta: B) — HyDE genera primero un documento hipotético que respondería la pregunta del usuario, y luego usa el embedding de ese documento para buscar. La ventaja es la alineación documento-documento (en vez de query-documento), mejorando la calidad de retrieval.

2. En Corrective RAG (CRAG), ¿qué acción se toma cuando los documentos recuperados son evaluados como "INCORRECTO"?

  • A) Se retorna un mensaje de error al usuario
  • B) Se usa Web Search como fallback para encontrar información relevante
  • C) Se aumenta el top_k y se busca de nuevo en la misma base
  • D) Se ignoran los documentos y se genera la respuesta sin contexto

Respuesta: B) — CRAG evalúa la calidad de los documentos recuperados y toma acciones diferenciadas: si son CORRECTOS los usa, si son AMBIGUOS refina la query, y si son INCORRECTOS recurre a Web Search como fallback.

3. ¿Por qué se usa Redis para cachear embeddings en un sistema RAG de producción?

  • A) Redis es la única base de datos que soporta embeddings
  • B) Porque queries repetidos ahorran costo ($0.02-$0.13/1M tokens) y latencia (~100-300ms se reduce a <1ms)
  • C) Redis es más preciso que las vector databases para búsqueda semántica
  • D) Solo se usa Redis para desarrollo, no para producción

Respuesta: B) — El caching con Redis (TTL de 24h, key = MD5 del texto) elimina llamadas repetidas a la API de embeddings, ahorrando costos significativos y reduciendo la latencia de ~100-300ms a menos de 1ms para queries ya vistos.

4. En el sistema de guardrails para RAG, ¿qué sucede cuando el top_score del contexto recuperado es menor a 0.7?

  • A) Se genera la respuesta normalmente pero con una advertencia
  • B) Se rechaza la consulta y se responde "No tengo información suficiente" en vez de generar una respuesta potencialmente incorrecta
  • C) Se busca en una base de datos secundaria
  • D) Se aumenta automáticamente la temperatura del modelo

Respuesta: B) — Si el score máximo del contexto recuperado cae por debajo de 0.7, el sistema rechaza generar una respuesta para evitar alucinaciones, respondiendo "No tengo información suficiente para responder esa pregunta con confianza." Esto es un guardrail crítico para producción.

¿Te gustó esta lección?

Con Pro puedes marcar progreso, hacer ejercicios, tomar quizzes, ganar XP y obtener tu constancia.

Ver planes desde $9/mes
← Embeddings e Indexación Avanzada 🔒 Desbloquear siguiente lección →