ruta ai/agentes / módulo 1

A1 — RAG core

Outcome del módulo: construir el pipeline RAG completo (chunk → embed → pgvector → retrieval top-k) y poder explicar por qué cada decisión de diseño determina si el sistema responde fundado en datos o alucina.

Concepto 1 — Qué es RAG y qué problema resuelve

Idea núcleo: RAG separa “lo que el modelo sabe” (pesos fijos de entrenamiento) de “lo que el modelo necesita saber ahora” (tu corpus), inyectando fragmentos recuperados en el prompt antes de generar.

Entender (capa 1)

Texto: un LLM entrenado tiene un corte de conocimiento fijo y no conoce datos privados ni actualizados después de ese corte. Preguntado sobre algo que no sabe, puede alucinar una respuesta plausible en vez de admitir que no tiene la información. RAG (Retrieval-Augmented Generation) resuelve esto con tres pasos: retrieval (buscar los fragmentos más relevantes de un corpus propio, indexado por significado), augmentation (insertar esos fragmentos como contexto en el prompt) y generation (el modelo responde fundado en ese contexto, idealmente citando la fuente). RAG reduce la alucinación, no la elimina: si el retrieval trae basura, el modelo genera fundado en basura.

Visual:

flowchart LR
    Q[pregunta del usuario] --> R[retriever]
    C[(corpus propio<br/>indexado por significado)] --> R
    R --> CTX[fragmentos relevantes]
    CTX --> LLM[LLM genera respuesta<br/>fundada en el contexto]
    LLM --> RESP[respuesta + fuente citada]

Video: What is Retrieval-Augmented Generation (RAG)? — IBM Technology

Ejemplo (mal → bien):

# mal: preguntarle al LLM directo sobre un documento privado que nunca vio
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "¿Cuál es la política de reembolsos de mi empresa?"}],
)
# el modelo no tiene ese documento: inventa una política plausible, o dice que no sabe

# bien: recuperar el fragmento relevante del corpus propio y fundamentar la respuesta ahí
contexto = retriever.buscar("política de reembolsos", k=3)
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": f"Contexto:\n{contexto}\n\nPregunta: ¿cuál es la política de reembolsos?"
    }],
)

Fijar (capa 2)

Nota atómica:

Feynman: “un LLM sin RAG es un examen a libro cerrado: contesta de memoria, y si no se acuerda bien, inventa con confianza. RAG es el mismo examen pero a libro abierto: le das el capítulo correcto antes de preguntar, así que responde leyendo, no adivinando.”

Aplicar (capa 3)

Elegí un documento que el modelo no puede conocer (un PDF interno, un README de un repo tuyo). Preguntale algo puntual sin darle el documento y anotá qué responde. Después pegale el fragmento relevante en el prompt y compará la respuesta.

Límites

Concepto 2 — Chunking (size / overlap)

Idea núcleo: los documentos se parten en fragmentos (chunks) porque no entran completos en el contexto ni conviene recuperarlos enteros; el tamaño y el overlap determinan si el retrieval encuentra la respuesta completa o la corta a la mitad.

Entender (capa 1)

Texto: un chunk demasiado grande diluye la señal semántica (el embedding representa “un poco de todo”, el retrieval se vuelve impreciso) y desperdicia presupuesto de contexto. Un chunk demasiado chico pierde contexto propio (una oración sola, sin el párrafo que la explica, puede ser ambigua). El overlap (superposición entre chunks consecutivos) evita que una idea quede cortada justo en el borde entre dos chunks. El chunking naive corta por cantidad fija de caracteres sin mirar la estructura; el chunking semántico respeta párrafos, headers y oraciones antes de cortar a lo bruto.

Visual:

flowchart LR
    DOC[documento completo] --> CH1[chunk 1]
    DOC --> CH2[chunk 2]
    DOC --> CH3[chunk 3]
    CH1 -.overlap.- CH2
    CH2 -.overlap.- CH3

