Inicio / Artículos

ragpythontf-idfinformation-retrievalllm

RAG en Python sencillo: fragmentación, TF-IDF, similitud cosénica y generación de prompts

Construir generación mejorada por recuperación utilizando únicamente el módulo matemático: particionamiento en ventanas deslizantes, vectores TF-IDF, clasificación por coseno, un prompt fundamentado y sus limitaciones.

Editor: REACTAPP.TOP10 min de lectura2359 palabras

Fuentes: Building RAG From First Principles with just plain python .

Un tutorial típico sobre RAG te hace instalar cinco bibliotecas y llamar a .from_documents(). Diez minutos después la demostración funciona, pero no se entiende nada al respecto. Cuando la recuperación de información devuelve datos inútiles o el modelo genera respuestas erróneas, te quedas atascado, porque la biblioteca hizo todo el trabajo por ti.

Aquí el flujo se construye de manera opuesta, en Python puro, utilizando únicamente el módulo math y una sola llamada a un LLM de texto plano al final. Sin modelos de embedding, sin FAISS, sin Pinecone. Conocer los mecanismos implica que, cuando más adelante utilices LangChain o una base de datos vectorial real, entenderás qué ocurre en el fondo. Al final se obtiene un flujo funcional y, lo que es más útil, una comprensión de cada número que genera.

El problema que resuelve RAG

El conocimiento de un modelo de lenguaje se congela al final del entrenamiento y reside exclusivamente en sus pesos. Para que responda preguntas sobre documentos internos de la empresa, un código fuente o las noticias de esta mañana, las dos opciones obvias decepcionan tanto a la vez:

  1. Reentrenar el modelo. Es lento y costoso, además de que los cambios del día siguiente requieren otra ronda de entrenamiento.
  2. Pegar todo en el prompt. Este método no es escalable. Incluso una ventana de un millón de tokens no puede contener 50,000 documentos, y las investigaciones sobre modelos de contexto largo señalan repetidamente un efecto de “perderse en la mitad”, donde la información enterrada profundamente en un prompt enorme se utiliza peor aunque quepa. Además, más tokens significan mayor latencia y costo por llamada.

El verdadero truco de RAG es más sencillo de lo que parece. Los documentos permanecen en almacenamiento externo, y solo los fragmentos relevantes para esa pregunta se recuperan y se entregan al modelo. El LLM nunca contiene tus datos; recibe siempre un fragmento nuevo y relevante.

Por lo tanto, cualquier sistema RAG está formado por dos sistemas combinados: la recuperación, que encuentra el texto adecuado en una colección mucho más grande para una consulta determinada, y la generación, que pasa ese texto junto con la consulta a un LLM para obtener una respuesta. El resto de esta explicación describe cómo se construyen ambas partes a partir de elementos básicos.

Paso 1: fragmentación

Convertir un PDF completo de 50 páginas en un único vector elimina casi toda la información útil, ya que los detalles necesarios para responder a una pregunta se mezclan y pierden su significado. La primera tarea es dividir los documentos en fragmentos pequeños y específicos.

El método más sencillo es una ventana deslizante de tamaño fijo contada en palabras, con superposición entre los bloques adyacentes:

def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")

    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap  # step back so chunks share context
    return chunks

Esa cláusula de protección no es solo decorativa. Si overlap alcanza o supera chunk_size, la posición de inicio deja de avanzar y el bucle nunca termina. Ajustar overlap hacia arriba “para tener más contexto” hace que esto sea un error fácil de cometer, por lo que la función se niega a ejecutarse en lugar de quedarse bloqueada en silencio.

El valor de overlap es más importante de lo que parece. Sin él, una oración que responde a una pregunta podría cruzar un límite entre bloques y dividirse, de modo que ninguno de los bloques contenga una idea completa que pueda ser recuperada. Con overlap, la región del límite permanece intacta en al menos un bloque.

Un caso límite importante: las ventanas de palabras fijas ignoran la estructura de oraciones y párrafos, por lo que pueden dividir una oración o una fila de tabla por la mitad. La división que tiene en cuenta las oraciones o los párrafos evita eso, pero genera tamaños de fragmentos desiguales, ya que una oración de 10 palabras y otra de 150 palabras se convierten en un mismo fragmento. Esto distorsiona el presupuesto de tokens y la influencia que tiene cada fragmento en los cálculos de recuperación posteriores. Ningún método de fragmentación está exento de compromisos, y la elección depende realmente de qué tipo de problema preferirías depurar.

Paso 2: convertir texto en números con TF-IDF

Una computadora no puede comparar directamente la similitud entre dos cadenas de texto. Cada paso posterior necesita vectores, listas de números sobre los cuales se pueden realizar operaciones aritméticas. La forma clásica y libre de dependencias para crearlos es TF-IDF, que es la frecuencia de término multiplicada por la frecuencia inversa de documento.

  • TF indica con qué frecuencia aparece una palabra en este fragmento específico. Un valor alto para “kidney” sugiere que el fragmento trata sobre los riñones.
  • IDF mide qué tan rara es una palabra en toda la colección. Palabras como “the” y “is” aparecen por todas partes y no proporcionan información sobre el fragmento que se está leyendo, por lo que obtienen un puntaje cercano a cero. Una palabra rara en un fragmento es una señal mucho más fuerte.

