Memoria entre hilos para los agentes de retail: PostgresStore, el cliente que vuelve y lo que no se debe recordar

Índice

TL;DR

El checkpointer es memoria de un hilo; el almacén es memoria entre hilos. Un cliente que vuelve mañana abre un hilo nuevo y el agente no sabe nada de ayer salvo lo que alguien haya puesto en el almacén. PostgresStore es ese almacén: una tabla store con espacio de nombres, clave y valor JSONB, y opcionalmente una tabla store_vectors con embeddings para buscar por significado.

El almacén llega al agente por store= en create_agent y se lee desde las herramientas y desde el prompt dinámico por runtime.store. Verificado: un dato escrito en un hilo por el cliente C9 aparece en otro hilo del mismo cliente y no aparece en un hilo de C10. El aislamiento es el espacio de nombres, ("clientes", cliente_id), y el identificador lo pone el contexto de ejecución, no el modelo.

Lo que se guarda lo decide la aplicación. Las preferencias que el cliente declara sí, por una herramienta con esquema cerrado. Las que el modelo deduce, no. Los hechos de negocio (última devolución, pedidos abiertos) no se duplican en el almacén: ya están en sus tablas y la herramienta los consulta ahí.

El TTL existe y tiene tres partes: default_ttl en minutos, refresh_on_read que renueva la caducidad al leer, y un barredor que borra lo caducado. El barredor es un hilo dentro del proceso; con tres réplicas son tres barredores. Se desactiva y sweep_ttl() se llama desde el CronJob de retención del artículo 5.

No hay borrado por espacio de nombres. Para cumplir una solicitud de supresión hay que listar las claves del cliente con search y borrarlas una a una con delete. Es un bucle de tres líneas y una consulta SQL de comprobación, y tiene que existir antes de guardar el primer dato.

InMemoryStore no admite TTL: put(..., ttl=...) lanza NotImplementedError. Los tests del almacén con TTL van en el grupo de Postgres, no en el de memoria.

Estás aquí: lo que el hilo no recuerda

Los cuatro agentes de la serie de retail tienen persistencia por hilo. Un hilo es un turno de trabajo del dependiente, una conversación con un cliente sobre un pedido, una pasada de compras o una ficha. Cuando el hilo termina, lo que se aprendió en él se va con la retención.

Hay cosas que deberían quedarse. Un cliente que ha dicho tres veces que prefiere recoger en la tienda de al lado no debería decirlo una cuarta. Un dependiente de la sección de menaje no debería tener que decir “menaje” en cada pregunta. Un proveedor que avisó de un retraso el lunes no debería sorprender al planificador del jueves.

Eso es el almacén, y este artículo lo añade a los agentes 1, 2 y 3. Es el primero de una tanda de dos; el segundo pone un supervisor delante de los cuatro.

La analogía: la ficha de cliente detrás del mostrador

Un buen dependiente lleva en la cabeza a los clientes habituales: la señora que recoge los jueves, el que siempre pide factura. Cuando el dependiente cambia de turno, eso se pierde, y por eso las tiendas que lo hacen bien tienen una ficha de cliente detrás del mostrador con dos o tres líneas escritas a mano. Lo que no tiene la ficha es lo que el sistema ya sabe: los pedidos, las devoluciones, la dirección. Eso se consulta.

El almacén es la ficha. Las tablas de negocio son el sistema. La regla de qué va en cada sitio es la misma que en la tienda: en la ficha, lo que el cliente ha dicho y no está en ningún otro sitio; en el sistema, todo lo demás.

Checkpointer y almacén

CheckpointerAlmacén
UnidadHilo (thread_id)Espacio de nombres (tupla de cadenas) y clave
ContenidoEstado completo del grafo por superstepDocumentos JSONB pequeños
Quién escribeEl motor, en cada superstepHerramientas y aplicación, cuando quieren
BúsquedaPor thread_idPor prefijo de espacio de nombres, filtro sobre el valor y, con índice, por significado
CaducidadRetención por hilo entero, externaTTL por documento, con barredor
Tablacheckpoints y compañíastore y store_vectors

