LangGraph, agenci z trwalym stanem i grafem

LangGraph, budowa agentów jako grafu z trwałym stanem

LangGraph opisuje agenta jako graf: węzły wykonują pracę, krawędzie decydują, co dzieje się dalej, a wspólny stan przechodzi między nimi i jest zapisywany po każdym kroku. Dzięki temu proces przerwany awarią wznawia się od miejsca zatrzymania, a nie od początku. Wersja 1.0 wyszła razem z LangChain 1.0 17 października 2025 roku, a linia rozwojowa doszła od tego czasu do 1.2.

Czym różni się od LangChain

LangChain daje gotowego agenta w jednej funkcji i pętlę, która sama decyduje, kiedy się zakończyć. LangGraph odbiera tę decyzję modelowi i oddaje ją Tobie: sam rysujesz przepływ, wskazujesz warunki przejścia i miejsca, w których proces ma się zatrzymać.

Ta różnica ma konsekwencje. W prostym agencie nie wiesz, ile obrotów wykona model ani w jakiej kolejności sięgnie po narzędzia. W grafie wiesz, bo przepływ jest zapisany w kodzie i daje się przetestować bez wywoływania modelu. Za to każdy graf trzeba zaprojektować, co przy zadaniu na trzy kroki jest pracą wykonaną niepotrzebnie.

Warto też spojrzeć na to od strony zespołu, który będzie utrzymywał efekt. Pętla agenta jest zrozumiała dla każdego, kto zna prompt, natomiast graf wymaga przeczytania kodu, żeby wiedzieć, co się wydarzy. W zamian daje jednoznaczną odpowiedź na pytanie, dlaczego proces poszedł tą, a nie inną ścieżką, bo warunek przejścia jest zapisany wprost. Przy zgłoszeniu od klienta ta różnica decyduje o tym, czy diagnoza zajmuje kwadrans, czy pół dnia zgadywania, co model miał na myśli.

Praktyczna granica przebiega tam, gdzie proces przestaje mieścić się w jednym wywołaniu. Rozmowa z jednym narzędziem to zadanie dla LangChain. Proces, który czeka na akceptację człowieka, rozgałęzia się zależnie od wyniku walidacji i musi przeżyć restart serwera, to zadanie dla LangGraph.

Graf nie jest przy tym jedynym sposobem opisania takiego procesu. LlamaIndex Workflows modeluje to samo zdarzeniami: krok nasłuchuje zdarzenia określonego typu i sam emituje kolejne, więc dopisanie etapu nie wymaga przerysowania krawędzi. Cena jest dokładnie odwrotna niż tutaj, bo przepływu nie widać w jednym miejscu i trzeba go odtworzyć z typów zdarzeń rozsianych po krokach, co przy diagnozie zgłoszenia bywa trudniejsze niż przeczytanie grafu.

Instalacja i pierwszy graf

pip install langgraph langchain-openai

Minimalny graf ma trzy elementy: schemat stanu, węzły i krawędzie.

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END

class Stan(TypedDict):
    zapytanie: str
    fragmenty: Annotated[list[str], add]
    odpowiedz: str

def wyszukaj(stan: Stan) -> dict:
    return {"fragmenty": retriever.invoke(stan["zapytanie"])}

def odpowiedz(stan: Stan) -> dict:
    kontekst = "\n".join(stan["fragmenty"])
    return {"odpowiedz": model.invoke(f"{kontekst}\n\nPytanie: {stan['zapytanie']}").content}

graf = StateGraph(Stan)
graf.add_node("wyszukaj", wyszukaj)
graf.add_node("odpowiedz", odpowiedz)
graf.add_edge(START, "wyszukaj")
graf.add_edge("wyszukaj", "odpowiedz")
graf.add_edge("odpowiedz", END)

aplikacja = graf.compile()
aplikacja.invoke({"zapytanie": "jak dziala zwrot towaru"})

Każdy węzeł to zwykła funkcja, która dostaje stan i zwraca słownik ze zmianami. Nie musi zwracać całego stanu, tylko pola, które modyfikuje. To drobna rzecz, która sporo upraszcza przy dziesięciu węzłach.

Stan, reducery i typowanie

Adnotacja Annotated[list[str], add] w schemacie stanu wskazuje reducer, czyli sposób łączenia nowej wartości ze starą. Bez niej każdy węzeł nadpisuje pole. Z nią wartości się sumują, co jest potrzebne wszędzie tam, gdzie kilka węzłów dokłada coś do wspólnej listy.

