Краткосрочная память и сессии | Курс LangChain Agents урок 6

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

В уроке 2 вы уже видели InMemorySaver и thread_id. В этом уроке разбираемся как checkpointer работает внутри, добавляем SqliteSaver для хранения истории между перезапусками и учимся инспектировать состояние треда.

Теория

Что такое тред

Без checkpointer каждый agent.invoke() - это чистый лист. Агент не помнит предыдущих сообщений. Это поведение по умолчанию.

Тред - это именованная сессия. Все вызовы с одинаковым thread_id работают с одним и тем же накопленным состоянием. Агент помнит историю, модель видит историю вызовов инструментов, контекст сохраняется.

thread_id "research-1"   thread_id "research-2"
  invoke("Anthropic")      invoke("OpenAI")
  invoke("Продолжи...")    invoke("Продолжи...")
  invoke("Выводы?")        invoke("Выводы?")
       ↓                        ↓
   своя история             своя история

Два треда не смешивают историю. Это принципиально отличается от ConversationBufferMemory из первого курса, которая хранилась в объекте Python и терялась при перезапуске. Checkpointer сохраняет состояние в базе данных и может переживать перезапуск процесса.

Как работает checkpointer

Checkpointer сохраняет снимок (snapshot) состояния агента после каждого шага: после ответа модели, после каждого вызова инструмента. При следующем invoke() с тем же thread_id агент загружает последний снимок и продолжает с него.

Состояние агента - это AgentState, который по умолчанию содержит только messages. Его можно расширить своими полями.

Примеры кода

Базовый тред: агент помнит имя

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

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

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver(),
)

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

agent.invoke(
    {"messages": [{"role": "user", "content": "Меня зовут Миша."}]},
    config,
)

response = agent.invoke(
    {"messages": [{"role": "user", "content": "Как меня зовут?"}]},
    config,
)
print(response["messages"][-1].content)
# Вас зовут Миша.

Несколько независимых сессий

Разные thread_id не знают друг о друге.

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

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

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer,
)

# Сессия 1
agent.invoke(
    {"messages": [{"role": "user", "content": "Исследуем компанию Anthropic."}]},
    {"configurable": {"thread_id": "research-anthropic"}},
)

# Сессия 2
agent.invoke(
    {"messages": [{"role": "user", "content": "Исследуем компанию OpenAI."}]},
    {"configurable": {"thread_id": "research-openai"}},
)

# Сессия 1 продолжается, не видит сессию 2
response = agent.invoke(
    {"messages": [{"role": "user", "content": "О какой компании мы говорили?"}]},
    {"configurable": {"thread_id": "research-anthropic"}},
)
print(response["messages"][-1].content)
# Мы говорили об Anthropic.

Инспекция состояния треда

agent.get_state() возвращает текущий снимок без запуска модели.

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

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

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "inspect-demo"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "Привет, я тестирую память."}]},
    config,
)

# Читаем состояние треда
state = agent.get_state(config)
messages = state.values["messages"]

print(f"Сообщений в треде: {len(messages)}")
for msg in messages:
    print(f"  [{msg.type}] {str(msg.content)[:60]}")

Поле state.next показывает, какой узел будет следующим (пустой список значит агент завершил работу). state.metadata содержит номер шага и идентификатор запуска.

Расширение состояния: `CustomAgentState`

По умолчанию AgentState хранит только messages. Если нужно хранить дополнительные данные между шагами, расширяйте его.

import os
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()


class ResearchState(AgentState):
    topic: str
    notes: list[str]


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

agent = create_agent(
    model=model,
    tools=[],
    state_schema=ResearchState,
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "custom-state"}}

result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Расскажи о LangGraph."}],
        "topic": "LangGraph",
        "notes": [],
    },
    config,
)

state = agent.get_state(config)
print(f"Тема: {state.values.get('topic')}")
print(f"Заметки: {state.values.get('notes')}")
print(f"Ответ: {result['messages'][-1].content[:80]}")

SqliteSaver: персистентность без сервера

Если Postgres нет, SqliteSaver сохраняет состояние в локальный файл. Подходит для локальных инструментов и экспериментов.

pip install langgraph-checkpoint-sqlite

import os
import sqlite3
from dotenv import load_dotenv
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

load_dotenv()

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

checkpointer = SqliteSaver(sqlite3.connect("agent_memory.db", check_same_thread=False))

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "local-1"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "Запомни: задача 42."}]},
    config,
)
# Перезапустите скрипт, тред загрузится из agent_memory.db
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Что я просил запомнить?"}]},
    config,
)
print(response["messages"][-1].content)

Production: PostgresSaver

Для production используйте checkpointer с базой данных.

pip install langgraph-checkpoint-postgres

На Windows дополнительно нужна бинарная сборка драйвера, иначе psycopg не найдёт библиотеку libpq:

pip install "psycopg[binary]"

