LangGraph Human-in-the-Loop: interrupt(), State Edits and Time Travel

Everything below assumes you know the standard seven-part layout of a LangGraph file (imports, state, tools, nodes, edges, assembly, entrypoint) and have used checkpointers with a thread_id. Each pattern here sits on top of a checkpointer. Without one, none of it works.

Why a human needs to stay in the loop

The moment an LLM can act, by sending an email, deleting a file, charging a card or pushing code, it can also make an expensive mistake at machine speed while nobody is looking. Human-in-the-loop (HITL) isn't a bonus feature stapled onto an agent. It's the design that makes agents with real-world tools safe to ship, and LangGraph was built with that problem in mind. Its checkpointing system exists largely to make pauses possible.

Four patterns cover most real systems, listed by how often they turn up:

1. Pausing before a risky action with interrupt(), the current standard.
2. A guaranteed review checkpoint in a workflow.
3. Editing state while paused, to correct the agent before it continues.
4. Time travel: replaying and branching from any earlier point in a conversation.

A few things are left out on purpose because they rarely appear in real codebases: token-level streaming internals, the full LangGraph Studio SDK and the older NodeInterrupt exception, which interrupt() replaced.

Interrupt and resume

Each pattern reduces to the same two-step move:

1. interrupt()  →  Graph pauses. State is saved. Control returns to your code.
2. Command(resume=...)  →  Graph picks up exactly where it paused, with your input.

Everything else is variation: where you call interrupt(), what data you hand to it and what you do with the answer.

Vocabulary

interrupt(value) is imported from langgraph.types and called inside a node. It halts the graph right there, saves the entire state through the checkpointer and hands value to whoever started the run. The pause is real. No thread spins in the background and no resources are held, so a thread can sit paused for months.

Command(resume=value) goes into graph.invoke() or graph.stream() to wake a paused graph. The value becomes the return value of the interrupt() call, as though interrupt() were a synchronous input() that just got an answer.

__interrupt__ is the key that shows up in the output when execution has paused. Check for it to learn whether the agent stopped and wants input.

The checkpointer (MemorySaver, SqliteSaver, PostgresSaver) is mandatory. With none, there's no saved state to pause or resume. Forgetting to compile the graph with one is the most frequent reason beginners' HITL code fails.

Pattern 1: approve a tool call before it runs

By far the most common production use. The agent decides to call a tool that has real effects, like sending an email, querying a database or placing an order, and a person has to approve it first.

Picture a manager co-signing a check. The employee (the LLM) fills it out and hands it over, and nothing moves until the manager (the human) signs. If the manager refuses, the check is voided.

The review node:

# ── IMPORTS needed ───────────────────────────────────────────
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
# ── MODULE 4: NODES ───────────────────────────────────────────
def review_tool_call(state: AgentState) -> dict:
    """Pauses execution to let a human approve, edit, or reject a tool call
    before it actually runs."""

    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]  # Assume one tool call for simplicity

    # interrupt() pauses HERE. The dict below is shown to whoever resumes the graph.
    human_response = interrupt({
        "question": "Approve this action?",
        "tool_name": tool_call["name"],
        "tool_args": tool_call["args"],
    })

    # Execution resumes here, with human_response being whatever
    # was passed to Command(resume=...)

    if human_response["type"] == "approve":
        # Do nothing - let the tool call proceed as-is
        return {}

    elif human_response["type"] == "edit":
        # Replace the tool call's arguments with the human's edits
        last_message.tool_calls[0]["args"] = human_response["edited_args"]
        return {"messages": [last_message]}

    elif human_response["type"] == "reject":
        # Short-circuit: skip the tool entirely, tell the LLM why
        return {
            "messages": [
                ToolMessage(
                    content="User rejected this action.",
                    tool_call_id=tool_call["id"]
                )
            ]
        }

Then wire it into the graph:

graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
# Agent decides: call a tool (go through review) or finish
graph_builder.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "review_tool_call", "__end__": END}
)
graph_builder.add_edge("review_tool_call", "tools")
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

Routing changes slightly from the basic agent: a tool request now goes to review_tool_call before the tools node. The agent's choice and the actual execution have the reviewer in between, which is the co-signature step before cashing.

