LangChain almacenamiento entre sesiones —— Store
Checkpointer resuelve el problema de la "memoria dentro de una sola conversación". Pero si necesitas compartir datos entre diferentes conversaciones —como preferencias del usuario, progreso de aprendizaje— necesitarás usarStore。
Checkpointer vs Store
| Dimensión | Checkpointer | Store |
|---|---|---|
| Alcance | Hilo de conversación individual (thread_id) | A través de todos los hilos de conversación |
| Tipo de datos | Instantánea de estado del Agent (gestión automática) | Datos arbitrarios clave-valor (gestión manual) |
| Usos típicos | Memoria de conversación de múltiples turnos | Preferencias del usuario, base de conocimiento, configuración |
| Organización de datos | thread_id → checkpoint | (namespace, key) → value |
Operaciones básicas de Store
Uso de StoreEspacio de nombres + clavela estructura jerárquica para organizar datos:
Ejemplo
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Escribir datos: put(namespace, key, value)
# namespace es una tupla, key es una cadena, value es un diccionario
store.put(
("users", "user_001"), # Espacio de nombres
"preferences", # Clave
{ # Valor
"theme": "dark",
"language": "zh-CN",
"level": "Principiante",
}
)
store.put(
("users", "user_001"),
"progress",
{
"completed_courses": ["Fundamentos de HTML", "Fundamentos de Python"],
"total_hours": 35,
}
)
# Leer datos: get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"Preferencias: {prefs.value}")
progress = store.get(("users", "user_001"), "progress")
print(f"Progreso de aprendizaje: {progress.value}")
# Buscar datos: search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n"Todos los datos del usuario ({len(all_user_data)} elementos):")
for item in all_user_data:
print(f" {item.key}: {item.value}")
# Eliminar datos: delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n"Después de eliminar: {deleted}") # None
store = InMemoryStore()
# Escribir datos: put(namespace, key, value)
# namespace es una tupla, key es una cadena, value es un diccionario
store.put(
("users", "user_001"), # Espacio de nombres
"preferences", # Clave
{ # Valor
"theme": "dark",
"language": "zh-CN",
"level": "Principiante",
}
)
store.put(
("users", "user_001"),
"progress",
{
"completed_courses": ["Fundamentos de HTML", "Fundamentos de Python"],
"total_hours": 35,
}
)
# Leer datos: get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"Preferencias: {prefs.value}")
progress = store.get(("users", "user_001"), "progress")
print(f"Progreso de aprendizaje: {progress.value}")
# Buscar datos: search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n"Todos los datos del usuario ({len(all_user_data)} elementos):")
for item in all_user_data:
print(f" {item.key}: {item.value}")
# Eliminar datos: delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n"Después de eliminar: {deleted}") # None
Resultado de la ejecución:
Preferencias: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
Progreso de aprendizaje: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}
用户的所有数据 (2 项):
preferences: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
progress: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}
Después de eliminar: None
Usar Store en Agent
Pasa Store a create_agent(), y todas las herramientas en Agent pueden acceder a él a través de InjectedStore:
Ejemplo
from dotenv import load_dotenv
load_dotenv()
from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Crear Store y preestablecer datos
store = InMemoryStore()
store.put(("ejemplo", "courses"), "catalog", {
"Tutorial básico de Python3": {"price": "Gratis", "hours": 20, "level": "Principiante"},
"Análisis de datos con Python": {"price": "Miembro", "hours": 30, "level": "Avanzado"},
"Java orientado a objetos": {"price": "Gratis", "hours": 25, "level": "Avanzado"},
})
store.put(("ejemplo", "users"), "user_vip_001", {
"name": "Xiaoming",
"membership": "VIP",
"joined": "2024-01-15",
})
@tool
def query_course_info(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Consulta la información detallada de los cursos en EJEMPLO.
Args:
course_name: nombre del curso
"""
item = store.get(("ejemplo", "courses"), "catalog")
catalog = item.value if item else {}
if course_name in catalog:
info = catalog[course_name]
return (
f"《{course_name}》- Precio: {info['price']},"
f"Duración: {info['hours']} horas, dificultad: {info['level']}"
)
return f"No se encontró el curso《{course_name}》"
@tool
def get_user_membership(
user_id: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Consulta la información de membresía del usuario.
Args:
user_id: ID de usuario
"""
item = store.get(("ejemplo", "users"), user_id)
if item is None:
return f"No se encontró el usuario {user_id}"
user = item.value
return (
f"Usuario {user['name']}, miembro {user['membership']},"
f"Fecha de registro {user['joined']}"
)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[query_course_info, get_user_membership],
store=store,
system_prompt="Eres el asesor de cursos de EJEMPLO.",
)
# Consultar información del curso (datos de Store)
result = agent.invoke({
"messages": [HumanMessage(content="¿Cuánto cuesta el tutorial básico de Python3?")]
})
print(f"Consultar curso: {result['messages'][-1].content}")
# Consultar información del usuario (datos de Store)
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a consultar la información del usuario user_vip_001")]
})
print(f"Consultar usuario: {result['messages'][-1].content}")
load_dotenv()
from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Crear Store y preestablecer datos
store = InMemoryStore()
store.put(("ejemplo", "courses"), "catalog", {
"Tutorial básico de Python3": {"price": "Gratis", "hours": 20, "level": "Principiante"},
"Análisis de datos con Python": {"price": "Miembro", "hours": 30, "level": "Avanzado"},
"Java orientado a objetos": {"price": "Gratis", "hours": 25, "level": "Avanzado"},
})
store.put(("ejemplo", "users"), "user_vip_001", {
"name": "Xiaoming",
"membership": "VIP",
"joined": "2024-01-15",
})
@tool
def query_course_info(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Consulta la información detallada de los cursos en EJEMPLO.
Args:
course_name: nombre del curso
"""
item = store.get(("ejemplo", "courses"), "catalog")
catalog = item.value if item else {}
if course_name in catalog:
info = catalog[course_name]
return (
f"《{course_name}》- Precio: {info['price']},"
f"Duración: {info['hours']} horas, dificultad: {info['level']}"
)
return f"No se encontró el curso《{course_name}》"
@tool
def get_user_membership(
user_id: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Consulta la información de membresía del usuario.
Args:
user_id: ID de usuario
"""
item = store.get(("ejemplo", "users"), user_id)
if item is None:
return f"No se encontró el usuario {user_id}"
user = item.value
return (
f"Usuario {user['name']}, miembro {user['membership']},"
f"Fecha de registro {user['joined']}"
)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[query_course_info, get_user_membership],
store=store,
system_prompt="Eres el asesor de cursos de EJEMPLO.",
)
# Consultar información del curso (datos de Store)
result = agent.invoke({
"messages": [HumanMessage(content="¿Cuánto cuesta el tutorial básico de Python3?")]
})
print(f"Consultar curso: {result['messages'][-1].content}")
# Consultar información del usuario (datos de Store)
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a consultar la información del usuario user_vip_001")]
})
print(f"Consultar usuario: {result['messages'][-1].content}")
Resultado de la ejecución:
Consulta de curso: 《Python3 基础教程》是免费的，学习时长约20小时，难度为入门级别。 Consulta de usuario: 用户小明是 VIP 会员，注册日期为 2024年1月15日。
Persistencia de Store
Los datos de InMemoryStore se pierden después de reiniciar el programa. En producción se pueden usar soluciones de persistencia como PostgresStore:
Ejemplo
# Fase de desarrollo
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Entorno de producción (requiere PostgreSQL)
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Entorno de producción (requiere PostgreSQL)
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")
Recomendaciones de uso de Store
| Escenario | Ejemplo de namespace | Ejemplo de key | Descripción |
|---|---|---|---|
| Preferencias del usuario | ("users", user_id) | preferences | Tema, idioma, configuración de notificaciones |
| Progreso de aprendizaje | ("users", user_id) | progress | Cursos completados, tiempo de estudio |
| Base de conocimiento | ("kb", collection) | doc_id | Documentos, FAQ, información del producto |
| Resumen de la sesión | ("sessions", thread_id) | summary | Resumen de conversaciones largas, para usar en escenarios fuera de Checkpointer |
Otras extensionesCheckpointer se encarga de "cómo va la conversación", y Store se encarga de "quién es el usuario, qué sabe, qué le gusta". Solo con la cooperación de ambos se puede construir un Agent inteligente con memoria continua.