История чатов, факты, состояние рабочего процесса и контрольные точки — это четыре разных хранилища.

Часть 9 из 14: Отдельная история чатов, сохранённые факты и данные рабочего процесса с возможностью продолжения

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

В предыдущей части была реализована функция поиска в справочнике: запрос к отобранному корпусу данных, сохранение метаданных происхождения и блокировка советов, основанных на документах, которые поиск не вернул.

Кто-то из сотрудников, находящихся в режиме дежурства, спрашивает, может ли система «помнить» инцидент на завтра. Вопрос сформулирован недостаточно чётко: сохранять ли ходы чата? Предпочтения команды? Метки и извлечённые фрагменты текста? Запись, ожидающая одобрения? Все промежуточные поля графа? Люди объединяют всё это в одну расплывчатую категорию. Каждому элементу нужны собственный ключ, время действия, правила доступа и политика обработки сбоев.

В этой части рассматриваются четыре отдельные идеи:

chat history
  ordered messages for one conversation

saved facts
  selected application data about a user or accountworkflow state
  the current named values for one runcheckpoint
  a saved snapshot of workflow state that can be loaded later

Передача предыдущих сообщений в статическую версию LangChain, готовую к запуску, не обеспечивает автоматической возможности приостановки и продолжения работы. Функции долговечных потоков и создания снимков реализованы благодаря модели состояния и чекпоинтов LangGraph.

Актуальные задачи

Для примеров используем знакомый случай:

После выпуска в 14:05 запросы на оплату из региона ЕС терпят неудачу. Журналы checkout-api указывают на отказ базы данных в установлении новых соединений.

Каждому типу записей следует выделить собственное пространство имен:

chat session:    chat:INC-2048
user facts:      user-17
workflow thread: ticket:INC-2048

Если рассматривать идентификатор инцидента как идентификатор человека, это приводит к слиянию несвязанных пространств имен. Аналогичным образом единый список транскрипций смешивает отдельные случаи.

Во-первых, перестаньте говорить «память»

Укажите точное название записи, о которой идет речь.

История чатов

Это может быть упорядоченный список:

human: The failure began after 14:05.
assistant: I recorded the start time.
human: The failed requests are only in the EU region.
assistant: I added the affected region to the investigation context.

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

Сохранённые факты

Выбранные поля, такие как:

{
  "team": "commerce-platform",
  "timezone": "America/Los_Angeles"
}

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

Состояние рабочего процесса

Текущие данные для одного процесса обработки заявок:

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

Состояние меняется по мере выполнения шагов.

Точка контроля

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

Поиск не относится ни к одному из этих вариантов

Индекс руководства с отбором — это корпус для поиска. Открытие его во время инцидента не превращает найденные результаты в историю чата. Фрагменты текста не должны автоматически преобразовываться в постоянные данные профиля. Индексы эмбеддингов не являются базами данных для сохранения состояний. Изолируйте хранилища даже в том случае, если один HTTP-запрос затрагивает несколько из них.

Что по умолчанию запоминает фиксированная цепочка

При независимых вызовах цепочка ничего не запоминает, если только ваше приложение не вводит или не сохраняет контекст.

Этот вызов:

result = chain.invoke(current_input)

не автоматически передает предыдущие входные данные или результаты. Вы можете сами добавлять предыдущие элементы истории или использовать специальные обертки для старых данных. В версии LangChain, рассмотренной здесь, RunnableWithMessageHistory выдаёт предупреждения и направляет новую работу на сохранение в LangGraph.

При использовании статической цепочки обычно проще управлять историей в коде приложения:

read permitted messages
  -> select the messages needed for this request
  -> call the chain
  -> store the new turn under the correct session ID

Именно это и делает вспомогательный код.

Структура проекта

Снимок версии из Части 9 содержит:

langchain-helpdesk/
├── app.py
├── checkpoint_graph.py
├── facts.py
├── history.py
└── tests/
    └── test_state.py

Установка пакетов:

python -m pip install -U langchain-core langgraph pydantic pytest

В примере не выполняются вызовы к поставщикам услуг.

Шаг 1: Хранение сообщений чата по сессиям

Создайте файл history.py:

from dataclasses import dataclass, field

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
)
@dataclass
class ChatHistoryStore:
    histories: dict[str, list[BaseMessage]] = field(
        default_factory=dict
    )    def read(self, session_id: str) -> list[BaseMessage]:
        return list(self.histories.get(session_id, []))    def add_turn(
        self,
        session_id: str,
        user_text: str,
        reply_text: str,
    ) -> None:
        history = self.histories.setdefault(session_id, [])
        history.extend(
            [
                HumanMessage(content=user_text),
                AIMessage(content=reply_text),
            ]
        )    def prior_turn_count(self, session_id: str) -> int:
        return len(self.histories.get(session_id, [])) // 2

Ключи histories соотносят каждую сессию с отсортированным списком. Функция read создаёт копии, чтобы вызывающие код не могли изменить хранилище побочными эффектами. Функция add_turn фиксирует пару «человек/ассистент». Параметр prior_turn_count уменьшает длину списка вдвое, поскольку пример хранит только полные пары. В реальных транскрипциях также присутствуют сообщения инструментов, незавершённые записи и ошибки — в производственных условиях не стоит ожидать идеальной парности данных.

История требует правила хранения

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

Шаг 2: Хранение выбранных фактов отдельно

Создайте файл facts.py:

from dataclasses import dataclass, field

@dataclass
class UserFactsStore:
    records: dict[str, dict[str, str]] = field(
        default_factory=dict
    )    def put(self, user_id: str, key: str, value: str) -> None:
        self.records.setdefault(user_id, {})[key] = value    def get(self, user_id: str) -> dict[str, str]:
        return dict(self.records.get(user_id, {}))

Индексируйте факты по user_id, никогда не по идентификатору сессии или инцидента. Принимайте только поля с именами; никогда не храните целый текст переписки под одним ключом. Настоящие пути put требуют списков разрешённых значений, проверки данных, авторизации и записей аудита. Совет модели не является разрешением на хранение данных.

Шаг 3: Определение состояния рабочего процесса

Переходите к рабочему процессу с состояниями. Используйте TypedDict в файле checkpoint_graph.py:

from operator import add
from typing import Annotated, TypedDict

class TicketWorkflowState(TypedDict, total=False):
    ticket_id: str
    details: str
    classification: str
    recommendation: str
    audit: Annotated[list[str], add]

total=False позволяет полям оставаться пропущенными до тех пор, пока узел их не заполнит. Для событий аудита используется функция-сокращатель:

Annotated[list[str], add]

Когда узел возвращает больше строк аудита, функция-сокращатель их соединяет вместо того, чтобы перезаписывать. Выбирайте функции-сокращатели осознанно: используйте append для потоков событий и replace для скобичных полей.

Шаг 4: Создание небольших детерминированных узлов

В примере обучения графа используется обычный Python, поэтому поведение точек контроля видно наглядно:

def classify_node(state: TicketWorkflowState) -> TicketWorkflowState:
    details = state["details"].lower()

    if "database" in details or "connection refused" in details:
        category = "database"
    elif "access" in details or "role" in details:
        category = "access"
    else:
        category = "unknown"    return {
        "classification": category,
        "audit": [f"classified:{category}"],
    }

Каждый узел считывает состояние и возвращает исправления; он никогда не изменяет входный словарь. Узел рекомендаций использует результаты классификации:

def recommend_node(state: TicketWorkflowState) -> TicketWorkflowState:
    category = state["classification"]

    if category == "database":
        recommendation = (
            "Compare database settings with the last good release."
        )
    elif category == "access":
        recommendation = (
            "Confirm the requested role and current access policy."
        )
    else:
        recommendation = "Ask a person to classify the ticket."    return {
        "recommendation": recommendation,
        "audit": ["recommendation_created"],
    }

Это обычные функции Python — в этом разделе рассматриваются состояние и сохранение данных, а не точность классификатора.

Шаг 5: Сборка графа

from langgraph.graph import END, START, StateGraph

def build_checkpointed_graph(checkpointer=None):
    builder = StateGraph(TicketWorkflowState)
    builder.add_node("classify", classify_node)
    builder.add_node("recommend", recommend_node)
    builder.add_edge(START, "classify")
    builder.add_edge("classify", "recommend")
    builder.add_edge("recommend", END)    return builder.compile(
        checkpointer=checkpointer or InMemorySaver()
    )

