Agente de Soporte Interno con n8n + LangChain / LangGraph

5. Fase 1 — Cuentas: OpenAI y LangSmith TU

5.1 OpenAI (el cerebro)

1. Entra a platform.openai.com y crea una cuenta.
2. Menu Billing -> agrega 5 USD de credito (alcanza y sobra).
3. Menu API keys -> Create new secret key. Copiala AHORA (no se vuelve a mostrar). Empieza con sk-....
4. Guardala en un lugar seguro. La vas a pegar en n8n y en el .env de Python.

5.2 LangSmith (ver por dentro al agente) — gratis

1. Entra a smith.langchain.com y crea cuenta (plan gratuito).
2. Settings -> API Keys -> crea una. Empieza con lsv2_....
3. La usaremos en la Fase 7 para ver el "cerebro" del agente traza por traza.

n8n no necesita cuenta externa: es self-host. El "owner account" se crea local la primera vez que abris la interfaz (Fase 2).

6. Fase 2 — Levantar n8n con Docker TU

Con Docker Desktop abierto y corriendo:

docker volume create n8n_data

docker run -d --name n8n -p 5678:5678 -v n8n_data:/home/node/.n8n docker.n8n.io/n8nio/n8n

Explicacion: -d lo corre en segundo plano; -p 5678:5678 publica el puerto; -v n8n_data:... guarda tus flujos aunque reinicies el contenedor.

Comandos utiles de manejo:

docker ps                 # ver que n8n esta corriendo
docker stop n8n           # apagarlo
docker start n8n          # prenderlo
docker logs -f n8n        # ver sus logs en vivo

7. Fase 3 — Lab A: agente de soporte 100% en n8n TU

Objetivo: en 20-30 min tener un agente que responde en un chat, usando OpenAI. Primero sin RAG (para ver el patron), luego le sumamos la base de conocimiento en la Fase 4.

8. Fase 4 — RAG: la base de conocimiento TU

8.1 Crear documentos internos falsos

En c:\prueba-agente\knowledge crea 3-4 archivos .txt o .md con politicas inventadas. Ejemplo de contenido para vacaciones.md:

Politica de vacaciones ACME
- Cada empleado tiene 21 dias habiles por ano.
- Se solicitan con 15 dias de anticipacion en el portal RRHH.
- Aprobacion a cargo del jefe directo en 48 horas.
- Los dias no usados se pierden el 31 de marzo del ano siguiente.

Crea tambien vpn.md, gastos.md, onboarding.md. Cuanto mas concretos, mejor demo.

8.2 Opcion rapida en n8n: Vector Store como tool

n8n trae un Simple Vector Store (en memoria) perfecto para la demo:

1. Agrega un nodo Vector Store (Simple / In-Memory) en modo "insert" para cargar, y otro en modo "retrieve as tool" para consultar.
2. Necesita un nodo Embeddings OpenAI enchufado (usa tu misma credencial).
3. Cargas los documentos (podes pegarlos con un nodo de texto o leerlos con "Read Binary Files"). Para la demo, pegar el texto directo es lo mas rapido.
4. Conecta el Vector Store (modo retrieve) como Tool del AI Agent. Ahora, cuando preguntes "cuantos dias de vacaciones tengo?", el agente busca en la KB y responde con el dato real.

8B. Fase 4B — Base vectorial en PostgreSQL (pgvector) TU

Esta es la version "de empresa" del RAG. Cuando el stack corporativo usa PostgreSQL, la base vectorial suele ser pgvector: la extension que agrega a Postgres un tipo de dato vector y operadores de distancia, convirtiendolo en base vectorial. Ventaja: una sola base para datos y embeddings, sin sumar un servicio aparte.

8B.1 Levantar Postgres con pgvector (Docker)

La imagen oficial ya trae la extension compilada:

docker run -d --name pgvector -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=soporte -p 5432:5432 pgvector/pgvector:pg16

Habilitar la extension dentro de la base soporte:

