Why a graph, not a chain

LCEL chains are linear — data flows one way, start to finish. Real agents aren't linear. They need to:

• Loop — call a tool, look at the result, maybe call another tool… (a cycle, which a chain can't express).
• Branch — take different paths depending on intermediate state (route to a tool vs. answer vs. ask a human).
• Share state — accumulate messages, tool results, and flags that many steps read and write.
• Pause and resume — stop for human approval, then continue exactly where it left off.

LangGraph models the agent as a graph: nodes (steps) connected by edges (transitions) operating over a single shared state object. It's the difference between a script and a state machine — and production agents are state machines.

The StateGraph

Three pieces: a state schema (what's shared), nodes (functions that update state), and edges (how control moves). Here's a minimal two-node graph:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):          # the shared state every node reads/writes (like a TS interface)
    question: str
    answer: str

def retrieve(state: State) -> dict:                  # a NODE: takes state, returns the fields to update
    docs = search(state["question"])                 # (your retriever from the RAG track)
    return {"answer": f"context: {docs}"}            # LangGraph MERGES this into the shared state

def generate(state: State) -> dict:                  # second node: uses what retrieve wrote
    reply = llm(state["question"], state["answer"])  # generate from question + retrieved context
    return {"answer": reply}                          # overwrite `answer` with the final text

graph = StateGraph(State)                             # build the graph over the State schema
graph.add_node("retrieve", retrieve)                  # register nodes by name
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")                     # entry: START -> retrieve
graph.add_edge("retrieve", "generate")                # retrieve -> generate
graph.add_edge("generate", END)                       # generate -> END (done)
app = graph.compile()                                 # compile into a runnable

app.invoke({"question": "What does the Pro plan cost?"})   # runs the graph, returns the final state

That's RAG as a graph — overkill for two linear steps (a chain is simpler), but it's the scaffold everything else hangs on.

Conditional edges: branching

A conditional edge picks the next node by running a function on the current state. This is how a graph routes — e.g. answer directly for a simple question, or go to a tool for a hard one:

from typing import Literal

def route(state: State) -> Literal["tool", "answer"]:   # returns the NAME of the next node
    if needs_lookup(state["question"]):                 # decide based on state
        return "tool"
    return "answer"

graph.add_conditional_edges("classify", route, {       # after "classify", call route() to choose…
    "tool": "call_tool",                                # …if route returns "tool" -> go to call_tool node
    "answer": "direct_answer",                          # …if "answer" -> go to direct_answer node
})

The routing function is plain Python reading plain state — fully testable, no magic. This is the branching a linear chain fundamentally can't do.

Cycles: the agent loop as a graph

The agent loop from Topics 1–3 is a cycle: an "agent" node decides an action, a "tools" node runs it, and an edge loops back — until a condition says stop. LangGraph expresses this directly:

def should_continue(state: State) -> Literal["tools", "end"]:
    last = state["messages"][-1]                        # the model's most recent message
    return "tools" if last.tool_calls else "end"        # tool calls pending? loop. Otherwise finish.

graph.add_node("agent", call_model)                     # the model decides (may emit tool calls)
graph.add_node("tools", run_tools)                      # execute the tool calls
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})  # branch after agent
graph.add_edge("tools", "agent")                        # THE CYCLE: after tools, go back to the agent
app = graph.compile()

The tools -> agent edge is the reason→act→observe loop you hand-wrote, now as an explicit, inspectable cycle with the stop condition as a first-class edge. LangGraph enforces a recursion limit so the cycle is bounded by default.

Durability & human-in-the-loop

Two production features fall out of the graph model almost for free:

Checkpointing. Attach a checkpointer and the graph persists its state after every node, keyed by a thread id. A crash, a restart, or a days-long pause resumes from the last checkpoint — essential for long-running or interruptible agents.

from langgraph.checkpoint.memory import MemorySaver     # a demo checkpointer (use a DB-backed one in prod)
app = graph.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "session-1"}}      # which conversation/run this belongs to
app.invoke({"question": "..."}, config=cfg)             # state is saved after each node under this thread

