LangChain Expression Language (LCEL) je elegantní, to bez debaty. Pro lineární pipeline typu prompt | model | parser je čitelný, kompaktní a střelíte v něm spoustu věcí během odpoledne. Jakmile ale potřebujete agenta, který se sám rozhoduje, jestli zavolat nástroj a kolikrát, narazíte na strop.
Ze své zkušenosti — když jsem poprvé zkoušel cyklus přes LCEL pomocí RunnableLambda a vlastního while, výsledek se sice zkompiloval, ale debugovat se to nedalo prakticky vůbec. LangGraph řeší tři klíčové problémy:
- Cykly — agent může volat nástroj opakovaně, dokud nedostane uspokojivou odpověď. V LCEL byste museli psát vlastní while smyčku mimo framework (a já to nedoporučuji nikomu).
- Stav jako prvotřídní občan — explicitně definovaný TypedDict drží historii zpráv, mezivýsledky a metadata mezi kroky. Žádné ruční žonglování slovníky.
- Persistence a obnova — vestavěné checkpointery (SQLite, Postgres, Redis) ukládají stav po každém kroku, takže agenta lze pozastavit, zeptat se uživatele a klidně i o pár hodin později pokračovat.
Mimochodem, LangGraph není nahrazením LangChainu. Je to jeho rozšíření. Modely, nástroje a parsery z LangChainu fungují uvnitř LangGraph uzlů beze změny — což je hezké, protože nemusíte nic přepisovat.
Instalace a základní koncepty
pip install langgraph langchain-anthropic langchain-community
pip install langgraph-checkpoint-sqlite
Každý LangGraph workflow se skládá ze tří entit:
- State — TypedDict (nebo Pydantic model) popisující data sdílená mezi uzly.
- Nodes — Python funkce přijímající state a vracející částečnou aktualizaci.
- Edges — pravidla pro přechod mezi uzly. Mohou být statické, nebo podmíněné (rozhodují se podle stavu).
To je celé. Tři pojmy, na kterých postavíte celou architekturu.
Příklad 1: ReAct agent s nástroji od základů
ReAct (Reasoning + Acting) je nejrozšířenější vzor pro agenty s nástroji: model uvažuje, zvolí nástroj, dostane výsledek a uvažuje znovu. LangGraph má sice prebuilt funkci create_react_agent (k té se ještě dostaneme), ale postavíme si ho ručně. Důvod? Když chápete mechaniku, debugování v produkci je výrazně rychlejší.
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import tool
# 1. Definice stavu — add_messages je reducer, ktery pripojuje zpravy
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# 2. Definice nastroju
@tool
def get_weather(city: str) -> str:
"""Vrati aktualni pocasi pro dane mesto."""
return f"V {city} je 18 stupnu, polojasno."
@tool
def calculate(expression: str) -> str:
"""Vyhodnoti matematicky vyraz."""
try:
return str(eval(expression, {"__builtins__": {}}))
except Exception as e:
return f"Chyba: {e}"
tools = [get_weather, calculate]
tools_by_name = {t.name: t for t in tools}
# 3. Model s navazanymi nastroji
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# 4. Uzly grafu
def call_model(state: AgentState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def call_tools(state: AgentState):
last_msg = state["messages"][-1]
tool_messages = []
for call in last_msg.tool_calls:
tool = tools_by_name[call["name"]]
result = tool.invoke(call["args"])
tool_messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": call["id"],
})
return {"messages": tool_messages}
# 5. Podminena hrana — rozhodne, jestli pokracovat nebo skoncit
def should_continue(state: AgentState) -> str:
last = state["messages"][-1]
if last.tool_calls:
return "tools"
return END
# 6. Sestaveni grafu
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent") # cyklus zpet na model
app = graph.compile()
# 7. Spusteni
result = app.invoke({
"messages": [HumanMessage(content="Kolik je 17 * 23 a jake je pocasi v Praze?")]
})
for msg in result["messages"]:
msg.pretty_print()
Klíčová pasáž je add_conditional_edges. Po každé odpovědi modelu se zavolá should_continue, která se podívá, jestli model požádal o nástroj. Pokud ano, jdeme do uzlu tools, vykonáme volání a vrátíme se zpět do agent. A cyklus pokračuje, dokud model nedá finální odpověď bez tool_calls.
Reducer add_messages — co to vlastně dělá
Reducer je funkce, která určí, jak se aktualizace z uzlu sloučí do stavu. Defaultně LangGraph hodnoty přepisuje — to je důležité si zapamatovat. add_messages je speciální reducer, který:
- Připojí nové zprávy k existující historii (místo přepsání).
- Deduplikuje zprávy podle
id — pokud uzel vrátí zprávu se stejným ID, ta v historii se aktualizuje.
- Automaticky doplní ID, pokud chybí.
Pro vlastní pole stavu (například seznam nálezů u výzkumného agenta) si můžete napsat vlastní reducer:
from operator import add
class ResearchState(TypedDict):
query: str
sources: Annotated[list[str], add] # konkatenace seznamu
summary: str # prepisovani (default)
Příklad 2: Výzkumný workflow s human-in-the-loop
Tak. Reálná aplikace často potřebuje schválení od člověka, než provede nákladnou nebo nevratnou akci — pošle e-mail, utratí peníze, vytvoří PR. LangGraph na to má primitivu interrupt, která zastaví běh grafu a vrátí kontrolu volajícímu kódu. Po obdržení odpovědi se graf obnoví přesně z místa, kde skončil — to funguje díky checkpointeru.
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
class ResearchState(TypedDict):
topic: str
plan: str
approved: bool
report: str
def create_plan(state: ResearchState):
plan = llm.invoke(
f"Vytvor 3-bodovy plan vyzkumu na tema: {state['topic']}"
).content
return {"plan": plan}
def human_approval(state: ResearchState):
# Zastavime graf a pockame na vstup
decision = interrupt({
"question": "Schvalujete tento plan?",
"plan": state["plan"],
})
return {"approved": decision == "yes"}
def execute_research(state: ResearchState):
if not state["approved"]:
return {"report": "Vyzkum zrusen uzivatelem."}
report = llm.invoke(
f"Provedl vyzkum podle planu:\n{state['plan']}"
).content
return {"report": report}
def route_after_approval(state: ResearchState) -> str:
return "execute" if state["approved"] else END
# Sestaveni s checkpointerem
checkpointer = SqliteSaver.from_conn_string("research.db")
graph = StateGraph(ResearchState)
graph.add_node("plan", create_plan)
graph.add_node("approval", human_approval)
graph.add_node("execute", execute_research)
graph.add_edge(START, "plan")
graph.add_edge("plan", "approval")
graph.add_conditional_edges("approval", route_after_approval, ["execute", END])
graph.add_edge("execute", END)
app = graph.compile(checkpointer=checkpointer)
# Spusteni do interruptu
config = {"configurable": {"thread_id": "user-42"}}
result = app.invoke({"topic": "GraphRAG vs vektorove RAG"}, config=config)
print("Plan ke schvaleni:", result["__interrupt__"][0].value)
# Pokracovani po schvaleni (muze byt i o hodiny pozdeji)
final = app.invoke(Command(resume="yes"), config=config)
print("Report:", final["report"])
Všimněte si thread_id v konfiguraci — to je identifikátor konverzace pro checkpointer. Můžete spustit tisíce paralelních agentů s různými thread ID a každý si drží vlastní stav. Při restartu serveru se obnoví ze stejného bodu, jako by se nic nestalo.
Příklad 3: Persistentní chat agent s pamětí
Teď spojíme předchozí koncepty: agent s nástroji, který si pamatuje historii konverzace mezi voláními a může klidně běžet měsíce.
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent
# Persistentni spojeni (ne in-memory)
conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
agent = create_react_agent(
model=llm,
tools=tools,
checkpointer=checkpointer,
)
# Prvni zprava
config = {"configurable": {"thread_id": "user-7"}}
agent.invoke(
{"messages": [HumanMessage(content="Jmenuji se Tomas.")]},
config=config,
)
# O tyden pozdeji — agent si pamatuje
response = agent.invoke(
{"messages": [HumanMessage(content="Jak se jmenuji?")]},
config=config,
)
print(response["messages"][-1].content)
# Vystup: "Jmenujes se Tomas."
Pro produkci doporučujeme PostgresSaver z balíku langgraph-checkpoint-postgres — zvládne tisíce konkurentních threadů a podporuje async API. SQLite nechte na vývoj a malé deploye, opravdu (mám s ním úspěšně rozjeté pet projekty, ale nikdy bych na něm neprovozoval B2B produkt).
Optimalizace velikosti stavu
Při dlouhých konverzacích historie zpráv roste a každý krok posílá celý kontext modelu. Což rychle zabolí na fakturaci. Tři techniky, jak to zvládnout:
- Trimming — pomocí
trim_messages z LangChainu nechat jen posledních N zpráv nebo M tokenů.
- Sumarizace — speciální uzel, který periodicky shrne staré zprávy do jedné systémové.
- Externí paměť — uložit relevantní fakta do vektorové DB nebo Mem0 a načítat jen potřebné.
Streamování průběhu agenta
Pro UX je naprosto kritické zobrazovat uživateli, co agent právě dělá — ne nechat ho zírat na spinner a čekat na finální odpověď. LangGraph podporuje streamování na úrovni uzlů i tokenů:
# Stream udalosti — jeden event za kazdy krok grafu
for chunk in app.stream(input_data, config=config, stream_mode="updates"):
for node_name, update in chunk.items():
print(f"[{node_name}] {update}")
# Stream tokenu z LLM uvnitr uzlu
async for event in app.astream_events(input_data, config=config, version="v2"):
if event["event"] == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="", flush=True)
Režim stream_mode="values" vrátí celý stav po každém kroku, "updates" jen rozdíly a "messages" proudí jednotlivé tokeny zpráv. Pro produkci běžně kombinujeme updates (pro stav) s messages (pro tokeny) přes stream_mode=["updates", "messages"].
Subgrafy a hierarchická orchestrace
Komplexní systémy rozdělujeme do subgrafů. Každý agent (například výzkumník, kritik, editor) je vlastní StateGraph, který se zkompiluje a použije jako uzel v nadřazeném grafu. Subgrafy mohou mít vlastní stav, který se mapuje na rodičovský přes shodná jména klíčů, případně přes explicitní transformaci.
researcher = build_researcher_graph() # vraci zkompilovany StateGraph
critic = build_critic_graph()
supervisor = StateGraph(SupervisorState)
supervisor.add_node("research", researcher)
supervisor.add_node("critique", critic)
supervisor.add_node("route", routing_node)
supervisor.add_conditional_edges("route", decide_next_agent)
Tenhle vzor — supervisor + workers — je dnes nejčastější architektura pro multi-agentní systémy. Pro hlubší pohled doporučujeme náš článek o multi-agentní orchestraci.
Časté chyby a jak se jim vyhnout
1. Zapomenutý reducer u stavu se zprávami
Bez Annotated[list[BaseMessage], add_messages] se každý uzel přepíše historií, kterou vrátí. Symptom: agent ztrácí kontext po prvním kroku. (Tuhle chybu jsem udělal asi třikrát, než mi to docvaklo.)
2. Nekonečný cyklus mezi modelem a nástroji
Pokud model trvale generuje tool_calls (typicky kvůli špatnému promptu), graf běží donekonečna. Řešení: graph.compile(...) přijímá recursion_limit (default 25). Také lze přidat počítadlo do stavu a kontrolovat ho v podmíněné hraně.
3. Mutace stavu místo návratu rozdílu
Uzel musí vrátit slovník s aktualizacemi, ne mutovat state přímo. Mutace fungují někdy, ale rozbíjejí checkpointing a paralelní vykonávání. Tedy: rozbíjejí přesně to, kvůli čemu LangGraph používáte.
4. Sdílená SQLite spojení mezi vlákny
Při použití SqliteSaver v multi-threaded prostředí (například FastAPI s více workery) musí být spojení vytvořeno s check_same_thread=False, jinak dostanete ProgrammingError.
Kdy LangGraph nepoužít
LangGraph přidává abstrakci, která má smysl jen u dostatečně komplexních workflow. Pokud váš úkol zní:
- "Vezmi text, nech ho přeložit, vrať výsledek" — použijte
llm.invoke().
- "Sumarizuj dokument po částech" — stačí LCEL nebo prostá Python smyčka.
- "Jednorázové RAG dotazování bez agenta" — LCEL pipeline je čitelnější.
LangGraph se vyplatí, když potřebujete cykly, větvení podle obsahu, persistentní stav nebo human-in-the-loop. Pro jednoduché případy je to overkill, žádný rozumný člověk to nezpochybní.
Často kladené otázky
Jaký je rozdíl mezi LangGraph a LangChain Agent Executor?
LangChain Agent Executor je černá skříňka — nevidíte detaily smyčky a obtížně se ladí. LangGraph je naopak explicitní graf, kde vidíte každý uzel, hranu i stav. Od konce roku 2024 LangChain tým doporučuje LangGraph místo Agent Executoru pro všechny nové projekty a starší API postupně deprecuje.
Funguje LangGraph s jinými LLM než OpenAI?
Ano, LangGraph je nezávislý na poskytovateli. Funguje s libovolným LangChain chat modelem — Anthropic Claude, Google Gemini, Mistral, lokální modely přes Ollama nebo vLLM. Klíčové je, aby model podporoval tool calling, pokud stavíte ReAct agenta.
Jak nasadit LangGraph agenta do produkce?
Tři osvědčené cesty: (1) obalit graf FastAPI endpointem a spravovat vlastní infrastrukturu; (2) použít LangGraph Platform (managed služba od LangChain Inc.) s vestavěným škálováním a monitoringem; (3) deploy do Kubernetes s PostgresSaver pro persistenci a Redis pro pub/sub událostí. Pro jednodušší případy postačí Modal nebo Railway s SQLite checkpointerem.
Jak monitorovat běh LangGraph agenta?
LangGraph se nativně integruje s LangSmith a Langfuse přes OpenTelemetry. Stačí nastavit příslušné environment proměnné a každý krok grafu se automaticky loguje s vstupy, výstupy a latencí. Pro vlastní metriky doporučujeme přidat callbacks do konfigurace při compile().
Mohu LangGraph použít synchronně i asynchronně?
Ano, každá metoda má sync i async variantu: invoke/ainvoke, stream/astream. Pro produkční API server vždy preferujte async — během čekání na LLM odpověď se uvolní event loop pro další požadavky. Uzly mohou být sync i async funkce; LangGraph je zvládne smíchat.
Závěr
LangGraph posunul stavbu AI agentů z hackování smyček kolem LangChainu na inženýrskou disciplínu s explicitním stavem, persistencí a kontrolou toku. Tři klíčové vzory, které byste si měli odnést: ReAct cyklus s podmíněnou hranou, human-in-the-loop přes interrupt, a persistence přes checkpointer. S těmito třemi nástroji postavíte 90 % produkčních agentů — zbylých 10 % je ladění promptů a infrastruktury, ale to už je jiný příběh.
V dalším článku se podíváme, jak kombinovat LangGraph s GraphRAG a jak měřit kvalitu agentních workflow přes DeepEval. Mezitím doporučujeme projít oficiální LangGraph Academy kurzy, které pokrývají pokročilé vzory jako planning, reflection a tool routing.
