Agent AI jako graf stanów w LangGraph: checkpointy, przerwania i odtwarzanie

Najprostszy agent to pętla: wyślij rozmowę do modelu, wykonaj wywołania narzędzi, o które prosi, dopisz wyniki i powtarzaj, aż przestanie prosić. Na demo i przy jednym czy dwóch narzędziach to działa. Potem agent musi przetrwać restart procesu albo poczekać na człowieka, albo ktoś chce zdebugować przebieg kilka dni po fakcie.

Alternatywa to jawny przepływ sterowania. Agent staje się grafem stanów: węzły to kroki, krawędzie to nazwane przejścia, stan to typowana struktura, a runtime zapisuje go po każdym kroku. Model podejmuje decyzję w węźle, a graf wybiera następny krok. Przykłady są w LangGraph, ale to samo da się zrobić w każdym orkiestratorze z podobnymi prymitywami.

Czego brakuje pętli

Zacznijmy od ponowień. Nieudane wywołanie narzędzia w pętli ponawia model, twój kod albo nikt, a jedynym globalnym bezpiecznikiem jest licznik kroków. Nie da się powiedzieć „wyszukiwanie próbuj najwyżej trzy razy, płatności nie ponawiaj nigdy”.

Stan pośredni nigdzie się nie zapisuje, bo pętla trzyma go w zmiennych lokalnych. Jeśli proces padnie w kroku 7, wszystko przepada, a start od zera powtarza kroki 1–6 razem z ich efektami ubocznymi. Z tego samego powodu nie da się czekać na człowieka: akceptacja oznacza zatrzymanie w połowie przebiegu, czasem na kilka dni, i wznowienie w innym procesie, który nie ma się do czego odwołać.

Zostaje odtwarzanie. Pętla zostawia co najwyżej log, a do zdebugowania złego przebiegu potrzebny jest dokładny stan sprzed błędnej decyzji i możliwość uruchomienia od tego miejsca jeszcze raz.

Sam warunek końca pętli też potrafi zawieść. Testowałem gotowy harness agentowy z lokalnym modelem: w trybie nieinteraktywnym wystarczyła jedna odpowiedź bez wywołania narzędzia i pętla się kończyła, więc model stawał w połowie zadania. Harness przekazał wznowionemu przebiegowi 26,6 tys. tokenów historii, a model i tak twierdził, że jej nie widzi. To był błąd modelu. Z poleceniem „pracuj do końca, nie pytaj” poprawka w końcu się udała, ale zmiana jednego znaku trwała 19 minut. Same wywołania narzędzi były bezbłędne w każdym przebiegu.

Graf

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy


class AgentState(TypedDict):
    task: str
    plan: list[str]
    context: Annotated[list[str], operator.add]  # dopisywane do listy, bez nadpisywania
    result: str | None
    needs_context: bool
    done: bool
    reflections: int


def route_after_plan(state: AgentState) -> str:
    return "retrieve" if state["needs_context"] else "act"


def route_after_act(state: AgentState) -> str:
    if state["done"] or state["reflections"] >= 2:
        return END
    return "reflect"


builder = StateGraph(AgentState)
builder.add_node("plan", planner)
builder.add_node("retrieve", retriever, retry_policy=RetryPolicy(max_attempts=3))
builder.add_node("act", tool_runner)
builder.add_node("reflect", critic)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", route_after_plan, ["retrieve", "act"])
builder.add_edge("retrieve", "act")
builder.add_conditional_edges("act", route_after_act, ["reflect", END])
builder.add_edge("reflect", "plan")

planner, retriever, tool_runner i critic to zwykłe funkcje, które dostają stan i zwracają częściową aktualizację. Funkcje routujące to czysty Python i da się je testować jednostkowo bez modelu.

reflections to reguła biznesowa trzymana w stanie: najwyżej dwie rundy krytyki. Niezależnie od niej recursion_limit w konfiguracji przebiegu ogranicza technicznie łączną liczbę kroków. Po jego przekroczeniu leci GraphRecursionError i przebieg się kończy. Domyślny limit w langgraph 1.2 to 10007 kroków, więc własny warto podać jawnie, jak w przykładzie niżej. Polityka ponowień dotyczy tylko węzła retrieve, więc narzędzia z efektami ubocznymi nie ponowi. Samo wyszukiwanie też nie przy każdym błędzie. Dopóki nie ustawisz własnego retry_on, RetryPolicy nie ponawia ValueError, RuntimeError, OSError i kilku innych wbudowanych błędów ani odpowiedzi 4xx z httpx. Ponawia wbudowany ConnectionError, odpowiedzi 5xx i każdy inny wyjątek, także twoje własne klasy. Z requests jest odwrotnie, niż można by się spodziewać: HTTPError ponawia przy każdym statusie, także 404, a requests.ConnectionError i Timeout (podklasy OSError) nie ponawia wcale. Jeśli to ma znaczenie, podaj własne retry_on.

