RAG: Retrieval Augmented Generation

¿Qué es RAG?

RAG combina la capacidad generativa de un LLM con la recuperación de información de fuentes externas. En lugar de confiar solo en el conocimiento del modelo (que puede estar desactualizado o alucinar), RAG busca documentos relevantes y los inyecta como contexto.

¿Por Qué RAG?

Sin RAG:
  Usuario: "¿Cuál es la política de devolución?"
  LLM: "Generalmente, las tiendas aceptan devoluciones en 30 días..." 
  ❌ Respuesta genérica, no específica de TU empresa

Con RAG:
  1. Buscar en knowledge base → encuentra doc de devoluciones
  2. Inyectar doc en el contexto del LLM
  3. LLM responde basándose en TU documentación real
  ✓ "Según nuestra política, aceptamos devoluciones en 15 días..."

Pipeline RAG Completo

┌──────────────────────────────────────────────────┐
│                 PIPELINE RAG                      │
│                                                   │
│  INDEXACIÓN (offline, una vez):                   │
│  ┌──────┐   ┌───────┐   ┌──────────┐            │
│  │ Docs │ → │Chunker│ → │Embeddings│ → VectorDB │
│  └──────┘   └───────┘   └──────────┘            │
│                                                   │
│  CONSULTA (online, cada request):                 │
│  ┌──────┐   ┌──────────┐   ┌────────┐           │
│  │Query │ → │ Embedding│ → │Similary│→ Top-K docs│
│  └──────┘   └──────────┘   │ Search │            │
│                             └────────┘            │
│  ┌──────────────────────────────────┐            │
│  │   Prompt = System + Docs + Query │ → LLM     │
│  └──────────────────────────────────┘            │
└──────────────────────────────────────────────────┘

Implementación Paso a Paso

1. Carga de Documentos

from pathlib import Path

def load_documents(directory: str) -> list[dict]:
    docs = []
    for path in Path(directory).rglob("*"):
        if path.suffix in [".md", ".txt", ".pdf"]:
            docs.append({
                "content": path.read_text(),
                "metadata": {
                    "source": str(path),
                    "type": path.suffix,
                }
            })
    return docs

2. Chunking (División de Documentos)

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Dividir texto en chunks con overlap."""
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        
        # Intentar cortar en un punto natural
        if end < len(text):
            # Buscar el último punto, salto de línea o espacio
            for sep in ['\n\n', '\n', '. ', ' ']:
                last_sep = text[start:end].rfind(sep)
                if last_sep > chunk_size * 0.5:
                    end = start + last_sep + len(sep)
                    break
        
        chunk = text[start:end].strip()
        if chunk:
            chunks.append(chunk)
        
        start = end - overlap
    
    return chunks

Estrategias de Chunking

3. Generación de Embeddings

from openai import OpenAI

client = OpenAI()

def embed_texts(texts: list[str], model="text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(
        model=model,
        input=texts,
    )
    return [item.embedding for item in response.data]

def embed_query(query: str) -> list[float]:
    return embed_texts([query])[0]

4. Búsqueda y Generación

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def retrieve(query: str, chunks: list, embeddings: list, top_k: int = 5):
    query_emb = embed_query(query)
    
    scores = [cosine_similarity(query_emb, emb) for emb in embeddings]
    top_indices = np.argsort(scores)[-top_k:][::-1]
    
    return [(chunks[i], scores[i]) for i in top_indices]

def generate_with_context(query: str, retrieved_docs: list) -> str:
    context = "\n\n---\n\n".join([doc for doc, score in retrieved_docs])
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"""Responde basándote ÚNICAMENTE en el contexto proporcionado.
Si la información no está en el contexto, di "No tengo esa información".

Contexto:
{context}"""},
            {"role": "user", "content": query}
        ],
        temperature=0,
    )
    
    return response.choices[0].message.content

RAG Avanzado

Hybrid Search (Embeddings + BM25)

from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, chunks, embeddings):
        self.chunks = chunks
        self.embeddings = embeddings
        tokenized = [chunk.lower().split() for chunk in chunks]
        self.bm25 = BM25Okapi(tokenized)
    
    def search(self, query, top_k=5, alpha=0.5):
        # Semantic search
        query_emb = embed_query(query)
        sem_scores = [cosine_similarity(query_emb, e) for e in self.embeddings]
        
        # Keyword search (BM25)
        bm25_scores = self.bm25.get_scores(query.lower().split())
        
        # Normalizar y combinar
        sem_norm = self._normalize(sem_scores)
        bm25_norm = self._normalize(bm25_scores)
        
        combined = [alpha * s + (1 - alpha) * b 
                    for s, b in zip(sem_norm, bm25_norm)]
        
        top_idx = np.argsort(combined)[-top_k:][::-1]
        return [(self.chunks[i], combined[i]) for i in top_idx]
    
    def _normalize(self, scores):
        min_s, max_s = min(scores), max(scores)
        if max_s == min_s:
            return [0] * len(scores)
        return [(s - min_s) / (max_s - min_s) for s in scores]

Re-ranking

def rerank(query: str, documents: list[str], top_k: int = 3) -> list:
    """Re-rankear documentos usando el LLM como juez."""
    prompt = f"""Dado la pregunta: "{query}"

Ordena los siguientes documentos del más relevante al menos relevante.
Responde solo con los números en orden.

{chr(10).join(f'{i+1}. {doc[:200]}' for i, doc in enumerate(documents))}"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    
    # Parsear el orden
    order = [int(x) - 1 for x in response.choices[0].message.content.split(",")]
    return [documents[i] for i in order[:top_k]]

Evaluación de RAG

Resumen

RAG es el patrón más usado en LLMOps para dar a los LLMs acceso a datos actualizados y específicos. Su efectividad depende de una buena estrategia de chunking, embeddings de calidad, búsqueda híbrida y evaluación continua.