De tu primer chunk de texto y tu primer embedding, hasta un sistema RAG de producción con re-ranking, búsqueda híbrida y evaluación de fidelidad.
0 / 10 completado
Nivel 1 · Fundamentos
Qué es RAG y por qué existe
Objetivo: entender el problema que resuelve RAG y preparar el entorno de trabajo.
Retrieval-Augmented Generation (RAG) resuelve un problema muy concreto: un LLM solo "sabe" lo que vio en su entrenamiento, con una fecha de corte fija, y no conoce tus documentos internos, tu base de conocimiento ni información que cambió ayer. RAG conecta el modelo a una fuente de información externa y actualizada en el momento de la consulta, en vez de depender solo de lo que el modelo memorizó.
La arquitectura básica tiene dos partes: un retriever que busca los fragmentos de texto más relevantes para una pregunta, y un generator (el LLM) que redacta la respuesta usando esos fragmentos como contexto.
Instala las dependencias
TERMINAL
pip install openai psycopg[binary] pgvector
export OPENAI_API_KEY="sk-tu-api-key-aqui"
# Postgres con la extensión pgvector (Docker es lo más rápido para probar)
docker run -d --name pg-rag -e POSTGRES_PASSWORD=postgres -p 5432:5432 pgvector/pgvector:pg17
setup.sql
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documentos (
id SERIAL PRIMARY KEY,
contenido TEXT NOT NULL,
embedding VECTOR(1536), -- dimensión de text-embedding-3-small
fuente TEXT
);
Reto
Levantá el contenedor de Postgres con pgvector y confirmá que la extensión se instaló corriendo SELECT * FROM pg_extension WHERE extname = 'vector';
Nivel 2 · Chunking
Dividir documentos en fragmentos
Objetivo: partir documentos largos en fragmentos (chunks) que el modelo de embeddings pueda procesar bien.
No podés meter un documento de 50 páginas entero en un embedding y esperar buenos resultados — hay que dividirlo en fragmentos manejables. El chunking es una de las decisiones que más impacta la calidad final de un RAG, y sin embargo se suele tratar como un detalle menor.
Recursive character splitting con overlap
chunking.py
def dividir_en_chunks(texto: str, tamano: int = 800, overlap: int = 150) -> list[str]:
"""Divide texto en chunks con solapamiento, respetando saltos de párrafo cuando puede."""
separadores = ["\n\n", "\n", ". ", " "]
chunks = []
inicio = 0
while inicio < len(texto):
fin = min(inicio + tamano, len(texto))
fragmento = texto[inicio:fin]
# Si no llegamos al final, cortar en el separador más cercano hacia atrás
if fin < len(texto):
for sep in separadores:
pos = fragmento.rfind(sep)
if pos > tamano * 0.5: # no cortar demasiado corto
fragmento = fragmento[:pos + len(sep)]
break
chunks.append(fragmento.strip())
inicio += len(fragmento) - overlap # el overlap evita perder contexto en el borde
return [c for c in chunks if c]
Por qué el overlap importa
Sin overlap, una idea que cruza el límite entre dos chunks queda cortada a la mitad en ambos — ni el chunk A ni el B tienen el contexto completo. Un overlap de 15-20% del tamaño del chunk suele ser un buen punto de partida; para documentos muy técnicos con definiciones largas, vale la pena subirlo.
Reto
Tomá un documento real (un artículo de este blog, por ejemplo) y probá dividirlo con tamano=400 y tamano=1200. Compará cuántos chunks salen y si alguna idea queda cortada de forma incómoda.
Nivel 3 · Embeddings
Convertir texto en vectores
Objetivo: generar embeddings de los chunks con la API de OpenAI.
Un embedding es una representación numérica (un vector) del significado de un texto — textos con significado similar producen vectores cercanos entre sí en ese espacio. Esa cercanía matemática es lo que permite "buscar por significado" en vez de por coincidencia exacta de palabras.
embeddings.py
from openai import OpenAI
client = OpenAI()
def generar_embedding(texto: str) -> list[float]:
respuesta = client.embeddings.create(
model="text-embedding-3-small", # 1536 dimensiones, buen default para producción
input=texto,
)
return respuesta.data[0].embedding
vector = generar_embedding("RAG conecta un LLM a información externa actualizada")
print(len(vector)) # 1536
text-embedding-3-small cuesta $0.02 por millón de tokens y es suficiente para el 80% de los casos de RAG en producción. text-embedding-3-large (3072 dimensiones) da un poco más de calidad a más costo — solo vale la pena si ya mediste que small se queda corto para tu caso.
Reto
Generá embeddings de 3 frases: dos con significado similar pero palabras distintas, y una completamente distinta. Calculá la similitud coseno entre pares (nivel 5 te muestra cómo) y confirmá que las dos similares quedan más cerca.
Nivel 4 · Vector store
Guardar los embeddings en pgVector
Objetivo: insertar chunks y sus embeddings en Postgres usando la extensión pgvector.
pgVector agrega un tipo de dato vector y operadores de distancia a Postgres — así podés guardar tus embeddings en la misma base de datos que ya usás para todo lo demás, sin sumar un vector store separado a tu infraestructura.
ingesta.py
import psycopg
from pgvector.psycopg import register_vector
conn = psycopg.connect("postgresql://postgres:postgres@localhost:5432/postgres")
register_vector(conn)
def guardar_chunk(contenido: str, embedding: list[float], fuente: str):
with conn.cursor() as cur:
cur.execute(
"INSERT INTO documentos (contenido, embedding, fuente) VALUES (%s, %s, %s)",
(contenido, embedding, fuente),
)
conn.commit()
# Pipeline completo: documento -> chunks -> embeddings -> guardado
texto = open("manual_producto.txt").read()
for chunk in dividir_en_chunks(texto):
emb = generar_embedding(chunk)
guardar_chunk(chunk, emb, fuente="manual_producto.txt")
Reto
Ingerí 2-3 documentos de texto distintos y confirmá con SELECT count(*) FROM documentos; que los chunks de todos quedaron guardados con su fuente correcta.
Nivel 5 · Retrieval
Búsqueda por similitud de coseno
Objetivo: buscar los chunks más relevantes para una pregunta usando el operador de distancia coseno de pgvector.
pgvector expone el operador <=> para distancia coseno — mientras más chica la distancia, más similares los vectores. Como la distancia coseno va de 0 a 2, 1 - distancia te da un puntaje de similitud entre -1 y 1, más intuitivo de leer.
retrieval.py
def buscar_relevantes(pregunta: str, top_k: int = 4) -> list[dict]:
embedding_pregunta = generar_embedding(pregunta)
with conn.cursor() as cur:
cur.execute(
"""
SELECT contenido, fuente, 1 - (embedding <=> %s) AS similitud
FROM documentos
ORDER BY embedding <=> %s
LIMIT %s
""",
(embedding_pregunta, embedding_pregunta, top_k),
)
return [
{"contenido": r[0], "fuente": r[1], "similitud": r[2]}
for r in cur.fetchall()
]
resultados = buscar_relevantes("¿Cómo configuro las notificaciones?")
for r in resultados:
print(f"[{r['similitud']:.3f}] {r['fuente']}: {r['contenido'][:80]}...")
Para datasets grandes (cientos de miles de vectores), agregá un índice HNSW a la columna embedding — sin índice, pgvector hace un escaneo secuencial completo en cada búsqueda, que no escala.
Reto
Corré CREATE INDEX ON documentos USING hnsw (embedding vector_cosine_ops); y compará el tiempo de una búsqueda antes y después con EXPLAIN ANALYZE.
Nivel 6 · Generación aumentada
De los chunks a una respuesta con fuentes
Objetivo: combinar los chunks recuperados con la pregunta original en un prompt, y generar una respuesta citando de dónde salió la información.
rag.py
def responder_con_rag(pregunta: str) -> str:
chunks = buscar_relevantes(pregunta, top_k=4)
contexto = "\n\n".join(
f"[Fuente: {c['fuente']}]\n{c['contenido']}" for c in chunks
)
prompt = f"""Respondé la pregunta usando SOLO la información del contexto.
Si el contexto no tiene la respuesta, decí explícitamente que no tenés esa información.
Citá la fuente entre corchetes al final de cada afirmación.
Contexto:
{contexto}
Pregunta: {pregunta}"""
respuesta = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
return respuesta.choices[0].message.content
print(responder_con_rag("¿Cómo configuro las notificaciones?"))
La instrucción "si el contexto no tiene la respuesta, decilo explícitamente" es la línea más importante del prompt — sin ella, el modelo va a rellenar los huecos con información inventada en vez de admitir que no sabe.
Reto
Hacele una pregunta que sepas que NO está en tus documentos ingeridos y confirmá que el modelo admite no tener esa información en vez de alucinar una respuesta.
Nivel 7 · Re-ranking
Mejorar la calidad de lo recuperado
Objetivo: reordenar los resultados de la búsqueda vectorial con un modelo especializado antes de pasarlos al LLM.
La búsqueda por similitud de coseno es rápida pero aproximada — trae candidatos razonables, no necesariamente los mejores en orden correcto. Un re-ranker (típicamente un cross-encoder) evalúa el par pregunta-chunk con más precisión, a costa de ser más lento — por eso se usa solo sobre los top-N candidatos que ya trajo la búsqueda vectorial, no sobre toda la base.
reranking.py
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def buscar_con_reranking(pregunta: str, top_k_inicial: int = 15, top_k_final: int = 4) -> list[dict]:
candidatos = buscar_relevantes(pregunta, top_k=top_k_inicial) # búsqueda vectorial amplia
pares = [[pregunta, c["contenido"]] for c in candidatos]
scores = reranker.predict(pares) # el cross-encoder puntúa cada par
for c, score in zip(candidatos, scores):
c["score_rerank"] = float(score)
candidatos.sort(key=lambda c: c["score_rerank"], reverse=True)
return candidatos[:top_k_final]
Este patrón de "búsqueda amplia y barata + re-ranking preciso sobre pocos candidatos" es el mismo principio detrás de casi todos los sistemas de búsqueda modernos, desde motores de búsqueda hasta sistemas de recomendación.
Reto
Compará el orden de los top-4 resultados con y sin re-ranking para una pregunta ambigua — vas a notar que el re-ranker suele mover al primer lugar un chunk que la búsqueda vectorial había puesto más abajo.
Nivel 8 · Búsqueda híbrida
Combinar búsqueda semántica y léxica
Objetivo: combinar la búsqueda vectorial con búsqueda de texto completo para no perder coincidencias exactas.
La búsqueda vectorial es excelente para significado, pero floja para términos exactos: códigos de error, nombres de producto, números de parte, siglas específicas. Postgres ya trae búsqueda de texto completo (tsvector) que sí encuentra coincidencias exactas — combinarlas da mejores resultados que cualquiera de las dos por separado.
busqueda_hibrida.py
def buscar_hibrido(pregunta: str, top_k: int = 4) -> list[dict]:
embedding_pregunta = generar_embedding(pregunta)
with conn.cursor() as cur:
cur.execute(
"""
WITH busqueda_vectorial AS (
SELECT id, contenido, fuente,
ROW_NUMBER() OVER (ORDER BY embedding <=> %s) AS rank_vector
FROM documentos
ORDER BY embedding <=> %s LIMIT 20
),
busqueda_texto AS (
SELECT id, contenido, fuente,
ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('spanish', contenido), plainto_tsquery('spanish', %s)) DESC) AS rank_texto
FROM documentos
WHERE to_tsvector('spanish', contenido) @@ plainto_tsquery('spanish', %s)
LIMIT 20
)
-- Reciprocal Rank Fusion: combina ambos rankings en un solo puntaje
SELECT COALESCE(v.contenido, t.contenido) AS contenido,
COALESCE(v.fuente, t.fuente) AS fuente,
COALESCE(1.0 / (60 + v.rank_vector), 0) + COALESCE(1.0 / (60 + t.rank_texto), 0) AS score_rrf
FROM busqueda_vectorial v
FULL OUTER JOIN busqueda_texto t ON v.id = t.id
ORDER BY score_rrf DESC
LIMIT %s
""",
(embedding_pregunta, embedding_pregunta, pregunta, pregunta, top_k),
)
return [{"contenido": r[0], "fuente": r[1]} for r in cur.fetchall()]
Reciprocal Rank Fusion (RRF) es la técnica estándar para combinar dos rankings distintos sin tener que normalizar puntajes de escalas diferentes (similitud coseno vs. rank de texto) — simplemente suma 1/(k + posición) de cada lista, dándole más peso a lo que aparece bien rankeado en ambas búsquedas.
Reto
Probá una pregunta que incluya un código o término exacto (ej. "error E404" o un nombre de producto específico) y compará los resultados de buscar_relevantes vs. buscar_hibrido.
Nivel 9 · Evaluación
Medir si tu RAG realmente funciona
Objetivo: evaluar la fidelidad de las respuestas y detectar cuándo el sistema debería admitir que no sabe.
Un RAG que "se ve bien" en unas pocas pruebas manuales puede estar fallando silenciosamente en producción. Dos métricas clave: faithfulness (¿la respuesta se sostiene solo con el contexto recuperado, o el modelo agregó información de su entrenamiento?) y relevancy (¿el contexto recuperado realmente tenía que ver con la pregunta?).
evaluacion.py
def evaluar_fidelidad(pregunta: str, contexto: str, respuesta: str) -> dict:
"""Usa un segundo LLM como juez para verificar que la respuesta se sostiene con el contexto."""
prompt = f"""Evaluá si la RESPUESTA está completamente sustentada por el CONTEXTO.
Respondé en JSON: {{"fiel": true/false, "razon": "..."}}
CONTEXTO:
{contexto}
RESPUESTA:
{respuesta}"""
resultado = client.chat.completions.create(
model="gpt-4o-mini", # un modelo económico alcanza para esta tarea de verificación
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
import json
return json.loads(resultado.choices[0].message.content)
Manejar el caso "no tengo información suficiente"
Si la similitud del mejor chunk recuperado está muy por debajo de tus otros resultados típicos (por ejemplo, menor a 0.3 en tu dataset), es una señal de que probablemente no hay contexto relevante — vale la pena responder "no tengo información sobre esto" antes de intentar generar una respuesta con contexto débil.
Reto
Armá un set de 10 preguntas de prueba (algunas con respuesta clara en tus documentos, otras sin ella) y corré evaluar_fidelidad sobre cada una — contá cuántas fallan.
Nivel 10 · Arquitectura final
RAG de producción: ingesta continua y escalado
Objetivo: ensamblar todo lo anterior en una arquitectura que se mantiene actualizada sola y escala con el volumen de documentos y consultas.
1. Pipeline de ingesta incremental
En producción los documentos cambian — no querés reprocesar toda la base cada vez que un archivo se actualiza. La solución estándar: guardar un hash del contenido de cada documento, y solo reingerir (re-chunkear + re-embedear) los que cambiaron desde la última corrida.
ingesta_incremental.py
import hashlib
def necesita_reingesta(doc_id: str, contenido: str) -> bool:
hash_actual = hashlib.sha256(contenido.encode()).hexdigest()
with conn.cursor() as cur:
cur.execute("SELECT hash FROM documentos_meta WHERE doc_id = %s", (doc_id,))
row = cur.fetchone()
if row is None or row[0] != hash_actual:
return True
return False
# En el pipeline de ingesta: solo re-procesar lo que cambió
for doc_id, contenido in documentos_fuente.items():
if necesita_reingesta(doc_id, contenido):
borrar_chunks_antiguos(doc_id)
for chunk in dividir_en_chunks(contenido):
guardar_chunk(chunk, generar_embedding(chunk), fuente=doc_id)
2. Caching de embeddings de consultas frecuentes
Si tu RAG recibe preguntas repetidas o similares (típico en soporte al cliente), cachear el embedding de la pregunta y los resultados de retrieval evita llamadas redundantes a la API de embeddings y a la base de datos.
3. Escalado del vector store
arquitectura final
Documentos fuente (S3, CMS, base de datos)
│
▼
Pipeline de ingesta incremental (solo re-procesa lo que cambió)
│
▼
Chunking + Embeddings (batch, con caché de embeddings ya generados)
│
▼
pgVector (con índice HNSW, réplicas de lectura si el volumen de consultas crece)
│
▼
Consulta usuario → Embedding → Búsqueda híbrida (vector + texto) → Re-ranking
│
▼
LLM genera respuesta con fuentes citadas
│
▼
Evaluación de fidelidad (muestreo continuo, no en cada request)
Con esto tenés el camino completo: de un chunk de texto suelto en el Nivel 2, a un sistema RAG de producción con ingesta que se mantiene al día sola, búsqueda híbrida, re-ranking, y evaluación continua de si las respuestas realmente se sostienen en tus documentos.
Reto final
Tomá el pipeline completo de los niveles 1-9 y envolvelo en un endpoint FastAPI POST /preguntar, con el pipeline de ingesta incremental corriendo aparte como un job programado.