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.

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

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

Test matrix

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.