Construye una tubería RAG local en Python sin costos de API
Un tutorial práctico para crear generación mejorada por recuperación en CPU utilizando sentence-transformers, numpy y Flan-T5, que incluye la división en fragmentos, la búsqueda y una prueba A/B.
Fuentes: Build a RAG System From Scratch — Hands-On, No API Costs
Se puede crear y ejecutar un pipeline completo de generación ampliada por recuperación de información únicamente en una computadora personal, utilizando solo la CPU, modelos de código abierto gratuitos, sin clave de OpenAI, sin tarjeta de crédito y sin límites de uso. Lo importante es comprender cada componente en lugar de copiar un sistema cerrado.
La mejor forma de aprender es de manera práctica: mantenga abierto un terminal y un archivo de Python llamado rag.py, ejecute cada bloque y prediga la salida antes de ejecutarlo. Después de la mayoría de los pasos aparecen ejercicios breves, y es allí donde ocurre la mayor parte del aprendizaje.
El problema que aborda RAG
Un modelo de lenguaje solo conoce lo que contenían sus datos de entrenamiento. Las notas privadas, una wiki corporativa, la reunión de la semana pasada o cualquier documento no visto le resultan invisibles. Al preguntársele sobre ellos, o bien admite su ignorancia o, peor aún, inventa algo con total confianza.
RAG divide la tarea en dos partes:
- Recuperar: encontrar los fragmentos de tus propios documentos que están relacionados con la pregunta. Eso es un problema de búsqueda.
- Generar: proporcionar esos fragmentos al modelo con la instrucción de responder a partir de ellos.
El modelo deja de adivinar y comienza a leer. Todo lo demás son mecanismos para llevar a cabo bien esas dos etapas.
question ──► [embed] ──► [search your docs] ──► top chunks ──┐
▼
[LLM: "answer using this context"] ──► answer
La herramienta gratuita
Tres componentes abarcan todo el sistema. sentence-transformers convierte el texto en vectores, utilizando un modelo pequeño que funciona en la CPU y se descarga una sola vez. numpy se encarga de los cálculos de similitud, y toda la “base de datos de vectores” cabe en unas cinco líneas, lo que permite ver cómo funciona. transformers con Flan-T5 escribe la respuesta, ya que está ajustado para seguir instrucciones, funciona en la CPU y no necesita ninguna API.
Instale todo una sola vez:
pip install sentence-transformers transformers torch numpy
Al cargarse por primera vez, cada modelo obtiene sus pesos de Hugging Face. El módulo de incrustación ocupa aproximadamente 80 MB y flan-t5-base unos 1 GB. Después de eso, se almacenan en caché y permiten trabajar sin conexión. Cuando el espacio en disco o la RAM es limitado, flan-t5-small (alrededor de 300 MB) puede reemplazar al modelo base en todo momento, aunque con menor calidad pero mucha mayor velocidad.
Paso 1: un corpus que el modelo no conoce
Es más fácil comprender RAG con documentos que ningún modelo podría haber visto, por lo que el script comienza con una pequeña wiki ficticia de una empresa. Cree rag.py y pegue en él los documentos:
# rag.py
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
La wiki continúa con más entradas:
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""", """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""", """The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""", """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""", """Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]
Antes de construir cualquier cosa, demuestre que el problema existe. En una shell de Python, pregunte a un modelo simple sobre Nimbus:
from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])
Producirá algo, y ese algo estará incorrecto o vacío porque es solo una suposición. Tenga en cuenta esa respuesta, ya que el resto del tutorial la corrige.
Paso 2: fragmentación
Los documentos completos no se envían al modelo. Se dividen en fragmentos más pequeños por dos razones: la recuperación de información se vuelve más precisa, ya que el modelo encuentra el párrafo que responde a la pregunta en lugar de todo un manual, y la ventana de contexto del modelo es limitada.
Agregue una función de fragmentación:
def chunk_text(text, chunk_size=60, overlap=15):
"""Split text into overlapping chunks of `chunk_size` words."""
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 by `overlap` so context isn't cut
return chunks
Luego, cree la lista de fragmentos mientras registra de qué documento proviene cada uno:
# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
print("-", c["text"][:70], "...")
La superposición existe porque un hecho puede abarcar más de un límite, como una política de reembolso definida entre “dentro de 5” y “días hábiles”. Sin superposición, ninguna parte contiene todo el hecho. Repetir algunas palabras en cada punto de división permite que los hechos sobrevivan a dicha división.
Ejercicio: ejecute nuevamente con chunk_size=20 para obtener fragmentos más pequeños, y luego pruebe con chunk_size=500, donde cada uno de estos documentos cortos se convierte en un único fragmento. Antes de ejecutar, intente adivinar qué configuración proporciona una precisión mayor pero corre el riesgo de perder el contexto circundante. Los fragmentos pequeños son precisos pero carecen de contexto, y esa tensión es el núcleo del proceso de fragmentación. No existe un tamaño universalmente correcto.
Paso 3: embeddings
Un embedding es un vector de números que codifica el significado de un texto. Las frases con significados similares terminan siendo vectores que apuntan en la misma dirección, incluso sin palabras en común; por eso “¿Cuánto cuesta?” aparece junto a “¿Cuál es el precio?”
Importe el modelo de embedding:
from sentence_transformers import SentenceTransformer
Luego incrusta cada fragmento:
embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape) # (num_chunks, 384)
Cada bloque ahora corresponde a un vector de 384 dimensiones, donde 384 es simplemente la cantidad de números que este modelo específico utiliza para expresar significado.
Ejercicio: observe cómo funciona el significado con un pequeño experimento:
import numpy as np
pairs = embedder.encode(
["the price of the pro plan", "how much does it cost", "the weather in Paris"],
normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3)) # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW
La expresión a @ b representa un producto escalar, que para vectores unitarios equivale a la similitud por coseno. Un valor de 1.0 indica la misma dirección, mientras que 0 significa que no hay relación entre ellos. La primera pareja debería obtener una puntuación muy superior a la segunda, aun cuando las frases no compartan palabras; eso es lo que se conoce como búsqueda semántica. Intente adivinar ambos valores dentro de un margen de 0.2 antes de ejecutar el proceso.
Paso 4: recuperación
En la fase de búsqueda, se debe incrustar la pregunta, compararla con cada vector de fragmento y conservar los mejores resultados. Dado que los vectores están normalizados, la “comparación” se reduce a una multiplicación matricial.
import numpy as np
La función de recuperación:
def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec # cosine similarity to every chunk
top_idx = np.argsort(scores)[::-1][:k] # indices of the k highest scores
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
print(f"[{score:.3f}] {text[:80]}...")
El bloque de precios debería aparecer en la parte superior con la mejor puntuación, aunque la pregunta mencionaba “costo” mientras que el documento decía “8 dólares al mes”. Se trata de una coincidencia de significado y no de palabras clave.
Ejercicio: pruebe con tres preguntas y vea qué resultados aparecen:
retrieve("¿Puedo recuperar mi dinero?")debería devolver el bloque relacionado con los reembolsos, aunque la palabra “reembolso” nunca apareció en la pregunta.retrieve("¿Están seguros mis datos?")debería mostrar el bloque de encriptación.retrieve("¿Todavía funciona sin conexión a Internet?")se espera que muestre el bloque de sincronización sin conexión.
Muchas personas subestiman esta etapa. Cuando la recuperación devuelve fragmentos incorrectos, ningún tipo de instrucción inteligente puede salvar la respuesta. Lo que se ingresa es lo que se obtiene, por lo que los esfuerzos de depuración deben centrarse primero aquí.
Paso 5: generación
Ahora las dos partes se unen. Se debe crear una instrucción que envíe los fragmentos recuperados a Flan-T5 y le indique que responda únicamente con base en ellos.
from transformers import pipeline
La función de generación:
generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved) prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:""" out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
print(f" [{score:.3f}] {text[:70]}...")
Compare el resultado con la respuesta incorrecta o vacía del Paso 1. El modelo y la pregunta son idénticos, y el único cambio es que se proporcionó el texto relevante para su lectura. Esa diferencia constituye todo el valor de RAG.
Ejercicio: formule una pregunta que los documentos no aborden, por ejemplo sobre un cliente de escritorio para Linux de Nimbus, y llame a rag_answer con ella. Una configuración adecuadamente fundamentada reconocerá que no lo sabe. Si la suya inventa una respuesta, se trata de un verdadero modo de fallo denominado alucinación a pesar de la recuperación de información, y una sección posterior trata este tema.
El script completo
Aquí está todo el rag.py como un único programa ejecutable con un bucle interactivo. Comienza con las importaciones:
# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
A continuación se muestra el cuerpo completo:
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""",
"""Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""",
"""The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""",
"""Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""",
"""Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
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
return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
[c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec
top_idx = np.argsort(scores)[::-1][:k]
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved)
prompt = (
"Answer the question using only the context below. "
"If the answer is not in the context, say you don't know.\n\n"
f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
)
out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip()if __name__ == "__main__":
print("RAG ready. Ask about Nimbus (or type 'quit').\n")
while True:
q = input("You: ").strip()
if q.lower() in {"quit", "exit", ""}:
break
print("Nimbus bot:", rag_answer(q), "\n")
Comience con:
python rag.py
Luego pruebe preguntas como cuál es el plazo para el reembolso, si los usuarios gratuitos tienen acceso al chat en vivo o cuál es el tamaño del espacio de almacenamiento gratuito.
Demasiendo el valor con una prueba A/B
Un experimento deja claro este punto al hacer la misma pregunta con y sin contexto obtenido. Primero, una función de referencia:
def no_rag(question):
out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
return out[0]["generated_text"].strip()
Luego, la comparación:
q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context :", rag_answer(q))
La respuesta sin contexto es una conjetura, mientras que la que incluye contexto debe indicar que el chat en vivo está disponible solo para usuarios Pro y que los usuarios gratuitos reciben respuestas por correo electrónico. Pruebe con tres preguntas diferentes. Esta demostración A/B es la más convincente de RAG que existe, así que guárdela para explicar la idea a otros.
Mejoras según su beneficio
Con un sistema funcional en mano, estas mejoras son las más importantes, aproximadamente en orden de beneficio:
- Mostrar fuentes. La función ya devuelve
retrieved, por lo que se deben imprimir los fragmentos correspondientes a cada respuesta. Los productos reales de RAG tienen éxito o fracasan en este aspecto. Una respuesta citada es fiable, mientras que una sin citar no es más que una conjetura presentada de forma atractiva.
k a 5 y establezca una puntuación mínima, descartando todo aquello por debajo de aproximadamente 0.25. Cuando ningún fragmento cumpla los requisitos, responda “No lo sé” y omita por completo la llamada al modelo. Esto elimina la mayoría de las alucinaciones, como la que aparece en el ejercicio del Paso 5.chunk_vectors @ q_vec escanea cada fragmento, lo cual es aceptable para cientos de ellos pero lento para millones. Chroma (pip install chromadb) indexa los vectores para una búsqueda rápida y los guarda en disco, de modo que no se vuelvan a procesar al iniciar el programa. Los conceptos de incrustación, almacenamiento y búsqueda por coseno permanecen los mismos; Chroma simplemente los aplica a gran escala.generator(...) por una solicitud a su API local. Una versión gratuita de la API de Gemini o Groq de Google también funciona al sustituir únicamente la línea de generación en rag_answer. Solo cambia una función, mientras que la recuperación de información permanece igual.rag_answer y verifique su corrección. No es algo atractivo, pero es lo que distingue un prototipo de un producto, ya que lo que no se mide no puede mejorarse.Diagnóstico de problemas
RAG es búsqueda más asistente de escritura, combinados, y casi todos los problemas se deben a una de estas dos partes. Una respuesta incorrecta con el fragmento correcto recuperado es un problema de generación o de prompt. Una respuesta incorrecta cuando el fragmento correcto nunca apareció es un problema de recuperación relacionado con la segmentación, los embeddings o el valor k. Decida cuál de las dos partes está defectuosa antes de cambiar algo. Mantener retrieve() y rag_answer() como funciones separadas existe precisamente para que cada parte pueda inspeccionarse por separado, y esa separación es el hábito más valioso aquí.
Solución de problemas
- Descarga lenta o fallida: los modelos provienen de Hugging Face en la primera ejecución y se almacenan en caché en
~/.cache/huggingface. Una vez almacenados en caché, funcionan sin conexión.
google/flan-t5-small con un chunk_size de 40. El modelo de embedding es lo suficientemente pequeño como para no ralentizar el proceso.pip install torch --index-url https://download.pytorch.org/whl/cpu.max_new_tokens, ya que Flan-T5 tiende a detenerse temprano. Un generador más grande, como el proporcionado por Ollama, es más efectivo que cualquier ajuste adicional.Realizar los ejercicios permite adquirir una comprensión más profunda de RAG que la que la mayoría de las personas obtienen con solo usar una capa de abstracción proporcionada por bibliotecas.