Checkpointy w Postgresie

Checkpointer zapisuje stan po każdym kroku (w terminologii LangGraph: po każdym super-stepie) pod identyfikatorem wątku. Implementacja dla Postgresa jest w pakiecie langgraph-checkpoint-postgres i korzysta z psycopg 3.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://agent:secret@localhost:5432/agents"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # tworzy tabele checkpointów; uruchamiać raz, jak migrację
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "loan-app-20931"}, "recursion_limit": 30}
    graph.invoke({"task": "...", "context": [], "reflections": 0}, config)

Po awarii graph.invoke(None, config) z tym samym thread_id rusza od ostatniego zakończonego kroku. graph.get_state(config) zwraca bieżący stan i następny węzeł, a graph.get_state_history(config) wszystkie wcześniejsze snapshoty.

Z każdego checkpointu da się odtworzyć cały stan. Duże teksty w stanie (pełne pobrane dokumenty, pliki w base64) trafiają do każdego wiersza checkpointu, a lista dopisywana przez reducer zapisuje się w całości przy każdej zmianie. W stanie trzymaj referencje, a treść gdzie indziej. Checkpointy w każdym wątku przybywają też bez końca, dopóki nie dodasz zadania retencji. Pakiet daje do tego delete_thread() i prune(), które ze strategią keep_latest zostawia tylko ostatni checkpoint. ShallowPostgresSaver jest przestarzały od wersji 2.0.20 pakietu. W środowisku regulowanym retencja i tak musi się zgadzać z polityką przechowywania danych.

Czekanie na człowieka

interrupt() zatrzymuje graf wewnątrz węzła, zapisuje stan i oddaje payload wywołującemu. Przebieg wznawia się później, z dowolnego procesu, przez Command(resume=...).

from langgraph.types import Command, interrupt


def approve_refund(state: AgentState) -> dict:
    # Zakłada pola `amount` i `approved` zadeklarowane w `AgentState`.
    answer = interrupt({"question": "Zatwierdzić zwrot?", "amount": state["amount"]})
    return {"approved": answer == "yes"}


# Kilka godzin później, np. w handlerze HTTP ekranu akceptacji:
graph.invoke(Command(resume="yes"), {"configurable": {"thread_id": "loan-app-20931"}, "recursion_limit": 30})

Przy wznowieniu węzeł wykonuje się ponownie od pierwszej linii, a interrupt() zamiast zatrzymywać zwraca wartość z resume. Cały kod przed interrupt() wykona się więc dwa razy. Ta część nie może mieć efektów ubocznych. Jeśli jakiś jest potrzebny, przenieś go do osobnego węzła za akceptacją.

Sprawdź też, jak długo runtime w ogóle chce czekać. Jeden orkiestrator agentów miał zaszyty limit 15 minut na pauzę dla zgody człowieka. Po nim cały przebieg kończył się błędem i dalsze kroki już nie ruszały. Zgoda często przychodzi po kilku godzinach, więc pauzę trzeba zapisać w stanie przebiegu.

Przed czym checkpoint nie chroni

Checkpoint zapisuje się po zakończeniu węzła. Jeśli węzeł wyśle maila, obciąży kartę albo strumieniuje odpowiedź do użytkownika, a proces padnie przed zapisem checkpointu, wznowienie uruchomi węzeł jeszcze raz. Gwarancja na poziomie węzła to at-least-once.

Z tego wynika kształt węzłów. W jednym węźle jeden zewnętrzny efekt uboczny, żeby to, co może się powtórzyć, było małe i wiadomo było co. Klucz idempotencji składasz z thread_id, nazwy węzła i numeru kroku (langgraph_step w metadanych configu), bo w grafie z pętlą ten sam węzeł wykonuje się w wątku więcej niż raz, a powtórka po awarii dostaje ten sam numer. Klucz przekazujesz do systemu zewnętrznego, jeśli go obsługuje (Stripe przyjmuje na przykład nagłówek Idempotency-Key). W API v1 Stripe może usunąć klucz po 24 godzinach, więc powtórka po kilkudniowej pauzie może już przejść jako nowe żądanie. Przy wznowieniu węzeł strumieniuje jeszcze raz, więc klient powinien traktować strumień jako podgląd i zastąpić to, co już wyświetlił, wartością, którą węzeł zwrócił do stanu w checkpoincie, zamiast doklejać drugą kopię.

