Create a stateful tool-using agent with nodes, edges, and checkpoints.
Define a typed state schema using TypedDict and wire it into a StateGraph so your agent has a single source of truth. You'll model the running example — a research agent that tracks a goal, message history, and tool results — as a concrete TypedDict.
Defines a typed state schema with TypedDict and wires it into a StateGraph so the agent has a single shared source of truth.
Why this matters: Every node in your stateful tool-using agent reads from and writes to this schema — getting it right is the foundation everything else builds on.
Your research agent will call tools, accumulate results, and loop — but how does each step know what happened before it?
In LangGraph, every reads from and writes to a single shared . The runtime owns that state object and passes it into each node on every step.
You define the schema once as a TypedDict — a plain Python dict with declared field types. That declaration is the contract every node must honour.
The decision this module forces: what belongs in state, and what should a node compute on the fly?
Put a field in state when another node downstream needs it, or when you need it to survive across loop iterations. Leave it out when it's a local intermediate value that only one node uses and never needs to be inspected or persisted.
goal — the research question. Set once, read by every node. → state fieldmessages — the growing conversation history. Accumulated across turns. → state fieldtool_results — raw outputs from tool calls. The summariser node needs them. → state fieldformatted_prompt — a string assembled inside the agent node from goal + messages. Only that node uses it. → compute on the flyA bloated state schema slows checkpointing and makes the graph harder to reason about. Err toward fewer fields; add one only when a second node genuinely needs it.
from langgraph.graph import StateGraph # Naive: pass a plain dict as the schema graph = StateGraph({"goal": str, "messages": list}) graph.compile()
StateGraph({...})The obvious first attempt is to hand a plain dict literal to . This looks reasonable but fails immediately.
TypeError: Expected a TypedDict class or a dataclass, got <class 'dict'>.
StateGraph requires a class (a TypedDict or dataclass) so it can introspect field names and types at compile time. A plain dict literal has no field metadata — the runtime can't build its internal state channels from it.
from typing import TypedDict, Annotated from langgraph.graph import StateGraph from langgraph.graph.message import add_messages class ResearchState(TypedDict): goal: str messages: Annotated[list, add_messages] tool_results: list[dict] graph = StateGraph(ResearchState) print(graph.nodes) # {} — no nodes yet
TypedDictAnnotated[list, add_messages]add_messagesStateGraph(ResearchState)This is the working schema for the running example. goal is a plain replacement field; messages uses the add_messages so new messages are appended, not overwritten.
{} — an empty dict. Nodes are registered separately with graph.add_node(...) after the graph is constructed. The StateGraph constructor only wires up the state channels; it doesn't know about nodes yet.
A returns a — a dict containing only the keys it changed. The runtime merges that dict into the current state.
For plain fields (goal, tool_results), the new value replaces the old one. For fields annotated with a (like messages), the runtime calls reducer(current, new) and stores the result.
messages grows across every node that appends to it.from typing import TypedDict, Annotated from langgraph.graph import StateGraph from langgraph.graph.message import add_messages class ResearchState(TypedDict): goal: str messages: Annotated[list, add_messages] tool_results: list[dict] # TODO: add a `status` field (str) that tracks # whether the agent is "researching" or "done". # Should it use a reducer, or plain replacement? graph = StateGraph(ResearchState)
status: strStop — attempt the TODO before revealing. The crux is deciding whether status needs a reducer or plain replacement, and why.
Hints: only one node sets the final status; no accumulation is needed; the last writer should win.
Replace the TODO with:
status: str
Changed lines: one field added, no Annotated wrapper.
Why no reducer: status is a single scalar that the agent node overwrites on each transition ("researching" → "done"). Only one node writes it, and you always want the latest value — accumulation would be wrong. Plain replacement is correct. Use a reducer only when multiple nodes contribute to a field and you need to merge their contributions.
Three failure modes to watch for — each with a concrete symptom:
{"messages": [new_msg]}. Without Annotated[list, add_messages], the runtime replaces the list — your agent sees only the latest message, never the history. The graph runs silently; no error is raised.{"tool_result": ...} (missing the 's'). The runtime raises InvalidUpdateError: Key 'tool_result' not found in schema at runtime, not at graph construction time.tool_results: list = [] as a class-level default shares one list across all graph runs. Symptom: results from a previous call bleed into the next one. Always pass initial state explicitly to graph.invoke({"goal": ..., "messages": [], "tool_results": []}).When an AI tool generates your ResearchState and wiring, check these four things before running it:
Annotated[list, add_messages] or a custom reducer — not a bare list.[] or {}) are set at the class level — initial state is passed to graph.invoke().StateGraph(ResearchState), not StateGraph(ResearchState()).With a verified schema in place, the next module adds the first real — the agent node that calls the LLM and returns a partial state update back into this same ResearchState.
Write the agent node that calls the LLM and returns a partial state update, then add a second node stub for tool execution. The running example adds an 'agent' node that reads the message list, calls the model, and appends the model's response — leaving the tool node signature for you to complete.
Write the agent node function that calls the LLM and returns a partial state update, then register it alongside a tool node stub on the StateGraph.
Why this matters: Nodes are the units of work in your stateful agent — getting their signature and return type right is the prerequisite for every edge, router, and tool call that follows.
Decision this forces: Should a single node handle both reasoning and tool dispatch, or should those be separate nodes?
A merges a node's into the existing — it appends to the messages list rather than replacing it. That's why nodes only need to return the keys they changed.
Now the question this module answers: once you have a wired to that schema, how do you write the functions that actually run inside it?
A is a plain Python function with the signature state → dict: it receives the full current state and returns only the keys it wants to change.
Returning a — not the full state — keeps nodes decoupled. Each node is responsible for exactly one slice of the state, so two nodes can update different keys without clobbering each other.
The runtime merges the returned dict into the shared state using the you declared on each field. You never touch the full state object directly.
Your research agent receives a user question. The needs to: read the current message history, pass it to the LLM, and append the model's reply.
The node does NOT decide which tool to call next — that's the router's job (module 3). It only produces the model's response and returns {"messages": [ai_message]}. The reducer appends that single message to the existing list.
A second stub sits alongside it. It will eventually execute the the model requested and return a — but for now it's just a placeholder so the graph compiles.
# Naive attempt — returns the FULL state, not a partial update def agent_node(state): messages = state["messages"] ai_message = model.invoke(messages) state["messages"].append(ai_message) # mutates in place return state # returns full state
state["messages"].append(...)return stateThis looks reasonable but breaks in two ways — predict what goes wrong before reading on.
Two problems: (1) Mutating the shared state dict directly bypasses the reducer, so the append may be applied twice — once by your code and once by the runtime merge — producing duplicate messages. (2) Returning the full state object means every key is treated as 'changed', which can overwrite keys other nodes updated in the same step. The fix: return only {"messages": [ai_message]} and let the reducer handle the merge.
def agent_node(state: AgentState) -> dict: ai_message = model.invoke(state["messages"]) return {"messages": [ai_message]} # partial update only def tool_node(state: AgentState) -> dict: # stub — tool execution wired in module 3 return {} graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node)
state: AgentState-> dictmodel.invoke(state["messages"])graph.add_node("agent", agent_node)return {}Each node returns only the keys it owns. agent_node returns {"messages": [ai_message]} and the reducer appends it; tool_node returns an empty dict so the graph compiles without error.
Both nodes are registered with add_node before any edges are added — the graph needs to know every node exists before it can route between them.
graph.nodes returns {"agent": agent_node, "tools": tool_node} — the two registered callables. START and END are not in this dict; they are special sentinels added by the runtime, not user-defined nodes. Edges (including the entry edge from START) are also absent — those come in module 3.
def tool_node(state: AgentState) -> dict: last_msg = state["messages"][-1] # the AI message with tool_calls tool_call = last_msg.tool_calls[0] # first requested call result = run_tool(tool_call) # executes the tool tool_msg = ToolMessage( content=result, tool_call_id=tool_call["id"] ) # TODO: return the partial update that appends tool_msg # Hint 1: which key holds the conversation history? # Hint 2: the value must be a list so the reducer can append
last_msg.tool_calls[0]ToolMessage(content=..., tool_call_id=...)tool_call["id"]Stop — write the return statement before revealing the answer. The crux is applying what you just learned: return only the changed key, wrapped in a list.
return {"messages": [tool_msg]}
Changed lines vs. the stub: the empty 'return {}' is replaced with a real partial update. tool_msg must be in a list because the messages reducer expects a list to append — passing a bare ToolMessage object would raise a TypeError or silently store the wrong type depending on the reducer implementation.
TypeError: 'AIMessage' object is not a mapping at merge time. Fix: always return a plain dict.{"messages": ai_message} instead of {"messages": [ai_message]}. Symptom: the reducer tries to iterate a single object and raises TypeError: 'AIMessage' object is not iterable.add_edge before add_node. Symptom: ValueError: Node 'tools' not found at compile time. Always register all nodes first.add_node calls must all appear before any add_edge or add_conditional_edges calls.graph.compile() and print compiled.nodes — both 'agent' and 'tools' must appear before you add any edges.With both nodes registered and verified, the next step is connecting them: module 3 adds the from and the that inspects the agent's output to decide whether to call the tool node or finish.
Add direct edges from START and conditional edges that inspect the last message to decide whether to call a tool or end. You'll complete the router function for the research agent — the worked version shows the 'has tool call' branch; you supply the 'no tool call → END' branch.
How to wire direct and conditional edges into a StateGraph so the research agent decides at runtime whether to call a tool or stop.
Why this matters: Without routing logic your agent can't loop — it either always calls tools or always stops; this module adds the decision point that makes the agent genuinely reactive.
The agent node returns a — only the keys that changed. The merges that dict into shared state using each field's . That merged state is exactly what your router will read.
This module answers: once the agent node runs and appends its response, how does the graph decide whether to call a tool or stop?
A (added with add_edge) always routes from one to the same destination — no branching.
A (added with add_conditional_edges) calls a function that receives the current state and returns a node name or .
For the research agent: connects directly to the agent node. The agent node connects conditionally to either the tool node or based on whether the model requested a tool call.
add_edge(START, "agent") — always enter via the agent nodeadd_conditional_edges("agent", router) — branch after the agent runsadd_edge("tools", "agent") — after tools run, loop back to the agentImagine the research agent answered: "The GDP of France in 2023 was $3.0 trillion." No tool was needed — the model answered from its own knowledge.
The router inspects the last message in the state's message list. If that message contains a (a structured request like {"name": "web_search", "args": {...}}), it routes to "tools". If not, it routes to and the graph stops.
In the GDP example the last message has no tool call, so the router returns and the agent's answer is the final output. If the user asks "What is the current stock price of Apple?", the model emits a tool call — the router catches that and sends execution to the tool node.
from langgraph.graph import END def router(state): last_msg = state["messages"][-1] if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: return "tools" return END # What does router return when the model's last message # contains tool_calls = [{"name": "web_search", ...}]?
state["messages"][-1]hasattr(last_msg, "tool_calls")last_msg.tool_callsreturn ENDreturn "tools"The router reads the last message from state and checks for a — the model's structured request to invoke a tool. Returning a string routes to that node; returning terminates the graph.
It returns the string "tools" because last_msg.tool_calls is a non-empty list. Execution moves to the "tools" node.
from langgraph.graph import StateGraph, START, END graph = StateGraph(ResearchState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node_stub) # Direct: always enter via the agent graph.add_edge(START, "agent") # Conditional: router decides what happens after agent runs graph.add_conditional_edges("agent", router) # Direct: after tools run, loop back to agent graph.add_edge("tools", "agent")
add_edge(START, "agent")add_conditional_edges("agent", router)add_edge("tools", "agent")This wires the full edge topology for the research agent. The on line 10 is the only branching point — everything else is a fixed route.
START → agent → (router returns END) → graph stops. Only the agent node runs. The tools node is never visited.
def router(state): last_msg = state["messages"][-1] if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: return "tools" # TODO: what should this return when there are no tool calls? # Hint 1: the graph needs a signal to stop. # Hint 2: it's a constant imported from langgraph.graph. ???
???ENDStop — attempt this before revealing. The tool-call branch is complete; you supply the no-tool-call branch. The missing piece is the crux of this module: what terminates the graph?
return END
# Changed line: the last line replaces ??? with return END.
# Why: END is the sentinel that tells the StateGraph to halt — no further nodes run.
# Without it the function returns None, which causes a KeyError at runtime
# because None is not a valid node name.
Slide to see what the router returns at each count of tool_calls in the last message.
None (missing return statement) — the graph raises KeyError: None. Fix: always have an explicit return END as the fallback.state["messages"][0] reads the original user message, which never has tool calls. Fix: always use [-1] to read the model's most recent reply.add_edge("tools", "agent"), the graph terminates after the first tool call. The model never sees the tool result. Fix: add the direct edge from tools back to agent.END, not a string or None.-1 (last), not 0 (first).add_conditional_edges is called on the "agent" node, not on "tools".With routing solid, the next module replaces the tool node stub with a real implementation. It binds tools to the model and dispatches each tool call the router sends its way.
Bind a list of tools to the model with bind_tools, then implement the tool-execution node using ToolNode or a manual dispatch loop. The research agent gains a web-search tool; you'll complete the node that reads tool_calls from the last message and appends ToolMessage results back to state.
How to bind tools to a chat model and implement the node that executes tool calls and writes results back into graph state.
Why this matters: This is the step that makes your research agent actually do things — without tool binding and a working tool node, the agent can only generate text, not search, query, or act.
Decision this forces: Use LangGraph's built-in ToolNode or write a custom dispatch loop — when does each make sense?
Answer: the router checks whether the last message contains a tool_calls list. If it does, execution routes to the tool node.END. That branch to the tool node is exactly where this module picks up — the tool node has been a stub until now.
This module wires the two pieces the stub was missing: binding tools to the model so it cantool_call messages, andToolMessage results back into state.
attaches a list of tool schemas to the chat model. The model then knows what functions it may call and what arguments each expects.
tool_calls field instead of plain text. It does not run the tool itself. Your graph's tool node reads that field and does the actual execution.
You bind tools once, at graph-build time, by calling model.bind_tools(tools). The returned model is a drop-in replacement. exactly as before.
The key tradeoff: binding more tools increases token cost and the chance the model picks the wrong one. Keep the list focused on what the agent actually needs.
tool_call message, the tool node must read each call from the last message. It runs the matching function.ToolMessage (with the call's tool_call_id.
ToolNode handles that loop automatically. You pass it your tool list and it dispatches, catches errors, and formats results. Zero boilerplate for the common case.
A custom dispatch loop gives you control when you need it. Examples: per-tool auth headers, conditional retries, logging each call, or mixing sync and async tools that ToolNode can't handle cleanly.
ToolNode when your tools are plain Python callables and you want the default error-wrapping behaviour.def web_search(query: str) -> str: """Search the web and return a result string.""" return f"[search result for: {query}]" tools = [web_search] model_with_tools = model.bind_tools(tools) def agent_node(state): response = model_with_tools.invoke(state["messages"]) return {"messages": [response]}
model.bind_tools(tools)response.tool_callsbind_tools to the agent node from module 2. The model now knows about web_searchtool_calls field when it decides to use it.
Notice the agent node itself is unchanged — only the model object is swapped.
response.tool_calls is a list of dicts, e.g. [{"name": "web_search", "args": {"query": "LangGraph"}, "id": "call_abc123"}]. If the model answers directly, tool_calls is an empty list [].
from langgraph.prebuilt import ToolNode tool_node = ToolNode(tools) # same list as bind_tools # Wire into the graph (from module 3 skeleton) graph.add_node("tools", tool_node) graph.add_node("agent", agent_node) graph.add_conditional_edges("agent", router) graph.add_edge("tools", "agent")
ToolNode(tools)graph.add_edge("tools", "agent") reads tool_callsToolMessage to messages.
The edge "tools" → "agent" sends results back so the model can reason over them on the next turn.
from langchain_core.messages import ToolMessage tool_registry = {t.__name__: t for t in tools} def tool_node_custom(state): last_msg = state["messages"][-1] results = [] for call in last_msg.tool_calls: fn = tool_registry[call["name"]] output = fn(**call["args"]) results.append( # TODO: construct the ToolMessage here ) return {"messages": results}
tool_registry = {t.__name__: t for t in tools}ToolMessage(content=..., tool_call_id=...)last_msg.tool_callsStop — attempt the TODO before revealing. ToolNodeToolMessage correctly — the rest of the loop is already given.
ToolMessage(content=str(output), tool_call_id=call["id"])
Changed lines vs. Stage 2: instead of ToolNode handling dispatch, you iterate last_msg.tool_calls manually (line 6), look up the function in tool_registry (line 7), call it (line 8), and build the ToolMessage yourself (line 9-11). The tool_call_id field is critical — it ties the result back to the specific call the model made; a wrong or missing id causes the model to misread which result belongs to which request.
| Option | Error handling | Per-call control | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| ToolNode (built-in) | Wraps exceptions into ToolMessage automatically | No hooks between dispatch and execution | Tools are plain callables; you want automatic error-wrapping and zero boilerplate. | Minimal overhead | Low — one constructor call |
| Custom dispatch loop | You write try/except and format ToolMessage manually | Full middleware, logging, and retry control | You need per-call auth, retries, audit logging, or async/sync mixing. | More code to maintain | Medium — write the loop yourself |
ToolNode. Symptom: KeyError: 'tool_name' inside ToolNode, or a silent hang in a custom loop if you don't guard the lookup. Fix: ensure bind_tools and ToolNode receive the same list.tool_call_id in a custom ToolMessage. The model receives the result but can't match it to its request. It either ignores it or hallucinates a follow-up. No error is raised; the agent just behaves oddly. Always copy call["id"] verbatim.ToolNode wraps exceptions automatically; your custom loop must do the same.bind_tools and to ToolNode are identical. A mismatch is the most common AI-generated bug.ToolMessage in the generated loop carries tool_call_id=call["id"]. Not a hardcoded string or None.state["messages"] — verify you see AIMessage → ToolMessage in that order before the next agent turn.thread_id. Each conversation run is then isolated and resumable.
Compile the graph with a MemorySaver or SqliteSaver checkpointer and pass a thread_id in the config so each run is scoped to a conversation. You'll configure the research agent to save after every step, then inspect the saved checkpoint to verify state was written — revisiting the state schema from Module 1 to confirm which fields were persisted.
How to compile a LangGraph StateGraph with a checkpointer and supply a thread_id so each conversation's state is saved and resumable.
Why this matters: Your research agent needs to remember what it has done across turns — checkpointing is the mechanism that makes that possible without any extra code in your nodes.
Decision this forces: MemorySaver for development vs. SqliteSaver (or Postgres) for production — what drives that choice?
TypedDict. Name the three fields you put on it and the reducer that handles the message list. Write your answer, then continue.The schema held goal, messages, and tool_results — with an add_messages on the message list so appends never overwrite.
Those same fields are exactly what the checkpointer will freeze after every node — so knowing the schema tells you what you can read back later.
After each node runs, a serialises the full and writes it to a backing store.
Every write is tagged with a — a string you supply at invocation time that scopes the checkpoint to one conversation.
On the next invocation with the same , the graph loads the latest checkpoint and resumes from that state instead of starting fresh.
This is the mechanism behind interrupts, human-in-the-loop pauses, and crash recovery — all three rely on the same persistence layer.
from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph from agent_state import ResearchState # your TypedDict from Module 1 builder = StateGraph(ResearchState) # … add nodes and edges as in modules 2–4 … checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)
MemorySaver()builder.compile(checkpointer=checkpointer)Passing checkpointer= at compile time is the only change needed to make the graph persistent.
Every node that runs will now write a snapshot of ResearchState — goal, messages, tool_results — to the in-memory store.
It starts from the saved checkpoint. The graph loads the latest snapshot for that thread_id and resumes from there, so messages and tool_results from the first run are already present.
config = {"configurable": {"thread_id": "research-session-42"}}
initial_input = {
"goal": "Summarise recent advances in RAG",
"messages": [],
"tool_results": [],
}
result = graph.invoke(initial_input, config=config)
print(result["messages"][-1].content){"configurable": {"thread_id": "..."}}graph.invoke(initial_input, config=config)The config dict is how you pass the to the runtime — it must live under the "configurable" key, or the checkpointer ignores it.
Use a stable, unique string per conversation — a UUID, a user ID, or a session token all work.
LangGraph raises a runtime error — something like 'thread_id is required when a checkpointer is used'. Without a thread_id the checkpointer has no key to write or read under.
snapshot = graph.get_state(config) print("goal :", snapshot.values["goal"]) print("msg count :", len(snapshot.values["messages"])) print("tool hits :", len(snapshot.values["tool_results"])) print("next nodes :", snapshot.next)
graph.get_state(config)snapshot.valuessnapshot.nextgraph.get_state(config) fetches the latest checkpoint for that and returns a StateSnapshot object.
snapshot.values is a plain dict matching your — you can read any field by name to confirm it was persisted correctly.
snapshot.next tells you which nodes would run next — an empty tuple means the graph reached .
It returns an empty snapshot — snapshot.values will be an empty dict (or None depending on the version) because no checkpoint has been written for that thread yet.
from langgraph.checkpoint.sqlite import SqliteSaver # TODO: create a SqliteSaver that writes to "checkpoints.db" # and compile the graph with it. # Hint 1: SqliteSaver takes a connection string or a path. # Hint 2: the compile call is identical to the MemorySaver version. checkpointer = ??? graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "research-session-42"}} result = graph.invoke(initial_input, config=config)
SqliteSaver(...)builder.compile(checkpointer=checkpointer)Stop — attempt the TODO before revealing the answer. The only lines that change from Stage 1 are the import and the checkpointer constructor.
Everything downstream — compile, invoke, get_state — is identical, which is why swapping backends is a one-line change.
Changed line: checkpointer = SqliteSaver("checkpoints.db")
That is the only difference. SqliteSaver writes each checkpoint to a SQLite file on disk, so state survives a process restart. The compile, invoke, and get_state calls are unchanged — the checkpointer interface is the same across backends.
| Option | Durability across restarts | Setup required | Multi-process safe | When to choose | Cost | Complexity |
|---|---|---|---|---|---|---|
| MemorySaver | None — state is lost when the process exits | Import and instantiate — no config | No — in-process only | Local development, unit tests, or single-session demos where losing state on restart is acceptable. | Free; state lives in RAM | Zero — no dependencies |
| SqliteSaver | Full — survives process restarts | Pass a DB path; library handles schema | Limited — SQLite write-lock under high concurrency | Single-server production or staging where you need durable checkpoints without standing up a separate database. | Free; writes to a local SQLite file | Low — provide a DB file path |
| PostgresSaver | Full — ACID-compliant writes | Connection string + optional schema migration | Yes — designed for concurrent workers | High-concurrency production deployments where multiple workers must share checkpoint state reliably. | Hosting cost for Postgres | Medium — requires a running Postgres instance and connection string |
"thread_id is required" the moment you invoke a checkpointed graph without the config envelope. Fix: always pass {"configurable": {"thread_id": "..."}}.graph.update_state(config, {}) to reset."configurable" — a typo like "config" silently skips persistence.graph.get_state(config) after invoke and assert snapshot.values["messages"] is non-empty — this proves the write actually happened..db file exists on disk after the first invoke — a missing file means the path was wrong or the import failed silently.snapshot.next == () for a completed run; a non-empty tuple means the graph stalled mid-execution and the checkpoint is partial.With checkpoints verified, the next module — Invoking, Resuming, and Debugging the Agent — shows you how to stream step-by-step output and resume a paused run by passing the same with new input, which is where the persistence you just wired in pays off.
Invoke the compiled research agent with graph.invoke and graph.stream, then resume a paused run by passing the same thread_id with a new input. You'll also work through the three most common failure modes — wrong router return value, non-idempotent side effects before an interrupt, and missing thread_id — and learn how to verify AI-generated graph code before shipping it.
How to invoke, stream, and resume a compiled LangGraph agent, plus how to diagnose the three most common runtime failures.
Why this matters: This is the final operational skill — without it, your compiled graph stays a static artifact; with it, you can run, observe, recover, and ship a production-ready stateful agent.
Decision this forces: When should you use invoke (blocking) vs. stream (step-by-step visibility) for your agent?
thread_id do when you pass it in the config to a compiled graph?Answer: it scopes the run to a named conversation. It acts as the cursor the uses to save and reload state between calls.
Module 5 left you with a compiled graph that saves a checkpoint after every step. This module covers what you do with that checkpoint. You'll learn how to start a run, watch it execute step by step, and pick it back up after a pause.
The driving question: when your research agent finishes a tool call and waits for human approval, how do you resume it? How do you know if something went wrong?
graph. blocks until the graph reaches and returns the final state dict. Use it when you only care about the result, not the intermediate steps.
graph. yields a dict after every node completes, keyed by node name. Each yielded chunk is a partial state update — the same shape as a node's return value. Use stream when you need live progress, want to surface tool calls to a UI, or are debugging which node ran and what it returned.
Both accept the same two arguments: the initial input dict and a config dict containing the under the configurable key.
# Stage 1 — blocking invoke config = {"configurable": {"thread_id": "research-001"}} final_state = graph.invoke( {"messages": [HumanMessage(content="Find recent AI safety papers")]}, config=config, ) print(final_state["messages"][-1].content) # Stage 2 — streaming the same run (new thread to avoid replaying) config2 = {"configurable": {"thread_id": "research-002"}} for step in graph.stream( {"messages": [HumanMessage(content="Find recent AI safety papers")]}, config=config2, ): node_name, update = next(iter(step.items())) print(f"[{node_name}] messages: {len(update.get('messages', []))}")
graph.invoke(input, config=config)graph.stream(input, config=config2)next(iter(step.items())){"configurable": {"thread_id": "..."}}Stage 1 shows the minimal invoke call — input dict plus config, result is the full final state. Stage 2 replaces invoke with stream and iterates over chunks; each chunk is a one-key dict mapping the node name to its partial update.
Each line looks like: [agent] messages: 1 or [tools] messages: 1 — the node name in brackets, then the count of messages that node appended in that step. The agent node typically adds one AIMessage; the tools node adds one ToolMessage per tool call.
# The graph paused at an interrupt — human approved the tool call. # Resume by invoking with the SAME thread_id and a new HumanMessage. resume_config = {"configurable": {"thread_id": "research-001"}} result = graph.invoke( {"messages": [HumanMessage(content="Approved — go ahead")]}, config=resume_config, ) print(result["messages"][-1].content)
graph.invoke({"messages": [HumanMessage(...)]}, config=resume_config)thread_id: "research-001"Resuming is just another invoke call — the magic is the matching . The checkpointer loads the saved state, appends the new HumanMessage, and continues from the interrupted node.
The graph does NOT restart from the beginning — it picks up exactly where the paused it.
The checkpointer loads the existing state for that thread and the graph re-runs from the last checkpoint with no new input appended. Depending on where the interrupt was, this may re-execute the interrupted node — which can cause side effects to fire a second time (the non-idempotent side-effect failure mode).
Your returns a string that doesn't match any key in add_conditional_edges's path map. LangGraph raises: ValueError: Branch condition returned 'tool_call' but no branch named 'tool_call' exists. Fix: print the router's return value in isolation before wiring it into the graph, and make sure every string it can return has a matching key.
A node sends an email or writes to a database, then the graph hits an . When you resume, that node re-runs — and the email goes out twice. There's no error; the symptom is duplicate records or messages in your external system. Fix: move side effects to AFTER the interrupt, or guard them with a state flag that the reducer tracks.
You compile with a checkpointer but forget to pass config (or pass an empty dict). LangGraph raises: ValueError: thread_id is required when using a checkpointer. Fix: always construct config as {"configurable": {"thread_id": some_id}} before every invoke or stream call.
# The research agent paused for approval on thread "research-007". # Task: stream the resumed run and print each node name as it executes. resume_config = # TODO: build the config dict with thread_id="research-007" for step in graph.stream( {"messages": [HumanMessage(content="Approved")]}, config=resume_config, ): node_name, _ = next(iter(step.items())) print(f"Resumed step: {node_name}")
graph.stream({"messages": [...]}, config=resume_config)next(iter(step.items()))# TODOStop — attempt the TODO before revealing. The rest of the loop is complete; only the config construction is missing. Hint 1: the key structure is {"configurable": {...}}. Hint 2: the thread_id must exactly match the paused run.
resume_config = {"configurable": {"thread_id": "research-007"}}
Changed line: the config dict is the only addition — everything else was already correct. This is the crux: without the matching thread_id, the checkpointer cannot find the saved state and raises ValueError. The stream loop then prints 'Resumed step: agent' and 'Resumed step: tools' (or 'Resumed step: __end__') depending on how many nodes run after the interrupt clears.
With this pattern in hand, you have everything you need for the solo capstone: build, run, pause, resume, and debug a full stateful research agent from scratch.
AI-generated LangGraph code looks plausible but fails in predictable spots. Run through these checks before shipping:
add_conditional_edges.config with a non-empty .graph.compile(checkpointer=...). Not just graph.compile().step.items() or equivalent. Not step["messages"] directly.Before looking at the build order below, try to reconstruct it from memory: starting from the state schema, what did you add at each step, and what did each addition unlock? Name the six stages and the key decision each one forced.
Apply what you learned to a stateful tool-using agent with nodes, edges, and checkpoints.
You define a TypedDict state schema with a messages field. After a node runs, LangGraph merges the node's returned dict into the existing state. Which behavior correctly describes what happens to messages if your schema uses the add_messages reducer?
The add_messages reducer merges incoming messages onto the existing list — that is the whole point of reducer-annotated fields. Full-state replacement only happens on fields with no reducer. Nodes always return partial dicts; returning the full state is unnecessary and would still be treated as a partial update. LangGraph does not raise a TypeError for list reducers.
A teammate proposes one large node that calls the LLM, checks whether a tool is needed, and executes the tool — all inside a single function. What is the strongest argument against this design?
Separating reasoning and tool dispatch into distinct nodes lets you place a conditional edge between them, so the graph can skip tool execution entirely when no tool call is emitted. It also makes each node unit-testable in isolation. LangGraph imposes no restriction on calling external APIs inside a node, no limit on how many keys a node returns, and no line-count check at compile time.
Read this router function:
def route(state):
last = state["messages"][-1]
return "tools" if last.tool_calls else END
The agent node has just run and the model returned a plain text answer with no tool calls. Which node executes next?
An empty list is falsy in Python, so the condition is False and the router returns END, halting the graph. tool_calls is present on AIMessage objects even when empty — it is not missing, so no KeyError occurs. An empty list is not truthy, so the tools branch is not taken. Returning END does not loop back to the agent node.
You are moving a LangGraph agent from a weekend prototype to a production service that must survive process restarts and serve multiple concurrent users across sessions. Which checkpointer choice and reason are correct?
SqliteSaver (and Postgres-backed savers) write state to disk or a database, so checkpoints survive process restarts and are accessible to any process that connects to the same store — exactly what production requires. MemorySaver is in-process only: it is lost on restart and not shared across processes. SqliteSaver is open-source and requires no paid subscription. Without a checkpointer, thread_id has nothing to look up — there is no saved state to resume from.
You have a long-running research agent and a user-facing chat endpoint. Explain when you would call graph.invoke() versus graph.stream(), and name one concrete benefit of the non-blocking option in a chat context.
invoke blocks until the entire graph finishes and returns the final state. stream yields each step's output as it is produced, which is valuable in interactive contexts because the user sees progress immediately. The key decision driver is whether intermediate output has value to the caller — if yes, stream; if only the final answer matters and latency is acceptable, invoke is simpler.