Notas prácticas: Construir un sistema RAG desde cero — Enfoque práctico, sin costos de API
Guía paso a paso práctica: Construye un sistema RAG desde cero — Enfoque práctico, sin costos de API: contratos, verificaciones y espacios para código listo para usar para los equipos que implementan este patrón.
Las notas siguientes reconstruyen un camino práctico para “Construir un sistema RAG desde cero: enfoque práctico, sin costos de API”. Se da énfasis a los contratos, las verificaciones y los marcadores de posición para código listo para usar, en lugar de un enfoque motivacional. Al trabajar en la etapa de descripción general, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto el camino óptimo como el de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras realizadas posteriormente.
Qué es realmente RAG (60 segundos)
La etapa de determinar qué es realmente RAG funciona mejor cuando se trata como una superficie medible. Capture un transcripte ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.
question ──► [embed] ──► [search your docs] ──► top chunks ──┐
▼
[LLM: "answer using this context"] ──► answer
Paso 0 — Configuración
La etapa de configuración Paso 0 funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace completaciones parciales silenciosas. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.
pip install sentence-transformers transformers torch numpy
Paso 1: Una base de conocimientos que el modelo nunca ha visto
La etapa de conocimiento de Paso 1 A funciona mejor cuando se trata como una superficie medible. Capture un transcripto ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la demostración a entornos compartidos. Asigne un presupuesto de tokens por turno y por sesión. Las herramientas agenciales amplían el contexto de manera intensiva; los límites estrictos impiden que las demostraciones se conviertan en facturas inesperadas. La etapa de conocimiento de Paso 1 A funciona mejor cuando se trata como una superficie medible. Capture un transcripto ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente tanto el camino óptimo como el camino de recuperación. Los intentos repetidos, los controles humanos y el manejo de correos no entregados forman parte del producto, no son mejoras realizadas posteriormente.
# 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.""",
"""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.""",
]
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"])
Paso 2: Fragmentación
En la fase de fragmentación del Paso 2, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Cite los pasajes que realmente sustentan la respuesta. Sin citaciones, los operadores no pueden distinguir entre alucinaciones y brechas en el indexado.
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
# 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], "...")
Paso 3: Embeddings: convertir texto en vectores
En la fase de inserción de los pasos 3, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Cite los pasajes que realmente sirvieron de base para la respuesta. Sin citaciones, los operadores no podrán distinguir entre alucinaciones y lagunas en el indexado.
from sentence_transformers import SentenceTransformer
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)
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
Paso 4: Recuperación: encontrar los fragmentos que responden a una pregunta
En la fase de búsqueda del Paso 4, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Cite los pasajes que realmente sirvieron de base para la respuesta; sin citas, los operadores no pueden distinguir entre alucinaciones y lagunas en el indexado. Documente tanto la ruta óptima como la ruta de recuperación. Las intentonas repetidas, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras realizadas posteriormente.
import numpy as np
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]}...")
Paso 5: Generación: permitir que el modelo responda a partir del contexto
Al trabajar en la fase de generación del Paso 5, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones del código. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una sola responsabilidad y no a un proceso complicado. Almacene en caché las instrucciones del sistema estables y los esquemas de las herramientas. Reenviar un preámbulo idéntico es una causa común de ineficiencias.
from transformers import pipeline
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]}...")
Paso 6: Unirlo todo
Al trabajar en la fase 6 “Ponlo en práctica”, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Trate esta fase como un contrato entre los datos de entrada y los resultados validados. Asigne nombres a los artefactos, defina las comprobaciones de éxito y rechace las completaciones parciales silenciosas. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
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")
python rag.py
Paso 7: Demuestre que RAG está funcionando (prueba A/B)
Al trabajar en la fase 7 de Prueba RAG, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mida el rendimiento en la recuperación de información con un conjunto fijo de preguntas antes de ajustar los prompts. Cambiar constantemente los prompts rara vez soluciona un sistema deficiente de recuperación de datos. Al trabajar en la fase 7 de Prueba RAG, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras adicionales realizadas posteriormente.
def no_rag(question):
out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
return out[0]["generated_text"].strip()
q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context :", rag_answer(q))
Paso 8: Mejóralo (elige lo que te interese)
La fase de Mejorarlo del Paso 8 funciona mejor cuando se trata como una superficie medible. Registra una transcripción exitosa, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiere unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una sola responsabilidad y no a un proceso complicado. Separa la política de fragmentación de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.
El modelo mental a mantener
El modelo mental de esta etapa funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Asigne un presupuesto de tokens por turno y por sesión; las herramientas agentes amplían el contexto de forma agresiva; los límites máximos evitan que las demostraciones se conviertan en facturas inesperadas.
Resolución de problemas
La etapa de resolución de problemas funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad. La etapa de resolución de problemas funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son ajustes realizados posteriormente.
Lista de verificación operativa
En la fase de lista de verificación operativa, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto.
Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema.
Cite los pasajes que realmente sustentan la respuesta. Sin citaciones, los operadores no pueden distinguir entre una alucinación y una laguna en el indexado.
Escriba un manual breve: cómo rotar las claves, cómo vaciar la cola de tareas y cómo revertir la última operación de ingestión.
Documente tanto el camino óptimo como el proceso de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son algo que se añada posteriormente.
Cite los pasajes que realmente sustentan la respuesta. Sin citaciones, los operadores no pueden distinguir entre alucinaciones y brechas en el indexado.
Antes de promocionar la pila, congele las versiones, capture una transcripción de referencia para el camino crítico y confirme los pasos de reversión. Los entornos compartidos necesitan límites de velocidad, verificaciones de tenencia y un propietario claro para la rotación de secretos. Prefiera una fiabilidad aburrida a demostraciones ingeniosas únicas.
Nota por lotes para 5223acdafa84: mantenga las claves del proveedor fuera del repositorio, establezca un límite máximo para tokens por sesión y almacene las transcripciones junto a los archivos de evaluación para que los cambios posteriores en el modelo sigan siendo comparables.