How to add dynamic breakpoints with interrupt

How to add dynamic breakpoints with `interrupt`#

!!! tip “Prerequisites”

This guide assumes familiarity with the following concepts:

* [Breakpoints](../../../concepts/breakpoints)
* [LangGraph Glossary](../../../concepts/low_level)
* [Human-in-the-loop conceptual guide](../../../concepts/human_in_the_loop)

Human-in-the-loop (HIL) interactions are crucial for agentic systems. Breakpoints are a common HIL interaction pattern, allowing the graph to stop at specific steps and seek human approval before proceeding (e.g., for sensitive actions).

In LangGraph you can add breakpoints before / after a node is executed. But oftentimes it may be helpful to dynamically interrupt the graph from inside a given node based on some condition. When doing so, it may also be helpful to include information about why that interrupt was raised.

This guide shows how you can dynamically interrupt the graph using the interrupt function from langgraph.types. Let’s see it in action!

%%capture --no-stderr
%pip install -U langgraph

Set up LangSmith for LangGraph development

Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started here.

Define the graph#

from IPython.display import Image, display
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

from langgraph.checkpoint.redis import RedisSaver

# Set up Redis connection
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp


class State(TypedDict):
    input: str


def step_1(state: State) -> State:
    print("---Step 1---")
    return state


def step_2(state: State) -> State:
    # Let's optionally raise an interrupt
    # if the length of the input is longer than 5 characters
    if len(state["input"]) > 5:
        # Use the new interrupt function instead of NodeInterrupt
        value = interrupt(
            {
                "reason": f"Input exceeds 5 characters: '{state['input']}'",
                "input_length": len(state["input"]),
                "input": state["input"]
            }
        )
        # If the interrupt is resumed with a value, we can use it
        # For example, the user could provide a shortened input
        if value and isinstance(value, str):
            return {"input": value}
    
    print("---Step 2---")
    return state


def step_3(state: State) -> State:
    print("---Step 3---")
    return state


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# Compile the graph with memory
graph = builder.compile(checkpointer=memory)

# View
display(Image(graph.get_graph().draw_mermaid_png()))

Run the graph with dynamic interrupt#

First, let’s run the graph with an input that is <= 5 characters long. This should safely ignore the interrupt condition we defined and run through the entire graph.

initial_input = {"input": "hello"}
thread_config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

If we inspect the graph at this point, we can see that there are no more tasks left to run and that the graph indeed finished execution.

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

Now, let’s run the graph with an input that’s longer than 5 characters. This should trigger the dynamic interrupt we defined via the interrupt function inside the step_2 node.

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "2"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

We can see that the graph now stopped while executing step_2. If we inspect the graph state at this point, we can see information about the interrupt and what node is set to execute next.

state = graph.get_state(thread_config)
print("Next node:", state.next)
print("Tasks:", state.tasks)

# Check if there are interrupts and display their information
if hasattr(state, 'interrupts') and state.interrupts:
    print("\nInterrupts:")
    for interrupt in state.interrupts:
        print(f"  - Value: {interrupt.value}")
        print(f"  - Resumable: {interrupt.resumable}")

If we try to resume the graph from the breakpoint without providing a new value, we will interrupt again as our inputs & graph state haven’t changed. However, with the new interrupt pattern, we can use the Command object to provide a value when resuming.

# Let's try resuming with a shorter input value using the Command object
# This will provide a value to the interrupt, which our step_2 function can use
for event in graph.stream(Command(resume="short"), thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print("Next node:", state.next)
print("Final state:", state.values)

Update the graph state#

To get around the interrupt, we can do several things:

Provide a shorter input via Command.resume: We can use Command(resume="foo") to provide a new value that the interrupt function will return, which our step_2 function uses to update the state.
Update the state directly: We can update the state to have an input that’s shorter than 5 characters before resuming.
Skip the node: We can update the state as the interrupted node to skip it altogether.

Let’s demonstrate updating the state directly:

# Start fresh with a new thread
initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "3"}}

# Run until interrupt
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

print("\n--- Interrupted due to long input ---\n")

# Update the state with a shorter input
graph.update_state(config=thread_config, values={"input": "foo"})
print("State updated with shorter input\n")

# Resume execution
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(f"\nFinal state: {state.values}")
print(f"Next nodes: {state.next}")

You can also update the state as node step_2 (the interrupted node) which would skip over that node altogether:

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "4"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)
    
print("\n--- Interrupted due to long input ---")

# Update the state as node `step_2` to skip it altogether
graph.update_state(config=thread_config, values=None, as_node="step_2")
print("Skipped step_2 by updating state as that node\n")

# Resume execution - this will go directly to step_3
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(f"\nFinal state: {state.values}")
print(f"Next nodes: {state.next}")

Using Command.resume to provide a value#

The most elegant way to handle interrupts is to use Command(resume=...) to provide a value that the interrupt function will return. This allows the interrupted node to handle the value appropriately:

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "5"}}

# Run the graph until the interruption
print("Running with long input...")
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

# Check the interrupt information
state = graph.get_state(thread_config)
if hasattr(state, 'interrupts') and state.interrupts:
    print("\n--- Interrupt Information ---")
    for interrupt in state.interrupts:
        print(f"Interrupt value: {interrupt.value}")

# Resume with a shorter value using Command
print("\n--- Resuming with shorter input via Command ---")
for event in graph.stream(Command(resume="hi"), thread_config, stream_mode="values"):
    print(event)

# Check final state
state = graph.get_state(thread_config)
print(f"\nFinal state: {state.values}")
print(f"Completed: {state.next == ()}")

Summary#

This guide demonstrated how to use the interrupt function from langgraph.types to dynamically interrupt graph execution based on conditions within a node.

Key takeaways:

Use interrupt(value) to pause execution and surface information to humans
The value passed to interrupt can be any JSON-serializable data
Resume execution using Command(resume=...) to provide input back to the interrupted node
You can also update state directly or skip nodes using update_state()

This pattern is particularly useful for:

Requesting human approval for sensitive actions
Getting human input when the graph needs clarification
Implementing conditional breakpoints based on runtime state
Building human-in-the-loop workflows with dynamic decision points

Version

How to add dynamic breakpoints with interrupt

Contents

How to add dynamic breakpoints with `interrupt`#

Define the graph#

Run the graph with dynamic interrupt#

Update the graph state#

Using Command.resume to provide a value#

Summary#

Version

How to add dynamic breakpoints with interrupt

Contents

How to add dynamic breakpoints with interrupt#

Define the graph#

Run the graph with dynamic interrupt#

Update the graph state#

Using Command.resume to provide a value#

Summary#

How to add dynamic breakpoints with `interrupt`#