Skip to main content

Human-in-Loop Agents: Interrupts & Approval (2026)

High-stakes agent workflows (financial transfers, data deletions, content publication) require human oversight. A human-in-the-loop agent pauses after critical decisions and asks for approval before proceeding. LangGraph's interrupt system lets you pause at any node, display the state to a human, collect feedback, and resume. This article covers interrupt patterns, approval flows, and designing safe agent-human collaboration.

The Human-in-the-Loop Pattern​

The basic pattern is:

  1. Agent reasons and decides on an action (e.g., "delete user data").
  2. Before executing, the agent emits a request for human approval.
  3. The workflow pauses; a human reviews the decision and its reasoning.
  4. Human approves, rejects, or modifies the decision.
  5. Workflow resumes with human feedback incorporated.

This is critical for:

  • Financial Agents: Before transferring money, ask a human.
  • Data Deletion: Before purging records, confirm with a manager.
  • Content Publishing: Before posting, ask an editor.
  • Model Fine-tuning: Before training, human reviews the dataset.

LangGraph Interrupts​

LangGraph has a built-in interrupt system. Mark a node as an interrupt point, and the graph pauses after that node, allowing external input.

from langgraph.graph import StateGraph
from langgraph.types import interrupt

class ApprovalState(TypedDict):
    request: str  # What the agent wants to do
    reasoning: str  # Why
    approval_status: str  # "pending", "approved", "rejected"
    feedback: str  # Human feedback

def decision_node(state: ApprovalState) -> dict:
    """Agent decides on an action and requests approval."""
    model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
    
    decision = model.invoke(
        f"What should we do about: {state['request']}?"
    )
    
    # Return the decision and mark as needing approval
    return {
        "reasoning": decision.content,
        "approval_status": "pending"
    }

def approval_interrupt_node(state: ApprovalState) -> dict:
    """Pause here and wait for human approval."""
    if state["approval_status"] == "pending":
        # Emit an interrupt: the graph pauses here
        approval = interrupt(f"Approve this action: {state['reasoning']}")
        
        return {
            "approval_status": approval,  # "approved" or "rejected"
            "feedback": approval  # Or parse structured feedback
        }
    
    return state

def execute_node(state: ApprovalState) -> dict:
    """Execute the approved action."""
    if state["approval_status"] != "approved":
        return {"request": "Action rejected by human review."}
    
    # Perform the action (delete, transfer, publish, etc.)
    result = execute_action(state["request"])
    
    return {"request": f"Action completed: {result}"}

After approval_interrupt_node emits an interrupt, the graph pauses. The calling code receives the interrupt value and displays it to a human. The human provides feedback (approval or rejection), and the calling code resumes the graph with the feedback.

Resuming After Interrupts​

Here's how the calling code handles interrupts:

from langgraph.checkpoint.sqlite import SqliteSaver

graph = StateGraph(ApprovalState)
graph.add_node("decide", decision_node)
graph.add_node("get_approval", approval_interrupt_node)
graph.add_node("execute", execute_node)

graph.add_edge("decide", "get_approval")
graph.add_conditional_edges(
    "get_approval",
    lambda s: "approved" if s["approval_status"] == "approved" else "rejected"
)
graph.add_edge("approved", "execute")
graph.add_edge("rejected", "end")

graph.set_entry_point("decide")
graph.set_finish_point("execute")

saver = SqliteSaver()
compiled_graph = graph.compile(checkpointer=saver)

# Run the agent
thread_id = "approval-workflow-001"
config = {"thread_id": thread_id}

try:
    result = compiled_graph.invoke(
        {"request": "Delete user account #12345", "approval_status": "pending"},
        config=config
    )
except Exception as e:
    # Check if this is an interrupt
    if "interrupt" in str(e):
        print(f"Agent paused and needs approval. Message: {e}")
        
        # Display to human (e.g., in a web UI)
        human_decision = input("Approve? (yes/no): ")
        
        # Resume with human decision
        result = compiled_graph.invoke(
            {"approval_status": "approved" if human_decision == "yes" else "rejected"},
            config=config
        )
        print(f"Resumed workflow: {result}")

The flow is: invoke → interrupt → human reviews → provide feedback → invoke again with same thread_id → resume from interrupt.

Web UI for Approvals​