docker exec -it pgvector psql -U postgres -d soporte -c "CREATE EXTENSION IF NOT EXISTS vector;"

8B.2 Camino A — desde n8n (nodo Postgres PGVector Store)

1. Crea una credencial Postgres en n8n:
   • Host: host.docker.internal (n8n esta en su contenedor; Postgres en otro publicado en tu host)
   • Port: 5432 · Database: soporte · User: postgres · Password: la que pusiste
2. Agrega el nodo Postgres PGVector Store. Usalo en modo Insert para indexar la KB, y otro en modo Retrieve (as tool) para que el AI Agent consulte.
3. Enchufa un nodo Embeddings OpenAI. Tabla p.ej. soporte_kb, dimension 1536 (la de text-embedding-3-small).

8B.3 Camino B — desde Python/LangChain (clase PGVector)

Reemplaza el InMemoryVectorStore del ingest.py de la Fase 5.

pip install langchain-postgres "psycopg[binary]"

Nuevo ingest.py (variante Postgres):

import os, glob
from langchain_postgres import PGVector
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from dotenv import load_dotenv

load_dotenv()

# Desde Python (en tu PC, fuera de Docker) el host es localhost.
# Desde n8n (en Docker) seria host.docker.internal. Mismo Postgres,
# distinta vista de red.
CONN = "postgresql+psycopg://postgres:postgres@localhost:5432/soporte"

def build_retriever(kb_dir=r"..\knowledge"):
    docs = []
    for path in glob.glob(os.path.join(kb_dir, "*.md")):
        with open(path, encoding="utf-8") as f:
            docs.append(Document(page_content=f.read(),
                                 metadata={"source": os.path.basename(path)}))
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(docs)

    store = PGVector.from_documents(          # <-- ingesta: embebe + inserta
        documents=chunks,
        embedding=OpenAIEmbeddings(),
        collection_name="soporte_kb",
        connection=CONN,
        use_jsonb=True,
    )
    return store.as_retriever(search_kwargs={"k": 3})

El resto del agente (agent.py, server.py) no cambia: sigue usando build_retriever(). Solo cambiaste donde viven los vectores.

Esto crea automaticamente dos tablas: langchain_pg_collection (las colecciones) y langchain_pg_embedding (los chunks + su vector).

8B.4 El indice ANN (para que la busqueda sea rapida)

Cargar los vectores no es lo mismo que indexarlos para busqueda veloz. Crea el indice aproximado sobre la columna embedding:

-- HNSW: mejor recall y velocidad de consulta (recomendado por defecto)
docker exec -it pgvector psql -U postgres -d soporte -c \
  "CREATE INDEX IF NOT EXISTS idx_kb_hnsw ON langchain_pg_embedding USING hnsw (embedding vector_cosine_ops);"

-- IVFFlat: build mas barato, PERO requiere datos ya cargados y elegir 'lists'
-- CREATE INDEX ON langchain_pg_embedding USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

• vector_cosine_ops = distancia coseno, la tipica para embeddings de OpenAI. Tambien existen vector_l2_ops (euclidea) y vector_ip_ops (producto interno).
• HNSW vs IVFFlat: HNSW da mejor calidad y no necesita datos previos (mas memoria, build mas lento). IVFFlat es mas liviano pero hay que cargar datos antes y ajustar lists.

Verificar que quedo todo:

docker exec -it pgvector psql -U postgres -d soporte -c "\d langchain_pg_embedding"
docker exec -it pgvector psql -U postgres -d soporte -c "SELECT count(*) FROM langchain_pg_embedding;"

8B.5 Puntos clave

• "pgvector convierte Postgres en base vectorial con el tipo vector y operadores de distancia. Lo elijo cuando la empresa ya tiene Postgres: una sola base para datos y embeddings, menos infra que mantener."
• "Indexar son dos pasos: la ingesta (embeber e insertar) y el indice ANN (HNSW/IVFFlat) para busqueda rapida. Sin indice es scan secuencial."
• "Uso distancia coseno porque los embeddings de OpenAI estan pensados para similitud coseno."
• "HNSW por defecto por recall; IVFFlat si el build tiene que ser barato y ya tengo los datos."