Video: Chunking Strategies in RAG: Optimising Data for AI Responses — Mervin Praison

Ejemplo (mal → bien):

# mal: corte ciego por caracteres fijos, sin overlap, ideas partidas al medio
def chunk_mal(texto: str, tamano: int = 500) -> list[str]:
    return [texto[i:i + tamano] for i in range(0, len(texto), tamano)]

# bien: overlap para no perder contexto en el borde, respeta estructura antes de cortar a lo bruto
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # tamaño objetivo por chunk
    chunk_overlap=50,     # se repite al final de un chunk y al inicio del siguiente
    separators=["\n\n", "\n", ". ", " "],   # intenta cortar por párrafo antes que por caracter
)
chunks = splitter.split_text(texto)

Fijar (capa 2)

Nota atómica:

Feynman: “cortar un documento sin overlap es como cortar una foto en cuadrados sin superponer los bordes: si la cara de alguien queda justo en el corte, ningún cuadrado tiene la cara completa. El overlap es superponer un poco los bordes para que nada importante quede partido.”

Aplicar (capa 3)

Tomá un documento de al menos 3000 caracteres. Chunkealo con chunk_size=200, overlap=0 y con chunk_size=500, overlap=50. Elegí un dato puntual del documento y verificá en cuál de los dos chunkings queda completo dentro de un solo chunk.

Límites

Concepto 3 — Embeddings

Idea núcleo: un embedding convierte texto en un vector numérico donde la cercanía geométrica representa cercanía semántica — eso habilita “buscar por significado” en vez de por palabra exacta.

Entender (capa 1)

Texto: un modelo de embeddings mapea cualquier texto a un vector de dimensión fija (por ejemplo, 1536 números). Textos con significado similar producen vectores cercanos en ese espacio, aunque no compartan ninguna palabra literal (sinónimos, paráfrasis, otro idioma). Es crítico usar el mismo modelo de embeddings para indexar el corpus y para embedear la consulta en tiempo de búsqueda — mezclar modelos produce vectores en espacios distintos, no comparables.

Visual:

flowchart LR
    T1["texto: 'cómo cancelar mi suscripción'"] --> M[modelo de embeddings]
    T2["texto: 'dar de baja el plan'"] --> M
    M --> V1[vector A]
    M --> V2[vector B]
    V1 -.cerca en el espacio.- V2

Video: Vectoring Words (Word Embeddings) — Computerphile

Ejemplo (mal → bien):

from openai import OpenAI
import numpy as np

client = OpenAI()

def embed(textos: list[str], modelo: str = "text-embedding-3-small") -> list[list[float]]:
    resp = client.embeddings.create(model=modelo, input=textos)
    return [d.embedding for d in resp.data]

# mal: buscar coincidencia textual exacta, no entiende sinónimos ni paráfrasis
def buscar_mal(pregunta: str, docs: list[str]) -> list[str]:
    return [d for d in docs if pregunta.lower() in d.lower()]

