Strona główna / Artykuły / Historia czatu, fakty, stan przepływu pracy oraz punkty kontrolne to cztery różne zbiory danych.

langchainlanggraphmemorycheckpointsagentspersistence

Historia czatu, fakty, stan przepływu pracy oraz punkty kontrolne to cztery różne zbiory danych.

Przestań nazywać wszystko pamięcią. Oddziel transkrypcje sesji, trwałe fakty, stan procesu obsługi zgłoszeń oraz punkty kontrolne LangGraph – z regułami przechowywania i autoryzacji dla każdego z nich.

2467 słów

Część 9 z 14: Oddzielna historia rozmów, zapisane fakty oraz dane do kontynuacji pracy

Dziewiąty odcinek z serii czternastu wpisów dotyczących budowy stanowiska pomocy technicznej, które pokazuje, jak LangChain przechodzi od pierwszego wywołania modelu do standardowych praktyk w produkcji. Późniejsze wpisy przekształcają gotowy system w narzędzia do ćwiczeń interwencyjnych.

W poprzednim odcinku omówiono wyszukiwanie w książce instrukcji: zapytanie do dobrze przygotowanego korpusu, przechowywanie metadanych pochodzenia oraz blokowanie porad opartych na dokumentach, których wyszukiwanie nigdy nie zwróciło.

Ktoś na zmianie pyta, czy system może „pamiętać” o incydencie jutro. To pytanie jest zbyt ogólne. Czy należy przechowywać kolejne rozmowy? Ustawienia zespołu? Etykiety i odzyskane fragmenty tekstu? Zapisy czekające na zatwierdzenie? Każde pośrednie pole z grafu? Ludzie łączą to wszystko pod jednym niejasnym etykiemietem. Każdy z tych elementów wymaga własnego klucza, okresu ważności, uprawnień dostępu oraz polityki radzenia sobie z awariami.

W tej części rozdzielono cztery różne koncepcje:

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

Przekazywanie wcześniejszych wiadomości do statycznego programu wykonywalnego w LangChain nie sprawia, że magicznie pojawi się możliwość przerwania i wznowienia działania. Trwałe wątki oraz zrzuty stanu pochodzą z modelu stanu i checkpointer w LangGraph.

Bieżący problem

Niech przykładami będą znane przypadki incydentów:

Po wersji z godziny 14:05 połączenia typu checkout z regionu UE zawodzą. Logi checkout-api wskazują, że baza danych odrzuciła nowe połączenia.

Należy przypisać każdemu rodzajowi rekordu własną przestrzeń kluczy:

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

Traktowanie identyfikatora incydentu jak identyfikatora osoby powoduje łączenie niespowiązanych przestrzeni nazw. Podobnie pojedyncza lista transkrypcji łączy odrębne przypadki.

Najpierw przestańmy mówić „pamięć”

Nazwij konkretny rekord, o którym mówimy.

Historia rozmowy

Listę uporządkowaną w następujący sposób:

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.

Sekwencja ta ma znaczenie, gdy później łączysz prompty z tych rund.

Zapisane fakty

Wybrane pola, takie jak:

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

Fakty mogą przetrwać poza jedną rozmową. Zachowaj je tylko za pomocą wyraźnych zasad aplikacji – a nie poprzez przeszukiwanie każdego twierdzenia wymyślonego przez model.

Stan procesu

Bieżące dane dla jednego procesu ticketów:

{
  "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"
  ]
}

Stan zmienia się w miarę wykonywania kroków.

Punkt kontrolny

Punkt kontrolny to jakby zamarznięty obraz procesu wraz z informacjami potrzebnymi do kontynuacji działania. Ładujesz najnowszy obraz dla danej ścieżki, wznowiasz pracę po przerwie, sprawdzasz, co widział dany krok, oraz przywracasz stan po awarii. Narzędzia do zapisu lokalne znikają po zamknięciu; przywracanie w środowisku produkcyjnym wymaga narzędzia zapisującego opartego na bazie danych.

Pobieranie informacji to nic z tego

Indeks uporządkowanej książki instrukcji to korpus do wyszukiwania. Otwarcie jej podczas incydentu nie przekształca wyników wyszukiwania w historię rozmowy. Fragmenty nie powinny automatycznie stawać się trwałymi danymi profilu. Indeksy embeddingów nie są bazami danych checkpointów. Izolujcie magazyny, nawet jeśli pojedynczy żądanie HTTP dotyka kilku z nich.

Co domyślnie pamięta stała łańcuchowa struktura

Pomiędzy niezależnymi wywołaniami łańcuch nic nie pamięta, chyba że wasza aplikacja wstrzykuje lub przechowuje kontekst.