9. Fase 5 — Lab B: el agente en LangGraph (Python) TU

Ahora el mismo agente, pero como codigo. Esto demuestra que dominas LangGraph, no solo arrastrar nodos.

9.1 Preparar el entorno Python

cd c:\prueba-agente\langgraph-agent
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip

requirements.txt

Crea el archivo requirements.txt con:

langgraph
langchain
langchain-openai
langchain-community
langchain-core
fastapi
uvicorn
python-dotenv

pip install -r requirements.txt

.env

Crea .env (mismo folder) con tus keys:

OPENAI_API_KEY=sk-tu-key-aca
# Para la Fase 7 (dejalas listas):
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2-tu-key-aca
LANGSMITH_PROJECT=soporte-interno

Y un .gitignore con .env y .venv/ dentro.

9.2 Ingesta del conocimiento — ingest.py

Este script lee los .md de knowledge\, los parte en chunks, los vectoriza y arma un retriever en memoria que reutilizamos.

import os, glob
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from dotenv import load_dotenv

load_dotenv()

def build_retriever(kb_dir=r"..\knowledge"):
    docs = []
    for path in glob.glob(os.path.join(kb_dir, "*.md")):
        with open(path, encoding="utf-8") as f:
            docs.append(Document(page_content=f.read(),
                                 metadata={"source": os.path.basename(path)}))
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(docs)
    store = InMemoryVectorStore.from_documents(chunks, OpenAIEmbeddings())
    return store.as_retriever(search_kwargs={"k": 3})

Nota: RecursiveCharacterTextSplitter vive en el paquete langchain-text-splitters, que se instala junto con langchain. Si faltara: pip install langchain-text-splitters.

9.3 El agente como grafo — agent.py

Un grafo LangGraph con estado y una rama de decision: recupera contexto, genera respuesta, y si el modelo no esta seguro, marca escalar.

from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from ingest import build_retriever

retriever = build_retriever()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class State(TypedDict):
    question: str
    context: str
    answer: str
    escalate: bool

def retrieve(state: State):
    docs = retriever.invoke(state["question"])
    ctx = "\n\n".join(f"[{d.metadata['source']}] {d.page_content}" for d in docs)
    return {"context": ctx}

def generate(state: State):
    system = SystemMessage(content=(
        "Sos el soporte interno de ACME. Responde SOLO con el contexto dado. "
        "Si el contexto no alcanza, responde exactamente 'ESCALAR' y nada mas."))
    user = HumanMessage(content=(
        f"Contexto:\n{state['context']}\n\nPregunta: {state['question']}"))
    resp = llm.invoke([system, user]).content
    return {"answer": resp, "escalate": resp.strip().upper().startswith("ESCALAR")}

def escalate(state: State):
    return {"answer": ("No tengo esa informacion en la base. "
                       "Abri un ticket para un agente humano: [TICKET-CREADO]")}

def route(state: State):
    return "escalate" if state["escalate"] else END

graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_node("escalate", escalate)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_conditional_edges("generate", route, {"escalate": "escalate", END: END})
graph.add_edge("escalate", END)

app = graph.compile()

if __name__ == "__main__":
    out = app.invoke({"question": "cuantos dias de vacaciones tengo?"})
    print(out["answer"])

Probalo directo:

python agent.py

Deberia responder con el dato de tu vacaciones.md. Preguntas fuera de la KB disparan la rama escalate. Eso es un agente con estado y control de flujo: justo lo que evalua LangGraph.

9.4 Exponerlo como API — server.py

from fastapi import FastAPI
from pydantic import BaseModel
from agent import app as agent_app

api = FastAPI()

class Query(BaseModel):
    question: str

