Skip to content

Human-in-the-Loop (HITL)

1. Why this matters

Autonomous agents are great until they do something dumb or dangerous:

  • Send the wrong email to the wrong customer.
  • Approve a $50,000 wire transfer that should have been $500.
  • Delete production data because a tool misfired.

For high-stakes actions, you want a human gate. HITL gives you a first-class way to insert that gate without breaking the rest of the graph.

2. Mental model

A LangGraph interrupt is a structured pause that:

  1. Stops the engine before/after a specific node.
  2. Persists the snapshot via the checkpointer.
  3. Returns control to your application code.
  4. Lets you inspect state, ask the user, possibly edit state.
  5. Resume the graph from where it stopped.
flowchart LR
    A[step 1] --> B[step 2 risky]
    B -.pause before.-> H[Checkpoint]
    H --> Hum[Human reviews<br/>approves / edits / rejects]
    Hum -->|approve & resume| B
    Hum -->|edit state, resume| B
    Hum -->|reject → END| EN((END))

3. Architecture / Flow

flowchart TD
    INV[invoke with thread_id] --> RUN[Run nodes]
    RUN --> CHK{Reach interrupt point?}
    CHK -->|no| END([End normally])
    CHK -->|yes| PAUSE[Save snapshot, return partial result]
    PAUSE --> APP[Your app code]
    APP -->|inspect get_state| Q{Decision}
    Q -->|approve| RESUME[invoke None, config]
    Q -->|edit| EDIT[update_state, then resume]
    Q -->|abort| ABORT[clear thread]
    RESUME --> RUN
    EDIT --> RUN

4. Core concepts

  • interrupt_before=["node_name"] at compile time — pause every time before that node runs.
  • interrupt_after=[...] — pause after the node finishes (less common).
  • Dynamic interrupt(value) — call from inside a node to pause based on logic. Returns the human's response when the graph is resumed via Command(resume=...).
  • graph.get_state(config) — at a pause, inspect the snapshot to know what's about to happen (.next tells you which node is queued).
  • graph.update_state(config, values, as_node=...) — modify state mid-pause. as_node decides which reducer logic applies.
  • Resume — pass None as input to .invoke(None, config); the engine continues from the latest checkpoint.
  • Requires a checkpointer. HITL relies entirely on persistence. No checkpointer → no interrupts.

5. Code — minimal working example

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class S(TypedDict):
    approved: bool
    result: str

def prepare(state: S):
    return {"result": "Drafted: 'Send refund of $500 to user 42.'"}

def execute(state: S):
    return {"result": state["result"] + "  ← EXECUTED"}

b = StateGraph(S)
b.add_node("prepare", prepare)
b.add_node("execute", execute)
b.add_edge(START, "prepare")
b.add_edge("prepare", "execute")
b.add_edge("execute", END)

graph = b.compile(checkpointer=MemorySaver(), interrupt_before=["execute"])

cfg = {"configurable": {"thread_id": "tx-99"}}

# First invoke — runs prepare, then pauses before execute
out = graph.invoke({"approved": False, "result": ""}, config=cfg)
print(out)         # {'approved': False, 'result': 'Drafted: ...'}
print(graph.get_state(cfg).next)   # ('execute',)

# --- HUMAN GATE ---
human_approves = True   # imagine a Slack button click here
if not human_approves:
    print("Aborted.")
else:
    # Resume by passing None (no new input — keeps the existing state)
    final = graph.invoke(None, config=cfg)
    print(final)   # 'Drafted: ...  ← EXECUTED'

6. Code — real-world pattern

Editable HITL — human can change the draft email before it goes out:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import sqlite3

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

class EmailState(TypedDict):
    request: str
    draft: str
    sent: bool

def draft_email(state: EmailState):
    out = llm.invoke([
        SystemMessage("Write a short, polite customer email."),
        HumanMessage(state["request"]),
    ])
    return {"draft": out.content}

def send_email(state: EmailState):
    # actually call SendGrid / SES here
    print("→ SENDING:", state["draft"][:80], "...")
    return {"sent": True}

b = StateGraph(EmailState)
b.add_node("draft", draft_email)
b.add_node("send", send_email)
b.add_edge(START, "draft")
b.add_edge("draft", "send")
b.add_edge("send", END)

cp = SqliteSaver(sqlite3.connect("hitl.db", check_same_thread=False))
graph = b.compile(checkpointer=cp, interrupt_before=["send"])

cfg = {"configurable": {"thread_id": "ticket-1234"}}

# 1. Generate draft and pause
graph.invoke({"request": "Apologize for delayed shipment, offer 10% off.",
              "draft": "", "sent": False}, config=cfg)

snap = graph.get_state(cfg)
print("DRAFT:", snap.values["draft"])
print("NEXT :", snap.next)         # ('send',)

# 2. Human edits the draft via update_state
graph.update_state(
    cfg,
    {"draft": snap.values["draft"] + "\n\nP.S. We've also expedited your replacement."},
    as_node="draft",                # apply 'draft' node's reducer behavior
)

# 3. Resume — runs `send` with the edited draft
final = graph.invoke(None, config=cfg)
print("SENT?", final["sent"])

Dynamic interrupt inside a node (pause only when an amount is above a threshold):

from langgraph.types import interrupt, Command

def execute_transfer(state):
    if state["amount"] > 1000:
        # Pause and wait for human input. The human's reply replaces this expression.
        approval = interrupt({"reason": "high amount", "amount": state["amount"]})
        if not approval.get("approve"):
            return {"status": "cancelled"}
    # ... actually transfer
    return {"status": "done"}