Najczęstszy błąd na tym etapie polega na trzymaniu w stanie zbyt wiele. Stan jest zapisywany po każdym kroku, więc wrzucenie do niego całych dokumentów zamiast ich identyfikatorów podnosi koszt zapisu i rozmiar checkpointu. Trzymaj w nim to, co jest potrzebne do podjęcia kolejnej decyzji, a resztę pobieraj w węźle.

Rozgałęzienia opisujesz funkcją, która zwraca nazwę następnego węzła.

def czy_wystarczy(stan: Stan) -> str:
    return "odpowiedz" if len(stan["fragmenty"]) >= 3 else "rozszerz_zapytanie"

graf.add_conditional_edges("wyszukaj", czy_wystarczy)

Funkcja warunkowa może też zwrócić listę nazw, co uruchamia kilka węzłów równolegle i scala ich wyniki reducerem. Przydaje się przy zapytaniach do kilku źródeł naraz, gdzie sekwencyjne odpytywanie niepotrzebnie wydłuża odpowiedź.

Warunek jest zwykłym Pythonem, więc przetestujesz go bez sieci i bez tokenów. To jedna z lepszych stron tego podejścia: logika sterowania jest testowalna osobno od modelu.

Trwałe wykonanie i checkpointery

Checkpointer zapisuje stan po każdym kroku pod identyfikatorem wątku. Jeśli proces padnie w połowie, wznawiasz go tym samym identyfikatorem i graf rusza od ostatniego zapisanego węzła.

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as saver:
    aplikacja = graf.compile(checkpointer=saver)
    aplikacja.invoke(
        {"zapytanie": "jak dziala zwrot towaru"},
        config={"configurable": {"thread_id": "zgloszenie-8842"}}
    )

Tryb trwałości ustawia się parametrem durability. Wartość async zapisuje stan w tle, podczas gdy rusza kolejny krok, i jest domyślnym kompromisem. Wartość sync zapisuje przed rozpoczęciem następnego kroku, przez co nic nie ginie nawet przy nagłym ubiciu procesu, kosztem kilkudziesięciu milisekund na węzeł. Wartość exit zapisuje dopiero na końcu i nadaje się do krótkich przebiegów, w których wznawianie nie ma sensu.

Wybór ma znaczenie tam, gdzie węzeł wywołuje operację z efektem ubocznym. Jeśli węzeł wysyła przelew, a stan zapisze się dopiero po nim asynchronicznie, restart w złym momencie potrafi doprowadzić do powtórzenia operacji. Przy takich węzłach ustaw sync i dołóż klucz idempotencji po swojej stronie.

Sam PostgresSaver nie przychodzi z pakietem langgraph, tylko z osobnym langgraph-checkpoint-postgres, więc dopisz go do zależności, zanim skopiujesz powyższy import. Przy pierwszym uruchomieniu wywołaj też setup() na checkpointerze, bo to on zakłada tabele.

W testach lokalnych wystarczy InMemorySaver z pakietu langgraph-checkpoint, który jest zależnością podstawową. W produkcji Postgres, a baza z Neona albo Supabase obsługuje ten scenariusz bez dodatkowej infrastruktury, bo checkpointer potrzebuje tylko kilku tabel.

Human in the loop

Zatrzymanie procesu na człowieku jest w LangGraph elementem języka, nie obejściem. Wywołanie interrupt przerywa wykonanie, zwraca dane do sprawdzenia i czeka.

from langgraph.types import interrupt, Command

def zatwierdz_zwrot(stan: Stan) -> dict:
    decyzja = interrupt({"kwota": stan["kwota"], "zamowienie": stan["numer"]})
    if decyzja["akcja"] == "odrzuc":
        return {"status": "odrzucone", "powod": decyzja.get("powod", "")}
    return {"status": "zatwierdzone"}

aplikacja.invoke(Command(resume={"akcja": "zatwierdz"}), config=konfiguracja)

Proces może stać w tym miejscu godzinami, bo stan siedzi w bazie, a nie w pamięci procesu. Serwer wolno w tym czasie zrestartować, wdrożyć nową wersję albo przeskalować w dół. Wznawiasz przez podanie tego samego thread_id i decyzji operatora.