@api.post("/chat")
def chat(q: Query):
    result = agent_app.invoke({"question": q.question})
    return {"answer": result["answer"], "escalated": result.get("escalate", False)}

uvicorn server:api --reload --port 8000

Probalo desde otra terminal PowerShell:

curl.exe -X POST http://localhost:8000/chat -H "Content-Type: application/json" -d "{\"question\":\"como pido vacaciones?\"}"

10. Fase 6 — Conectar LangGraph con n8n TU

Ahora n8n es la puerta y LangGraph el cerebro. En un workflow nuevo ("Soporte Lab B"):

1. Chat Trigger (o Webhook) como entrada.
2. Nodo HTTP Request:
   • Metodo: POST
   • URL: http://host.docker.internal:8000/chat
   • Body (JSON): { "question": "{{ $json.chatInput }}" }
3. Devolve {{ $json.answer }} como respuesta del chat.

11. Fase 7 — Observabilidad con LangSmith TU

Ya pusiste las variables LANGSMITH_* en el .env. Con LANGSMITH_TRACING=true, cada corrida del agente aparece en smith.langchain.com: veras cada paso del grafo, que documentos recupero, que prompt mando al LLM, cuantos tokens gasto y cuanto tardo.

Corre python agent.py otra vez y abri LangSmith: vas a ver la traza. Saber depurar un agente mirando la traza es una habilidad clave para operarlo en produccion.

11B. Escalado: performance para volumen y concurrencia corporativa TEORIA

El demo responde a un usuario. Una plataforma corporativa responde a cientos en paralelo, todo el dia, sin caerse ni fundir el presupuesto. El principio rector:

11B.0 Mapa de cuellos de botella

11B.1 n8n — del proceso unico al modo cola

Por defecto n8n corre en regular mode (un solo proceso, base SQLite): no escala. Para volumen se pasa a queue mode: un proceso main reparte las ejecuciones por Redis a varios procesos worker que escalan horizontalmente.

# Variables clave (en el/los contenedores n8n)
EXECUTIONS_MODE=queue
QUEUE_BULL_REDIS_HOST=redis
DB_TYPE=postgresdb                 # NUNCA SQLite en produccion (no soporta concurrencia)
DB_POSTGRESDB_HOST=postgres
N8N_CONCURRENCY_PRODUCTION_LIMIT=10 # ejecuciones simultaneas POR worker
N8N_DEFAULT_BINARY_DATA_MODE=filesystem   # binarios fuera de memoria/DB
EXECUTIONS_DATA_PRUNE=true          # que la tabla de ejecuciones no crezca infinito
EXECUTIONS_DATA_MAX_AGE=336         # horas (14 dias)

• Se levantan N procesos worker (comando n8n worker) y se escalan segun la cola. El main solo orquesta.
• Webhooks dedicados: instancias de webhook separadas para que la ingesta de eventos no compita con la ejecucion.
• SQLite es solo para el demo; con concurrencia se bloquea. Postgres es obligatorio.

11B.2 LLM (OpenAI) — el recurso mas caro y con limite duro

• Rate limits (RPM/TPM): el 429 es tu enemigo #1. Retry con backoff exponencial y un semaforo que limite las llamadas concurrentes por debajo del techo de la cuenta.
• Right-sizing: usa el modelo mas barato que pase tus evals (gpt-4o-mini para el 90%); patron router que escala a un modelo grande solo cuando hace falta.
• Prompt caching: cachea el prefijo estatico (system prompt + contexto recuperado) — gran ahorro de costo y latencia a volumen.
• Streaming: baja la latencia percibida y libera la conexion antes.
• Fallback a un modelo/proveedor secundario ante caida.

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0,
                 max_retries=5, timeout=30, streaming=True)
backup = ChatOpenAI(model="gpt-4o", max_retries=3)
llm = llm.with_fallbacks([backup])       # si el primario falla, usa el backup

11B.3 Embeddings — batch y no recomputar