To wywołanie:

result = chain.invoke(current_input)

nie przekazuje automatycznie wcześniejszych danych wejściowych ani wyjściowych. Możecie sami dodać wcześniejsze etapy rozmowy lub użyć starszych narzędzi do przechowywania historii. W wersji LangChain omówionej tutaj RunnableWithMessageHistory ostrzega i skierowuje nowe operacje w stronę przechowywania danych w LangGraph.

W przypadku statycznej łańcuchowej struktury zarządzanie historią w kodzie aplikacji zazwyczaj jest łatwiejsze do zrozumienia:

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

To właśnie robi kod towarzyszący.

Struktura projektu

Zrzut stanu z Części 9 zawiera:

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

Zainstaluj pakiety:

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

Przykład nie wykonywa żadnych wywołań do dostawcy.

Krok 1: Przechowywanie wiadomości czatowych według sesji

Stwórz plik 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

Klucze histories łączą jedną sesję z jedną uporządkowaną listą. Funkcja read tworzy kopie, dzięki czemu użytkownicy nie mogą dodawać elementów do pamięci poprzez efekt uboczny. Funkcja add_turn rejestruje parę człowiek/asystent. Funkcja prior_turn_count zmniejsza długość listy o połowę, ponieważ przykład przechowuje tylko kompletne pary. Transkrypcje na żywo zawierają również wiadomości narzędzi, nieukończone wypowiedzi oraz błędy — nie zakładaj idealnego dopasowywania w środowisku produkcyjnym.

Historia wymaga zasady przechowywania

Zachowywanie każdej rundy na zawsze nie jest funkcją produktu. Polityka musi określić, co może być przechowywane, jak długo, kto może to przeczytać, które pola są maskowane, w jaki sposób odbywa się usuwanie danych oraz ile rund trafia do następnego wywołania modelu. Zbyt duże historie marnują tokeny i pieniądze. Streszczenia mogą pomóc, ale mogą również powodować błędy — traktuj je jako artefakty pochodne z jasnymi regułami pochodzenia.

Krok 2: Przechowywanie wybranych faktów osobno

Stwórz plik 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, {}))

Utwórz indeks faktów według user_id, nigdy według identyfikatora sesji lub incydentu. Przyjmuj tylko nazwane pola; nigdy nie przechowuj całego transkryptu pod jednym kluczem. Prawdziwe ścieżki put wymagają list dostępu, walidacji, autoryzacji oraz zdarzeń audytowych. Wskazówka modelu nie stanowi upoważnienia do trwałego przechowywania danych.

Krok 3: Definicja stanu procesu

Pерейдź do procesu z możliwością przechowywania stanów. TypizedDict w pliku 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 pozwala polom pozostać pustymi, dopóki węzeł je nie zapisze. Zdarzenia audytowe wykorzystują reduktora:

Annotated[list[str], add]

Gdy węzeł zwraca więcej wierszy audytowych, reduktor łączy je ze sobą zamiast je nadpisywać. Wybieraj reduktory celowo — użyj append dla strumieni zdarzeń, a replace dla pól skalarowych.

Krok 4: Tworzenie małych, deterministycznych węzłów

Teaching graph wykorzystuje zwykły Python, dzięki czemu zachowanie punktów kontrolnych jest widoczne:

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}"],
    }

Każdy węzeł odczytuje stan i zwraca patch; nigdy nie modyfikuje przychodzącego słownika. Węzeł rekomendacji wykorzystuje wyniki klasyfikacji:

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"],
    }

Są to zwykłe funkcje w Pythonie — ta część koncentruje się na stanie i trwałości danych, a nie na dokładności klasyfikatora.

Krok 5: Budowanie grafu

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) łączy wspólny stan z słownikiem typowanym. Węzły i krawędzie określają kolejność; funkcja compile weryfikuje strukturę i łączy punkty kontrolne. Ścieżka pozostaje liniowa – graf funkcjonuje dzięki temu, że stan i punkty kontrolne mają pierwszorzędne znaczenie, a nie dlatego, że diagram jest wyrafinowany.

Krok 6: Przypisz każdemu procesowi robociemu identyfikator wątku

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

Zrób test:

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 dzieli historię punktów kontrolnych. Ponowne użycie tego samego wątku w niepowiązanych incydentach powoduje wyciek stanu pomiędzy nimi.

Krok 7: Odczytaj zapisany stan

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

Wartości zawierają:

{
  "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"
  ]
}

Ten zrzut ekranu to wyłącznie dane procesu roboczego – nie jest ani magazynem profili, ani zbiorem instrukcji.

Czego InMemorySaver może, a czego nie może zrobić

