Implement LangGraph Human in the Loop: Interrupt and Resume Patterns Explained

Mastering LangGraph’s Human in the Loop: Interrupt and Resume Patterns Explained

Imagine you’re building a smart assistant. Sometimes, this assistant needs a little help from you. Maybe it needs approval before sending an email, or it asks for a specific piece of information only you know. This is where LangGraph’s special interrupt and resume patterns come in handy.

This guide will show you how to pause your smart assistant, let a human step in, and then smoothly continue the assistant’s work. You’ll learn all about how to implement these powerful langgraph interrupt resume patterns in your projects. We’ll explore various ways to stop a process, get human input, and kickstart it again.

What is LangGraph and Why Does It Need You?

LangGraph is like a blueprint for your smart assistant. It lets you draw out steps and decisions your assistant will make. Think of it as a flowchart for AI.

Each step in this flowchart is called a “node,” and the arrows between them are “edges.” Your assistant follows these steps to get things done, always keeping track of its current “state.”

Sometimes, your assistant might reach a point where it can’t move forward without human help. This is where the concept of Human-in-the-Loop (HITL) comes in. You, the human, become an important part of the assistant’s brain, especially when dealing with complex tasks or sensitive decisions.

Why Interrupt? Real-World Scenarios Where You Step In

Pausing your LangGraph assistant isn’t about it breaking down; it’s about smart design. There are many times your AI system will benefit from your intervention. You might need to step in for several good reasons.

Think about these common situations where Graph interruption mechanics become essential. Your AI might need an approval, more information, or a correction. All these scenarios require you to jump in and guide the process.

When a Human Needs to Approve Something

Imagine your AI assistant has drafted a sensitive email to a customer. Before sending it, it’s wise to have a human review and approve the content. You wouldn’t want an AI to accidentally send something wrong or embarrassing.

This is a classic case for an interrupt node configuration. The AI pauses, waits for your “OK” or “No,” and only then proceeds. This pattern ensures quality control and prevents costly mistakes.

When a Human Needs to Provide Missing Information

Sometimes, the AI might not have all the data it needs to complete a task. For example, if it’s booking a trip, it might need your specific travel dates or budget. The AI can ask you directly.

In such cases, the handling user input mechanism allows the AI to pause, ask for the missing details, and then incorporate them into its state updates after approval. This way, you effectively fill in the gaps for the AI. You become the source of critical data.

When a Human Needs to Correct an AI’s Mistake

Even the smartest AI can make mistakes or misunderstand a request. If your AI generates a report with incorrect figures or misinterprets a customer’s query, you need a way to step in. You can correct the error.

By interrupting the process, you can provide the correct information or steer the AI back on track. This error recovery patterns approach ensures the AI’s output is always accurate and reliable, learning from your corrections.

The Basics of Graph Interruption Mechanics

LangGraph is built to be flexible, and that includes pausing. When you design your graph, you can tell it to stop at certain points. This is like putting a “stop sign” in your assistant’s flowchart.

The key to understanding this is knowing about checkpoint management and the graph’s state preservation during pause. These features ensure that when the graph stops, it doesn’t forget where it was or what it was doing. When you resume from interrupt, it picks up exactly where it left off.

Pattern 1: Simple Approval - Pause and Wait

Let’s start with a common and straightforward langgraph interrupt resume pattern: getting a simple “yes” or “no” approval from you. The AI does some work, needs your sign-off, pauses, and then waits for your decision. This is perfect for review steps.

The idea is that the graph stops at a specific node, and you, the human, get to look at what happened. You then tell the graph to either continue or perhaps go down a different path. This provides a crucial human touch point.

How It Works: Defining an “Interrupt” Node

To make your graph pause, you essentially create a node where this pause can happen. When the graph reaches this node, it automatically stops running and waits for an external command to proceed. This is part of the interrupt node configuration.

Crucially, when it pauses, LangGraph keeps a snapshot of everything your assistant was thinking or working on. This state preservation during pause means no information is lost. You can imagine the assistant literally freezing in time, holding all its thoughts.

When you’re ready, you’ll send a signal to the graph, telling it to resume from interrupt. You can also provide new information if needed, like your approval. This new information can then trigger state updates after approval to guide the next steps.

Code Example: Simple Approval Graph

Let’s build a small example using LangGraph. We’ll use a very basic setup. First, you need to install LangGraph:

pip install -U langgraph langchain_core

Now, let’s set up a simple graph that asks for approval.

from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
import operator
import json

# 1. Define the graph state
class AgentState(TypedDict):
    current_plan: str
    needs_approval: bool
    approved: bool
    history: Annotated[List[BaseMessage], operator.add]

# 2. Define the nodes (functions)
def generate_plan_node(state: AgentState):
    print("Agent: Generating a plan...")
    plan = "Research market trends for Q3 and draft a summary report."
    print(f"Agent: Proposed plan: '{plan}'")
    return {"current_plan": plan, "needs_approval": True, "approved": False}

def human_approval_node(state: AgentState):
    # This node is where the human would normally interact.
    # For a simple interrupt, we just set needs_approval to True to trigger the pause.
    print(f"Agent: Reached human approval node. Waiting for approval for plan: '{state['current_plan']}'")
    # In a real system, the graph would pause here.
    # The 'needs_approval' flag helps us identify where to pause from outside.
    return state # No direct state change here, just a "stop" point.

def execute_plan_node(state: AgentState):
    if state["approved"]:
        print(f"Agent: Plan '{state['current_plan']}' approved. Executing now!")
        # Simulate execution
        executed_result = f"Summary report drafted for: '{state['current_plan']}'"
        print(f"Agent: Execution complete: {executed_result}")
        return {"current_plan": executed_result, "needs_approval": False}
    else:
        print(f"Agent: Plan '{state['current_plan']}' was not approved. Halting.")
        return {"current_plan": "Plan rejected, process halted.", "needs_approval": False}

# 3. Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("generate_plan", generate_plan_node)
workflow.add_node("human_approval", human_approval_node)
workflow.add_node("execute_plan", execute_plan_node)

workflow.set_entry_point("generate_plan")