• Batch en la ingesta: mandar muchos textos por request, no de a uno.
• Cache de embeddings (CacheBackedEmbeddings): no re-embeber chunks identicos.
• Ingesta incremental: hashear cada doc y re-embeber solo lo que cambio.
• Dimension menor si el recall lo permite (text-embedding-3-small 1536 dims < -large 3072): busqueda mas rapida y menos almacenamiento.

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
from langchain_openai import OpenAIEmbeddings

store = LocalFileStore("./emb_cache")
embeddings = CacheBackedEmbeddings.from_bytes_store(
    OpenAIEmbeddings(), store, namespace="soporte")

11B.4 Postgres / pgvector — indice y pooling

• Indice HNSW ajustado: m y ef_construction en el build; hnsw.ef_search en consulta (sube recall a costa de latencia).
• Connection pooling con PgBouncer (o pool de SQLAlchemy): abrir conexiones a Postgres es caro; hay que reutilizarlas.
• Recursos: maintenance_work_mem alto para construir el indice, work_mem/shared_buffers para consultas.
• Filtro por metadata en SQL (por ejemplo tenant/area) para achicar el set candidato; particionado o coleccion por tenant en multi-inquilino.
• Cuantizacion (halfvec / binaria en pgvector reciente) para cortar almacenamiento y acelerar. Read replicas si el retrieval domina.

-- Build ajustado y parametro de consulta
CREATE INDEX ON langchain_pg_embedding
  USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);
SET hnsw.ef_search = 40;   -- por sesion: mas alto = mas recall, mas latencia

11B.5 Servicio LangGraph (FastAPI/uvicorn) — async, sin estado, replicado

• Async de punta a punta: async def + await graph.ainvoke(...) + cliente LLM/DB async. Un endpoint sincrono serializa TODAS las peticiones — mata la concurrencia.
• Varios workers detras de un proxy: Gunicorn con workers uvicorn.
• Sin estado en el proceso: el checkpointer de LangGraph en Postgres (o Redis), no en memoria — asi cualquier replica atiende cualquier conversacion y escalas horizontalmente.
• Clientes singleton: crear retriever/vectorstore/LLM una vez al arranque, no por request. Semaforo de backpressure para no reventar los rate limits aguas abajo.
• Timeouts, reintentos y circuit breaker en cada llamada externa; health-checks y autoscaling (HPA en Kubernetes por CPU o profundidad de cola).

# estado externo -> cualquier replica sirve cualquier hilo
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(CONN)
app = graph.compile(checkpointer=checkpointer)

# arranque multi-worker
# gunicorn server:api -k uvicorn.workers.UvicornWorker -w 4 --timeout 60

11B.6 Caching transversal — la palanca de mayor impacto

En soporte interno muchas preguntas se repiten ("como reseteo la clave", "como pido vacaciones"). Una cache semantica responde esas desde Redis sin tocar retrieval ni LLM: recorta costo y latencia de forma dramatica.

from langchain_core.globals import set_llm_cache
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(redis_url="redis://redis:6379",
                                 embedding=embeddings, score_threshold=0.1))

Capas de cache utiles: respuesta del LLM (semantica), resultado del retrieval, y los embeddings (11B.3).

11B.7 Colas, observabilidad y costo

• Desacoplar ingesta de servicio: la carga/re-indexado de KB va por una cola de mensajes (RabbitMQ como bus durable, o Celery sobre Redis/RabbitMQ), nunca en el camino de respuesta. (Ojo: la cola de n8n es Redis/Bull, cosa distinta — ver 11B.11.)
• Tickets no urgentes async: encolar y responder por callback.
• Muestreo de trazas: a alto volumen no se traza el 100% en LangSmith/Langfuse — se muestrea.
• Metricas que importan: latencia p50/p95/p99, tokens y costo por request, cache hit rate, tasa de 429/timeout, profundidad de cola.
• Controles de costo: alertas de presupuesto y cuotas por usuario/tenant.

11B.8 Resiliencia y correccion bajo carga