Zachowuje punkty kontrolne tylko przez cały czas trwania procesu Pythona – co jest wystarczające do testów jednostkowych i notatek. Nie przetrwa restartów, nie obejmie replik usług ani nie spełni wymagań dotyczących przechowywania, szyfrowania czy tworzenia kopii zapasowych. Obecne dokumentacje skłaniają do wykorzystania pamięci agenta produkcyjnego oraz wątków możliwych do wznowienia z zapisem w bazie danych, takiej jak Postgres.

Kształt w produkcji:

from langgraph.checkpoint.postgres import PostgresSaver

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

Tajemnice połączeń, migracje, zarządzanie zasobami i czyszczenie pozostają sprawami aplikacji. Unikaj umieszczania URI bazy danych w komitowanych plikach źródłowych.

Gdzie kończy się LangChain, a zaczyna LangGraph

Stabilna wersja LangChain wystarcza, gdy

sekwencja jest ustalona; pojedyncza prośba może zostać zakończona bez przerwy ze strony człowieka; restart całej prośby nie stanowi problemu; stan w trakcie realizacji nie musi być trwały; zwykły kod aplikacji może przechowywać potrzebną małą historię.

LangGraph jest lepszy, gdy

Łącza przepływu sterowania lub pętle nad nazwanym stanem; osoba musi zatwierdzić działanie w trakcie jego wykonywania; praca jest kontynuowana później w tym samym wątku; ponowne uruchomienie procesu musi zachować stan w oczekiwaniu; operatorzy potrzebują możliwości inspekcji zrzutów stanu; przywracanie powinno odbywać się z zapisanego punktu, a nie od początku.

Dzisiejsza funkcja pomocnicza create_agent już zwraca agenta umieszczonego w LangGraph. Należy dostarczyć punkt kontrolny, a dalsza praca odbywa się w oparciu o ten czas wykonywania — to jest oczekiwana architektura, a nie przypadkowy wyciek danych.

Punkt kontrolny to nie dziennik audytu

Punkty kontrolne istnieją po to, aby czas wykonywania mógł być kontynuowany. Dzienniki audytu istnieją po to, aby specjaliści ds. bezpieczeństwa i biznesu mogli odtworzyć dokonane działania. Czasami mają one wspólne pola, ale ich zadania się różnią. Wiersz w dzienniku audytu powinien zawierać nazwę osoby proszącej o działanie, proponowane narzędzie, osobę dokonującą zatwierdzenia, użyte argumenty, wynik oraz datę i godzinę. Nie należy traktować wewnętrznego zserializowanego zrzutu stanu jako dziennika audytu o standardzie zgodności.

Punkty kontrolne i skutki uboczne

Ciągłe przechowywanie stanu nie czyni zewnętrznego zapisu idempotentnym. Jeśli proces aktualizuje bilet, a następnie ulega zakończeniu przed kolejnym punktem kontrolnym, przy wznowieniu działania zapis może zostać powtórzony. Narzędzia wymagają kluczy idempotencji lub sprawdzeń typu „już zastosowane”. Skutki uboczne należy umieszczać po zatwierdzeniu; oznaczać identyfikatory stabilnej pracy; dokumentować semantykę ponawiania prób. Kolejny etap polega na zatrzymaniu działania przed zapisem do biletu oraz rozstrzygnięciu, czy akcja ma być zatwierdzona, czy odrzucona.

Testowanie separacji

Trzy testy offline w towarzyszącym snapshotie.

Historie wiadomości pozostają oddzielone

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

Zapisane fakty to nie wiadomości czatowe

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

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

Punkty kontrolne pozostają oddzielone według wątku biletu

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"

Rozpocznij:

pytest -q

Oczekiwane wyniki:

3 passed

Zapewniają one granice przestrzeni nazw oraz izolację wątków — nie gwarantują natomiast trwałości bazy danych przy użyciu InMemorySaver.

Częste błędy

Jedna globalna lista historii

Powoduje to kolizje między niepowiązanymi użytkownikami lub incydentami. Zawsze identyfikuj historię za pomocą autoryzowanego, ściśle określonego identyfikatora.

Zapisywanie każdej instrukcji modelu jako faktu

Modele tworzą dane bez ostrożności. Zachowuj tylko pola z listy dozwolonych poprzez zweryfikowaną ścieżkę zapisu.

Zachowywanie tajemnic w stanie aplikacji

Snapy są kopiowane, sprawdzane i przechowywane. Trzymaj tajemnice w sejfie i przekazuj zamiast tego odniesienia do nich.

Używanie thread_id jako mechanizmu autoryzacji

ID wątku służy do znajdowania stanu; nigdy nie potwierdza, że osoba wywołująca może go odczytać. Autoryzuj oddzielnie.