Przy projektowaniu tego kroku zadbaj o dwie rzeczy. Po pierwsze, dane przekazywane do przerwania powinny wystarczyć operatorowi do decyzji bez zaglądania w inny system, bo inaczej akceptacja zamienia się w przełączanie okien. Po drugie, przewiduj wygaśnięcie: proces czekający trzy tygodnie na akceptację zwykle powinien zostać zamknięty automatycznie, a nie wisieć w bazie bez końca.

Ten mechanizm jest głównym powodem, dla którego zespoły przechodzą z prostego agenta na graf. Każdy proces dotykający pieniędzy, danych osobowych albo komunikacji z klientem prędzej czy później dostaje wymóg akceptacji, a doklejenie go do pętli agenta kończy się własnym systemem kolejek.

Topologie wieloagentowe

Kilku agentów da się połączyć na kilka sposobów i wybór wpływa na koszt oraz na to, jak trudno zdiagnozować błąd.

Zacznij od pojedynczego agenta i rozbijaj dopiero wtedy, gdy zestaw narzędzi przestaje się mieścić w jednym promptcie. Podział na pięciu agentów w pierwszym tygodniu projektu zwykle kończy się systemem, w którym nikt nie potrafi wskazać, który z nich podjął złą decyzję.

Streaming i podgląd przebiegu

Graf strumieniuje na kilku poziomach: tokeny z modelu, aktualizacje stanu po każdym węźle albo pełne migawki stanu. W interfejsie użytkownika najbardziej przydaje się drugi tryb, bo pozwala pokazać, na jakim etapie jest proces.

for typ, dane in aplikacja.stream(wejscie, config=konfiguracja, stream_mode=["updates", "messages"]):
    if typ == "updates":
        print("wezel:", list(dane.keys()))

Do diagnostyki poza interfejsem zostaje śledzenie w LangSmith, gdzie każdy przebieg widać jako drzewo węzłów z czasami i tokenami. Bez tego debugowanie grafu z dwunastoma węzłami sprowadza się do czytania logów, co przy rozgałęzieniach przestaje wystarczać.

Jak testować graf

Graf ma tę przewagę nad pętlą agenta, że da się go testować warstwami, a każda warstwa wymaga innego rodzaju testu.

Najniżej leżą funkcje warunkowe. To zwykły kod bez modelu, więc pokrywasz je testami jednostkowymi i sprawdzasz przypadki brzegowe: pusty wynik wyszukiwania, wartość na granicy progu, brak wymaganego pola w stanie. Te testy chodzą w milisekundach i wyłapują większość błędów sterowania.

Wyżej są węzły wywołujące model. Tu podmieniasz model na atrapę zwracającą ustaloną odpowiedź i sprawdzasz, czy węzeł poprawnie przekształca ją w zmianę stanu. Chodzi o kontrakt, nie o jakość odpowiedzi, więc atrapa wystarczy.

Na końcu jest cały graf. Uruchamiasz go z InMemorySaver na zestawie przypadków i porównujesz ścieżkę, którą przeszedł, z oczekiwaną listą węzłów.

from langgraph.checkpoint.memory import InMemorySaver

def test_krotkie_wyszukiwanie_rozszerza_zapytanie():
    aplikacja = graf.compile(checkpointer=InMemorySaver())
    konfiguracja = {"configurable": {"thread_id": "test-1"}}
    aplikacja.invoke({"zapytanie": "zwrot", "fragmenty": []}, config=konfiguracja)
    odwiedzone = [s.next for s in aplikacja.get_state_history(konfiguracja)]
    assert "rozszerz_zapytanie" in str(odwiedzone)

Osobnego testu wymagają przerwania. Sprawdź obie ścieżki: wznowienie z akceptacją i wznowienie z odrzuceniem, bo w praktyce to gałąź odrzucenia bywa nieprzetestowana i to ona zostawia proces w stanie, z którego nie ma wyjścia.

Warto też trzymać zestaw kilkunastu realnych zapytań z oczekiwanym efektem końcowym i przepuszczać go po każdej zmianie promptu. To już nie jest test jednostkowy, tylko pomiar jakości, więc uruchamiaj go świadomie, bo kosztuje tokeny.

Koszty i LangSmith Deployment

Sama biblioteka jest open source na licencji MIT i uruchomisz ją na własnym serwerze bez opłat. Płatny jest hosting grafów z kolejką zadań, wznawianiem i panelem do podglądu przebiegów. Nazwa się zmieniła: to, co LangChain sprzedawał jako LangGraph Platform, występuje teraz w cenniku jako LangSmith Deployment, a plany są wspólne dla śledzenia i wdrożeń.