Run it and resume it:

config = {"configurable": {"thread_id": "session-001"}}
# ── First call: runs until interrupt() pauses it ─────────────
result = graph.invoke(
    {"messages": [HumanMessage(content="Send an email to the team about the delay")]},
    config=config
)
# Check if the graph paused
if "__interrupt__" in result:
    interrupt_data = result["__interrupt__"][0].value
    print(f"Agent wants to: {interrupt_data['tool_name']}")
    print(f"With arguments: {interrupt_data['tool_args']}")

    # Show this to a human (in a real app: render a UI button, Slack message, etc.)
    user_decision = input("Approve? (yes/no): ")

    # ── Resume with the human's decision ──────────────────────
    if user_decision.lower() == "yes":
        final_result = graph.invoke(
            Command(resume={"type": "approve"}),
            config=config
        )
    else:
        final_result = graph.invoke(
            Command(resume={"type": "reject"}),
            config=config
        )

    print(final_result["messages"][-1].content)

That's the whole pattern. What changes elsewhere is the content passed to interrupt() and what you do with the answer that comes back.

Pattern 2: a mandatory feedback gate

Sometimes you don't want a conditional pause depending on what the LLM chose. You want a pause at a fixed stage, every time. A content-generation agent that always stops for review before publishing is the usual example, regardless of what it wrote.

Think of a quality-control station on an assembly line. Each item stops there, no exceptions. The worker can pass it as is, send it back for rework or adjust it directly.

The feedback node:

def human_feedback(state: AgentState) -> dict:
    """A guaranteed pause point. Shows the human the latest draft
    and lets them approve, request changes, or edit directly."""

    draft = state["messages"][-1].content

    feedback = interrupt({
        "question": "Review this draft. Approve, or provide edit instructions.",
        "draft": draft,
    })

    if feedback.get("approved"):
        return {}  # Pass through unchanged

    # Feedback contains revision instructions — feed back to the agent
    return {
        "messages": [
            HumanMessage(content=f"Please revise based on this feedback: {feedback['notes']}")
        ]
    }

And the loop wiring:

graph_builder = StateGraph(AgentState)
graph_builder.add_node("writer", writer_node)
graph_builder.add_node("human_feedback", human_feedback)
graph_builder.set_entry_point("writer")
graph_builder.add_edge("writer", "human_feedback")
# After feedback, loop back to writer for revisions, OR end if approved
graph_builder.add_conditional_edges(
    "human_feedback",
    lambda state: "writer" if state["messages"][-1].content.startswith("Please revise") else "__end__",
    {"writer": "writer", "__end__": END}
)
graph = graph_builder.compile(checkpointer=MemorySaver())

This is the review-and-revise cycle found in nearly every content, code-review or research agent: generate, pause for a person, revise if needed, pause again, repeat until approved.

Pattern 3: fix the state by hand

At times a person doesn't want to approve or reject. They want to repair something in the agent's state before it goes on, because the agent was slightly off and retrying is slower than correcting it.

Three calls matter here:

• graph.get_state(config) returns the current StateSnapshot, with .values (the state dict) and .next (the node that would run on resume).
• graph.update_state(config, values) writes new values at the current checkpoint. It creates a new checkpoint and leaves history intact, which matters later for time travel.
• as_node="node_name" is an optional argument to update_state that tells LangGraph to treat the update as coming from that node. LangGraph picks the next edges based on the source node of the last update, so getting this wrong can resume the graph somewhere unexpected.

In practice:

# ── Pause first (using interrupt_before at compile time) ────
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["agent"]  # Always pause before the agent node runs
)
config = {"configurable": {"thread_id": "session-002"}}
# Run until the breakpoint
result = graph.invoke({"messages": [HumanMessage(content="What's 5 + 5?")]}, config)
# ── Inspect what's about to happen ───────────────────────────
snapshot = graph.get_state(config)
print(f"About to run: {snapshot.next}")  # e.g. ('agent',)
print(f"Current messages: {snapshot.values['messages']}")
# ── Directly correct the state ───────────────────────────────
# Suppose the human wants to inject a clarification before the agent responds
graph.update_state(
    config,
    {"messages": [HumanMessage(content="Actually, I meant 5 * 5, not 5 + 5")]},
)
# ── Resume - agent now sees the corrected message ────────────
final = graph.invoke(None, config)  # None = "just resume, no new input"
print(final["messages"][-1].content)

