Podstawy LangGraph

LangGraph to framework od LangChain do budowy aplikacji opartych na grafach stanow. Umozliwia tworzenie cyklicznych przeplywow, checkpointow i zaawansowanych wzorcow agentowych — wszystko czego brakuje w klasycznym LangChain.

Dlaczego LangGraph?

Klasyczne lancuchy LangChain sa liniowe — dane plyna w jednym kierunku. Ale agenci potrzebuja petli: wywolaj narzedzie, przeanalizuj wynik, zdecyduj co dalej. LangGraph rozwiazuje ten problem przez grafy stanow.

Kluczowe koncepty

Koncept	Opis	Przyklad
State	Struktura danych przeplywajaca przez graf	TypedDict z messages, result
Node	Funkcja przetwarzajaca stan	def agent(state) -> dict
Edge	Stale polaczenie miedzy wezlami	add_edge(“a”, “b”)
Conditional Edge	Dynamiczny routing na podstawie stanu	add_conditional_edges()
START / END	Specjalne wezly poczatku i konca	add_edge(START, “agent”)
Checkpointer	Zapisuje stan miedzy wywolaniami	MemorySaver, SqliteSaver

Instalacja

# Podstawowa instalacja
pip install langgraph langchain-openai

# Z dodatkowymi checkpointerami
pip install langgraph[sqlite]  # SqliteSaver
pip install langgraph[postgres]  # PostgresSaver

# Sprawdz wersje
pip show langgraph

Prosty graf — krok po kroku

Zacznijmy od najprostszego grafu bez LLM, aby zrozumiec podstawy:

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 1. DEFINICJA STANU
# Stan to TypedDict - definiuje jakie dane przeplywaja przez graf
class State(TypedDict):
    input: str
    processed: str
    final: str

# 2. DEFINICJA WEZLOW
# Kazdy wezel to funkcja: State -> dict (aktualizacje stanu)
def step_one(state: State) -> dict:
    """Pierwszy krok - przetwarzanie."""
    text = state["input"].upper()
    return {"processed": f"PROCESSED: {text}"}

def step_two(state: State) -> dict:
    """Drugi krok - finalizacja."""
    return {"final": state["processed"] + " | DONE"}

# 3. BUDOWA GRAFU
graph = StateGraph(State)

# Dodaj wezly
graph.add_node("process", step_one)
graph.add_node("finalize", step_two)

# Dodaj krawedzie (przejscia)
graph.add_edge(START, "process")      # START -> process
graph.add_edge("process", "finalize")  # process -> finalize
graph.add_edge("finalize", END)         # finalize -> END

# 4. KOMPILACJA
app = graph.compile()

# 5. URUCHOMIENIE
result = app.invoke({
    "input": "hello world",
    "processed": "",
    "final": ""
})

print(result)
# {'input': 'hello world', 'processed': 'PROCESSED: HELLO WORLD', 'final': 'PROCESSED: HELLO WORLD | DONE'}

Reducery — jak aktualizowac stan

Domyslnie nowe wartosci nadpisuja stare. Ale dla list (np. wiadomosci) chcemy dodawac, nie nadpisywac. Do tego sluza reducery:

from typing import Annotated
from langgraph.graph.message import add_messages
import operator

# Przyklad 1: add_messages dla wiadomosci czatu
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]  # dodaje wiadomosci

# Przyklad 2: operator.add dla list
class ListState(TypedDict):
    items: Annotated[list, operator.add]  # laczy listy

# Przyklad 3: wlasny reducer
def keep_last_5(current: list, new: list) -> list:
    """Zachowaj tylko 5 ostatnich elementow."""
    return (current + new)[-5:]

class LimitedState(TypedDict):
    history: Annotated[list, keep_last_5]

Warunkowe krawedzie

Conditional edges pozwalaja na dynamiczny routing — rozne sciezki w zaleznosci od stanu:

from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    value: int
    result: str

def check_value(state: State):
    return {"result": "checking..."}

def handle_positive(state: State):
    return {"result": "Value is positive!"}

def handle_negative(state: State):
    return {"result": "Value is negative!"}

def handle_zero(state: State):
    return {"result": "Value is zero!"}

# Funkcja routingu - zwraca nazwe nastepnego wezla
def route_by_value(state: State) -> str:
    value = state["value"]
    if value > 0:
        return "positive"
    elif value < 0:
        return "negative"
    else:
        return "zero"

# Budowa grafu
graph = StateGraph(State)
graph.add_node("check", check_value)
graph.add_node("positive", handle_positive)
graph.add_node("negative", handle_negative)
graph.add_node("zero", handle_zero)

graph.add_edge(START, "check")

# Warunkowa krawedz - rozne sciezki
graph.add_conditional_edges(
    "check",           # z jakiego wezla
    route_by_value,    # funkcja decyzji
    {                  # mapowanie zwracanych wartosci na wezly
        "positive": "positive",
        "negative": "negative",
        "zero": "zero"
    }
)

graph.add_edge("positive", END)
graph.add_edge("negative", END)
graph.add_edge("zero", END)

app = graph.compile()

# Test
print(app.invoke({"value": 5, "result": ""}))   # "Value is positive!"
print(app.invoke({"value": -3, "result": ""}))  # "Value is negative!"