# Conditional edge from generate_plan: always goes to human_approval
workflow.add_edge("generate_plan", "human_approval")

# Conditional edge from human_approval:
# This is where we implicitly handle the 'interrupt'.
# We want to pause *before* moving to execute_plan.
# The graph execution itself will be managed externally.
# For demo purposes, we'll connect it, but in a real scenario,
# an external loop would capture the interrupt and then invoke with new state.
workflow.add_conditional_edges(
    "human_approval",
    # This function determines the next node based on state
    lambda state: "execute_plan" if state["approved"] else END,
    {
        "execute_plan": "execute_plan",
        "end_rejected": END # Define a path if rejected right away (optional)
    }
)

workflow.add_edge("execute_plan", END)

app = workflow.compile()

Step-by-Step: Building, Pausing, and Resuming

Now, let’s simulate the run, the interrupt, and then the resume from interrupt. This usually involves running the graph in a loop and checking its state.

1. Initial Run and Interruption

When you run the graph, it will generate a plan and then hit the human_approval node. At this point, we want to pause. LangGraph offers stream() for this.

# Function to simulate external human input
def get_human_decision():
    decision = input("Do you approve the plan? (yes/no): ").lower()
    return True if decision == "yes" else False

# --- Simulate the first run, leading to interrupt ---
print("\n--- Starting first run (AI generates plan) ---")
# We'll use graph.stream to observe the state changes and capture the interrupt
# LangGraph's "interrupt" is often managed by observing the state or a specific node.
# When a node completes, LangGraph yields its output.
# If we want explicit interruption, we typically use "checkpointing" and re-invocation.

# For this simple example, we'll manually check the state from each step.
# A more robust system would use checkpointer with stream_json() or a custom runner.

current_state = {}
thread_id = "user_123" # A unique identifier for this conversation thread

# Running the graph up to the point of needing approval
# The actual LangGraph interrupt mechanism uses stream() and checks for the 'next_steps'
# or specifically, if the graph is waiting for `tools` from the Human or is in an `interrupt` state.

# Let's re-conceptualize slightly for a more direct LangGraph interrupt experience using checkpoints.
from langgraph.checkpoint.sqlite import SQLiteSaver
memory = SQLiteSaver.from_conn_string(":memory:")

# Recompile the app with a checkpointer
app = workflow.compile(checkpointer=memory)

print("\n--- Starting initial process ---")
config = {"configurable": {"thread_id": thread_id}}
# First invocation to generate the plan
# We don't stream here because we want the graph to run till it needs human input.
# The `human_approval_node` doesn't inherently pause, it's a conceptual pause.
# Real interruption happens when a node is set up to `yield ToolCalls`
# or when you explicitly configure it to `interrupt` at a certain point.

# Let's adjust our `human_approval_node` to actually trigger an implicit interrupt
# by returning a `ToolCall` that requires human action.
# This is a common pattern in LangGraph for HITL.

from langchain_core.messages import ToolMessage, AIMessage, HumanMessage

class AgentState(TypedDict):
    current_plan: str
    approved: bool
    history: Annotated[List[BaseMessage], operator.add]
    # We add `next_action` to store tool calls if needed
    next_action: dict

def generate_plan_node(state: AgentState):
    plan = "Research market trends for Q3 and draft a summary report."
    print(f"Agent: Proposed plan: '{plan}'")
    return {"current_plan": plan, "history": [AIMessage(content=f"Proposed plan: {plan}")]}

def human_approval_tool(plan: str):
    """Call this tool to get human approval for a given plan."""
    print(f"\n--- HUMAN INTERVENTION REQUIRED ---")
    print(f"Agent needs your approval for plan: '{plan}'")
    decision = input("Do you approve the plan? (yes/no): ").lower()
    print(f"--- HUMAN RESPONSE RECEIVED ---")
    return {"approved": True if decision == "yes" else False}

# We need to register this human approval "tool" for the agent to call it
from langchain_core.tools import tool

@tool
def human_review_tool(plan_to_review: str) -> dict:
    """Provides a plan to a human for review and returns their decision."""
    print(f"\n--- HUMAN INTERVENTION REQUIRED ---")
    print(f"Agent needs your approval for plan: '{plan_to_review}'")
    decision = input("Do you approve the plan? (yes/no): ").lower()
    return {"approved": True if decision == "yes" else False, "review_plan": plan_to_review}


def call_human_review_node(state: AgentState):
    print("Agent: Requesting human approval via tool...")
    # Simulate the agent calling a tool that needs human input
    tool_call_message = AIMessage(
        content="",
        tool_calls=[
            {
                "id": "human_review_123",
                "name": human_review_tool.name,
                "args": {"plan_to_review": state["current_plan"]}
            }
        ]
    )
    return {"history": [tool_call_message]}

def decide_to_execute(state: AgentState):
    # This function will determine the next step after tool execution.
    # We look into the history for the ToolMessage's output.
    for msg in reversed(state["history"]):
        if isinstance(msg, ToolMessage) and msg.name == human_review_tool.name:
            tool_output = msg.content
            try:
                output_data = json.loads(tool_output)
                if output_data.get("approved"):
                    print("Decision: Plan approved.")
                    return "execute_plan"
                else:
                    print("Decision: Plan rejected.")
                    return "end_rejected"
            except json.JSONDecodeError:
                print("Error parsing tool output, defaulting to rejection.")
                return "end_rejected"
    print("Decision: No approval found, defaulting to rejection.")
    return "end_rejected"

# 3. Build the graph (Revised)
workflow = StateGraph(AgentState)

workflow.add_node("generate_plan", generate_plan_node)
workflow.add_node("call_human_review", call_human_review_node)
workflow.add_node("execute_plan", execute_plan_node) # execute_plan_node modified to use approved state
# The `execute_plan_node` from before needs to be updated slightly
def execute_plan_node(state: AgentState):
    # The decision is now in the state via `decide_to_execute`
    print(f"Agent: Plan '{state['current_plan']}' approved. Executing now!")
    executed_result = f"Summary report drafted for: '{state['current_plan']}'"
    print(f"Agent: Execution complete: {executed_result}")
    return {"current_plan": executed_result, "history": [AIMessage(content=f"Execution complete: {executed_result}")]}

