Long-term Memory: Store y Cross-Session

Descripción de la cápsula

Long-term memory almacena información que persiste entre diferentes conversaciones. A diferencia de los checkpoints (que rastrean una sola ejecución), long-term memory acumula conocimiento a lo largo del tiempo — preferencias del usuario, temas de investigaciones pasadas, patrones aprendidos.

En las cápsulas 03-06 aprendiste a guardar y restaurar el estado de UNA ejecución con checkpointing. Eso resuelve "se cayó el proceso, perdí todo." Pero no resuelve "el usuario vuelve mañana y el agente no recuerda nada." Long-term memory resuelve eso.

Short-term vs Long-term: la distinción fundamental

Son dos mecanismos completamente separados:

El checkpointer responde: "¿Qué pasó en ESTA conversación?" El store responde: "¿Qué sé sobre este usuario de TODAS sus conversaciones?"

Sesión 1 (thread_001):           Sesión 2 (thread_002):
┌─────────────────────┐          ┌─────────────────────┐
│ Checkpoint 1        │          │ Checkpoint 1        │
│ Checkpoint 2        │          │ Checkpoint 2        │
└─────────────────────┘          └─────────────────────┘
         │                                │
         └──────────┬─────────────────────┘
                    ▼
         ┌─────────────────────┐
         │    Long-term Store  │
         │  (preferences,      │
         │   history, patterns)│
         └─────────────────────┘

Los checkpoints son verticales — dentro de una sesión. El store es horizontal — cruza todas las sesiones.

InMemoryStore: el store para desarrollo

LangGraph provee InMemoryStore como implementación en memoria. Perfecto para desarrollo y testing — rápido, sin dependencias, pero se pierde si el proceso muere.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
print(type(store))
# Output esperado:
# <class 'langgraph.store.memory.InMemoryStore'>

Organización por namespaces

Los datos se organizan con namespaces — tuplas de strings que funcionan como directorios:

("users", "mike", "preferences")     → preferencias de mike
("users", "mike", "history")         → historial de mike
("users", "ana", "preferences")      → preferencias de ana (aislada de mike)
("projects", "research-ai", "notes") → notas del proyecto

Operaciones core del Store

put y get

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

namespace = ("users", "mike", "preferences")
store.put(namespace, "sources", {"value": ["academic", "arxiv"], "reason": "Prefers peer-reviewed"})
store.put(namespace, "format", {"value": "bullet_points"})

item = store.get(namespace, "sources")
print(f"Key: {item.key}")
print(f"Value: {item.value}")
print(f"Namespace: {item.namespace}")

missing = store.get(namespace, "nonexistent")
print(f"Inexistente: {missing}")
# Output esperado:
# Key: sources
# Value: {'value': ['academic', 'arxiv'], 'reason': 'Prefers peer-reviewed'}
# Namespace: ('users', 'mike', 'preferences')
# Inexistente: None

put recibe namespace (tupla), key (string), value (dict). Si el key ya existe, sobrescribe. get retorna un Item con .key, .value, .namespace, .created_at, .updated_at, o None si no existe.

search y delete

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
namespace = ("users", "mike", "preferences")
store.put(namespace, "sources", {"value": ["academic"]})
store.put(namespace, "format", {"value": "bullet_points"})
store.put(namespace, "language", {"value": "es"})

items = store.search(namespace)
print(f"Items: {len(items)}")
for item in items:
    print(f"  {item.key}: {item.value}")

store.delete(namespace, "language")
print(f"\nDespués de delete: {len(store.search(namespace))} items")
# Output esperado:
# Items: 3
#   sources: {'value': ['academic']}
#   format: {'value': 'bullet_points'}
#   language: {'value': 'es'}
#
# Después de delete: 2 items

search retorna todos los items en un namespace. Acepta limit y offset para paginación.

3 casos de uso concretos

Caso 1: Recordar fuentes favoritas del usuario

Después de varias investigaciones, el agente detecta que siempre pides más detalle de fuentes académicas. Prioriza arXiv automáticamente.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

def record_source_preference(store, user_id: str, source_type: str):
    namespace = ("users", user_id, "source_preferences")
    existing = store.get(namespace, source_type)
    count = (existing.value.get("count", 0) + 1) if existing else 1
    store.put(namespace, source_type, {"count": count, "source": source_type})

