Fully autonomous agents make irreversible mistakes. Human-in-the-loop (HITL) patterns let agents pause at critical decision points and wait for human approval before taking actions that are expensive, dangerous, or hard to undo.
Not every action needs human approval — that defeats the purpose of agents. Select a trigger rule to see how to detect it programmatically:
Example: Deleting records, sending emails, deploying to production
def plan_node(state: State) -> dict:
plan = model.invoke(f"Plan: {state['task']}").content
return {"action_plan": plan}
def classify_reversibility(plan: str) -> bool:
"""True if the planned action is irreversible."""
irreversible_keywords = [
"delete", "drop", "send email", "deploy", "publish",
"remove", "purge", "terminate",
]
return any(kw in plan.lower() for kw in irreversible_keywords)
# In your graph: only interrupt if irreversible
def should_interrupt(state: State) -> str:
if classify_reversibility(state["action_plan"]):
return "approval" # Pause for human review
return "execute" # Safe to run directlyNot every action needs human approval — that defeats the purpose of agents. Interrupt when:
Deleting a database record, sending an email, deploying to production
Buying something, running a 1-hour compute job, making an external API call with side effects
Agent is taking an unusual code path or has low confidence on a critical extraction
Handling PII, financial data, or credentials
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from typing import TypedDict
class State(TypedDict):
task: str
action_plan: str
human_approved: bool
result: str
def plan_node(state: State) -> dict:
"""Agent proposes an action."""
plan = model.invoke(f"Plan how to accomplish: {state['task']}").content
return {"action_plan": plan}
def approval_node(state: State) -> dict:
"""Pause and wait for human review."""
human_decision = interrupt({
"message": "Please review this action plan before I proceed.",
"action_plan": state["action_plan"],
"options": ["approve", "reject", "modify"],
})
# Execution is PAUSED here — the human's response will resume it
return {"human_approved": human_decision == "approve"}
def execute_node(state: State) -> dict:
"""Only runs if approved."""
if not state["human_approved"]:
return {"result": "Action cancelled by human."}
result = execute_action(state["action_plan"])
return {"result": result}
graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("approval", approval_node)
graph.add_node("execute", execute_node)
graph.set_entry_point("plan")
graph.add_edge("plan", "approval")
graph.add_edge("approval", "execute")
graph.add_edge("execute", END)
checkpointer = MemorySaver() # Checkpointer required for interrupts
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["approval"], # Pause BEFORE running approval_node
)
config = {"configurable": {"thread_id": "task-001"}}
# Run until interrupt
result = app.invoke({"task": "Delete all test records from the database"}, config=config)
print("Agent paused. Review the plan:")
print(result["action_plan"]) # Human reviews this
# Human approves — resume with their input
final = app.invoke(
Command(resume="approve"), # Pass human decision
config=config,
)from fastapi import FastAPI
from pydantic import BaseModel
api = FastAPI()
class ApprovalRequest(BaseModel):
thread_id: str
action_plan: str
class ApprovalResponse(BaseModel):
thread_id: str
decision: str # "approve" | "reject"
notes: str = ""
# Pending approvals (use Redis in production)
pending: dict[str, ApprovalRequest] = {}
@api.post("/agent/start")
async def start_agent(task: str):
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
state = app.invoke({"task": task}, config=config)
# Graph paused — queue the approval request
pending[thread_id] = ApprovalRequest(
thread_id=thread_id,
action_plan=state["action_plan"],
)
return {"thread_id": thread_id, "status": "awaiting_approval", "action_plan": state["action_plan"]}
@api.post("/agent/approve")
async def approve_action(response: ApprovalResponse):
config = {"configurable": {"thread_id": response.thread_id}}
final = app.invoke(Command(resume=response.decision), config=config)
del pending[response.thread_id]
return {"result": final["result"], "status": "completed"}Approval fatigue is real. If humans approve 99% of agent actions without reading them, the HITL gate becomes security theater. Keep interrupts rare and significant — the rarer the interruption, the more attention it gets.
In the final lesson before the capstone, we cover how to test AI agents — a significantly harder problem than testing deterministic functions.