Sześć podstawowych elementów LangGraph i ukryty w każdym z nich sposób awarii

Pierwszy agent LangGraph zazwyczaj powstaje szybko: odpowiada na pytanie, udoskonala odpowiedź i przestaje działać. Następnie ktoś dodaje gałąź do ponownej próby, i nagle graf przestaje się kończyć, zużywając kredyty API aż do zatrzymania procesu. Rozwiązaniem jest często brakujący pojedynczy krawędź, ale prawdziwym problemem jest brak modelu mentalnego wyjaśniającego, dlaczego graf zachowuje się w taki sposób.

To przewodnik buduje ten model na podstawie sześciu elementów bazowych, z których składa się LangGraph: stanu, węzłów, krawędzi bezpośrednich, krawędzi warunkowych, mechanizmu przechowywania stanów w punktach kontrolnych oraz udziału człowieka w procesie. Dla każdego z tych elementów przedstawiono minimalny przykład, najczęstsze błędy popełniane przez zespoły przy ich używaniu oraz wersję nadającą się do wprowadzenia do produkcji. Jeśli chcesz szerszego przeglądu wzorców agentów opartych na tych elementach, LangGraph w praktyce: stan, węzły, krawędzie i pięć wzorców agentów omawia ten temat; tutaj natomiast skupiamy się na możliwych sposobach awarii.

Dlaczego graf zamiast łańcucha

Syntaks rury LangChain, prompt | llm | parser, jest wygodna przy jednorazowym przetwarzaniu danych przez model. Przestaje być skuteczna, gdy agent musi podjąć decyzję: czy szukać, czy odpowiadać bezpośrednio, czy spróbować ponownie, czy poddać się, czy zapytać człowieka, czy kontynuować. Łańcuch nie ma pojęcia o „zależy to od okoliczności”, dlatego programiści otaczają wywołania łańcucha instrukcjami if, a wkrótce tworzą własną, nieudokumentowaną maszynę stanową trudną do debugowania.

LangGraph czyni tę maszynę stanową jawną. Otrzymujemy węzły, krawędzie oraz jeden wspólny obiekt stanu, który można sprawdzić w dowolnym momencie. Nie ma w tym nic magicznego, a to właśnie jest jej zaletą: każda decyzja podejmowana przez agenta odpowiada elementom widocznym w definicji grafu.

1. Stan: jeden wspólny obiekt oraz istotne reduktory

Stan jest jedynym obiektem, z którego każdy węzeł odczytuje dane i do którego je zapisuje. Bez niego kontekst jest zwykle przekazywany jako argumenty funkcji, co utrudnia określenie, co dokładnie wiedział dany krok. Poniższa definicja to TypedDict zawierający pytanie, odpowiedź oraz listę wiadomości, których aktualizacje są łączone przez reduktora add_messages.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add nie jest reduktorem wiadomości

Wiele tutoriali zamiast tego oznacza pole wiadomości jako operator.add. Wygląda to sensownie: add dodaje element do listy zamiast ją nadpisywać, co jest konieczne w przypadku rozwijającej się rozmowy. Problem polega na tym, że łączy elementy bez żadnej kontroli. Gdy tylko potrzebujesz zaktualizować lub usunąć istniejącą wiadomość, na przykład podczas edycji historii lub zastępowania wyniku wywołania narzędzia, dodaje się wtedy duplikat, a historia rozmowy wypełnia się przestarzałymi wpisami bez żadnego błędu.

add_messages został stworzony specjalnie do tego celu. Porównuje wiadomości pod kątem ID i zastępuje istniejącą wiadomość, jeśli ten identyfikator już się pojawił, dodając jedynie te naprawdę nowe. Zasada jest prosta: używaj add_messages dla pól zawierających obiekty HumanMessage i AIMessage, a operator.add dla zwykłych list akumulacyjnych, takich jak bieżący rejestr narzędzi, które zostały wywołane.

Zachowaj minimalną strukturę stanu

