Checkpointy w LangGraph: zapis stanu w SQLite i wznawianie grafu

Dotychczas nasz graf działał od początku do końca w jednym wywołaniu. W prawdziwym programie coś w końcu się wysypie: zewnętrzna usługa nie odpowie, proces zostanie zabity albo operator zamknie okno przed zatwierdzeniem. Bez zapisu stanu zaczynasz od zera i powtarzasz wszystko, w tym kroki, które już się udały. Checkpointy rozwiązują ten problem, a w następnej części cyklu będą nam potrzebne do zatrzymania grafu na akceptację człowieka.

Zakładam, że znasz stan, węzły i krawędzie w LangGraph oraz pseudonimizację danych w Pythonie, bo zapiszemy stan właśnie takiego grafu. Kod uruchomiłem na Pythonie 3.13, LangGraph 1.2.18 i langgraph-checkpoint-sqlite 3.1.2. Instalacja:

pip install langgraph langgraph-checkpoint-sqlite

Czym jest checkpoint i po co jest thread_id?

LangGraph wykonuje graf krokami. Po każdym kroku checkpointer zapisuje migawkę stanu (checkpoint) razem z informacją, który węzeł ma wykonać się następny. Migawki należą do wątku (thread): to identyfikator jednego przebiegu, który podajesz przy każdym wywołaniu. Dwa wątki nie widzą swoich danych.

Gdy graf skończy, next jest puste, a zapisany stan zostaje dostępny pod tym samym thread_id.

Jak przygotować stan, żeby dał się zapisać?

Stan trafia do bazy w serializowanej postaci, więc najbezpieczniej trzymać w nim zwykłe dane: napisy, liczby, listy i słowniki. Gdy w poprzedniej części wynikiem detektora była lista obiektów Finding (dataclass), przy zapisie LangGraph 1.2.18 wypisał takie ostrzeżenie:

Deserializing unregistered type detectors.Finding from checkpoint. This will be blocked in a future version. Set LANGGRAPH_STRICT_MSGPACK=true to block now, or add to allowed_msgpack_modules to allow explicitly: [('detectors', 'Finding')]

Czyli własne klasy w stanie działają dziś, ale w przyszłej wersji mają być blokowane. Najprościej uniknąć problemu: w stanie przechowujemy findings jako listę słowników (asdict), a obiekt Finding odtwarzamy dopiero wewnątrz węzła. Po tej zmianie ostrzeżenie zniknęło.

Jak zbudować graf z checkpointerem?

Graf ma trzy węzły: wykrywanie, podmiana i zapis wyniku. Węzeł save udaje zapis do zewnętrznego magazynu, który może się nie powieść. Przełącznik FAIL pozwoli nam wywołać awarię na żądanie:

from dataclasses import asdict
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

from detectors import Finding, find_sensitive
from pseudonymize import pseudonymize


class State(TypedDict):
    text: str
    findings: list[dict]
    pseudonymized: str
    mapping: dict[str, str]


FAIL = {"on": False}


def detect(state: State) -> dict:
    return {"findings": [asdict(f) for f in find_sensitive(state["text"])]}


def replace(state: State) -> dict:
    findings = [Finding(**f) for f in state["findings"]]
    result = pseudonymize(state["text"], findings)
    return {"pseudonymized": result.text, "mapping": result.mapping}


def save(state: State) -> dict:
    if FAIL["on"]:
        raise ConnectionError("Brak połączenia z magazynem")
    return {}


def build(checkpointer):
    builder = StateGraph(State)
    builder.add_node("detect", detect)
    builder.add_node("replace", replace)
    builder.add_node("save", save)
    builder.add_edge(START, "detect")
    builder.add_edge("detect", "replace")
    builder.add_edge("replace", "save")
    builder.add_edge("save", END)
    return builder.compile(checkpointer=checkpointer)

Jedyna różnica względem grafu bez zapisu to argument checkpointer w compile(). Węzły detect i replace korzystają z kodu z poprzednich części (wykrywanie danych wrażliwych i pseudonimizacja).

Jak uruchomić graf z zapisem do SQLite i wywołać awarię?

