LangChain Stockage inter-sessions — Store
Checkpointer résout le problème de la « mémoire au sein d'une seule conversation ». Mais si vous devez partager des données entre différentes conversations — comme les préférences utilisateur ou la progression d'apprentissage — vous devez utiliserStore。
Checkpointer vs Store
| Dimension | Checkpointer | Store |
|---|---|---|
| Portée | Fil de conversation unique (thread_id) | À travers tous les fils de conversation |
| Type de données | Instantané de l'état de l'agent (géré automatiquement) | Données clé-valeur arbitraires (gérées manuellement) |
| Cas d'utilisation typiques | Mémoire de conversation multi-tours | Préférences utilisateur, base de connaissances, configuration |
| Organisation des données | thread_id → checkpoint | (namespace, key) → value |
Opérations de base du Store
Utilisation du StoreNamespace + keypour organiser les données dans une structure hiérarchique :
Exemple
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Écrire des données : put(namespace, key, value)
# namespace est un tuple, key est une chaîne de caractères, value est un dictionnaire
store.put(
("users", "user_001"), # Espace de noms
"preferences", # Clé
{ # Valeur
"theme": "dark",
"language": "zh-CN",
"level": "Débutant",
}
)
store.put(
("users", "user_001"),
"progress",
{
"completed_courses": ["HTML Basics", "Python Basics"],
"total_hours": 35,
}
)
# Lire les données : get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"Préférences : {prefs.value}")
progress = store.get(("users", "user_001"), "progress")
print(f"Progression d'apprentissage : {progress.value}")
# Rechercher des données : search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n"Toutes les données utilisateur ({len(all_user_data)} éléments) :")
for item in all_user_data:
print(f" {item.key}: {item.value}")
# Supprimer les données : delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n"Après suppression : {deleted}") # None
store = InMemoryStore()
# Écrire des données : put(namespace, key, value)
# namespace est un tuple, key est une chaîne de caractères, value est un dictionnaire
store.put(
("users", "user_001"), # Espace de noms
"preferences", # Clé
{ # Valeur
"theme": "dark",
"language": "zh-CN",
"level": "Débutant",
}
)
store.put(
("users", "user_001"),
"progress",
{
"completed_courses": ["HTML Basics", "Python Basics"],
"total_hours": 35,
}
)
# Lire les données : get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"Préférences : {prefs.value}")
progress = store.get(("users", "user_001"), "progress")
print(f"Progression d'apprentissage : {progress.value}")
# Rechercher des données : search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n"Toutes les données utilisateur ({len(all_user_data)} éléments) :")
for item in all_user_data:
print(f" {item.key}: {item.value}")
# Supprimer les données : delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n"Après suppression : {deleted}") # None
Sortie :
Preferences: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
Learning progress: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}
用户的所有数据 (2 项):
preferences: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
progress: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}
After deletion: None
Utilisation du Store dans un Agent
Passer Store à create_agent(), et tous les outils de l'Agent peuvent y accéder via InjectedStore :
Exemple
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
# Créer un Store et prédéfinir des données
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
"Tutoriel des bases de Python3": {"price": "Gratuit", "hours": 20, "level": "Débutant"},
"Analyse de données avec Python": {"price": "Membre", "hours": 30, "level": "Avancé"},
"Java orienté objet": {"price": "Gratuit", "hours": 25, "level": "Avancé"},
})
store.put(("example", "users"), "user_vip_001", {
"name": "Xiao Ming",
"membership": "VIP",
"inscrit": "2024-01-15",
})
@tool
def query_course_info(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Interroger des informations détaillées sur les cours du site de tutoriel EXAMPLE.
Arguments :
course_name: Nom du cours
"""
item = store.get(("example", "courses"), "catalog")
catalog = item.value if item else {}
if course_name in catalog:
info = catalog[course_name]
return (
f"《{course_name}》 - Prix: {info['price']},"
f"Durée: {info['hours']} heures, Difficulté: {info['level']}"
)
return f"Cours 《{course_name}》 introuvable"
@tool
def get_user_membership(
user_id: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Interroger les informations d'adhésion de l'utilisateur.
Arguments :
user_id: ID utilisateur
"""
item = store.get(("example", "users"), user_id)
if item is None:
return f"Utilisateur {user_id} introuvable"
user = item.value
return (
f"Utilisateur {user['name']}, membre {user['membership']},"
f"Date d'inscription {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="Vous êtes le consultant de cours pour EXAMPLE.",
)
# Interroger les informations du cours (les données proviennent du Store)
result = agent.invoke({
"messages": [HumanMessage(content="Combien coûte le Python3 Basics Tutorial ?")]
})
print(f"Requête de cours : {result['messages'][-1].content}")
# Interroger les informations utilisateur (les données proviennent du Store)
result = agent.invoke({
"messages": [HumanMessage(content="Aidez-moi à rechercher les informations de l'utilisateur user_vip_001")]
})
print(f"Requête utilisateur : {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
# Créer un Store et prédéfinir des données
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
"Tutoriel des bases de Python3": {"price": "Gratuit", "hours": 20, "level": "Débutant"},
"Analyse de données avec Python": {"price": "Membre", "hours": 30, "level": "Avancé"},
"Java orienté objet": {"price": "Gratuit", "hours": 25, "level": "Avancé"},
})
store.put(("example", "users"), "user_vip_001", {
"name": "Xiao Ming",
"membership": "VIP",
"inscrit": "2024-01-15",
})
@tool
def query_course_info(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Interroger des informations détaillées sur les cours du site de tutoriel EXAMPLE.
Arguments :
course_name: Nom du cours
"""
item = store.get(("example", "courses"), "catalog")
catalog = item.value if item else {}
if course_name in catalog:
info = catalog[course_name]
return (
f"《{course_name}》 - Prix: {info['price']},"
f"Durée: {info['hours']} heures, Difficulté: {info['level']}"
)
return f"Cours 《{course_name}》 introuvable"
@tool
def get_user_membership(
user_id: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Interroger les informations d'adhésion de l'utilisateur.
Arguments :
user_id: ID utilisateur
"""
item = store.get(("example", "users"), user_id)
if item is None:
return f"Utilisateur {user_id} introuvable"
user = item.value
return (
f"Utilisateur {user['name']}, membre {user['membership']},"
f"Date d'inscription {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="Vous êtes le consultant de cours pour EXAMPLE.",
)
# Interroger les informations du cours (les données proviennent du Store)
result = agent.invoke({
"messages": [HumanMessage(content="Combien coûte le Python3 Basics Tutorial ?")]
})
print(f"Requête de cours : {result['messages'][-1].content}")
# Interroger les informations utilisateur (les données proviennent du Store)
result = agent.invoke({
"messages": [HumanMessage(content="Aidez-moi à rechercher les informations de l'utilisateur user_vip_001")]
})
print(f"Requête utilisateur : {result['messages'][-1].content}")
Sortie :
Course query: 《Python3 基础教程》是免费的，学习时长约20小时，难度为入门级别。 User query: 用户小明是 VIP 会员，注册日期为 2024年1月15日。
Persistance du Store
Les données d'InMemoryStore sont perdues après un redémarrage du programme. Pour les environnements de production, vous pouvez utiliser des solutions persistantes telles que PostgresStore :
Exemple
# Étape de développement
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Environnement de production (nécessite PostgreSQL)
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Environnement de production (nécessite PostgreSQL)
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")
Recommandations d'utilisation du Store
| Scénario | exemple de namespace | exemple de clé | Description |
|---|---|---|---|
| Préférences utilisateur | ("users", user_id) | preferences | Thème, langue, paramètres de notification |
| Progression d'apprentissage | ("users", user_id) | progression | Cours terminés, durée d'étude |
| Base de connaissances | ("kb", collection) | doc_id | Documents, FAQ, informations produit |
| Résumés de conversation | ("sessions", thread_id) | summary | Résumés de longues conversations, pour une utilisation dans des scénarios au-delà de Checkpointer |
Autres extensionsCheckpointer est responsable de « où se trouve la conversation », et Store est responsable de « qui est l'utilisateur, ce qu'il sait et ce qu'il aime ». Ce n'est qu'en utilisant les deux ensemble que vous pouvez construire un Agent intelligent doté d'une mémoire persistante.