import os
from dotenv import load_dotenv
from langgraph.checkpoint.postgres import PostgresSaver
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI


load_dotenv()


DB_URI = os.getenv("DATABASE_URL")  # postgresql://user:pass@host/db

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # создаёт таблицы при первом запуске

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

    agent = create_agent(
        model=model,
        tools=[],
        checkpointer=checkpointer,
    )

    response = agent.invoke(
        {"messages": [{"role": "user", "content": "Запомни: задача 42."}]},
        {"configurable": {"thread_id": "persistent-1"}},
    )
    print(response["messages"][-1].content)

После перезапуска процесса тред "persistent-1" сохранится в базе и агент продолжит с последнего состояния.

Сквозной проект: несколько параллельных исследований

import asyncio
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

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


async def run_research(agent, topic: str, thread_id: str) -> str:
    """Запускает одно исследование в своём треде."""
    config = {"configurable": {"thread_id": thread_id}}

    await agent.ainvoke(
        {"messages": [{"role": "user", "content": f"Кратко расскажи о компании {topic}."}]},
        config,
    )

    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "О какой компании мы говорили?"}]},
        config,
    )

    return response["messages"][-1].content


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

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

    agent = create_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer,
        middleware=[
            ModelCallLimitMiddleware(run_limit=6, exit_behavior="end"),
        ],
    )

    # Параллельный запуск двух исследований
    results = await asyncio.gather(
        run_research(agent, "Anthropic", "research-anthropic"),
        run_research(agent, "OpenAI", "research-openai"),
    )

    for result in results:
        print(result)
        print("---")

    # Инспекция состояний
    for thread_id in ["research-anthropic", "research-openai"]:
        state = agent.get_state({"configurable": {"thread_id": thread_id}})
        msg_count = len(state.values["messages"])
        print(f"Тред {thread_id!r}: {msg_count} сообщений")


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

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

Пропустили thread_id, история не сохраняется

# Неправильно: каждый invoke независим, агент ничего не помнит
agent.invoke({"messages": [{"role": "user", "content": "Меня зовут Иван."}]})
agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]})

# Правильно: передавать config с thread_id
config = {"configurable": {"thread_id": "my-session"}}
agent.invoke({"messages": [{"role": "user", "content": "Меня зовут Иван."}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, config)

get_state() вернул пустое значение

# get_state() требует тот же config, что и invoke()
config = {"configurable": {"thread_id": "session-1"}}
agent.invoke({"messages": [...]}, config)

# Правильно: тот же thread_id
state = agent.get_state(config)

# Неправильно: другой thread_id, другой тред
state = agent.get_state({"configurable": {"thread_id": "session-2"}})

Выбор checkpointer

Три варианта с разным назначением:

InMemorySaver - данные в RAM, теряются при перезапуске. Для разработки и тестов.
SqliteSaver - данные в файле .db, переживают перезапуск. Для локальных проектов и экспериментов.
PostgresSaver - данные в базе данных. Для production.

Задание

Создайте агента с InMemorySaver. Запустите три отдельных треда: "user-alice", "user-bob", "user-carol". В каждом сообщите агенту имя пользователя.

Затем в каждом треде спросите имя. Убедитесь, что ответы не смешиваются. После выполните get_state() для каждого треда и выведите количество сообщений.

<< урок 5

урок 7 >>

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

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

Краткосрочная память и сессии | Курс LangChain Agents урок 6

Теория

Что такое тред

Как работает checkpointer

Примеры кода

Базовый тред: агент помнит имя

Несколько независимых сессий

Инспекция состояния треда

Расширение состояния: `CustomAgentState`

SqliteSaver: персистентность без сервера

Production: PostgresSaver

Сквозной проект: несколько параллельных исследований

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

Задание

Похожие туториалы

Checkpointer и память сессии | Курс LangChain Agents урок 2

Память и история диалога в LangChain | Курс по LangChain урок 4

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

Субагенты и Handoffs | Курс LangChain Agents урок 11

Агенты: автоматизация цикла tool use | Курс по LangChain урок 7

Context engineering | Курс LangChain Agents урок 4

Краткосрочная память и сессии | Курс LangChain Agents урок 6

Теория

Что такое тред

Как работает checkpointer

Примеры кода

Базовый тред: агент помнит имя

Несколько независимых сессий

Инспекция состояния треда

Расширение состояния: CustomAgentState

SqliteSaver: персистентность без сервера

Production: PostgresSaver

Сквозной проект: несколько параллельных исследований

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

Задание

Похожие туториалы

Checkpointer и память сессии | Курс LangChain Agents урок 2

Память и история диалога в LangChain | Курс по LangChain урок 4

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

Субагенты и Handoffs | Курс LangChain Agents урок 11

Агенты: автоматизация цикла tool use | Курс по LangChain урок 7

Context engineering | Курс LangChain Agents урок 4

Расширение состояния: `CustomAgentState`