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

Цель урока

Вы поймёте, почему LLM не помнят предыдущие сообщения. Научитесь вручную управлять историей диалога и подключать RunnableWithMessageHistory, чтобы память между запросами велась автоматически и изолированно по сессиям.

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

Уроки 2–3 (LCEL, цепочки, шаблоны промптов)
Базовое понимание списков и словарей в Python

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

Stateless LLM (без сохранения состояния)
Ручное управление историей с помощью списка сообщений
ChatMessageHistory и InMemoryChatMessageHistory
MessagesPlaceholder
RunnableWithMessageHistory
Сессии и session_id

Почему модель не помнит предыдущие сообщения

LLM stateless по природе, где каждый вызов .invoke() - это независимый HTTP запрос к API. Модель получает входные данные, возвращает ответ и ничего не сохраняет.

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

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

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

model.invoke("Меня зовут Алексей.")
response = model.invoke("Как меня зовут?")
print(response.content)
# "Я не знаю вашего имени, так как у меня нет информации о вас. ..." — модель не помнит первое сообщение

Ручное управление историей

Самый простой подход это хранить историю в обычном списке и передавать его в модель.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

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

history = [
    SystemMessage(content="Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
]

def chat(user_input: str) -> str:
    history.append(HumanMessage(content=user_input))
    response = model.invoke(history)
    history.append(response)  # response уже AIMessage
    return response.content

print(chat("Меня зовут Михаил."))
# "Привет, Михаил! 👋 Рад познакомиться. Чем я могу помочь?"

print(chat("Как меня зовут?"))
# "Тебя зовут Михаил! 😊"

print(chat("Напиши функцию на Python для вычисления факториала."))
# def factorial(n): ...

Каждый раз в model.invoke() передается весь список history. Модель видит полный контекст диалога.

Проблемы ручного подхода:

Одна глобальная история, она не масштабируется для нескольких пользователей
Нет изоляции между сессиями
Историю нужно везде передавать

ChatMessageHistory

LangChain предоставляет объект для хранения истории сообщений.

ChatMessageHistory - это просто обертка над списком сообщений. Её можно сериализовать, сохранить в БД или Redis. Сейчас воспользуемся её базовой версией.

from langchain_community.chat_message_histories import ChatMessageHistory

history = ChatMessageHistory()

history.add_user_message("Меня зовут Алексей.")
history.add_ai_message("Привет, Алексей!")
history.add_user_message("Как меня зовут?")

print(history.messages)
# [HumanMessage(content='Меня зовут Алексей.'),
#  AIMessage(content='Привет, Алексей!'),
#  HumanMessage(content='Как меня зовут?')]

RunnableWithMessageHistory

RunnableWithMessageHistory — обертка над любым Runnable, которая автоматически:

Загружает историю нужной сессии перед вызовом
Добавляет новое сообщение пользователя и ответ модели в историю после вызова

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

load_dotenv()

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
    MessagesPlaceholder(variable_name="history"),  # сюда подставится история
    ("human", "{input}"),
])

chain = prompt | model

# Хранилище сессий: session_id -> ChatMessageHistory
store: dict[str, ChatMessageHistory] = {}


def get_session_history(session_id: str) -> ChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


# Оборачиваем цепочку
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",       # ключ со входным сообщением
    history_messages_key="history",   # ключ для MessagesPlaceholder
)

Теперь общаемся через chain_with_history.invoke(), передавая config с session_id:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | model

store: dict[str, ChatMessageHistory] = {}


def get_session_history(session_id: str) -> ChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)


config_alice = {"configurable": {"session_id": "alice"}}
config_bob   = {"configurable": {"session_id": "bob"}}

# Диалог с Алисой
response = chain_with_history.invoke(
    {"input": "Меня зовут Алиса."},
    config=config_alice,
)
print(response.content)
# "Привет, Алиса!"

response = chain_with_history.invoke(
    {"input": "Как меня зовут?"},
    config=config_alice,
)
print(response.content)
# "Тебя зовут Алиса."

# Боб - отдельная сессия, не видит диалог Алисы
response = chain_with_history.invoke(
    {"input": "Как меня зовут?"},
    config=config_bob,
)
print(response.content)
# "Извини, я не знаю твоего имени — ты ещё не представился."

Каждый session_id это изолированная история. Это базовая модель для многопользовательских чат приложений.

Потоковый вывод с историей

RunnableWithMessageHistory поддерживает .stream() так же, как обычный Runnable.

История обновляется автоматически после того, как стриминг завершен.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | model
store: dict[str, ChatMessageHistory] = {}


def get_session_history(session_id: str) -> ChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

config = {"configurable": {"session_id": "alice"}}

print("Ответ: ", end="")
for chunk in chain_with_history.stream(
    {"input": "Напиши функцию на Python для бинарного поиска с комментариями."},
    config=config,
):
    print(chunk.content, end="", flush=True)
print()

Ограничение длины истории

Бесконечная история это проблема, рано или поздно она превысит контекстное окно модели и увеличит стоимость запросов. Нужно обрезать историю до разумного размера.

Простой способ, хранить только последние N сообщений.

from langchain_community.chat_message_histories import ChatMessageHistory

store: dict[str, ChatMessageHistory] = {}

def get_limited_session_history(session_id: str, max_messages: int = 10) -> ChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()

    history = store[session_id]

    # Если сообщений больше лимита - обрезаем, оставляем последние max_messages
    if len(history.messages) > max_messages:
        history.messages = history.messages[-max_messages:]

    return history

LangChain также предоставляет готовый инструмент trim_messages для управления размером контекста.

trim_messages обрезает историю по количеству токенов, а не сообщений.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import trim_messages
from langchain_core.runnables import RunnablePassthrough

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

