LangGraph w Pythonie: stan, węzły równoległe, punkty kontrolne i przegląd przez człowieka

Każdy, kto stworzył coś więcej niż zwykłą zabawkę na bazie modeli LLM, zna trzy powtarzające się problemy. Seria połączeń powoduje wzrost opóźnień. Długi proces może się zawieść, niszcząc cały postęp. A model w najgorszym możliwym momencie może dostarczyć pewnego siebie, ale całkowicie błędnego odpowiedzi.

LangGraph rozwiązuje wszystkie te problemy. Jest to framework orkiestracji oparty na grafach, składający się z kilku prostych elementów: stan reprezentowany jest przez TypedDict, węzły to zwykłe funkcje, a krawędzie stanowią logikę routingu. Te elementy łączą się w procesy, które uruchamiają węzły równolegle, przechowują stan w celu możliwości przywrócenia po awarii oraz pozwalają na przerwę, by osoba mogła sprawdzić sytuację, gdy stawką jest wiele.

Kolejne sekcje omawiają temat krok po kroku. Najpierw pojawia się pojedynczy węzeł, potem wykonywanie równoległe, routowanie warunkowe i pamięć trwała, a następnie przerwy z udziałem człowieka. Na końcu znajduje się kompletny proces pracy z e-mailemi, który łączy wszystkie te elementy. Każdy przykład ma być uruchamiany bez zmian. SDK TypeScript udostępnia niemal te same interfejsy API, więc tam również obowiązuje ten sam model myślowy.

Ustawienia

Najpierw potrzebne są dwa pakiety:

pip install langgraph langchain-openai

Następnie zmienne środowiskowe dla OpenAI oraz, opcjonalnie, śledzenie działania LangSmith:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Przykłady zostały przetestowane z langgraph 1.1.3 (marzec 2026 r.) oraz langchain-openai 1.1.10 i wymagają Pythona 3.10 lub nowszego. Starsze przewodniki czasami używają LANGCHAIN_TRACING_V2 i LANGCHAIN_API_KEY; oba nadal działają, ale LANGSMITH_TRACING i LANGSMITH_API_KEY są obecnie preferowanymi nazwami.

Wszystko w LangGraph opiera się na dwóch zasadach: stan przechowuje dane, a węzły to funkcje, które je przekształcają. Gdy już opanujemy te zasady, reszta frameworku staje się jasna.

Definiowanie stanu za pomocą TypedDict

Stan w LangGraph to zwykły Pythonowy TypedDict – bez specjalnych klas bazowych ani dekoratorów specyficznych dla frameworku, po prostu skatalogowane słowniki:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

Klasa określa strukturę danych przepływających przez graf, a każdy węzeł czyta z tej wspólnej struktury i do niej zapisuje.

Dlaczego TypedDict, a nie zwykły słownik lub klasa dataclass? Dzięki niemu możliwe jest statyczne sprawdzanie typów, ponieważ zarówno mypy, jak i Pyright ją rozumieją, a także umożliwia dodawanie wskazówek typów; ponadto unika się dodatkowego obciążenia podczas wykonywania kodu charakterystycznego dla klasy dataclass lub modelu Pydantic. LangGraph może również analizować adnotacje, aby poznać schemat, co jest ważne przy strukturyzowanym wyjściu i walidacji.

Linią, którą należy zrozumieć, jest Annotated[list[str], operator.add] – określa ona reduktora. Jest to najważniejszy koncept w strukturze stanu LangGraph i prawie zawsze sprawia trudności osobom po raz pierwszy się z nią spotykającym.

Bez reduktora węzeł zwracający {"summary": "new value"} po prostu zastępuje przechowywaną wartość. W przypadku pól skalarnych takich jak summary to właśnie jest pożądane. Problemy pojawiają się, gdy dwa równoległe węzły piszą do messages: zachowana zostaje tylko ta wartość, która zostanie zapisana jako ostatnia. Deklaracja operator.add zamiast tego poleca LangGraph dodawanie wartości, dzięki czemu każdy wpis do messages jest bezpiecznie akumulowany.

Złożona logika łączenia wymaga użycia specjalnego reduktora:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

Ten reduktor łączy istniejącą listę z nowymi wiadomościami, a następnie ogranicza ją do dziesięciu najnowszych elementów, co pomaga w długich rozmowach, w których w przeciwnym razie liczba tokenów gwałtownie by wzrosła. Funkcja operator.add służy do prostego dodawania elementów na końcu listy. Wyborem w przypadku usuwania duplikatów, ograniczania rozmiaru listy lub jakiejkolwiek innej fuzji poza zwykłym dodawaniem jest dostosowany reduktor.

Klasycznym błędem jest zapomnienie o użyciu reduktora w polu, do którego zapisują się węzły równoległe. Bez niego gałęzie w tle nadpisują się nawzajem, zwycięża ta, która skończy pierwsza, i nie pojawia się żaden błąd.

TypedDict versus Pydantic