Stawka za stanowisko to dopiero pierwsza połowa rachunku i tabela sama tego nie pokaże. Wszystko, co agent faktycznie zużyje, liczy się osobno w dwóch jednostkach: LCU po 1,50 USD za pracę i moc obliczeniową oraz LSU po 1 USD za ślady i magazyn. Pula śladów wliczona w plan nie jest twardym sufitem, tylko progiem, po którym wchodzi rozliczenie za zużycie, więc rachunek za miesiąc z jednym stanowiskiem bywa wielokrotnością tych 39 USD. Wdrożenie w chmurze wymaga przy tym co najmniej planu Plus, a uruchomienie panelu sterowania we własnym klastrze dopiero planu Enterprise.

Rachunek za tokeny nadal dominuje. Graf z nadzorcą i trzema podagentami wykonuje przy jednym zapytaniu od pięciu do kilkunastu wywołań modelu, więc różnica między modelem dużym a mniejszym w węzłach pomocniczych zmienia koszt kilkukrotnie. Kieruj do dużego modelu tylko te węzły, w których decyzja jest naprawdę trudna.

Typowe błędy

Pierwszy to graf bez limitu kroków. Rozgałęzienie, które w pewnym układzie danych wraca do wcześniejszego węzła, tworzy pętlę wykonującą się do wyczerpania budżetu. Ustaw recursion_limit i potraktuj jego przekroczenie jako błąd wymagający poprawy przepływu.

Drugi to stan rosnący w nieskończoność. Lista wiadomości dokładana reducerem w każdym obrocie przekroczy okno kontekstu po kilkudziesięciu krokach. Dołóż węzeł streszczający albo przycinający historię.

Trzeci to trzymanie sekretów w stanie. Wszystko, co wejdzie do stanu, ląduje w bazie checkpointów i w śledzeniu, więc token dostępu użytkownika przekazuj przez konfigurację, a nie przez pola stanu.

Czwarty to projektowanie grafu pod obecny prompt. Jeśli węzły są nazwane jak fragmenty instrukcji dla modelu, każda zmiana promptu wymusza zmianę struktury. Nazywaj węzły od kroków procesu biznesowego, bo te zmieniają się rzadziej.

Kiedy nie sięgać po LangGraph

Ostatni wiersz to jedyny, w którym graf wygrywa bez zastrzeżeń. W pozostałych dokłada warstwę, którą trzeba utrzymywać.

FAQ

Czy LangGraph wymaga LangChain?

Nie. LangGraph korzysta z abstrakcji langchain-core, ale nie wymusza używania łańcuchów ani agentów z LangChain. Węzeł może wywoływać SDK dostawcy bezpośrednio, a graf nadal zapewni stan, wznawianie i zatrzymania.

Czy działa w TypeScripcie?

Tak, biblioteka ma wersję JavaScript i TypeScript z tym samym modelem grafu, stanu i checkpointerów. Nowe funkcje pojawiają się zwykle najpierw w Pythonie, więc przy egzotycznych integracjach warto sprawdzić, czy odpowiednik już istnieje. Typowanie stanu po stronie TypeScriptu działa dobrze i wyłapuje literówki w nazwach pól.

Jak wznowić proces po restarcie serwera?

Wywołaj graf ponownie z tym samym thread_id w konfiguracji. Checkpointer odtworzy stan z ostatniego zapisanego kroku i wykonanie ruszy od następnego węzła. Warunkiem jest checkpointer na trwałym magazynie, bo wersja w pamięci znika razem z procesem.

Czy da się podejrzeć, którędy poszedł graf?

Tak, każdy przebieg zostawia historię stanów dostępną przez get_state_history, a przy włączonym śledzeniu także drzewo węzłów w LangSmith. Historia stanów pozwala też cofnąć proces do wybranego kroku i puścić go inną gałęzią.

Ile węzłów to za dużo?

Graf powyżej dwudziestu węzłów zwykle sygnalizuje, że jeden proces robi kilka rzeczy naraz. Zamiast rozbudowywać go dalej, wydziel podgraf i wywołaj go jako pojedynczy węzeł. Podgraf ma własny stan i testuje się osobno.

Dokumentacja trwałego wykonania stoi w docs.langchain.com, a kod w repozytorium na GitHubie.