trimmer = trim_messages(
    max_tokens=20,            # максимум сообщений в истории
    strategy="last",          # оставляем последние сообщения
    token_counter=len,        # считаем по количеству сообщений, не токенов
    include_system=True,      # системное сообщение сохраняем всегда
    allow_partial=False,      # не обрезать сообщения посередине
    start_on="human",         # история начинается с HumanMessage
)

# Встраиваем trimmer в цепочку перед моделью
chain = (
    RunnablePassthrough.assign(history=lambda x: trimmer.invoke(x["history"]))
    | prompt
    | model
)

Персистентная история: хранение между перезапусками

ChatMessageHistory хранит данные в памяти, при перезапуске приложения история стирается. В продакшн история хранится в базе данных.

LangChain поддерживает несколько бэкендов.

Пример с SQLite:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты дружелюбный ассистент-разработчик. Отвечай кратко."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | model


def get_persistent_session_history(session_id: str) -> SQLChatMessageHistory:
    return SQLChatMessageHistory(
        session_id=session_id,
        connection="sqlite:///chat_history.db",  # путь к файлу БД
    )


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_persistent_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

Интерфейс не изменился, функция get_session_history теперь возвращает объект, который читает и пишет в SQLite. После перезапуска история сохраняется.

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

1. Ключ history_messages_key не совпадает с MessagesPlaceholder

# Ошибка: в промпте variable_name="history", но в обертке history_messages_key="chat_history"
MessagesPlaceholder(variable_name="history")
RunnableWithMessageHistory(..., history_messages_key="chat_history")  # не совпадает!

Решение: убедитесь, что variable_name в MessagesPlaceholder и history_messages_key в RunnableWithMessageHistory одна и та же строка.

2. Нет MessagesPlaceholder в промпте

KeyError: 'history'

Причина: в промпте нет места для подстановки истории.

Решение: добавьте MessagesPlaceholder(variable_name="history") между сообщениями system и human.

3. Один store на всё приложение

store = {}  # глобальный словарь — ок для разработки

В продакшн это не масштабируется. Используйте Redis, PostgreSQL или другой персистентный бэкенд.

4. Рост истории без ограничений

Симптом: со временем запросы становятся медленнее и дороже.

Решение: добавьте trim_messages в цепочку или ограничь количество хранимых сообщений в get_session_history.

Рабочий пример кода

Пример демонстрирует все концепции урока: RunnableWithMessageHistory, изоляцию сессий, ограничение истории и персистентное хранение в SQLite.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import trim_messages
from langchain_core.runnables import RunnablePassthrough
from langchain_community.chat_message_histories import SQLChatMessageHistory

load_dotenv()

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

# Ограничение истории — не более 20 сообщений
trimmer = trim_messages(
    max_tokens=20,
    strategy="last",
    token_counter=len,
    include_system=True,
    allow_partial=False,
    start_on="human",
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты ассистент-разработчик. Отвечай кратко и по делу."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = (
    RunnablePassthrough.assign(history=lambda x: trimmer.invoke(x["history"]))
    | prompt
    | model
)

# История сохраняется в SQLite — не стирается после перезапуска
def get_session_history(session_id: str) -> SQLChatMessageHistory:
    return SQLChatMessageHistory(
        session_id=session_id,
        connection="sqlite:///chat_history.db",
    )

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

def chat(session_id: str, message: str) -> str:
    response = chain_with_history.invoke(
        {"input": message},
        config={"configurable": {"session_id": session_id}},
    )
    return response.content

# Изоляция сессий: alice и bob не видят историю друг друга
print("=== Сессия: alice ===")
print(chat("alice", "Меня зовут Алиса, я пишу на Python."))
print(chat("alice", "Какой язык я упомянула?"))

print("\n=== Сессия: bob ===")
print(chat("bob", "Привет! Я разрабатываю на Go."))
print(chat("bob", "Что я сказал о своём стеке?"))

print("\n=== Алиса снова ===")
print(chat("alice", "Посоветуй библиотеку для HTTP-запросов на моём языке."))
# Модель помнит, что Алиса пишет на Python → посоветует requests/httpx

История сохраняется в chat_history.db. При повторном запуске сессии alice и bob продолжатся с того же места.

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

Создайте консольный чат бот с памятью и двумя режимами работы.

Требования:

1) Бот должен поддерживать два режима, задаваемых при старте:

"tutor" - терпеливый преподаватель Python, объясняет концепции с примерами
"reviewer" - строгий code reviewer, указывает на проблемы и предлагает улучшения

2) Реализуйте хранение истории с помощью RunnableWithMessageHistory с изоляцией по session_id

3) Добавьте команду /history - выводит количество сообщений в текущей сессии

4) Добавьте команду /clear - очищает историю текущей сессии

5) Ограничьте историю: не более 20 сообщений (используй trim_messages или обрезку в get_session_history)

Пример сессии:

Режим (tutor/reviewer): tutor
Session ID: user_42

Вы: Объясни list comprehension
Бот: List comprehension — это краткий способ создать список...

Вы: Покажи пример с условием
Бот: Конечно, вот пример с фильтрацией... (помнит контекст)

Вы: /history
История: 4 сообщения

Вы: /clear
История очищена.

Вы: Что я спрашивал раньше?
Бот: Ты ещё ничего не спрашивал — история пуста.

Итоги уроком

Сейчас цепочки обрабатывают данные линейно: вход → обработка → выход. Но многие задачи требуют ветвления, выбрать стратегию в зависимости от контекста, запустить несколько шагов параллельно, объединить результаты. В следующем уроке разберем RunnableParallel, RunnableBranch и RunnableLambda - это инструменты для построения нелинейных пайплайнов.

<< Урок 3

Урок 5 >>

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

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