When your agent has a shape, draw the shape. Graphs, nodes, edges, checkpoints.
Agents Module 3 LangGraph StateGraph
LangGraph is a Python library for building agent control flow as an explicit state graph.
You declare:
The runtime walks the graph, persists state at every step, and can resume from any checkpoint.
anthropic directly from nodes.StateGraph(State)A typed graph builder. State is usually a TypedDict (or Pydantic model) describing what flows between nodes.
Plain Python functions (state) -> dict-of-state-updates. The runtime merges your return dict into the global state.
add_edge(a, b) for “always go a → b” and add_conditional_edges(a, decide, mapping) for “after a, call decide(state) and route to one of mapping's targets.”
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class S(TypedDict):
n: int
log: list[str]
def add_one(s: S) -> dict:
return {"n": s["n"] + 1, "log": s["log"] + ["add_one"]}
def double(s: S) -> dict:
return {"n": s["n"] * 2, "log": s["log"] + ["double"]}
g = StateGraph(S)
g.add_node("add_one", add_one)
g.add_node("double", double)
g.add_edge(START, "add_one")
g.add_edge("add_one", "double")
g.add_edge("double", END)
app = g.compile()
print(app.invoke({"n": 3, "log": []}))
# {'n': 8, 'log': ['add_one', 'double']}
Three nodes (START, add_one, double, END), two edges, one compile. No LLM in the hello world on purpose — get the shape first.
| Chain (linear) | Graph (today) |
|---|---|
| Always A → B → C | Branch on state after B |
| One path through | Loops, retries, conditional skips |
| No checkpoint surface | Persist state at every node |
| Hard to draw | Trivially renderable as a DAG |
def route(s: S) -> str:
return "double" if s["n"] > 10 else "add_one"
g.add_conditional_edges(
"add_one",
route,
{"add_one": "add_one", "double": "double"},
)
g.add_edge("double", END)
After add_one, call route(state); the string it returns is keyed into the mapping to pick the next node. This is how branching, looping, and early-exit are all expressed.
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")
app = g.compile(checkpointer=memory)
app.invoke({"n": 3, "log": []}, config={"configurable": {"thread_id": "run-1"}})
Pass a checkpointer to .compile() and a thread_id per run, and every node transition writes the merged state to sqlite.
Crash, restart, resume — state is on disk.
.invoke({}, config={...}) with the same thread_id, the graph picks up at the last completed node.app.get_state_history(...)) and re-run from there.interrupt_before=["approve"] to compile; the graph stops before the named node and waits for an external resume call.Sweet spot.
anthropic directly so tool functions port verbatim.@tool-decorated functions as plain functions called from nodes.SqliteSaver — resume, time-travel, human-in-the-loop.