LangGraph human-in-the-loop patterns let you freeze an agent mid-execution, wait for a person to review or edit state, then resume without losing context. This guide walks through wiring checkpoints into a stateful graph using LangGraph’s interrupt primitive and a durable checkpointer, so you can drop approval gates into any agent workflow.
Step 1: Install and import the right packages
Use LangGraph 0.2.x or newer. The interrupt and Command types live in langgraph.types, and checkpointers ship under langgraph.checkpoint.
pip install langgraph langchain-coreFor a durable backend we’ll use SQLite in this example; swap to Postgres later by changing the import.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.sqlite import SqliteSaverStep 2: Define state and nodes with an interrupt
State is a typed dict. The node that needs human sign-off calls interrupt() with a payload describing what the human should see. The return value of interrupt() is whatever you pass to Command(resume=...) later.
class State(TypedDict):
draft: str
approved: bool
final: str
def draft_node(state: State):
# In real code, call an LLM here. If you route through n4n.ai, its
# OpenAI-compatible endpoint with automatic fallback keeps this node
# resilient when a provider is degraded.
return {"draft": "Invoice #42 is ready for your review."}
def approval_node(state: State):
# Pause for human. The dict is surfaced to your UI or CLI.
review = interrupt({"draft": state["draft"], "action": "approve?"})
# review is the human's response
return {"approved": bool(review.get("approved", False))}
def send_node(state: State):
if state["approved"]:
return {"final": state["draft"] + " (sent)"}
return {"final": "discarded"}The interrupt call raises a control flow exception that LangGraph catches, snapshots state, and returns control to the caller.
Step 3: Compile the graph with a durable checkpointer
A checkpointer is what makes LangGraph human-in-the-loop survivable across process restarts. MemorySaver is fine for tests; SqliteSaver or PostgresSaver is mandatory for production.
builder = StateGraph(State)
builder.add_node("draft", draft_node)
builder.add_node("approval", approval_node)
builder.add_node("send", send_node)
builder.add_edge(START, "draft")
builder.add_edge("draft", "approval")
builder.add_edge("approval", "send")
builder.add_edge("send", END)
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
graph = builder.compile(checkpointer=saver)The thread_id in the config scopes state. Every invocation with the same thread_id reads/writes the same checkpoint sequence.
Step 4: Run the graph and detect the interruption
Invoke with a config that includes thread_id. On first run the graph executes draft_node, hits approval_node, and stops at the interrupt.
config = {"configurable": {"thread_id": "job-1"}}
result = graph.invoke({"draft": ""}, config)
print(result)LangGraph returns the state up to the interrupt, with an __interrupt__ key containing your payload. Your application should detect this and surface the question to a human instead of proceeding.
if "__interrupt__" in result:
prompt = result["__interrupt__"][0].value
print("Human needed:", prompt)Step 5: Resume with human input via Command
Once the human responds, resume the exact same thread with Command(resume=...). The value you pass becomes the return value of interrupt() inside approval_node.
human_response = {"approved": True}
result = graph.invoke(Command(resume=human_response), config)
print(result)The graph replays from the saved checkpoint, executes the remainder of approval_node, then runs send_node. Final state shows approved: True and final containing the sent string.
If the human rejects, pass {"approved": False} and the send_node branches accordingly.
Step 6: Verify the checkpoint and resume worked
Verification is two-fold: confirm the interrupt fired, and confirm the resumed run completed the remaining nodes.
Inspect state directly:
state = graph.get_state(config)
assert state.values["approved"] is True
assert "sent" in state.values["final"]You can also list the checkpoint history to prove persistence:
for checkpoint in graph.get_state_history(config):
print(checkpoint.step, checkpoint.values)A successful run shows at least two steps: one ending at approval, one ending at END.
Step 7: Handle restarts and production persistence
Kill the Python process after Step 4, then restart and run Step 5 with the same thread_id and same checkpoints.db file. The graph resumes because the checkpoint is on disk, not in memory.
For multi-worker deployments, use PostgresSaver:
from langgraph.checkpoint.postgres import PostgresSaver
saver = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")
saver.setup()
graph = builder.compile(checkpointer=saver)Two caveats from production:
- Idempotency: nodes after an interrupt may re-execute if you change graph topology. Keep side effects (sending email, writing DB) behind a flag in state or use LangGraph’s built-in retry controls.
- Serialization: state must be JSON-serializable or registered with the checkpointer’s serializer. Don’t put raw model objects in state.
Alternative: interrupt_before
If you don’t want to modify the node, compile with interrupt_before=["approval"]. The graph pauses before entering the node; resume with Command(resume=None) and read prior state via get_state. The in-node interrupt() approach is better when you need to pass a dynamic payload to the human.
What you shipped
You now have a LangGraph human-in-the-loop gate that persists across restarts, surfaces a structured prompt to a person, and resumes exactly where it left off. The same pattern works for editing tool calls, confirming deletions, or any step where an agent shouldn’t act unilaterally.