Межсессионное хранилище LangChain — Store

Checkpointer решает проблему «памяти в рамках одного диалога». Но если вам нужно обмениваться данными между разными диалогами — например, пользовательскими предпочтениями или прогрессом обучения — вам нужно использоватьStore。

Checkpointer против Store

ИзмерениеCheckpointerStore
Область действияОдин поток диалога (thread_id)По всем потокам диалога
Тип данныхСнимок состояния агента (управляется автоматически)Произвольные данные ключ-значение (управляются вручную)
Типичные случаи использованияПамять многоходового диалогаПользовательские предпочтения, база знаний, конфигурация
Организация данныхthread_id → checkpoint(namespace, key) → value

Основные операции Store

Использование StoreNamespace + keyдля организации данных в иерархической структуре:

Пример

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# Запись данных: put(namespace, key, value)
# namespace — это кортеж, key — строка, value — словарь
store.put(
    ("users", "user_001"),           # Пространство имён
    "preferences",                    # Ключ
    {                                 # Значение
        "theme": "dark",
        "language": "zh-CN",
        "level": "Новичок",
    }
)

store.put(
    ("users", "user_001"),
    "progress",
    {
        "completed_courses": ["HTML Basics", "Python Basics"],
        "total_hours": 35,
    }
)

# Чтение данных: get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"Настройки: {prefs.value}")

progress = store.get(("users", "user_001"), "progress")
print(f"Прогресс обучения: {progress.value}")

# Поиск данных: search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n"Все данные пользователя ({len(all_user_data)} элементов):")
for item in all_user_data:
    print(f" {item.key}: {item.value}")

# Удаление данных: delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n"После удаления: {deleted}")  # None

Вывод:

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

Использование Store в агенте

Передайте Store в create_agent(), и все инструменты в Agent смогут получить к нему доступ через InjectedStore:

Пример

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

# Создайте Store и предварительно заполните данные
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
    "Учебник по основам Python3": {"price": "Бесплатно", "hours": 20, "level": "Начальный"},
    "Анализ данных на Python": {"price": "Для участников", "hours": 30, "level": "Продвинутый"},
    "Объектно-ориентированное программирование на Java": {"price": "Бесплатно", "hours": 25, "level": "Продвинутый"},
})

store.put(("example", "users"), "user_vip_001", {
    "name": "Сяо Мин",
    "membership": "VIP",
    "joined": "2024-01-15",
})


@tool
def query_course_info(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Запрос подробной информации о курсах на учебном сайте EXAMPLE.

    Аргументы:
course_name: Название курса
    """
    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}》 - Цена: {info['price']},"
            f"Продолжительность: {info['hours']} hours, Сложность: {info['level']}"
        )
    return f"Курс 《{course_name}》 не найден"


@tool
def get_user_membership(
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Запрос информации о членстве пользователя.

    Аргументы:
user_id: ID пользователя
    """
    item = store.get(("example", "users"), user_id)
    if item is None:
        return f"Пользователь {user_id} не найден"

    user = item.value
    return (
        f"Пользователь {user['name']}, участник {user['membership']},"
        f"Дата регистрации {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="Вы — консультант по курсу для EXAMPLE.",
)

# Запрос информации о курсе (данные поступают из Store)
result = agent.invoke({
    "messages": [HumanMessage(content="Сколько стоит Python3 Basics Tutorial?")]
})
print(f"Запрос курса: {result['messages'][-1].content}")

# Запрос информации о пользователе (данные поступают из Store)
result = agent.invoke({
    "messages": [HumanMessage(content="Помогите найти информацию о пользователе user_vip_001")]
})
print(f"Запрос пользователя: {result['messages'][-1].content}")

Вывод:

Course query: 《Python3 基础教程》是免费的，学习时长约20小时，难度为入门级别。
User query: 用户小明是 VIP 会员，注册日期为 2024年1月15日。

Персистентность хранилища

Данные InMemoryStore теряются после перезапуска программы. Для производственных сред можно использовать персистентные решения, такие как PostgresStore:

Пример

# Этап разработки
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

# Производственная среда (требуется PostgreSQL)
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")

Рекомендации по использованию хранилища

Сценарийпример пространства имёнпример ключаОписание
Пользовательские предпочтения("users", user_id)preferencesТема, язык, настройки уведомлений
Прогресс обучения("users", user_id)прогрессПройденные курсы, продолжительность обучения
База знаний("kb", collection)doc_idДокументы, FAQ, информация о продукте
Резюме разговоров("sessions", thread_id)summaryРезюме длинных разговоров для использования в сценариях за пределами Checkpointer

Checkpointer отвечает за «где находится разговор», а Store — за «кто пользователь, что он знает и что ему нравится». Только используя оба вместе, можно построить интеллектуального агента с постоянной памятью.

Другие расширения