Oba pojawiają się w LangGraph, a rozdzielenie między nimi jest celowe. TypedDict służy do reprezentacji stanu grafu, czyli kontenera, który przemieszcza się po grafie. LangGraph został zaprojektowany wokół niego – jego pola obsługują składnię reduktora Annotated, umożliwia czystą serializację w celu tworzenia punktów kontrolnych i nie dodaje żadnego obciążenia podczas wykonywania.

Pydantic służy do tworzenia schematów używanych przez modele językowe. Wywołanie .with_structured_output(SomeModel) dostarcza modelowi JSON schematu, według którego powinien generować odpowiedź, oraz weryfikuje, czy odpowiedź jest zgodna z tym schematem. Odpowiedź, która nie pasuje, natychmiast wywołuje błąd i nie pozwala na przesyłanie nieprawidłowych danych przez graf.

Proces pracy w końcowej fazie dokładnie follows tę podział: EmailState to TypedDict pełniący rolę kontenera grafu, natomiast EmailClassification to model Pydantic opisujący wynik generowany przez model językowy.

Tworzenie pierwszego węzła

Każdy węzeł przyjmuje pełny stan jako dane wejściowe i zwraca słownik z częściową aktualizacją. LangGraph łączy tę aktualizację ze stanem i stosuje wszystkie dekrementatory, które zostały zadeklarowane:

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Przeanalizujmy kod:

• StateGraph(ChatState) tworzy graf, który wykorzystuje ChatState jako swoją strukturę, a LangGraph czyta adnotacje, aby poznać reduktory i typy pól.
• graph.add_node("chatbot", chatbot) rejestruje funkcję pod nazwą "chatbot", do której odnoszą się definicje krawędzi.
• graph.add_edge(START, "chatbot") sprawia, że "chatbot" staje się pierwszym węzłem do uruchomienia. START to wbudowany sygnał początkowy.
• graph.add_edge("chatbot", END) kończy wykonywanie po zakończeniu pracy chatbot. END to kolejny wbudowany sygnał końcowy.
• graph.compile() weryfikuje strukturę, sprawdzając istnienie niepowiązanych krawędzi i niedostępnych węzłów, oraz zwraca gotowy do uruchomienia CompiledGraph.
• app.invoke({"messages": [...]}) uruchamia graf synchronicznie i zwraca końcowy stan.

Węzeł umieszcza swoją ciągłą wartość w liście, ponieważ messages używa operator.add jako reduktora. Zwrócenie {"messages": [response.content]} dodaje listę zawierającą jeden element do istniejącej listy, natomiast zwrócenie samej ciągłej wartości spowodowałoby błąd w czasie wykonywania, ponieważ operator.add nie może dodać ciągu do listy.

Zasada polega na dopasowaniu tego, co zwraca węzeł, do typu pola, włączając reduktora. Pole typu list[str] z operator.add przyjmuje wartość typu list[str], natomiast zwykłe pole typu str bez reduktora przyjmuje wartość typu str.

To jest kompletna, działająca aplikacja LangGraph składająca się z około 20 linii kodu. Kolejnym pytaniem jest to, co dzieje się wewnątrz funkcji invoke().

Super kroki: jak działa wykonywanie

LangGraph uruchamia węzły w super krokach. Każdy węzeł zaplanowany do tego samego kroku jest wykonywany równocześnie, a wszystkie one muszą zostać zakończone, zanim rozpocznie się następny krok. Przykład jednokrotnego węzła obejmuje jeden super krok zawierający jeden węzeł. Dodanie krawędzi równoległych umieszcza kilka węzłów w tym samym super kroku.

To ma znaczenie z dwóch powodów. Wykonywanie równoległe jest wbudowane w model od samego początku i nie jest czymś dodawanym później. Ponadto dzięki checkpointerowi LangGraph zapisuje kompletne zdjęcie stanu po zakończeniu każdego super kroku. To umożliwia przywrócenie po awarii: jeśli proces zostanie przerwany w połowie, wykonywanie jest kontynuowane od ostatniego ukończonego super kroku.

Gdy już rozumiemy, jak stan przepływa przez pojedynczy węzeł, następnym pytaniem jest to, jak kontrolować kolejność lub wykonywać węzły jednocześnie. To właśnie jest zadaniem krawędzi.

Krawędzie: sekwencyjne, równoległe i warunkowe

Krawędzie kontrolują, w jaki sposób wykonywanie zadań przechodzi między węzłami. LangGraph oferuje trzy rodzaje krawędzi, które łącznie obejmują wszystkie możliwe wzory routingu. Najprostsza forma pojawiła się już w przykładzie chatbota.

Krawędzie sekwencyjne

Najprostszym przypadkiem jest jeden węzeł następujący po drugim.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft jest wykonywany najpierw. Gdy tylko zostanie zakończony, a LangGraph połączy pole draft ze stanem, uruchamia się review_draft, który może odczytać projekt z state["draft"]. Sekwencyjne uruchamianie jest gwarantowane, ponieważ krawędź zapewnia zależność, a nie jest potrzebny żaden kod sekwencjonujący. Krawędzie sekwencyjne pasują wtedy, gdy węzeł B zależy od wyniku węzła A: najpierw generowanie, potem przeglądanie, najpierw pobieranie, potem przetwarzanie, najpierw klasyfikacja, potem kierowanie.

