LangGraph - Graph-Based Agent Workflows

Łańcuch z poprzednich lekcji idzie zawsze w jedną stronę: badacz, analityk, pisarz, koniec. A co, jeśli pisarz musi odesłać pracę do badacza albo agent przed zakupem biletów powinien zapytać człowieka o zgodę? Prawdziwy szlak przez sawannę ma zawrócenia i rozwidlenia.

LangGraph to framework do budowania złożonych, stanowych aplikacji AI opartych na grafach. Pozwala tworzyć cykle, rozgałęzienia i zaawansowane przepływy sterowania dla agentów. Myśl o nim jak o mapie szlaków: węzły to obozy, krawędzie to ścieżki, a stan to plecak, który nosisz od obozu do obozu.

Czym jest LangGraph?

LangGraph rozszerza możliwości LangChain o:

• State Management - zarządzanie stanem między krokami
• Cycles - możliwość tworzenia pętli (w przeciwieństwie do DAG)
• Branching - warunkowe rozgałęzienia przepływu
• Human-in-the-loop - punkty wstrzymania dla interakcji z człowiekiem

Schemat pokazuje typowy graf agenta z narzędziami, w którym przepływ może wracać do agenta wiele razy:

1┌─────────────────────────────────────────────────────────────┐
2│                     LangGraph Flow                          │
3├─────────────────────────────────────────────────────────────┤
4│                                                             │
5│     ┌──────────┐                                            │
6│     │  START   │                                            │
7│     └────┬─────┘                                            │
8│          │                                                   │
9│          ▼                                                   │
10│     ┌──────────┐     ┌──────────────┐                       │
11│     │  Agent   │────▶│   Condition  │                       │
12│     │  Node    │     │   (Router)   │                       │
13│     └──────────┘     └──────┬───────┘                       │
14│          ▲                  │                               │
15│          │        ┌─────────┼─────────┐                     │
16│          │        ▼         ▼         ▼                     │
17│          │   ┌────────┐ ┌────────┐ ┌────────┐              │
18│          │   │ Tool A │ │ Tool B │ │  END   │              │
19│          │   └───┬────┘ └───┬────┘ └────────┘              │
20│          │       │          │                               │
21│          └───────┴──────────┘                               │
22│              (cycle back)                                    │
23│                                                             │
24└─────────────────────────────────────────────────────────────┘

Agent decyduje, router wybiera ścieżkę do narzędzia albo do końca, a po użyciu narzędzia przepływ wraca do agenta.

Instalacja i podstawy

Najważniejszy element LangGraph to stan. Definiujemy go jako TypedDict, czyli słownik z opisanymi typami kluczy:

1# pip install langgraph langchain-openai
2
3from langgraph.graph import StateGraph, END
4from langgraph.checkpoint.memory import MemorySaver
5from typing import TypedDict, Annotated, Literal
6from langchain_openai import ChatOpenAI
7from langchain_core.messages import HumanMessage, AIMessage
8import operator
9
10# Definicja stanu - kluczowy element LangGraph
11class AgentState(TypedDict):
12    messages: Annotated[list, operator.add]  # Lista wiadomości
13    current_step: str
14    iteration_count: int
15
16# Model LLM
17llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Annotated[list, operator.add] mówi LangGraph, jak łączyć aktualizacje: nowe wiadomości są doklejane do listy, a nie ją zastępują. Klucze bez adnotacji są nadpisywane. Dla czatu jest też gotowy reduktor add_messages z modułu langgraph.graph.message.

Podstawowy graf agenta

Budowa grafu ma zawsze tę samą kolejność: definicja stanu, StateGraph, węzły (add_node), krawędzie (add_edge) i kompilacja (compile). Zaczynamy od stanu i węzła agenta:

1from langgraph.graph import StateGraph, START, END
2
3# Definicja stanu
4class State(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_action: str
7
8# Funkcje węzłów
9def agent_node(state: State) -> State:
10    """Węzeł agenta - podejmuje decyzje."""
11    messages = state["messages"]
12
13    response = llm.invoke(messages)
14
15    return {
16        "messages": [response],
17        "next_action": "tool" if response.tool_calls else "end"
18    }

Węzeł to zwykła funkcja: dostaje stan, zwraca słownik z tymi kluczami, które chce zmienić. Żeby model w ogóle zwracał tool_calls, trzeba mu podpiąć narzędzia metodą llm.bind_tools(tools), inaczej lista będzie zawsze pusta.

Drugi węzeł wykonuje narzędzia, a router wybiera kolejny krok:

1def tool_node(state: State) -> State:
2    """Węzeł narzędzi - wykonuje akcje."""
3    last_message = state["messages"][-1]
4
5    # Wykonaj tool calls
6    tool_results = []
7    for tool_call in last_message.tool_calls:
8        result = execute_tool(tool_call)
9        tool_results.append(result)
10
11    return {"messages": tool_results, "next_action": "agent"}
12
13def router(state: State) -> Literal["tool", "end"]:
14    """Router - decyduje o następnym kroku."""
15    return state["next_action"]

execute_tool to miejsce na Twoją logikę. W praktyce zastąpisz ten węzeł gotowym ToolNode z langgraph.prebuilt.

Teraz łączymy wszystko w graf:

1# Budowa grafu
2workflow = StateGraph(State)
3
4# Dodaj węzły
5workflow.add_node("agent", agent_node)
6workflow.add_node("tool", tool_node)
7
8# Dodaj krawędzie
9workflow.add_edge(START, "agent")
10workflow.add_conditional_edges(
11    "agent",
12    router,
13    {
14        "tool": "tool",
15        "end": END
16    }
17)
18workflow.add_edge("tool", "agent")  # Cykl z powrotem do agenta
19
20# Kompilacja grafu
21app = workflow.compile()
22
23# Uruchomienie
24result = app.invoke({
25    "messages": [HumanMessage(content="Jaka jest pogoda w Serengeti?")],
26    "next_action": ""
27})

add_conditional_edges pyta router o nazwę ścieżki i tłumaczy ją słownikiem na węzeł. Krawędź z "tool" z powrotem do "agent" tworzy cykl.

ReAct Agent z LangGraph

Ten wzorzec jest tak popularny, że istnieje gotowiec. Najpierw narzędzia z dekoratorem @tool, znanym z lekcji o systemach multi-agentowych:

1from langchain_core.tools import tool
2from langgraph.prebuilt import create_react_agent
3
4# Definicja narzędzi
5@tool
6def get_weather(location: str) -> str:
7    """Pobiera pogodę dla lokalizacji Safari."""
8    weather_data = {
9        "Serengeti": "28°C, słonecznie",
10        "Masai Mara": "25°C, pochmurnie",
11        "Kruger": "30°C, gorąco"
12    }
13    return weather_data.get(location, f"Brak danych dla {location}")
14
15@tool
16def search_animals(query: str) -> str:
17    """Wyszukuje informacje o zwierzętach Safari."""
18    return f"Informacje o {query}: Fascynujące zwierzę afrykańskiej sawanny!"
19
20@tool
21def calculate_safari_cost(days: int, people: int) -> str:
22    """Oblicza koszt Safari."""
23    cost_per_day = 200
24    total = days * people * cost_per_day
25    return f"Koszt Safari: {total} USD ({days} dni, {people} osoby)"

Każde narzędzie ma docstring, który model czyta jako opis, i typy parametrów, które mówią mu, co podać.

create_react_agent buduje cały graf agent-narzędzia w jednej linijce:

1# Tworzenie agenta ReAct
2tools = [get_weather, search_animals, calculate_safari_cost]
3react_agent = create_react_agent(llm, tools)
4
5# Uruchomienie
6result = react_agent.invoke({
7    "messages": [HumanMessage(content="Oblicz koszt 5-dniowego Safari dla 4 osób")]
8})
9print(result["messages"][-1].content)

Model sam wywoła calculate_safari_cost z days=5 i people=4. Uwaga: w LangGraph 1.0 create_react_agent jest oznaczony jako deprecated, a nowy kod powinien używać create_agent z pakietu langchain.agents.

Multi-Agent z LangGraph

Graf świetnie nadaje się do zespołu agentów. Każdy agent to węzeł, a pole current_agent w stanie mówi, kto jest następny:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class MultiAgentState(TypedDict):
5    messages: Annotated[list, operator.add]
6    current_agent: str
7    task_complete: bool
8
9# Agenci specjaliści
10def researcher_agent(state: MultiAgentState) -> MultiAgentState:
11    """Agent badawczy - zbiera informacje."""
12    system_prompt = "Jesteś badaczem Safari. Zbieraj informacje o zwierzętach i lokalizacjach."
13
14    response = llm.invoke([
15        {"role": "system", "content": system_prompt},
16        *state["messages"]
17    ])
18
19    return {
20        "messages": [AIMessage(content=f"[Researcher]: {response.content}")],
21        "current_agent": "analyzer"
22    }

Badacz dopisuje swoją odpowiedź do messages i wskazuje analityka jako kolejnego.

Analityk i pisarz działają tak samo, tylko pisarz kończy pracę flagą task_complete:

1def analyzer_agent(state: MultiAgentState) -> MultiAgentState:
2    """Agent analityczny - analizuje dane."""
3    system_prompt = "Jesteś analitykiem. Analizuj zebrane informacje i wyciągaj wnioski."
4
5    response = llm.invoke([
6        {"role": "system", "content": system_prompt},
7        *state["messages"]
8    ])
9
10    return {
11        "messages": [AIMessage(content=f"[Analyzer]: {response.content}")],
12        "current_agent": "writer"
13    }
14
15def writer_agent(state: MultiAgentState) -> MultiAgentState:
16    """Agent pisarski - tworzy końcowy raport."""
17    system_prompt = "Jesteś pisarzem. Stwórz angażujący raport na podstawie analizy."
18
19    response = llm.invoke([
20        {"role": "system", "content": system_prompt},
21        *state["messages"]
22    ])
23
24    return {
25        "messages": [AIMessage(content=f"[Writer]: {response.content}")],
26        "current_agent": "end",
27        "task_complete": True
28    }

Każdy agent widzi całą historię wiadomości, bo messages rośnie z każdym węzłem.

Pozostaje router i graf:

1def router(state: MultiAgentState) -> Literal["analyzer", "writer", "end"]:
2    """Router między agentami."""
3    return state["current_agent"]
4
5# Budowa grafu multi-agent
6multi_agent_graph = StateGraph(MultiAgentState)
7
8multi_agent_graph.add_node("researcher", researcher_agent)
9multi_agent_graph.add_node("analyzer", analyzer_agent)
10multi_agent_graph.add_node("writer", writer_agent)
11
12multi_agent_graph.add_edge(START, "researcher")
13multi_agent_graph.add_conditional_edges(
14    "researcher",
15    router,
16    {"analyzer": "analyzer"}
17)
18multi_agent_graph.add_conditional_edges(
19    "analyzer",
20    router,
21    {"writer": "writer"}
22)
23multi_agent_graph.add_conditional_edges(
24    "writer",
25    router,
26    {"end": END}
27)
28
29multi_agent_app = multi_agent_graph.compile()
30
31# Uruchomienie pipeline
32result = multi_agent_app.invoke({
33    "messages": [HumanMessage(content="Przygotuj raport o lwach w Serengeti")],
34    "current_agent": "researcher",
35    "task_complete": False
36})

Kolejność jest liniowa, ale router pozwala łatwo dodać powrót do badacza.

Checkpointing i Persistence

Domyślnie graf zapomina wszystko po invoke. Checkpointer zapisuje stan po każdym kroku, co pozwala kontynuować konwersację:

1from langgraph.checkpoint.memory import MemorySaver
2from langgraph.checkpoint.sqlite import SqliteSaver  # pip install langgraph-checkpoint-sqlite
3import sqlite3
4
5# Memory checkpoint (w pamięci)
6memory_checkpointer = MemorySaver()
7
8# SQLite checkpoint (trwały)
9sqlite_checkpointer = SqliteSaver(sqlite3.connect("safari_agent.db", check_same_thread=False))
10
11# Kompilacja z checkpointerem
12app_with_memory = workflow.compile(checkpointer=memory_checkpointer)

MemorySaver trzyma stan w pamięci i traci go przy restarcie. SqliteSaver z pakietu langgraph-checkpoint-sqlite zapisuje go w pliku bazy SQLite.

Rozmowy rozróżnia thread_id w konfiguracji:

1# Uruchomienie z thread_id dla kontynuacji
2config = {"configurable": {"thread_id": "safari-session-1"}}
3
4# Pierwsza wiadomość
5result1 = app_with_memory.invoke(
6    {"messages": [HumanMessage(content="Opowiedz mi o lwach")]},
7    config
8)
9
10# Kontynuacja rozmowy (ten sam thread_id)
11result2 = app_with_memory.invoke(
12    {"messages": [HumanMessage(content="A co z ich polowaniami?")]},
13    config
14)
15
16# Historia jest zachowana!
17print(result2["messages"])

Drugie wywołanie z tym samym thread_id widzi pierwszą wiadomość, więc model wie, że "ich" oznacza lwy.

Human-in-the-loop

Niektórych decyzji agent nie powinien podejmować sam. Najpierw stan, węzły i router:

1from langgraph.graph import StateGraph, START, END
2from langgraph.checkpoint.memory import MemorySaver
3
4class HumanLoopState(TypedDict):
5    messages: Annotated[list, operator.add]
6    needs_approval: bool
7    approved: bool
8
9def agent_node(state: HumanLoopState) -> HumanLoopState:
10    """Agent proponuje akcję."""
11    response = llm.invoke(state["messages"])
12
13    # Sprawdź czy akcja wymaga zatwierdzenia
14    needs_approval = "delete" in response.content.lower() or "buy" in response.content.lower()
15
16    return {
17        "messages": [response],
18        "needs_approval": needs_approval
19    }
20
21def human_approval_node(state: HumanLoopState) -> HumanLoopState:
22    """Punkt zatwierdzenia przez człowieka."""
23    # Ten węzeł zatrzymuje wykonanie
24    # Użytkownik musi zatwierdzić przed kontynuacją
25    return {"approved": state.get("approved", False)}
26
27def execute_node(state: HumanLoopState) -> HumanLoopState:
28    """Wykonuje zatwierdzoną akcję."""
29    return {"messages": [AIMessage(content="Akcja wykonana!")]}
30
31def approval_router(state: HumanLoopState) -> Literal["human", "execute", "end"]:
32    """Router sprawdzający czy potrzeba zatwierdzenia."""
33    if state.get("needs_approval") and not state.get("approved"):
34        return "human"
35    elif state.get("approved"):
36        return "execute"
37    return "end"

Węzeł human niczego nie liczy, jest tylko miejscem postoju. Zwraca jeden klucz bez zmian, bo zwrócenie całego stanu dokleiłoby wszystkie wiadomości drugi raz przez reduktor operator.add.

Kompilujemy graf z checkpointerem i interrupt_before:

1# Graf z human-in-the-loop
2human_loop_graph = StateGraph(HumanLoopState)
3
4human_loop_graph.add_node("agent", agent_node)
5human_loop_graph.add_node("human", human_approval_node)
6human_loop_graph.add_node("execute", execute_node)
7
8human_loop_graph.add_edge(START, "agent")
9human_loop_graph.add_conditional_edges(
10    "agent",
11    approval_router,
12    {"human": "human", "execute": "execute", "end": END}
13)
14human_loop_graph.add_edge("human", "agent")  # Po zatwierdzeniu wróć do agenta
15human_loop_graph.add_edge("execute", END)
16
17# Kompilacja z interrupt_before dla human node
18app = human_loop_graph.compile(
19    checkpointer=MemorySaver(),
20    interrupt_before=["human"]  # Zatrzymaj przed human node
21)

Router zwraca "end", dlatego słownik ścieżek tłumaczy tę nazwę na END. Bez niego LangGraph zgłosi błąd nieznanego węzła end.

Uruchomienie zatrzymuje się przed węzłem human, a my zatwierdzamy i wznawiamy:

1# Uruchomienie - zatrzyma się przed human node
2config = {"configurable": {"thread_id": "approval-1"}}
3result = app.invoke(
4    {"messages": [HumanMessage(content="Usuń wszystkie pliki")]},
5    config
6)
7
8# Sprawdź stan
9state = app.get_state(config)
10print(f"Needs approval: {state.values.get('needs_approval')}")
11
12# Użytkownik zatwierdza
13app.update_state(config, {"approved": True})
14
15# Kontynuuj wykonanie
16final_result = app.invoke(None, config)

update_state zapisuje zgodę, a invoke(None, config) wznawia graf i router kieruje przepływ do execute. W nowszych wersjach wygodniejsza jest funkcja interrupt() wywoływana wewnątrz węzła.

Branching i Parallel Execution

Dwie niezależne gałęzie mogą działać jednocześnie. Najpierw funkcje węzłów:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class ParallelState(TypedDict):
5    query: str
6    weather_result: str
7    animal_result: str
8    combined_result: str
9
10def weather_branch(state: ParallelState) -> ParallelState:
11    """Pobiera pogodę równolegle."""
12    # Symulacja wywołania API
13    return {"weather_result": "Pogoda: 28°C, słonecznie"}
14
15def animal_branch(state: ParallelState) -> ParallelState:
16    """Pobiera info o zwierzętach równolegle."""
17    return {"animal_result": "Zwierzęta: Lwy, słonie, żyrafy"}
18
19def combine_results(state: ParallelState) -> ParallelState:
20    """Łączy wyniki z równoległych gałęzi."""
21    combined = f"{state['weather_result']}. {state['animal_result']}"
22    return {"combined_result": combined}

Każda gałąź zapisuje inny klucz stanu, więc nie ma konfliktu.

Dwie krawędzie z START rozgałęziają przepływ:

1# Graf z równoległym wykonaniem
2parallel_graph = StateGraph(ParallelState)
3
4parallel_graph.add_node("weather", weather_branch)
5parallel_graph.add_node("animals", animal_branch)
6parallel_graph.add_node("combine", combine_results)
7
8# Równoległe krawędzie
9parallel_graph.add_edge(START, "weather")
10parallel_graph.add_edge(START, "animals")
11parallel_graph.add_edge("weather", "combine")
12parallel_graph.add_edge("animals", "combine")
13parallel_graph.add_edge("combine", END)
14
15parallel_app = parallel_graph.compile()
16
17result = parallel_app.invoke({"query": "Safari info"})
18print(result["combined_result"])

weather i animals wykonują się w tym samym kroku, a combine rusza dopiero, gdy obie skończą.

Supervisor Pattern

W tym wzorcu jeden agent zarządza resztą. Supervisor pyta model, kto ma pracować:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class SupervisorState(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_worker: str
7    completed_workers: list[str]
8
9def supervisor_node(state: SupervisorState) -> SupervisorState:
10    """Supervisor decyduje który worker ma działać."""
11    system_prompt = """Jesteś supervisorem zarządzającym zespołem:
12    - researcher: zbiera informacje
13    - analyst: analizuje dane
14    - writer: tworzy raporty
15
16    Na podstawie zadania i dotychczasowych wyników, wybierz następnego workera.
17    Odpowiedz TYLKO nazwą workera lub 'FINISH' jeśli zadanie ukończone."""
18
19    completed = state.get("completed_workers", [])
20
21    response = llm.invoke([
22        {"role": "system", "content": system_prompt},
23        {"role": "user", "content": f"Completed: {completed}\nTask: {state['messages'][-1].content}"}
24    ])
25
26    next_worker = response.content.strip().lower()
27
28    return {"next_worker": next_worker}

Supervisor zwraca tylko nazwę kolejnego workera.

Fabryka tworzy węzły workerów, a router tłumaczy decyzję:

1def worker_node(worker_name: str):
2    """Fabryka węzłów workerów."""
3    def node(state: SupervisorState) -> SupervisorState:
4        prompts = {
5            "researcher": "Jesteś badaczem. Zbierz informacje.",
6            "analyst": "Jesteś analitykiem. Przeanalizuj dane.",
7            "writer": "Jesteś pisarzem. Napisz raport."
8        }
9
10        response = llm.invoke([
11            {"role": "system", "content": prompts[worker_name]},
12            *state["messages"]
13        ])
14
15        completed = state.get("completed_workers", []) + [worker_name]
16
17        return {
18            "messages": [AIMessage(content=f"[{worker_name}]: {response.content}")],
19            "completed_workers": completed
20        }
21    return node
22
23def supervisor_router(state: SupervisorState) -> Literal["researcher", "analyst", "writer", "end"]:
24    """Router supervisora."""
25    next_worker = state.get("next_worker", "").lower()
26    if next_worker == "finish" or next_worker == "end":
27        return "end"
28    return next_worker

Jeśli model odpowie czymś spoza listy, router zwróci nieznaną nazwę i graf się wysypie, dlatego w produkcji wymuś format odpowiedzi.

Każdy worker oddaje sterowanie supervisorowi:

1# Budowa grafu supervisora
2supervisor_graph = StateGraph(SupervisorState)
3
4supervisor_graph.add_node("supervisor", supervisor_node)
5supervisor_graph.add_node("researcher", worker_node("researcher"))
6supervisor_graph.add_node("analyst", worker_node("analyst"))
7supervisor_graph.add_node("writer", worker_node("writer"))
8
9supervisor_graph.add_edge(START, "supervisor")
10
11# Wszystkie krawędzie prowadzą z powrotem do supervisora
12for worker in ["researcher", "analyst", "writer"]:
13    supervisor_graph.add_edge(worker, "supervisor")
14
15supervisor_graph.add_conditional_edges(
16    "supervisor",
17    supervisor_router,
18    {
19        "researcher": "researcher",
20        "analyst": "analyst",
21        "writer": "writer",
22        "end": END
23    }
24)
25
26supervisor_app = supervisor_graph.compile()
27
28# Uruchomienie
29result = supervisor_app.invoke({
30    "messages": [HumanMessage(content="Przygotuj kompletny raport o Safari w Serengeti")],
31    "completed_workers": []
32})

Polecam zaczynać od grafu liniowego, a cykle i supervisora dodawać, gdy naprawdę potrzebujesz decyzji w trakcie.

LangGraph to potężne narzędzie do budowania złożonych przepływów AI. W następnej lekcji poznasz subagentów i hierarchie agentów!

Zapamiętaj: LangGraph to mapa szlaków, na której wolno zawrócić, rozdzielić się i poczekać na decyzję przewodnika.