Human-in-the-loop. Because state is checkpointed, the graph can interrupt before a sensitive node (sending an email, spending money, deleting data), surface the pending action for approval, and resume on a human's go-ahead:

# Pause BEFORE the "send_email" node so a human can approve the drafted action:
app = graph.compile(checkpointer=MemorySaver(), interrupt_before=["send_email"])
app.invoke(initial, config=cfg)          # runs until it hits send_email, then STOPS and returns
# … a human reviews the drafted email in state …
app.invoke(None, config=cfg)             # resume: pass None to continue from the checkpoint through send_email

This is the pattern behind every "the agent drafted this — approve to send?" workflow, and it's why checkpointing matters beyond crash recovery.

When to reach for LangGraph

Same discipline as always: don't over-build. A linear task is a chain; a simple tool loop is fine with an AgentExecutor. Reach for LangGraph when you need explicit branching, cycles you must control, durable state across pauses, or human-in-the-loop gates — i.e. when the agent is a real state machine you need to inspect, checkpoint, and steer.

Pitfalls

• Graphifying a linear task. If there's no branching/cycle/state, a chain is simpler — don't add a graph for ceremony.
• Unbounded cycles. Keep the recursion limit and a clear stop edge; a cycle with a broken condition spins.
• In-memory checkpointer in production. MemorySaver is volatile — use a DB-backed checkpointer for real durability.
• Fat state objects. Put only what nodes actually share in state; dumping everything in makes runs hard to reason about.
• No human gate on irreversible actions. Interrupt before send/pay/delete — don't let an agent take one-way actions unsupervised.

Recap

• Agents need graphs, not linear chains: branching, cycles, shared state, and pause/resume.
• A StateGraph = a typed state + nodes (state -> partial update) + edges; conditional edges branch on state and a cycle expresses the agent loop.
• Checkpointing persists state per thread for durability; interrupts enable human-in-the-loop approval before sensitive nodes.
• Reach for LangGraph when you need real control (branching/cycles/durability/HITL) — otherwise a chain or AgentExecutor is simpler.
• Worked examples map to the exercises: a StateGraph (Ex 1), conditional routing (Ex 2), the agent↔tools cycle (Ex 3), and an interrupt-based approval gate (Ex 4).

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    answer: str

def retrieve(state):
    print("after retrieve sees:", state)
    return {"answer": f"[context for: {state['question']}]"}

def generate(state):
    print("generate sees:", state)     # note it can read what retrieve wrote
    return {"answer": state["answer"] + " -> final answer"}

g = StateGraph(State)
g.add_node("retrieve", retrieve); g.add_node("generate", generate)
g.add_edge(START, "retrieve"); g.add_edge("retrieve", "generate"); g.add_edge("generate", END)
app = g.compile()
print(app.invoke({"question": "Pro plan price?"}))

from typing import Literal

def classify(state): return {}          # no state change; it's just a branch point
def route(state) -> Literal["short", "long"]:
    return "short" if len(state["question"]) < 40 else "long"

g.add_node("classify", classify)
g.add_node("direct_answer", lambda s: {"answer": "quick answer"})
g.add_edge(START, "classify")
g.add_conditional_edges("classify", route, {"short": "direct_answer", "long": "retrieve"})

from typing import Literal

def should_continue(state) -> Literal["tools", "end"]:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else "end"

g.add_node("agent", call_model)      # returns {"messages": [ai_message]}
g.add_node("tools", run_tools)       # executes tool calls, returns {"messages": [tool_results]}
g.add_edge(START, "agent")
g.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
g.add_edge("tools", "agent")          # the cycle
app = g.compile()

from langgraph.checkpoint.memory import MemorySaver

app = g.compile(checkpointer=MemorySaver(), interrupt_before=["send_email"])
cfg = {"configurable": {"thread_id": "t1"}}

app.invoke({"question": "email the client the quote"}, config=cfg)   # runs, then STOPS before send_email
state = app.get_state(cfg)                                            # inspect the pending/drafted action
print("paused, about to run:", state.next)                           # -> ('send_email',)

# A human approves here…
app.invoke(None, config=cfg)      # pass None to resume from the checkpoint through send_email