Przypadkowe cykle stanowią częste zagrożenie. Domyślnie LangGraph oczekuje grafu skierowanego acyklicznego, więc dodanie krawędzi od review z powrotem do generate powoduje błąd podczas kompilacji, chyba że graf został wyraźnie ustawiony jako cykliczny. Pętle iteracyjne są zazwyczaj budowane przy użyciu krawędzi warunkowych, które mogą kierować z powrotem do wcześniejszego węzła.

Krawędzie równoległe

Gdy dwa krawędzie wychodzą od jednego węzła, ich cele są przetwarzane jednocześnie w tym samym superkroku. Tutaj niezbędne stają się reduktory:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Dwie krawędzie wychodzące z START planują operacje research i outline w tym samym superkroku, a LangGraph wykonywa je jednocześnie. Obie mają również krawędź prowadzącą do combine, a LangGraph nie zaplanuje tej operacji, dopóki wszystkie węzły do niej prowadzące nie zakończą pracy, co stanowi automatyczną barierę synchronizacyjną.

findings posiada reduktora, mimo że tutaj do niego zapisuje się tylko research, ponieważ w rzeczywistym systemie mogłyby istnieć cztery równoległe agenty badawcze dodające elementy do tej samej listy, a bez reduktora przetrwałby tylko jeden wynik.

Matematyka opóźnień jest prosta. Jeśli research zajmuje 2 sekundy, a outline – 3, wersja sekwencyjna potrzebuje 5 sekund, natomiast wersja równoległa – 3 sekundy, co odpowiada czasowi trwania dłuższego z zadań. W przypadku niezależnych wywołań LLM, wykonywanie zadań równolegle daje niemal całkowite oszczędności czasowe.

Kandydatami do paralelizacji są wszystkie pary węzłów, które nie zależą od wyników drugiego z nich. Nawet przeniesienie dwóch wywołań trwających po sekundę do tego samego superkroku ma znaczenie przy tysiącach zapytań. Zwracaj uwagę na węzły, które wydają się niezależne, ale dzielą stan w mniej oczywisty sposób. Jeśli dwa węzły działające równolegle zapisują dane do tego samego pola bez reduktora, występuje sytuacja wyścigu, przy której wartość określa ten, który skończy pierwszy, a LangGraph nie wydaje żadnego ostrzeżenia.

To rozwiązanie oparte na rozgałęzianiu i łączeniu jest podstawą projektu końcowego dotyczącego e-maili, który równolegle przeprowadza wyszukiwanie w bazie wiedzy oraz sprawdzanie w systemie śledzenia błędów przed sformułowaniem odpowiedzi.

Krawędzie warunkowe

Gdy następny węzeł zależy od tego, co wygenerował poprzedni, stosuje się funkcję add_conditional_edges:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

Funkcja route_by_sentiment analizuje stan i zwraca klucz w postaci ciągu znaków, a słownik mapujący przekazany funkcji add_conditional_edges przekształca ten klucz w nazwę węzła. Funkcja routingu, która pozostaje czysta (wejście: stan, wyjście: ciąg znaków, brak efektów ubocznych), jest łatwa do testowania, ponieważ można ją przetestować w formie testów jednostkowych bez udziału modeli językowych.

add_conditional_edges przyjmuje trzy argumenty: węzeł źródłowy, funkcję routingu oraz słownik mapujący klucze w postaci ciągów znaków na nazwy węzłów docelowych. Każdy klucz, który może zwrócić funkcja routingu, musi występować w tym słowniku; w przeciwnym razie LangGraph wywoła błąd podczas wykonywania.

Alternatywa w postaci Command

Gdy węzeł musi zaktualizować stan i wybrać trasę w tym samym kroku, może bezpośrednio zwrócić obiekt Command i pominąć funkcję add_conditional_edges:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Jedna wartość zwracana przez Command zawiera zarówno aktualizację stanu (update=), jak i wybór trasy (goto=), dzięki czemu węzeł zapisuje dane i wybiera następny cel w jednym kroku.

add_conditional_edges jest odpowiednim rozwiązaniem, gdy logika routingu tylko odczytuje stan i nic nowego nie zapisuje, gdy routowanie powinno pozostać oddzielone od logiki węzła, lub gdy kilka węzłów źródłowych korzysta z tej samej funkcji routingu.

Command jest odpowiednim wyborem, gdy węzeł wytwarza nowe dane, które wpływają na jego własne routowanie, gdy pojedynczy przypadek nie uzasadnia osobnej funkcji routingu, lub gdy cel zależy od czegoś obliczonego wewnątrz węzła, a nie znajdującego się już w stanie.