# bien: comparar en espacio semántico
def similitud_coseno(a: list[float], b: list[float]) -> float:
    a_arr, b_arr = np.array(a), np.array(b)
    return float(a_arr @ b_arr / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

Fijar (capa 2)

Nota atómica:

Feynman: “el embedding es como traducir una frase a coordenadas GPS de un mapa inventado por ese modelo específico. Dos modelos distintos dibujan mapas distintos: las coordenadas de uno no significan nada en el mapa del otro.”

Aplicar (capa 3)

Embedeá 5 frases donde 2 sean parafraseos entre sí y las otras 3 no tengan relación. Calculá similitud coseno entre todos los pares y confirmá que el par parafraseado da el score más alto sin compartir palabras literales.

Límites

Concepto 4 — Vector store con pgvector (similitud coseno, HNSW)

Idea núcleo: pgvector agrega un tipo de columna vectorial y operadores de distancia a Postgres, así el retrieval semántico vive en la misma base relacional que ya tenés, sin sumar otra pieza de infraestructura.

Entender (capa 1)

Texto: pgvector es una extensión de Postgres que agrega el tipo vector(N) y operadores de distancia: <-> (euclidiana), <=> (coseno), <#> (producto interno negativo). Para buscar top-k más cercanos sin índice, Postgres compara contra cada fila (exact search) — funciona, pero no escala pasado unos miles/decenas de miles de filas. El índice HNSW (Hierarchical Navigable Small World) construye una estructura de grafo en capas que permite aproximar el vecino más cercano mucho más rápido, a costa de recall no-100% (aproximado, no exacto) — trade-off velocidad/precisión aceptable en la gran mayoría de casos de RAG.

Visual:

flowchart TD
    T[tabla chunks<br/>id, contenido, embedding vector] --> IDX[índice HNSW<br/>sobre la columna embedding]
    Q[vector de la consulta] --> IDX
    IDX --> TOPK[top-k por distancia coseno<br/>aproximado, muy rápido]

Video: 18 Months of Pgvector Learnings in 47 Minutes — Tiger Data

Ejemplo (mal → bien):

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id BIGSERIAL PRIMARY KEY,
    documento_id TEXT NOT NULL,
    contenido TEXT NOT NULL,
    embedding VECTOR(1536) NOT NULL
);

-- HNSW: aproximado pero rápido a escala; sin índice, exact search no escala
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
# mal: traer TODOS los embeddings a Python y comparar en memoria
filas = db.execute("SELECT id, embedding FROM chunks").fetchall()
similitudes = [(f.id, similitud_coseno(query_vec, f.embedding)) for f in filas]   # no escala

# bien: la distancia se calcula en la DB, con índice HNSW
# setup una vez por conexión: register_vector adapta list[float] ↔ tipo vector
from pgvector.psycopg import register_vector
register_vector(conn)          # sin esto, psycopg manda un ARRAY y falla el operador <=>

resultados = db.execute(
    "SELECT id, contenido, embedding <=> %s AS distancia "
    "FROM chunks ORDER BY distancia LIMIT 5",
    (query_vec,),              # ya adaptado a vector; alternativa sin register: %s::vector con string '[...]'
).fetchall()

Fijar (capa 2)

Nota atómica:

Feynman: “exact search es buscar tu llave revisando cajón por cajón, uno por uno: exacto pero lento si hay muchos cajones. HNSW es como tener un mapa de atajos entre cajones parecidos: llegás casi siempre al correcto, mucho más rápido, con una probabilidad chica de no encontrar el óptimo absoluto.”

Aplicar (capa 3)

Creá la tabla chunks con pgvector, insertá al menos 100 filas de prueba, corré la misma consulta top-5 con y sin índice HNSW (EXPLAIN ANALYZE) y compará el plan de ejecución y el tiempo.

Límites

Concepto 5 — Retrieval top-k y armado de contexto

Idea núcleo: recuperar los k chunks más cercanos no es el final del trabajo; cómo se arma el prompt con esos chunks (cuántos, en qué orden, con qué fuente, dentro de qué presupuesto de tokens) determina si el LLM responde bien o se pierde.

Entender (capa 1)

Texto: elegir k es un trade-off: muy chico y falta información para responder; muy grande y metés ruido (chunks poco relevantes), costo extra, y el riesgo de “lost in the middle” (concepto A0-2). Además de recortar por k, conviene filtrar por un umbral de distancia mínima: si el chunk más cercano igual está lejos, mejor no usarlo que forzar una respuesta mal fundada. Cada chunk insertado en el prompt debería llevar su fuente (documento, sección) para que el modelo pueda citar de dónde sale cada afirmación — eso es lo que hace la respuesta auditable.

Visual:

flowchart LR
    Q[pregunta] --> E[embed de la pregunta]
    E --> V[(pgvector: top-k por <=>)]
    V --> F[filtro por umbral de distancia]
    F --> A[armar contexto:<br/>chunk + fuente, dentro del presupuesto de tokens]
    A --> LLM[prompt final al LLM]

Video: Making Retrieval Augmented Generation Better — Pinecone

Ejemplo (mal → bien):

# mal: pegar los k chunks tal cual, sin fuente, sin filtro de relevancia, sin límite
def armar_contexto_mal(resultados: list[dict]) -> str:
    return "\n".join(r["contenido"] for r in resultados)

# bien: filtra por umbral, limita cantidad, atribuye fuente
UMBRAL_DISTANCIA = 0.35   # [verificar] calibrar por corpus y modelo de embeddings

def armar_contexto(resultados: list[dict], max_chunks: int = 5) -> str:
    relevantes = [r for r in resultados if r["distancia"] < UMBRAL_DISTANCIA][:max_chunks]
    if not relevantes:
        return ""   # sin contexto real: mejor admitirlo que alucinar
    return "\n\n".join(
        f"[Fuente: {r['documento_id']}]\n{r['contenido']}" for r in relevantes
    )

def prompt_rag(pregunta: str, contexto: str) -> str:
    if not contexto:
        return f"No hay información suficiente en el corpus para responder: {pregunta}"
    return (
        "Respondé SOLO con la información del contexto. Citá la fuente de cada dato. "
        "Si el contexto no alcanza, decilo explícitamente en vez de inventar.\n\n"
        f"Contexto:\n{contexto}\n\nPregunta: {pregunta}"
    )

Fijar (capa 2)

Nota atómica:

Feynman: “top-k sin umbral es como preguntarle a las 5 personas más cercanas en la calle, aunque estén a 3 cuadras. El umbral es decir: solo pregunto si hay alguien a menos de 10 metros; si no, prefiero decir ‘no sé’ que inventar con la info de un desconocido lejano.”

Aplicar (capa 3)

Sobre tu tabla chunks de pgvector, escribí la función completa: recibe una pregunta, la embedea, trae top-5 de la DB, filtra por umbral, arma el contexto con fuente, y arma el prompt final. Probá con una pregunta que SÍ está en el corpus y una que NO, y confirmá que la segunda devuelve “no tengo información” en vez de una respuesta inventada.

Límites

Build step

Ingesta de un corpus público (chunk + embed → pgvector) y endpoint de retrieval top-k sobre el servicio del build step anterior (A0):

def ingerir_documento(documento_id: str, texto: str) -> None:
    chunks = splitter.split_text(texto)
    vectores = embed(chunks)
    for contenido, vector in zip(chunks, vectores):
        db.execute(
            "INSERT INTO chunks (documento_id, contenido, embedding) VALUES (%s, %s, %s)",
            (documento_id, contenido, vector),
        )

class ResultadoBusqueda(BaseModel):
    documento_id: str
    contenido: str
    distancia: float

@app.get("/buscar", response_model=list[ResultadoBusqueda])
def buscar(pregunta: str, k: int = 5) -> list[ResultadoBusqueda]:
    query_vec = embed([pregunta])[0]
    filas = db.execute(
        "SELECT documento_id, contenido, embedding <=> %s AS distancia "
        "FROM chunks ORDER BY distancia LIMIT %s",
        (query_vec, k),        # conn con register_vector(conn) para adaptar el vector
    ).fetchall()
    return [ResultadoBusqueda(**f) for f in filas]   # **f requiere filas como dict → conn con row_factory=dict_row (psycopg3)

Checklist de dominio A1

Marcás cuando podés explicar (Feynman) + aplicar cada uno:

Salida verificable del módulo: un corpus público ingerido en pgvector (chunkeado + embebido) y un endpoint /buscar que devuelve top-k con fuente y distancia, capaz de admitir “no tengo información” cuando corresponde en vez de alucinar.