SqliteSaver.from_conn_string("checkpoints.db") otwiera plik SQLite jako menedżer kontekstu. Wszystko, co dzieje się w bloku with, zapisuje się do pliku. Konfiguracja z thread_id jest drugim argumentem wywołania:

from langgraph.checkpoint.sqlite import SqliteSaver

import flow

with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    graph = flow.build(saver)
    config = {"configurable": {"thread_id": "zgloszenie-1"}}
    flow.FAIL["on"] = True
    state = {"text": "Klient 44051401359 pisze z a@b.pl", "findings": [], "pseudonymized": "", "mapping": {}}
    try:
        graph.invoke(state, config)
    except ConnectionError as e:
        print("Błąd:", e)
    print("next:", graph.get_state(config).next)

Wyjście:

Błąd: Brak połączenia z magazynem
next: ('save',)

Graf przerwał się w węźle save, a get_state(config).next pokazuje, że czeka właśnie na niego. Dwa wcześniejsze węzły już wykonały się i ich wynik jest w pliku. Możesz go odczytać: get_state(config).values["pseudonymized"] zwraca Klient [PESEL_1] pisze z [EMAIL_1].

Jak wznowić graf po awarii, także po restarcie programu?

Wznawianie to wywołanie invoke z None zamiast stanu wejściowego i z tym samym thread_id. None oznacza: „nie zaczynaj od nowa, kontynuuj z zapisanego miejsca”. Sprawdziłem to w osobnym procesie Pythona, żeby mieć pewność, że stan jest w pliku, a nie w pamięci:

with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    graph = flow.build(saver)
    config = {"configurable": {"thread_id": "zgloszenie-1"}}
    print("przed:", graph.get_state(config).next)
    result = graph.invoke(None, config)
    print("wynik:", result["pseudonymized"], result["mapping"])
    print("po:", graph.get_state(config).next)

Wyjście z drugiego uruchomienia (przełącznik FAIL jest tam domyślnie wyłączony):

przed: ('save',)
wynik: Klient [PESEL_1] pisze z [EMAIL_1] {'[PESEL_1]': '44051401359', '[EMAIL_1]': 'a@b.pl'}
po: ()

Nowy proces odczytał stan z pliku, wykonał tylko save i zakończył graf (next jest puste). Węzły detect i replace się nie powtórzyły. Mówi o tym też test w pytest poniżej.

Co dokładnie jest zapisane i jak obejrzeć historię?

Po każdym kroku powstaje osobny checkpoint. get_state_history(config) zwraca je od najnowszego:

for snapshot in graph.get_state_history(config):
    print(snapshot.metadata["step"], snapshot.next)

3 ()
2 ('save',)
1 ('replace',)
0 ('detect',)
-1 ('__start__',)

Każda linijka to jeden checkpoint: numer kroku i węzły, które miały się wykonać po nim. Krok -1 to stan wejściowy, zapisany, zanim cokolwiek się wykonało. Dzięki historii wiesz, co dokładnie widział każdy węzel, co ułatwia szukanie błędów.

Wątki są niezależne: get_state dla innego thread_id zwróciło {}, czyli pusty stan, mimo że plik ten sam.

Jak nazwać thread_id?

Użyj identyfikatora zdarzenia, którego dotyczy przebieg: numeru zgłoszenia albo dokumentu, a nie losowego napisu czy znacznika czasu. Wtedy po awarii od razu wiesz, który wątek wznowić, i możesz bezpiecznie powtórzyć wywołanie dla tego samego dokumentu, zamiast tworzyć drugi wątek z duplikatem.

Czego checkpoint nie robi za Ciebie?

Dwie rzeczy warto wiedzieć, zanim oprzesz na tym prawdziwy system:

Pierwszy punkt jest istotny dla węzła save: w naszej demonstracji wykonał się dwa razy (raz nieudanie, raz po wznowieniu). Gdyby zapisywał rekord do zewnętrznej bazy, powtórzenie nie może stworzyć duplikatu, więc taki węzeł powinien być idempotentny (kolejne wykonanie daje ten sam skutek, np. zapis po kluczu zamiast dopisania na końcu).

Drugi punkt jest ważniejszy w tym cyklu. Sprawdziłem plik bazy po uruchomieniu grafu:

from pathlib import Path

print(b"44051401359" in Path("checkpoints.db").read_bytes())