• Idempotencia: claves de idempotencia para que un reintento no cree el ticket dos veces.
• Rate limiting por usuario/tenant para que un abusador no tire el servicio.
• Degradacion elegante: si el LLM esta caido, caer a busqueda-solo o a "escalar a humano" en vez de fallar.
• Cola de human-in-the-loop que no bloquee a los workers.

11B.9 Como dimensionar (capacity planning)

No se adivina, se calcula. Necesitas tres numeros: (1) el techo de RPM/TPM de tu cuenta LLM, (2) tu objetivo de usuarios concurrentes, y (3) tu presupuesto de latencia p95. Con eso: workers ~= concurrencia objetivo / throughput por worker, y el pool de Postgres y el semaforo del LLM se dimensionan para no pasar el techo de cada recurso. Despues se valida con una prueba de carga (k6/Locust) y se mueve el cuello.

11B.10 Puntos clave

• "Escalo por capas: n8n a queue mode con workers y Postgres; el servicio del agente async, sin estado y replicado con checkpointer en Postgres; y protejo el LLM con retry, semaforo y cache."
• "La palanca de mayor ROI en soporte es la cache semantica: muchas preguntas se repiten; contestarlas desde Redis recorta costo y latencia sin tocar el LLM."
• "El estado va afuera del proceso (checkpointer en Postgres) para poder escalar horizontal: cualquier replica atiende cualquier conversacion."
• "Mido p95 y costo por request, y hago capacity planning contra el techo de RPM/TPM antes de prometer un SLA."

11B.11 Diagrama de arquitectura escalable de referencia

Todo lo anterior, junto. Esta es la forma "de produccion" del sistema: cada mejora de las secciones 11B.1–11B.8 ubicada en su lugar. Es el diagrama que conviene poder dibujar de memoria en la pizarra.

Leyenda — donde se aplica cada mejora

12. Ruta rapida: un primer agente en ~90 minutos

Un orden de construccion sugerido para tener un primer agente funcionando rapido. Objetivo: ~90 minutos.

13. Puntos clave (resumen conceptual)

• Que es RAG y por que: "Recupero contexto de docs internos vectorizados y se lo doy al modelo para que responda con la fuente, no de memoria. Reduce alucinaciones y se actualiza cambiando documentos, sin reentrenar."
• LangChain vs LangGraph: "LangChain son las piezas; LangGraph orquesta un agente con estado, ciclos y ramas — para decidir entre responder y escalar."
• Rol de n8n: "La capa de integracion y disparo: recibe el evento (chat, email, webhook), llama al agente y conecta con Slack, tickets o base de datos."
• Como evito alucinaciones: "Instruyo responder solo con el contexto; si no alcanza, escala. Y lo verifico con las trazas de LangSmith."
• Human-in-the-loop: "Para acciones sensibles, el grafo se interrumpe y pide aprobacion humana antes de ejecutar."
• Como escalaria a produccion: "Vector store persistente (Qdrant/pgvector), evaluaciones automaticas, control de costos con un modelo mas chico para el retrieval, y permisos por rol en las tools."

14. Apendice A — Ollama (LLM local gratis) TU

Si preferis practicar sin gastar en OpenAI:

winget install Ollama.Ollama
ollama pull llama3.1
ollama pull nomic-embed-text   # para embeddings

En Python cambias dos lineas:

pip install langchain-ollama

# en vez de ChatOpenAI:
from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)

# en vez de OpenAIEmbeddings:
from langchain_ollama import OllamaEmbeddings
embeddings = OllamaEmbeddings(model="nomic-embed-text")

En n8n hay nodo "Ollama Chat Model". Como n8n esta en Docker, la URL de Ollama es http://host.docker.internal:11434.

15. Apendice B — Problemas frecuentes

16. Apendice C — RAG solo con Redis (alternativa a pgvector) TU

