LangGraph Cheat Sheet

1Install & importspip / packages

pip install langgraph★
Pulls langgraph-checkpoint (in-memory saver + base classes). Add a model package (langchain-openai) to call LLMs inside nodes.
from langgraph.graph import StateGraph, START, END★
Core builder + the two sentinel nodes. START is the entry, END is the exit.
from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command, Send, interrupt
Persistence + the dynamic-control primitives. Production checkpointers: langgraph-checkpoint-sqlite, -postgres.
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages
State is a TypedDict; reducers are attached with Annotated. Python 3.10+.

2Build & run a graphthe whole loop

class State(TypedDict): count: int builder = StateGraph(State)★
A graph is typed by its state schema. Nodes receive the state and return a partial update.
def inc(state: State): return {"count": state["count"] + 1} builder.add_node("inc", inc)★
A node is any callable state -> partial-state. The dict you return is merged into state (via reducers).
builder.add_edge(START, "inc") builder.add_edge("inc", END) graph = builder.compile()★
Wire the flow, then compile() into a runnable. compile() is where you attach a checkpointer/store.
graph.invoke({"count": 0}) # -> {'count': 1}★
The compiled graph is a Runnable: .invoke / .stream / .batch and async .ainvoke / .astream.

3State schema & reducershow updates merge

import operator class State(TypedDict): items: Annotated[list, operator.add]★
Without a reducer, a returned key overwrites. With operator.add, lists append/concatenate instead — how parallel nodes accumulate.
total: int # no reducer -> last write wins
Default behavior: each node's value replaces the previous one. Fine for scalars/flags.
def merge(old, new): return {**old, **new} data: Annotated[dict, merge]
Reducers are just (current, update) -> new functions — write your own for custom merge logic.
from langgraph.graph import StateGraph StateGraph(State, input_schema=In, output_schema=Out)
Optionally separate the public input/output schemas from the internal state (hide scratch fields).
builder.add_node("n", fn); Command(update={"items":[x]})
Nodes update state either by returning a dict or by returning a Command(update=...) (card 6).

4Messages & MessagesStatechat state

class State(TypedDict): messages: Annotated[list, add_messages]★
add_messages is the reducer for chat: it appends new messages and updates existing ones by id (not blind overwrite).
from langgraph.graph import MessagesState class State(MessagesState): extra: str★
MessagesState is a prebuilt base with the messages field already wired — subclass to add your own keys.
def call_model(state: MessagesState): return {"messages": [llm.invoke(state["messages"])]}★
The canonical LLM node: read history, return the new AIMessage in a one-element list — the reducer appends it.
graph.invoke({"messages": [{"role":"user","content":"hi"}]})
Dict messages are coerced to LangChain message objects. Read the answer at result["messages"][-1].

5Edges & conditional routingwho runs next

builder.add_edge("a", "b")★
Static edge: after a, always run b. Add edges from the same source to two nodes to run them in parallel.
def route(state) -> str: return "tools" if state["messages"][-1].tool_calls else END builder.add_conditional_edges("agent", route)★
The routing function returns the name(s) of the next node(s) — the heart of the agent loop (call tools, or finish).
builder.add_conditional_edges("agent", route, {"tools": "tools", END: END})
Optional mapping translates the router's return value to node names — and documents the possible targets for visualization.
builder.add_edge("tools", "agent") # loop back = a cycle★
Cycles are first-class — this back-edge is exactly what makes a ReAct agent loop until the model stops calling tools.

6Command — dynamic controlupdate + goto

from langgraph.types import Command def node(state) -> Command: return Command(update={"foo": "bar"}, goto="next")★
Return a Command to update state and pick the next node in one step — combines a node and a conditional edge.
Command(goto=Command.PARENT, graph=Command.PARENT)
Navigate to a node in the parent graph — the mechanism behind multi-agent handoffs across subgraphs.
Command(resume=value)★
Resume a graph that paused at an interrupt(), feeding value back in as the interrupt's return (card 9).
# tool returning Command -> updates agent state / hands off
Tools may return a Command too, letting a tool write to state or route control.

7Send — map-reducefan-out

from langgraph.types import Send def fan_out(state): return [Send("worker", {"item": x}) for x in state["items"]]★
Return a list of Send from a conditional edge to spawn N parallel copies of a node, each with its own payload — the map step.
builder.add_conditional_edges("start", fan_out, ["worker"])
Wire the fan-out through a conditional edge. Each Send gets a distinct state, independent of the main graph state.
results: Annotated[list, operator.add] # the reduce★
Workers write into a reducer-backed key; results concatenate as they finish — the reduce step. Great for parallel per-item LLM calls.

