Chapter 11: Human-in-the-Loop and Interrupts
In Chapter 10, we learned that Checkpointing allows LangGraph to pause execution, serialize the State to a database, and perfectly resume the workflow later.
In this chapter, we will intentionally exploit that mechanic.
We will learn how to intentionally halt an Agent mid-thought, ask a human for permission, and resume the Agent once the human is satisfied. This pattern is known as Human-in-the-Loop (HITL).
11.1 Why Human Approval Matters
As Agents become more autonomous, the blast radius of their mistakes grows exponentially.
It is perfectly fine to let an autonomous agent summarize a PDF. It is catastrophic to let an autonomous agent execute a database DROP TABLE command, wire a $50,000 payment, or push code directly to the production branch.
You never want an agent doing high-risk actions automatically.
By inserting a Human-in-the-Loop, the Agent does the heavy lifting (drafting the code, calculating the payment, preparing the query), but a human holds the keys to the final execution.
11.2 The interrupt() API
To pause a graph, LangGraph provides the interrupt() function.
When a node calls interrupt(), the Graph Runtime immediately suspends execution. It saves a Checkpoint, puts the thread to sleep, and yields control back to the application layer.
Mandatory Checkpointer:
interrupt()requires a checkpointer because the graph must persist its state before pausing. For development,MemorySaveris sufficient. For production, you must use a persistent checkpointer likePostgresSaver.
from langgraph.types import interrupt
def security_node(state: AgentState):
print("Preparing high-risk operation...")
# 1. The Graph halts here!
# 2. It saves a Checkpoint.
# 3. It surfaces the string "Approve database deletion?" to the application.
human_response = interrupt("Approve database deletion?")
# 4. When the graph is resumed, human_response will contain the human's answer.
if human_response == "APPROVED":
print("Executing deletion...")
return {"status": "deleted"}
else:
print("Operation aborted by human.")
return {"status": "aborted"}
11.3 Inspecting Pending Interrupts
Once the graph halts, how does your backend application know it is waiting for a human?
You can query the database using app.get_state(). LangGraph stores the pending interrupt metadata inside the checkpoint's tasks array.
config = {"configurable": {"thread_id": "thread_999"}}
state = app.get_state(config)
for task in state.tasks:
# This reads the exact string passed into the interrupt() function!
print("Graph is paused. Waiting for:", task.interrupts[0].value)
# Output: "Graph is paused. Waiting for: Approve database deletion?"
Enterprise developers use this exact API to build "Pending Approval" dashboards where managers can log in and see exactly which workflows are waiting for their review.
11.4 Resuming the Graph (Command)
Three days later, a Security Manager logs into the dashboard, sees the pending request, and clicks the "Approve" button.
To wake the graph up and pass the manager's approval back into the interrupt() function, we use the Command object with resume.
from langgraph.types import Command
# 1. We use the EXACT same thread_id
config = {"configurable": {"thread_id": "thread_999"}}
# 2. We invoke the graph again, passing a Command instead of state updates
app.invoke(Command(resume="APPROVED"), config=config)
Crucial Accuracy Note: When you resume the graph, the node does not rerun from the top. Execution resumes exactly from the line where
interrupt()was called! The graph wakes up, assigns"APPROVED"to thehuman_responsevariable, and continues.
11.5 Building an Approval Workflow
Let's build an end-to-end Human-in-the-Loop workflow for sending invoices. Notice how we treat rejection as a standard state update rather than crashing the application with an Exception.
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
class InvoiceState(TypedDict):
client: str
amount: int
status: str
def generate_invoice(state: InvoiceState):
print(f"Drafting invoice for {state['client']}...")
return {"status": "drafted"}
def human_approval_node(state: InvoiceState):
# Pause the graph and ask for approval!
decision = interrupt(f"Review Invoice for {state['client']}: ${state['amount']}. Approve?")
if decision == "YES":
return {"status": "approved"}
else:
return {"status": "rejected"} # Business rejection, not a system failure
def send_invoice(state: InvoiceState):
print(f"SUCCESS: Invoice sent to {state['client']}!")
return {"status": "sent"}
# Routing logic
def route_approval(state: InvoiceState):
if state["status"] == "approved":
return "sender"
return END # If rejected, gracefully exit the graph.
builder = StateGraph(InvoiceState)
builder.add_node("generator", generate_invoice)
builder.add_node("approval", human_approval_node)
builder.add_node("sender", send_invoice)
builder.add_edge(START, "generator")
builder.add_edge("generator", "approval")
builder.add_conditional_edges("approval", route_approval)
builder.add_edge("sender", END)
memory = MemorySaver()
app = builder.compile(checkpointer=memory)
# --- Execution ---
config = {"configurable": {"thread_id": "invoice_001"}}
print("\n--- Starting Workflow ---")
app.invoke({"client": "Acme Corp", "amount": 5000, "status": "pending"}, config=config)
print("\n--- Graph is Asleep. Human is Reviewing ---")
print("\n--- Resuming Workflow ---")
app.invoke(Command(resume="YES"), config=config)
11.6 Human Feedback Loops (Editing)
interrupt() isn't just for binary "Yes/No" approvals. It is incredibly powerful for Editing.
Imagine an Agent drafting a marketing email. Instead of just approving it, the human can rewrite parts of the email before it sends!
def review_draft(state: MarketingState):
draft = state["draft_text"]
# Send the draft to the human for editing
edited_draft = interrupt(f"Please review and edit this draft:\n{draft}")
# Overwrite the state with the human's edited version!
return {"draft_text": edited_draft}
When the graph resumes via Command(resume="Human's modified text"), the State is permanently updated with the human's exact edits.
11.7 Multi-Step Approval Chains
Interrupts are not limited to a single pause. You can build massive organizational chains by placing multiple interrupt nodes in sequence.
Draft Contract
↓
[INTERRUPT] Legal Review
↓
[INTERRUPT] Finance Review
↓
[INTERRUPT] Executive Approval
↓
Sign Contract
At each stage, the graph goes to sleep, waits for the specific department to provide their Command(resume=...) input, saves the checkpoint, and routes to the next department.
11.8 Summary
By combining Checkpointing with the interrupt() API, LangGraph allows you to seamlessly weave human intelligence into autonomous agent loops.
You can build fail-safes for high-risk actions, collaborative editing workflows, and multi-tier enterprise approval chains.
At this point, we have mastered single-agent architectures. But what happens when one agent isn't enough? What if you need a specialized Research Agent to collaborate with a specialized Writing Agent?
In the next chapter, we will unlock the power of composability by diving into Subgraphs.