Cómo resolver la memoria en LangGraph usando DynamoDB
Cómo persistir la memoria a largo plazo de tus agentes con DynamoDB sin sumar bases relacionales innecesarias a tu stack.
Tu usuario entra a tu app, te cuenta que prefiere respuestas breves en español y cierra la pestaña. Al otro día vuelve, le habla a tu agente y el bot le pregunta todo de nuevo desde cero.
El agente no se acuerda de nada entre sesiones.
El problema es común: el checkpointer de LangGraph solo guarda el estado dentro de un thread de chat. No mantiene la memoria que sobrevive entre sesiones. La receta típica te manda a levantar un Postgres con pgvector. Si tu arquitectura ya corre serverless sobre AWS con Lambda, Fargate y DynamoDB, meter una base relacional solo para esto suma fricción y costos fijos al vicio.
Para resolverlo escribí un BaseStore para LangGraph respaldado por DynamoDB. Se llama langgraph-dynamodb-store y podés ver el código en GitHub.
En este post te muestro cómo funciona por dentro, cómo integrarlo en tu grafo y cuándo te conviene frente a Postgres.
Checkpointer vs. Store: dos formas de memoria en LangGraph
Es común pensar que con el checkpointer ya alcanza para todo. En la práctica no alcanza, porque resuelven problemas distintos:
- Checkpointer: guarda el estado dentro de un thread. Es la memoria de la conversación en curso. Si el usuario abre otra sesión, arranca un thread nuevo y el agente pierde el hilo anterior.
- Store: guarda memoria a largo plazo entre distintos threads. Es la memoria del usuario, no la de la charla puntual.
Un turno del agente con ambos componentes funciona 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 sostiene la charla actual. El store permite que el agente reconozca al usuario cuando vuelve mañana. Como vimos al comparar LangGraph vs Vercel AI SDK, una máquina de estados con persistencia granular resulta clave para manejar flujos complejos.
El mecanismo: namespaces, tipos y TTL
La librería implementa BaseStore de LangGraph (get, put, search, en versiones sincrónicas y asincrónicas). Encaja directo donde el framework espera un store.
El mapeo a DynamoDB resuelve tres puntos clave:
- Namespaces como fronteras de propiedad: el namespace es una tupla que define al dueño del dato. Podés usar
("user", cognito_sub)para memorias privadas o("instance", tenant_id)para datos de todo un tenant. Consultás con la misma tupla que usaste al guardar. - Tipos de memoria con TTL propio: cada dato tiene un ciclo de vida distinto y usa 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 |
La memoria episódica expira sola en DynamoDB. No necesitás programar cron jobs ni tareas de limpieza como en Postgres.
El uso básico es deliberadamente simple:
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 abajo, la tabla usa owner_id como clave de partición y sort_key como rango. Un GSI sobre created_at resuelve el orden temporal y la paginación corre con el LastEvaluatedKey nativo. Todo esto viene resuelto en el diseño base.
Integración con LangGraph
Para sumarlo a tu flujo tenés que hacer dos pasos: pasar el store al compilar el grafo y sumar dos nodos dedicados.
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 busca las N memorias más recientes sin usar embeddings. Es una opción barata que rinde bien para la mayoría de los asistentes.
Para guardar datos, MemoryWriter usa with_structured_output del LLM. Extrae hechos limpios de la charla y los guarda categorizados:
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"],
)
Al extraer información con un esquema evitamos guardar transcripciones enteras que ensucien el contexto futuro. Guardás hechos puntuales, como que el usuario prefiere respuestas breves. Este desacople sigue los criterios que repasamos en patrones de agentes en producción.
DynamoDB vs. Postgres + pgvector: cuándo conviene
Cada opción tiene tradeoffs claros de arquitectura.
Si tu sistema ya corre sobre DynamoDB, este enfoque te ahorra mantener una base extra. Si tu producto depende de búsquedas vectoriales densas con miles de registros por usuario, te conviene evaluar Postgres u otro motor.
| 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 |
Para recordar preferencias y hechos de usuarios por clave y fecha, DynamoDB encaja directo y te saca mantenimiento operativo. Si necesitás calcular similitud semántica sobre miles de notas por usuario, un vector store dedicado te va a dar mejor resultado. La librería ofrece el extra [semantic], pero DynamoDB no reemplaza a un motor como Pinecone.
Qué hacer el lunes
Para implementarlo en tu agente, podés seguir estos pasos:
- Instalá el paquete: corré
pip install langgraph-dynamodb-store(o sumápip install "langgraph-dynamodb-store[semantic]"si vas a usar búsqueda semántica). - Creá la tabla: ejecutá
aws dynamodb create-tableconfigurando particiónowner_id, rangosort_keyy el índiceowner_id-created_at-index. Después habilitáupdate-time-to-liveen el campottl. Tenés el comando listo en el README. - Pasá el store: agregalo en
builder.compile(checkpointer=..., store=store). - Sumá los nodos: usá
SimpleRetrievalantes del LLM para inyectar el contexto yMemoryWriteral final para persistir los hechos nuevos. - Definí los vencimientos: configurá los TTL según tu caso de uso (los eventos episódicos vencen solos y las preferencias semánticas persisten).
El proyecto tiene licencia MIT. Podés usarlo, probarlo y mandar un PR si encontrás mejoras. Si lo sumás a tu stack en producción, me interesa conocer tu experiencia.
Podés revisar el paquete en langgraph-dynamodb-store en PyPI y el código fuente de memory-layer en GitHub.