Долгосрочная память в LangChain | Курс LangChain Agents урок 7

Цель урока: научить агента помнить данные между сессиями: сохранять результаты исследований в хранилище и использовать их в следующих запусках.

Теория

Краткосрочная vs долгосрочная память

Checkpointer из урока 6 хранит историю сообщений конкретного треда. Это краткосрочная память: она живёт в рамках одной сессии и к ней нет доступа из других тредов.

Долгосрочная память хранится отдельно от тредов в store. Это словарь документов, который агент может читать и писать из любой сессии. Когда пользователь вернётся через неделю с новым thread_id, агент всё равно вспомнит предыдущие результаты.

Тред A (thread_id="user-1-session-1")  ─┐
Тред B (thread_id="user-1-session-2")  ─┤─→  Store (долгосрочная память)
Тред C (thread_id="user-1-session-3")  ─┘

Структура хранилища

Каждый документ в store имеет два адреса:

namespace - кортеж строк, аналог папки: ("user-42", "research")
key - строка, имя документа внутри namespace: "anthropic"

Namespace позволяет разделять данные: память одного пользователя не смешивается с памятью другого.

Примеры кода

Работа со store напрямую

import os
from dotenv import load_dotenv
from langgraph.store.memory import InMemoryStore

load_dotenv()

store = InMemoryStore()

namespace = ("user-42", "research")

# Сохранить документ
store.put(namespace, "anthropic", {
    "topic": "Anthropic",
    "summary": "AI-компания, создатель Claude. Основана в 2021.",
    "date": "2026-02-01",
})

# Получить по ключу
item = store.get(namespace, "anthropic")
print(item.value["summary"])

# Список всех документов в namespace
items = store.search(namespace)
for item in items:
    print(f"  {item.key}: {item.value['topic']}")

Чтение из store в инструменте

Инструмент получает доступ к store через ToolRuntime. Для этого нужно объявить параметр runtime: ToolRuntime в сигнатуре инструмента.

import os
from dataclasses import dataclass
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI
from langgraph.store.memory import InMemoryStore

load_dotenv()


@dataclass
class UserContext:
    user_id: str


store = InMemoryStore()

# Заранее кладём данные
store.put(("user-42", "preferences"), "settings", {
    "language": "ru",
    "style": "краткий",
})


@tool
def get_user_preferences(runtime: ToolRuntime[UserContext]) -> str:
    """Получить настройки пользователя."""
    user_id = runtime.context.user_id
    item = runtime.store.get((user_id, "preferences"), "settings")
    if item:
        return str(item.value)
    return "Настройки не найдены."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[get_user_preferences],
    store=store,
    context_schema=UserContext,
)

response = agent.invoke(
    {"messages": [{"role": "user", "content": "Какой у меня стиль ответов?"}]},
    context=UserContext(user_id="user-42"),
)
print(response["messages"][-1].content)

Запись в store из инструмента

Агент может не только читать, но и сохранять данные с помощью инструментов.

import os
from dataclasses import dataclass
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI
from langgraph.store.memory import InMemoryStore

load_dotenv()


@dataclass
class UserContext:
    user_id: str


class ResearchResult(TypedDict):
    topic: str
    summary: str


store = InMemoryStore()


@tool
def save_research(result: ResearchResult, runtime: ToolRuntime[UserContext]) -> str:
    """Сохранить результат исследования для дальнейшего использования."""
    user_id = runtime.context.user_id
    key = result["topic"].lower().replace(" ", "-")
    runtime.store.put((user_id, "research"), key, result)
    return f"Исследование по теме '{result['topic']}' сохранено."


@tool
def get_research(topic: str, runtime: ToolRuntime[UserContext]) -> str:
    """Получить сохранённый результат исследования по теме."""
    user_id = runtime.context.user_id
    key = topic.lower().replace(" ", "-")
    item = runtime.store.get((user_id, "research"), key)
    if item:
        return f"Найдено: {item.value['summary']}"
    return f"Исследование по теме '{topic}' не найдено."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[save_research, get_research],
    store=store,
    context_schema=UserContext,
    system_prompt=(
        "Перед ответом на вопрос об объекте или теме всегда проверяй "
        "сохранённые данные через get_research. "
        "Если данные найдены, используй их в ответе."
    ),
)

config = {"configurable": {"thread_id": "session-1"}}

# Первый запрос, сохранить исследование
agent.invoke(
    {"messages": [{"role": "user", "content": "Сохрани: Anthropic, основана в 2021, делает Claude."}]},
    config,
    context=UserContext(user_id="user-42"),
)

# Второй запрос в новом треде, данные доступны
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Что ты знаешь об Anthropic?"}]},
    {"configurable": {"thread_id": "session-2"}},
    context=UserContext(user_id="user-42"),
)
print(response["messages"][-1].content)

Store не привязан к треду: данные, сохранённые в session-1, доступны в session-2 под тем же user_id.

Семантический поиск

Если передать функцию эмбеддинга при создании store, store.search() будет ранжировать результаты по смысловому сходству с запросом.