Se puede armar el RAG completo usando Redis como base vectorial, sin Postgres. Aca Redis actua como fuente de la verdad (no como cache). La pieza que lo habilita es el motor de busqueda vectorial: en Redis 8 viene integrado en el core (antes era el modulo RediSearch dentro de Redis Stack).

C.1 Ingesta: analizar, embeber y persistir en Redis

El analisis (parsing) y el chunking son iguales a cualquier RAG (chunks de ~500/50). Los embeddings los genera el LLM/API; Redis solo los guarda e indexa. Cada chunk es un key de Redis (Hash), con el vector como bytes float32:

import redis, numpy as np
r = redis.Redis()

# un key por chunk: doc:{docId}:{chunkId}
r.hset("doc:vac:001", mapping={
    "content":   chunk_text,            # el texto, para devolverlo al LLM
    "source":    "vacaciones.md",       # metadata: cita / gestion
    "tenant":    "acme",                # metadata: filtro
    "embedding": np.array(vec, dtype=np.float32).tobytes(),  # el vector
})
# indice inverso para poder actualizar/borrar el documento entero:
r.sadd("chunks:vac", "doc:vac:001")

Una sola vez se crea el indice vectorial sobre el prefijo doc::

FT.CREATE idx:kb ON HASH PREFIX 1 doc:
  SCHEMA
    content   TEXT
    source    TAG
    tenant    TAG
    embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE

Usa HNSW (ANN, para volumen) o FLAT (fuerza bruta exacta, para pocos datos). COSINE es la metrica de los embeddings de OpenAI, DIM 1536 la de text-embedding-3-small.

C.2 Consulta: KNN con filtro hibrido

Embebes la pregunta y pasas el vector como parametro; Redis devuelve los top-k chunks con su texto, metadata y score:

FT.SEARCH idx:kb "(@tenant:{acme})=>[KNN 3 @embedding $BLOB AS score]"
  PARAMS 2 BLOB $query_bytes
  SORTBY score RETURN 2 content source DIALECT 2

El (@tenant:{acme})=>[KNN ...] es busqueda hibrida: primero filtra por metadata, despues hace el KNN. Ese resultado va al LLM como contexto.

C.3 Atajo con LangChain

pip install langchain-redis

from langchain_redis import RedisVectorStore, RedisConfig
from langchain_openai import OpenAIEmbeddings

config = RedisConfig(index_name="kb", redis_url="redis://localhost:6379")
vs = RedisVectorStore.from_documents(chunks, OpenAIEmbeddings(), config=config)
retriever = vs.as_retriever(search_kwargs={"k": 3})

El resto del agente (agent.py, server.py) no cambia: solo cambia de donde build_retriever() saca los vectores.

C.4 Que asume Redis para ser fuente de la verdad

• Persistencia en disco (sin esto, un reinicio borra todo): appendonly yes (AOF) + snapshots RDB. Al reiniciar, Redis recarga los keys y reconstruye el indice desde los hashes.
• Ciclo de vida por documento: el SET chunks:{docId} lista los keys de sus chunks. Actualizar = borrar los viejos (los del SET) e insertar los nuevos. Para ingesta incremental, un docmeta:{docId} con el hash del contenido.
• Alta disponibilidad y escala: replicas (replicaof) para HA; Redis Cluster para repartir el keyspace.
• El limite real es la RAM: todo (vectores + indice) vive en memoria. Un vector 1536 float32 = ~6 KB, mas el overhead del grafo HNSW (~1.5-2x). Con millones de chunks, la RAM es el cuello y el costo. Mitigacion: FLOAT16/BFLOAT16 o la compresion SVS-VAMANA de Redis 8.

C.5 Redis vs pgvector: cuando cada uno

Nota: los nombres exactos (Redis 8 core vs Redis Stack, el paquete langchain-redis, DIALECT 2) son estables a mi corte, pero la API vectorial de Redis evoluciono rapido — confirmalos contra la doc oficial al montarlo.

Fin de la guia. Cualquier error o paso que no corra, anotalo y se resuelve: la depuracion tambien es parte de dominar el stack.