Raising GraphInterrupt Manually — Is This Correct?

Product Help LangGraph
,
1

I came across GraphInterrupt, and read the description:

"Raised when a subgraph is interrupted, suppressed by the root graph. Never raised directly, or surfaced to the user."

Can someone help me understand what this means in practice?

Also, I’m experimenting with LangGraph and tried something like this:

  try:
      result = interrupt("Do you want to proceed?")
      if result == "proceed":
          eval(orm_query)
              return "✅ Operation completed successfully."
 except GraphInterrupt as gi:  
      raise gi

What exactly happens if I raise GraphInterrupt manually like this? Is this the correct way to handle interrupt() inside a tool?

2

GraphInterrupt is an internal exception that LangGraph uses for control flow - you shouldn’t catch or raise it manually. The description means it’s used internally by subgraphs to signal interruption to parent graphs, but it’s handled automatically.

Your code is incorrect. When you call interrupt(), it immediately halts execution and waits for user input - it doesn’t return a value like "proceed". The correct pattern is:

def my_node(state):
    # This will pause execution and wait for user input
    interrupt("Do you want to proceed?")
    
    # Execution resumes here after user provides input
    # User input will be in the next state
    eval(orm_query)
    return "✅ Operation completed successfully."

Don’t catch GraphInterrupt let LangGraph handle it automatically. The interrupt mechanism is designed to pause and resume, not return values synchronously.

3

how to save any state in a node before interrupt?
i am saving state like this in a node but when i fetch using threads id the state is empty
request: HumanInterrupt = {

        "action_request": {

            "action": "collect_text",

            "args": {"prompt": question_text}

        },

        "config": {

            "allow_ignore": False,

            "allow_respond": True,

            "allow_edit": False,

            "allow_accept": False

        },

        "description": question_text

    }

    print(f"🔔 Sending interrupt request")

    

    *# Record interrupt in history BEFORE the interrupt call*

    interrupt_record = {

        "timestamp": datetime.utcnow().isoformat(),

        "action": "collect_text",

        "request": request,

        "thread_id": thread_id,

        "status": "sent"

    }

    

    new_interrupt_records.append(interrupt_record)

    print(f"🔍 DEBUG: Added interrupt record to new records. Count now: {len(new_interrupt_records)}")

    

    *# IMPORTANT: Save interrupt record immediately using graph.update_state()*

    print(f"🔍 DEBUG: Current new_interrupt_records before interrupt: {new_interrupt_records}")

    

    *# Prepare updated interrupt_history for return (let checkpointer handle persistence)*

    current_interrupt_history = existing_interrupt_history + new_interrupt_records

    *state*\["interrupt_history"\] = current_interrupt_history

    

    user_response = interrupt(request)