Un pipeline Mini RAG en 90 líneas de Python: recuperar, mejorar y generar.
Construir una generación mejorada por recuperación desde cero utilizando embeddings, similitud coseno y una lista ordenada, para luego observar dónde falla la versión simplificada a gran escala.
La generación mejorada mediante recuperación de información no es una característica de ningún modelo. Se trata de un problema de empaquetamiento: la ventana de contexto es pequeña, los documentos son enormes, y algo tiene que decidir qué partes se incluyen. Ese algo puede implementarse en unas 90 líneas de Python puro, sin LangChain ni servicio de base de datos vectorial. Tiene tres pasos:
- Recuperar. Incorporar la pregunta, clasificar los fragmentos por similitud y conservar los k mejores.
- Mejorar. Unir esos fragmentos seleccionados en el mensaje del sistema.
- Generar. Realizar la misma llamada de chat que se utiliza en cualquier script básico de LLM.
Toda abstracción que ofrece un framework como LangChain se relaciona con uno de estos tres pasos, por lo que ver primero la versión básica facilita la comprensión del framework.
Los embeddings convierten el significado en geometría
# pip install openai
# export OPENAI_API_KEY=sk-...
import math
from openai import OpenAI
client = OpenAI()
def embed(texts):
resp = client.embeddings.create(
model="text-embedding-3-small",
input=texts,
)
return [d.embedding for d in resp.data]
def cosine(a, b):
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
return dot / (na * nb)
q, hit, miss = embed([
"How do I get my money back?",
"Refunds are possible within 14 days "
"of purchase.",
"Data is stored in Frankfurt and "
"encrypted at rest.",
])
print(f"refund doc: {cosine(q, hit):.3f}")
print(f"storage doc: {cosine(q, miss):.3f}")
La salida típica de ese script es una puntuación de 0.531 para el documento de reembolso y 0.126 para el documento de almacenamiento. Observe cómo ocurre la coincidencia: la pregunta dice “devolución de dinero” mientras que el documento indica “reembolsos”. No hay palabras clave en común, pero la similitud es alta. Por eso se utiliza incrustación en lugar de búsqueda por patrones.
Un almacén vectorial es una lista que se puede ordenar
“Almacén vectorial” suena como una infraestructura. A pequeña escala, se trata de una lista de pares (vector, texto) junto con un mecanismo de ordenamiento. Pinecone, Chroma y pgvector funcionan bien con millones de vectores, pero lo importante es la interfaz, y esa se enmarca en una sola categoría:
class VectorStore:
def __init__(self):
self.texts = []
self.vectors = []
def add(self, texts):
self.texts.extend(texts)
self.vectors.extend(embed(texts))
def top_k(self, query, k=3):
q = embed([query])[0]
scored = [
(cosine(q, v), t)
for v, t in zip(
self.vectors, self.texts
)
]
scored.sort(
key=lambda s: s[0], reverse=True
)
return scored[:k]
store = VectorStore()
store.add([
"Refunds are possible within 14 days.",
"The Pro plan costs 12 EUR per month.",
"Data is stored in Frankfurt.",
])
for score, text in store.top_k(
"can I cancel and get a refund?", k=2
):
print(f"{score:.3f} {text}")
top_k representa la decisión sobre qué elementos se incluyen, y ahora es ejecutable: se clasifica todo, se conservan k elementos y el resto se descarta. El valor de k funciona como un control de ajuste que permite equilibrar la capacidad de recuperación con el espacio del ventanal de contexto y los tokens utilizados.
Combinar la recuperación, el enriquecimiento y la generación
Solo hay una función nueva aquí, build_messages. Esta función numera los fragmentos seleccionados, los incluye en el mensaje del sistema y le indica al modelo que responda únicamente a partir de ellos. El mensaje del usuario contiene solo la pregunta. La lista de mensajes es la misma que se utiliza en cualquier llamada de chat, por lo que RAG solo modifica su contenido. Este archivo se ejecuta tal como está:
# examples/part07_rag.py
# Mini RAG in ~90 lines: embed -> store -> top-k
# -> stitch into messages -> generate.
#
# Setup:
# pip install openai
# export OPENAI_API_KEY=sk-...
#
# Run:
# python part07_rag.py
import math
from openai import OpenAI
client = OpenAI()
EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
# A tiny corpus. In real life these are chunks
# of your docs, not whole documents.
DOCS = [
"Support hours are 9am to 6pm CET, "
"Monday to Friday.",
"Refunds are possible within 14 days "
"of purchase, no questions asked.",
"The Pro plan costs 12 EUR per month "
"and includes API access.",
"The Free plan allows 3 projects and "
"community support only.",
"Data is stored in Frankfurt and "
"encrypted at rest with AES-256.",
"You can export all your data as "
"JSON from the settings page.",
]
def embed(texts):
resp = client.embeddings.create(
model=EMBED_MODEL,
input=texts,
)
return [d.embedding for d in resp.data]
def cosine(a, b):
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
return dot / (na * nb)
class VectorStore:
"""A list of (vector, text) pairs. That is
all a vector store is, before you need
scale."""
def __init__(self):
self.texts = []
self.vectors = []
def add(self, texts):
self.texts.extend(texts)
self.vectors.extend(embed(texts))
def top_k(self, query, k=3):
q = embed([query])[0]
scored = [
(cosine(q, v), t)
for v, t in zip(
self.vectors, self.texts
)
]
scored.sort(key=lambda s: s[0], reverse=True)
return scored[:k]
def build_messages(question, hits):
context = "\n\n".join(
f"[{i + 1}] {text}"
for i, (_, text) in enumerate(hits)
)
system = (
"Answer using only the context below. "
"If the answer is not in the context, "
"say you do not know.\n\n"
"Context:\n" + context
)
return [
{"role": "system", "content": system},
{"role": "user", "content": question},
]
def ask(store, question):
hits = store.top_k(question, k=3)
print(f"Q: {question}")
for score, text in hits:
print(f" {score:.3f} {text[:44]}...")
messages = build_messages(question, hits)
resp = client.chat.completions.create(
model=CHAT_MODEL,
messages=messages,
)
return resp.choices[0].message.content
if __name__ == "__main__":
store = VectorStore()
store.add(DOCS)
answer = ask(
store,
"Where is my data stored, and can "
"I get it out?",
)
print(f"A: {answer}")
Una prueba de ejemplo pregunta dónde se almacenan los datos del usuario y si se pueden exportar. Las tres mejores puntuaciones son 0.588 para un fragmento sobre el almacenamiento en Fráncfort con encriptación, 0.472 para un fragmento sobre la exportación de todos los datos en formato JSON y 0.244 para la política de reembolsos. El modelo responde que los datos se encuentran en Fráncfort, encriptados en reposo con AES-256, y que la página de configuraciones permite una exportación completa en formato JSON.
Tanto el fragmento relacionado con el almacenamiento como el relacionado con la exportación lograron incluirse. El fragmento sobre reembolsos quedó en tercer lugar y el modelo lo ignoró. La recuperación de información puede ser imperfecta siempre y cuando el material adecuado quede entre los primeros resultados.
Dónde falla la versión simplificada
Estas 90 líneas tienen límites reales, y cada una señala algo que los sistemas reales añaden:
- Fragmentación. La demostración almacena oraciones completas. Los documentos reales necesitan ser divididos con solapamiento, y esa es una decisión de preprocesamiento, no del modelo.
- Volver a incrustar. Es aceptable incrustar la consulta en cada llamada. No lo es volver a incrustar todo el corpus en cada reinicio, por lo que se deben conservar los vectores.
- Búsqueda lineal. La similitud coseno con cada vector es O(n). Después de aproximadamente 100 mil fragmentos, un índice ANN resulta útil, y ahí es donde las bases de datos vectoriales demuestran su valor.
- Calidad del ranking. El top-k simple basado en coseno es una herramienta rudimentaria; los sistemas de re-ranking o las búsquedas híbridas por palabras clave la hacen más precisa.
LangChain encapsula cada uno de estos elementos como un componente intercambiable. Una vez que el bucle básico se ha escrito a mano, esos componentes muestran claramente su función real: son simples conveniencias y no algo místico.