Память диалога: checkpointer, thread_id, обрезка истории — AI агенты на Python: шаг за шагом

Пользователь пишет агенту поддержки: «Меня зовут Анна, мой заказ 1042». Через минуту: «А когда он приедет?». Если агент не помнит первое сообщение, он переспросит номер заказа, и клиент справедливо решит, что с ним разговаривает автомат. Без памяти диалог превращается в череду несвязанных вопросов.

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

Что уже умеем

Вы создаёте агента через create_agent, даёте ему инструменты и задаёте поведение системным промптом. Из урока 1.4 вы помните, что модель сама ничего не запоминает: историю нужно передавать в каждом запросе, а из урока 1.2 — что длина истории напрямую влияет на стоимость. Сегодня поручим хранение истории LangChain и научимся её обрезать.

Что вы узнаете в рамках статьи

почему агент без памяти забывает предыдущие сообщения;
как подключить память через checkpointer и InMemorySaver;
как thread_id разделяет диалоги разных пользователей;
как обрезать историю своим middleware с @before_model;
как сжимать старую часть диалога через SummarizationMiddleware.

Ключевые термины

Краткосрочная память — история сообщений одного диалога, которую агент учитывает при ответе.
Checkpointer — объект, который сохраняет состояние агента после каждого шага и восстанавливает его при следующем вызове.
thread_id — идентификатор диалога; у каждого разговора свой, и истории разных диалогов не смешиваются.
Middleware — функция или объект, который встраивается в цикл агента и выполняется в определённый момент, например перед вызовом модели.
Суммаризация — замена старых сообщений их кратким пересказом, сделанным моделью.

Агент без памяти

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

from langchain.agents import create_agent

agent = create_agent(model="openai:gpt-6-luna", tools=[], system_prompt="Отвечай кратко.")

agent.invoke({"messages": [{"role": "user", "content": "Меня зовут Анна."}]})
result = agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]})
print(result["messages"][-1].content)

Что увидим в консоли: «Я не знаю вашего имени, вы его не называли». Второй вызов ничего не знает о первом. Можно было бы, как в уроке 1.4, хранить список сообщений самим и передавать его целиком, но в LangChain для этого есть готовый механизм.

Checkpointer и thread_id

Память подключается одним параметром checkpointer. Для обучения и тестов подходит InMemorySaver — он хранит состояние в оперативной памяти процесса. При каждом вызове нужно передать config с thread_id: по нему агент находит сохранённую историю и дописывает в неё новые сообщения.

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[],
    system_prompt="Отвечай кратко.",
    checkpointer=InMemorySaver(),
)

anna = {"configurable": {"thread_id": "anna-1"}}
agent.invoke({"messages": [{"role": "user", "content": "Меня зовут Анна."}]}, anna)
result = agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, anna)
print(result["messages"][-1].content)
print(len(result["messages"]))

Что увидим: «Вас зовут Анна» и число 4. Во втором вызове мы передали только новое сообщение, а агент сам достал из checkpointer два сообщения первого обмена и добавил к ним два новых. В result["messages"] теперь лежит вся история диалога, а не только последний обмен.

Обратите внимание: с памятью в invoke передаётся только новая реплика пользователя. Если по привычке отправить всю историю заново, сообщения задвоятся.

Разные пользователи — разные потоки

Один агент обслуживает многих пользователей одновременно. Их разговоры разделяет thread_id: для каждого диалога — свой идентификатор, обычно связанный с id пользователя или сессии.

boris = {"configurable": {"thread_id": "boris-1"}}
result = agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, boris)
print(result["messages"][-1].content)

result = agent.invoke({"messages": [{"role": "user", "content": "Напомни, как меня зовут?"}]}, anna)
print(result["messages"][-1].content)

Что увидим: Борису агент ответит, что не знает его имени, а Анне — что её зовут Анна. Потоки полностью изолированы. Если по ошибке дать всем пользователям один thread_id, агент будет путать их данные — это и неудобно, и небезопасно.

InMemorySaver теряет всё при перезапуске программы. В рабочих системах используют checkpointer с базой данных, например PostgresSaver из пакета langgraph-checkpoint-postgres. Код агента при этом не меняется — меняется только объект в параметре checkpointer.

Обрезка истории через @before_model