Los dos viven en la base agentes, en tablas distintas, y los dos tienen setup() propio que corre en el mismo Job de migraciones del artículo 5.

Qué se guarda y qué no

La decisión más importante del artículo se toma antes de escribir código. Tres categorías:

Se guarda: lo que el cliente o el empleado declara y no está en ninguna tabla. Preferencia de recogida, canal de contacto preferido, idioma, “no me llaméis por la mañana”, la sección en la que trabaja el dependiente. Son datos que la persona ha dicho y que quiere que se recuerden.

No se guarda: lo que el modelo deduce. “Parece un cliente exigente”, “suele devolver”. Lo primero es una opinión del modelo sobre una persona y no tiene sitio en ningún almacén de una empresa que responda ante el RGPD. Lo segundo es un hecho que ya está en devoluciones y que la herramienta puede contar cuando haga falta, con la fecha y el importe, en vez de como etiqueta.

No se duplica: lo que está en las tablas de negocio. Pedidos, devoluciones, stock, incidencias. El almacén no es una caché de la base de datos; es la ficha escrita a mano.

Para el caso 3 hay una cuarta categoría: avisos de proveedores que no encajan en incidencias porque no son incidencias todavía (“nos han dicho que la fábrica cierra en agosto”). Van al almacén con espacio de nombres ("proveedores", proveedor_id) y los escribe compras desde su aplicación, no el modelo.

El almacén

from psycopg_pool import ConnectionPool
from langgraph.store.postgres import PostgresStore
from langgraph.store.base import TTLConfig

pool_agentes = ConnectionPool(os.environ["PG_AGENTES"], min_size=2, max_size=8,
                              kwargs={"autocommit": True, "prepare_threshold": 0})

almacen = PostgresStore(
    pool_agentes,
    index={"dims": 1024, "embed": emb, "fields": ["texto"]},
    ttl=TTLConfig(default_ttl=60 * 24 * 365, refresh_on_read=True, omit_expired=True, sweep_interval_minutes=None),
)

emb es el mismo OpenAIEmbeddings del caso 1, contra el modelo de embeddings servido en vLLM detrás de LiteLLM, con check_embedding_ctx_length=False. dims tiene que coincidir con la dimensión del modelo; el índice se crea con esa dimensión en setup() y no se cambia después sin recrear la tabla.

fields dice qué campos del valor se indexan. Solo texto. El resto del documento (origen, fecha, quién lo escribió) es metadato y se filtra, no se busca.

El TTL de un año con renovación al leer es la política para preferencias: un cliente que vuelve las mantiene; uno que no vuelve en un año las pierde. sweep_interval_minutes=None desactiva el barredor en proceso; se explica más abajo.

setup() con index configurado aplica las migraciones de la tabla store y además las de store_vectors, empezando por CREATE EXTENSION IF NOT EXISTS vector dentro de un bloque DO. Eso exige que el rol del Job pueda crear extensiones, o que la extensión ya exista en la base. En un cluster de CloudNativePG lo segundo es lo habitual: la extensión se declara en el Cluster y el rol del Job no necesita ser superusuario.

Las herramientas

@tool
def recordar_preferencia(tipo: Literal["recogida", "contacto", "idioma", "horario"], valor: str,
                         runtime: ToolRuntime[ContextoCliente]) -> dict:
    """Guarda una preferencia que el cliente ha declarado explícitamente en esta conversación."""
    ns = ("clientes", runtime.context.cliente_id)
    runtime.store.put(ns, f"pref-{tipo}", {
        "texto": f"{tipo}: {valor}",
        "tipo": tipo,
        "valor": valor,
        "origen": "cliente",
        "hilo": runtime.config["configurable"]["thread_id"],
    })
    return {"ok": True, "tipo": tipo}

