LangGraph — Stateful Workflows & Routing

The mental model: State + Nodes + Edges advanced

You've hit this wall already, even if you didn't have a name for it. A LangChain chain (L2) is a straight pipe: A feeds B feeds C, no branches, no loops, no going back. It's perfect until the day your agent needs to decide — "is this a billing question or a bug?" — or to repeat — "call the tool, look at the result, then think again." A chain can't turn and a chain can't loop, so people reach for the fully prebuilt agent instead, which can do both but hides how behind a single function call. LangGraph is the missing middle: the branching and looping you need, but written out in the open where you can see and change every step.

Here's the one analogy to hold onto: LangGraph is a flowchart made real. You've drawn flowcharts on whiteboards your whole career — boxes for the steps, arrows for "what happens next," diamonds for decisions. LangGraph is that exact picture turned into runnable code, with one addition that makes it powerful: a shared clipboard that every box writes on as work flows through. Get those two ideas — boxes-and-arrows plus one shared clipboard — and the rest of this chapter is just naming the parts.

There are exactly three parts, and they map cleanly onto the flowchart. A LangGraph app is a graph over a single state object — that's the shared clipboard, a plain typed dictionary that holds everything the app knows so far. Each node is one box: a function that reads the state and returns an update to it (never the whole thing — just the fields it touched). Edges are the arrows: they decide which node runs next. Most edges are plain ("always go A→B"), but a conditional edge is the decision diamond — it runs a little function on the current state and picks the next node at runtime. You wire it all up, call compile() to freeze it into a runnable app, then invoke it — and execution flows from START to END, threading the state through every node it passes.

The mistake almost everyone makes first is thinking a node "calls" the next node, the way one function calls another. It doesn't. Nodes never call each other and never reach out to grab the next step — they just read the clipboard, write their update back, and return. The edges are what move execution along; the graph runtime is the thing driving the whole flow. That separation is exactly why LangGraph is testable: a node is a nearly pure function of State → partial State, so you can call it with a fake state in a unit test and check what it returns, with no graph running at all. Keep the flowchart in your head as you read the diagram below — boxes do work, arrows decide flow, and one clipboard ties them together.

Lab L4.1 · Typed state & reducers advanced

Here's a bug that has cost people whole afternoons. They build a chat graph, run it, and the assistant has amnesia — every turn it acts as if the conversation just started. The nodes look correct. The model is fine. The culprit is one missing annotation on the state: without it, each node's new message overwrote the entire history instead of adding to it. This section is about the single mechanism that fixes that — and it's the mechanism people most often skip because it looks like a fussy type hint.

Start with the shape of state. State in LangGraph is a typed dict — a dictionary with a fixed set of named, typed fields, so the graph (and your editor) knows exactly what data flows through. Simple enough. The subtlety is what happens when a node returns a value for one of those fields. By default the rule is dead simple and probably what you'd guess: last write wins. A node returns {"category": "billing"} and the graph drops that value into the category slot, replacing whatever was there. Overwrite. That's perfect for a field like "the current category" — you want the newest value to win.

But now think about the message list. Every turn, a node produces one new message. If "last write wins" applied here, each node would slam its single message into the messages slot and erase every message before it — that's the amnesia bug. What you actually want is append: add the new message to the running list, keep the old ones. That "how do I combine the new value with the old value" decision is called a reducer, and it's the whole idea of this section. A reducer is a rule attached to a specific field that says how a node's update merges into the existing state, instead of just overwriting it.

The analogy that makes it click: think of a bank account. Most fields are like your address — when it changes, the new one simply replaces the old (overwrite). But your balance is different — a $50 deposit doesn't become your balance, it gets added to it (a reducer). LangGraph lets you pick, per field, which kind you want. For the messages list the built-in reducer is add_messages, and attaching it is what turns a pile of stateless node calls into a graph with memory: because messages append, each node can return just its own new message and trust the graph to accumulate the full conversation. The common mistake — worth burning into memory — is forgetting that annotation and then wondering why history keeps vanishing. If a field should grow, give it a reducer; if it should reset each time, leave it plain. The lab below declares both kinds side by side so the contrast is concrete.

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

Lab L4.2 · Build a two-node graph advanced

Enough theory — let's build the smallest graph that does something real. The mental model and the reducers only truly land once you've wired a graph with your own hands and watched data flow through it, so this lab keeps the graph deliberately tiny: a support message comes in, one node classifies it (bug / billing / other), a second node drafts a reply, done. Two nodes, one edge between them, plus the START and END bookends. If you can build this, you can build any graph — everything larger is just more of the same three moves.

Those three moves are worth naming before you read the code, because they're the same every single time. First you declare the state — the typed dict of fields the nodes will read and write. Second you write the nodes as plain functions, each one taking the state and returning a small dict of only the fields it changed. Third you wire the graph: create a StateGraph, register each function with add_node, lay the arrows with add_edge, and call compile() to freeze it into something you can invoke. Build → compile → invoke. That rhythm never changes, whether your graph has two nodes or twenty.