El código crea un vocabulario a partir de todos los fragmentos, calcula el IDF para cada término y luego convierte cada fragmento en un vector TF-IDF:

import math
from collections import Counter

def build_vocabulary(chunks: list[str]) -> list[str]:
    vocab = set()
    for chunk in chunks:
        vocab.update(chunk.lower().split())
    return sorted(vocab)

def compute_idf(chunks: list[str], vocab: list[str]) -> dict[str, float]:
    num_chunks = len(chunks)
    idf = {}
    for word in vocab:
        containing = sum(1 for c in chunks if word in c.lower().split())
        idf[word] = math.log(num_chunks / (1 + containing))
    return idf

def compute_tf(chunk: str, vocab: list[str]) -> dict[str, float]:
    words = chunk.lower().split()
    counts = Counter(words)
    total = len(words)
    return {word: counts.get(word, 0) / total for word in vocab}

def vectorize(chunk: str, vocab: list[str], idf: dict[str, float]) -> list[float]:
    tf = compute_tf(chunk, vocab)
    return [tf[word] * idf[word] for word in vocab]

Pruébalo con tres fragmentos y algo se hará claro. Una palabra presente en casi todos los fragmentos, como “the”, obtiene un IDF cercano a cero, por lo que repetirla no aporta nada al vector. Una palabra rara y específica del tema recibe un IDF claramente positivo. TF-IDF es, literalmente, una fórmula para determinar si una palabra ofrece información distintiva sobre un fragmento.

La desventaja es la escala. Con 100,000 documentos, el vocabulario puede contener más de 500,000 palabras únicas, por lo que cada vector de fragmento, incluso el de tres palabras, tiene 500,000 números y está compuesto en su mayor parte por ceros. Estos vectores dispersos son costosos de almacenar y comparar. Esa es la razón principal por la cual los sistemas en producción pasan a usar embeddings neuronales densos de 384 a 1536 dimensiones con casi ningún cero.

Paso 3: medir la relevancia con similitud coseno

Al considerar los fragmentos como vectores, la siguiente pregunta es cuán relevante es un fragmento para una consulta. La distancia bruta no es la herramienta adecuada, ya que en el caso del texto el ángulo entre los vectores es más importante que su longitud. Un fragmento que menciona “gato” una vez y otro que lo repite veinte veces siguen tratándose del mismo tema, y la similitud coseno los trata de manera similar sin importar cuál es más frecuente.

La implementación necesita un producto escalar, una magnitud y la relación entre ellos:

def dot_product(a, b):
    return sum(x * y for x, y in zip(a, b))

def magnitude(v):
    return math.sqrt(sum(x * x for x in v))

def cosine_similarity(a, b):
    denom = magnitude(a) * magnitude(b)
    return 0.0 if denom == 0 else dot_product(a, b) / denom

En principio, la similitud coseno varía de -1 a 1, y con esta configuración de TF-IDF casi siempre se mantiene en el rango de 0 a 1:

  • 1.0 significa que los vectores apuntan exactamente en la misma dirección, lo que representa la máxima similitud.
  • 0.0 significa que son ortogonales y no comparten ninguna dimensión significativa; en términos sencillos, esto indica que la consulta y el fragmento no tienen palabras en común que no sean triviales.

Una puntuación de 0 no es un error. Significa que el sistema informa correctamente de que el fragmento no tiene relación alguna con la consulta.

Un punto sobre la precisión: la fórmula de IDF suavizada puede asignar un valor ligeramente negativo a una palabra que aparece en todos los fragmentos, por lo que la similitud coseno ocasionalmente puede descender un poco por debajo de cero. En textos reales esto apenas tiene importancia, pero afirmar que nunca es negativa sería exagerado.

El punto ciego del TF-IDF con similitud coseno son los sinónimos. Una consulta que dice “automobile” frente a un documento que dice “car” obtiene una puntuación cercana a cero, ya que solo se cuenta la superposición literal del vocabulario y no existe concepto de significado. Los modelos de embedding denso, que colocan a los sinónimos cerca unos de otros, existen precisamente para cerrar esa brecha.

Paso 4: el recuperador

Una vez que la vectorización y la similitud están listas, la recuperación implica vectorizar cada fragmento una vez, vectorizar la consulta entrante, calcular la puntuación de cada fragmento en relación con ella, ordenarlos y conservar los k mejores:

class VectorStore:
    def __init__(self):
        self.chunks, self.vocab, self.idf, self.vectors = [], [], {}, []

    def index(self, documents, chunk_size=50, overlap=10):
        all_chunks = []
        for doc in documents:
            all_chunks.extend(chunk_text(doc, chunk_size, overlap))
        self.chunks = all_chunks
        self.vocab = build_vocabulary(all_chunks)
        self.idf = compute_idf(all_chunks, self.vocab)
        self.vectors = [vectorize(c, self.vocab, self.idf) for c in all_chunks]

    def search(self, query, top_k=3):
        query_vector = vectorize(query, self.vocab, self.idf)
        scored = [(c, cosine_similarity(query_vector, v))
                  for c, v in zip(self.chunks, self.vectors)]
        scored.sort(key=lambda pair: pair[1], reverse=True)
        return scored[:top_k]

Un detalle puede arruinar todo en silencio. La consulta debe ser convertida utilizando el vocabulario y los valores IDF calculados a partir de los documentos, y utilizar un vocabulario derivado únicamente de la consulta en sí es un error. La similitud coseno presupone vectores de igual longitud, donde la posición i corresponde a la misma palabra en cada uno. Si se le da a la consulta su propio vocabulario, la posición 5 podría ser “heart” en un vector y “and” en otro, lo que hace que la puntuación de similitud carezca de sentido estructural, y no solo esté ligeramente desviada.

Paso 5: mejorar el prompt

Este paso no implica cálculos matemáticos, solo disciplina. Combina los fragmentos recuperados con la pregunta del usuario en un prompt que indica al modelo exactamente cómo comportarse:

def build_augmented_prompt(query: str, retrieved_chunks: list[str]) -> str:
    context = "\n\n".join(
        f"[Source {i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)
    )
    return f"""Answer the question using ONLY the context below.
If the answer is not contained in the context, say "I don't know based on the provided context."
Cite which source number(s) you used.

Context:
{context}

Question: {query}

Answer:"""

Dos detalles aquí no son meramente estéticos.

“Usar SOLO el contexto a continuación.” Sin esa regla explícita, el modelo mezcla los fragmentos obtenidos con lo que recuerda de su entrenamiento, y ya no se puede distinguir qué parte de una respuesta proviene de documentos fiables y cuál es una suposición. Al presentarse con confianza, tal mezcla causa más daño que omitir por completo el RAG, ya que parece tener una fuente cuando en realidad no la tiene.

Etiquetas numeradas de las fuentes e instrucciones de citación. Estas permiten rastrear las respuestas. Cuando una respuesta es incorrecta, lo cual ocurrirá, las fuentes etiquetadas muestran qué fragmento engañó al modelo, si el propio fragmento estaba desactualizado o incorrecto, o si el modelo interpretó mal uno correcto. Sin etiquetas, localizar la causa de una mala respuesta es en gran medida una conjetura.

Paso 6: integrarlo todo

El flujo completo consiste primero en la recuperación, luego en el aumento de datos y finalmente en una llamada al LLM. La decisión clave de diseño es mantener el indexado y las consultas como dos ciclos de vida separados, en lugar de que una sola función realice ambas tareas:

class RagPipeline:
    def __init__(self, client, model="openai/gpt-oss-20b"):
        self.store = VectorStore()
        self.client = client
        self.model = model

    def build_index(self, documents: list[str]):
        """Run once, whenever documents are loaded or updated."""
        self.store.index(documents, chunk_size=50, overlap=10)

    def answer(self, query: str, top_k: int = 3) -> str:
        """Run on every user query — never re-indexes."""
        results = self.store.search(query, top_k=top_k)
        chunks = [c for c, _ in results]
        prompt = build_augmented_prompt(query, chunks)

        response = self.client.chat.completions.create(
            model=self.model,
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

Si la indexación se realizara dentro de la función por consulta, cada consulta dividiría nuevamente el texto, reconstruiría el vocabulario y volvería a calcular todos los vectores TF-IDF para toda la colección, aunque nada de eso cambia entre consultas. Con 10,000 documentos y 1,000 consultas, eso supone 999 reconstrucciones idénticas innecesarias. Construir el índice una sola vez y luego realizar muchas consultas es el enfoque que recomiendan FAISS, Chroma y Pinecone; se trata de la arquitectura correcta desde el principio, no de una optimización prematura.

¿Qué tiene de malo este sistema?

Las limitaciones merecen ser abordadas con sinceridad, ya que no son errores. Surgen directamente de las decisiones de diseño mencionadas anteriormente.

  • Sin comprensión semántica. “Coche” y “automóvil” nunca coinciden, ya que TF-IDF solo considera la superposición literal del vocabulario.
  • Vectores escasos y muy amplios. El tamaño del vocabulario sigue al corpus, por lo que cada vector se ensancha junto con él. Más allá de unos pocos miles de documentos, esto se vuelve inviable.
  • Nada en el código verifica que la recuperación haya encontrado algo útil. Incluso cuando cada fragmento de los top-k obtiene una puntuación cercana a cero, el proceso aún los envía al modelo. La solicitud del prompt de admitir la ignorancia depende por completo de que el LLM coopere, por lo que una recuperación deficiente puede seguir pasando sin ser revisada a la fase de generación.

    • Lecturas relacionadas

    ← Todos los artículosEnviar el brief