@tool
def preferencias(runtime: ToolRuntime[ContextoCliente]) -> list[dict]:
    """Preferencias declaradas por el cliente en conversaciones anteriores."""
    ns = ("clientes", runtime.context.cliente_id)
    return [{"tipo": it.value["tipo"], "valor": it.value["valor"]} for it in runtime.store.search(ns, limit=10)]

Tres decisiones.

La clave es pref-{tipo}, no un identificador nuevo por escritura. Una preferencia nueva del mismo tipo sobreescribe la anterior. Sin eso, “recogida: T002” y “recogida: T001” conviven y el modelo elige.

El tipo es un Literal cerrado. El modelo no puede inventar categorías. Si el cliente dice algo que no encaja, no se guarda. Es la parte del esquema que hace que el almacén no acumule opiniones.

El espacio de nombres viene del contexto. runtime.context.cliente_id lo puso la aplicación a partir de la sesión. El modelo no ve ese parámetro y no puede escribir en la ficha de otro cliente. Verificado: un dato escrito con contexto C9 no aparece en una búsqueda con contexto C10, en otro hilo, con otro agente construido sobre el mismo almacén.

La búsqueda de preferencias no lleva query: devuelve todo el espacio de nombres, que son como mucho cuatro documentos. La búsqueda semántica se usa en el caso 3, donde los avisos de un proveedor pueden ser decenas:

@tool
def avisos_proveedor(consulta: str, runtime: ToolRuntime[ContextoCompras]) -> list[dict]:
    """Avisos recientes de compras sobre el proveedor relacionados con la consulta."""
    ns = ("proveedores", runtime.context.proveedor_id)
    return [
        {"texto": it.value["texto"], "fecha": it.value["fecha"], "similitud": round(it.score, 3)}
        for it in runtime.store.search(ns, query=consulta, limit=3)
        if it.score is not None and it.score > 0.4
    ]

Con el índice configurado, search con query calcula el embedding de la consulta con emb, lo compara por coseno contra store_vectors dentro del prefijo, y devuelve SearchItem con score. Sin query, es un listado por prefijo con filtro opcional sobre el valor.

El prompt que lee el almacén

Para el caso 2 las preferencias no van por herramienta. Van en el prompt, porque el modelo tiene que saberlas antes de proponer nada y no debería tener que acordarse de llamar a preferencias:

@dynamic_prompt
def prompt_cliente(request: ModelRequest) -> str:
    ctx = request.runtime.context
    prefs = request.runtime.store.search(("clientes", ctx.cliente_id), limit=10)
    lineas = "\n".join(f"- {p.value['texto']}" for p in prefs) or "- (ninguna)"
    return (
        f"Atiendes al cliente {ctx.cliente_id} de la tienda {ctx.tienda}.\n"
        f"Preferencias que el cliente ha declarado en otras conversaciones:\n{lineas}\n"
        "Respétalas al proponer recogida, contacto o plazo. No las cites como si las hubieras deducido."
        ...
    )

Verificado que request.runtime.store está disponible en el prompt dinámico y devuelve lo mismo que desde una herramienta. Es una consulta a Postgres por cada llamada al modelo; con refresh_on_read=True, además renueva la caducidad de esas preferencias en cada llamada, que es lo que se quiere para un cliente activo y lo que hay que tener en cuenta al estimar escrituras: cada lectura es también un UPDATE de expires_at.

La última frase del prompt existe porque un modelo que recibe “prefiere recogida en T002” tiende a decir “veo que sueles recoger en T002”, y eso a un cliente le suena a vigilancia. La preferencia se aplica, no se comenta.

El TTL y el barredor

TTLConfig tiene cuatro campos: default_ttl en minutos para lo que se escriba sin ttl explícito, refresh_on_read para renovar al leer, omit_expired para que las lecturas ignoren lo caducado aunque el barredor no haya pasado, y sweep_interval_minutes para el barredor.

