Память сессии: threads, checkpoints и цена истории | Курс LangChain урок 12

СЛУЖЕБНОЕ, НЕ ПУБЛИКУЕТСЯ

Поля публикации

1) type_content: tutorial

2) title (75): Память сессии: threads, checkpoints и цена истории | Курс LangChain урок 12

3) seo_description (154): Урок 12 курса по LangChain: память агента на checkpointer и thread_id, чтение checkpoints, режимы durability, SqliteSaver, обрезка истории и суммаризация.

4) seo_keywords: langchain checkpointer, langgraph checkpointer, thread_id, InMemorySaver, SqliteSaver, get_state_history, durability langgraph, SummarizationMiddleware, RemoveMessage, память агента langchain

5) category_id: 10

ТЕКСТ МАТЕРИАЛА, НИЖЕ ЭТОЙ СТРОКИ ВСЁ ИДЁТ В БЛОГ

Цель урока: включить агенту память диалога параметром checkpointer и ключом thread_id, читать checkpoints состояния thread, выбирать режим долговечности записи. И удерживать историю в пределах окна обрезкой или суммаризацией.

Необходимые знания:

1) урок 0: окружение собрано, ключ работает, переменные MODEL_NAME и MODEL_BASE_URL заполнены

2) урок 1: выдумки модели, уверенный ответ без опоры на факт

3) урок 3: блоки reasoning в ответе модели и ключ reasoning в output_token_details

4) урок 4: расход токенов в usage_metadata и счёт денег по нему

5) урок 5: системный промпт агента и middleware как точка настройки

6) урок 9: цикл вызова инструментов и ToolMessage в истории

7) урок 11: create_agent, состояние агента, узлы model и tools, редьюсер поля

8) Python на уровне джуниора: контекстный менеджер with, модуль pathlib для работы с путями

Ключевые концепции:

1) thread это единица памяти агента, и адресуется он ключом thread_id в конфигурации вызова

2) checkpointer сохраняет checkpoint состояния на каждом шаге графа и подставляет историю при следующем вызове

3) checkpoint читается объектом StateSnapshot: значения каналов, следующий узел, служебные поля и адрес родителя

4) режим долговечности durability задаёт, когда запись уходит на носитель и чем вы платите за надёжность

5) история растёт с каждым вызовом агента, и каждый вызов оплачивает её целиком

6) обрезка снимает старые сообщения дёшево и с потерей, суммаризация стоит лишнего вызова модели и оставляет вместо истории пересказ

7) память сессии привязана к одному thread, в соседнем thread этой истории нет

Чем закончился прошлый урок

Агент из урока 11 начинал каждый вызов с пустой истории. Внутри одного вызова история полная, между вызовами её нет. Поэтому вопрос "а когда он приедет" приходил к модели без всякой связи с предыдущим заказом.

Решить это можно двумя способами. Первый вам уже знаком: ваш код держит список сообщений у себя и целиком передаёт его в следующий вызов. Такой способ годится, пока диалог один и лежит в памяти процесса. Когда появляются разные пользователи, перезапуски программы и параллельные сессии, хранение истории становится отдельной подсистемой, и писать и отлаживать её приходится самим.

Второй способ: отдать хранение фреймворку. Документация называет этот слой persistence и делит его на две части. Checkpointer отвечает за состояние одного thread, хранилище (store) за данные, общие для всех threads. Checkpointer разобран в этом уроке, хранилище в уроке 13.

Thread это последовательность вызовов агента с общим состоянием. Каждый следующий вызов в thread получает всё сказанное в нём раньше, а сообщения другого thread в него не попадают.

Общая сборка модели

Модуль сборки модели тот же, что в уроках 4-11. Положите его рядом с примерами под именем course_model.py, и дальше каждый пример подключает его одной строкой.

"""Общая сборка модели для примеров урока 12.

Тот же модуль, что course_model.py уроков 4-11, без изменений: у build_model есть
необязательный первый аргумент с именем модели, а рядом лежит gateway_kwargs()
для примеров, которые собирают модель сами.

Файл .env берётся тот же, что в уроке 0. Положите его рядом с этой папкой или
выше по дереву: load_dotenv() ищет файл начиная с папки этого модуля и поднимается
вверх.
"""

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()


def gateway_kwargs():
    """Возвращает аргументы доступа к провайдеру: имя, адрес, ключ.

    Нужны примерам, которые вызывают init_chat_model сами. У настраиваемой модели
    имени модели при создании нет, поэтому build_model ей не подходит.
    """
    base_url = os.getenv("MODEL_BASE_URL")

    if base_url:
        return {
            "model_provider": "openai",
            "base_url": base_url,
            "api_key": os.environ["OPENAI_API_KEY"],
        }

    return {}


def build_model(model_name=None, **kwargs):
    """Собирает модель курса.

    model_name без значения означает модель из переменной MODEL_NAME. Явное имя
    нужно примерам, где моделей в приложении больше одной.

    Все именованные аргументы уходят в init_chat_model как есть: temperature,
    max_tokens, timeout, max_retries, rate_limiter, profile и прочее из раздела
    Parameters.
    """
    model_name = model_name or os.environ["MODEL_NAME"]
    access = gateway_kwargs()

    if access:
        # Путь для любого адреса, совместимого с OpenAI Chat Completions API.
        return init_chat_model(model=model_name, **access, **kwargs)

    # Путь напрямую к провайдеру.
    return init_chat_model(model_name, **kwargs)

Как включить память: checkpointer и thread_id

Память включается параметром checkpointer у create_agent: в него передаётся объект, который хранит checkpoints. Дальше в каждый вызов агента вы передаёте конфигурацию, где в словаре configurable лежит ключ thread_id.

В начале вызова граф читает из thread последний checkpoint, а после каждого шага записывает новый. Во вход invoke вы кладёте только новую реплику, а история подставляется из checkpoint. Поле messages склеивает старое и новое редьюсером add_messages, с которым вы познакомились в уроке 11.