In production, you'd expose approvals via a web interface (not stdin). Here's a sketch:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

@app.post("/agent/request-approval")
async def request_approval(thread_id: str):
    """Get the pending approval request for a thread."""
    # Fetch the state from checkpointer
    saver = SqliteSaver()
    checkpoint = saver.get({"thread_id": thread_id})
    
    if not checkpoint:
        raise HTTPException(status_code=404)
    
    state = checkpoint["values"]
    
    # Return the request for human review
    return {
        "thread_id": thread_id,
        "request": state.get("request"),
        "reasoning": state.get("reasoning")
    }

@app.post("/agent/submit-approval")
async def submit_approval(thread_id: str, approved: bool, feedback: str = ""):
    """Submit human approval decision."""
    saver = SqliteSaver()
    
    # Resume the graph with the decision
    compiled_graph = build_graph()  # Rebuild your graph
    result = compiled_graph.invoke(
        {"approval_status": "approved" if approved else "rejected", "feedback": feedback},
        config={"thread_id": thread_id}
    )
    
    return {"status": "resumed", "result": result}

A web UI calls /request-approval to fetch the pending decision, displays it to the human, and calls /submit-approval with their choice.

CrewAI Human-in-the-Loop​

CrewAI has a simpler human-in-the-loop mechanism: callbacks and task-level hooks.

from crewai import Agent, Task, Crew

# Define a callback that pauses for human input
def human_review_callback(agent_output):
    """Pause and ask human for feedback."""
    print(f"Agent output: {agent_output}")
    
    approval = input("Approve this output? (yes/no/modify): ")
    
    if approval == "no":
        raise ValueError("Human rejected the output.")
    elif approval == "modify":
        modified = input("Enter modified output: ")
        return modified
    
    return agent_output

# Assign callback to a task
approval_task = Task(
    description="Generate a financial report.",
    agent=financial_agent,
    callback=human_review_callback
)

crew = Crew(agents=[financial_agent], tasks=[approval_task])
result = crew.kickoff()

CrewAI's callback system is lighter than LangGraph's interrupts—it's task-level, not node-level. Choose based on your needs: LangGraph for fine-grained control, CrewAI for simplicity.

Designing Safe Approval Flows​

When designing human-in-the-loop workflows, consider:

  • Clarity: Show the human exactly what the agent is about to do and why.
  • Urgency: Indicate time-sensitive decisions (e.g., data deletion is permanent).
  • Context: Provide the agent's reasoning and any relevant supporting data.
  • Audit Trail: Log every human decision and the agent's response.
def format_approval_request(state: ApprovalState) -> str:
    """Format a clear approval request for a human."""
    return f"""
REQUEST FOR APPROVAL
====================
Action: {state['request']}

Reasoning: {state['reasoning']}

Impact: This action will affect 42 user records.

Approved by: [Select one]
- [ ] Approve
- [ ] Reject
- [ ] Approve with modifications

Timestamp: {datetime.now().isoformat()}
    """

Key Takeaways​

  • Interrupts pause an agent at critical points and wait for human feedback.
  • LangGraph interrupts are explicit node-level pauses; CrewAI callbacks are task-level hooks.
  • Thread IDs enable resumable workflows; same ID resumes from the pause point.
  • Web UIs expose approvals to humans; use checkpoints to persist state.
  • Clear formatting and audit trails are non-negotiable for high-stakes workflows.

Frequently Asked Questions​

How long can an agent wait for human approval?​

LangGraph doesn't enforce timeouts; you manage this at the application level. Set a timeout (e.g., 24 hours) and escalate if no response.

Can multiple humans approve a single action?​

Yes. Modify the state to track approvals from multiple users. For example, state["approvals"] = ["alice", "bob"]. Check that required approvers have signed off before proceeding.

What if a human rejects an action mid-workflow?​

The workflow pauses. You can either terminate the entire workflow or route to an alternative action (e.g., notify the agent to try a different approach).

Can an agent re-request approval after rejection?​

Yes. After rejection, the agent can reason about what went wrong and request again with a modified plan. Set a max retry count to prevent loops.

How do I audit human decisions?​

Log every approval request and response to a database or audit log. Include timestamp, human ID, decision, and reasoning. This is essential for compliance.

Further Reading​