El barredor es start_ttl_sweeper(): un hilo dentro del proceso que cada N minutos ejecuta sweep_ttl(), que es un DELETE FROM store WHERE expires_at IS NOT NULL AND expires_at < NOW(). Con tres réplicas del servicio del caso 2, son tres hilos haciendo el mismo DELETE sobre la misma tabla. No rompe nada; es trabajo repetido y una fuente de bloqueos breves que no hace falta tener.

Se deja sweep_interval_minutes=None y el barrido va en el CronJob de retención del artículo 5, una línea más:

borrados = almacen.sweep_ttl()
log.info("store: %d documentos caducados borrados", borrados)

omit_expired=True es lo que hace que entre barrido y barrido lo caducado no se lea. Con False, una preferencia caducada hace once meses seguiría apareciendo hasta la noche.

Los plazos por espacio de nombres: un año con renovación para clientes, 90 días sin renovación para proveedores (un aviso de hace tres meses ya es una incidencia o no fue nada), y 180 días con renovación para empleados. Se pasan como ttl= en cada put, en minutos, y default_ttl es solo la red.

El borrado que la ley pide

Un cliente puede pedir que se borre lo que la empresa sabe de él, y la respuesta tiene que ser completa. El almacén no tiene “borra todo lo de este espacio de nombres”. BaseStore tiene delete(namespace, key) y list_namespaces, y nada más.

def suprimir_cliente(cliente_id: str) -> int:
    ns = ("clientes", cliente_id)
    items = almacen.search(ns, limit=1000)
    for it in items:
        almacen.delete(ns, it.key)
    return len(items)

Y la comprobación, que es la que se guarda como evidencia de la solicitud:

SELECT count(*) FROM store WHERE prefix = 'clientes.' || %s;
SELECT count(*) FROM store_vectors WHERE prefix = 'clientes.' || %s;

El prefijo en la tabla es la tupla unida por puntos. Las dos consultas tienen que dar cero, y las dos van en el registro de la solicitud junto con la fecha.

Esta función existe y está probada antes de que el agente guarde la primera preferencia. Y va acompañada de la del checkpointer: los hilos del cliente en hilos se borran con delete_thread, y las trazas de Langfuse con su API de borrado por user_id, que es otro artículo. Lo que este artículo fija es que el almacén no puede ser el sitio donde un dato sobre una persona se queda sin que nadie sepa borrarlo.

Lo que se prueba y lo que no

Con InMemoryStore se prueba todo el aislamiento y la lectura: el almacén admite index con FakeEmbeddings de langchain-core, y los tests del artículo de pruebas se extienden con tres:

def test_preferencia_sobrevive_al_hilo(store):
    agente_a = build_cliente(ModeloGuion(responses=[llamada("recordar_preferencia", tipo="recogida", valor="T002"), AIMessage(content="ok")]), InMemorySaver(), store)
    agente_a.invoke(entrada("me lo llevo a T002"), context=Ctx("C9", "T001"), config=hilo("h1"))
    agente_b = build_cliente(ModeloGuion(responses=[llamada("preferencias"), AIMessage(content="ok")]), InMemorySaver(), store)
    r = agente_b.invoke(entrada("hola"), context=Ctx("C9", "T003"), config=hilo("h2"))
    assert json.loads(tool_msg(r).content) == [{"tipo": "recogida", "valor": "T002"}]

def test_otro_cliente_no_ve_nada(store):
    ...
    assert json.loads(tool_msg(r).content) == []

def test_sin_almacen_las_herramientas_no_rompen():
    agente = build_cliente(ModeloGuion(responses=[llamada("preferencias"), AIMessage(content="ok")]), InMemorySaver(), store=None)
    r = agente.invoke(entrada("hola"), context=Ctx("C9", "T001"), config=hilo("h1"))
    assert json.loads(tool_msg(r).content) == []

