Memoria para agentes LangGraph sin Postgres: BaseStore sobre DynamoDB
Cómo darle memoria persistente entre sesiones a un agente LangGraph usando DynamoDB en vez de Postgres, con la librería que publiqué.
Si construiste un agente con LangGraph y lo pusiste en producción, tarde o temprano chocás con la misma pared: el agente no se acuerda de nada entre sesiones. El checkpointer de LangGraph persiste el estado dentro de un thread —la conversación de una sesión—, pero no la memoria que tiene que sobrevivir entre sesiones: las preferencias del usuario, lo que ya te contó, cómo le gusta que le respondas. La respuesta de manual para eso es levantar Postgres con pgvector. Pero si tu stack ya vive serverless sobre AWS —Lambda o Fargate y DynamoDB—, montar un Postgres solo para la memoria del agente es fricción pura: otra pieza que aprovisionar, parchear y pagar aunque no la uses.
Me pasó exactamente eso, así que escribí una librería para resolverlo: un BaseStore de LangGraph respaldado por DynamoDB. Se llama langgraph-dynamodb-store (código en GitHub). Este post es sobre el problema que resuelve, cómo funciona por dentro y cuándo conviene —y cuándo no—.
Checkpointer vs. Store: son dos memorias distintas
La confusión más común es creer que el checkpointer alcanza. No alcanza, porque resuelven cosas diferentes:
- Checkpointer: guarda el estado del grafo dentro de un thread. Es la memoria de “esta conversación”. Si el usuario cierra la app y vuelve mañana en otra sesión, ese thread es nuevo y arranca en blanco.
- Store: guarda memoria de largo plazo compartida entre todos los threads de un mismo usuario. Es la memoria de “este usuario”, no la de “esta charla”.
Un turno del agente, con las dos memorias en juego, se ve así:
flowchart TD
A["Mensaje del usuario"] --> B["Grafo LangGraph (thread)"]
B --> C["Nodo: recuperar memoria"]
C -->|"search(namespace)"| S[("DynamoDBStore")]
S -->|"memorias del usuario"| C
C --> D["LLM con bloque de memoria"]
D --> E["Respuesta"]
E --> F["Nodo: escribir memoria"]
F -->|"extract_and_save"| S
B -.->|"estado del thread"| CK[("Checkpointer")]
El checkpointer mantiene el hilo; el store es lo que hace que el agente te reconozca la próxima vez.
El mecanismo: namespaces, tipos y TTL
La librería implementa la interfaz BaseStore de LangGraph (get, put, search, en versión sync y async), así que enchufa donde LangGraph ya espera un store. Lo interesante es cómo mapea eso a DynamoDB sin pedirte nada raro:
- Namespaces como fronteras de propiedad. Un namespace es una tupla que define de quién es la memoria.
("user", cognito_sub)es memoria privada por usuario;("instance", tenant_id), memoria compartida por tenant. La tupla con la que hacésput()tiene que ser la misma con la que después consultás ensearch(). - Tipos de memoria con TTL propio. No toda memoria dura lo mismo, y eso lo modela con el TTL nativo de DynamoDB:
| Tipo | Para qué | TTL por defecto |
|---|---|---|
semantic |
Preferencias y hechos estables | Sin vencimiento |
episodic |
Eventos puntuales del pasado | 90 días |
procedural |
Patrones recurrentes | Sin vencimiento |
Que lo episódico expire solo, sin un cron que limpie, es justo el tipo de cosa que DynamoDB te da gratis y Postgres te hace programar.
El uso básico es deliberadamente aburrido —que es lo que querés en infraestructura—:
from memory_layer import DynamoDBStore
store = DynamoDBStore(table_name="my-app-memories")
namespace = ("user", "user-123")
store.put(namespace, "mem-1", {"content": "Prefiere respuestas en español", "type": "semantic"})
memories = store.search(namespace, limit=10)
Por debajo, la clave de partición es el owner_id y el rango es un sort_key; un GSI por created_at resuelve las consultas por recencia, y la paginación usa el LastEvaluatedKey nativo. Nada de esto lo tocás vos: es el diseño de tabla que la librería espera.
Integración con LangGraph
En la práctica son dos movimientos: pasar el store al compilar el grafo, y agregar dos nodos —uno que lee memoria antes del LLM y otro que la escribe después—.
from memory_layer import DynamoDBStore
from memory_layer.retrieval import SimpleRetrieval
store = DynamoDBStore(table_name="my-app-memories")
graph = builder.compile(checkpointer=checkpointer, store=store)
def supervisor_node(state, config, store):
retrieval = SimpleRetrieval(store, limit=5)
memories = retrieval.fetch(("user", state["context"]["user_id"]))
memory_block = retrieval.to_prompt_block(memories)
# ...inyectás memory_block en el prompt del LLM
SimpleRetrieval trae las N memorias más recientes sin embeddings —barato y suficiente para la mayoría de los casos—. Para escribir, MemoryWriter usa with_structured_output del LLM para extraer hechos de la conversación y persistirlos clasificados por tipo:
from memory_layer.writer import MemoryWriter
writer = MemoryWriter(llm=your_chat_model, store=store)
async def memory_writer_node(state):
await writer.extract_and_save(
namespace=("user", state["context"]["user_id"]),
messages=state["messages"],
session_id=state["context"]["session_id"],
)
El extractor con salida estructurada es la parte que evita el clásico “guardé toda la conversación cruda y ahora mi contexto es un chiquero”: guardás hechos, no transcripciones.
DynamoDB vs. Postgres + pgvector: cuándo conviene
Esto no es una bala de plata; es una decisión de arquitectura. La regla corta: si ya estás en DynamoDB, esto te ahorra una pieza entera de infra; si necesitás búsqueda semántica pesada a gran escala, pensalo dos veces.
| DynamoDB store | Postgres + pgvector | |
|---|---|---|
| Infra a mantener | Ninguna (serverless) | Un Postgres vivo |
| Costo en reposo | Cero (pay-per-request) | Pagás la instancia igual |
| Acceso por clave y recencia | Nativo y rápido (GSI) | Bien |
| Expiración automática | TTL nativo | Cron / job propio |
| Búsqueda vectorial | Opcional ([semantic]), no es su fuerte |
Nativa y potente |
| Filtros ricos / SQL | Limitados, diseñás por clave | Flexibles |
Traducido: para un asistente que necesita recordar preferencias y hechos por usuario, con acceso por clave y recencia, DynamoDB es un encaje natural y una pieza menos que operar. Si tu caso es recuperar por similitud semántica sobre miles de recuerdos por usuario, ahí un vector store dedicado sigue ganando —la librería trae un extra [semantic] para ese camino, pero no le pidas a DynamoDB que sea Pinecone—.
Qué hacer el lunes
Si querés probarlo en un agente real, el camino más corto:
- Instalá:
pip install langgraph-dynamodb-store(opip install "langgraph-dynamodb-store[semantic]"si vas a querer recuperación semántica). - Creá la tabla con su GSI y activá TTL. Es una sola llamada de
aws dynamodb create-tablecon particiónowner_id, rangosort_keyy el índiceowner_id-created_at-index; despuésupdate-time-to-livesobre el atributottl. El comando exacto está en el README. - Enchufá el store al compilar:
builder.compile(checkpointer=..., store=store). - Agregá dos nodos: uno con
SimpleRetrievalque inyecte el bloque de memoria antes del LLM, y otro conMemoryWriterque extraiga y guarde hechos después. - Ajustá los TTL por tipo a tu dominio (lo episódico vence solo; lo semántico queda).
Es MIT, así que úsalo, rompelo y mandá un PR si le falta algo. Si lo probás en producción, me interesa el feedback: para eso lo publiqué.
— Librería: langgraph-dynamodb-store en PyPI · memory-layer en GitHub