Drugim powszechnym błędem jest projektowanie stanu na wzór schematu bazy danych, z polami dla każdej potrzeby, jaką ktoś mógłby później mieć. Dodawaj pole tylko wtedy, gdy węzeł rzeczywiście je odczytuje lub zapisuje. Konsekwencje ignorowania tego faktu są konkretne: weźmy na przykład graf do przetwarzania dokumentów, który przechowuje w stanie pełne surowe odpowiedzi LLM, włączając metadane dotyczące użycia tokenów. Przetwarzanie 50 dokumentów w pętli powodowało, że każdy punkt kontrolny zajmował około 180 KB, a czas zapisu w Postgresie przekraczał 400 ms – co jest na tyle wolne, że użytkownicy czekający na odpowiedź to zauważają. Rozwiązanie nie było efektowne: wystarczyło ograniczyć stan do trzech pól, których rzeczywiście używały węzły następne. Pamiętaj, że przy użyciu punktu kontrolnego wszystko w stanie jest serializowane i zapisywane na każdym kroku.

2. Węzły: zwracaj tylko to, co się zmieniło

Węzeł to zwykła funkcja w Pythonie. Otrzymuje stan, wykonywa swoją pracę i zwraca słownik zawierający tylko te pola, które zmienił. To cała umowa. Pierwszy przykład wywołuje model czatowy OpenAI z pytaniem i zapisuje odpowiedź do answer.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Gdy iterujesz po strukturze grafu, co stanowi większość wstępnej pracy, możesz nie chcieć, aby każdy projekt korzystał z płatnej API. Lokalny model dostarczany przez Ollama implementuje tę samą interfejs, więc treść węzła pozostaje identyczna, a debugowanie nie kosztuje nic:

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Wersja ta wymaga uruchomienia Ollamy lokalnie z pobrałym modelem (ollama pull llama3.1) oraz zainstalowanym pakietem integracyjnym (pip install langchain-ollama). Ustawienie temperature=0 sprawia również, że wykonywania są bardziej powtarzalne, co jest przydatne podczas testowania logiki routingu.

Zwracanie całego stanu zakłóca inne aktualizacje

Częstym błędem jest zwracanie całego słownika stanu z węzła, a nie tylko zmienionych kluczy. W małym grafie liniowym może to wydawać się skuteczne, ponieważ żadne inne elementy nie wpływają na te pola. Gdy dwa węzły aktualizują nakładające się pola, pełne dane zwrócone przez jeden węzeł nadpisują zmiany drugiego starymi wartościami. Objawy przypominają problem z routowaniem, dlatego programiści często szukają usterki w logice łączy, podczas gdy rzeczywistą przyczyną jest zbyt obszerne zwracanie danych przez węzeł. Zwracanie minimalnej aktualizacji pozwala również reduktorom prawidłowo wykonywać swoją funkcję – pole bez reduktora jest po prostu zastępowane tym, co zwraca węzeł.

3. Bezpośrednie łącza: zawsze podłącz wyjście

Krawędzie decydują, co będzie wykonywane dalej. Bezpośrednia krawędź jest bezwarunkowa: gdy węzeł A kończy swoją pracę, uruchamia się węzeł B. Poniższy graf rejestruje dwa węzły, łączy answer z refine, łączy refine z END, określa punkt wejścia i kompiluje się.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

Krawędź END to element, o którym ludzie zapominają, i stanowi klasyczne źródło problemów z grafem, który wydaje się działać w nieskończoność. Pełnie niezawodną praktyką jest wyraźne umieszczanie końca każdej ścieżki w grafie przy END, dzięki czemu można sprawdzić jej zakończenie poprzez przeczytanie definicji. Jest to szczególnie ważne, gdy pojawiają się cykle: pętla ponawiania bez ścieżki do END, lub z warunkiem, który nigdy nie staje się prawdziwy, powoduje dalsze cyklenie, aż limit rekurencji LangGraph zatrzyma ją błędem GraphRecursionError. Ten limit służy jako zabezpieczenie, a nie element projektu – każda z tych iteracji nadal kosztuje tokeny. Gdy graf wydaje się zawieszony, najpierw sprawdź jego definicję.

4. Krawędzie warunkowe: gdzie agent faktycznie podejmuje decyzję

Krawędzie warunkowe sprawiają, że graf funkcjonuje jak agent, a nie jako stała ścieżka przetwarzania. Funkcja routingu sprawdza stan i zwraca etykietę; funkcja mapowania przekształca każdą taką etykietę w następny węzeł. W tym przykładzie krótka odpowiedź (poniżej 50 znaków) jest wysyłana do refine, a wszystko inne trafia do END.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Należy zauważyć, że ta krawędź warunkowa zastępuje bezpośrednią krawędź answer do refine z poprzedniego fragmentu kodu. Jeśli zarejestrujesz obie, zostaną wybrane obie ścieżki, co rzadko jest tym, czego chcesz.

