Implementation:Langchain ai Langgraph Interrupt Function

Property	Value
API	interrupt(value: Any) -> Any
Type	API Doc
Workflow	Human_in_the_Loop_Agent
Pipeline Stage	Interrupt Declaration
Repository	Langchain_ai_Langgraph
Source	libs/langgraph/langgraph/types.py:L420-543
Also covers	HumanInterrupt TypedDict at libs/prebuilt/langgraph/prebuilt/interrupt.py:L51-85

Overview

The interrupt() function pauses graph execution from within a node and surfaces a value to the client. On the first call, it raises a GraphInterrupt exception. When the graph is resumed with a Command(resume=...), the node re-executes and interrupt() returns the resume value instead of pausing.

The HumanInterrupt TypedDict provides a structured schema for interrupt values in human-in-the-loop patterns, describing the action being requested, allowed response types, and a description.

Description

The interrupt() function performs the following logic:

1. Retrieves the current configuration from the execution context via get_config().
2. Increments an interrupt counter in the scratchpad to track the position of this interrupt within the node.
3. Checks if a resume value exists at the current index in scratchpad.resume. If so, it sends the resume values to the output channel and returns the value at the current index.
4. If no prior resume value exists, checks for a "null resume" value (set during the resume phase). If found, appends it to the resume list and returns it.
5. If no resume value is found at all, raises a GraphInterrupt containing an Interrupt object with the provided value and an ID derived from the checkpoint namespace.

The HumanInterrupt TypedDict provides a structured convention for interrupt values:

• action_request: An ActionRequest TypedDict containing action (str) and args (dict)
• config: A HumanInterruptConfig TypedDict with boolean flags: allow_ignore, allow_respond, allow_edit, allow_accept
• description: An optional string describing what input is needed

Usage

from langgraph.types import interrupt, Command
from langgraph.prebuilt import HumanInterrupt

# Simple interrupt with a string value
def my_node(state):
    answer = interrupt("What is your age?")
    return {"human_value": answer}

# Structured interrupt using HumanInterrupt
def review_node(state):
    response = interrupt([HumanInterrupt(
        action_request=ActionRequest(
            action="approve_action",
            args={"tool": "send_email", "to": "user@example.com"}
        ),
        config=HumanInterruptConfig(
            allow_ignore=True,
            allow_respond=True,
            allow_edit=False,
            allow_accept=True,
        ),
        description="Please review the email action before execution"
    )])[0]
    return {"response": response}

Code Reference

Source Location

Item	Path
interrupt() function	libs/langgraph/langgraph/types.py, Lines 420-543
Interrupt dataclass	libs/langgraph/langgraph/types.py, Lines 159-214
HumanInterrupt TypedDict	libs/prebuilt/langgraph/prebuilt/interrupt.py, Lines 51-85
HumanInterruptConfig TypedDict	libs/prebuilt/langgraph/prebuilt/interrupt.py, Lines 11-27
ActionRequest TypedDict	libs/prebuilt/langgraph/prebuilt/interrupt.py, Lines 33-45

Signature

def interrupt(value: Any) -> Any:

class HumanInterrupt(TypedDict):
    action_request: ActionRequest
    config: HumanInterruptConfig
    description: str | None

Import

from langgraph.types import interrupt
from langgraph.prebuilt import HumanInterrupt

I/O Contract

interrupt()

Parameter	Type	Default	Description
value	Any	(required)	The value to surface to the client when the graph is interrupted. Can be a string, dict, list, or any serializable Python object.

Returns: Any -- On subsequent invocations (after resume), returns the value provided via Command(resume=...).

Raises: GraphInterrupt -- On the first invocation within the node, halts execution and surfaces the provided value to the client.

Interrupt dataclass

Attribute	Type	Description
value	Any	The value associated with the interrupt.
id	str	The deterministic ID of the interrupt, derived from the checkpoint namespace via xxh3_128_hexdigest. Can be used to resume the interrupt directly.

HumanInterrupt TypedDict

Field	Type	Description
action_request	ActionRequest	The specific action being requested from the human.
config	HumanInterruptConfig	Configuration defining what response actions are allowed.
description	None	Optional detailed description of what input is needed.

Usage Examples

Basic interrupt with resume

import uuid
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

class State(TypedDict):
    foo: str
    human_value: str | None

def node(state: State):
    answer = interrupt("what is your age?")
    return {"human_value": answer}

builder = StateGraph(State)
builder.add_node("node", node)
builder.add_edge(START, "node")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": uuid.uuid4()}}

# First invocation: graph pauses at interrupt
for chunk in graph.stream({"foo": "abc"}, config):
    print(chunk)
# Output: {'__interrupt__': (Interrupt(value='what is your age?', ...),)}

# Second invocation: resume with human input
for chunk in graph.stream(Command(resume="25"), config):
    print(chunk)
# Output: {'node': {'human_value': '25'}}

Multiple interrupts in a single node

def multi_interrupt_node(state):
    name = interrupt("What is your name?")
    age = interrupt("What is your age?")
    return {"name": name, "age": age}

When resumed, each resume value maps to the corresponding interrupt by position. The first Command(resume=...) resolves the first interrupt; the second resolves the second.

Related Pages

• Langchain_ai_Langgraph_Interrupt_Definition
• Heuristic:Langchain_ai_Langgraph_Deprecation_Migration_Guide
• Langchain_ai_Langgraph_Command_Resume
• Langchain_ai_Langgraph_Pregel_Stream_With_Interrupts