def get_preferred_sources(store, user_id: str) -> list[str]:
    namespace = ("users", user_id, "source_preferences")
    items = store.search(namespace)
    if not items:
        return ["web", "academic", "news"]
    sorted_sources = sorted(items, key=lambda x: x.value["count"], reverse=True)
    return [item.value["source"] for item in sorted_sources]

for _ in range(5):
    record_source_preference(store, "mike", "academic")
for _ in range(2):
    record_source_preference(store, "mike", "web")
record_source_preference(store, "mike", "news")

preferred = get_preferred_sources(store, "mike")
top = store.get(("users", "mike", "source_preferences"), preferred[0])
print(f"Fuentes preferidas: {preferred}")
print(f"Agente: 'Priorizaré {preferred[0]} ({top.value['count']} selecciones).'")
# Output esperado:
# Fuentes preferidas: ['academic', 'web', 'news']
# Agente: 'Priorizaré academic (5 selecciones).'

Caso 2: Acumular un perfil de investigación

Después de varias sesiones, el agente sabe tus áreas de interés y puede sugerir temas relacionados.

from langgraph.store.memory import InMemoryStore
from datetime import datetime

store = InMemoryStore()

def record_research(store, user_id: str, topic: str):
    topics_ns = ("users", user_id, "topics")
    existing = store.get(topics_ns, topic)
    if existing:
        store.put(topics_ns, topic, {"count": existing.value["count"] + 1, "last": datetime.now().isoformat()})
    else:
        store.put(topics_ns, topic, {"count": 1, "last": datetime.now().isoformat()})

def get_profile(store, user_id: str) -> list[tuple[str, int]]:
    topics = store.search(("users", user_id, "topics"))
    return sorted([(t.key, t.value["count"]) for t in topics], key=lambda x: x[1], reverse=True)

for topic in ["RAG techniques", "prompt engineering", "RAG techniques", "AI agents", "RAG techniques", "AI agents"]:
    record_research(store, "mike", topic)

profile = get_profile(store, "mike")
print("Perfil de investigación:")
for topic, count in profile:
    print(f"  {topic}: {count} sesiones")
print(f"\nAgente: 'Tus áreas de interés son {profile[0][0]} y {profile[1][0]}.'")
# Output esperado:
# Perfil de investigación:
#   RAG techniques: 3 sesiones
#   AI agents: 2 sesiones
#   prompt engineering: 1 sesiones
#
# Agente: 'Tus áreas de interés son RAG techniques y AI agents.'

Caso 3: Almacenar preferencias de formato de reporte

El usuario dice "prefiero bullet points." El agente lo recuerda para siempre.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

def save_format_pref(store, user_id: str, key: str, value: str):
    store.put(("users", user_id, "format_prefs"), key, {"value": value})

def get_format_config(store, user_id: str) -> dict:
    defaults = {"structure": "paragraphs", "detail_level": "standard", "max_findings": 5}
    items = store.search(("users", user_id, "format_prefs"))
    user_prefs = {item.key: item.value["value"] for item in items}
    return {**defaults, **user_prefs}

save_format_pref(store, "mike", "structure", "bullet_points")
save_format_pref(store, "mike", "detail_level", "detailed")
save_format_pref(store, "mike", "max_findings", 8)

config = get_format_config(store, "mike")
print(f"Config de formato: {config}")
print(f"Agente: 'Formateando con {config['structure']}, nivel {config['detail_level']}.'")
# Output esperado:
# Config de formato: {'structure': 'bullet_points', 'detail_level': 'detailed', 'max_findings': 8}
# Agente: 'Formateando con bullet_points, nivel detailed.'

Acceder al Store desde un grafo (StateGraph)

En un grafo LangGraph, el store se inyecta automáticamente a los nodos cuando compilas con store=:

from typing import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

class State(TypedDict):
    user_input: str
    response: str