Niepasujące etykiety tras powodują wyraźne, ale trudne do zdiagnozowania błędy

Częstym błędem jest tutaj funkcja routingu, która zwraca ciąg znaków nieobecny w słowniku mapowania. Powstający błąd to dość ogólny błąd klucza, ukryty na kilku poziomach głęboko w śladzie stosu, a coś tak prostego jak spacja na końcu może kosztować zaskakująco dużo czasu. Niezawodny nawyk to napisanie najpierw mapowania, a następnie implementacji routera poprzez skopiowanie z niego dokładnych kluczy. Jeszcze lepiej jest zdefiniować etykiety raz jako stałe lub oznaczyć typ zwracany przez router za pomocą Literal["refine", "done"], aby narzędzia do sprawdzania typów i czytelnicy mogli od razu zobaczyć dozwolone wartości.

5. Punkty kontrolne: pamięć przetrwająca między wywołaniami

Checkpointer przekształca bezstanową funkcję wywoławczą w rozmowę z pamięcią. Bez niego każde app.invoke() rozpoczyna się od zera. Z jego użyciem stan jest przechowywany dla każdego wątku, a każde wywołanie, które zawiera tę samą wartość thread_id w swojej konfiguracji, kontynuuje to, co zakończyło poprzednie wywołanie. W przykładzie drugie wywołanie na wątku user-session-42 pamięta pierwsze pytanie.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver nadaje się tylko do lokalnego rozwoju. Znajduje się w pamięci procesu, więc ponowne uruchomienie serwera usuwa wszystkie rozmowy. Wszystko, od czego zależą rzeczywiści użytkownicy, wymaga trwałego backendu: SQLite dla pojedynczego serwera lub Postgres, gdy kilka instancji musi dzielić stan.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Każdy backend jest dostarczany jako odrębny pakiet, jak wskazują komentarze dotyczące instalacji. W obecnych wersjach takie narzędzia do zapisu są zazwyczaj tworzone na podstawie ciągu połączeniowego (na przykład za pomocą from_conn_string), a Postgres wymaga jednorazowego wywołania funkcji setup() w celu utworzenia swoich tabel, dlatego sprawdź dokumentację do checkpointera w celu poznania dokładnej procedury inicjalizacji w Twojej wersji.

Ryzyko polega na użyciu narzędzia do zapisu w pamięci operacyjnej w środowisku produkcyjnym i odkryciu tego problemu dopiero wtedy, gdy restart środowiska testowego usunie aktywną demonstrację. Dobrą wiadomością jest to, że zmiana jest prosta, jeśli struktura grafu jest dobrze zaprojektowana: checkpointer to argument określany podczas kompilacji, a nie element wymagający przebudowy, a przechodzenie na SqliteSaver może zająć mniej niż godzinę.

6. Człowiek w procesie: punkty zatrzymania statyczne versus przerwy dynamiczne

Wzorzec pokazywany w większości instrukcji to interrupt_before – lista nazw węzłów, przy których skompilowana struktura zatrzymuje się przed wykonaniem:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Funkcjonuje to i jest łatwe do wyjaśnienia, ale jest to rozwiązanie statyczne. Punkt zatrzymania jest określony przez nazwę węzła, nie można go uczynić warunkowym ani dołączyć do niego informacji opisujących, na co powinien zwrócić uwagę sprawdzający. Rzeczywiste wymagania szybko przewyższają możliwości tego podejścia, ponieważ „zatrzymać się przed tym węzłem” oraz „zatrzymać się tylko wtedy, gdy zwrot pieniędzy przekracza 500 dolarów” to różne zasady, a tylko pierwszą można wyrazić w ten sposób.

Zatrzymywanie z wnętrza węzła za pomocą interrupt()

Bardziej elastycznym rozwiązaniem jest wywołanie interrupt() bezpośrednio z wnętrza węzła. Poniższy węzeł sprawdza kwotę zwrotu; jeśli przekracza 500 dolarów, zatrzymuje się i pokazuje projekt oraz kwotę osobie odpowiedzialnej. Pierwsze wywołanie invoke trwa do chwili tego zatrzymania. Drugie wywołanie przekazuje Command(resume="approve") na tym samym wątku, a wartość podana do resume staje się wartością zwracaną przez interrupt(); w rezultacie węzeł albo kontynuuje wysyłkę, albo zwraca status anulowania. Potrzebny jest wskaźnik stanu, ponieważ stan zatrzymania musi być gdzieś przechowywany podczas oczekiwania.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