Dla etapu ludzkiej weryfikacji mechanizm email capstone opiera się na funkcji add_conditional_edges, ponieważ routowanie jest prostsze – sprawdza jedynie priorytet i poziom pewności już zapisane w stanie.

Pamięć i mechanizmy checkpointingu

Gdy uporządkowanie i rozgałęzianie jest już pod kontrolą, następnym problemem jest trwałość danych: co się dzieje, gdy proces zostanie uruchomiony ponownie w trakcie rozmowy?

Do tej pory każde uruchomienie zaczynało się od zera. Rozmowy oraz procesy wieloetapowe wymagają stanu, który przetrwa między poszczególnymi wywołaniami. LangGraph radzi sobie z tym za pomocą mechanizmów checkpointingu, a API jest na tyle proste, że przechodzenie od środowiska rozwojowego do produkcji wymaga tylko jednej zmiany.

MemorySaver do celów rozwojowych

MemorySaver przechowuje stan w pamięci, co jest przydatne podczas rozwoju i testowania:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

Przy użyciu checkpointer=memory skompilowany graf przechowuje pełny zrzut stanu po każdym superkroku. Ponowne wywołanie funkcji invoke() z tym samym thread_id sprawia, że LangGraph najpierw ładuje najnowszy punkt kontrolny dla danego wątku. thread_id funkcjonuje jak identyfikator sesji: można go ponownie użyć, aby kontynuować pracę, lub zmienić, aby rozpocząć od zera.

Druga próba wywołania może pominąć wcześniejsze wiadomości, ponieważ punkt kontrolny już je przechowuje. Jeśli Alice wywoła funkcję z wartością {"messages": ["What should I learn first?"]}, LangGraph ładuje punkt kontrolny "user-alice", znajduje przechowywaną listę i dodaje nowy wpis za pomocą reduktora operator.add. Gdy chatbot już działa, state["messages"] zawiera całą historię rozmowy.

Używanie ponownie tego samego thread_id dla niepowiązanych rozmów to błąd. Jeśli "user-alice" zostanie ponownie użyty dla innego tematu, chatbot będzie miał dostęp do całej wcześniejszej rozmowy Alice, więc każda nowa sesja powinna otrzymać nowy thread_id. Podczas rozwoju app.get_state_history(config) pokazuje pełną historię punktów kontrolnych dla danej nitki rozmowy, a ten sam mechanizm służy do debugowania z użyciem funkcji podróży w czasie w LangSmith.

Trwałe narzędzia do przechowywania stanu w produkcji

MemorySaver znajduje się w pamięci procesu, więc gdy proces się zakończy, stan również znika. W produkcji potrzebny jest trwały backend:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() tworzy wymagane tabele bazy danych podczas pierwszego uruchomienia i dalej nie wykonuje żadnych działań, więc należy je wywołać tylko raz podczas inicjalizacji. API pozostaje niezmienne pomiędzy różnymi backendami, ponieważ mechanizmy checkpointingu działają za wspólną abstrakcją. MemorySaver, PostgresSaver oraz RedisSaver stosują ten sam protokół, więc kod grafu nie wie, z którym z nich komunikuje się.

Wybór spośród nich:

• MemorySaver nadaje się do rozwoju, testowania i demonstracji, gdzie istotne nie jest przetrwanie po restartowaniu.
• PostgresSaver pasuje do procesów produkcyjnych, które wymagają trwałości danych oraz możliwości przeglądania historii checkpointów.
• RedisSaver z pakietu langgraph-redis nadaje się do zastosowań o wysokiej przepustowości, które wymagają szybkiego odczytu checkpointów i są gotowe pogodzić się z ograniczeniami związanymi z trwałością danych w Redis.

Co jest zapisywane

Pełny zrzut stanu jest przechowywany przy każdym superkroku. To umożliwia podróż w czasie: stan w dowolnym momencie historii może zostać zbadany, odtworzony z wybranego punktu kontrolnego lub sprawdzony pod kątem błędów, aby ustalić dokładnie, gdzie doszło do problemu. Ten sam mechanizm leży u podstaw przerwań. Gdy węzeł wywołuje interrupt(), LangGraph potrzebuje miejsca na przechowywanie zatrzymanego stanu, aby móc go później wznowić. Bez punktu kontrolnego nic nie może tego utrzymać, a struktura przerwań się rozpada.

Człowiek w procesie z przerwami

Część decyzji jest zbyt ważna, by ją zautomatyzować. Zanim agent wyśle e-mail do klienta, zatwierdzi zwrot pieniędzy lub wdroży kod, osoba powinna to przejrzeć i zatwierdzić. LangGraph zarządza tym za pomocą funkcji interrupt().

Funkcja interrupt()

Umieszczenie interrupt() wewnątrz węzła zatrzymuje wykonywanie, zapisuje stan do checkpointera i zwraca ładunek do wywołującego:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

W kolejności, wywołanie wykonuje cztery czynności:

1. Zatrzymuje wykonywanie dokładnie w wierszu, w którym zostało wywołane.
2. Zapisuje pełny stan do checkpointera, dlatego jest on konieczny.
3. Zwraca przekazany payload, czyli tekst w wersji roboczej, do osoby, która wywołała invoke().
4. Czeka, pozostawiając graf w stanie zawieszenia.

Zmienna decision początkowo nic nie otrzymuje, ponieważ interrupt() nie zwraca wartości, dopóki graf nie zostanie wznowiony za pomocą Command(resume=value). W tym momencie decision otrzymuje to, co zostało przekazane do resume=.

Payload informuje ludzkiego recenzenta o tym, co ogląda, więc może zawierać wszystko, czego potrzebuje: wersję roboczą, streszczenie lub ustrukturyzowany słownik z kontekstem. Osoba, która wywołała funkcję, otrzymuje go jako część powiadomienia o przerwaniu.

Wznowienie działania za pomocą Command

Kontynuowanie po przerwie oznacza ponowne wywołanie grafu za pomocą Command(resume=value):

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

W pierwszym kroku LangGraph uruchamia funkcję generate, przechowuje wersję roboczą w stanie i dociera do funkcji interrupt() w węźle review. Tworzy tam punkt kontrolny stanu i wraca, a result z tego pierwszego wywołania invoke() zawiera dane przerwania oraz wersję roboczą przeznaczoną dla osoby dokonującej przeglądu. W drugim kroku funkcja Command(resume="approve") sprawia, że LangGraph ładuje punkt kontrolny o nazwie "email-review-1", kontynuuje działanie od wywołania interrupt() i zwraca wartość "approve" jako wynik. Wartość decision węzła review staje się "approve", ustawia się approved=True, a dalszy proces przechodzi do kroku send.

Do obu wywołań musi być użyty ten sam config, ponieważ LangGraph znajduje odpowiedni punkt kontrolny za pomocą thread_id. Inny thread_id przy drugim wywołaniu wskazałby na punkt kontrolny, który nie istnieje, co spowoduje niepowodzenie wywołania.

Punkt kontrolny jest obowiązkowy przy przerwaniach. Bez niego stan wstrzymany nie ma dokąd przejść, dlatego app.compile(checkpointer=memory) nie jest tu opcjonalne.

Od wersji LangGraph 1.1.0 funkcja invoke() przyjmuje parametr version="v2" i zwraca obiekt GraphOutput zawierający atrybuty .value oraz .interrupts, co umożliwia bezpieczniejsze obsługiwanie przerwań – warto to wypróbować po opanowaniu podstaw. Treść przerwania może być dowolną wartością serializowalną, czy to łańcuchem, słownikiem, czy listą. W środowiskach produkcyjnych zazwyczaj przekazywany jest ustrukturyzowany słownik zawierający wszystkie niezbędne informacje, takie jak wersja projektu, kategoria, adresat oraz URL recenzji, które warstwa aplikacyjna może wyświetlić w interfejsie lub przesłać przez Slacka lub e-mail.

Istnieje jeszcze jedna pułapka. Ponowne wywołanie invoke() z pełnym początkowym stanem zamiast Command(resume=value) jest błędem: app.invoke({"messages": [...]}, config=config) na tym samym thread_id restartuje graf dla tego wątku i usuwa stan wstrzymany. Przerwany graf powinien być kontynuowany wyłącznie za pomocą Command(resume=value).

Kluczowy element: proces triażu e-maili

Gdy wszystkie elementy są już gotowe, można je połączyć w system triażu e-maili w stylu produkcyjnym. System ten odczytuje przychodzący e-mail, klasyfikuje jego cel za pomocą ustrukturyzowanego wyniku, jednocześnie wyszukuje informacje w bazie wiedzy i trackerze błędów, tworzy projekt odpowiedzi i może zatrzymać się w celu przeglądu przez człowieka przed wysłaniem.

Każda z wcześniejszych koncepcji odpowiada określonej części tego procesu:

• Stan TypedDict z reduktorami jest reprezentowany jako EmailState, z operator.add na search_results.
• Krawędzie serialne łączą read_email z classify_intent.
• Krawędzie równoległe umożliwiają classify_intent na rozgałęzienie się do search_docs i bug_tracking.
• Synchronizacja typu fan-in oznacza, że oba węzły równoległe dostarczają dane do write_response.
• Ukierunkowanie warunkowe to route_by_priority, które decyduje między przeglądem przez człowieka a automatycznym wysłaniem.
• Strukturyzowany wyjście w formacie Pydantic to EmailClassification, który określa ramy odpowiedzi LLM.
• Przerwanie procesu to human_review, które wstrzymuje działanie w oczekiwaniu na zatwierdzenie.
• Checkpointer to MemorySaver, który umożliwia cykl przerwania i wznowienia działania.

Stan i schematy

Najpierw przychodzi stan oraz schemat Pydantic do strukturalnej klasyfikacji:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification to model Pydantic, a nie część EmailState, który stanowi przykład rozdzielenia typów między TypedDict a Pydantic. Jest to struktura przekazywana modelowi LLM jako cel schematu JSON, a Pydantic weryfikuje odpowiedź i natychmiast wywołuje błąd, jeśli nie jest zgodna. EmailState to kontener grafu, reprezentowany jako TypedDict, ponieważ LangGraph wymaga składni reduktora oraz efektywnej serializacji.

classification jest przechowywany jako dict w EmailState, ponieważ po weryfikacji odpowiedzi przez Pydantic, funkcja .model_dump() przekształca ją w zwykły słownik przed umieszczeniem w stanie. Dzięki temu stan pozostaje możliwy do serializacji przez dowolny checkpointer, bez konieczności używania Pydantic podczas przywracania danych.

Najważniejszym polem jest search_results: Annotated[list[str], operator.add]. Ponieważ search_docs i bug_tracking działają równolegle i oba mogą zapisywać wyniki, reducer łączy ich wyjścia razem, podczas gdy bez niego jedna gałąź po cichu nadpisałaby drugą.

Budowanie węzłów

Węzły pozostają małe i służą do jednego celu. Każdy z nich pobiera z stanu tylko te wartości, których potrzebuje, wykonuje swoją funkcję i zwraca częściową aktualizację:

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

Na pierwszy rzut oka read_email wydaje się bezużyteczny, ponieważ zwraca jedynie pola już obecne w stanie, ale pełni rolę wyraźnego kroku parsowania. Wersja produkcyjna przyjmowałaby surowy payload webhooka i wyodrębniała z niego pola, więc ta funkcja zastępuje tamto wyodrębnianie.

classify_intent to miejsce, gdzie Pydantic okazuje się przydatny. W structured_llm = llm.with_structured_output(EmailClassification) model LLM jest otoczony tak, aby odpowiedzi musiały odpowiadać schematowi. Otrzymuje on schemat JSON wywiedziony z EmailClassification i musi zwracać odpowiadający mu ważny JSON; LangChain automatycznie próbuje ponownie, zanim wywoła błąd, jeśli tego nie zrobi. Następnie .model_dump() przekształca zweryfikowany obiekt na zwykły słownik przeznaczony do przechowywania w stanie.

Dwa węzły równoległe zapisują dane do oddzielnych pól, search_results i bug_info, dzięki czemu ich wywołania LLM odbywają się jednocześnie. W przypadku search_results funkcja search_docs zwraca listę ([f"[Docs] {response.content}"]), ponieważ operator.add wymaga danych we formie listy. Funkcja bug_tracking zwraca zwykły ciąg znaków dla bug_info, który nie posiada funkcji redukcyjnej, więc wystarczy zwykłe przypisanie.

Węzłem o strukturze fan-in jest write_response. Gdy on rozpoczyna działanie, oba węzły równoległe już przetworzyły dane, a ich wyniki znajdują się w stanie, więc ten węzeł buduje prompt na podstawie wszystkich tych informacji i przygotowuje odpowiedź.

human_review przekazuje do interrupt() ustrukturyzowany słownik, a nie zwykłą ciągłą literę, co stanowi dobrą praktykę w środowiskach produkcyjnych. Słownik zawiera wersję roboczą, kategorię klasyfikacji oraz wiadomość skierowaną do osoby dokonującej oceny, dzięki czemu każdy system po drugiej stronie – czy to interfejs internetowy, bot Slacka czy powiadomienie e-mail – otrzymuje pełny kontekst.

send_reply wykorzystuje state.get("approved", True) jako bezpieczny standard domyślny; jeśli wartość approved w jakikolwiek sposób nie występuje w stanie, funkcja i tak wysyła wiadomość. Ten domyślny wartość może zostać zmieniona na False, gdy profil ryzyka wymaga większej ostrożności.

W środowisku produkcyjnym search_docs i bug_tracking łączyłyby się z rzeczywistymi systemami, takimi jak baza danych wektorowych, Jira czy GitHub Issues, a dekorator LangChain @tool ułatwia realizację tych wywołań. Struktura węzłów pozostaje niezmienna – wywołanie narzędzia zastępuje wywołanie LLM. Jeszcze jedna pułapka: write_response odczytuje state["search_results"], więc jeśli jest to nadal pusta lista, dostęp może się nie powieść; unika się tego poprzez inicjalizację search_results: [] w początkowym pakiecie wywołania, dzięki czemu reducer może od razu do niej dodawać dane.

Łączenie krawędzi

Krawędzie sekwencyjne, równoległe i warunkowe łączą wtedy węzły:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority wysyła e-maile o wysokim priorytecie do ludzkiej oceny, niezależnie od poziomu pewności. Klasyfikacje o poziomie pewności poniżej 0,8 również trafiają do oceny, ponieważ jeśli model nie jest pewien tematu e-mailem, decyzję powinien podjąć człowiek. Wszystko inne jest wysyłane automatycznie, co stanowi rozsądny standard w produkcji, który należy dostosować pod kątem rzeczywistych wskaźników błędów.

To kierowanie przepływem odbywa się za pośrednictwem add_conditional_edges, a nie Command, ponieważ route_by_priority jest funkcją czystą: sprawdza jedynie stan i zwraca klucz, nie tworząc nowych danych, co dokładnie odpowiada przeznaczeniu add_conditional_edges.

Oto wszystkie połączenia w grafie:

• Sekwencyjna eksploatacja przebiega od START przez read_email do classify_intent.
• Rozgałęzienie od classify_intent uruchamia jednocześnie search_docs oraz bug_tracking.
• Rozgałęzienie przychodzące przekazuje wyniki tych dwóch węzłów do write_response.
• Po write_response warunkowe połączenie wybiera albo human_review, albo send_reply.
• Gdy przegląd zostanie zakończony, human_review przekazuje dane do send_reply, które kończy się w punkcie END.

Uruchamianie całego procesu od początku do końca

Najpierw wpływa przykładowy e-mail, a pierwsze uruchomienie trwa aż do przerwania:

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

Do tego czasu graf wykonał już wiele zadań:

1. read_email przetworzył wiadomość.
2. classify_intent oznaczył ją jako raport o błędzie o wysokim priorytecie przy użyciu strukturyzowanego wyniku z Pydantic.
3. Poszukiwanie w dokumentacji oraz sprawdzenie w systemie trackingu błędów odbyły się równolegle w ramach jednego kroku.
4. write_response przygotował odpowiedź, wykorzystując cały ten kontekst.
5. Węzeł human_review wywołał interrupt() i wstrzymał proces.

Cały proces przetwarzania, włączając oba równoległe wywołania modeli językowych, zakończył się przed wstrzymaniem. Gdy graf się zatrzymuje, stan jest w pełni zapisany, więc recenzent może odpowiedzieć po kilku minutach lub godzinach bez utraty żadnych informacji.

Następnie recenzent daje zgodę:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph załadował punkt kontrolny dla "email-001", wznowił proces human_review po wywołaniu interrupt(), przekazał wartość "approve" jako wynik funkcji, ustawił approved=True w stanie i przełączył się na funkcję send_reply, która zwróciła status potwierdzenia.

Każda z wcześniejszych koncepcji pojawia się w ostatecznym rozwiązaniu: stan typu TypedDict wspierany reduktorem, krawędzie sekwencyjne i równoległe, warunkowe kierowanie ruchem, strukturyzowany wyjście w formacie Pydantic, przegląd przez człowieka poprzez przerwanie działania oraz pamięć przechowywana przez punkt kontrolny.

Dla e-maila o niskim priorytecie, ale o wysokim poziomie pewności, funkcja route_by_priority zwraca wartość "auto_send", a etap human_review jest pomijany. Proces przeprowadza się bezpośrednio: czytanie, klasyfikacja, wyszukiwanie równolegle, tworzenie projektu wiadomości i jej wysyłka – przerwanie występuje jedynie na ścieżce, która tego wymaga. To właśnie jest zaleta warunkowego routingu.

Ostatnia uwaga dotyczy stanu początkowego. Struktura TypedDict nie ustala domyślnych wartości dla brakujących pól, w odróżnieniu od Pydantic. Jeśli funkcja write_response spróbuje uzyskać dostęp do state["search_results"], a klucz ten nie znajdzie się w stanie początkowym, pojawi się błąd KeyError. Każde pole musi być obecne w słowniku stanu początkowego, nawet jeśli początkowo ma postać pustej strony, pustej listy lub wartości False.

Dalej co robić

Wybrana tutaj ścieżka zaczyna się od pojedynczego węzła i obiektu TypedDict, a kończy się procesem w stylu produkcyjnym, który umożliwia jednoczesne wykonywanie zapytań do modeli językowych, weryfikację strukturyzowanych wyników, warunkowe rozgałęzianie ścieżek, przetrwanie awarii dzięki punktom kontrolnym oraz oczekiwanie na recenzenta. Stamtąd istnieje kilka możliwych kierunków rozwoju:

• Platforma LangGraph zajmuje się wdrażaniem, zarządzaniem punktami kontrolnymi oraz skalowaniem poziomym, co eliminuje obciążenie związane z obsługą infrastruktury, gdy celem jest obsługa rzeczywistego ruchu.
• LangSmith śledzi i debuguje każde wykonywanie grafu, pokazując dokładny stan na każdym etapie w przypadku nieprawidłowego zachowania węzła. Jest to szczególnie ważne, gdy grafy składają się z pięciu lub więcej węzłów z warunkowymi ścieżkami.
• Podgrafy łączą mniejsze grafy w większe systemy wieloagentowe. Każdy z nich zarządza własnym stanem i komunikuje się za pośrednictwem grafu nadrzędnego, co stanowi naturalny krok po procesach opartych na pojedynczym grafie.
• Narzędzie wywoływane przez LangChain za pomocą @tool i ToolNode umożliwia agentom korzystanie z zewnętrznych API, baz danych i usług bezpośrednio w ramach grafu, a kluczowe węzły search_docs i bug_tracking to oczywiste miejsca do ich wypróbowania.
• Trwałość w środowisku produkcyjnym wymaga zainstalowania langgraph-checkpoint-postgres w celu zachowania trwałego stanu po ponownym uruchomieniu, a wymiana API odbywa się w jednej linii kodu.

Szybki podsumowanie

• Stan jest reprezentowany jako TypedDict. Każde pole typu Annotated[list, operator.add] gromadzi wartości z różnych zapisów, natomiast pole bez funkcji redukcyjnej jest zastępowane przy każdym zapisie.
• Węzeł to po prostu funkcja: otrzymuje pełny stan i zwraca częściową tablicę aktualizacji, przy czym LangGraph zajmuje się ich połączeniem.
• Krawędzie określają przepływ. Używaj krawędzi sekwencyjnych, gdy ważny jest porządek, krawędzi równoległych do obsługi niezależnych zadań wykonywanych jednocześnie, oraz krawędzi warunkowych, gdy dane powinny wybrać odpowiednią ścieżkę.
• Do reprezentacji stanu grafu użyj TypedDict, a do określenia tego, co zwraca model LLM, zastosuj Pydantic z metodą .with_structured_output().
• Mechanizmy checkpointingu zapewniają trwałość danych i umożliwiają przerwy: MemorySaver podczas rozwoju, PostgresSaver w środowisku produkcyjnym. Bez checkpointingu nie ma możliwości przerwania.
• Przerwa to nic innego jak zapisany, wstrzymany stan. Funkcja interrupt(payload) zatrzymuje proces i tworzy punkt checkpointu, natomiast Command(resume=value) wznowia działanie dokładnie w tym miejscu.
• Równoległość jest wbudowana: dwie krawędzie wychodzące z jednego węzła wykonują swoje zadania jednocześnie, a każde pole zapisywane przez te węzły wymaga reduktora.

Pytania do przemyślenia

1. MemorySaver nadaje się do środowiska rozwojowego, a PostgresSaver do środowiska produkcyjnego. Które inne rozwiązania do przechowywania zasługują na uwagę w systemach o wysokiej przepustowości i jakie kompromisy wpłyną na wybór?
2. route_by_priority wysyła e-maile o wysokim priorytecie do sprawdzenia, a resztę wysyła automatycznie. Jak można dostosować tę granicę w rzeczywistym systemie i jakie wskaźniki pokażą, czy została prawidłowo ustawiona?
3. Ostateczna ścieżka routingu następuje po write_response za pomocą add_conditional_edges. Czy istnieje jakiś scenariusz w procesie obsługi e-maili, w którym wzorzec Command pasowałby lepiej niż oddzielna funkcja routingu?

Literatura pokrewna

• Praktyczne notatki: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph — Szczegółowy przewodnik po praktycznych notatkach: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph: umowy, sprawdzenia oraz gotowe fragmenty kodu dla zespołów wdrażających ten model.
• Budowanie agenta do badania akcji z ReAct i LangGraph — Agent oparty na LangGraph i ReAct przyjmuje symbol giełdowy, wywołuje dane fundamentalne, wiadomości, analizy techniczne oraz narzędzia analityczne w ustalonej kolejności i tworzy krótki raport, z pauzami umożliwiającymi interwencję człowieka.

Praca z lokalnym agentem LLM zbudowanym z folderów i plików Markdown — Stwórz zabezpieczony, wieloetapowy pipeline lokalnego agenta, w którym każdy etap to folder, prompty są edytowalnymi plikami Markdown, a człowiek sprawdza wyniki przed kosztownym krokiem.

Notatki praktyczne: Agent Text-to-SQL w Pythonie: Przewodnik po wywoływaniu narzędzi LLM — Szczegółowy przewodnik po notatkach praktycznych: Agent Text-to-SQL w Pythonie: Przewodnik po wywoływaniu narzędzi LLM: umowy, sprawdzenia oraz gotowe fragmenty kodu dla zespołów wdrażających ten wzorzec.

Notatki praktyczne: Dlaczego nikt nie mówi o LangChain, LangGraph czy AutoAgent — Szczegółowy przewodnik po Notatkach praktycznych: Dlaczego nikt nie mówi o LangChain, LangGraph czy AutoAgent: kontrakty, sprawdzania oraz gotowe elementy kodu dla zespołów wdrażających ten wzorzec.

Zespół wielu agentów zdolny do samoregeneracji w LangGraph: architekt, tester, debugger — Stwórz maszynę stanową w LangGraph w Pythonie, w której architekt, programista, środowisko testowe pytest, debugger oraz recenzent współpracują i powtarzają proces, aż testy się udają.

Notatki praktyczne: Budowa pętli przeglądu kodu z udziałem dwóch agentów przy użyciu A2A, LangGraph i — Szczegółowy przewodnik po notatkach praktycznych dotyczących budowy pętli przeglądu kodu z udziałem dwóch agentów przy użyciu A2A, LangGraph i: umów, sprawdzeń oraz miejsc na kod do wstawienia dla zespołów wdrażających ten wzorzec.