Історыя чату, факты, стан рабочага процесу і пункты контролю — гэта чатыры разныя базы дадзейнаў.

Частка 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 з регіону ЕС не выкананы. Журналы 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: Даць кожнаму рабочаму процесу ID адгалужэння

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. Неабяжна падаць чекпойнтар, і тады працэс будзе продовжвацца — гэта ўскладненая архітектура, а не випадковы прасач.

Чекпойнт — гэта не журнал аудыту

Чекпойнты існуюць каб працэс могаў продовжыцца. Журналы аудыту існуюць каб спецялісты з безпекі та бізнесу моглі восстанавіць выконаныя дзеяння. Інодзе ў іх існуюць спольныя поля; але ў яных разнае праблематыка. У рэкордзе аудыту павінны быць указаны запрошчальнік, выбраны інструмент, той, хто затвердзіў, параметры, якія былі выкараны, рэзультат і час запісу. Не трэба спрыяжваць внутршній серыялізаваны кадр стану з журналам аудыту высокага рэгламенту.

Тачкі пазначэння і побачныя эфекты

Зберагаецца стан не робіць зовнішню запісь ідэмпотентной. Якщо процес апдейтуе квітанцыю, а потым зупініцца прычынай наступнай тачкі пазначэння, пры вярненні роботы запіс можа буць павтараная. Інструменты патрабуюць ключоў ідэмпотентнасі або пераканання “вже застосавана”. Побачныя эфекты трэба размістіць пасля затверджэння; ставіць пазначкі стабільных ID роботы; задокументаваць семантіку павторных спроб. Наступны етап запускаецца прычынай паўтарной запісі квітанцыі і пераканваецца, чы характерыстыкі “затвердзіць” чы “адмовіць”.

Тэставацыя раздзелення

Тры тэсты без падключэння да сеті ў супутнім скрапшоте.

Історыі паведамленняя застаюцца окремымі

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 пад назвай “memory”. Калі прыймаеце новага колегу, папросіце яго накрасіць чатыры квадраты і пазначыць ключы; якщо ён не можа гэтага зробіць, дизайн не гатовы для перерыву/праўернення.

Калі пазнейшае вы введзеце механізм людскай апраўкі (Частка 10), пункт пераконтролу стане месцам, дзе працэс выконання завяршваецца у чаканне. Історыя чату продовжваецца самастоятельна, таму інжынер можа задаваць уточняючыя запитанні без змены ўсё яшчэ не апраўленых дадзенняў. Факты не прабіваюць у шляху перерыву, як толькі не будзе спецыяльнага правіла, яке копіюе певны поле. Самэ гэта раздзеленне не дазволяе сцэнарію “працаваць пасля обеду” ператворыцца на “переглед усіх розмов у апдэйте тыкета”.

Сумешанне правіл зберагчэння дадзеных у разных сховішчах

Транскрыпцыі чату, стойкія факты, пункты пераконтролу працэса выконання і вбудованыя элементы інструкцыйяў практычна ніколи не выкарыстоўваюць аднойчы той самы час зберагчэння. Їх супаралелізацыя “для простасці” зазвычай наражае на рызык або запиты на выдаленне дадзеных з метай прыватнасці, або патрэбы ў переглядзе інцидэтаў. Запісуйце чатыры часавыя меткі, чатырох адпаведальных і чатыро пункты выдалення — нават якщо два з іх наразе вядуць да той самай інстанцыі Redis.

Адміністрування паведамленняў пра занепад як неабяжных

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

Забыванне пра тое, што редукеры ўскладнююць схему

Команды гадзіны спарчваюцься над назвамі полей, а потым проста дадаюць редукер для дапамогі да поля, якое павінна заменіць іх. Багі з’яўляюцца за некалькі тыдзён у вигледе дуплікацый класыфікацый чы выдаленых рэядкав аудыту. Пераглядайце редукеры ў той жа PR, што і TypedDict.