StateGraph(TicketWorkflowState) связывает общее состояние с словарем определенного типа. Узлы и ребра задают порядок; функция compile проверяет структуру и подключает механизм контроля состояния. Маршрут остается линейным — граф функционален благодаря тому, что состояние и точки контроля считаются первостепенными элементами, а не из-за сложности диаграммы.

Шаг 6: Присвоить каждому рабочему процессу идентификатор потока

def thread_config(thread_id: str) -> dict[str, dict[str, str]]:
    return {"configurable": {"thread_id": thread_id}}

Запустите обработку заявки:

config = thread_config("ticket:INC-2048")

result = graph.invoke(
    {
        "ticket_id": "INC-2048",
        "details": (
            "checkout-api reports database connection refused"
        ),
        "audit": ["ticket_received"],
    },
    config,
)

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

Шаг 7: Чтение сохраненного состояния

snapshot = graph.get_state(config)
print(snapshot.values)

Значения содержат:

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

Этот снимок представляет собой только данные рабочего процесса — он не является ни хранилищем профилей, ни корпусом инструкций.

Что может и чего не может делать InMemorySaver

Он сохраняет точки контроля только в течение срока работы процесса Python — это подходит для тестов на единицы и ноутбуков. Однако он не выдержит перезагрузок, не будет работать с репликами сервиса и не удовлетворит требования к хранению, шифрованию или резервному копированию. Текущая документация рекомендует использовать для хранения в памяти агента в производственных условиях и для потоков, способных к возобновлению работы, решения, основанное на базе данных, такой как Postgres.

Структура для производства:

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = build_checkpointed_graph(checkpointer)

Секреты подключения, миграции, управление пулами соединений и очистка остаются задачами самого приложения. Не включайте URI баз данных в финальный исходный код.

Где заканчивается LangChain и начинается LangGraph

Фиксированная версия LangChain достаточна, когда

последовательность действий фиксирована; один запрос может быть завершен без перерыва со стороны человека; возможна перезагрузка всего запроса; состояние на промежуточных этапах не должно быть долговечным; обычный код приложения может хранить необходимую небольшую историю действий.

LangGraph предпочтительнее, когда

Разветвления потока управления или циклы осуществляются в рамках определенного состояния; человек должен дать разрешение во время выполнения процесса; работа продолжается позже на том же потоке; при перезапуске процесса необходимо сохранить текущее состояние; операторам требуются доступные к просмотру снимки состояния; восстановление должно происходить с сохраненной точки, а не с нуля.

Сегодняшний вспомогательный функционал create_agent уже возвращает агента, находящегося в LangGraph. Необходимо предоставить точку контроля, после чего происходит сохранение состояния во время выполнения — это предусмотренная архитектура, а не случайная утечка данных.

Точка контроля — это не журнал аудита

Точки контроля существуют для того, чтобы процесс мог продолжаться. Журналы аудита существуют для того, чтобы специалисты по безопасности и бизнесу могли восстановить ход действий. Иногда они имеют общие поля, но их назначения различаются. В строке журнала аудита должны указываться имя заявителя, используемый инструмент, лицо, давшее разрешение, аргументы, использованные при выполнении, результат и время события. Не следует рассматривать внутренний сериализованный снимок состояния как журнал аудита, соответствующий стандартам соблюдения норм.

Точки контроля и побочные эффекты

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

Проверка разделения функций

Три теста в автономном снимке состояния.

Истории сообщений остаются разделенными

history.add_turn("chat:first", "First note", "First reply")
history.add_turn("chat:first", "Second note", "Second reply")
history.add_turn("chat:second", "Other ticket", "Other reply")

assert history.prior_turn_count("chat:first") == 2
assert history.prior_turn_count("chat:second") == 1

Сохраненные данные не являются сообщениями в чате

facts.put("user-17", "team", "commerce-platform")

assert facts.get("user-17") == {
    "team": "commerce-platform"
}
assert history.read("user-17") == []