В первом примере чекпойнтер держит данные в оперативной памяти. Настраивать его не нужно, но хранит он их только до конца процесса: программа завершилась, и thread пропал вместе с ней. Для первого знакомства этого достаточно, а дальше в уроке checkpoints будут храниться в файле на диске.

Пример 01_thread.py

"""Пример 1 урока 12: история одного thread, соседний thread её не получает.

Агент собран с чекпойнтером в памяти. Три запроса: два уходят в thread
"seat-1", третий в thread "seat-2". Между вызовами код не передаёт историю
руками, её подставляет чекпойнтер по thread_id из конфигурации вызова.
"""

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

from course_model import build_model

agent = create_agent(
    model=build_model(temperature=0, max_tokens=256),
    tools=[],
    system_prompt=(
        "Вы помощник склада. Отвечайте по-русски, одной короткой фразой. "
        "Если нужного факта в диалоге не было, так и скажите."
    ),
    checkpointer=InMemorySaver(),
)


def ask(question, thread_id):
    """Один запрос в названный thread. В invoke уходит только новая реплика."""
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config,
    )
    answer = result["messages"][-1].text.replace("\n", " ")
    print(f"[{thread_id}] вопрос: {question}")
    print(f"[{thread_id}] ответ:  {answer}")
    print(f"[{thread_id}] сообщений в thread после вызова: {len(result['messages'])}")
    print()


ask("Запомните: моя смена начинается в 7:40.", "seat-1")
ask("Во сколько начинается моя смена?", "seat-1")
ask("Во сколько начинается моя смена?", "seat-2")

# Вывод:
# [seat-1] вопрос: Запомните: моя смена начинается в 7:40.
# [seat-1] ответ:  Запомнил: смена в 7:40.
# [seat-1] сообщений в thread после вызова: 2
#
# [seat-1] вопрос: Во сколько начинается моя смена?
# [seat-1] ответ:  Ваша смена начинается в 7:40.
# [seat-1] сообщений в thread после вызова: 4
#
# [seat-2] вопрос: Во сколько начинается моя смена?
# [seat-2] ответ:  Извините, в нашем диалоге не было информации о времени начала вашей смены.
# [seat-2] сообщений в thread после вызова: 2

В третьем вызове тот же вопрос, что во втором, только он уходит в thread "seat-2", и ответ другой. В этом thread истории нет, поэтому модель отвечает, что времени смены в диалоге не было. Такой ответ бывает не при каждом запуске: модель может назвать и время, которого не получала, например 8:00. Это выдумка из урока 1, и пустая память от неё не защищает.

Разделение по thread задано самим устройством хранения: thread_id работает первичным ключом, по которому чекпойнтер ищет checkpoint. Два оператора за соседними столами не увидят диалогов друг друга, пока вы не дадите им один и тот же идентификатор.

Имя thread задаёт ваше приложение. Это может быть идентификатор диалога в мессенджере, номер обращения в поддержку или пара из пользователя и задачи. Делайте имя коротким: документация предупреждает, что у PostgresSaver колонка под thread_id ограничена по длине, и советует держаться в пределах 255 знаков. Длинный ключ замените UUID или хешем.

Посмотрите и на счётчик сообщений в выводе: thread растёт с каждым вызовом. Значит, в модель каждый раз уходит больше, чем вы написали в этом вызове. Сколько именно в токенах, посчитано в разделе "Что занимает контекстное окно".

Два имени одного класса

У чекпойнтера в оперативной памяти два имени, и это сбивает.

# Оба импорта дают один и тот же класс
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.memory import MemorySaver

В документации на страницах про persistence и про чекпойнтеры стоит InMemorySaver, а на страницах про тестирование и стриминг MemorySaver. Если читать их подряд, кажется, что это два разных инструмента. А на самом деле класс один. В модуле langgraph/checkpoint/memory/__init__.py закреплённой версии определён InMemorySaver, а в конце файла стоит строка MemorySaver = InMemorySaver с комментарием о совместимости со старым кодом.

Пишите InMemorySaver. Встретив MemorySaver в чужом коде, знайте, что это тот же класс под прежним именем.

InMemorySaver годится для отладки и тестов. Для продакшна ставьте чекпойнтер на базе данных, например PostgresSaver. Чем хранение в памяти отличается от хранения в базе, видно на примере SQLite в разделе "Долговечность записи и thread на диске".

Checkpoint: что лежит в thread

Чекпойнтер записывает checkpoint после каждого супершага (super-step). Супершаг это один такт графа: в нём отрабатывают все узлы, запланированные на этот такт. У агента из урока 11 первый такт уходит на вход в граф, а дальше в каждом такте работает один узел, model или tools. Поэтому на один ваш вопрос приходится несколько checkpoints. В примере ниже запрос с одним вызовом инструмента даёт их пять, вместе с записью самого входа.

Прочитанный checkpoint приходит к вам объектом StateSnapshot с такими полями.

Документация показывает в metadata ещё поле writes с правками узлов. В закреплённых версиях его нет: langgraph 1.2.11 это поле не заполняет, и в описании CheckpointMetadata из пакета langgraph-checkpoint 4.2.0 его тоже нет. Сами правки узлов чекпойнтер сохраняет отдельно от checkpoint, записями узлов (в документации pending writes). Каждая такая запись привязана к checkpoint, от которого узел начал работу.

Читаются checkpoints двумя методами графа. get_state(config) отдаёт последний checkpoint в thread, get_state_history(config) отдаёт всю цепочку, начиная со свежего. Если в configurable вместе с thread_id положить ещё и checkpoint_id, get_state вернёт именно этот checkpoint.

Пример 02_snapshots.py

