RAG desde cero con Python: entiende cómo funciona

TL;DR

Qué aprenderás: cómo funciona RAG por dentro — chunking, embeddings, similitud coseno, recuperación y cómo construir el prompt aumentado — implementando cada pieza en Python vanilla antes de tocar ningún framework.

Requisitos:

• Python 3.10+
• pip install anthropic sentence-transformers numpy
• Una API key de Anthropic (ANTHROPIC_API_KEY en tu entorno)

Tiempo estimado: 45-60 minutos leyendo y ejecutando el código.

---

¿Qué es RAG y por qué no basta con un prompt largo?

RAG (Retrieval-Augmented Generation) es una arquitectura que añade un paso de recuperación antes de llamar al LLM: en lugar de meter todo el conocimiento en el prompt, buscas solo los fragmentos relevantes y los inyectas en contexto.

La razón por la que no basta con un prompt largo es triple. Primero, los modelos degradan su atención en contextos muy largos — no procesan igual la información al principio que en el medio. Segundo, el coste de tokens crece linealmente con el tamaño del contexto: si tu base de conocimiento tiene 500 páginas, meter todo en cada llamada es inviable. Tercero, y más importante: el modelo no necesita todo — necesita las piezas correctas. RAG resuelve exactamente eso.

---

Paso 1: trocear el documento (chunking)

El chunking convierte un documento largo en fragmentos pequeños que tienen sentido por sí solos. El tamaño importa: muy pequeño pierde contexto, muy grande introduce ruido.

Un chunk debe contener una idea completa. La regla práctica es 200-500 palabras, con overlap entre chunks consecutivos para no perder frases que caen en el límite.

def chunk_text(text: str, chunk_size: int = 400, overlap: int = 50) -> list[str]:
    """
    Divide un texto en fragmentos con solapamiento.
    
    Args:
        text: Texto completo a trocear.
        chunk_size: Tamaño aproximado de cada chunk en caracteres.
        overlap: Caracteres de solapamiento entre chunks consecutivos.
    
    Returns:
        Lista de strings, uno por chunk.
    """
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        # Intentar cortar en un espacio para no romper palabras
        if end < len(text):
            last_space = chunk.rfind(" ")
            if last_space > 0:
                chunk = chunk[:last_space]
                end = start + last_space
        chunks.append(chunk.strip())
        start = end - overlap  # El overlap conecta chunks adyacentes
    return [c for c in chunks if c]  # Eliminar chunks vacíos


# Ejemplo de uso
documento = """
Los transformers son una arquitectura de red neuronal propuesta en 2017
en el paper 'Attention is All You Need'. Se basan en mecanismos de
atención que permiten al modelo relacionar cualquier par de posiciones
en la secuencia de entrada. Esto los hace especialmente eficientes
para procesar texto de forma paralela, a diferencia de las RNNs que
procesan secuencialmente.

La arquitectura encoder-decoder original se usa en tareas de
traducción. Los modelos solo-decoder, como GPT, generan texto
de forma autoregresiva. Los modelos solo-encoder, como BERT,
se usan para clasificación y comprensión de texto.
"""

chunks = chunk_text(documento, chunk_size=300, overlap=50)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk[:80]}...")

En producción usarás splitters más sofisticados (por párrafos, por frases, o semánticos), pero la lógica es la misma: divide y controla el overlap.

---

Paso 2: convertir chunks en embeddings

Un embedding es una representación vectorial de texto — un array de números donde la distancia entre vectores refleja similitud semántica. Dos frases con el mismo significado tienen vectores cercanos aunque usen palabras distintas.

sentence-transformers es la librería estándar para esto. El modelo all-MiniLM-L6-v2 es ligero (80MB), rápido y buen punto de partida para texto en inglés. Para español, paraphrase-multilingual-MiniLM-L12-v2 da mejores resultados (ver FAQ).

from sentence_transformers import SentenceTransformer
import numpy as np

# Cargar el modelo de embeddings (se descarga la primera vez, ~80MB)
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def embed_chunks(chunks: list[str]) -> np.ndarray:
    """
    Convierte una lista de textos en una matriz de embeddings.
    
    Args:
        chunks: Lista de strings a embeddear.
    
    Returns:
        Array de shape (n_chunks, embedding_dim). Para all-MiniLM-L6-v2,
        embedding_dim es 384.
    """
    embeddings = model.encode(chunks, convert_to_numpy=True)
    return embeddings


# Embeddear los chunks del paso anterior
chunk_embeddings = embed_chunks(chunks)

print(f"Shape de los embeddings: {chunk_embeddings.shape}")
# Output esperado: (n_chunks, 384)
print(f"Primer vector (primeros 5 dims): {chunk_embeddings[0][:5]}")

El método model.encode() acepta una lista de strings y devuelve un np.ndarray. El parámetro convert_to_numpy=True se incluye explícitamente para que el código sea auto-documentado; es el valor por defecto en versiones actuales de sentence-transformers.

---

Paso 3: almacenar embeddings (vector store mínimo)

Un vector store es una estructura que permite buscar vectores similares a una query de forma eficiente. En producción usarías Chroma, Qdrant o Pinecone. Para entender el mecanismo, un diccionario en memoria es suficiente.

from dataclasses import dataclass

@dataclass
class VectorStore:
    """Vector store mínimo en memoria."""
    chunks: list[str]
    embeddings: np.ndarray  # Shape: (n_chunks, embedding_dim)
    
    @classmethod
    def from_chunks(cls, chunks: list[str], embed_fn) -> "VectorStore":
        """Construye el store a partir de chunks y una función de embedding."""
        embeddings = embed_fn(chunks)
        return cls(chunks=chunks, embeddings=embeddings)


# Construir el store
store = VectorStore.from_chunks(chunks, embed_chunks)
print(f"Store con {len(store.chunks)} chunks indexados.")

En un sistema real, los embeddings se persisten en disco o en una base de datos vectorial para no recalcularlos en cada arranque. El coste de embeddings es el coste de indexación — solo lo pagas una vez.

---

Paso 4: recuperar los chunks más relevantes (similitud coseno)

La similitud coseno mide el ángulo entre dos vectores. Si el ángulo es 0°, los vectores apuntan en la misma dirección — máxima similitud. Si es 90°, no tienen relación. La fórmula: cos(θ) = (A · B) / (‖A‖ × ‖B‖).

Para recuperar los chunks más relevantes a una query: embeddeas la query, calculas su similitud con cada chunk, y te quedas con los K más altos.

def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """
    Calcula la similitud coseno entre dos vectores.
    
    Args:
        vec_a: Vector 1D.
        vec_b: Vector 1D del mismo tamaño que vec_a.
    
    Returns:
        Float entre -1 y 1. Más cercano a 1 = más similar.
    """
    dot_product = np.dot(vec_a, vec_b)
    norm_a = np.linalg.norm(vec_a)
    norm_b = np.linalg.norm(vec_b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)


