Історія чату, факти, стан робочого процесу та контрольні точки — це чотири різні сховища даних.

Частина 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 не дозволяє магічним чином відновити роботу після перерви. Механізми тривалих потоків та знімків стану походять від моделей стану та checkpointer у LangGraph.

Актуальний випадок

Для прикладів можна скористатися вже відомим інцидентом:

Після випуску о 14:05 запити checkout з регіону ЄС припиняють працювати. Журнали 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.