HOW-TO · RAG
How to Add Human-in-the-Loop to LangGraph Workflows
Target environment
Ubuntu 24.04 · Ollama 0.4.x
PREREQUISITES
LangGraph workflow running, checkpointer configured, Python 3.10+
What this does
Human-in-the-loop pauses execution before or after specific nodes, allowing a human to approve, reject, or modify the agent's action before continuing.
Steps
- Add a checkpointer. The checkpointer saves state snapshots for later resumption.
from langgraph.checkpoint import MemorySaver
memory = MemorySaver()
- Compile the graph with the checkpointer and an interrupt point. Use
interrupt_beforeto pause before a node.
from langgraph.graph import StateGraph, END
from typing import TypedDict
class State(TypedDict):
input: str
approved: bool
def process(state: State) -> dict:
return {"approved": False}
def execute(state: State) -> dict:
return {"approved": True}
workflow = StateGraph(State)
workflow.add_node("process", process)
workflow.add_node("execute", execute)
workflow.set_entry_point("process")
workflow.add_edge("process", "execute")
workflow.add_edge("execute", END)
app = workflow.compile(
checkpointer=memory,
interrupt_before=["execute"]
)
- Run with a thread ID. The thread ID ties state across interruptions.
config = {"configurable": {"thread_id": "session-1"}}
# First run — pauses before "execute"
result = app.invoke({"input": "transfer $100", "approved": False}, config)
print(result) # State is partial, execution is paused
- Resume with human approval. Update state and continue.
# Human reviews and approves
state = app.get_state(config)
state.values["approved"] = True
app.update_state(config, {"approved": True})
# Resume execution
result = app.invoke(None, config)
print(result) # Now "execute" node ran
- Reject by updating state and ending. Set a field that causes the execute node to skip.
state.values["approved"] = False
app.update_state(config, {"approved": False})
# The execute node can check approved before proceeding
Verification
python -c "
from langgraph.checkpoint import MemorySaver
m = MemorySaver()
print(type(m).__name__)
# Expected: MemorySaver
"
Common failures
- Missing thread_id. The checkpointer requires a
thread_idin the config; otherwise, state is not persisted and interrupts fail silently. - Resuming with wrong state schema.
update_statemust match the state schema. Extra keys cause validation errors. - Interrupt prevents graph from reaching END. If execution is interrupted and never resumed, the graph stays in a paused state. Check
app.get_state(config).nextto see pending nodes. - Version mismatch - The installed package or runtime differs from the command shown; check the version first and rerun the smallest verification command.
- Local environment drift - Another service, virtual environment, model, or path is being used; print the active binary path and configuration before changing the guide steps.
Related guides
- How to Set Up Basic LangGraph Agent from Scratch
- How to Implement Memory in LangGraph State