workflow.set_entry_point("generate_plan")

workflow.add_edge("generate_plan", "call_human_review")

# The graph will implicitly "interrupt" when `call_human_review` yields a ToolCall.
# The external runner then needs to call `human_review_tool` and send back the result.

# We need a decision node after the tool call.
workflow.add_conditional_edges(
    "call_human_review",
    decide_to_execute, # This function will read the tool's output from history
    {
        "execute_plan": "execute_plan",
        "end_rejected": END
    }
)

workflow.add_edge("execute_plan", END)

app_with_checkpointer = workflow.compile(checkpointer=memory, tools=[human_review_tool]) # Pass tools to graph

# --- Running the graph with explicit interruption and resume ---
print("\n--- Starting process with explicit human tool call and resume ---")
config = {"configurable": {"thread_id": "simple_approval_thread"}}

# Initial invocation. It will run until the 'call_human_review' node.
# The stream will yield `ToolCalls` which signifies an interrupt.
first_run_output = None
for s in app_with_checkpointer.stream({"history": [HumanMessage(content="Generate a plan.")]}, config):
    first_run_output = s
    print(s)
    if "__end__" not in s: # Check if the graph has not finished
        # If the last message is an AIMessage with tool_calls, it's waiting for human input
        if "history" in s and s["history"] and isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
            print("--- Graph interrupted, waiting for human approval! ---")
            break # Break the stream to simulate the pause

# Check if we actually interrupted and got tool calls
if first_run_output and "history" in first_run_output and first_run_output["history"] and \
   isinstance(first_run_output["history"][-1], AIMessage) and first_run_output["history"][-1].tool_calls:
    ai_message_with_tool_calls = first_run_output["history"][-1]
    tool_call = ai_message_with_tool_calls.tool_calls[0]

    # --- Human provides input ---
    human_decision_result = human_approval_tool.invoke(tool_call["args"])
    print(f"Human decision: {human_decision_result}")

    # --- Resume the graph by sending the tool output back ---
    print("\n--- Resuming graph with human decision ---")
    # We send back a ToolMessage with the result of the human's interaction
    tool_message = ToolMessage(
        content=json.dumps(human_decision_result),
        tool_call_id=tool_call["id"],
        name=tool_call["name"]
    )
    # The graph resumes from the last state, processing the ToolMessage
    for s in app_with_checkpointer.stream({"history": [tool_message]}, config):
        print(s)
else:
    print("Graph completed without interruption or in an unexpected way.")

In this example:

1. generate_plan_node creates a plan.
2. call_human_review_node then “calls” a special human_review_tool. When a LangGraph agent calls a tool, and that tool isn’t immediately executed, the graph pauses.
3. Your external code (the loop) detects this pause by looking for tool_calls in the output stream.
4. You, the human, then interact with the human_review_tool outside the graph, providing your approval.
5. Finally, your code sends the result of your approval back into the graph as a ToolMessage. This resume from interrupt action makes the graph continue from where it left off, now knowing your decision.

You can observe how the state changes, allowing state updates after approval to happen. This ensures the thread continuation is smooth and accurate based on your input.

Pattern 2: Data Correction / Input Injection

Sometimes, you don’t just approve; you provide new information or correct existing data. This langgraph interrupt resume pattern is about injecting fresh data into the graph’s state. It’s like your assistant asks for a missing puzzle piece, and you give it to them.

This pattern is useful when the AI generates something that’s close but needs a tweak. Or when it realizes it needs specific details only you can provide to move forward. The graph pauses, you update the state, and then it continues.

How It Works: Updating State During Resume

The core idea here is that when you resume from interrupt, you don’t just signal to continue. You also send new data that modifies the current state of the graph. This is different from a simple “yes/no” approval.

You might provide a corrected sentence, a missing date, or even a completely new instruction. LangGraph allows you to pass an input dictionary during the resume operation. This input gets merged into the graph’s state or processed by the next node. This is a powerful form of handling user input.

Code Example: Adding Missing Details

Let’s expand our example. Imagine the AI needs to send a personalized message but realizes it doesn’t have the user’s name. It will pause, ask for the name, and then use it.

from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END
import operator
import json
from langgraph.checkpoint.sqlite import SQLiteSaver
from langchain_core.tools import tool

# 1. Define the graph state (updated)
class PersonalizationState(TypedDict):
    task: str
    message_draft: str
    user_name: str
    needs_user_name: bool # New flag
    history: Annotated[List[BaseMessage], operator.add]

# 2. Define the nodes
def initial_task_node(state: PersonalizationState):
    print("Agent: Starting task to draft a personalized message.")
    return {"task": "Draft a personalized welcome message.",
            "history": [AIMessage(content="Starting task to draft a personalized message.")]}

def draft_message_node(state: PersonalizationState):
    print("Agent: Drafting message...")
    if not state.get("user_name"):
        print("Agent: Realized user name is missing. Requesting it.")
        return {"message_draft": "", "needs_user_name": True}
    else:
        draft = f"Hello {state['user_name']},\n\nWelcome to our service! We're excited to have you."
        print(f"Agent: Drafted message: '{draft}'")
        return {"message_draft": draft, "needs_user_name": False,
                "history": [AIMessage(content=f"Drafted message: {draft}")]}

@tool
def get_user_name_tool() -> str:
    """Prompts the human to provide their name."""
    print(f"\n--- HUMAN INTERVENTION REQUIRED ---")
    user_name = input("Agent needs your name. Please enter it: ")
    print(f"--- HUMAN RESPONSE RECEIVED ---")
    return user_name

def call_get_user_name_node(state: PersonalizationState):
    print("Agent: Calling tool to get user name...")
    tool_call_message = AIMessage(
        content="",
        tool_calls=[
            {
                "id": "get_name_123",
                "name": get_user_name_tool.name,
                "args": {}
            }
        ]
    )
    return {"history": [tool_call_message]}