from langchain_core.runnables import RunnableConfig


def send_confirmation(state: AgentState, config: RunnableConfig) -> dict:
    # Zakłada pola `customer_email` i `confirmation_sent` zadeklarowane w `AgentState`.
    thread_id = config["configurable"]["thread_id"]
    step = config["metadata"]["langgraph_step"]
    # `mailer` to twój klient; dostawca usuwa duplikaty po tym kluczu.
    mailer.send(
        to=state["customer_email"],
        body=state["result"],
        idempotency_key=f"{thread_id}:send_confirmation:{step}",
    )
    return {"confirmation_sent": True}

Podobnie tnę zadania dla agentów, którzy piszą mi kod: paczki po około 250–300 linii diffu. Całe fazy (550–930 linii) wracały z poprawkami mniej więcej co trzecia.

Odtwarzanie do debugowania i regresji

Każdy snapshot z get_state_history ma własny config z ID checkpointu. Wywołanie grafu z None i tym configiem uruchamia przebieg od tego punktu i tworzy nową gałąź historii. Oryginał zostaje nietknięty. Jeśli węzły dostają narzędzia przez fabrykę (build_graph(tools)), można wziąć stan z nieudanego przebiegu na produkcji, zbudować graf z nagranymi albo zamockowanymi narzędziami i sprawdzić, czy zmieniony prompt albo reguła routingu prowadzi do innej decyzji. Stan z produkcji może zawierać dane osobowe. Środowisko odtwarzania potrzebuje takiej samej kontroli dostępu jak produkcja.

Obserwowalność

Loguj jedno ustrukturyzowane zdarzenie na każde wykonanie węzła. Wystarczy owinąć węzły przy rejestracji, ich kod zostaje bez zmian:

import inspect
import logging
import time

from langgraph.errors import GraphBubbleUp

log = logging.getLogger("agent")


def observed(name, fn):
    # Węzeł może przyjmować sam stan albo (state, config).
    takes_config = len(inspect.signature(fn).parameters) > 1

    # Bez functools.wraps: LangGraph czytałby wtedy sygnaturę fn i przy węźle
    # z samym stanem nie przekazałby configu do wrappera.
    def wrapper(state, config):
        started = time.perf_counter()
        error = None
        try:
            return fn(state, config) if takes_config else fn(state)
        except GraphBubbleUp:  # interrupt() zatrzymuje graf, rzucając ten wyjątek
            raise
        except Exception as exc:
            error = type(exc).__name__
            raise
        finally:
            log.info("agent_step", extra={
                "thread_id": config["configurable"]["thread_id"],
                "node": name,
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "error": error,
            })
    return wrapper


# Zamiast builder.add_node("act", tool_runner); nazwę węzła można zarejestrować tylko raz.
builder.add_node("act", observed("act", tool_runner))

Liczniki tokenów należą do tego samego zdarzenia. W modelach czatowych LangChain są dostępne na wiadomości odpowiedzi jako usage_metadata; zwróć je z węzła albo zaloguj tam, gdzie wołasz model. Gdy thread_id, węzeł, czas, tokeny i błąd są w jednym rekordzie, koszt przebiegu, wolne narzędzie albo węzeł, który zaczął padać po deployu, znajdziesz jednym zapytaniem.

Raz krok wykonywany przez Codexa dostał zły model, API odrzuciło żądanie w 2 sekundy, a orkiestrator, który go prowadził, oznaczył krok jako zrobiony. Testy i lint przeszły, tyle że na niezmienionym kodzie.

Kiedy graf to przesada

Pojedyncze wywołanie modelu ze strukturalnym wyjściem albo stała sekwencja kroków bez rozgałęzień lepiej wychodzi jako zwykła funkcja. Krótkie synchroniczne zadanie, którego nikt nie zatwierdza i nikt nie musi odtwarzać, nie uzasadnia tabeli checkpointów i dodatkowego zapisu do bazy na każdym kroku. Graf zaczyna się zwracać, kiedy przebieg może trwać na tyle długo, że coś go przerwie, kiedy ktoś w połowie coś zatwierdza albo kiedy nieudany przebieg trzeba potem odtworzyć.