"""Пример 2 урока 12: checkpoint состояния thread.

Один запрос, на который агенту нужен инструмент. После вызова читается то,
что осталось в thread: последний checkpoint и вся история checkpoints. Видно,
на каких шагах графа чекпойнтер делает запись.
"""

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

from course_model import build_model

SHIFTS = {"Казань": "7:40", "Омск": "9:00"}


def shift_start(city: str) -> str:
    """Возвращает время начала смены на складе в городе."""
    time = SHIFTS.get(city)
    if time is None:
        return f"Склада в городе {city} нет."
    return f"Смена на складе в городе {city} начинается в {time}."


agent = create_agent(
    model=build_model(temperature=0, max_tokens=256),
    tools=[shift_start],
    system_prompt="Вы помощник склада. Отвечайте по-русски, одной короткой фразой.",
    checkpointer=InMemorySaver(),
)

CONFIG = {"configurable": {"thread_id": "shift-1"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "Во сколько смена в Казани?"}]},
    CONFIG,
)

snapshot = agent.get_state(CONFIG)

print("ПОСЛЕДНИЙ CHECKPOINT")
print("  ключи values:  ", sorted(snapshot.values))
print("  сообщений:     ", len(snapshot.values["messages"]))
print("  next:          ", snapshot.next)
print("  шаг:           ", snapshot.metadata["step"])
print("  источник:      ", snapshot.metadata["source"])
print("  checkpoint_id: ", snapshot.config["configurable"]["checkpoint_id"])
print("  checkpoint_ns: ", repr(snapshot.config["configurable"]["checkpoint_ns"]))
print("  создан:        ", snapshot.created_at)
# У самого первого checkpoint в thread родителя нет, parent_config равен None.
parent = snapshot.parent_config
print(
    "  родитель:      ",
    parent["configurable"]["checkpoint_id"] if parent else None,
)
print()

print("ИСТОРИЯ CHECKPOINTS, сверху самый свежий")
print(f"  {'шаг':>4}  {'источник':<8}  {'следующий узел':<16}  сообщений")
for item in agent.get_state_history(CONFIG):
    following = ", ".join(item.next) or "-"
    count = len(item.values.get("messages", []))
    print(
        f"  {item.metadata['step']:>4}  {item.metadata['source']:<8}  "
        f"{following:<16}  {count}"
    )

# Вывод:
# ПОСЛЕДНИЙ CHECKPOINT
#   ключи values:   ['messages']
#   сообщений:      4
#   next:           ()
#   шаг:            3
#   источник:       loop
#   checkpoint_id:  1f1c356b-c194-6c18-8003-255554f4034a
#   checkpoint_ns:  ''
#   создан:         2026-10-08T20:27:51.557429+00:00
#   родитель:       1f1c356b-b4b6-6db9-8002-89ba5e3ca741
#
# ИСТОРИЯ CHECKPOINTS, сверху самый свежий
#    шаг  источник  следующий узел    сообщений
#      3  loop      -                 4
#      2  loop      model             3
#      1  loop      tools             2
#      0  loop      model             1
#     -1  input     __start__         0
#
# Значения checkpoint_id и время создания свои при каждом запуске, у вас будут другие.

Нижняя строка истории, шаг -1 с источником input, это запись вашего входа до первого узла. На шаге 0 отработал служебный узел __start__, он передал вход узлу model. Дальше идут такты цикла, по одному на узел: model, tools и снова model. У всех записей, начиная с шага 0, источник loop.

Кроме input и loop, бывают ещё два источника. update получает checkpoint, когда состояние правили вручную методом update_state. fork получает копия, которую граф делает, когда его вызывают от старого checkpoint. По этим двум источникам записи, сделанные правкой или вызовом от старой точки, отличают от записей обычного вызова.

Поле next у верхнего checkpoint пустое, и это формальный признак того, что вызов закончен. У промежуточных записей там стоит узел, который пойдёт следующим, например tools у checkpoint перед вызовом инструмента. Если next не пустой у последнего checkpoint, вызов не закончен: например, агент ждёт подтверждения человека. С этого checkpoint вызов потом и продолжают.

Поле parent_config связывает checkpoints в цепочку. Если вызвать граф с адресом старого checkpoint, узлы до этой точки не выполняются повторно, их результаты уже сохранены, а узлы после неё выполняются заново. Новые записи пойдут от старой точки, и история из списка превратится в дерево. Вызов от выбранной точки и отладка по истории в этом курсе не разбираются, это тема отдельного курса по LangGraph. Для них нужна только история checkpoints, и чекпойнтер её уже записывает.

Долговечность записи и thread на диске

Запись checkpoint на носитель занимает время. Когда её делать, задаёт параметр durability у вызова графа.

Значение ставится прямо в вызове, рядом с конфигурацией thread, и действует на весь вызов. Без параметра действует "async", в этом режиме идут все примеры урока, кроме третьего.

Вторая тема раздела, это сам носитель. Чекпойнтеры поставляются отдельными пакетами:

1) langgraph-checkpoint идёт вместе с langgraph и даёт InMemorySaver

2) langgraph-checkpoint-sqlite даёт SqliteSaver и его асинхронную версию

3) langgraph-checkpoint-postgres даёт PostgresSaver для продакшна

4) langgraph-checkpoint-mongodb и langchain-azure-cosmosdb дают чекпойнтеры на MongoDB и Azure Cosmos DB

В курсе используется SQLite: база лежит в файле рядом с кодом, и сервер для неё не нужен. Пакет ставится отдельно:

pip install langgraph-checkpoint-sqlite

SqliteSaver открывается методом from_conn_string. Это контекстный менеджер: он отдаёт готовый чекпойнтер, а на выходе из блока with закрывает соединение. Строка ":memory:" создаёт базу в оперативной памяти, любая другая строка задаёт путь к файлу. Таблицы в новом файле чекпойнтер создаёт сам, при первом обращении.