def finalize_message_node(state: PersonalizationState):
    print(f"Agent: Finalizing message: {state['message_draft']}")
    final_message = state['message_draft'] + "\n\nBest regards,\nYour AI Assistant"
    print(f"Agent: Final message: {final_message}")
    return {"message_draft": final_message,
            "history": [AIMessage(content=f"Finalized message: {final_message}")]}

# 3. Build the graph
workflow = StateGraph(PersonalizationState)

workflow.add_node("initial_task", initial_task_node)
workflow.add_node("draft_message", draft_message_node)
workflow.add_node("call_get_user_name", call_get_user_name_node)
workflow.add_node("finalize_message", finalize_message_node)

workflow.set_entry_point("initial_task")

workflow.add_edge("initial_task", "draft_message")

# Conditional routing from draft_message
workflow.add_conditional_edges(
    "draft_message",
    lambda state: "call_get_user_name" if state["needs_user_name"] else "finalize_message",
    {
        "call_get_user_name": "call_get_user_name",
        "finalize_message": "finalize_message"
    }
)

# After calling the tool, we need to process its output.
# The `call_get_user_name` node yields a ToolCall. When the human responds
# with a ToolMessage, we want to update the state with the user_name.
# We'll use a specific node to process the tool output and then loop back to draft_message.
# Let's add a "process_user_name" node.

def process_user_name_node(state: PersonalizationState):
    for msg in reversed(state["history"]):
        if isinstance(msg, ToolMessage) and msg.name == get_user_name_tool.name:
            user_name = msg.content # The content of ToolMessage is the tool's return value
            print(f"Agent: Received user name: {user_name}")
            return {"user_name": user_name, "needs_user_name": False,
                    "history": [AIMessage(content=f"Received user name: {user_name}")]}
    return state # Should not happen if routing is correct

workflow.add_node("process_user_name", process_user_name_node)
workflow.add_edge("call_get_user_name", "process_user_name") # Tool call goes to processing
workflow.add_edge("process_user_name", "draft_message") # After processing, try drafting again

workflow.add_edge("finalize_message", END)

memory = SQLiteSaver.from_conn_string(":memory:")
app_with_data_injection = workflow.compile(checkpointer=memory, tools=[get_user_name_tool])

# --- Running the graph with data injection ---
print("\n--- Starting process with data injection ---")
config = {"configurable": {"thread_id": "data_injection_thread"}}

# First invocation to start the process
first_run_output = None
for s in app_with_data_injection.stream({"history": [HumanMessage(content="Draft a welcome message.")]}, config):
    first_run_output = s
    print(s)
    if "__end__" not in s:
        # Check if the graph is waiting for tool calls (i.e., human input)
        if "history" in s and s["history"] and \
           isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
            print("--- Graph interrupted, waiting for user name! ---")
            break

if first_run_output and "history" in first_run_output and first_run_output["history"] and \
   isinstance(first_run_output["history"][-1], AIMessage) and first_run_output["history"][-1].tool_calls:
    ai_message_with_tool_calls = first_run_output["history"][-1]
    tool_call = ai_message_with_tool_calls.tool_calls[0]

    # --- Human provides input ---
    user_name_input = get_user_name_tool.invoke({}) # Invoke the tool to get human input
    print(f"Human provided name: {user_name_input}")

    # --- Resume the graph by sending the tool output back ---
    print("\n--- Resuming graph with human provided name ---")
    tool_message = ToolMessage(
        content=user_name_input, # The actual name is the content here
        tool_call_id=tool_call["id"],
        name=tool_call["name"]
    )
    # The graph resumes, and `process_user_name_node` will update the state
    for s in app_with_data_injection.stream({"history": [tool_message]}, config):
        print(s)
else:
    print("Graph completed without needing user name or in an unexpected way.")

In this scenario:

1. initial_task_node sets up the goal.
2. draft_message_node tries to draft but finds user_name is missing, so it sets needs_user_name to True.
3. The conditional edge routes to call_get_user_name. This node then calls the get_user_name_tool.
4. The graph pauses (implicit interrupt) because it’s waiting for the tool’s result.
5. You, the human, provide your name to the get_user_name_tool.invoke({}) call.
6. Your code sends this name back as a ToolMessage.
7. The process_user_name_node processes this ToolMessage, extracting your name and adding it to the user_name in the graph’s state.
8. The graph then loops back to draft_message_node, which now finds user_name present and successfully drafts the message. This shows effective state updates after approval and seamless thread continuation.

This demonstrates how handling user input can involve more than simple approvals. You can inject specific pieces of data to steer the AI’s process.

Implementing Interruption in LangGraph

Understanding the mechanisms behind interruption is key to using langgraph interrupt resume patterns effectively. It’s not just about a “pause” button; it’s a well-defined process. You need to know how LangGraph manages runs, saves progress, and lets you re-engage.

Using graph.invoke and graph.stream

LangGraph provides two main ways to run your graph: invoke() and stream().

• invoke() runs the graph completely from start to finish (or until an error) and returns the final state. It doesn’t inherently support interrupt and resume directly because it aims for a single, complete run.
• stream() is much more powerful for langgraph interrupt resume patterns. It yields the state of the graph after each node (or even after each step within a node). This allows you to observe the graph’s progress in real-time. When a graph needs human input (like calling a tool that isn’t immediately resolved), stream() will yield a state that indicates this, allowing you to intercept.

Checkpoints: Your Graph’s Memory

Checkpoints are fundamental to state preservation during pause. Think of a checkpoint as a save file in a video game. When your game crashes, you can load from your last save point. In LangGraph, when you interrupt, the entire state of your graph is saved in a checkpoint.

When you resume from interrupt, LangGraph loads this checkpoint. It then continues processing from exactly where it left off, ensuring no context is lost. This is crucial for maintaining the flow of complex interactions over time.

How to Configure a Checkpoint Management System

LangGraph offers different checkpoint management systems. The simplest is MemorySaver, which keeps checkpoints in your computer’s memory. For real applications, you’ll want something more persistent, like SQLiteSaver or a custom database.

Here’s how you might configure a SQLiteSaver:

from langgraph.checkpoint.sqlite import SQLiteSaver