Agent ReAct z narzedziami

Teraz zbudujmy prawdziwego agenta ReAct (Reason + Act) z petla:

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

# 1. DEFINICJA STANU
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 2. DEFINICJA NARZEDZI
@tool
def search(query: str) -> str:
    """Search the web for current information.
    
    Args:
        query: The search query string
    """
    # Symulacja - w produkcji uzyj prawdziwego API
    return f"Search results for '{query}': AI is advancing rapidly in 2025..."

@tool
def calculator(expression: str) -> str:
    """Calculate a mathematical expression.
    
    Args:
        expression: Math expression like '2 + 2' or '15 * 7'
    """
    try:
        result = eval(expression)
        return f"Result: {result}"
    except:
        return "Error: Invalid expression"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city.
    
    Args:
        city: Name of the city
    """
    return f"Weather in {city}: 22C, sunny, humidity 45%"

tools = [search, calculator, get_weather]

# 3. LLM Z NARZEDZIAMI
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 4. WEZEL AGENTA
def agent(state: AgentState) -> dict:
    """Wezel agenta - wywoluje LLM."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 5. FUNKCJA ROUTINGU
def should_continue(state: AgentState) -> str:
    """Decyzja: czy kontynuowac do narzedzi czy zakonczyc?"""
    last_message = state["messages"][-1]
    
    # Jesli LLM chce uzyc narzedzi
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"
    
    # Jesli nie - koncz
    return END

# 6. BUDOWA GRAFU
graph = StateGraph(AgentState)

# Dodaj wezly
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))  # wbudowany wezel narzedzi

# Krawedzie
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # PETLA! tools -> agent

# 7. KOMPILACJA
app = graph.compile()

# 8. URUCHOMIENIE
result = app.invoke({
    "messages": [("user", "What's 25 * 17? Also, what's the weather in Warsaw?")]
})

# Wyswietl konwersacje
for msg in result["messages"]:
    print(f"{msg.type}: {msg.content[:100] if msg.content else '[tool call]'}...")

Checkpointy i persystencja

Checkpointer zapisuje stan grafu po kazdym wezle — umozliwia wznowienie, pamiec miedzy sesjami i time-travel debugging:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# Opcja 1: Pamiec (dla developmentu)
memory_checkpointer = MemorySaver()

# Opcja 2: SQLite (dla produkcji)
sqlite_checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# Kompilacja z checkpointerem
app = graph.compile(checkpointer=memory_checkpointer)

# KLUCZOWE: thread_id identyfikuje konwersacje
config = {"configurable": {"thread_id": "user-123"}}

# Pierwsza wiadomosc
result1 = app.invoke(
    {"messages": [("user", "My name is Alice and I like Python.")]},
    config=config
)

# Druga wiadomosc - PAMIEC zachowana dzieki checkpointerowi!
result2 = app.invoke(
    {"messages": [("user", "What's my name and what do I like?")]},
    config=config
)
# Agent odpowie: "Your name is Alice and you like Python"

# Inny uzytkownik - nowy thread_id = czysta historia
config2 = {"configurable": {"thread_id": "user-456"}}
result3 = app.invoke(
    {"messages": [("user", "What's my name?")]},
    config=config2
)
# Agent odpowie: "I don't know your name"

Human-in-the-loop

LangGraph umozliwia zatrzymanie grafu przed wezlem i czekanie na akceptacje uzytkownika:

# Kompilacja z przerwaniem przed narzedziami
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"]  # zatrzymaj PRZED wykonaniem narzedzi
)

config = {"configurable": {"thread_id": "review-123"}}

# Uruchom - zatrzyma sie przed "tools"
result = app.invoke(
    {"messages": [("user", "Search for latest AI news")]},
    config=config
)

# Sprawdz jaki tool chce byc wywolany
state = app.get_state(config)
pending_tools = state.values["messages"][-1].tool_calls
print(f"Agent chce wywolac: {pending_tools}")

# Uzytkownik akceptuje - kontynuuj
if input("Approve? (y/n): ") == "y":
    result = app.invoke(None, config=config)  # None = kontynuuj
else:
    print("Cancelled by user")

Streaming

LangGraph wspiera granularny streaming — mozesz streamowac po wezlach lub tokenach:

# Streaming po wezlach
for event in app.stream(
    {"messages": [("user", "Tell me about AI")]},
    config=config
):
    for node_name, node_output in event.items():
        print(f"[{node_name}]: {node_output}")

# Streaming tokenow (dla UI typu ChatGPT)
for event in app.stream(
    {"messages": [("user", "Tell me about AI")]},
    config=config,
    stream_mode="messages"  # streamuj wiadomosci
):
    if event[0].content:
        print(event[0].content, end="", flush=True)

Prebuilt agents

LangGraph oferuje gotowe agenty dla szybkiego startu:

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# Gotowy agent ReAct w jednej linii!
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools=[search, calculator])

# Uzycie
result = agent.invoke({
    "messages": [("user", "What's 15% of 200?")]
})

# Z checkpointerem
agent_with_memory = create_react_agent(
    llm,
    tools=tools,
    checkpointer=MemorySaver()
)