8Checkpointer & threadsdurable memory

graph = builder.compile(checkpointer=InMemorySaver())★
Attach a checkpointer at compile time. It snapshots state after every node — the basis of memory, resume, and time travel.
cfg = {"configurable": {"thread_id": "user-1"}} graph.invoke({"messages":[...]}, cfg)★
The thread_id is the conversation key. Reuse it and prior state is loaded automatically — no resending history.
from langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string(url) as cp: ...
Production persistence. Install langgraph-checkpoint-postgres / -sqlite; async variants exist (AsyncPostgresSaver).
graph.get_state(cfg) · graph.update_state(cfg, {"k": v})
Inspect the current snapshot or manually patch state between turns.

9Human-in-the-looppause for a human

from langgraph.types import interrupt def review(state): ok = interrupt({"question": "Approve?"}) return {"approved": ok}★v1
interrupt() pauses the graph mid-node and surfaces its payload to the caller. Requires a checkpointer. The dynamic, code-driven way to gate an action.
graph.invoke(Command(resume=True), cfg)★
Resume by re-invoking with Command(resume=value); the node re-runs and interrupt() now returns value.
graph = builder.compile(checkpointer=cp, interrupt_before=["tools"])
Static alternative: pause before (or interrupt_after) a named node without editing node code — approve, then resume.
# interrupt() needs a checkpointer + a thread_id, or it can't resumegotcha
On resume the whole node re-executes from the top — keep side effects after the interrupt() call, not before.

10Time travel & durabilityrewind / replay

for snap in graph.get_state_history(cfg): print(snap.config)★
Every checkpoint is inspectable. Each snapshot carries a checkpoint_id you can rewind to.
cfg2 = {"configurable": {"thread_id":"1", "checkpoint_id": "..."}} graph.invoke(None, cfg2) # replay from that point
Fork history: resume from an earlier checkpoint to explore an alternate path (great for debugging/what-ifs).
graph.invoke(inp, cfg, durability="sync")
Control checkpoint durability per run ("sync"/"async"/"exit") — trade write-latency against crash-recovery guarantees.

11Streamingwatch it run

for chunk in graph.stream(inp, cfg, stream_mode="updates"): print(chunk)★
"updates" = each node's delta as it runs; "values" = full state after each step.
graph.stream(inp, cfg, stream_mode="messages")★
"messages" streams LLM tokens (as (token, metadata)) from any model call inside a node — token-by-token UIs.
stream_mode=["updates", "messages", "custom"]
Pass a list to multiplex; each yield is (mode, chunk). "debug" adds full trace events.
from langgraph.config import get_stream_writer get_stream_writer()({"progress": 0.5})
Emit your own "custom" events from inside a node — progress bars, intermediate status.

12Prebuilt: ToolNode & agentsdon't reinvent

from langgraph.prebuilt import ToolNode, tools_condition builder.add_node("tools", ToolNode([search, calc]))★
ToolNode executes the tool calls in the last AIMessage and appends ToolMessages. tools_condition is the ready-made router for the agent loop.
builder.add_conditional_edges("agent", tools_condition)
Drop-in routing: go to "tools" if the model called a tool, else END. Pairs with a model node that does llm.bind_tools([...]).
# langgraph.prebuilt.create_react_agent -> deprecatedmigration
The old prebuilt agent is superseded by langchain.agents.create_agent (its middleware-based successor). For new high-level agents, start there; use raw LangGraph for custom graphs.
from langgraph.graph import StateGraph # subgraph as a node builder.add_node("sub", subgraph.compile())
A compiled graph is itself a Runnable, so it can be a node in a bigger graph — how multi-agent systems compose.

13Store, deploy & ecosystemship it

from langgraph.store.memory import InMemoryStore graph = builder.compile(checkpointer=cp, store=InMemoryStore())★
Long-term memory across threads: a store holds durable, namespaced key–value facts (vs. the per-thread checkpointer). Reach it via get_store().
langgraph dev · langgraph build · langgraph up★
LangGraph CLI (pip install "langgraph-cli[inmem]") runs the dev server + Studio and packages the app for the LangGraph Platform.
export LANGSMITH_TRACING=true
LangSmith traces every graph run automatically once set — step-by-step debugging, evals, monitoring. No code change.
png = graph.get_graph().draw_mermaid_png()
Visualize the compiled graph (Mermaid / PNG / ASCII) to sanity-check nodes & edges.

LangGraph cheat sheet

Outline

Getting started

State

Control flow

Persistence & HITL

Streaming & ecosystem

Getting Started

State

Control Flow

Persistence & Human-in-the-Loop

Streaming & Ecosystem

Worth memorizing