def retrieve(
    query: str,
    store: VectorStore,
    embed_fn,
    top_k: int = 3
) -> list[tuple[str, float]]:
    """
    Recupera los top_k chunks más similares a la query.
    
    Args:
        query: Pregunta o texto de búsqueda.
        store: El VectorStore indexado.
        embed_fn: Función que convierte texto en embedding.
        top_k: Número de chunks a recuperar.
    
    Returns:
        Lista de tuplas (chunk_text, score) ordenada por relevancia desc.
    """
    query_embedding = embed_fn([query])[0]  # Shape: (embedding_dim,)
    
    scores = [
        cosine_similarity(query_embedding, chunk_emb)
        for chunk_emb in store.embeddings
    ]
    
    # Ordenar por score descendente y devolver los top_k
    ranked = sorted(
        zip(store.chunks, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return ranked[:top_k]


# Probar la recuperación
query = "¿Cuál es la diferencia entre encoder y decoder?"
resultados = retrieve(query, store, embed_chunks, top_k=2)

for chunk, score in resultados:
    print(f"Score: {score:.4f} | Chunk: {chunk[:100]}...")

---

Paso 5: construir el prompt aumentado y llamar al LLM

Con los chunks recuperados, construyes un prompt que incluye el contexto relevante y la pregunta del usuario. El modelo responde basándose en ese contexto, no en su conocimiento de entrenamiento.

import anthropic
import os

def build_prompt(query: str, context_chunks: list[tuple[str, float]]) -> str:
    """
    Construye el prompt aumentado con los chunks recuperados.
    
    Args:
        query: Pregunta original del usuario.
        context_chunks: Lista de (texto, score) de retrieve().
    
    Returns:
        String con el prompt completo para enviar al LLM.
    """
    context_text = "\n\n---\n\n".join(
        f"[Fragmento {i+1}]\n{chunk}"
        for i, (chunk, _) in enumerate(context_chunks)
    )
    
    return f"""Usa ÚNICAMENTE la siguiente información para responder la pregunta.
Si la respuesta no está en el contexto proporcionado, di "No tengo información sobre eso en los documentos."

CONTEXTO:
{context_text}

PREGUNTA: {query}

RESPUESTA:"""


def rag_query(
    query: str,
    store: VectorStore,
    embed_fn,
    top_k: int = 3
) -> str:
    """
    Pipeline RAG completo: recupera + genera.
    
    Args:
        query: Pregunta del usuario.
        store: VectorStore indexado.
        embed_fn: Función de embedding.
        top_k: Chunks a recuperar.
    
    Returns:
        Respuesta generada por el LLM.
    """
    # 1. Recuperar contexto relevante
    context_chunks = retrieve(query, store, embed_fn, top_k=top_k)
    
    # 2. Construir prompt aumentado
    prompt = build_prompt(query, context_chunks)
    
    # 3. Llamar al LLM
    client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
    
    return response.content[0].text


# Ejecutar el pipeline completo
respuesta = rag_query(
    query="¿Qué son los transformers y para qué sirven?",
    store=store,
    embed_fn=embed_chunks,
    top_k=2
)
print(respuesta)

---

El pipeline completo de un vistazo

Juntando todas las piezas, el flujo es lineal:

# rag_pipeline.py — sistema RAG mínimo completo
import os
import numpy as np
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
import anthropic


# --- Configuración ---
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
LLM_MODEL = "claude-sonnet-4-6"
CHUNK_SIZE = 400
OVERLAP = 50
TOP_K = 3

embed_model = SentenceTransformer(EMBED_MODEL)


# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = OVERLAP) -> list[str]:
    chunks, start = [], 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if end < len(text):
            last_space = chunk.rfind(" ")
            if last_space > 0:
                chunk = chunk[:last_space]
                end = start + last_space
        chunks.append(chunk.strip())
        start = end - overlap
    return [c for c in chunks if c]


# --- Embeddings ---
def embed(texts: list[str]) -> np.ndarray:
    return embed_model.encode(texts, convert_to_numpy=True)


# --- Vector Store ---
@dataclass
class VectorStore:
    chunks: list[str]
    embeddings: np.ndarray

    @classmethod
    def build(cls, text: str) -> "VectorStore":
        chunks = chunk_text(text)
        return cls(chunks=chunks, embeddings=embed(chunks))


# --- Recuperación ---
def retrieve(query: str, store: VectorStore, top_k: int = TOP_K) -> list[str]:
    q_emb = embed([query])[0]
    scores = [
        np.dot(q_emb, emb) / (np.linalg.norm(q_emb) * np.linalg.norm(emb) + 1e-9)
        for emb in store.embeddings
    ]
    ranked = sorted(zip(store.chunks, scores), key=lambda x: x[1], reverse=True)
    return [chunk for chunk, _ in ranked[:top_k]]


# --- Generación ---
def generate(query: str, context: list[str]) -> str:
    ctx = "\n\n---\n\n".join(context)
    prompt = f"Contexto:\n{ctx}\n\nPregunta: {query}\n\nRespuesta:"
    client = anthropic.Anthropic()  # Usa ANTHROPIC_API_KEY del entorno
    response = client.messages.create(
        model=LLM_MODEL,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text


# --- Pipeline ---
def rag(documento: str, pregunta: str) -> str:
    store = VectorStore.build(documento)
    context = retrieve(pregunta, store)
    return generate(pregunta, context)


# --- Ejecución ---
if __name__ == "__main__":
    doc = open("mi_documento.txt").read()
    respuesta = rag(doc, "¿Cuál es la idea principal del documento?")
    print(respuesta)

---

De aquí a frameworks reales: qué cambia (y qué no)

Una vez entiendes las piezas, los frameworks solo son capas de abstracción sobre esto mismo.

LangChain envuelve el chunking, los embeddings y el retrieval en cadenas configurables. Tiene más opciones (splitters semánticos, BM25 híbrido, rerankers), pero el flujo es idéntico al que acabas de implementar.

LlamaIndex pone el foco en la indexación de documentos estructurados (PDFs, tablas, árboles de nodos). Útil cuando tu base de conocimiento no es texto plano.

Chroma / Qdrant / Pinecone reemplazan el dict en memoria por un vector store persistente con índices HNSW para búsqueda aproximada eficiente a millones de vectores.

Lo que no cambia: los conceptos. Chunk → Embed → Almacenar → Recuperar → Augmentar → Generar. Siempre.

Para profundizar en cómo los LLMs procesan estos vectores internamente, el post sobre razonar en espacio latente explica qué ocurre dentro del transformer cuando le das ese contexto. Y si quieres ver cómo RAG encaja en agentes más complejos, cómo crear agentes de IA gratis muestra el panorama completo de frameworks disponibles hoy.

---

FAQ

¿Qué tamaño de chunk es mejor? Depende del contenido. Para texto narrativo, 300-500 caracteres con 50-100 de overlap. Para documentación técnica con definiciones densas, chunks más pequeños (150-200). Lo mejor es experimentar midiendo la calidad de recuperación con preguntas de prueba que ya conoces la respuesta.

¿Por qué similitud coseno y no distancia euclídea? La distancia euclídea es sensible a la magnitud del vector. Dos textos con el mismo significado pero diferente longitud pueden tener vectores de distinta magnitud y salir “lejanos” en euclidiana. El coseno mide el ángulo, no la distancia, así que es robusto a la longitud del texto.

¿Cuántos chunks recuperar (top_k)? Empieza con 3-5. Más chunks = más contexto = más coste de tokens y más riesgo de “attention dilution”. Si el modelo ignora información relevante, baja el k. Si responde con “no tengo información” cuando debería saberlo, súbelo.

¿Puedo usar un modelo de embeddings diferente? Sí. all-MiniLM-L6-v2 es el punto de partida estándar. Para español puro, paraphrase-multilingual-MiniLM-L12-v2 da mejores resultados. Para máxima calidad, BAAI/bge-m3 es el estado del arte multilingüe en 2026.

¿Qué pasa si el chunk relevante no llega al top_k? Eso es un problema de recall. Las causas más comunes: chunks demasiado pequeños (pierden contexto), modelo de embeddings inadecuado para el dominio, o texto muy técnico con vocabulario específico. La solución estándar es el reranking: recuperas más candidatos (top_20) y luego los reordenas con un cross-encoder más preciso antes de quedarte con los mejores.

¿Cómo sé si mi RAG funciona bien? Construye un set de preguntas de evaluación con respuestas esperadas conocidas. Mide: precision@k (¿los chunks recuperados son relevantes?), recall@k (¿el chunk con la respuesta está entre los recuperados?) y la calidad de la respuesta final. Ragas es el framework más usado para esto.

¿RAG vs fine-tuning: cuándo usar cada uno? RAG cuando el conocimiento cambia con frecuencia o es voluminoso. Fine-tuning cuando necesitas cambiar el estilo de respuesta del modelo o inculcar conocimiento que se usa en prácticamente todas las consultas. En la mayoría de casos de negocio, RAG es más rápido, más barato y más mantenible.

¿Este código funciona con otros LLMs? El chunking, embeddings y recuperación son agnósticos al LLM. Solo tienes que cambiar la llamada en generate() por el SDK del modelo que uses — OpenAI, Mistral, un modelo local con Ollama. La arquitectura no cambia.

---

Para ver cómo MCP (Model Context Protocol) extiende este patrón hacia herramientas externas en tiempo real, el post sobre MCP servers cubre el salto de RAG estático a contexto dinámico.