Human-in-the-Loop Breakpoints and Interrupts

Some actions are too important to leave to the LLM:

• Sending an email.
• Issuing a refund.
• Deleting data.
• Posting to social media.

Human-in-the-loop (HITL) pauses execution before these actions, gets approval, then resumes. LangGraph makes this clean.

Pattern 1 — Interrupt before a node

The simplest pattern. Compile with interrupt_before:

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string("postgresql://...")

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_refund"],
)

When execution reaches the execute_refund node, it stops. State is saved. The caller gets control:

config = {"configurable": {"thread_id": "ticket-123"}}

# Run until interrupt
result = graph.invoke({"messages": [user_msg]}, config)

# Check if we're paused
state = graph.get_state(config)
if state.next:  # there's a next node to run
    next_node = state.next[0]
    print(f"Agent wants to execute: {next_node}")
    print(f"State: {state.values}")

    # Get human approval
    user_approves = ask_human(state.values)

    if user_approves:
        # Resume execution
        graph.invoke(None, config=config)
    else:
        # Update state to reflect rejection, then resume
        graph.update_state(
            config,
            {"messages": [HumanMessage("Refund rejected by reviewer")]},
        )
        graph.invoke(None, config=config)

The agent loops back into its decision node with the rejection note. It can revise its approach.

Pattern 2 — Dynamic interrupts (interrupt() function)

For more flexibility, interrupt within a node:

from langgraph.types import interrupt

def execute_refund_node(state):
    refund_data = {
        "order_id": state["order_id"],
        "amount": state["refund_amount"],
        "reason": state["refund_reason"],
    }

    # Interrupt and wait for human input
    approval = interrupt({
        "type": "refund_approval",
        "details": refund_data,
    })

    if approval == "approved":
        # Actually execute the refund
        result = refund_service.create(**refund_data)
        return {"messages": [AIMessage(f"Refund issued: {result['refund_id']}")]}
    else:
        return {"messages": [AIMessage("Refund rejected by reviewer")]}

When the node hits interrupt(), execution pauses. The state is saved with interrupt_value set to whatever was passed. Resumption:

graph.invoke(
    {"approval": "approved"},  # this becomes interrupt()'s return value
    config=config
)

This is more flexible than interrupt_before — you can interrupt mid-node, pass structured data both ways, and have multiple interrupts per node.

Pattern 3 — Conditional human-in-the-loop

Only require approval for certain conditions:

def maybe_request_approval(state):
    if state["refund_amount"] > 1000:
        return "human_review"
    return "execute_refund"

builder.add_conditional_edges("agent", maybe_request_approval)
builder.add_node("human_review", human_review_node)
builder.add_node("execute_refund", execute_refund_node)

For low-stakes (refund < $1000), auto-execute. For high-stakes, pause for review.

Pattern 4 — Feedback loop

Let the human suggest changes, then re-run:

def draft_email_node(state):
    draft = state["email_draft"]

    feedback = interrupt({
        "type": "draft_review",
        "draft": draft,
        "action_options": ["approve", "edit", "regenerate"],
    })

    if feedback["action"] == "approve":
        return {"final_email": draft}
    elif feedback["action"] == "edit":
        return {"email_draft": feedback["edited_content"], "iteration": state["iteration"] + 1}
    else:  # regenerate
        return {"messages": [HumanMessage(feedback["regenerate_instructions"])]}

The graph routes back to draft_email_node if the human asked for changes. Loops until approved.

Building the human-side UI

The HITL pattern requires a UI to:

• Show pending approvals.
• Display agent state and planned action.
• Provide approve/reject/edit buttons.
• Submit decision and resume.

# Pseudo-code for a Flask endpoint
@app.route("/agent/approve/<thread_id>", methods=["POST"])
def approve_action(thread_id):
    config = {"configurable": {"thread_id": thread_id}}
    decision = request.json["decision"]

    if decision == "approved":
        graph.invoke({"approval": "approved"}, config=config)
    else:
        graph.invoke({"approval": "rejected", "reason": request.json["reason"]}, config=config)

    return {"status": "resumed"}

A real system has:

• Queue of pending approvals (rows from graph.get_state_history where state.next is non-empty).
• Email/Slack notifications to reviewers.
• Audit log of all approvals.
• Timeout handling (if no human acts in N hours, auto-escalate).

When to use HITL

Always HITL for:

• Financial transactions (refunds, payments, transfers).
• Sending external communications (emails, social posts, customer messages above threshold).
• Database modifications (deletes, bulk updates).
• Actions affecting many users or high-value users.

Often HITL for:

• Complex customer support cases.
• Generating reports for executives.
• Anything where the cost of being wrong is high.

Usually skip HITL for:

• Read-only actions (lookups, retrieval).
• Internal logging.
• Low-stakes automation.

HITL and async workflows

Some HITL takes hours or days (e.g., manager approval). LangGraph handles this — checkpointed state lives in the DB until the human responds:

# Tuesday 10am: agent runs, hits interrupt, exits
graph.invoke({"messages": [user_msg]}, config)

# Tuesday 4pm: manager reviews via UI, approves
# (process running in another time/place)

graph.invoke({"approval": "approved"}, config)
# Picks up where it left off

The thread persists in Postgres/Redis. Resuming works hours or days later, possibly on a different machine.

Production HITL checklist

• Checkpointer set up (Postgres or Redis).
• interrupt_before or interrupt() at every action requiring approval.
• UI for pending approvals (queue, details, action buttons).
• Notification system (Slack/email to reviewers).
• Timeout handling (auto-escalate if no human acts).
• Audit log of all approval decisions.
• Test the full flow: agent → interrupt → human approves → agent resumes.
• Test the rejection path: agent → interrupt → human rejects → agent revises or ends gracefully.

Common HITL mistakes

1. No timeout. Agents wait indefinitely for humans who never respond.
2. No audit trail. Approvals happen but no record of who approved what when.
3. Approval UI shows raw state. Reviewers need a human-friendly summary, not JSON.
4. HITL for low-stakes actions. Friction without value.
5. No retry after rejection. Agent sees rejection but no path forward.

Takeaway

HITL is what makes high-stakes agents shippable. interrupt_before for pre-node interrupts; interrupt() for in-node flexibility; conditional routing for "only sometimes". Pair with a UI for reviewers, notifications, timeouts, and audit logs. Production HITL pattern is mature and well-supported.

---

Production Deep Dive: Modern interrupt() vs Legacy NodeInterrupt

In early LangGraph versions, pausing required raising a NodeInterrupt exception. LangGraph v0.2+ standardizes on the functional interrupt() pattern:

• Inside a node or tool: user_approval = interrupt({"prompt": "Confirm transfer of $500", "amount": 500})
• The graph execution halts immediately, writes the state snapshot to PostgreSQL, and returns control to the caller.
• To resume from an external API or CLI:

  from langgraph.types import Command
  graph.stream(Command(resume={"approved": True}), config={"configurable": {"thread_id": "session_1"}})
  

• This enables clean asynchronous human review across REST APIs without holding active worker threads open.