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

Частка 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.