import os
from dotenv import load_dotenv
from langgraph.store.memory import InMemoryStore

load_dotenv()

# Заглушка для демонстрации. В продакшне замените на реальные эмбеддинги:
#   from langchain_openai import OpenAIEmbeddings
#   embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
#   embed_fn = embeddings.embed_documents
#   dims = 1536
def embed_fn(texts: list[str]) -> list[list[float]]:
    import hashlib
    result = []
    for text in texts:
        h = int(hashlib.md5(text.encode()).hexdigest(), 16)
        vec = [(h >> i & 0xFF) / 255.0 for i in range(8)]
        result.append(vec)
    return result

store = InMemoryStore(
    index={
        "embed": embed_fn,
        "dims": 8,
    }
)

namespace = ("research",)

store.put(namespace, "anthropic", {"text": "Anthropic создала Claude. AI-безопасность."})
store.put(namespace, "openai", {"text": "OpenAI создала GPT. ChatGPT."})
store.put(namespace, "deepmind", {"text": "DeepMind работает над AGI. AlphaFold."})

# Поиск по смыслу (с реальными эмбеддингами результаты будут семантически релевантны)
results = store.search(namespace, query="безопасный ИИ")
for item in results:
    print(f"{item.key}: {item.value['text']}")

Без функции эмбеддинга search() возвращает все документы namespace без ранжирования. С реальными эмбеддингами результаты сортируются по смысловому сходству с запросом.

Сквозной проект: агент не повторяет исследования

import asyncio
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

load_dotenv()

SERVER_PATH = str(Path(__file__).parent.parent / "agents_course" / "tools_server.py")


@dataclass
class AnalystContext:
    user_id: str


class ResearchNote(TypedDict):
    topic: str
    summary: str


store = InMemoryStore()


@tool
def save_research_note(note: ResearchNote, runtime: ToolRuntime[AnalystContext]) -> str:
    """Сохранить результат исследования в долгосрочную память."""
    user_id = runtime.context.user_id
    key = note["topic"].lower().replace(" ", "-")
    runtime.store.put((user_id, "research"), key, note)
    return f"Заметка по теме '{note['topic']}' сохранена."


@tool
def get_research_note(topic: str, runtime: ToolRuntime[AnalystContext]) -> str:
    """Проверить, есть ли уже сохранённое исследование по теме."""
    user_id = runtime.context.user_id
    key = topic.lower().replace(" ", "-")
    item = runtime.store.get((user_id, "research"), key)
    if item:
        return f"Уже исследовано: {item.value['summary']}"
    return f"По теме '{topic}' исследований нет, нужно провести."


async def main():
    client = MultiServerMCPClient({
        "analyst_tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [SERVER_PATH],
        }
    })

    mcp_tools = await client.get_tools()
    model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

    agent = create_agent(
        model=model,
        tools=mcp_tools + [save_research_note, get_research_note],
        checkpointer=InMemorySaver(),
        store=store,
        context_schema=AnalystContext,
        system_prompt=(
            "Ты агент-аналитик. Перед исследованием темы проверяй долгосрочную память "
            "через get_research_note. Если данные уже есть, используй их. "
            "После нового исследования сохраняй результат через save_research_note."
        ),
    )

    context = AnalystContext(user_id="analyst-1")

    # Первая сессия: исследуем Anthropic
    r1 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Исследуй компанию Anthropic."}]},
        {"configurable": {"thread_id": "s1"}},
        context=context,
    )
    print("Сессия 1:", r1["messages"][-1].content[:80])

    # Вторая сессия (новый thread_id), агент должен найти данные в store
    r2 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Что ты знаешь об Anthropic?"}]},
        {"configurable": {"thread_id": "s2"}},
        context=context,
    )
    print("Сессия 2:", r2["messages"][-1].content[:80])

    # Проверяем store
    items = store.search(("analyst-1", "research"))
    print(f"\nВ store сохранено заметок: {len(items)}")


if __name__ == "__main__":
    asyncio.run(main())

Частые ошибки

store.get() вернул None, хотя данные были сохранены

# Неправильно: namespace не совпадает
store.put(("user-42", "research"), "topic", {...})
item = store.get(("user-42", "notes"), "topic")  # другой namespace

# Правильно: тот же namespace
item = store.get(("user-42", "research"), "topic")

ToolRuntime без передачи store в агента

# Неправильно: runtime.store будет None
agent = create_agent(model=model, tools=[my_tool])

# Правильно: передать store явно
agent = create_agent(model=model, tools=[my_tool], store=store)

Store и checkpointer это разные вещи

Checkpointer хранит историю сообщений треда. Store хранит произвольные документы. Они независимы, можно использовать оба одновременно.

Задание

Создайте агента с двумя инструментами:

remember(key, value) - сохраняет произвольный факт в store
recall(key) - достаёт факт из store

Запустите три отдельных сессии thread_id. В первой сохраните несколько фактов. В двух следующих запросите их. Убедитесь, что данные доступны в каждой сессии.

<< урок 6

урок 8 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия