LangGraph — jak budować produkcyjne workflow agentów AI

LangGraph to biblioteka Pythonowa od LangChain, która pozwala budować agentów AI jako skierowane grafy z zarządzanym stanem — gdzie każdy węzeł to krok logiki, a krawędzie (w tym warunkowe) decydują o dalszym przepływie. To de facto standard budowy produkcyjnych workflow agentowych w 2026 roku, bo rozwiązuje fundamentalny problem prostych pętli agentów: brak kontroli nad stanem, trudne debugowanie i niemożność wznowienia przerwanych zadań. Zamiast while-pętli z LLM w środku dostajesz jawny, sterowalny automat stanów — z możliwością pauzy, wznowienia, równoległych gałęzi i wbudowanego checkpointingu.

Kompletny przewodnik po LangGraph: jak działa StateGraph z węzłami, krawędziami i conditional routing, wzorzec ReAct krok po kroku, human-in-the-loop z interrupt/resume, persistence z checkpointerami PostgreSQL, równoległe gałęzie Send API, streaming i porównanie z CrewAI, AutoGen i Pydantic AI.

Wyobraź sobie agenta, który analizuje raport finansowy: najpierw wyodrębnia dane, potem równolegle sprawdza dwie bazy danych, następnie — jeśli dane są sprzeczne — prosi człowieka o rozstrzygnięcie, a na końcu generuje wynik. Zrealizowanie tego w zwykłej while-pętli to labirynt if-else. W LangGraph to czytelny graf z pięcioma węzłami. Ten artykuł przeprowadzi Cię przez kluczowe koncepcje, wzorce produkcyjne i porównanie z innymi frameworkami.

Co to jest LangGraph i kiedy go używać

LangGraph rozszerza LangChain o graf przepływu sterowania z zarządzanym stanem. Kluczowe pojęcia:

• StateGraph — główna klasa; definiujesz schemat stanu (TypedDict lub Pydantic) i dodajesz węzły
• Węzeł (node) — funkcja Pythona przyjmująca stan i zwracająca jego aktualizację
• Krawędź (edge) — połączenie między węzłami; może być bezwarunkowa lub warunkowa
• Conditional edge — funkcja routingu, która na podstawie stanu wybiera następny węzeł
• START / END — specjalne węzły frameworku; każdy graf zaczyna się w START i kończy w END

LangGraph jest odpowiednim wyborem, gdy: - Agent musi działać wielokrotnie (pętle, iteracje, retry) - Potrzebujesz human-in-the-loop (zatwierdzenia, korekty w trakcie) - Workflow ma rozgałęzienia warunkowe lub równoległe gałęzie - Zadanie może być przerywane i wznawialne (długie operacje, asynchroniczne) - Chcesz debugować i wizualizować przepływ krok po kroku

Kiedy LangGraph to za dużo: dla prostego agenta z jedną pętlą wystarczy czysty Python z LLM i tool calling. LangGraph świeci przy skomplikowanych, wieloetapowych przepływach.

Pierwszy graf w LangGraph — ReAct agent od zera

Wzorzec ReAct (Reasoning + Acting) to najpopularniejszy punkt startowy: agent myśli, decyduje o działaniu, wykonuje narzędzie, obserwuje wynik i idzie do przodu — lub kończy. W LangGraph mapuje to naturalnie na graf.

from typing import TypedDict, Annotated, Sequencefrom langgraph.graph import StateGraph, START, ENDfrom langgraph.prebuilt import ToolNodefrom langchain_anthropic import ChatAnthropicfrom langchain_core.messages import BaseMessage, HumanMessageimport operatorclass AgentState(TypedDict):    messages: Annotated[Sequence[BaseMessage], operator.add]model = ChatAnthropic(model="claude-haiku-4-5-20251001").bind_tools(tools)def agent_node(state: AgentState) -> dict:    response = model.invoke(state["messages"])    return {"messages": [response]}def should_continue(state: AgentState) -> str:    last = state["messages"][-1]    if last.tool_calls:        return "tools"    return ENDgraph = StateGraph(AgentState)graph.add_node("agent", agent_node)graph.add_node("tools", ToolNode(tools))graph.add_edge(START, "agent")graph.add_conditional_edges("agent", should_continue)graph.add_edge("tools", "agent")app = graph.compile()

Ten przykład pokazuje pełny cykl ReAct w 30 liniach. Kluczowy element to funkcja should_continue — warunkowy routing, który decyduje czy iść do węzła narzędzi czy kończyć. ToolNode to prebuilt utility, które automatycznie wywołuje narzędzia z tool calls w wiadomości.

Zarządzanie stanem — fundament LangGraph

Stan w LangGraph to zarządzana struktura z regułami scalania aktualizacji. Adnotacja Annotated[list, operator.add] mówi LangGraph, żeby przy każdej aktualizacji węzła dodawał nowe elementy do istniejącej listy zamiast nadpisywać.

from typing import TypedDict, Annotatedfrom langgraph.graph import StateGraphimport operatorclass WorkflowState(TypedDict):    messages: Annotated[list, operator.add]    current_step: str    tool_results: Annotated[list, operator.add]    final_answer: str | NoneDobre praktyki zarządzania stanem:- Trzymaj stan płaski i prosty — głęboko zagnieżdżone obiekty trudniej debugować- Używaj Pydantic zamiast TypedDict dla walidacji i domyślnych wartości- Każdy węzeł zwraca tylko te klucze, które zmienił — nie trzeba zwracać całego stanu- Przechowuj w stanie kontekst potrzebny do decyzji routingu, nie tylko dane wyjściowe

Persistence — checkpointing i wznawianie

To cecha, która odróżnia LangGraph od zwykłych pętli. Checkpointer zapisuje snapshot stanu po każdym kroku — tak, że możesz wznowić przerwaną sesję, przeglądać historię kroków do debugowania, cofnąć do poprzedniego stanu i zaimplementować human-in-the-loop ze wstrzymaniem.

from langgraph.checkpoint.sqlite import SqliteSaverfrom langgraph.checkpoint.postgres import PostgresSaver# Developmentmemory_saver = SqliteSaver.from_conn_string(":memory:")app = graph.compile(checkpointer=memory_saver)# Productionwith PostgresSaver.from_conn_string("postgresql://user:pass@host/db") as checkpointer:    checkpointer.setup()    app = graph.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "session-abc-123"}}result = app.invoke({"messages": [HumanMessage("Przeanalizuj raport Q1")]}, config)

thread_id to klucz sesji — identyczny ID = ta sama sesja, nowy ID = nowa niezależna sesja. W produkcji zawsze przekazuj thread_id — bez niego każde wywołanie to izolowana sesja bez historii.

Human-in-the-loop — zatrzymanie i zatwierdzenie

Jeden z najpotężniejszych wzorców: agent wstrzymuje się przed wrażliwą operacją i czeka na zatwierdzenie człowieka.

from langgraph.types import interrupt, Commanddef review_before_send(state: WorkflowState) -> dict:    draft_email = state["draft_email"]    approved = interrupt({        "action": "review_email",        "draft": draft_email,        "message": "Zatwierdz lub popraw email przed wyslaniem"    })    if approved["decision"] == "approve":        return {"send_email": True}    return {"send_email": False, "draft_email": approved.get("revised", draft_email)}app = graph.compile(interrupt_before=["review_before_send"])app.invoke(    Command(resume={"decision": "approve"}),    config)

Wzorzec human-in-the-loop jest niezastąpiony w systemach produkcyjnych: agent może przygotować treść emaila, ofertę czy decyzję zakupową — i zatrzymać się przed faktycznym działaniem. Człowiek zatwierdza, poprawia lub odrzuca. Dopiero potem system rusza dalej.

Równoległe gałęzie — Send API i fan-out

LangGraph pozwala uruchamiać węzły równolegle za pomocą Send API. Wzorzec map-reduce: jeden węzeł generuje listę elementów, każdy jest przetwarzany równolegle, wyniki są zbierane.

from langgraph.types import Senddef generate_tasks(state: WorkflowState) -> list:    documents = state["documents"]    return [Send("analyze_doc", {"doc": doc}) for doc in documents]def analyze_doc(state: dict) -> dict:    doc = state["doc"]    analysis = llm.invoke("Przeanalizuj: " + doc["content"])    return {"analyses": [{"doc_id": doc["id"], "result": analysis.content}]}graph.add_conditional_edges("prepare_tasks", generate_tasks)

Zamiast sekwencyjnego przetwarzania 10 dokumentów (10× czas), LangGraph uruchamia je równolegle — łączny czas to najdłuższy pojedynczy dokument. Kluczowy use case: przetwarzanie wsadowe, wiele API calls jednocześnie, równoległa analiza wielu źródeł.

Streaming — wyniki w czasie rzeczywistym

LangGraph obsługuje streaming na wielu poziomach — od tokenów przez aktualizacje stanu po eventy specyficzne dla węzłów.

async for chunk in app.astream_events(    {"messages": [HumanMessage("Przeanalizuj dane")]},    config,    version="v2"):    if chunk["event"] == "on_chat_model_stream":        token = chunk["data"]["chunk"].content        print(token, end="", flush=True)    elif chunk["event"] == "on_chain_end" and chunk["name"] == "agent":        print("\n[Agent skonczyl krok]")async for state_update in app.astream(inputs, config):    node_name = list(state_update.keys())[0]    print("Wezel " + node_name + " zakonczyl")

Streaming to konieczność w UI — użytkownik widzi postęp zamiast czekać kilkadziesiąt sekund na końcowy wynik. W praktyce strumieniujesz przez WebSocket lub Server-Sent Events do frontendu.

Subgrafy i systemy wieloagentowe

LangGraph obsługuje grafy zagnieżdżone — subgrafy jako węzły w głównym grafie. To fundament architektur wieloagentowych: orchestrator i wyspecjalizowani agenci.

from langgraph.graph import StateGraphfinance_graph = StateGraph(FinanceState)finance_graph.add_node("fetch_data", fetch_financial_data)finance_graph.add_node("analyze", analyze_financials)finance_graph.add_edge(START, "fetch_data")finance_graph.add_edge("fetch_data", "analyze")finance_graph.add_edge("analyze", END)finance_agent = finance_graph.compile()main_graph = StateGraph(OrchestratorState)main_graph.add_node("route_task", route_incoming_task)main_graph.add_node("finance_agent", finance_agent)main_graph.add_node("legal_agent", legal_agent)main_graph.add_node("synthesize", synthesize_results)

Każdy agent ma własny stan i logikę wewnętrzną. To podejście skaluje się lepiej niż jeden duży prompt z całą logiką, bo każdy agent jest wyspecjalizowany, testowalny niezależnie i łatwy do zamiany.

LangGraph vs inne frameworki

LangGraph — wybierz, gdy potrzebujesz pełnej kontroli, wieloetapowych workflow z state management, human-in-the-loop i streaming w produkcji. Najbardziej dojrzała opcja produkcyjna w 2026.

CrewAI — wybierz na szybki start z modelem role-based. Dobry dla prostszych pipeline'ów bez złożonego routingu.

AutoGen — wybierz do badań i prototypowania; model konwersacyjny jest intuicyjny, ale trudniejszy do kontrolowania w produkcji.

Pydantic AI — wybierz jeśli już używasz Pydantic i chcesz typowanego API. Szybko rośnie, dobra integracja z Logfire do monitoringu.

Wzorce produkcyjne

Reflection (Self-critique) — Agent generuje odpowiedź, ocenia własną jakość, poprawia lub kończy. Implementacja: węzeł "generate" + węzeł "critique" + conditional edge sprawdzający ocenę.

Plan-and-Execute — Planner tworzy listę kroków, executor przetwarza je sekwencyjnie lub równolegle, synthesizer łączy wyniki. Dobry dla długich zadań wymagających wielu narzędzi.