True

PESEL z naszego przykładu leży w pliku w otwartym tekście, tak samo jak e-mail i cała tabela mapowania, bo to wszystko jest częścią stanu. Pseudonimizacja chroni tekst przekazywany dalej, ale checkpointer zapisuje także oryginały. Plik checkpoints.db jest więc tak samo wrażliwy jak same dane. Dlatego jedna z kolejnych części cyklu dotyczy szyfrowania mapowania.

Jak zabezpieczyć plik z checkpointami?

Skoro plik checkpoints.db zawiera te same dane, które pseudonimizujesz, traktuj go jak same dane:

• nie dodawaj go do repozytorium: wpisz *.db do .gitignore,
• nie kopiuj go do chmury ani na zewnętrzne dyski bez szyfrowania,
• ustal okres przechowywania i usuwaj wątki po jego upływie, bo po zakończeniu przebiegu stary stan nie jest już potrzebny.

Jak przetestować wznawianie w pytest?

Testy używają tymczasowego pliku bazy (tmp_path), żeby nie zostawiać śladów. Ostatni test sprawdza, że wątki są niezależne:

import pytest
from langgraph.checkpoint.sqlite import SqliteSaver

import flow

TEXT = "Klient 44051401359 pisze z a@b.pl"


def start_state():
    return {"text": TEXT, "findings": [], "pseudonymized": "", "mapping": {}}


@pytest.fixture
def db(tmp_path):
    flow.FAIL["on"] = False
    return str(tmp_path / "checkpoints.db")


def config(thread="t1"):
    return {"configurable": {"thread_id": thread}}


def test_failure_leaves_graph_waiting_at_failed_node(db):
    flow.FAIL["on"] = True
    with SqliteSaver.from_conn_string(db) as saver:
        graph = flow.build(saver)
        with pytest.raises(ConnectionError):
            graph.invoke(start_state(), config())
        snapshot = graph.get_state(config())
    assert snapshot.next == ("save",)
    assert snapshot.values["pseudonymized"] == "Klient [PESEL_1] pisze z [EMAIL_1]"


def test_resume_does_not_rerun_finished_nodes(db, monkeypatch):
    runs = []
    original = flow.detect
    monkeypatch.setattr(flow, "detect", lambda state: runs.append(1) or original(state))
    flow.FAIL["on"] = True
    with SqliteSaver.from_conn_string(db) as saver:
        graph = flow.build(saver)
        with pytest.raises(ConnectionError):
            graph.invoke(start_state(), config())
        flow.FAIL["on"] = False
        result = graph.invoke(None, config())
    assert runs == [1]
    assert result["mapping"] == {"[PESEL_1]": "44051401359", "[EMAIL_1]": "a@b.pl"}


def test_new_connection_resumes_the_same_thread(db):
    flow.FAIL["on"] = True
    with SqliteSaver.from_conn_string(db) as saver:
        with pytest.raises(ConnectionError):
            flow.build(saver).invoke(start_state(), config())
    flow.FAIL["on"] = False
    with SqliteSaver.from_conn_string(db) as saver:
        graph = flow.build(saver)
        assert graph.get_state(config()).next == ("save",)
        graph.invoke(None, config())
        assert graph.get_state(config()).next == ()


def test_threads_are_independent(db):
    with SqliteSaver.from_conn_string(db) as saver:
        graph = flow.build(saver)
        graph.invoke(start_state(), config("a"))
        assert graph.get_state(config("b")).values == {}

Test test_resume_does_not_rerun_finished_nodes zlicza wywołania detect: po awarii i wznowieniu węzeł wykonał się raz, nie dwa razy. Test z nowym połączeniem (test_new_connection_resumes_the_same_thread) odtwarza sytuację restartu programu.

Co dalej?

Mamy graf, który pamięta swój stan i potrafi ruszyć od miejsca przerwania. W kolejnej części wykorzystamy to do zatrzymania grafu w środku i oddania decyzji człowiekowi: operator zobaczy proponowane zastąpienia, zatwierdzi je albo poprawi, a graf ruszy dalej z zapisanego miejsca. Całą ścieżkę Pythona znajdziesz na stronie ścieżki rozwoju w Pythonie.