# This creates an SQLite database file named 'langgraph_checkpoints.sqlite'
# You can also use ':memory:' for an in-memory database (data lost on program exit)
memory = SQLiteSaver.from_conn_string("langgraph_checkpoints.sqlite")

# When compiling your graph, pass the checkpointer:
# app = workflow.compile(checkpointer=memory, tools=[...])

Every time the graph runs, LangGraph automatically saves its state to this checkpointer. When you call stream() or invoke() with a thread_id, it first tries to load the last saved state for that thread_id. If it finds one, it continues from there.

Handling User Input for Resume

When your graph is interrupted and waiting for you, you’ll typically interact with it by invoking the graph again. You provide new input and a config dictionary.

The config dictionary is super important. It tells LangGraph which conversation thread to resume and what the last run was.

config = {"configurable": {"thread_id": "my_unique_thread_id"}}
# If you want to resume a specific run (less common for human-in-the-loop,
# more for debugging or specific advanced flows), you can add:
# config["configurable"]["thread_ts"] = "timestamp_of_last_run"

The thread_id is a unique name you give to each conversation. It allows LangGraph to fetch the correct saved state from your checkpoint management system. When resume from interrupt, your new input (e.g., your name, your approval) is then processed by the graph. The LangGraph framework ensures the thread continuation is smooth.

Advanced Interrupt and Resume Techniques

Beyond simple approvals and data injection, langgraph interrupt resume patterns can become quite sophisticated. You can design your graph to have multiple points where you might need to step in, or even add time limits to human responses.

Multiple Interrupt Points

Your graph can be designed with several different interrupt node configurations. Imagine a multi-stage approval process: first a junior manager approves, then a senior manager. Or a complex data entry task where different pieces of information might be requested at different stages.

Designing for this means making sure each interruption point clearly identifies what it’s waiting for. You might need to examine the tool_calls or the specific node where the graph paused to understand the context. This helps you present the right question to the human user. You can add specific flags in your state or use different tool names to distinguish between them.

Interrupt Timeout Handling

What if a human doesn’t respond? Sometimes, waiting indefinitely isn’t an option. For example, if a customer support agent is busy and doesn’t approve a response within a set time, the system might need to escalate or use a default action. This is interrupt timeout handling.

LangGraph itself doesn’t have a built-in timeout for human input. You’ll need to implement this externally. Here’s a conceptual approach:

1. When the graph yields an interrupt, record the timestamp.
2. Have an external process (e.g., a cron job, a background worker) periodically check for pending interruptions.
3. If an interruption for a specific thread_id has exceeded its timeout, you can programmatically resume from interrupt with a special “timeout” input.
4. This “timeout” input would then trigger error recovery patterns within your graph. For instance, it could send a default response, escalate to another agent, or mark the task as failed.

Here’s a conceptual snippet for handling a timeout:

# Assuming you have a way to store pending interruptions and their timestamps
# For example, a dictionary:
# pending_interrupts = {
#     "thread_id_1": {"timestamp": time.time(), "tool_call_id": "..."}
# }

# In your external checker function:
import time

def check_for_timeouts(app_instance, config_manager, timeout_seconds=3600): # 1 hour
    current_time = time.time()
    for thread_id, interrupt_data in list(config_manager.pending_interrupts.items()):
        if current_time - interrupt_data["timestamp"] > timeout_seconds:
            print(f"Timeout detected for thread {thread_id}. Resuming with default.")
            # Construct a ToolMessage that signifies a timeout
            timeout_tool_message = ToolMessage(
                content=json.dumps({"status": "timeout", "reason": "human did not respond"}),
                tool_call_id=interrupt_data["tool_call_id"],
                name=interrupt_data["tool_name"] # Store tool_name when interrupting
            )
            # Resume the graph with this message
            timeout_config = {"configurable": {"thread_id": thread_id}}
            for s in app_instance.stream({"history": [timeout_tool_message]}, timeout_config):
                print(f"Timeout resume output: {s}")
            # Remove from pending list after resume
            del config_manager.pending_interrupts[thread_id]

# Your graph would need a node to handle the "timeout" result from the tool output.
# The `decide_to_execute` in previous examples could be extended to check for "status": "timeout".

Conditional Interrupts

Not every task needs human intervention. Sometimes, you only want to interrupt if certain conditions are met. For example, if the AI’s confidence score for a generated answer is below a certain threshold, then it should ask for human review. If the confidence is high, it proceeds automatically.

You can achieve this using LangGraph’s conditional edges. Instead of directly sending to a human review node, you first send to a “check_condition” node. This node evaluates the AI’s output (e.g., its confidence score). Based on that, it directs the flow: either to the human interrupt node configuration or directly to the next AI step.

# Example state
class ConditionalState(TypedDict):
    ai_output: str
    confidence_score: float
    # ...

# Node to decide whether to interrupt
def decide_review_needed(state: ConditionalState):
    if state["confidence_score"] < 0.7:
        print("Confidence low, human review needed.")
        return "human_review"
    else:
        print("Confidence high, proceeding automatically.")
        return "auto_continue"

# ... (rest of your graph definition)
# Assuming 'generate_response' creates ai_output and confidence_score
workflow.add_conditional_edges(
    "generate_response",
    decide_review_needed,
    {
        "human_review": "call_human_review", # Your interrupt node
        "auto_continue": "final_action"
    }
)

This pattern allows for smarter, more efficient use of human resources, leveraging Graph interruption mechanics only when truly necessary. You can also explore [our detailed post on LangGraph conditional routing](/blog/langgraph-conditional-routing.md) for more insights.

Practical Examples & Code Walkthroughs

Let’s dive into some more elaborate examples to cement your understanding of langgraph interrupt resume patterns. These show how to integrate these concepts into real-world applications.

Example 1: Customer Support Ticket Escalation

Imagine a smart customer support agent. It tries to answer common questions. If it can’t, or if the customer’s sentiment is very negative, it escalates the ticket to a human agent for review. The human can then take over or guide the AI. This showcases robust error recovery patterns.

Graph Design:

• Receive Ticket: Entry point.
• Analyze Sentiment: AI determines if the customer is angry/frustrated.
• Draft Response: AI tries to answer the question based on FAQs.
• Review & Escalate (Conditional):
  • If sentiment is very negative OR AI’s confidence in its answer is low -> go to Human Agent Review.
  • Else -> go to Send Response.
• Human Agent Review: This is our interrupt node configuration. Human decides to approve, edit, or take over.
• Send Response: Final action.

Implementation:

from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END
import operator
import json
from langgraph.checkpoint.sqlite import SQLiteSaver
from langchain_core.tools import tool

# 1. Define the graph state
class SupportTicketState(TypedDict):
    ticket_id: str
    customer_query: str
    ai_draft_response: str
    sentiment_score: float # e.g., -1 (negative), 0 (neutral), 1 (positive)
    confidence_score: float # AI's confidence in its drafted response (0-1)
    human_action: str # "approve", "edit", "escalate"
    human_edited_response: str
    history: Annotated[List[BaseMessage], operator.add]

# 2. Define the nodes
def receive_ticket_node(state: SupportTicketState):
    query = state["customer_query"]
    ticket_id = state.get("ticket_id", f"TICKET_{hash(query) % 10000}")
    print(f"Agent: Received ticket {ticket_id} with query: '{query}'")
    return {"ticket_id": ticket_id, "history": [HumanMessage(content=query)]}

def analyze_sentiment_node(state: SupportTicketState):
    print("Agent: Analyzing sentiment...")
    # Simulate sentiment analysis
    sentiment = -0.8 if "angry" in state["customer_query"].lower() else 0.2
    print(f"Agent: Sentiment score: {sentiment}")
    return {"sentiment_score": sentiment,
            "history": [AIMessage(content=f"Sentiment analysis: {sentiment}")]}

def draft_response_node(state: SupportTicketState):
    print("Agent: Drafting response...")
    query = state["customer_query"]
    draft = f"Thank you for reaching out regarding '{query}'. We are looking into this."
    confidence = 0.6 if state["sentiment_score"] < -0.5 else 0.9 # Simulate lower confidence for negative sentiment
    print(f"Agent: Drafted: '{draft}' (Confidence: {confidence})")
    return {"ai_draft_response": draft, "confidence_score": confidence,
            "history": [AIMessage(content=f"Drafted response: {draft} (Confidence: {confidence})")]}

@tool
def human_agent_review_tool(ticket_id: str, query: str, ai_draft: str, sentiment: float, confidence: float) -> dict:
    """A human agent reviews the AI's draft and decides on the next action."""
    print(f"\n--- HUMAN AGENT INTERVENTION REQUIRED (Ticket: {ticket_id}) ---")
    print(f"Query: {query}")
    print(f"AI Draft: {ai_draft}")
    print(f"Sentiment: {sentiment}, Confidence: {confidence}")

    action = input("Action (approve/edit/escalate): ").lower()
    edited_response = ""
    if action == "edit":
        edited_response = input("Enter edited response: ")
    print(f"--- HUMAN AGENT RESPONSE RECEIVED ---")
    return {"action": action, "edited_response": edited_response}

def call_human_agent_review_node(state: SupportTicketState):
    print("Agent: Escalating to human agent for review...")
    tool_call_message = AIMessage(
        content="",
        tool_calls=[
            {
                "id": "human_review_ticket_123",
                "name": human_agent_review_tool.name,
                "args": {
                    "ticket_id": state["ticket_id"],
                    "query": state["customer_query"],
                    "ai_draft": state["ai_draft_response"],
                    "sentiment": state["sentiment_score"],
                    "confidence": state["confidence_score"]
                }
            }
        ]
    )
    return {"history": [tool_call_message]}

def process_human_agent_action_node(state: SupportTicketState):
    for msg in reversed(state["history"]):
        if isinstance(msg, ToolMessage) and msg.name == human_agent_review_tool.name:
            tool_output = json.loads(msg.content)
            print(f"Agent: Human agent decided: {tool_output['action']}")
            new_state = {
                "human_action": tool_output["action"],
                "human_edited_response": tool_output.get("edited_response", ""),
                "history": [AIMessage(content=f"Human agent action: {tool_output['action']}")]
            }
            return new_state
    return state

def send_response_node(state: SupportTicketState):
    final_response = state["ai_draft_response"]
    if state["human_action"] == "edit" and state["human_edited_response"]:
        final_response = state["human_edited_response"]
        print("Agent: Sending human-edited response.")
    elif state["human_action"] == "approve":
        print("Agent: Sending AI-drafted (approved) response.")
    elif state["human_action"] == "escalate":
        print("Agent: Response not sent, ticket fully escalated for manual handling.")
        return {"history": [AIMessage(content="Ticket fully escalated.")]} # No response sent by AI
    else:
        print("Agent: Sending default AI-drafted response (no human action specified).")

    print(f"Agent: Final response sent for {state['ticket_id']}: '{final_response}'")
    return {"history": [AIMessage(content=f"Final response sent: {final_response}")],
            "ai_draft_response": final_response}

# 3. Build the graph
workflow = StateGraph(SupportTicketState)

workflow.add_node("receive_ticket", receive_ticket_node)
workflow.add_node("analyze_sentiment", analyze_sentiment_node)
workflow.add_node("draft_response", draft_response_node)
workflow.add_node("call_human_agent_review", call_human_agent_review_node)
workflow.add_node("process_human_agent_action", process_human_agent_action_node)
workflow.add_node("send_response", send_response_node)

workflow.set_entry_point("receive_ticket")

workflow.add_edge("receive_ticket", "analyze_sentiment")
workflow.add_edge("analyze_sentiment", "draft_response")

# Conditional routing from draft_response to decide if human review is needed
def decide_review_or_send(state: SupportTicketState):
    if state["sentiment_score"] < -0.5 or state["confidence_score"] < 0.75:
        return "human_review"
    return "send_automatically"

workflow.add_conditional_edges(
    "draft_response",
    decide_review_or_send,
    {
        "human_review": "call_human_agent_review",
        "send_automatically": "send_response"
    }
)