# Resume with the human's answer
graph.invoke(Command(resume={"approve": True}), config=cfg)

7. Common pitfalls

  • ❗ Compiling without a checkpointer. Interrupts depend on the snapshot store. No checkpointer → interrupts don't work.
  • ❗ Hitting interrupt_before and forgetting to resume. State is stuck; no work happens. Always have a path that resumes or explicitly cancels.
  • ❗ Calling .invoke(state, cfg) to resume with new input. That replaces / merges state; the engine still resumes but you may have clobbered something. To pure-resume, pass None.
  • ❗ Editing the wrong field via update_state. Make sure as_node matches the node whose reducer behavior you want — important for add_messages-style fields.
  • ❗ Security: trusting user-supplied edits. Treat them like any external input — validate before resuming. A malicious "approve" can be expensive.
  • ❗ Long-pause leaks. Threads paused for hours can pile up. Implement a TTL / cleanup for stale paused threads.

8. When to use vs not use

Use HITL when Skip when
Action is destructive (delete, send, charge) Pure read-only / analysis
Output requires expert judgment Confidence is high enough for automation
Compliance requires human approval Internal-only, low-stakes
You're in early rollout / building trust Mature, well-evaluated agent

9. Cheatsheet

# Static interrupt at compile time
graph = builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["risky_node"],     # or list of node names
    interrupt_after=["audit_node"],
)

# Dynamic interrupt inside a node
from langgraph.types import interrupt, Command

def my_node(state):
    answer = interrupt({"question": "Proceed?", "data": state["x"]})
    # …after resume, answer is whatever was passed in Command(resume=...)

# Resume / control commands
graph.invoke(None, config=cfg)                       # plain resume
graph.invoke(Command(resume=human_response), config=cfg)   # answer a dynamic interrupt
graph.update_state(cfg, {"key": "new"}, as_node="x")  # edit state mid-pause

# Inspect pause point
snap = graph.get_state(cfg)
snap.next            # tuple of nodes about to run (empty if at END)
snap.values          # current state dict
snap.config          # includes checkpoint_id

# Time-travel: replay or branch from earlier checkpoint
old = list(graph.get_state_history(cfg))[5]
graph.invoke(None, config=old.config)

10. Q&A — recall test

  • Q: Two ways to pause a LangGraph run? A: Static (interrupt_before=[...] at compile time) and dynamic (call interrupt(...) inside a node).

  • Q: How do you resume? A: graph.invoke(None, config=cfg) — passing None means "no new input, just resume from latest checkpoint."

  • Q: What is update_state for? A: Editing state during a pause — e.g., a human corrects the draft before the next node consumes it.

  • Q: What does as_node="x" do? A: Applies the reducer behavior that node x would apply. Matters for fields with reducers (e.g., add_messages).

  • Q: Why must you compile with a checkpointer for HITL? A: Interrupts mean stopping mid-graph and resuming later. Without persistence, there's nothing to resume from.

  • Q: How do you respond to a dynamic interrupt(value)? A: graph.invoke(Command(resume=your_answer), config=cfg). The your_answer becomes the return value of interrupt(...) inside the node.

Practice

What does this print?

Expected: True

# HITL: graph pauses → human reviews → human approves/edits → graph resumes
states = ["running", "paused_for_human", "resumed", "completed"]
print("paused_for_human" in states)

Resume with Command(resume=value), not just calling invoke again

Expected: True

# Pattern: pass the human's response via Command(resume=...)
use_command = False           # bug: must use Command(resume=...) to inject the response
print(not use_command)

Quiz — Quick check

What you remember

Q1. What does interrupt() do inside a node?

  • Pauses the graph and returns control to the caller — waits for human input before continuing
  • Cancels the graph
  • Throws an exception
  • Logs a warning

Why: HITL is built on interrupts. Inside a node, interrupt(value) halts execution and surfaces value to the caller. The next invoke(Command(resume=...)) resumes the node with the supplied value as interrupt's return.

Q2. What's the prerequisite for using HITL?

  • A checkpointer must be configured — the graph needs to save state before pausing
  • An external database
  • Async execution
  • Streaming enabled

Why: Without a checkpointer, the paused state would be lost. The checkpointer stores the state at the interrupt point so a later invoke can resume from there.

Q3. Why use HITL for tool-using agents?

  • Sensitive actions (sending emails, executing code, transferring money) should be human-approved
  • LLMs are always wrong
  • Required by sklearn
  • For speed

Why: Autonomous agents are great until they aren't. HITL adds a human checkpoint before irreversible actions — risk management for production. The pattern is "agent proposes, human approves".

Common doubts

How does the frontend know the graph paused?

The graph's .invoke() or .astream() returns an interrupt event. Your endpoint sees this, returns a "needs approval" response to the frontend with the proposed action. The frontend renders an approve/edit UI; on submit, calls back with Command(resume=user_response).

Can I have multiple interrupt points in one graph?

Yes — any node can call interrupt(). Common pattern: an agent proposes a tool call → interrupt → human approves/edits → tool executes → continue. The graph remembers where it was at each pause.

What if the user never responds?

The graph state remains in the checkpointer indefinitely. To clean up: (1) set a TTL on threads in your storage layer, (2) add a periodic job that abandons threads paused for >N days, (3) at resume time, check if the proposed action is still valid (data may have changed).