Orchestrator-Workers — Orchestrator przydziela zadania, workers wykonują równolegle, orchestrator zbiera i decyduje o next step. Odpowiednik fan-out/fan-in.

Routing z klasyfikatorem — Pierwszy węzeł klasyfikuje intencję użytkownika, conditional edge kieruje do wyspecjalizowanego agenta. Zamiast jednego dużego agenta obsługującego wszystko.

Deployment — LangGraph Platform i LangGraph Studio

LangGraph Studio to wizualny debugger: widzisz graf, śledzisz stan kroku po kroku, możesz edytować stan i wznawiać z dowolnego punktu. Nieoceniony przy debugowaniu złożonych workflow.

LangGraph Platform to managed hosting z automatycznym skalowaniem, persistent storage, REST API do interakcji z grafami, webhooks dla asynchronicznych zadań i wbudowanym monitoringiem. Dla self-hosted: LangGraph aplikacja to zwykła aplikacja Pythonowa — deployujesz ją jak FastAPI z własnym PostgreSQL jako checkpointerem.

Typowe błędy i checklist wdrożenia na produkcję

1. 1.Infinite loop bez warunku wyjścia — każda conditional edge powinna mieć ścieżkę do END; dodaj licznik iteracji w stanie
2. 2.Zbyt duży stan — trzymaj tylko to, co potrzebujesz do decyzji; duże dane przechowuj zewnętrznie, w stanie tylko referencję
3. 3.Brak thread_id przy checkpointingu — bez niego każde wywołanie to nowa sesja bez historii
4. 4.Synchroniczny kod w async grafie — jeśli używasz astream, wszystkie węzły muszą być async
5. 5.Wyciek wrażliwych danych przez stan — stan trafia do checkpointera; nie przechowuj tokenów ani PII w czystym tekście
6. 6.Brak obsługi błędów narzędzi — twoja logika routingu musi obsługiwać scenariusz "all tools failed"
7. 7.Zdefiniuj stan jako Pydantic model — walidacja, domyślne wartości, dokumentacja
8. 8.Zawsze dodaj limit iteracji do każdej pętli (pole "iteration_count" w stanie)
9. 9.Podłącz PostgresSaver z dedykowaną tabelą — SQLite tylko w dev
10. 10.Każde wywołanie produkcyjne ma thread_id — identyfikator sesji lub zadania
11. 11.Włącz human-in-the-loop na wrażliwych operacjach (wysyłka, płatność, zapis)
12. 12.Dodaj tracing (LangSmith lub Langfuse) — logujesz każdy krok i widzisz gdzie agent się gubi

Najważniejsze wnioski

LangGraph to de facto standard orkiestracji agentów AI w Pythonie w 2026 roku. Zamienia agenta z prostej while-pętli w jawny, sterowalny automat stanów — z checkpointingiem, pełną historią kroków, wbudowanym human-in-the-loop, równoległymi gałęziami i streamingiem. Kluczowe koncepcje: StateGraph, węzły (funkcje Pythona), conditional edges (routing), checkpointer (persistence) i Send API (równoległość). W produkcji używaj PostgresSaver, zawsze przekazuj thread_id, strumieniuj przez astream_events i dodaj limit iteracji do każdej pętli. LangGraph wygrywa z CrewAI i AutoGen, gdy potrzebujesz pełnej kontroli, złożonego routingu i niezawodności produkcyjnej.

---

Pomagam firmom projektować i wdrażać produkcyjne workflow agentów AI na LangGraph — od architektury grafu i doboru checkpointera, przez human-in-the-loop i równoległe gałęzie, po deployment, tracing z LangSmith i optymalizację kosztów. Napisz do mnie — zaczynam od bezpłatnej 30-minutowej analizy Twojego przypadku.

Powiązane artykuły

• Systemy wieloagentowe
• Agenty AI — jak działają
• MCP: jeden protokół dla narzędzi AI
• Pamięć agenta AI