В документации SqliteSaver в примерах создаётся из готового соединения: SqliteSaver(sqlite3.connect("checkpoint.db")). Метод from_conn_string у синхронного SqliteSaver там не показан, он описан в коде пакета.

Пример 03_sqlite.py

"""Пример 3 урока 12: thread на диске и режим долговечности записи.

SqliteSaver держит checkpoints в файле. Программа открывает соединение, задаёт
вопрос и закрывает соединение, а потом открывает тот же файл заново, другим
соединением и другим объектом агента. Thread при этом остаётся на месте.

Первый сеанс пишет в режиме durability="sync", второй в режиме "exit", и по
числу checkpoints после каждого видна разница.

Файл session.sqlite удаляется на старте, чтобы вывод не зависел от того,
сколько раз пример уже запускали.

SqliteSaver ставится отдельным пакетом: pip install langgraph-checkpoint-sqlite
"""

from pathlib import Path

from langchain.agents import create_agent
from langgraph.checkpoint.sqlite import SqliteSaver

from course_model import build_model

DB_PATH = Path(__file__).with_name("session.sqlite")
DB_PATH.unlink(missing_ok=True)

CONFIG = {"configurable": {"thread_id": "shift-1"}}
SYSTEM_PROMPT = (
    "Вы помощник склада. Отвечайте по-русски, одной короткой фразой. "
    "Если нужного факта в диалоге не было, так и скажите."
)


def session(question, durability):
    """Отдельный сеанс работы: своё соединение, свой агент, общий файл thread."""
    with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
        agent = create_agent(
            model=build_model(temperature=0, max_tokens=256),
            tools=[],
            system_prompt=SYSTEM_PROMPT,
            checkpointer=checkpointer,
        )
        result = agent.invoke(
            {"messages": [{"role": "user", "content": question}]},
            CONFIG,
            durability=durability,
        )
        checkpoints = len(list(agent.get_state_history(CONFIG)))

    answer = result["messages"][-1].text.replace("\n", " ")
    print(f"вопрос:      {question}")
    print(f"ответ:       {answer}")
    print(f"durability:  {durability}")
    print(f"checkpoints в thread: {checkpoints}")
    print(f"файл thread:          {DB_PATH.name}, {DB_PATH.stat().st_size} байт")
    print()


session("Запомните: моя смена начинается в 7:40.", "sync")
session("Во сколько начинается моя смена?", "exit")

with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
    tuples = list(checkpointer.list(CONFIG))
    print("ЗАПИСИ THREAD В ФАЙЛЕ")
    print("  записей в файле:", len(tuples))
    print("  самая свежая:   ", tuples[0].config["configurable"]["checkpoint_id"])
    print("  самая старая:   ", tuples[-1].config["configurable"]["checkpoint_id"])

# Вывод:
# вопрос:      Запомните: моя смена начинается в 7:40.
# ответ:       Запомнил: ваша смена начинается в 7:40.
# durability:  sync
# checkpoints в thread: 3
# файл thread:          session.sqlite, 20480 байт
#
# вопрос:      Во сколько начинается моя смена?
# ответ:       Ваша смена начинается в 7:40.
# durability:  exit
# checkpoints в thread: 4
# файл thread:          session.sqlite, 28672 байт
#
# ЗАПИСИ THREAD В ФАЙЛЕ
#   записей в файле: 4
#   самая свежая:    1f1c356b-f827-6604-8004-4db80e1932a1
#   самая старая:    1f1c356b-db2d-67fd-bfff-85ee68d8874a
#
# Размер файла и checkpoint_id свои при каждом запуске.

Второй сеанс работает с новым объектом агента и новым соединением, а отвечает так, будто диалог не прерывался. Это и есть разница между памятью процесса и памятью приложения: файл переживает и объект, и процесс, и перезапуск машины.

Сравните прирост числа checkpoints после первого сеанса и после второго. Шагов графа в обоих одинаково, а значения durability разные. При "sync" в файл ушли три checkpoints, при "exit" один, финальный. Режим "exit" экономит обращения к носителю. Но если процесс упадёт посреди длинного вызова, промежуточных checkpoints не будет, и вызов придётся повторять с начала.

Конец примера читает записи thread прямо из чекпойнтера. Его метод list возвращает кортежи CheckpointTuple, из них граф и собирает StateSnapshot. У чекпойнтера есть и метод delete_thread: он удаляет checkpoints этого thread вместе с записями узлов. Им выполняют просьбу пользователя удалить переписку.

Что занимает контекстное окно

Память сессии решает одну задачу и создаёт другую. Модель не хранит состояние между запросами, поэтому каждый вызов отправляет ей всю историю заново. Чем длиннее thread, тем больше уходит в модель на каждый вопрос и тем дороже становится вопрос, который сам по себе не изменился.

С ростом thread:

1) полная история перестаёт помещаться в контекстное окно

2) качество ответов падает: устаревшие и посторонние сообщения размывают нужное

3) ответ приходит дольше, а счёт растёт

Пример 04_window.py

"""Пример 4 урока 12: чем длиннее thread, тем дороже каждый следующий вызов.

Четыре запроса подряд в один thread. После каждого печатается строка: сколько
сообщений лежит в thread, сколько входных токенов насчитал провайдер за этот
запрос, сколько их набежало с начала диалога и какой объём истории насчитал
счётчик LangChain.
"""

from langchain.agents import create_agent
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.checkpoint.memory import InMemorySaver

from course_model import build_model

agent = create_agent(
    model=build_model(temperature=0, max_tokens=384),
    tools=[],
    system_prompt=(
        "Вы помощник склада. Отвечайте по-русски, тремя-четырьмя развёрнутыми "
        "фразами."
    ),
    checkpointer=InMemorySaver(),
)

CONFIG = {"configurable": {"thread_id": "window-1"}}

QUESTIONS = [
    "Запомните: моя смена начинается в 7:40.",
    "Опишите порядок приёмки паллет на складе.",
    "А как оформлять брак, найденный при приёмке?",
    "Во сколько начинается моя смена?",
]

