Stateful agent tutorial
Build a fictional draft-review graph that remembers its state, pauses for a person and resumes safely—without sending anything externally.
Published 13 October 2026 · Barcelona Code School
A stateful LangGraph agent stores shared data across graph nodes and persists checkpoints under a stable thread ID. In this tutorial, a request is validated, a draft is created, execution pauses with interrupt(), and a person approves, edits or rejects it. Resume with Command(resume=...) and the same thread ID. Keep side effects after approval because the interrupted node restarts when resumed.
Key takeaways
- State stores raw facts needed across nodes; prompts are formatted inside nodes.
- Nodes return state updates and edges define allowed movement.
- A checkpointer and thread ID make pause/resume possible.
- Interrupt should precede non-idempotent side effects.
- Test duplicate resumes, wrong thread IDs and rejected approval before adding real tools.
What will you build?
The example is a low-risk review assistant. It accepts a fictional request, validates required text, creates a deterministic placeholder draft, pauses for review and stores the final decision. It does not call a model, send a message or modify an external system. That keeps the tutorial focused on LangGraph state and control.
Figure 1. Stateful draft-review execution
- Validate · Required input?
- Draft · Store raw result
- Interrupt · Checkpoint + review
- Resume · Same thread ID
- Finish · Record decision
No external side effect occurs in this learning graph.
Text equivalent: input is validated and converted into a draft stored in state. The review node pauses through an interrupt and the checkpointer saves the thread. A later call resumes the same thread with a human decision. The graph records an approved, edited or rejected final state and ends without sending data elsewhere.
Step 1: install LangGraph and define state
pip install -U langgraph
from typing import Literal, TypedDict
class ReviewState(TypedDict, total=False):
request_id: str
request_text: str
draft: str
status: Literal["new", "waiting", "approved", "rejected"]
reviewer_note: str
Store values that must survive across nodes. Keep the original request and draft as raw data; format review messages only when needed.
Step 2: add validation and draft nodes
def validate(state: ReviewState):
text = state.get("request_text", "").strip()
if not text:
raise ValueError("request_text is required")
return {"status": "new"}
def create_draft(state: ReviewState):
# Deterministic placeholder: replace with a tested model call later.
draft = f"Draft for review: {state['request_text']}"
return {"draft": draft, "status": "waiting"}
Each node receives current state and returns only its updates. Validation remains deterministic. A future model call should return structured output and have separate error handling.
Step 3: pause for human review
from langgraph.types import interrupt
def review(state: ReviewState):
# Put interrupt before any non-idempotent side effect.
decision = interrupt({
"request_id": state["request_id"],
"draft": state["draft"],
"allowed": ["approve", "edit", "reject"],
})
if decision["action"] == "reject":
return {"status": "rejected", "reviewer_note": decision.get("note", "")}
return {
"status": "approved",
"draft": decision.get("edited_draft", state["draft"]),
"reviewer_note": decision.get("note", ""),
}
On resume, the interrupted node starts again from its beginning. Never place an unguarded email, payment, database write or other irreversible operation before interrupt().
4-week live bootcamp · Barcelona or online
Go from one workflow to a complete AI automation system
In the AI Agent & Automation Bootcamp you build a no-code Virtual AI Department with n8n, RAG, connected agents and human approvals — live with an instructor and a small group.
Not sure it fits? Take the free AI Career Roadmap test
Step 4: compile with a checkpointer
from langgraph.graph import END, START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(ReviewState)
builder.add_node("validate", validate)
builder.add_node("create_draft", create_draft)
builder.add_node("review", review)
builder.add_edge(START, "validate")
builder.add_edge("validate", "create_draft")
builder.add_edge("create_draft", "review")
builder.add_edge("review", END)
graph = builder.compile(checkpointer=InMemorySaver())
The in-memory saver is suitable for this local tutorial, not a production deployment. Production needs durable storage and explicit retention, privacy and recovery decisions.
Step 5: invoke, inspect and resume
from langgraph.types import Command
config = {"configurable": {"thread_id": "review-case-1042"}}
initial = {
"request_id": "case-1042",
"request_text": "Prepare a short acknowledgement.",
}
paused = graph.invoke(initial, config=config)
# Inspect paused["__interrupt__"] in the invoke API.
finished = graph.invoke(
Command(resume={"action": "approve", "note": "Checked"}),
config=config,
)
assert finished["status"] == "approved"
The same thread ID is essential. Treat it as an access-controlled pointer to persisted state, not a guessable public token.
Permissions and recovery boundaries
| Risk | Control | Safe result |
|---|---|---|
| Wrong person resumes thread | Authenticate caller and authorize access to the business transaction | No cross-user state access |
| Code before interrupt repeats | Put side effect after approval or make operation idempotent | No duplicate write |
| Process crashes while waiting | Durable production checkpointer and stable thread ID | State can be recovered |
| Malformed resume payload | Validate action and fields against a schema | Graph remains paused or fails safely |
| Approval rejected | Terminal rejected status; no external action | Decision is preserved |
| External action times out later | Read destination state before retry | No blind duplicate action |
Test matrix
| Test | Pass condition |
|---|---|
| Valid input | Pauses with draft and request ID |
| Missing text | Fails before draft/review |
| Approve | Status approved; original draft retained |
| Edit | Status approved; edited draft stored |
| Reject | Status rejected; no side effect |
| Wrong thread ID | Cannot resume target state |
| Malformed decision | Rejected by validation |
| Duplicate resume | No external write; terminal state handled predictably |
Success criteria
- State contains only required raw values.
- Each node has one observable responsibility.
- The graph pauses and resumes with the same thread ID.
- Approval, edit and rejection paths are tested.
- No external action occurs before authorization.
- Production design names a durable checkpointer, retention policy and recovery owner.
Choose training that matches your stack
This tutorial is for code-first learners. If your goal is Python or JavaScript agent engineering, continue with official LangGraph documentation, software testing and deployment practice. Barcelona Code School's current AI Agent & Automation Bootcamp teaches visual business automation with n8n rather than LangGraph development.
Frequently asked questions
What makes a LangGraph agent stateful?
Nodes read and update shared graph state, while a checkpointer saves state for a thread. Reusing the same thread ID lets a later invocation load and continue that thread rather than starting with empty state.
Why does LangGraph need a thread ID?
The thread ID identifies which persisted checkpoints belong to a run or conversation. Use a stable unique identifier for the business transaction or session and enforce access so one user cannot resume another user's state.
How does interrupt work in LangGraph?
The interrupt function pauses a node, persists graph state through the configured checkpointer and returns a JSON-serializable request to the caller. The graph continues when invoked again with Command resume and the same thread ID.
Can code before interrupt run twice?
Yes. When a paused graph resumes, the interrupted node restarts from its beginning. Put interrupt before non-idempotent side effects, or protect any earlier operation with an idempotency key and destination-state check.
Is MemorySaver suitable for production?
Use an in-memory saver only for local learning and tests. Production applications need a durable checkpointer selected and configured for their infrastructure, retention, privacy, concurrency and recovery requirements.
Sources
- LangGraph overview.
- Graph API.
- Persistence.
- Durable execution.
- Interrupts.
- BCS: How to Build an AI Agent.
Continue learning
Design an agent from the business process
Build visual AI automation systems with n8n
Originally published at Barcelona Code School. Learn more about our AI, coding and design programmes at barcelonacodeschool.com.
Top comments (0)