Точки контроля остаются разделенными по потокам тикетов

first, first_config = run_ticket(
    graph,
    "INC-2048",
    "checkout-api reports database connection refused",
)
second, second_config = run_ticket(
    graph,
    "INC-2050",
    "identity-api denied an access role request",
)

assert graph.get_state(first_config).values["ticket_id"] == "INC-2048"
assert graph.get_state(second_config).values["ticket_id"] == "INC-2050"

Выполнить:

pytest -q

Ожидаемый результат:

3 passed

Они подтверждают наличие границ пространства имён и изоляцию потоков — они не гарантируют надёжность базы данных при использовании InMemorySaver.

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

Один глобальный список истории

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

Сохранение каждого оператора модели как факта

Модели создаются с уверенностью. Сохраняйте только разрешённые поля через проверенный путь записи.

Хранение секретов в состоянии

Снимки состояния копируются, анализируются и хранятся. Храните секреты в сейфе и передавайте вместо этого ссылки на них.

Использование thread_id в качестве механизма авторизации

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

Называние хранилища векторов «долгосрочной памятью»

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

Ожидание использования контрольной точки для исправления ошибок

Снимки сохраняют всё, что было записано, включая ошибки. Проверка и тестирование остаются обязательными.

Результат части 9

Четыре определенных границы хранения:

session ID -> ordered chat messages
user ID    -> selected saved facts
thread ID  -> current workflow state
checkpoint -> persisted workflow snapshot

Статическая цепочка по-прежнему подходит для задач с одним прохождением. LangGraph представляет собой более подходящую структуру, когда требуется долговременное хранение данных, возможность паузы, продолжения работы или восстановления. Далее следует первое настоящее решение агента: инструменты с только-чтением могут работать автоматически; изменения в тикетах требуют утверждения человека.

Проверка документации: была проведена сравнительная проверка с документацией LangChain по краткосрочной памяти и документацией LangGraph по сохранению данных 26 августа 2026 года. API пакетов меняются.

Для дальнейшего изучения: краткосрочная память LangChain, агенты LangChain, сохранение данных LangGraph.

Системы службы поддержки производства обычно требуют одновременно всех четырех типов хранилищ: буфер чата с ограничением по сессии для текущего инженера, хранилище данных с ограничением по пользователю для сохранения постоянных настроек, хранилище состояния рабочего процесса с ограничением по потоку для отслеживания состояния заявок, а также индекс руководства с возможностью поиска, который никогда не используется одновременно в качестве истории или точки контроля. Определение этих границ в ходе обзоров кода предотвращает распространенную ошибку — помещение всего в один список Redis с названием «память». При подборе нового коллеги попросите его нарисовать четыре квадрата и обозначить ключи; если он не сможет этого сделать, дизайн еще не готов к возобновлению работы после перерыва.

Когда позже будет введена проверка с участием человека (Часть 10), контрольная точка станет местом, где работа потока останавливается в ожидании. История чатов продолжает существовать независимо, что позволяет инженеру задавать уточняющие вопросы без изменения данных, находящихся в ожидании записи. Факты не попадают в путь прерывания, если только специальное правило не копирует соответствующее поле. Именно это разделение предотвращает ситуацию, когда «продолжение после обеда» превращается в «перепросмотр всего разговора в обновлении заявки».

Смешивание политик хранения в разных хранилищах

Транскрипты чатов, постоянные факты, контрольные точки потока работ и встроенные данные руководства по выполнению практически никогда не используют один и тот же таймер хранения. Согласование их «для простоты» обычно нарушает либо запросы на удаление данных в целях конфиденциальности, либо требования к перепросмотру инцидентов. Укажите четыре таймера, четырех ответственных и четыре точки удаления — даже если два из них в настоящее время указывают на одну и ту же инстанцию Redis.

Рассмотрение предупреждений о устаревании как необязательных

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

Забывание о том, что редьюсеры являются частью схемы

Команды часами обсуждают названия полей, а затем бездумно присоединяют редьюсер для дополнения к полю, которое должно заменить его. Ошибка проявляется спустя недели в виде дублирования классификаций или удаления записей аудита. Проверяйте редьюсеры в том же PR, что и TypedDict.