spent = 0
print(f"{'вызов':>5}  {'сообщений':>9}  {'вход':>6}  {'вход всего':>10}  {'история':>7}")

for number, question in enumerate(QUESTIONS, start=1):
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        CONFIG,
    )
    answer = result["messages"][-1]
    # Расход возвращает не каждый провайдер, поэтому пустой словарь как запасной путь.
    usage = answer.usage_metadata or {}
    incoming = usage.get("input_tokens", 0)
    spent += incoming
    history = count_tokens_approximately(result["messages"])
    print(
        f"{number:>5}  {len(result['messages']):>9}  {incoming:>6}  "
        f"{spent:>10}  {history:>7}"
    )

print()
print("ПОСЛЕДНИЙ ВОПРОС:", QUESTIONS[-1])
print("ПОСЛЕДНИЙ ОТВЕТ: ", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# вызов  сообщений    вход  вход всего  история
#     1          2      53          53       67
#     2          4     134         187      183
#     3          6     262         449      300
#     4          8     384         833      369
#
# ПОСЛЕДНИЙ ВОПРОС: Во сколько начинается моя смена?
# ПОСЛЕДНИЙ ОТВЕТ:  Ваша смена начинается в 7:40, как вы и указывали ранее. Напоминаю, что к этому времени желательно быть на складе, чтобы переодеться и принять смену. Я буду учитывать это время при планировании ваших задач.

Колонка "вход" показывает главное: число входных токенов растёт вместе с историей, а длина вопроса на него почти не влияет. Четвёртый вопрос короче первого, но обошёлся в 384 токена против 53, потому что вместе с ним в модель ушло всё сказанное раньше. Сумма за диалог растёт ещё быстрее, квадратично от числа вызовов: за четыре вызова набежало 833 входных токена.

Колонка "история" посчитана функцией count_tokens_approximately. Это приблизительная оценка, которую ваш код получает до запроса, без обращения к провайдеру. Той же функцией SummarizationMiddleware считает историю, когда сверяет её с порогом в токенах, и поправляет оценку вверх по расходу из последнего ответа модели. Кроме того, порог срабатывает, если его достигла цифра total_tokens из последнего ответа модели. Эта проверка работает, только если ответ пришёл от того же провайдера, что у модели сводки.

Историю раздувает и рассуждение модели. Оно приходит в содержимом сообщения блоками reasoning, их вы видели в уроке 3. Его токены входят в выходные токены того ответа модели, где рассуждение появилось, и в usage_metadata для них есть отдельный ключ reasoning внутри output_token_details. Например, в документации на 304 выходных токена приходится 256 токенов рассуждения.

Сообщение ложится в состояние thread целиком, вместе с рассуждением. Поэтому при модели с рассуждением история растёт быстрее видимого текста и достигает порога сжатия раньше. Дорожает ли от этого каждый вызов, зависит от провайдера: рассуждение может уходить ему повторно, а может отбрасываться. Проверьте это по колонке "вход" в примере выше.

Ни обрезка, ни суммаризация не отделяют рассуждение от ответа: оба приёма работают с сообщением целиком. Обрезка выбрасывает сообщение вместе с рассуждением, суммаризация заменяет пересказом целый кусок диалога.

Укоротить историю можно обрезкой или суммаризацией, у каждого способа своя цена.

Обрезка: дёшево и с потерей

Первый способ, это выкинуть старые сообщения. Для этого подходит хук before_model, он отрабатывает перед каждым вызовом модели.

Историю правят служебным сообщением RemoveMessage: редьюсер add_messages выполняет его как команду на удаление. С идентификатором конкретного сообщения RemoveMessage удаляет это сообщение, а с константой REMOVE_ALL_MESSAGES из langgraph.graph.message очищает весь список. Для обрезки удобен второй вариант: сначала удалить всё, а следом вернуть сообщения, которые вы решили оставить.

История после правки обязана остаться валидной для провайдера: часть провайдеров ждёт, что список начинается с реплики человека. Почти все требуют, чтобы вызов инструмента и его результат стояли в истории вместе. За сообщением модели с вызовами идут результаты, и у каждого результата есть свой вызов. Если граница обрезки легла между вызовом и результатом, провайдер возвращает ошибку, и запрос к модели не проходит.

Пример 05_trim.py

"""Пример 5 урока 12: обрезка истории перед вызовом модели.

Функция trim_history, повешенная на хук before_model, перед каждым вызовом
модели оставляет в thread последние KEEP сообщений и удаляет всё, что старше.
Диалог тот же, что в примере 4. По ответу на последний вопрос видно, пережил ли
обрезку факт из первой реплики.
"""

from typing import Any

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime

from course_model import build_model

KEEP = 4


@before_model
def trim_history(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Оставляет в thread последние KEEP сообщений, начиная с реплики человека."""
    messages = state["messages"]
    if len(messages) <= KEEP:
        return None

    kept = messages[-KEEP:]
    # История, начатая ответом модели или результатом инструмента, ломает часть
    # провайдеров: первым сообщением они ждут реплику человека.
    while kept and kept[0].type != "human":
        kept = kept[1:]
    if not kept:
        return None

    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *kept]}


agent = create_agent(
    model=build_model(temperature=0, max_tokens=384),
    tools=[],
    system_prompt=(
        "Вы помощник склада. Отвечайте по-русски, тремя-четырьмя развёрнутыми "
        "фразами. Если нужного факта в диалоге не было, так и скажите."
    ),
    middleware=[trim_history],
    checkpointer=InMemorySaver(),
)

CONFIG = {"configurable": {"thread_id": "trim-1"}}

QUESTIONS = [
    "Запомните: моя смена начинается в 7:40.",
    "Опишите порядок приёмки паллет на складе.",
    "А как оформлять брак, найденный при приёмке?",
    "Во сколько начинается моя смена?",
]

print(f"{'вызов':>5}  {'сообщений в thread':>18}  первое сообщение thread")

for number, question in enumerate(QUESTIONS, start=1):
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        CONFIG,
    )
    first = result["messages"][0].text.replace("\n", " ")
    if len(first) > 40:
        first = first[:37] + "..."
    print(f"{number:>5}  {len(result['messages']):>18}  {first!r}")

print()
print("ПОСЛЕДНИЙ ВОПРОС:", QUESTIONS[-1])
print("ПОСЛЕДНИЙ ОТВЕТ: ", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# вызов  сообщений в thread  первое сообщение thread
#     1                   2  'Запомните: моя смена начинается в 7:40.'
#     2                   4  'Запомните: моя смена начинается в 7:40.'
#     3                   4  'Опишите порядок приёмки паллет на скл...'
#     4                   4  'А как оформлять брак, найденный при п...'
#
# ПОСЛЕДНИЙ ВОПРОС: Во сколько начинается моя смена?
# ПОСЛЕДНИЙ ОТВЕТ:  Из нашего диалога мне неизвестно время начала вашей смены — эта информация не упоминалась. Пожалуйста, уточните расписание у своего супервизора или проверьте график смен во внутренней системе склада.

Число сообщений в thread доходит до четырёх и дальше не меняется, сколько бы вызовов ни было. В этом главное свойство приёма: история, которую получает модель, ограничена сверху. Колонка с первым сообщением показывает, как граница обрезки сдвигается вперёд по диалогу. При последнем вопросе модель получает только то, что осталось внутри этой границы.

Вашей реплики со временем смены внутри границы уже нет. В запуске выше модель ответила, что времени смены в диалоге не было. Гарантии такого ответа нет: при пропавшем факте модель может уверенно назвать время, которого не получала. Просьба в системном промпте сказать, что факта не было, этого не исключает.

Обрезка меняет состояние thread. Хук возвращает правку, редьюсер её применяет, и новый checkpoint записывается уже без выброшенных сообщений, так что на следующих вызовах модель их не получит. В прежних checkpoints этого thread они остаются: каждый checkpoint хранит полный набор каналов, и get_state_history отдаёт их, пока thread не удалён методом delete_thread.

Значит, копия истории до обрезки, например для разбора жалоб, у вас уже есть. Но и персональные данные обрезка из хранилища не убирает. В InMemorySaver и SqliteSaver их можно убрать только методом delete_thread, а он удаляет thread целиком.

Обрезка не стоит ни одного лишнего токена. Платите вы другим: всё, что вышло за границу обрезки, модель больше не получает.

Суммаризация: дороже и с пересказом

Второй способ, это заменить старую часть диалога кратким пересказом, сводкой. Для этого есть встроенный SummarizationMiddleware. Перед каждым вызовом модели он сверяет историю с порогом. Если порог достигнут, отдельный вызов модели пишет сводку, и она встаёт в историю вместо сжатой части.

Когда сжимать и сколько оставить, задают trigger и keep.

1) trigger, условие срабатывания. Один порог задаётся кортежем: ("tokens", 4000), ("messages", 10) или ("fraction", 0.8), это доля контекстного окна. В словаре {"tokens": 4000, "messages": 10} должны выполниться все пороги сразу, в списке порогов достаточно одного. Без trigger сжатие не срабатывает ни разу

2) keep, какую часть конца истории оставить нетронутой. Задаётся одним кортежем того же вида, по умолчанию ("messages", 20), то есть двадцать последних сообщений. Словарь и список здесь не принимаются

Кроме порогов, SummarizationMiddleware получает свою модель, обязательный параметр model: сводку пишет отдельный вызов, и его можно отдать модели подешевле основной. Вид fraction и в trigger, и в keep считается от размера контекстного окна модели сводки, той, что передана в model. Основная модель агента в этом счёте не участвует. Нужен профиль модели сводки, без него middleware падает с ошибкой при сборке. Профили разобраны в уроке 4.

Пример 06_summarize.py

"""Пример 6 урока 12: суммаризация вместо обрезки.

Тот же диалог, что в примерах 4 и 5. Старые сообщения здесь
заменяются сводкой, её пишет отдельный вызов модели внутри встроенного
middleware SummarizationMiddleware.

Middleware срабатывает при шести сообщениях в thread. После этого в thread
остаются два последних сообщения, а перед ними встаёт сводка.

Сводку пишет middleware по собственному промпту, system_prompt агента на неё
не действует. Поэтому промпт сводки задан отдельно, параметром summary_prompt,
и на русском языке.
"""

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver

from course_model import build_model

SUMMARY_PROMPT = (
    "Кратко перескажите по-русски историю переписки ниже. Сохраните все "
    "факты, которые пользователь просил запомнить, и принятые в диалоге "
    "решения.\n\n"
    "Переписка:\n{messages}"
)

agent = create_agent(
    model=build_model(temperature=0, max_tokens=384),
    tools=[],
    system_prompt=(
        "Вы помощник склада. Отвечайте по-русски, тремя-четырьмя развёрнутыми "
        "фразами. Если нужного факта в диалоге не было, так и скажите."
    ),
    middleware=[
        SummarizationMiddleware(
            model=build_model(temperature=0, max_tokens=512),
            trigger=("messages", 6),
            keep=("messages", 2),
            summary_prompt=SUMMARY_PROMPT,
        )
    ],
    checkpointer=InMemorySaver(),
)

CONFIG = {"configurable": {"thread_id": "sum-1"}}

QUESTIONS = [
    "Запомните: моя смена начинается в 7:40.",
    "Опишите порядок приёмки паллет на складе.",
    "А как оформлять брак, найденный при приёмке?",
    "Во сколько начинается моя смена?",
]


def find_summary(messages):
    """Сообщение со сводкой, если middleware её уже собрал."""
    for message in messages:
        if message.additional_kwargs.get("lc_source") == "summarization":
            return message
    return None


print(f"{'вызов':>5}  {'сообщений в thread':>18}  сводка в thread")

for number, question in enumerate(QUESTIONS, start=1):
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        CONFIG,
    )
    summary = find_summary(result["messages"])
    print(f"{number:>5}  {len(result['messages']):>18}  {summary is not None}")

print()
summary = find_summary(result["messages"])
if summary is None:
    print("СВОДКА: middleware не сработал, условие срабатывания не выполнилось")
else:
    print("СВОДКА:")
    print(summary.text)
    print()
    # После сводки в thread лежит хвост, сохранённый по keep, и ответ на последний вопрос.
    print("ХВОСТ, СОХРАНЁННЫЙ ПОСЛЕ СВОДКИ:")
    for message in result["messages"][1:-1]:
        print(f"  [{message.type}] {message.text.replace(chr(10), ' ')}")

print()
print("ПОСЛЕДНИЙ ВОПРОС:", QUESTIONS[-1])
print("ПОСЛЕДНИЙ ОТВЕТ: ", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# вызов  сообщений в thread  сводка в thread
#     1                   2  False
#     2                   4  False
#     3                   6  False
#     4                   4  True
#
# СВОДКА:
# Here is a summary of the conversation to date:
#
# Вот краткий пересказ переписки:
#
# 1. Пользователь попросил запомнить, что его смена начинается в **7:40**. Ассистент подтвердил запоминание этого факта.
# 2. Пользователь спросил о порядке приёмки паллет на складе. Ассистент ответил, что такой информации в диалоге нет, и порекомендовал обратиться к внутренней документации, напомнив о времени начала смены.
# 3. Пользователь спросил, как оформлять брак, найденный при приёмке. Ассистент не предоставил ответа на этот вопрос (в переписке ответ отсутствует).
#
# **Принятые решения и запомненные факты:**
# - Запомнено: смена пользователя начинается в 7:40.
# - Вопросы о складских процедурах (приёмка паллет и оформление брака) остались без ответа от ассистента — рекомендовано обратиться к внутренним инструкциям.
#
# ХВОСТ, СОХРАНЁННЫЙ ПОСЛЕ СВОДКИ:
#   [ai] Извините, но в нашем диалоге не было информации о порядке оформления брака при приёмке. Если у вас есть внутренние документы склада или регламенты на этот счёт, я могу помочь их структурировать. Ваша смена начинается в 7:40, и, возможно, к этому времени стоит уточнить процедуру у старшего смены.
#   [human] Во сколько начинается моя смена?
#
# ПОСЛЕДНИЙ ВОПРОС: Во сколько начинается моя смена?
# ПОСЛЕДНИЙ ОТВЕТ:  Ваша смена начинается в 7:40, и я запомнил это время. Пожалуйста, учитывайте его при планировании рабочих задач. Если понадобится уточнить другие детали графика, обращайтесь.

Сводку пишет middleware своим промптом, system_prompt агента на этот вызов не распространяется. Промпт по умолчанию лежит в коде пакета на английском, документация его текст не приводит. Он требует четыре раздела с английскими заголовками: SESSION INTENT, SUMMARY, ARTIFACTS и NEXT STEPS. Язык ответа в нём не задан. Поэтому русскоязычному агенту промпт сводки задают свой, параметром summary_prompt. Шаблон обязан содержать плейсхолдер {messages}. В примере выше шаблон написан на русском.

По выводу выше видно, что сводка пришла на русском, как и просит русский шаблон. Но первой строкой в ней стоит английская фраза Here is a summary of the conversation to date:. Её добавляет сам middleware перед текстом от модели, и шаблон summary_prompt на неё не влияет. Если сводку показывают пользователю, уберите эту строку сами, прежде чем выводить текст.

В thread после срабатывания middleware первым стоит сообщение с ролью человека, в нём лежит текст сводки. Внутри middleware это сделано так: сначала RemoveMessage с REMOVE_ALL_MESSAGES, потом сообщение со сводкой, потом сохранённый хвост. Пометка lc_source в служебном поле сообщения отличает сводку от реплики пользователя, по ней пример её и находит.

Если на границе сжатия оказался результат инструмента, middleware сдвигает границу, чтобы пара из вызова и результата не разъехалась. В своей обрезке за парами следите сами. В примере 5 срез начинается с реплики человека: вызовы инструментов и их результаты идут после неё, поэтому в срез они попадают вместе.

Суммаризация обходится дороже обрезки. Каждое срабатывание, это лишний вызов модели со своей задержкой и своим счётом. Кроме того, сводку модель пишет своими словами, и что в неё попадёт, заранее неизвестно.

Есть и ограничение по объёму. Модель сводки получает только конец сжимаемой части: параметр trim_tokens_to_summarize по умолчанию оставляет последние 4000 токенов. Всё, что старше, в том числе прежняя сводка, в новую сводку не попадёт, и ошибки при этом не будет. Значение None снимает это ограничение.

Порог проверяется перед вызовом модели. В третьем вызове перед моделью в thread пять сообщений, порог в шесть не достигнут, и шестым становится ответ. В четвёртом вызове перед моделью семь сообщений, и middleware срабатывает.

При keep=("messages", 2) после сжатия в thread остались два последних сообщения: ответ про брак и вопрос про смену. Время 7:40 в этом запуске стоит в двух местах: первым пунктом сводки и в самом ответе про брак. Какой из двух источников сработал, по выводу не сказать. В другом запуске сводка может выйти иначе и факт потерять, поэтому важное проверяйте запуском на своей модели.

Если в истории есть изображения, звук или видео, middleware их не уменьшает. Сообщения из сжатой части остаются в истории только текстом сводки, а в сохранённом хвосте вложения стоят как есть. Если изображений в диалоге много, храните файлы отдельно, а в истории передавайте ссылку на них.

Какой приём выбрать

Приёмы можно сочетать. Например, суммаризация не даёт истории разрастись, а точечное удаление заранее убирает сообщения, которые не должны попасть даже в сводку.

Распространённые ошибки

Ошибка 1: чекпойнтер задали, а thread_id не передали

# Неправильно: агент с памятью вызван без конфигурации thread
agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver())
agent.invoke({"messages": [{"role": "user", "content": question}]})

Что происходит: вызов падает с ValueError и текстом "Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id".

Почему так: без thread_id чекпойнтеру не по чему искать checkpoint. Граф до начала работы проверяет, что в вызов передан словарь configurable, и без него сразу падает. Thread со случайным именем граф не создаёт, поэтому вызов падает с ошибкой.

# Правильно: имя thread приходит из вашего приложения
config = {"configurable": {"thread_id": "support-4412"}}
agent.invoke({"messages": [{"role": "user", "content": question}]}, config)

Ошибка 2: историю передают руками поверх чекпойнтера

# Неправильно: приложение хранит свою копию истории словарями и отправляет её целиком
history.append({"role": "user", "content": question})
agent.invoke({"messages": history}, config)

Что происходит: в thread появляются копии сообщений, и модель получает диалог с дублями. Ошибки при этом нет, счёт за токены растёт, а качество ответов падает.

Почему так: редьюсер add_messages добавляет пришедшее к тому, что уже лежит в checkpoint. Сообщения с известным идентификатором он заменяет, а словари приходят без идентификаторов и становятся новыми записями.

# Правильно: во вход идёт только новая реплика, историю подставляет чекпойнтер
agent.invoke({"messages": [{"role": "user", "content": question}]}, config)

Ошибка 3: обрезка разорвала пару вызова и результата

# Неправильно: срез по числу сообщений, без оглядки на их роли
kept = state["messages"][-3:]
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *kept]}

Что происходит: в срез попадает ToolMessage, а сообщение модели с соответствующим tool_calls осталось за границей. Провайдер отвечает ошибкой о несогласованной истории. Ваш код при этом отработал без ошибок, падает запрос к модели.

Почему так: результат инструмента связан со своим вызовом полем tool_call_id. Провайдер принимает результат, только если этот вызов стоит в ближайшем перед результатами сообщении модели, иначе запрос возвращается с ошибкой. Кроме того, часть провайдеров принимает историю, только если она начинается с реплики человека.

# Правильно: границу двигают, пока история не станет валидной
kept = state["messages"][-3:]
while kept and kept[0].type != "human":
    kept = kept[1:]

Практическое задание

Соберите помощника night_shift.py, который ведёт диалог в thread на диске и не даёт ему разрастись.

Требования:

1) чекпойнтер SqliteSaver на файл рядом со скриптом, имя thread задаётся аргументом командной строки

2) один инструмент shift_rules(topic), который отдаёт короткое правило по теме из словаря в коде

3) SummarizationMiddleware, который срабатывает при десяти сообщениях и сохраняет четыре последних, модель для сводки берите ту же, что у агента

4) цикл из шести вопросов, где первый сообщает факт о пользователе, а последний этот факт запрашивает

5) после каждого вызова печатайте строку: номер вызова, число сообщений в thread, число checkpoints в thread, входные токены последнего запроса к модели

6) в конце печатайте, появилась ли в thread сводка, и первые двести знаков её текста

Как проверить результат:

1) второй запуск программы с тем же именем thread продолжает диалог: в первой строке вывода сообщений в thread больше двух, а checkpoints больше, чем в первой строке первого запуска

2) запуск с другим именем thread начинает диалог с нуля

3) число сообщений в thread после срабатывания middleware становится меньше, чем было после предыдущего вызова

4) уберите checkpointer из сборки и подсчёт checkpoints из печати. Ответ на последний вопрос перестанет опираться на факт из первого: модель скажет, что его не было, или назовёт другое значение

Подсказка: число checkpoints считается как len(list(agent.get_state_history(config))), а входные токены последнего запроса к модели берутся из usage_metadata последнего сообщения модели.

Итоги урока

Память сессии включается одним параметром и одним ключом конфигурации: checkpointer при сборке агента и thread_id при вызове. Дальше фреймворк читает состояние в начале вызова и записывает после каждого шага, а вы отправляете только новую реплику.

Checkpoint записывается после каждого супершага. У него есть адрес, родитель, служебные поля и полный набор каналов, поэтому историю thread можно прочитать целиком: get_state даёт последний checkpoint, get_state_history даёт всю цепочку.

Носитель вы выбираете сами. InMemorySaver работает до конца процесса и годится для отладки, SqliteSaver кладёт thread в файл, PostgresSaver ставят для продакшна. Момент записи задаёт durability: "exit" бережёт время, "sync" бережёт данные.

За память вы платите растущим числом входных токенов. Каждый вызов отправляет модели весь диалог, поэтому длинный thread дорожает сам по себе, даже когда вопросы остаются короткими. Обрезка ограничивает текущее состояние thread сверху, а удалённое остаётся в прежних checkpoints до вызова delete_thread. Суммаризация сохраняет смысл ранних реплик, а платите вы за это лишним вызовом модели и неточностью пересказа.

У памяти сессии есть предел: всё сохранённое привязано к одному thread. В соседнем thread у агента нет ни имени пользователя, ни его предпочтений, ни договорённостей вчерашнего диалога.

В уроке 13 разберу хранилище, общее для всех threads: как агент записывает в него факты о пользователе и читает их. Там же покажу, чем пространство имён отличается от thread_id, какие факты стоит хранить долго и как не копить в памяти лишнее.

Код урока

Примеры этого урока лежат в репозитории курса, папка lesson_12. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.

Предыдущий урок: Первый агент: цикл, состояние и рамка настройки

Следующий урок: Долгая память: хранилище между сессиями

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

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