LangGraph w Pythonie: stan, węzły i krawędzie od zera

Zwykły program w Pythonie to ciąg wywołań funkcji. Gdy przepływ się komplikuje (są gałęzie, powroty, równoległe kroki i miejsca, w których trzeba zatrzymać się na decyzję człowieka), taki ciąg robi się trudny do czytania i jeszcze trudniejszy do wznowienia po przerwaniu. LangGraph rozwiązuje to, zmuszając Cię do opisania programu jako grafu stanowego: co się dzieje w poszczególnych krokach i jak kroki są ze sobą połączone.

To tekst otwierający cykl o pseudonimizacji danych w LangGraph. Jeśli pojęcia węzła, krawędzi i cyklu są dla Ciebie nowe, zacznij od tekstu o tym, czym jest graf w programowaniu. Zbudujemy tu mały graf, który znajduje w tekście adresy e-mail i numery telefonów i je maskuje. Bez modelu językowego, bo chodzi o samą mechanikę grafu. Ten przykład rozbudujemy w kolejnych częściach o wykrywanie PESEL, zapis stanu, zatrzymanie na akceptację operatora i szyfrowanie mapowania. Jeśli wcześniej budowałeś pętlę agenta w Pythonie, zobaczysz, że graf przejmuje rolę, którą tam pełniła ręcznie napisana pętla while.

Kod uruchomiłem na Pythonie 3.13 i LangGraph 1.2.18. Instalacja:

pip install langgraph

Czym jest graf stanowy i z jakich elementów się składa?

Graf w LangGraph ma cztery składniki:

Do tego dochodzą dwa punkty specjalne: START (wejście do grafu) i END (zakończenie).

Jak zdefiniować stan grafu?

Stan opisujesz jako TypedDict. To zwykły słownik z podpowiedziami typów, więc edytor i mypy wiedzą, jakie klucze istnieją:

import re
from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
PHONE = re.compile(r"(?<!\d)(?:\+48[ -]?)?\d{3}[ -]?\d{3}[ -]?\d{3}(?!\d)")


class State(TypedDict):
    text: str
    hits: Annotated[list[str], add]
    count: int
    result: str

Zwróć uwagę na Annotated[list[str], add] przy kluczu hits. To reducer, czyli reguła łączenia wartości. Wyjaśniam go poniżej, bo bez niego graf z tego artykułu by się nie uruchomił.

Jak napisać węzły?

Węzeł to zwykła funkcja: dostaje cały stan i zwraca słownik tylko z tymi kluczami, które chce zmienić. Resztą stanu LangGraph zajmuje się sam:

def normalize(state: State) -> dict:
    return {"text": " ".join(state["text"].split())}


def find_emails(state: State) -> dict:
    return {"hits": EMAIL.findall(state["text"])}


def find_phones(state: State) -> dict:
    return {"hits": PHONE.findall(state["text"])}


def summarize(state: State) -> dict:
    return {"count": len(state["hits"])}


def mask(state: State) -> dict:
    text = EMAIL.sub("[EMAIL]", state["text"])
    return {"result": PHONE.sub("[TELEFON]", text)}


def keep(state: State) -> dict:
    return {"result": state["text"]}

Dzięki temu każdy węzeł da się przetestować osobno, zwykłym wywołaniem funkcji, bez uruchamiania całego grafu. Wyrażenia regularne w find_emails i find_phones rozbierzemy dokładniej w następnej części cyklu.

Jak połączyć węzły krawędziami?

Graf budujesz w StateGraph: najpierw rejestrujesz węzły, potem krawędzie, na końcu wywołujesz compile():

def choose_next(state: State) -> str:
    return "mask" if state["count"] else "keep"


builder = StateGraph(State)
builder.add_node("normalize", normalize)
builder.add_node("find_emails", find_emails)
builder.add_node("find_phones", find_phones)
builder.add_node("summarize", summarize)
builder.add_node("mask", mask)
builder.add_node("keep", keep)

builder.add_edge(START, "normalize")
builder.add_edge("normalize", "find_emails")
builder.add_edge("normalize", "find_phones")
builder.add_edge(["find_emails", "find_phones"], "summarize")
builder.add_conditional_edges("summarize", choose_next, ["mask", "keep"])
builder.add_edge("mask", END)
builder.add_edge("keep", END)

graph = builder.compile()

Trzy rodzaje krawędzi, które tu widać:

• Zwykła (add_edge("a", "b")): po a zawsze wykona się b.
• Rozgałęzienie: dwie krawędzie wychodzące z normalize powodują, że find_emails i find_phones wykonają się w tym samym kroku, równolegle.
• Łącząca (add_edge([...], "summarize")): summarize czeka na oba węzły z listy. Zapis z listą w pierwszym argumencie oznacza „po wszystkich naraz”, a nie „po każdym z osobna”.
• Warunkowa (add_conditional_edges): funkcja choose_next dostaje stan i zwraca nazwę następnego węzła. Trzeci argument to lista dozwolonych celów, dzięki której LangGraph potrafi narysować diagram.

Do czego służy reducer i dlaczego bez niego graf się wywala?

Dwa węzły (find_emails i find_phones) wykonują się w tym samym kroku i oba zapisują klucz hits. LangGraph musi wiedzieć, co zrobić z dwiema wartościami. Jeśli klucz nie ma reducera, nie zgaduje i zgłasza błąd. Sprawdziłem to na minimalnym grafie, w którym dwa węzły równolegle zapisują hits bez Annotated:

