Human-in-the-loop w LangGraph: interrupt i wznowienie grafu

Detektor z poprzednich części nie jest nieomylny: wyrażenia regularne znajdą PESEL i e-mail, ale potraktują też numer faktury jako telefon. Gdy pomyłka kosztuje (np. tekst idzie do zewnętrznego narzędzia), ostatnie słowo powinien mieć człowiek. Ten wzorzec nazywa się human-in-the-loop (człowiek w pętli), a LangGraph ma do niego gotowy mechanizm: graf zatrzymuje się w środku, czeka na decyzję i rusza dalej od tego samego miejsca. Zbudujemy graf, w którym operator zatwierdza, które z wykrytych danych mają zostać zastąpione tokenami.

Opieramy się na checkpointach w LangGraph i SQLite, bo zatrzymanie grafu działa tylko wtedy, gdy stan da się zapisać. Z poprzednich części użyjemy też detektora danych wrażliwych i pseudonimizacji. Kod uruchomiłem na Pythonie 3.13, LangGraph 1.2.18 i langgraph-checkpoint-sqlite 3.1.2. Jeśli interesuje Cię zatwierdzanie przez człowieka w zwykłym agencie, bez grafu, zobacz uprawnienia i zatwierdzanie akcji agenta AI w Pythonie.

Jak działa interrupt w LangGraph?

interrupt() to funkcja, którą wywołujesz wewnątrz węzła. Robi trzy rzeczy:

1. Zapisuje stan grafu w checkpoincie.
2. Zwraca do wywołującego wartość, którą jej przekazałeś (propozycje dla człowieka).
3. Gdy graf zostanie wznowiony, zwraca w tym samym miejscu decyzję człowieka.

Kluczowe jest to, że między krokiem 1 a 4 program może się zakończyć. Wznowienie odczyta stan z pliku, tak jak w artykule o checkpointach.

Jak zbudować graf z human-in-the-loop?

Graf ma trzy węzły: wykrywanie, przegląd przez człowieka i podmiana tylko zatwierdzonych trafień. Zatwierdzenie to lista numerów trafień w stanie:

from dataclasses import asdict
from typing import TypedDict

from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

from detectors import Finding, find_sensitive
from pseudonymize import pseudonymize


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


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


def review(state: State) -> dict:
    decision = interrupt({"text": state["text"], "findings": state["findings"]})
    return {"approved": decision["approved"]}


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


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

Wszystko, co robi review, to jedno wywołanie interrupt() z danymi dla człowieka i zwrócenie jego odpowiedzi do stanu. Węzeł replace korzysta z funkcji pseudonymize z poprzedniej części, ale przekazuje jej tylko te trafienia, których numery znalazły się w approved. Stan trzymamy w zwykłych słownikach, z powodów opisanych w artykule o checkpointach.

Jak zatrzymać graf i odczytać propozycje?

Pierwsze wywołanie dochodzi do interrupt() i kończy się, a stan czeka w checkpoincie. Aby sprawdzić, co dokładnie czeka na człowieka, odczytujemy propozycje ze stanu:

from langgraph.checkpoint.sqlite import SqliteSaver

import review

TEXT = "Klient 44051401359 pisze z a@b.pl. Faktura 600123456."
start = {"text": TEXT, "findings": [], "approved": [], "pseudonymized": "", "mapping": {}}

with SqliteSaver.from_conn_string("review.db") as saver:
    graph = review.build(saver)
    config = {"configurable": {"thread_id": "zgloszenie-1"}}
    graph.invoke(start, config)
    snapshot = graph.get_state(config)
    print(snapshot.next)
    print(snapshot.tasks[0].interrupts[0].value["findings"])

Wyjście:

('review',)
[{'kind': 'pesel', 'value': '44051401359', 'start': 7, 'end': 18}, {'kind': 'email', 'value': 'a@b.pl', 'start': 27, 'end': 33}, {'kind': 'phone', 'value': '600123456', 'start': 43, 'end': 52}]

next wskazuje, że graf czeka w węźle review. Propozycje to dokładnie to, co przekazaliśmy do interrupt(). Zwróć uwagę na trzecią pozycję: „telefon” 600123456 to w naszym tekście numer faktury. Detektor się pomylił, i właśnie po to jest operator.

Jak pokazać propozycje operatorowi i odczytać jego odpowiedź?

Tekstowy interfejs wystarczy do przykładu: numerujemy trafienia od jedynki i przyjmujemy listę numerów, a także słowa „wszystko” i „nic”. Obie funkcje są zwykłymi funkcjami, więc łatwo je przetestować:

def format_proposals(payload: dict) -> str:
    lines = [f"{i}. {f['kind']}: {f['value']}" for i, f in enumerate(payload["findings"], start=1)]
    return "\n".join(lines)


def parse_choice(raw: str, count: int) -> list[int]:
    raw = raw.strip().lower()
    if raw == "nic":
        return []
    if raw == "wszystko":
        return list(range(count))
    numbers = [int(part) - 1 for part in raw.split(",")]
    if any(n < 0 or n >= count for n in numbers):
        raise ValueError(f"Podaj numery od 1 do {count}")
    return sorted(set(numbers))

Dla naszych propozycji format_proposals wypisze:

1. pesel: 44051401359
2. email: a@b.pl
3. phone: 600123456

Operator wpisuje 1,2, czyli zatwierdza PESEL i e-mail, a odrzuca „telefon”. parse_choice zamienia to na numerowanie od zera: [0, 1]. Błędne dane (zero, numer spoza zakresu, litera, puste miejsce między przecinkami) kończą się ValueError jeszcze przed dotknięciem grafu. To celowe, bo walidację trzeba zrobić zanim wznowisz graf, o czym niżej.

Jak wznowić graf decyzją operatora?

Wznawiasz wywołaniem invoke z obiektem Command(resume=...) i tym samym thread_id. To, co podasz w resume, interrupt() zwróci w węźle review:

from langgraph.types import Command

with SqliteSaver.from_conn_string("review.db") as saver:
    graph = review.build(saver)
    config = {"configurable": {"thread_id": "zgloszenie-1"}}
    result = graph.invoke(Command(resume={"approved": [0, 1]}), config)
    print(result["pseudonymized"])
    print(result["mapping"])
    print(graph.get_state(config).next)

Klient [PESEL_1] pisze z [EMAIL_1]. Faktura 600123456.
{'[PESEL_1]': '44051401359', '[EMAIL_1]': 'a@b.pl'}
()

Numer faktury został w tekście, bo operator go nie zatwierdził. Graf skończył (next jest puste). To wznowienie działało w osobnym procesie Pythona, uruchomionym po zakończeniu pierwszego, więc decyzja operatora może przyjść później, nawet po restarcie programu. Jeśli operator odrzuci wszystko ({"approved": []}), tekst zostaje bez zmian, a mapowanie jest puste.

Co dzieje się z kodem węzła po wznowieniu?

Tu jest ważna pułapka. Po wznowieniu LangGraph wykonuje węzeł z interrupt() od początku, a interrupt() zwraca tym razem decyzję zamiast zatrzymywać graf. Zmierzyłem to licznikiem wywołań: węzeł review wywołał się raz przed wznowieniem i drugi raz po nim, razem dwa razy. Wniosek praktyczny: cokolwiek robisz w węźle przed interrupt(), wykona się dwa razy. Dlatego przed interrupt() nie wysyłaj maili, nie zapisuj do bazy ani nie wywołuj płatnych API. Kod z efektami ubocznymi umieść w osobnym węźle po przeglądzie, tak jak zrobiliśmy z węzłem replace.

Dlaczego decyzję operatora trzeba sprawdzić przed wznowieniem?

Naturalna pokusa to sprawdzać poprawność decyzji w samym węźle review: jeśli numer jest spoza zakresu, rzuć wyjątek. Sprawdziłem, co się wtedy dzieje w LangGraph 1.2.18: węzeł rzucił ValueError po wznowieniu, a zła decyzja została zapisana razem z checkpointem. Gdy potem wznowiłem ten sam wątek poprawną decyzją, dostałem ten sam ValueError, bo odtworzona została zapisana zła wartość. Wątek utknął.

Dlatego parse_choice działa poza grafem: błędna odpowiedź kończy się wyjątkiem, zanim Command(resume=...) w ogóle trafi do LangGraph, więc stan grafu zostaje czysty i operator może spróbować jeszcze raz. To zachowanie sprawdziłem na wersji 1.2.18, więc przy innej wersji zrób własny test. Zasada walidacji przed wznowieniem pozostaje jednak bezpieczna w każdej wersji.

Jakie są ograniczenia human-in-the-loop?

Drugi wiersz jest realny: propozycje przekazane do interrupt() zawierają oryginalny tekst i wartości, a cały stan zapisuje się do bazy. W artykule o checkpointach pokazałem, że PESEL jest widoczny w pliku w otwartym tekście, więc to samo dotyczy propozycji.