El tercero fija un detalle verificado: sin store= en create_agent, runtime.store es None dentro de la herramienta, y la herramienta tiene que comprobarlo y devolver vacío en vez de lanzar. Es el comportamiento que se quiere si el almacén se despliega después del agente.

Lo que no se prueba en memoria: el TTL. InMemoryStore.put(..., ttl=...) lanza NotImplementedError: TTL is not supported by InMemoryStore. Los tests de caducidad, refresh_on_read, omit_expired y sweep_ttl van en el grupo de Postgres, con un TTL de un minuto y un sleep.

Lo que se mide

MétricaDóndeQué dice
Documentos por clienteSELECT prefix, count(*) FROM store GROUP BY prefixCon el esquema cerrado, cuatro como máximo. Más, el Literal se ha abierto
Preferencias escritas por conversaciónSpans de recordar_preferencia en LangfusePor debajo de 0,1. Un modelo que guarda algo en cada conversación está guardando lo que no debe
Conversaciones del caso 2 en las que se aplicó una preferenciaComparar la propuesta con storeLa utilidad. Si no cambia ninguna propuesta, la memoria no sirve
Caducados borrados por barridoLog del CronJobDebe crecer a partir del mes doce y no antes
Solicitudes de supresión y tiempo hasta ceroRegistro de solicitudesLa obligación. Tiene que ser el mismo día

Checklist

  • Hay una lista escrita de qué se guarda, qué no y qué no se duplica, y el Literal de tipos la refleja.
  • El espacio de nombres viene de runtime.context, nunca de un argumento del modelo.
  • Las claves son por tipo de preferencia y sobreescriben.
  • index usa el modelo de embeddings de la serie y dims coincide.
  • setup() del almacén corre en el Job de migraciones y la extensión vector existe antes.
  • sweep_interval_minutes=None y sweep_ttl() en el CronJob de retención.
  • omit_expired=True.
  • TTL por espacio de nombres pasado en cada put.
  • suprimir_cliente existe, está probada y deja evidencia con las dos consultas a cero.
  • Las herramientas comprueban runtime.store is None.
  • Los tests de TTL están en el grupo de Postgres.

Trampas y cosas que no son lo que parecen

El almacén como caché de la base de negocio. Se desincroniza y se convierte en la fuente que el modelo prefiere. Solo lo que no está en ninguna tabla.

El modelo escribiendo lo que deduce. Un put con texto libre desde una herramienta sin esquema acaba con opiniones sobre personas en Postgres. Literal cerrado y origen cliente.

Tres réplicas, tres barredores. El barredor es un hilo en proceso. Desactivar y barrer desde el CronJob.

refresh_on_read escribe. Cada lectura desde el prompt renueva expires_at. Es un UPDATE por llamada al modelo; se cuenta.

No hay borrado por espacio de nombres. Listar y borrar, y comprobar en SQL. Antes de la primera escritura.

InMemoryStore no tiene TTL. NotImplementedError. Los tests de caducidad van contra Postgres.

setup() crea la extensión. Necesita permiso o la extensión ya creada. En CNPG, se declara en el Cluster.

“Veo que sueles…” El modelo comenta la preferencia y suena a vigilancia. El prompt dice que se aplique y no se cite.

Cierre

La memoria entre hilos es una tabla con espacio de nombres, un esquema cerrado de lo que se puede escribir, un TTL por tipo de dato y una función de borrado que existe antes que el primer put. LangGraph pone la tabla, la búsqueda y la caducidad; la aplicación pone lo demás, y lo demás es lo que decide si la ficha de cliente es una ayuda o un problema.

El siguiente artículo cierra la tanda con un supervisor: un punto de entrada que decide si una conversación es del asistente de sala, del agente de cliente o de compras, y la traspasa con Command(goto=...) sin perder el hilo ni el almacén.

Ver también

Fuentes