LangGraph — Stateful Workflows & Routing
A chain is a straight line; an agent is a black-box loop. LangGraph is the middle: you model your agent as an explicit graph — a shared state, nodes that update it, and edges (some conditional) that decide what runs next. This chapter builds one from scratch so the prebuilt agent stops being magic.
- an Anthropic API key (
ANTHROPIC_API_KEY) +pip install anthropic
Learning objectives
- Explain LangGraph's model: State, Nodes, Edges, and the compiled graph.
- Define a typed state with reducers (why
messagesappends instead of overwrites). - Build a graph with nodes and normal edges.
- Add conditional edges for routing — the branch a chain can't express.
- Rebuild the ReAct loop as an explicit model↔tools graph.
create_react_agent is a compiled LangGraph graph you didn't have to write. You drop to raw LangGraph when you need control it hides: custom routing, extra state fields, a reflection step, multiple agents, or a human-approval pause (L5). Learn the primitives here; wield them in L5 and the multi-agent module.The mental model: State + Nodes + Edges advanced
A LangGraph app is a graph over a single state object. Each node is a function that reads the state and returns an update to it. Edges decide which node runs next. You compile() the graph, then invoke it — execution flows from START to END, threading the state through.
This is the whole idea of LangGraph in one picture. A LangGraph app is a graph: little boxes (called nodes) joined by arrows (called edges). There is one shared State object — a bag of data — and every node reads it and writes back into it as execution moves left to right.
- The circles START and END are the entrance and exit. Execution always begins at
STARTand finishes atEND— you never call the nodes yourself. - node A is an ordinary node: a function that looks at the State and does some work. The plain arrow into it is a normal edge — it always goes to the same next node.
- After node A the two amber arrows are a conditional edge (labelled conditional): a small function looks at the State and picks whether to go to node B or node C. Only one of them runs. This runtime choice is the thing a straight-line chain cannot do.
- Both branches funnel into
END. The caption line at the bottom is the key rule: every node reads & updates the same shared State object — the State is how nodes pass information to each other.
In short: Think of nodes as workers and edges as arrows on a flowchart. LangGraph just makes that flowchart real code, with one shared clipboard (the State) that every worker writes on.
Lab L4.1 · Typed state & reducers advanced
State is a typed dict. By default a node's returned field overwrites that key. But for things like a message list you want to append — that's what a reducer does: it says how to merge a node's update into the existing state.
state.pyfrom typing import Annotated, TypedDict
from langgraph.graph import add_messages # the append reducer for messages
class State(TypedDict):
messages: Annotated[list, add_messages] # nodes APPEND to this
step_count: int # no reducer → nodes OVERWRITE it
Before you build a graph you must declare its State — the shape of the shared data every node reads and writes. Here State is a TypedDict (a dictionary with named, typed fields). The clever part is the reducer: a rule for how a node's update gets merged into the existing State.
TypedDictjust means "a dictionary with a fixed set of named keys". Here the State has two keys:messagesandstep_count.messages: Annotated[list, add_messages]attaches the reduceradd_messagesto themessagesfield.Annotated[type, extra]is Python's way of tagging a type with extra info — LangGraph reads that tag and uses it.- Because of that reducer, when a node returns a new message the graph APPENDS it to the list instead of replacing the list. That is how the conversation grows turn by turn.
step_counthas no reducer, so it is last-write-wins: whatever a node returns for it simply OVERWRITES the old value. Choosing a reducer per field is how you decide what accumulates versus what resets.
What the output means: Nothing runs yet — this file only declares the State's shape. But it decides the behaviour of the whole graph: messages will grow; step_count will be replaced each time.
Try this: Picture two nodes each returning one message. With add_messages you end up with both messages in the list; without it, the second node's message would erase the first. That single difference is what LangGraph calls "memory".
add_messages is why every node can return just its new message and the graph accumulates the full conversation — the resend-the-history pattern (C2) turned into a state rule. A field with no reducer is last-write-wins. Choosing the reducer per field is how you control what accumulates versus what resets.Lab L4.2 · Build a two-node graph advanced
Start concrete: a graph that classifies a message, then drafts a reply. Two nodes, one edge, plus START/END.
graph.pyfrom langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict
model = ChatAnthropic(model="claude-opus-4-8", max_tokens=512)
class State(TypedDict):
text: str
category: str
reply: str
def classify(state: State) -> dict:
r = model.invoke(f"One word — bug, billing, or other:\n{state['text']}")
return {"category": r.content.strip().lower()} # update to state
def draft(state: State) -> dict:
r = model.invoke(f"Write a short reply to a {state['category']} ticket:\n{state['text']}")
return {"reply": r.content}
g = StateGraph(State)
g.add_node("classify", classify)
g.add_node("draft", draft)
g.add_edge(START, "classify")
g.add_edge("classify", "draft")
g.add_edge("draft", END)
app = g.compile()
out = app.invoke({"text": "I was double charged."})
print(out["category"], "|", out["reply"])
This is your first real graph: a tiny pipeline that classifies a support message and then drafts a reply. Read it in three parts — the State, the two node functions, then the wiring that turns them into a runnable graph.
- The State (
class State(TypedDict)) has three fields:text(the incoming message),category, andreply. The nodes will fill these in as data flows through. - node
classifyasks the model for one word (bug / billing / other) and returns{"category": ...}— just the one key it changed. It does not return the whole State; LangGraph merges that small update in for you. - node
draftreadsstate['category'](set by the previous node) and writes areply. This is nodes passing data through the shared State. - The wiring:
StateGraph(State)creates the graph;add_noderegisters each function;add_edgelays the arrowsSTART → classify → draft → END.g.compile()freezes it into a runnableapp, andapp.invoke({...})runs it once with a starting State.
What the output means: Prints the model's guess and its drafted reply, e.g. billing | We're sorry about the double charge…. Note the two nodes ran in order because a plain edge always goes classify → draft.
Try this: Add a print(state) as the first line of draft and re-run: you'll see category is already filled in, proving the classify node's update landed in the shared State before draft ran.
State → partial State.Lab L4.3 · Conditional routing expert
Now the payoff a chain can't give you: pick the next node based on state. A routing function inspects the state and returns the name of the edge to follow.
This diagram shows the payoff a straight chain can't give you: one node with many exits. After classify figures out the category, a conditional edge sends the work to exactly one of three different reply nodes.
- On the left, the classify node produces a category. The three arrows leaving it are a single conditional edge — labelled route by category — not three normal edges.
- A little routing function looks at the State's
categoryand returns a name. That name decides which one arrow is actually followed this run. - billing_reply, tech_reply, and escalate_human are the three possible destinations. Only the chosen one runs; the other two are skipped entirely.
- The colours are just hints:
escalate_human(red) is the "send to a person" path — routing lets you peel off the hard cases automatically.
In short: This is the router / supervisor pattern: a decision step in the middle of a graph that picks the branch. It is the skeleton of every triage or dispatch agent.
Setup to run this snippet
class _Any:
'''stands in for any undefined demo value; supports call/attr/index/
iteration and basic arithmetic (as 0.7) so demo snippets run.'''
def __call__(self, *a, **k): return _Any()
def __getattr__(self, k): return _Any()
def __getitem__(self, k): return _Any()
def __iter__(self): return iter([])
def __len__(self): return 0
def __contains__(self, o): return True
def __enter__(self, *a): return _Any()
def __exit__(self, *a): return False
def __float__(self): return 0.7
def __int__(self): return 1
def __lt__(self, o): return True
def __gt__(self, o): return False
def __le__(self, o): return True
def __ge__(self, o): return False
def __add__(self, o): return o
def __radd__(self, o): return o
def __bool__(self): return True
def __repr__(self): return 'demo'
def __str__(self): return 'demo'
State = _Any()
class _g_t:
add_conditional_edges = 'demo'
def add_conditional_edges(self, *a, **k): return 'demo'
def __getattr__(self, k): return 'demo'
g = _g_t()routing.pydef route(state: State) -> str:
"""Return the name of the next node based on the category."""
if state["category"] == "billing": return "billing_reply"
if state["category"] == "bug": return "tech_reply"
return "escalate_human"
g.add_conditional_edges("classify", route, {
"billing_reply": "billing_reply",
"tech_reply": "tech_reply",
"escalate_human": "escalate_human",
}) # the dict maps route()'s return value → target node
Here is the code behind that routing diagram. Two pieces work together: a plain routing function that decides, and add_conditional_edges that wires its decision to real nodes.
route(state)is an ordinary function that readsstate["category"]and returns a string — the name of the branch to take. No LLM, no magic; justifstatements returning labels like"billing_reply".- The final
return "escalate_human"is the default: anything that isn't billing or bug falls through to the human-escalation path. Always give a router a catch-all. g.add_conditional_edges("classify", route, { ... })attaches the router to theclassifynode. The dictionary is a lookup table: it maps each stringroute()can return to the actual target node to jump to.- So the flow is: run
classify→ callroute(state)→ look its return value up in the dict → go to that node. That indirection is what lets one node have several possible next steps.
What the output means: No output on its own (this snippet only wires the edges). At run time it silently steers each ticket to one reply node — the branch you'd otherwise need a human to choose.
Try this: Add a new category like "refund": return "refund_reply" from route and add both a matching node and a "refund_reply": "refund_reply" entry to the dict. Forgetting the dict entry is the classic "maps to a missing node" bug.
The ReAct loop as a graph expert
Now L3's prebuilt agent stops being magic. It's two nodes and one conditional edge that loops back:
create_react_agent, unpacked. A model node and a tools node, with a conditional edge: if the model asked for tools, go run them and loop back; otherwise go to END. That back-edge — a cycle — is exactly what L5 explores next.
This is L3's prebuilt agent with the lid off. An agent that can use tools is just two nodes and one conditional edge that loops back — and this diagram is that whole machine.
- From
STARTwe go to the model node: it calls the LLM, which either answers directly or asks to use a tool. - The conditional edge in the middle asks tool_calls? — did the model request any tools? If yes, the arrow goes right to the tools node, which actually runs those tools.
- The dashed arrow loop back with results is the important one: after running tools we go back to the model so it can read the results and decide what to do next. That back-arrow is a cycle — a loop a plain chain can't have.
- If instead there were no tool_calls, the model is finished, so that arrow goes straight to
END. The loop keeps spinning (model → tools → model) until the model stops asking for tools.
In short: This model↔tools loop is the ReAct agent: think, act with a tool, observe the result, think again — repeat until done. L5 explores that back-edge (the cycle) in depth.
Requires: pip install langgraph
react_graph.pyfrom langgraph.graph import StateGraph, START, END, add_messages
from langgraph.prebuilt import ToolNode
from typing import Annotated, TypedDict
class S(TypedDict):
messages: Annotated[list, add_messages]
model_with_tools = model.bind_tools([get_weather, add])
def call_model(s): return {"messages": [model_with_tools.invoke(s["messages"])]}
def should_continue(s):
return "tools" if s["messages"][-1].tool_calls else END
g = StateGraph(S)
g.add_node("model", call_model)
g.add_node("tools", ToolNode([get_weather, add]))
g.add_edge(START, "model")
g.add_conditional_edges("model", should_continue) # branch: tools or END
g.add_edge("tools", "model") # the loop back
app = g.compile()
ToolNode (LangGraph's prebuilt tool executor), a model node, and a cycle. Now you can add anything: a reflection node before END, a step counter in state, a router to sub-agents. That freedom is the reason to drop below create_react_agent.And here is that agent as code — remarkably short. It is the same three moves as the diagram: define a tiny State, register a model node and a tools node, then add the conditional edge and the loop-back edge.
- The State
Sholds onlymessageswith theadd_messagesreducer — so every model reply and tool result appends to the running conversation (the reducer from Lab L4.1 doing its job). call_modelis the model node: it sends the current messages to the tool-aware model and returns its reply as a new message.ToolNode([...])is LangGraph's prebuilt node that runs whatever tools the model asked for — you don't write it.should_continueis the router: it peeks at the last message (s["messages"][-1]) and returns"tools"if the model requested tool calls, otherwiseEND. This is the tool_calls? decision from the diagram.- The wiring makes the cycle:
START → model, then a conditional edge frommodel(branch to tools or END), andadd_edge("tools", "model")— the loop back — sends tool results to the model to think again.
What the output means: After compile(), app is a working ReAct agent: invoke it with a question and it will call tools and loop until it can answer without them.
Try this: Compare this to the diagram box by box — model, tools, the should_continue branch, the loop-back edge. Everything create_react_agent hid is right here, which is exactly why dropping to raw LangGraph gives you control.
Common pitfalls expert
| Pitfall | Fix |
|---|---|
Forgetting the reducer on messages | Use Annotated[list, add_messages] or each node overwrites history |
| Returning the whole state from a node | Return only the keys you changed |
| Using a chain when you need branching/loops | That's exactly what LangGraph is for — model it as a graph |
| Conditional edge mapping to a missing node | Every return value must map to a real node or END |
| A cycle with no exit condition | Ensure the router can reach END; set a recursion limit (L5) |
| Over-engineering a linear task into a graph | If it's a straight line, a chain (L2) is simpler |
Exercises expert
Exercise L4.1 — Triage graph
Context: A routing graph that runs only the matching branch is the canonical use of conditional edges.
Your task: Build a triage graph end to end: a classify node and three reply nodes wired with a conditional edge, and test all three routes.
Requirements:
- One classify node plus three reply nodes
- A conditional edge routes to the matching reply node
- Exercise all three routes
- Confirm only the matching node runs for each input
💡 Hint: The classify node writes a category into state; the conditional edge maps each category to its reply node.
Exercise L4.2 — Add a reflection node
Context: Inserting a reflect node turns L1's reflection architecture into a graph — a cycle that must be bounded so it can't loop forever.
Your task: Take the ReAct graph and insert a reflect node between the model and END that critiques the draft, routing back to the model if weak or to END if good.
Requirements:
- Add a quality signal to the state that the reflect node sets
- A conditional edge routes 'revise' back to the model or 'ok' to END
- Cap the number of revisions with a counter
- Confirm a weak draft triggers a revision and a good one ends
💡 Hint: The reflect node writes a quality field; the edge reads it; a counter guarantees termination even if quality never clears.
Show hint
Add a quality field to state. The reflect node sets it; a conditional edge routes "revise" → "model" or "ok" → END. Cap revisions with a counter so it can't loop forever.
Exercise L4.3 — Visualize it
Context: Seeing your code rendered as a graph is the fastest way to catch a mis-wired edge.
Your task: Call get_graph().draw_mermaid() (or the ASCII variant) on your triage graph and compare the rendered diagram to the lab's diagram.
Requirements:
- Render your compiled graph as a diagram
- Compare it to the reference diagram from the lab
- Look for a mis-wired or missing edge
- Confirm the structure matches what you intended
💡 Hint: A wrong or missing edge jumps out visually far faster than it does by reading the builder code.
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: In LangGraph a node reads the shared State and returns an update that gets merged in — it never mutates the world directly, which is what makes runs reproducible.
Your task: Model a graph as a dict State plus node functions that each return a partial update.
Requirements:
- Each node takes the state and returns a partial update dict
- The driver merges each update into the shared state
- Nodes don't mutate the state in place
- Show two nodes running in sequence, the second seeing the first's update
💡 Hint: Return-then-merge, never mutate; the merge step is what a real StateGraph does between nodes.
Show solution
Nodes do work by returning partial state updates. Runnable stdlib:
def node_a(state):
return {"count": state.get("count", 0) + 1}
def node_b(state):
return {"label": f"count is {state['count']}"}
state = {}
for node in (node_a, node_b):
update = node(state)
state.update(update) # merge the update into shared state
print(state) # {'count': 1, 'label': 'count is 1'}
A node never mutates the world directly — it returns an update, and the graph merges it. That is what makes runs reproducible.
Context: Some state fields overwrite; others — like messages — append. That append behavior is a reducer, and it's exactly what add_messages is.
Your task: Model a state merge where messages uses an add-reducer and other keys overwrite.
Requirements:
- Register a reducer for
messagesthat appends - Keys with no reducer overwrite by default
- Merging a messages update grows the list rather than replacing it
- Show a non-message key being overwritten in the same merge
💡 Hint: A per-key reducer table decides how an update combines with the old value; without it, each node would clobber the conversation.
Show solution
A reducer decides how an update combines with the old value. Runnable:
REDUCERS = {"messages": lambda old, new: old + new} # append, don't replace
def merge(state, update):
out = dict(state)
for k, v in update.items():
if k in REDUCERS:
out[k] = REDUCERS[k](out.get(k, []), v)
else:
out[k] = v # default: overwrite
return out
s = {"messages": [{"role": "user", "content": "hi"}], "step": 1}
s = merge(s, {"messages": [{"role": "ai", "content": "hello"}], "step": 2})
print(len(s["messages"]), "messages; step =", s["step"]) # 2 messages; step = 2
Without the reducer, each node would clobber the conversation. add_messages is exactly this append-reducer.
Context: A normal edge always goes A→B; a conditional edge runs a function on the state to pick the next node — the branch a linear chain simply can't express.
Your task: Model a graph with a router that branches on state.
Requirements:
- A node classifies/updates the state
- A routing function reads the state and returns the next node's name
- Different states route to different nodes
- Show at least two inputs taking different branches
💡 Hint: The router is a plain function state→node-name; runtime routing on state is the reason to drop from a chain to a graph.
Show solution
Conditional edges are branch logic over state. Runnable:
def classify(state):
return {"kind": "refund" if "refund" in state["q"].lower() else "general"}
def route(state):
return "refund_node" if state["kind"] == "refund" else "general_node"
NODES = {
"refund_node": lambda s: {"answer": "Refunds take 30 days."},
"general_node": lambda s: {"answer": "How can I help?"},
}
def run(q):
state = {"q": q}
state.update(classify(state))
nxt = route(state) # conditional edge picks the node
state.update(NODES[nxt](state))
return state["answer"]
print(run("I want a refund")) # Refunds take 30 days.
print(run("hello")) # How can I help?
Routing on state at runtime is the branch a linear chain simply cannot express — the reason to drop to a graph.
Context: ReAct expressed as an explicit graph is a model node, a tools node, and a conditional edge that loops back or ends — which is precisely what create_react_agent compiles.
Your task: Model the compiled ReAct graph's execution offline.
Requirements:
- A model node either requests a tool or produces a final answer
- A tools node executes the requested tool and records the observation
- A
should_continueedge loops back to the model or routes to END - Drive the node transitions manually and bound the steps
- Show it returning the right answer
💡 Hint: Three pieces — model node, tools node, and one conditional edge — loop until the model stops asking for tools.
Show solution
ReAct = model node + tools node + a should_continue edge. Runnable:
TOOLS = {"add": lambda a, b: a + b}
END = "__end__"
def model_node(state):
if not state.get("tool_ran"):
return {"next_call": ("add", (2, 3))} # ask for a tool
return {"answer": state["last_obs"], "next_call": None}
def tools_node(state):
name, args = state["next_call"]
return {"last_obs": TOOLS[name](*args), "tool_ran": True}
def should_continue(state):
return END if state.get("next_call") is None else "tools"
def run():
state = {}
node = "model"
for _ in range(10):
if node == "model":
state.update(model_node(state))
node = should_continue(state)
elif node == "tools":
state.update(tools_node(state))
node = "model"
elif node == END:
return state["answer"]
return "recursion limit"
print(run()) # 5
This is what create_react_agent compiles: two nodes and one conditional edge that loops until the model stops asking for tools.
Context: Before you compile a graph it should be structurally sound: START and END reachable, and no edge pointing at an undefined node. LangGraph's compile() runs this check for you.
Your task: Write a validator over an adjacency map that catches dead edges and an unreachable END before 'compile'.
Requirements:
- Flag any edge target that isn't a defined node or END
- Verify END is reachable from START (e.g. via BFS/DFS)
- Return whether the graph is valid plus the list of problems
- Show it passing a good graph and failing a broken one
💡 Hint: Collect every referenced node, check each target exists, then traverse from START to confirm END is reachable — that's what compile validates.
Show solution
Catch dead edges and unreachable ends statically. Runnable:
def validate(edges, start="START", end="__end__"):
nodes = set(edges) | {t for ts in edges.values() for t in ts}
problems = []
# every edge target must be a real node or END
for src, targets in edges.items():
for t in targets:
if t not in edges and t != end:
problems.append(f"{src} -> undefined node '{t}'")
# END must be reachable from START (BFS)
seen, stack = set(), [start]
while stack:
n = stack.pop()
if n in seen: continue
seen.add(n)
stack += list(edges.get(n, []))
if end not in seen:
problems.append("END not reachable from START")
return (not problems), problems
g = {"START": ["model"], "model": ["tools", "__end__"], "tools": ["model"]}
print(validate(g)) # (True, [])
print(validate({"START": ["gone"]})) # (False, [... undefined ..., END not reachable])
LangGraph's compile() does this structural check for you; modeling it shows what "compile" is really validating.
Context: The production graph is a typed State with an add-messages reducer, a model node, a tools node, and a conditional edge — the same State+Nodes+Edges you modeled offline, now with the real reducer and Claude.
Your task: Write the real StateGraph using documented langgraph APIs. (Needs the libraries installed.)
Requirements:
- Define a typed State with
messages: Annotated[list, add_messages] - Add a model node that calls a Claude model bound to tools
- Add a
should_continueconditional edge to a tools node or END - Wire START → model and compile the builder
- Label it as needing
langgraph/langchain-anthropic+ a key; note aToolNodefills the tools node
💡 Hint: Same State/Nodes/Edges as the offline model; add_messages is the append reducer and bind_tools lets the model emit calls — verify imports against your version.
Show solution
Correct StateGraph. Needs pip install langgraph langchain-anthropic + ANTHROPIC_API_KEY:
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
@tool
def add(a: int, b: int) -> int:
"Add two integers."
return a + b
class State(TypedDict):
messages: Annotated[list, add_messages] # the append reducer
model = ChatAnthropic(model="claude-opus-4-8").bind_tools([add])
def call_model(state: State):
return {"messages": [model.invoke(state["messages"])]}
def should_continue(state: State):
last = state["messages"][-1]
return "tools" if getattr(last, "tool_calls", None) else END
builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph = builder.compile()
# graph.invoke({"messages": [{"role": "user", "content": "add 2 and 3"}]})
Same State+Nodes+Edges you modeled offline, now with the real reducer and Claude. A ToolNode would fill the "tools" node. Verify imports against your version.
✓ Checkpoint — you can move on when you can…
- Explain State, Nodes, Edges, and the compile→invoke flow.
- Define typed state and say what a reducer does (and why
messagesneeds one). - Build a multi-node graph where nodes return partial updates.
- Add a conditional edge that routes by state.
- Draw the ReAct agent as a two-node graph with a cycle.
diagnose node, a conditional edge that routes read-only findings straight to a report but risky fixes through a safety_gate node, and a loop back to re-diagnose after acting. The routing function is the safety policy. L5 adds the human-approval pause that makes that gate real. See the safety-gate build →Knowledge check check yourself
What does a reducer like add_messages do to a state field, and how does a field with no reducer behave?
Show answer
add_messages makes the messages field append new messages instead of overwriting — which is what gives the graph its accumulating conversation "memory." A field with no reducer is last-write-wins: a node's return value simply overwrites the old value.What can a conditional edge do that a normal edge (and a linear chain) cannot?