The one habit to form right now — and the thing beginners most often get wrong — is what a node returns. Your instinct from ordinary Python might be to grab the whole state, mutate it, and hand the whole thing back. Don't. A LangGraph node returns only the keys it touched: classify returns {"category": ...} and nothing else, even though the state also has text and reply. The graph merges that little update into the shared state for you (using reducers where you defined them). Returning partial updates is what keeps nodes composable and independently testable — each one is a self-contained "read these fields, produce those fields" unit that doesn't need to know the rest of the graph exists. Notice too that the second node reads state['category'], a value the first node wrote: that's the shared clipboard doing its job, carrying data from one node to the next without either function calling the other. Watch for that hand-off as you read — it's the moment the graph stops being a diagram and starts being a program.

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"])

graph.pyfrom langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

model = ChatOpenAI(model="gpt-5.5", 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"])

Lab L4.3 · Conditional routing expert

This is the section that justifies the whole chapter. Everything so far — state, reducers, nodes, edges — a chain could arguably fake if you squinted. But a chain runs the same steps in the same order every time; it cannot look at what just happened and change course. The moment your app needs to send a billing question one way and a bug report another, or escalate the hard cases to a human, you've left chain territory for good. Conditional routing is how a graph makes that decision at runtime, and it is the single most important primitive in LangGraph.

The idea is smaller than it sounds. A normal edge is hard-wired: "after classify, always go to draft." A conditional edge replaces that fixed arrow with a routing function — a plain Python function that reads the current state and returns the name of the node to go to next. No LLM, no framework magic: just if statements returning strings. If the category is "billing", return "billing_reply"; if it's "bug", return "tech_reply"; otherwise return "escalate_human". The graph runs your function, takes the string it hands back, and jumps to the matching node. One node, many possible exits — but only one is taken per run.

Think of it as a railway switch, or the classic fork-in-the-road. The train arrives at one point; a switch reads the signal and throws it toward exactly one of several tracks. The other tracks exist, but this train won't touch them. That's precisely what happens here: after classify, only the chosen reply node runs — the others are skipped entirely. This shape has names you'll hear constantly: it's the router or supervisor pattern from L1, and it's the skeleton of every triage, dispatch, and "which specialist handles this?" agent you'll ever build. The router function is literally where L1's architecture diagrams turn into executable code.

Two traps to avoid, because both bite in practice. First, always give your router a catch-all: a final return that handles anything you didn't explicitly match — here, escalate to a human. A model that answers with an unexpected category shouldn't crash your graph; it should fall through to a safe default. Second — and this is the classic LangGraph bug — a conditional edge is wired with a mapping dict that translates each string your router can return into a real target node. If your router can return "refund_reply" but that string isn't in the dict (or the node doesn't exist), the graph fails with a "maps to a missing node" error. Every value the router can produce must map to a real node or to END. Keep the router and the dict in sync and conditional routing becomes the reliable workhorse it's meant to be.

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

The ReAct loop as a graph expert

In L3 you called create_react_agent and a working tool-using agent popped out — magic. This is the section where the magic dissolves. Because here's the reveal that surprises people: that prebuilt agent, the one that felt like a black box, is itself just a compiled LangGraph graph. And a small one. Once you've seen it drawn out, you'll never look at a "tool-calling agent" as anything mystical again — it's two nodes and one conditional edge that loops back. That's the entire machine.

Recall the ReAct rhythm from the agents module: think, act, observe, repeat. The model thinks and either answers or asks to use a tool; if it used a tool, you run the tool, feed the result back, and let the model think again with that new information; you keep going until the model is satisfied and answers directly. Now map that onto the two graph primitives you already know. "Think" is a model node that calls the LLM. "Act" is a tools node that runs whatever tools the model requested. And the "did it ask for a tool?" question — the decision that either loops or finishes — is a conditional edge, exactly the routing you just learned, wired to a tiny function that peeks at the last message: if it contains tool calls, go to tools; otherwise go to END.

There's one genuinely new thing here, and it's the star of L5: the graph loops back. After the tools node runs, its edge doesn't go forward — it goes back to the model node. That back-edge is a cycle, and it's the one structure a chain fundamentally cannot have (a chain is a straight line; a line has no loops). The cycle is what lets the agent take as many think-act-observe turns as a problem needs — one tool call or ten — without you hard-coding the number. The model itself decides when to stop by simply not asking for another tool, at which point the conditional edge routes to END.

Watching this makes concrete why you'd ever drop below the convenient prebuilt function. When the agent is just these primitives laid bare, you can reach in and change them: insert a reflection node that critiques the draft before END, add a step_count field to the state to cap runaway loops, or slot in a router to hand off to sub-agents. The prebuilt agent trades that control for convenience; raw LangGraph hands it back. One honest caution as you read the code below — an unbounded cycle can loop forever if the model keeps asking for tools, so real graphs set a recursion limit (that's an L5 topic). For now, focus on recognizing each piece of the diagram in the code: the model node, the ToolNode, the should_continue branch, and the loop-back edge that closes the ring.

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()

Common pitfalls expert

Exercises expert

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'}

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

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?

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

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])

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic      # Claude
# from langchain_openai import ChatOpenAI          # OpenAI
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])   # Claude
# model = ChatOpenAI(model="gpt-5.5").bind_tools([add])            # OpenAI

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"}]})

Knowledge check check yourself