С памятью каждый новый вопрос обходится дороже предыдущего: модели уходит вся накопленная история. Простое решение — перед каждым вызовом модели оставлять только последние сообщения. Для этого в LangChain есть middleware: функция с декоратором @before_model выполняется перед каждым обращением к модели и может изменить состояние.

from typing import Any
from langchain.agents import AgentState
from langchain.agents.middleware import before_model
from langchain.messages import HumanMessage, RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime

MAX_MESSAGES = 6

@before_model
def keep_last_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    messages = state["messages"]
    if len(messages) <= MAX_MESSAGES:
        return None  # история короткая, ничего не меняем
    recent = messages[-MAX_MESSAGES:]
    while not isinstance(recent[0], HumanMessage):
        recent = recent[1:]  # история должна начинаться с реплики пользователя
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *recent]}

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[],
    system_prompt="Отвечай кратко.",
    checkpointer=InMemorySaver(),
    middleware=[keep_last_messages],
)

Что увидим: пока в диалоге не больше шести сообщений, всё работает как раньше. Дальше старые сообщения удаляются, и в result["messages"] их становится не больше шести. RemoveMessage(id=REMOVE_ALL_MESSAGES) очищает список, а следом записываются оставшиеся сообщения. Цикл while сдвигает начало до сообщения пользователя: так история не начнётся с ToolMessage, оторванного от вызова, который его породил, — многие провайдеры такую историю отклоняют.

Цена простоты — забывчивость: если Анна представилась десять сообщений назад, после обрезки агент её имени уже не вспомнит. Удалённые сообщения пропадают и из checkpointer.

Суммаризация старых сообщений

Чтобы не терять важные факты, старую часть истории можно не выбрасывать, а пересказать. Готовый SummarizationMiddleware следит за размером истории и, когда он превышает порог, заменяет старые сообщения кратким пересказом, оставляя последние без изменений.

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[],
    system_prompt="Отвечай кратко.",
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-6-luna",
            trigger=("tokens", 4000),
            keep=("messages", 10),
        ),
    ],
)

Что увидим: пока история меньше 4 000 токенов, ничего не происходит. Когда порог превышен, middleware просит модель пересказать всё, кроме последних 10 сообщений, и заменяет их одним сообщением с пересказом. Имя Анны и номер заказа сохранятся в пересказе, а история снова станет короткой. Пересказ — это отдельный вызов модели, он тоже стоит денег, но происходит редко, а экономия на каждом следующем шаге его окупает.

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

Забыть передать config с thread_id. Агент с checkpointer без thread_id не знает, какую историю загрузить, и вызов завершится ошибкой.
Давать всем пользователям один thread_id. Истории смешаются, и агент сможет пересказать одному клиенту данные другого.
Передавать всю историю при включённой памяти. Checkpointer уже хранит прошлые сообщения. Отправляйте в invoke только новую реплику.
Обрезать историю посреди вызова инструмента. Если история начнётся с ToolMessage без соответствующего AIMessage, модель вернёт ошибку. Начинайте обрезанную историю с сообщения пользователя.

Наиболее частые вопросы

Чем thread_id отличается от previous_response_id из урока 1.4? — Идея та же — продолжить диалог, но историю хранит ваш checkpointer, а не сервер OpenAI. Поэтому вы сами решаете, что хранить и что обрезать.
Сохранится ли память после перезапуска программы? — С InMemorySaver нет. Для постоянного хранения используйте checkpointer с базой данных.
Что выбрать: обрезку или суммаризацию? — Обрезка проще и бесплатна, подходит для коротких разговоров. Суммаризация нужна, когда важно помнить факты из начала длинного диалога.
Помнит ли агент Анну в новом диалоге с другим thread_id? — Нет, краткосрочная память живёт внутри одного потока. Память между диалогами — тема урока 3.5.

Итоги

Отличный результат! Ваш агент теперь ведёт настоящий диалог: помнит, что говорил пользователь, не путает собеседников и не раздувает контекст до бесконечности. Вы подключили checkpointer, разобрались с thread_id и написали первый собственный middleware, а заодно познакомились с готовой суммаризацией.

Middleware — мощный механизм, и обрезка истории лишь первый пример. В следующем уроке вы научитесь получать от агента ответ строго по схеме Pydantic и с помощью middleware ограничите число шагов агента и обработаете ошибки инструментов.