workflow.add_edge("call_human_agent_review", "process_human_agent_action")

# After human action, decide if we send the response or end (if escalated)
workflow.add_conditional_edges(
    "process_human_agent_action",
    lambda state: "send_response" if state["human_action"] in ["approve", "edit"] else END,
    {
        "send_response": "send_response",
        "end_escalated": END # For "escalate" action
    }
)

workflow.add_edge("send_response", END)

memory = SQLiteSaver.from_conn_string("support_tickets.sqlite")
app_support = workflow.compile(checkpointer=memory, tools=[human_agent_review_tool])

# --- Simulate a problematic ticket ---
print("\n--- Simulating a problematic ticket needing human review ---")
problem_query = "I am extremely angry about your service! It's terrible!"
config_problem = {"configurable": {"thread_id": "ticket_angry_customer"}}

for s in app_support.stream({"customer_query": problem_query}, config_problem):
    print(s)
    if "__end__" not in s:
        if "history" in s and s["history"] and \
           isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
            print("--- GRAPH INTERRUPTED: Human agent review required ---")
            break

# Simulate human agent action: Edit the response
if "history" in s and s["history"] and isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
    ai_msg = s["history"][-1]
    tool_call = ai_msg.tool_calls[0]

    # Human decides to edit
    human_response_data = human_agent_review_tool.invoke(tool_call["args"])
    # Example: human_response_data = {"action": "edit", "edited_response": "We sincerely apologize for your experience and are actively looking into your concerns. Please expect a follow-up from a senior agent soon."}

    print("\n--- Resuming with human agent's decision ---")
    tool_message = ToolMessage(
        content=json.dumps(human_response_data),
        tool_call_id=tool_call["id"],
        name=tool_call["name"]
    )
    for s_resume in app_support.stream({"history": [tool_message]}, config_problem):
        print(s_resume)

This example demonstrates a sophisticated use of langgraph interrupt resume patterns. The call_human_agent_review_node serves as the interrupt node configuration. It gathers all relevant information (ticket_id, query, draft, sentiment, confidence) and passes it to the human_agent_review_tool. When the graph yields this tool call, it implicitly pauses.

The human agent provides handling user input by deciding the action (approve, edit, escalate) and potentially providing state updates after approval (an edited_response). This input is then sent back as a ToolMessage, allowing thread continuation and ensuring the ticket is handled appropriately. This process also shows robust error recovery patterns as it addresses negative sentiment or low confidence.

Example 2: Content Generation with Human Editor

Imagine an AI drafting blog posts or marketing copy. Before publishing, a human editor needs to review, make changes, and give final approval. The AI can then finalize the content.

Graph Design:

• Generate Draft: AI creates initial content.
• Human Editor Review: This is the interrupt node configuration. Human reviews, edits, and approves.
• Refine & Finalize: AI makes final formatting adjustments or checks based on human input.
• Publish: End point.

Implementation:

# Reusing most of the setup from previous examples for brevity
# Only showing relevant nodes and graph structure for this specific pattern

class ContentState(TypedDict):
    topic: str
    ai_draft_content: str
    human_edited_content: str
    approved_for_publish: bool
    history: Annotated[List[BaseMessage], operator.add]

def generate_draft_content_node(state: ContentState):
    print(f"Agent: Generating draft content for topic: '{state['topic']}'")
    draft = f"Initial draft about {state['topic']}:\n\n" \
            f"This is a placeholder content. The AI would write an extensive article here.\n" \
            f"It covers key points and structure for the topic. " \
            f"Please review and provide your edits."
    return {"ai_draft_content": draft,
            "history": [AIMessage(content=f"Generated initial draft for: {state['topic']}")]}

@tool
def human_editor_tool(topic: str, draft_content: str) -> dict:
    """Human editor reviews and edits the AI-generated content."""
    print(f"\n--- HUMAN EDITOR INTERVENTION REQUIRED ---")
    print(f"Topic: {topic}")
    print(f"AI Draft:\n{draft_content}")
    edited = input("Enter your edited content (or type 'approve' to use AI draft as is): ")
    action = "approve" if edited.lower() == "approve" else "edited"
    final_content = edited if action == "edited" else draft_content
    return {"action": action, "final_content": final_content}

def call_human_editor_node(state: ContentState):
    print("Agent: Sending draft to human editor for review...")
    tool_call_message = AIMessage(
        content="",
        tool_calls=[
            {
                "id": "human_editor_review_123",
                "name": human_editor_tool.name,
                "args": {
                    "topic": state["topic"],
                    "draft_content": state["ai_draft_content"]
                }
            }
        ]
    )
    return {"history": [tool_call_message]}

def process_editor_action_node(state: ContentState):
    for msg in reversed(state["history"]):
        if isinstance(msg, ToolMessage) and msg.name == human_editor_tool.name:
            tool_output = json.loads(msg.content)
            print(f"Agent: Editor action: {tool_output['action']}")
            approved_status = True if tool_output['action'] in ["approve", "edited"] else False
            new_state = {
                "human_edited_content": tool_output["final_content"],
                "approved_for_publish": approved_status,
                "history": [AIMessage(content=f"Editor response: {tool_output['action']}")]
            }
            return new_state
    return state

def finalize_content_node(state: ContentState):
    if state["approved_for_publish"]:
        print("Agent: Finalizing content for publishing...")
        final_version = state["human_edited_content"] + "\n\n-- Published --"
        print(f"Agent: Final Content:\n{final_version}")
        return {"human_edited_content": final_version,
                "history": [AIMessage(content=f"Content finalized and published.")]}
    else:
        print("Agent: Content not approved. Halting publishing.")
        return {"history": [AIMessage(content="Publishing halted due to no approval.")]}

# Build the graph
workflow_content = StateGraph(ContentState)

workflow_content.add_node("generate_draft", generate_draft_content_node)
workflow_content.add_node("call_human_editor", call_human_editor_node)
workflow_content.add_node("process_editor_action", process_editor_action_node)
workflow_content.add_node("finalize_content", finalize_content_node)