Jak przetestować human-in-the-loop w pytest?

Testy sprawdzają zatrzymanie, zatwierdzanie podzbioru, odrzucenie wszystkiego, wznowienie po restarcie oraz ponowne wykonanie węzła. Osobne testy dotyczą funkcji parse_choice:

import pytest
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command

import review

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


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


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


@pytest.fixture
def db(tmp_path):
    return str(tmp_path / "review.db")


def test_graph_stops_at_review_and_shows_proposals(db):
    with SqliteSaver.from_conn_string(db) as saver:
        graph = review.build(saver)
        graph.invoke(start_state(), config())
        snapshot = graph.get_state(config())
    assert snapshot.next == ("review",)
    proposals = snapshot.tasks[0].interrupts[0].value["findings"]
    assert [f["kind"] for f in proposals] == ["pesel", "email", "phone"]


def test_only_approved_findings_are_replaced(db):
    with SqliteSaver.from_conn_string(db) as saver:
        graph = review.build(saver)
        graph.invoke(start_state(), config())
        result = graph.invoke(Command(resume={"approved": [0, 1]}), config())
    assert result["pseudonymized"] == "Klient [PESEL_1] pisze z [EMAIL_1]. Faktura 600123456."


def test_rejecting_everything_leaves_text_unchanged(db):
    with SqliteSaver.from_conn_string(db) as saver:
        graph = review.build(saver)
        graph.invoke(start_state(), config())
        result = graph.invoke(Command(resume={"approved": []}), config())
    assert result["pseudonymized"] == TEXT
    assert result["mapping"] == {}


def test_decision_can_arrive_after_restart(db):
    with SqliteSaver.from_conn_string(db) as saver:
        review.build(saver).invoke(start_state(), config())
    with SqliteSaver.from_conn_string(db) as saver:
        graph = review.build(saver)
        result = graph.invoke(Command(resume={"approved": [0]}), config())
    assert result["pseudonymized"].startswith("Klient [PESEL_1] pisze z a@b.pl")


def test_review_node_runs_again_after_resume(db, monkeypatch):
    runs = []
    original = review.review
    monkeypatch.setattr(review, "review", lambda state: runs.append(1) or original(state))
    with SqliteSaver.from_conn_string(db) as saver:
        graph = review.build(saver)
        graph.invoke(start_state(), config())
        graph.invoke(Command(resume={"approved": []}), config())
    assert len(runs) == 2


@pytest.mark.parametrize("raw,expected", [
    ("1,3", [0, 2]), (" 2 ", [1]), ("wszystko", [0, 1, 2]), ("nic", []), ("2,2,1", [0, 1]),
])
def test_parse_choice_accepts_valid_input(raw, expected):
    assert review.parse_choice(raw, 3) == expected


@pytest.mark.parametrize("raw", ["0", "4", "a", "1,,2"])
def test_parse_choice_rejects_invalid_input(raw):
    with pytest.raises(ValueError):
        review.parse_choice(raw, 3)

Test test_review_node_runs_again_after_resume dokumentuje opisaną wyżej pułapkę: jeśli kiedyś zmieni się zachowanie LangGraph, test to wykryje.

Jak traktować odpowiedź człowieka w grafie?

Traktuj odpowiedź człowieka jak dane wejściowe z zewnątrz: sprawdź ją i zamień na bezpieczny typ jeszcze przed Command(resume=...). Błędna decyzja, która dotarła do grafu, zostaje w checkpoincie, a kolejna próba może odtworzyć ten sam błąd (na LangGraph 1.2.18 tak właśnie było). Operatorowi łatwiej powiedzieć „spróbuj jeszcze raz”, gdy graf nic jeszcze nie zapisał.

Co pokazać operatorowi, żeby zatwierdzenie nie było formalnością?

Lista „1. phone: 600123456” nie powie operatorowi, że to numer faktury. Pokaż fragment tekstu wokół każdego trafienia (kilkadziesiąt znaków przed i po). Decyzja „zatwierdź wszystko” bez kontekstu to najczęstsza droga do tego, że zatwierdzanie staje się formalnością.

Co dalej?

Mamy graf, który wykrywa dane, pokazuje je człowiekowi i podmienia tylko to, co zatwierdzi. Zostaje najważniejsze: mapowanie token → oryginał leży w stanie, a stan w pliku, w otwartym tekście. W ostatniej części cyklu zaszyfrujemy mapowanie biblioteką cryptography. Całą ścieżkę Pythona znajdziesz na stronie ścieżki rozwoju w Pythonie.