langgraph
LangGraph is LangChain's low-level orchestration framework for stateful, long-running agents. You model your application as a graph: nodes are Python functions that update a shared typed state, edges decide what runs next, and the runtime handles persistence, streaming, and resumption. Its headline features are durable execution (a checkpointer saves state after every step, so a crashed or interrupted run resumes exactly where it stopped), human-in-the-loop interrupts, and per-thread memory. It is inspired by Google's Pregel model and can be used entirely without LangChain, though most teams pair the two. It hit 1.0 in late 2025 after two years as the de facto standard for production agent orchestration.
The most capable orchestration layer for production agents that need state, persistence, and human oversight, and now stable at 1.x. Expect a real learning curve and stale pre-1.0 tutorials; if your problem fits a prebuilt agent loop, you may not need the graph at all.
Use it if
- You are building agents that need cycles, branching, retries, or multiple cooperating steps, which plain chain-style pipelines cannot express
- You need human-in-the-loop approval gates; interrupt() pauses a run mid-graph and Command(resume=...) picks it up later, even days later, from the checkpoint
- You need durable, resumable execution: per-thread conversation state persisted to memory, SQLite, or Postgres via pluggable checkpointers
- You want fine control over agent behavior that higher-level frameworks hide, while keeping streaming and persistence off your plate
- Your workflow is a straight line of LLM calls; a graph framework adds state schemas, reducers, and compile steps for no benefit over plain functions
- You want a batteries-included multi-agent abstraction with roles and conversation patterns out of the box; CrewAI or AutoGen get you a demo faster, and LangGraph makes you build those patterns yourself
- You are wary of the LangChain ecosystem's churn; the 1.0 transition moved docs, renamed prebuilt helpers (create_react_agent is deprecated in favor of langchain's create_agent), and split checkpointers into separate packages, so older tutorials mislead
- Concurrency needs are simple background jobs, not agent state machines; a task queue like Celery is the boring, right tool
Setup reality
pip install langgraph is clean on Python 3.10+, but the real setup is conceptual: you define a TypedDict state, learn reducers (why messages need the add_messages annotation), and remember to compile the graph before running it. Persistence beyond memory means extra packages (langgraph-checkpoint-sqlite or langgraph-checkpoint-postgres) plus a setup() call people forget. Every stateful call needs a config with configurable.thread_id, and forgetting it is the most common runtime error. Docs moved to docs.langchain.com at 1.0, so many search results point at dead langchain-ai.github.io pages.
Patterns
Define state, add nodes, compile, invokebuild-basic-graph
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
topic: str
draft: str
def write(state: State) -> dict:
return {"draft": f"An essay about {state['topic']}"}
builder = StateGraph(State)
builder.add_node("write", write)
builder.add_edge(START, "write")
builder.add_edge("write", END)
graph = builder.compile()
print(graph.invoke({"topic": "graphs"}))Nodes return partial state updates as dicts, not full state. Nothing runs until compile(); forgetting it and calling invoke on the builder is a common first error.
Accumulate chat messages with a reducerchat-state-with-messages
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
def chatbot(state: State) -> dict:
reply = {"role": "assistant", "content": "hi"}
return {"messages": [reply]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
graph = builder.compile()Without the add_messages reducer each node would overwrite the message list instead of appending. langgraph.graph.MessagesState is a prebuilt shortcut for exactly this schema.
Route to different nodes based on stateconditional-edges
def route(state: State) -> str:
if state["needs_search"]:
return "search"
return "answer"
builder.add_conditional_edges("classify", route, {"search": "search", "answer": "answer"})The routing function returns a key from the mapping, not a node call. The third argument mapping is optional if the function returns node names directly, but keeping it makes the drawn graph accurate.
Give the graph memory across invocationspersist-with-checkpointer
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [{"role": "user", "content": "hi"}]}, config)
# same thread_id later resumes with full history
graph.invoke({"messages": [{"role": "user", "content": "remember me?"}]}, config)Every call with a checkpointer requires configurable.thread_id or you get a runtime error. InMemorySaver is for development; use langgraph-checkpoint-postgres or -sqlite in production.
Stream state updates as nodes finishstream-graph-progress
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "hi"}]},
config,
stream_mode="updates",
):
print(chunk)stream_mode="updates" yields each node's delta, "values" yields the full state after each step, and "messages" yields LLM tokens. You can pass a list of modes to get several at once.
Pause for human approval and resumehuman-in-the-loop-interrupt
from langgraph.types import interrupt, Command
def approval(state: State) -> dict:
answer = interrupt({"question": "Approve this action?"})
return {"approved": answer}
# run until the interrupt
result = graph.invoke(inputs, config)
print(result["__interrupt__"])
# later, resume with the human's answer
graph.invoke(Command(resume=True), config)interrupt() requires a checkpointer and a thread_id. On resume the node re-runs from its start, so keep side effects after the interrupt call or make them idempotent.
Prebuilt tool-calling agent loopprebuilt-react-agent
from langgraph.prebuilt import create_react_agent
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Sunny in {city}"
agent = create_react_agent(
model="anthropic:claude-sonnet-4-5",
tools=[get_weather],
)
agent.invoke({"messages": [{"role": "user", "content": "weather in SF?"}]})Still works in 1.x but is deprecated; new code is steered toward create_agent in the langchain package. The model string form requires the matching provider package installed.
Fan out dynamic parallel work with Sendparallel-fanout-send
from langgraph.types import Send
def fan_out(state: State):
return [Send("summarize", {"doc": d}) for d in state["docs"]]
builder.add_conditional_edges("plan", fan_out, ["summarize"])Send launches one copy of the target node per item, each with its own private input. Results merge back through your state reducers, so the collecting field needs one (like operator.add).
Use a compiled graph as a nodesubgraph-as-node
child_builder = StateGraph(State)
# ... add child nodes/edges ...
child = child_builder.compile()
parent = StateGraph(State)
parent.add_node("child_flow", child)
parent.add_edge(START, "child_flow")
graph = parent.compile()If parent and child share state keys you can add the compiled child directly. Different schemas require wrapping the child in a function that translates state in and out.
Combine a state update with routing in one nodecommand-goto
from langgraph.types import Command
def decide(state: State) -> Command:
if state["score"] > 0.8:
return Command(update={"status": "done"}, goto="finish")
return Command(update={"attempts": state["attempts"] + 1}, goto="retry")Command replaces the pattern of a node plus a separate conditional edge. Annotate the return type so graph rendering knows the possible targets.
Read a thread's saved state and historyinspect-and-rewind-state
snapshot = graph.get_state(config)
print(snapshot.values) # current state
print(snapshot.next) # nodes that would run next
for state in graph.get_state_history(config):
print(state.config["configurable"]["checkpoint_id"])Invoking with a config that includes a past checkpoint_id forks the thread from that point, which is how time travel and re-running from an earlier step work.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| crewai | PyPI | You want role-based multi-agent crews with high-level abstractions and faster time to demo |
| openai-agents | PyPI | You want a lighter agent SDK with handoffs and guardrails, and OpenAI-first defaults |
| autogen-agentchat | PyPI | You want Microsoft's conversation-driven multi-agent patterns like group chats |
| langchain | PyPI | You just need create_agent and model integrations without designing a custom graph |