Przykładowy stan wykorzystuje pola takie jak refund_amount, draft i task, których nie ma we wcześniejszym typie AgentState; w rzeczywistym grafie należałoby je tam zadeklarować.

Kontynuacja uruchamia ponownie cały węzeł

Zachowanie, które zaskakuje ludzi: w kontekście kontynuacji wykonywania kodu LangGraph nie przechodzi dalej od wiersza interrupt(). Ponownie wykonywa cały węzeł od początku, a tym razem interrupt() zwraca wartość kontynuacji zamiast wstrzymywać wykonywanie. Wszystki kod znajdujący się przed tą funkcją jest wykonywany ponownie. Węzeł, który zwiększa licznik przed przerwą, zrobi to dwa razy za każdą akceptację. Upewnij się, że wszystko przed interrupt() jest idempotentne, albo przenieś efekty uboczne do wcześniejszego węzła. Ten sam rozsądek dotyczy wywołań API lub zapisów do bazy danych umieszczonych przed momentem wstrzymania.

Model, który sam siebie akceptuje, nie stanowi systemu z udziałem człowieka

Niezależnie od wybranego mechanizmu, zadawanie modelowi pytania „Czy powinienem kontynuować?” i ufanie odpowiedzi nie stanowi nadzoru ludzkiego, bez względu na to, jak jest to określane. Jest to agent potwierdzający własną decyzję. Rzeczywisty krok zatwierdzenia przekazuje kontrolę osobie spoza struktury i czeka na jej odpowiedź.

Sześć podstawowych elementów w pigułce

Poniższy streszczenie łączy każdą koncepcję z jej funkcją oraz typowym błędem, który się z nią wiąże.

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Rozsądna kolejność budowy

Dla pierwszego prawdziwego grafu sprawdź, czy pełny pętla działa poprawnie od początku do końca przy użyciu InMemorySaver bez przerwań. Zachowaj mały stan, ograniczony do tego, co potrzebują węzły, i upewnij się, że każdy warunkowy krawędź zwraca dokładnie te etykiety, których oczekuje jego mapowanie. Dopiero gdy wszystko będzie działać bez problemów, warto zastąpić go trwałym wskaźnikiem stanu i dodać przerwanie w tym jedynym kroku, który rzeczywiście wymaga interwencji człowieka – zazwyczaj w przypadku operacji przekazywania pieniędzy, wysyłania e-maili na zewnątrz lub usuwania danych.

Bardziej zaawansowane funkcje, w tym niestandardowe reduktory oprócz add_messages, podgrafy dzielące duży graf na części nadające się do testowania oraz przepływ danych na poziomie tokenów, opierają się na tym samym fundamentzie. Są one znacznie łatwiejsze do wdrożenia po tym, jak stworzysz, uszkodzisz i naprawisz graf, korzystając wyłącznie z tych sześciu koncepcji.

Główne wnioski

• Używaj add_messages do historii czatu, a operator.add tylko dla zwykłych list, i utrzymuj lżeki stan, ponieważ jest on zapisywany na każdym kroku.
• Zwracaj tylko zmienione pola z węzłów; pełny stan przywracany w tle bezgłośnie nadpisuje równoległe lub wcześniejsze aktualizacje.
• Daj każdej ścieżce wyraźną trasę do END, a także zdefiniuj pętle ponawiania prób zamiast polegać na limicie rekurencji.
• Pochodź etykiety trasowania z mapowania, aby nie rozchodziły się one od siebie.
• Traktuj InMemorySaver wyłącznie jako narzędzie do rozwoju; zamiana checkpointer jest tania, więc wykonaj ją przed tym, jak użytkownicy zaczną polegać na grafie.
• Najlepiej używać dynamicznego interrupt() do warunkowych zatwierdzeń i utrzymywać kod przed jego idempotentnością, ponieważ wznowienie uruchamia ponownie węzeł.

Dokumentacja referencyjna: dokumentacja Graph API dla LangGraph, przewodnik dotyczący przerwania oraz interrupt() – referencja API.