def greet_user(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    user_id = config["configurable"].get("user_id", "anonymous")
    items = store.search(("users", user_id, "preferences"))
    prefs = {item.key: item.value for item in items}

    if prefs:
        pref_summary = ", ".join(f"{k}={v.get('value', v)}" for k, v in prefs.items())
        return {"response": f"¡Bienvenido! Preferencias: {pref_summary}"}
    return {"response": "¡Hola! Primera vez aquí."}

def save_preference(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    user_id = config["configurable"].get("user_id", "anonymous")
    if "bullet" in state["user_input"].lower():
        store.put(("users", user_id, "preferences"), "format", {"value": "bullet_points"})
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("greet", greet_user)
graph_builder.add_node("save_pref", save_preference)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "save_pref")
graph_builder.add_edge("save_pref", END)

my_store = InMemoryStore()
graph = graph_builder.compile(store=my_store)

r1 = graph.invoke(
    {"user_input": "Quiero bullet points", "response": ""},
    config={"configurable": {"user_id": "mike"}},
)
print(f"Sesión 1: {r1['response']}")

r2 = graph.invoke(
    {"user_input": "Investiga AI", "response": ""},
    config={"configurable": {"user_id": "mike"}},
)
print(f"Sesión 2: {r2['response']}")
# Output esperado:
# Sesión 1: ¡Hola! Primera vez aquí.
# Sesión 2: ¡Bienvenido! Preferencias: format={'value': 'bullet_points'}

Los nodos declaran *, store: BaseStore como keyword argument. LangGraph inyecta el store automáticamente. El user_id viene de config["configurable"].

Combinando Store + Checkpointer

El store y el checkpointer son complementarios:

from typing import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langgraph.checkpoint.memory import MemorySaver
import datetime

class State(TypedDict):
    topic: str
    result: str

def research_with_memory(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    user_id = config["configurable"].get("user_id", "anonymous")
    prefs_ns = ("users", user_id, "preferences")
    history_ns = ("users", user_id, "history")

    prefs = {item.key: item.value for item in store.search(prefs_ns)}
    sessions = store.search(history_ns)
    session_num = len(sessions) + 1

    fmt = prefs.get("format", {}).get("value", "paragraphs")
    result = f"Reporte #{session_num} sobre '{state['topic']}' (formato: {fmt})"

    store.put(history_ns, f"session_{session_num}", {
        "topic": state["topic"],
        "timestamp": datetime.datetime.now().isoformat(),
    })
    return {"result": result}

graph_builder = StateGraph(State)
graph_builder.add_node("research", research_with_memory)
graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", END)

store = InMemoryStore()
checkpointer = MemorySaver()
store.put(("users", "mike", "preferences"), "format", {"value": "bullet_points"})

graph = graph_builder.compile(checkpointer=checkpointer, store=store)

for topic in ["RAG techniques", "Prompt engineering", "AI agents"]:
    result = graph.invoke(
        {"topic": topic, "result": ""},
        config={"configurable": {"thread_id": f"session-{topic}", "user_id": "mike"}},
    )
    print(f"  {result['result']}")

print(f"\nHistorial en store:")
for item in store.search(("users", "mike", "history")):
    print(f"  {item.key}: {item.value['topic']}")
# Output esperado:
#   Reporte #1 sobre 'RAG techniques' (formato: bullet_points)
#   Reporte #2 sobre 'Prompt engineering' (formato: bullet_points)
#   Reporte #3 sobre 'AI agents' (formato: bullet_points)
#
# Historial en store:
#   session_1: RAG techniques
#   session_2: Prompt engineering
#   session_3: AI agents

El checkpointer guarda el estado de cada ejecución individual (cada thread_id). El store acumula el historial del usuario a través de todas las ejecuciones.

Store en la Functional API

Con @entrypoint y @task, el store se pasa como parámetro al entrypoint:

from langgraph.func import entrypoint, task
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langgraph.checkpoint.memory import MemorySaver

store = InMemoryStore()
checkpointer = MemorySaver()

@task
def generate_greeting(session_count: int, last_topic: str) -> str:
    if session_count == 0:
        return "¡Hola! Soy tu asistente de investigación."
    return f"¡Bienvenido de vuelta! Llevas {session_count} sesiones. Última: '{last_topic}'."

@entrypoint(checkpointer=checkpointer, store=store)
def research_agent(topic: str, *, store: BaseStore) -> dict:
    config = entrypoint.get_config()
    user_id = config["configurable"].get("user_id", "anonymous")

    history_ns = ("users", user_id, "history")
    prefs_ns = ("users", user_id, "preferences")
    sessions = store.search(history_ns)

    last_topic_item = store.get(prefs_ns, "last_topic")
    last_topic = last_topic_item.value["value"] if last_topic_item else ""

    greeting = generate_greeting(len(sessions), last_topic).result()

    import datetime
    store.put(history_ns, f"session_{len(sessions) + 1}", {
        "topic": topic, "timestamp": datetime.datetime.now().isoformat(),
    })
    store.put(prefs_ns, "last_topic", {"value": topic})

    return {"greeting": greeting, "topic": topic, "session_number": len(sessions) + 1}

for i, topic in enumerate(["RAG", "Prompt engineering", "AI agents"]):
    r = research_agent.invoke(
        topic, config={"configurable": {"thread_id": f"s{i+1}", "user_id": "mike"}},
    )
    print(f"Sesión {r['session_number']}: {r['greeting']}")
# Output esperado:
# Sesión 1: ¡Hola! Soy tu asistente de investigación.
# Sesión 2: ¡Bienvenido de vuelta! Llevas 1 sesiones. Última: 'RAG'.
# Sesión 3: ¡Bienvenido de vuelta! Llevas 2 sesiones. Última: 'Prompt engineering'.

El store se declara como *, store: BaseStore en el @entrypoint. Los @task no reciben el store directamente — pásales los datos que necesiten como argumentos.

Stores de producción

InMemoryStore es para desarrollo. Para producción, necesitas persistencia real.

PostgresStore

from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://user:password@localhost:5432/mydb"
with PostgresStore.from_conn_string(DB_URI) as store:
    store.put(("users", "mike", "preferences"), "format", {"value": "bullet_points"})
    item = store.get(("users", "mike", "preferences"), "format")

La API es idéntica a InMemoryStore. El cambio es una línea. La estrategia recomendada:

import os

if os.getenv("ENVIRONMENT") == "production":
    from langgraph.store.postgres import PostgresStore
    store = PostgresStore.from_conn_string(os.getenv("DATABASE_URL"))
else:
    from langgraph.store.memory import InMemoryStore
    store = InMemoryStore()

Diseño de namespaces

# Nivel usuario
("users", user_id, "preferences")       → formato, fuentes, idioma
("users", user_id, "history")           → historial de sesiones
("users", user_id, "topics")            → temas con frecuencia

# Nivel proyecto
("projects", project_id, "notes")       → notas compartidas

# Nivel global
("system", "templates")                 → plantillas de reportes

Reglas:

• ✅ Nivel más específico posible — ("users", "mike", "preferences") no ("preferences", "mike")
• ✅ Consistencia jerárquica — entidad → instancia → categoría
• ✅ IDs estables, no nombres que cambien
• ❌ No mezclar entidades diferentes en el mismo namespace
• ❌ No más de 4 niveles — se vuelve difícil de navegar

Troubleshooting

Problema 1: "store.get retorna None aunque hice put"

Causa: Namespace diferente. ("users", "mike", "prefs") y ("users", "mike", "preferences") son distintos.

Solución: Define namespaces como constantes:

USER_PREFS = lambda uid: ("users", uid, "preferences")
store.put(USER_PREFS("mike"), "format", {"value": "bullets"})
item = store.get(USER_PREFS("mike"), "format")  # Siempre coincide

Problema 2: "Los datos desaparecen al reiniciar"

Causa: InMemoryStore vive en RAM.

Solución: Serializa a JSON al terminar, carga al iniciar. Para producción, usa PostgresStore. En la cápsula 08 (proyecto) implementarás persistencia a archivo.

Problema 3: "El store no está disponible en mis nodos"

Causa: No pasaste store= al compilar.

Solución:

graph = graph_builder.compile(store=my_store)  # ← no olvides esto

Problema 4: "Conflicto de datos entre usuarios"

Causa: No incluyes user_id en el namespace.

Solución: Siempre usa el user_id:

user_id = config["configurable"]["user_id"]
namespace = ("users", user_id, "preferences")  # Aislamiento por usuario

Ejercicios

Ejercicio 1: CRUD completo con InMemoryStore (Fácil)

Crea un InMemoryStore, guarda 3 items en ("projects", "alpha", "notes"), lee uno específico, lista todos, elimina uno, y verifica que quedan 2.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
ns = ("projects", "alpha", "notes")
store.put(ns, "idea_1", {"text": "Usar RAG para búsqueda interna", "priority": "high"})
store.put(ns, "idea_2", {"text": "Agregar caché de embeddings", "priority": "medium"})
store.put(ns, "idea_3", {"text": "Dashboard de métricas", "priority": "low"})

item = store.get(ns, "idea_2")
print(f"idea_2: {item.value['text']}")

print(f"\nTodos ({len(store.search(ns))}):")
for item in store.search(ns):
    print(f"  {item.key}: {item.value['text']}")

store.delete(ns, "idea_1")
print(f"\nDespués de delete ({len(store.search(ns))}):")
for item in store.search(ns):
    print(f"  {item.key}: {item.value['text']}")
# Output esperado:
# idea_2: Agregar caché de embeddings
#
# Todos (3):
#   idea_1: Usar RAG para búsqueda interna
#   idea_2: Agregar caché de embeddings
#   idea_3: Dashboard de métricas
#
# Después de delete (2):
#   idea_2: Agregar caché de embeddings
#   idea_3: Dashboard de métricas

Ejercicio 2: Detección automática de preferencias (Fácil)

Implementa detect_and_save_preferences(store, user_id, text) que detecta preferencias en texto del usuario: "resumen corto" → detail_level: brief, "fuentes académicas" → preferred_sources: academic, "bullet" → format: bullet_points. Prueba con 3 inputs y muestra las preferencias acumuladas.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

def detect_and_save_preferences(store, user_id: str, text: str):
    ns = ("users", user_id, "auto_prefs")
    t = text.lower()
    if "resumen corto" in t or "breve" in t:
        store.put(ns, "detail_level", {"value": "brief", "from": text[:50]})
    if "académic" in t or "arxiv" in t or "paper" in t:
        store.put(ns, "preferred_sources", {"value": "academic", "from": text[:50]})
    if "bullet" in t or "lista" in t:
        store.put(ns, "format", {"value": "bullet_points", "from": text[:50]})

inputs = [
    "Dame un resumen corto sobre RAG con papers de arxiv",
    "Quiero una lista con bullet points",
    "Investiga en profundidad los avances en AI agents",
]

for text in inputs:
    detect_and_save_preferences(store, "mike", text)

print("Preferencias de mike:")
for item in store.search(("users", "mike", "auto_prefs")):
    print(f"  {item.key}: {item.value['value']}")
# Output esperado:
# Preferencias de mike:
#   detail_level: brief
#   preferred_sources: academic
#   format: bullet_points

Ejercicio 3: Store integrado en un StateGraph (Medio)

Crea un grafo con 3 nodos: load_profile (lee perfil del store), process (simula investigación), update_profile (actualiza perfil). Compila con InMemoryStore. Ejecuta 3 veces con el mismo user_id y muestra cómo el perfil se enriquece.

from typing import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

class State(TypedDict):
    topic: str
    result: str
    session_count: int

def load_profile(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]
    items = store.search(("users", uid, "profile"))
    profile = {item.key: item.value for item in items}
    return {"session_count": profile.get("count", {}).get("value", 0)}

def process(state: State, config: RunnableConfig) -> dict:
    return {"result": f"Investigación #{state['session_count'] + 1} sobre '{state['topic']}' completada."}

def update_profile(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    uid = config["configurable"]["user_id"]
    ns = ("users", uid, "profile")
    store.put(ns, "count", {"value": state["session_count"] + 1})
    store.put(ns, "last_topic", {"value": state["topic"]})

    existing = store.get(ns, "all_topics")
    topics = existing.value["value"] if existing else []
    if state["topic"] not in topics:
        topics.append(state["topic"])
    store.put(ns, "all_topics", {"value": topics})
    return {}

builder = StateGraph(State)
builder.add_node("load", load_profile)
builder.add_node("process", process)
builder.add_node("update", update_profile)
builder.add_edge(START, "load")
builder.add_edge("load", "process")
builder.add_edge("process", "update")
builder.add_edge("update", END)

my_store = InMemoryStore()
graph = builder.compile(store=my_store)

for topic in ["RAG", "Prompt Engineering", "AI Agents"]:
    r = graph.invoke(
        {"topic": topic, "result": "", "session_count": 0},
        config={"configurable": {"user_id": "mike"}},
    )
    print(r["result"])

print(f"\nPerfil final:")
for item in my_store.search(("users", "mike", "profile")):
    print(f"  {item.key}: {item.value}")
# Output esperado:
# Investigación #1 sobre 'RAG' completada.
# Investigación #2 sobre 'Prompt Engineering' completada.
# Investigación #3 sobre 'AI Agents' completada.
#
# Perfil final:
#   count: {'value': 3}
#   last_topic: {'value': 'AI Agents'}
#   all_topics: {'value': ['RAG', 'Prompt Engineering', 'AI Agents']}

Ejercicio 4: Multi-usuario con namespaces aislados (Medio)

Crea un sistema donde 3 usuarios guardan preferencias en el mismo store. Verifica que los datos están aislados — un usuario no ve las preferencias de otro.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

users = {
    "alice": {"format": "bullet_points", "language": "en"},
    "bob": {"format": "paragraphs", "language": "es"},
    "carol": {"format": "detailed", "language": "fr"},
}

for uid, prefs in users.items():
    for key, value in prefs.items():
        store.put(("users", uid, "preferences"), key, {"value": value})

for uid in users:
    items = store.search(("users", uid, "preferences"))
    print(f"{uid}: {', '.join(f'{i.key}={i.value[\"value\"]}' for i in items)}")

alice_lang = store.get(("users", "alice", "preferences"), "language").value["value"]
bob_lang = store.get(("users", "bob", "preferences"), "language").value["value"]
print(f"\nAlice lang={alice_lang}, Bob lang={bob_lang}, diferentes={alice_lang != bob_lang}")
# Output esperado:
# alice: format=bullet_points, language=en
# bob: format=paragraphs, language=es
# carol: format=detailed, language=fr
#
# Alice lang=en, Bob lang=es, diferentes=True

Ejercicio 5: Diseño de namespaces multi-tenant (Avanzado)

Diseña un esquema para organizaciones con múltiples usuarios. Cada org tiene config compartida, cada usuario tiene preferencias. Implementa get_effective_config(store, org_id, user_id) que merge org defaults + user overrides (user wins). Prueba con 2 orgs, 2 usuarios cada una.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

for key, val in [("model", "gpt-4.1"), ("max_sources", 5), ("language", "en")]:
    store.put(("orgs", "acme", "config"), key, {"value": val})

store.put(("orgs", "acme", "users", "alice", "prefs"), "language", {"value": "es"})

for key, val in [("model", "gpt-4.1-mini"), ("max_sources", 3), ("language", "fr")]:
    store.put(("orgs", "globex", "config"), key, {"value": val})

store.put(("orgs", "globex", "users", "carol", "prefs"), "language", {"value": "en"})

def get_effective_config(store, org_id: str, user_id: str) -> dict:
    org_items = store.search(("orgs", org_id, "config"))
    org_config = {item.key: item.value["value"] for item in org_items}

    user_items = store.search(("orgs", org_id, "users", user_id, "prefs"))
    user_prefs = {item.key: item.value["value"] for item in user_items}

    return {**org_config, **user_prefs}

for org, user, desc in [("acme", "alice", "org en, user es"), ("acme", "bob", "no user prefs"), ("globex", "carol", "org fr, user en")]:
    config = get_effective_config(store, org, user)
    print(f"{org}/{user} ({desc}): {config}")
# Output esperado:
# acme/alice (org en, user es): {'model': 'gpt-4.1', 'max_sources': 5, 'language': 'es'}
# acme/bob (no user prefs): {'model': 'gpt-4.1', 'max_sources': 5, 'language': 'en'}
# globex/carol (org fr, user en): {'model': 'gpt-4.1-mini', 'max_sources': 3, 'language': 'en'}

Resumen

En esta cápsula aprendiste:

• Short-term (checkpoints) y long-term (store) son mecanismos diferentes — checkpoints rastrean una ejecución, el store acumula conocimiento entre ejecuciones
• InMemoryStore tiene 4 operaciones (put, get, search, delete) organizadas por namespaces jerárquicos
• 3 casos de uso concretos: fuentes favoritas, perfil de investigación, preferencias de formato — cada uno transforma la experiencia del usuario
• El store se inyecta en nodos al compilar con store= — los nodos declaran *, store: BaseStore
• Store + Checkpointer trabajan juntos: checkpointer para resiliencia, store para inteligencia entre sesiones
• En la Functional API, el store se pasa al @entrypoint y se accede como *, store: BaseStore
• Para producción, PostgresStore reemplaza a InMemoryStore con la misma API
• Diseño de namespaces determina la escalabilidad — usa jerarquías claras y aísla datos entre usuarios

Próxima cápsula: Proyecto Evolutivo v3 — tu Research Agent gana memoria. Cierra la terminal, ábrela de nuevo, y el agente recuerda quién eres.

Recursos adicionales

1. LangGraph Memory Store — Documentación oficial de stores y long-term memory
2. LangGraph Persistence — Checkpointing vs Store
3. LangGraph Functional API — Store en @entrypoint y @task
4. InMemoryStore Reference — API reference completa
5. LangGraph Cross-thread Memory — Memoria entre threads
6. PostgresStore Guide — Migrar a PostgresStore

Módulo 8 — LangChain & LangGraph: From Chains to Agents