Two distinctions trip people up. First, interrupt_before=["node_name"] is set when you compile and always pauses ahead of that node on every run, while interrupt() lives inside a node and pauses only when your logic says so. Use interrupt_before for simple debugging or guaranteed stops, and interrupt() whenever the pause depends on runtime conditions, which covers most real applications.

Second, graph.invoke(None, config) means "add no new input, continue from the last checkpoint." That's different from Command(resume=...), which answers a pending interrupt(). After interrupt_before plus update_state, resume with None. After interrupt(), resume with Command(resume=...).

Pattern 4: time travel

Checkpoint history works like commits in Git. You can look at any past state and branch from it to make an alternate timeline without disturbing the original. Another comparison: save points in a video game. get_state_history lists every one, and you can load any of them, not just the latest.

• graph.get_state_history(config) is a generator that yields every StateSnapshot saved for the thread, newest first. Each carries its own config with a unique checkpoint_id.
• checkpoint_id lives at snapshot.config["configurable"]["checkpoint_id"]. Pass a config containing a specific one to invoke or update_state and the operation targets that point, not the most recent.
• Replay means calling graph.invoke(None, config_with_old_checkpoint_id). It re-executes from that point forward. Checkpoints already stored for that branch are reused rather than recomputed, though node logic after that point runs again.
• Branch means calling update_state on an old checkpoint, which forks a new checkpoint and a new timeline alongside the first.

config = {"configurable": {"thread_id": "session-003"}}
# Run a few turns normally...
graph.invoke({"messages": [HumanMessage(content="My name is Sam")]}, config)
graph.invoke({"messages": [HumanMessage(content="What's my name?")]}, config)
# ── Find a past checkpoint ────────────────────────────────────
history = list(graph.get_state_history(config))
for snapshot in history:
    print(f"Step {snapshot.metadata['step']}: {len(snapshot.values['messages'])} messages")
# Pick an earlier checkpoint - say, right after "My name is Sam"
target_checkpoint = history[-2]  # Adjust index based on what you printed above
target_config = target_checkpoint.config
# ── Branch: update state AT that old checkpoint ──────────────
# This creates a NEW checkpoint that forks from the old one.
# The original conversation history is untouched.
branch_config = graph.update_state(
    target_config,
    {"messages": [HumanMessage(content="Actually, my name is Alex")]},
)
# ── Continue from the new branch ──────────────────────────────
result = graph.invoke(None, branch_config)
print(result["messages"][-1].content)
# Agent now responds based on the BRANCHED history (name = Alex),
# while the original thread (name = Sam) still exists if you go back to it.

Most chatbots don't need this at runtime. Its value is in three places. Debugging: replay exactly what happened in a failed run to find where it broke. Evaluation: rerun the same conversation with another prompt or model and compare, which is A/B testing without replaying everything. Error recovery: when a node fails halfway, rewind to just before it, fix the cause and resume instead of restarting the whole conversation.

A full agent with tool approval

Here's a complete template combining the basic structure, memory and tool-call approval. It's the realistic shape of a production agent with a human safety net.

# ============================================================
# LANGGRAPH AGENT WITH HUMAN-IN-THE-LOOP TOOL APPROVAL
# Extends: Part 1 (core structure) + Part 2 (memory)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── MODULE 2: STATE ──────────────────────────────────────────
class State(MessagesState):
    pass  # messages field is inherited; add custom fields if needed