InvalidUpdateError: At key 'hits': Can receive only one value per step. Use an Annotated key to handle multiple values.

Błąd mówi wprost, co zrobić: oznacz klucz jako Annotated z funkcją łączącą. Funkcja operator.add dla list robi konkatenację, więc wyniki obu węzłów trafiają do jednej listy. Reducer ma także drugą konsekwencję: działa na każdym zapisie do tego klucza, nie tylko w kroku równoległym. Jeśli węzeł zwróci {"hits": [...]}, nowe elementy zostaną dopisane do istniejących, a nie je zastąpią. Klucze bez reducera (text, count, result) są po prostu nadpisywane.

Jak uruchomić graf i zobaczyć, co się w nim dzieje?

Całość wywołujesz przez invoke(), podając początkowy stan:

result = graph.invoke({"text": "Proszę   o kontakt:\n jan.kowalski@example.com, tel. 600 123 456.",
                     "hits": [], "count": 0, "result": ""})
print(result)

Wyjście:

{'text': 'Proszę o kontakt: jan.kowalski@example.com, tel. 600 123 456.', 'hits': ['jan.kowalski@example.com', '600 123 456'], 'count': 2, 'result': 'Proszę o kontakt: [EMAIL], tel. [TELEFON].'}

Dla tekstu bez danych ("Dzień dobry, mam pytanie o fakturę.") graf przechodzi gałęzią keep: count wynosi 0, a result jest identyczny jak text.

Chcesz zobaczyć krok po kroku, co zmienił każdy węzeł? Użyj stream() w trybie updates:

for step in graph.stream({"text": "Pisz: a@b.pl", "hits": [], "count": 0, "result": ""},
                        stream_mode="updates"):
    print(step)

{'normalize': {'text': 'Pisz: a@b.pl'}}
{'find_emails': {'hits': ['a@b.pl']}}
{'find_phones': {'hits': []}}
{'summarize': {'count': 1}}
{'mask': {'result': 'Pisz: [EMAIL]'}}

Każda linijka to jedna zmiana stanu, z nazwą węzła, który ją wprowadził. To najszybsze narzędzie do debugowania grafu, a w dalszych częściach cyklu to samo stream() pokaże nam moment zatrzymania na akceptację operatora.

Jak narysować graf?

Skompilowany graf potrafi wygenerować diagram w formacie Mermaid:

print(graph.get_graph().draw_mermaid())

W wyniku dostajesz kod diagramu (węzły, krawędzie zwykłe oraz krawędzie warunkowe zaznaczone liniami przerywanymi), który wkleisz do dokumentacji projektu albo do edytora obsługującego Mermaid. Rozbieżność diagramu z tym, co zamierzałeś, bywa pierwszym sygnałem, że pomyliłeś nazwę węzła w krawędzi.

Jak przetestować graf w pytest?

Węzły to zwykłe funkcje, więc testujesz je bezpośrednio, a graf jako całość osobnym testem:

import pytest
from graph import PHONE, graph, mask, normalize


def run(text):
    return graph.invoke({"text": text, "hits": [], "count": 0, "result": ""})


def test_normalize_collapses_whitespace():
    assert normalize({"text": "a \n  b"}) == {"text": "a b"}


def test_mask_podmienia_mail_i_telefon():
    result = mask({"text": "x@y.pl i 600 123 456"})
    assert result == {"result": "[EMAIL] i [TELEFON]"}


def test_graph_collects_hits_from_both_branches():
    result = run("jan@example.com 600123456")
    assert sorted(result["hits"]) == ["600123456", "jan@example.com"]
    assert result["count"] == 2


def test_text_without_data_passes_unchanged():
    result = run("Dzień dobry")
    assert result["result"] == "Dzień dobry"
    assert result["hits"] == []


@pytest.mark.parametrize("text", ["12345678", "1234567890"])
def test_phone_ignores_wrong_length(text):
    assert PHONE.search(text) is None

Test z sorted(...) jest celowy: kolejność wyników z dwóch równoległych węzłów w liście hits nie jest gwarantowana, więc nie opieraj na niej asercji. Podstawy pisania takich testów, w tym używanie atrap, opisujemy w artykule o testowaniu kodu z API LLM w pytest.

Kiedy graf ma sens, a kiedy wystarczy zwykła funkcja?

Nie każdy program potrzebuje LangGraph. Prosta zasada:

Nasz przykład ma gałąź i krok równoległy, więc to granica, od której graf zaczyna się opłacać. Prawdziwą przewagą LangGraph jest ostatni wiersz tabeli: zapis stanu i wznawianie, o czym jest następna część cyklu.

Co dalej?

W kolejnych częściach rozbudujemy ten graf: pokażemy, jak wiarygodnie wykrywać dane wrażliwe wyrażeniami regularnymi (w tym PESEL z sumą kontrolną), jak zastępować je tokenami, jak zapisywać stan grafu w SQLite i wznawiać go po przerwaniu, jak zatrzymać graf na akceptację operatora i jak zaszyfrować mapowanie. Zanim do tego dojdziesz, zobacz ścieżkę nauki Pythona, w której ten materiał jest jednym z modułów.