workflow_content.set_entry_point("generate_draft")
workflow_content.add_edge("generate_draft", "call_human_editor")
workflow_content.add_edge("call_human_editor", "process_editor_action")

workflow_content.add_conditional_edges(
    "process_editor_action",
    lambda state: "finalize_content" if state["approved_for_publish"] else END,
    {
        "finalize_content": "finalize_content",
        "end_rejected": END
    }
)
workflow_content.add_edge("finalize_content", END)

memory_content = SQLiteSaver.from_conn_string("content_editor.sqlite")
app_content = workflow_content.compile(checkpointer=memory_content, tools=[human_editor_tool])

# --- Simulate content generation and editing ---
print("\n--- Simulating content generation with human editor ---")
topic_to_write = "Benefits of AI in daily life"
config_content = {"configurable": {"thread_id": "content_thread_ai_benefits"}}

for s in app_content.stream({"topic": topic_to_write}, config_content):
    print(s)
    if "__end__" not in s:
        if "history" in s and s["history"] and \
           isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
            print("--- GRAPH INTERRUPTED: Human editor review required ---")
            break

if "history" in s and s["history"] and isinstance(s["history"][-1], AIMessage) and s["history"][-1].tool_calls:
    ai_msg = s["history"][-1]
    tool_call = ai_msg.tool_calls[0]

    # Human editor provides input (edited content or 'approve')
    human_editor_output = human_editor_tool.invoke(tool_call["args"])
    # Example: human_editor_output = {"action": "edited", "final_content": "Human edited version: AI makes life easier by automating tasks, providing smart insights, and enhancing creativity. This is the revised text."}

    print("\n--- Resuming with human editor's input ---")
    tool_message = ToolMessage(
        content=json.dumps(human_editor_output),
        tool_call_id=tool_call["id"],
        name=tool_call["name"]
    )
    for s_resume in app_content.stream({"history": [tool_message]}, config_content):
        print(s_resume)

In this example, the call_human_editor_node triggers an interrupt. The human_editor_tool then gets invoked by your external script, allowing you (the human editor) to review and handling user input. You can either approve the AI’s draft or provide completely new state updates after approval (your edited content).

The graph resumes, processes your input in process_editor_action_node, and then the finalize_content_node takes your edited content (or the approved AI draft) to publish the final version. This shows a clear thread continuation workflow for content creation.

Building a Human Interface

While we’ve used console inputs in our examples, in a real application, you’d connect your LangGraph system to a web interface. Think of tools like Flask, FastAPI, or Streamlit.

Your web UI would:

1. Initiate a LangGraph run by sending an initial input and thread_id to an API endpoint.
2. Listen for the stream() output from LangGraph.
3. When an interrupt (a ToolCall) is detected, it would render a form or a dialog box to the human user, displaying the relevant information and asking for input.
4. Once the human provides input, the UI sends this input (along with the thread_id and tool_call_id) back to a different API endpoint on your server.
5. Your server-side code then uses this information to resume from interrupt the LangGraph thread. This makes the handling user input seamless for the user. You can find more information on [integrating LangGraph with web frameworks](/blog/langgraph-web-integration.md).

Best Practices for LangGraph Interrupts

To make your langgraph interrupt resume patterns reliable and user-friendly, follow these best practices:

• Design Clear Interrupt Nodes: Make sure each interrupt node configuration clearly states why it’s pausing and what kind of human input it needs. This helps you build clear prompts for your users.
• Ensure Robust State Preservation: Always use a persistent checkpoint management system (like SQLiteSaver or a database) for production systems. This ensures state preservation during pause even if your application restarts.
• Implement Effective Checkpoint Management: Regularly review your checkpointing strategy. Understand how thread_ids work and how to retrieve specific states if needed.
• Think About Error Recovery: What happens if the human provides invalid input? What if there’s a network error during resume from interrupt? Design error recovery patterns into your graph to gracefully handle these situations.
• Provide Clear Instructions: When the graph pauses, the human user needs to know exactly what’s expected of them. Provide clear, concise prompts and options.
• Test Thoroughly: Test all possible langgraph interrupt resume patterns: approvals, rejections, data inputs, edge cases, and even timeouts. This ensures your system is robust.

Common Challenges and Solutions

Even with the best planning, you might encounter some challenges when implementing langgraph interrupt resume patterns.

• Challenge: State getting lost or corrupted.
  • Solution: This almost always points to an issue with checkpoint management. Ensure you’re using a persistent checkpointer (not just MemorySaver for long-running processes) and that your thread_ids are consistent. Double-check that your state updates are immutable or correctly merged.
• Challenge: Complex resume logic.
  • Solution: If your graph needs to do too much logic just to figure out what to do after an interrupt, simplify your graph. Use dedicated nodes (like process_human_agent_action_node in our examples) to handle parsing and integrating human input. This keeps the thread continuation clean.
• Challenge: Multiple active threads (conversations) interfering.
  • Solution: Each conversation must have a unique thread_id. This is how LangGraph isolates states. When resume from interrupt, always pass the correct thread_id in the config dictionary.
• Challenge: UI integration feels clunky.
  • Solution: Design clear API endpoints for handling user input. One endpoint to start a thread, another to send input to an interrupted thread. The UI should map human actions to these API calls, translating human input into ToolMessage or specific state updates for resume from interrupt. Consider using a framework like FastAPI for simple API development.

Conclusion

Implementing LangGraph Human in the Loop: Interrupt and Resume Patterns is a powerful way to build more intelligent, reliable, and user-centric AI applications. You’ve learned how to design your LangGraph assistant to pause, ask for your input, and then seamlessly continue its work.

By mastering Graph interruption mechanics, configuring interrupt node configuration, ensuring state preservation during pause with checkpoint management, and effectively handling user input to resume from interrupt, you empower your AI systems. These langgraph interrupt resume patterns allow for crucial state updates after approval and smooth thread continuation, making your AI not just smart, but truly collaborative.

Explore these patterns further, experiment with different scenarios, and see how you can create AI assistants that work hand-in-hand with human intelligence. The ability to pause and get your input makes your AI solutions far more robust, flexible, and ultimately, more useful.