# ── MODULE 3: TOOLS ──────────────────────────────────────────
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email. This is a SENSITIVE action requiring human approval."""
    return f"Email sent to {to} with subject '{subject}'"
tools = [send_email]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ──────────────────────────────────────────
def agent_node(state: State) -> dict:
    messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}
def review_tool_call(state: State) -> dict:
    """Pauses for human approval before any tool executes."""

    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]

    decision = interrupt({
        "tool_name": tool_call["name"],
        "tool_args": tool_call["args"],
    })

    if decision["type"] == "reject":
        return {
            "messages": [
                ToolMessage(
                    content="Action rejected by user.",
                    tool_call_id=tool_call["id"]
                )
            ]
        }

    if decision["type"] == "edit":
        last_message.tool_calls[0]["args"] = decision["edited_args"]
        return {"messages": [last_message]}

    return {}  # approved - pass through unchanged
# ── MODULE 5: ROUTING ─────────────────────────────────────────
def should_continue(state: State) -> Literal["review_tool_call", "__end__"]:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "review_tool_call"
    return "__end__"
def after_review(state: State) -> Literal["tools", "agent"]:
    """If the tool call was rejected, the last message is a ToolMessage -
    skip the tools node and go straight back to the agent."""
    last_message = state["messages"][-1]
    if isinstance(last_message, ToolMessage):
        return "agent"
    return "tools"
# ── MODULE 6: GRAPH ASSEMBLY ──────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges(
    "agent", should_continue,
    {"review_tool_call": "review_tool_call", "__end__": END}
)
graph_builder.add_conditional_edges(
    "review_tool_call", after_review,
    {"tools": "tools", "agent": "agent"}
)
graph_builder.add_edge("tools", "agent")
graph = graph_builder.compile(checkpointer=MemorySaver())
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "session-001"}}

    result = graph.invoke(
        {"messages": [HumanMessage(content="Email bob@example.com that the meeting moved to 3pm")]},
        config
    )

    while "__interrupt__" in result:
        interrupt_info = result["__interrupt__"][0].value
        print(f"\n Approval needed:")
        print(f"  Tool: {interrupt_info['tool_name']}")
        print(f"  Args: {interrupt_info['tool_args']}")

        choice = input("Approve (a) / Edit (e) / Reject (r)? ").strip().lower()

        if choice == "a":
            result = graph.invoke(Command(resume={"type": "approve"}), config)
        elif choice == "e":
            new_to = input("New 'to' address: ")
            result = graph.invoke(
                Command(resume={
                    "type": "edit",
                    "edited_args": {**interrupt_info["tool_args"], "to": new_to}
                }),
                config
            )
        else:
            result = graph.invoke(Command(resume={"type": "reject"}), config)

    print(f"\nAgent: {result['messages'][-1].content}")

Quick reference

Pausing and resuming. interrupt(value) pauses inside a node, saves state and hands value to the caller, then returns the resume value later. Command(resume=value) answers a pending interrupt. "__interrupt__" is the output key signaling a pause, and result["__interrupt__"][0].value shows what was surfaced. interrupt_before=["node_name"] always pauses ahead of the named node.

Editing state. graph.get_state(config) gives a StateSnapshot (.values, .next, .config). graph.update_state(config, values) adds a checkpoint without erasing history. as_node influences which edges are followed on resume. graph.invoke(None, config) continues without new input.

Time travel. graph.get_state_history(config) lists snapshots newest first. checkpoint_id identifies a save point. Replay continues a timeline from an old checkpoint, and branch forks a new one.

Which pattern fits

• Tools that spend money, send messages or change external systems: approval with interrupt(). For production agents with real side effects, treat it as required.
• A content step that must always be reviewed before the workflow advances: the dedicated feedback node with a revise loop.
• A person correcting or adding information mid-conversation with no formal approval step: interrupt_before plus update_state.
• Debugging a failed run, testing prompt changes or recovering from an error without restarting: time travel with get_state_history.

Safety is the shape of the graph

Memory, summarization, persistence and human oversight are all expressed as structure in LangGraph: a node here, an edge there, a field in state. There's no special safety mode to flip on. Safety is another shape the graph can take.

Agents rarely fail in production because the LLM made a bad call, which is expected and recoverable. They fail when a bad call had no checkpoint between the decision and its consequence. interrupt() is that checkpoint. It costs one extra node and one extra edge, and it may be the most valuable node in the graph.

With a standard structure, a way to keep conversations affordable and persistent, and a safety net around anything that matters, you have the scaffold under nearly every serious LangGraph application. The natural next topics are multi-agent orchestration with supervisors and subgraphs, and long-term memory such as cross-session user profiles, both of which build on state, nodes and checkpointers.