Nazywanie magazynu wektorów „długoterminową pamięcią”

Slogan ukrywa kwestię własności i usuwania danych. Należy podać nazwy zapisów, autorów, ścieżkę zapytania oraz politykę usuwania.

Oczekiwanie na punkt kontrolny do naprawy błędnego kroku

Zdjęcia stanu zachowują wszystko, co zostało napisane – włącznie z błędami. Walidacja i testy pozostają obowiązkowe.

Wynik części 9

Cztery określone granice przechowywania:

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

Statyczna łańcuchowa struktura nadal pasuje do zadań wykonywanych w jednej rundzie. LangGraph jest bardziej przejrzystym rozwiązaniem, gdy potrzebny jest trwały stan, możliwość pauzowania, kontynuacji lub odzyskiwania danych. Następnie pojawia się pierwsza prawdziwa decyzja agenta – narzędzia tylko do odczytu mogą działać automatycznie; modyfikacje biletów wymagają interwencji człowieka.

Kontrola dokumentacji: sprawdzona w porównaniu z dokumentacją dotyczącą krótkoterminowej pamięci w LangChain oraz trwałości danych w LangGraph z dnia 26 sierpnia 2026 roku. API pakietów uległy zmianie.

Dalsza lektura: krótka pamięć w LangChain, agenci w LangChain, trwałość danych w LangGraph.

Systemy helpdesku produkcyjnego zazwyczaj wymagają jednocześnie wszystkich czterech magazynów: bufora do rozmów skonfigurowanego na poziomie sesji dla aktualnego inżyniera, magazynu faktów skonfigurowanego na poziomie użytkownika do przechowywania trwałych ustawień, stanu przepływu pracy skonfigurowanego na poziomie wątku dla grafu zgłoszeń oraz indeksu podręcznika operacyjnego umożliwiającego wyszukiwanie, który nigdy nie pełni funkcji historii ani punktu kontrolnego. Określenie tych granic podczas przeglądów kodu zapobiega klasycznemu sposobowi obejścia, polegającemu na umieszczaniu wszystkiego w jednej liście Redis o nazwie „memory”. Podczas wprowadzania nowego kolegi do zespołu poproś go o narysowanie tych czterech pól i oznaczenie kluczy – jeśli nie będzie w stanie tego zrobić, projekt nie jest gotowy do przerwania i kontynuowania pracy.

Gdy później wprowadzisz ludzką aprobatę (Część 10), punkt kontrolny staje się miejscem, w którym przepływ pracy jest zatrzymywany w oczekiwaniu. Historia rozmów kontynuuje się niezależnie, dzięki czemu inżynier może zadawać pytania wyjaśniające bez modyfikowania bieżącej operacji zapisu. Fakty nie trafiają na ścieżkę przerwania, chyba że jakaś wyraźna zasada kopiuje dane z danego pola. To rozdzielenie zapobiega temu, by funkcja „przywrócenie po przerwie na lunch” zamieniła się w „ponowne odtworzenie całej rozmowy w aktualizacji ticketu”.

Mieszanie polityk przechowywania w różnych systemach

Transkrypcje rozmów, trwałe dane, punkty kontrolne przepływu pracy oraz elementy z runbooków prawie nigdy nie korzystają z tego samego mechanizmu określania czasu przechowywania. Ich dostosowanie „dla uproszczenia” zazwyczaj narusza albo prośby o usunięcie danych ze względów prywatności, albo potrzeby ponownego odtworzenia incydentu. Zapisz cztery mechanizmy określania czasu, czterech odpowiedzialnych oraz cztery punkty usunięcia – nawet jeśli dwa z nich obecnie wskazują na tę samą instancję Redis.

Traktowanie ostrzeżeń o deprecjacji jako opcjonalnych

Gdy biblioteka ostrzega, że wrapperzy historii przechodzą na mechanizm persistencji LangGraph, należy to traktować jako sygnał projektowy. Wprowadzanie nowej funkcji w dziale pomocy na zastarzałej ścieżce oznacza konieczność przepisania jej później w terminie. Dla wszelkich procesów, które mogą zostać wstrzymane, lepiej zastosować model checkpointer.

Zapominanie o tym, że reducerzy są częścią schematu

Zespoły godzinami dyskutują nad nazwami pól, a potem bez zastanowienia dodają reducer do pola, które powinno je zastąpić. Błąd pojawia się kilka tygodni później w postaci powtórzonych kategorii lub usuniętych rekordów audytu. Należy przejrzeć reducerzy w tym samym PR co TypedDict.

Literatura pokrewna

Wszystkie artykułyWyślij brief