Does LangGraph support using HITL within subgraphs?

OSS Product Help LangGraph
1

I tried to use interrupt in the subgraph. but it doesn’t work

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
from langgraph.types import interrupt


class SubgraphState(TypedDict):
    bar: str
    node_2: str

# Subgraph

def subgraph_node_1(state: SubgraphState):
    print(state["bar"])
    return {"bar": "hi! " + state["bar"]}

def subgraph_node_2(state: SubgraphState):
    user_answer = interrupt({
        "question": "continue"
    })
    if user_answer == "Y":
        print(user_answer["question"])
    else:
        print("this time in loop is deprecated")


subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge(START, "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Parent graph

class State(TypedDict):
    foo: str

def call_subgraph(state: State):
    # Transform the state to the subgraph state
    for i in range(1,10):
        subgraph_output = subgraph.invoke({"bar": f"{state["foo"]}_{i}"})
        if subgraph_output["__interrupt__"]:
            print(subgraph_output["__interrupt__"])
    # Transform response back to the parent state
    return {"foo": subgraph_output["bar"]}

builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
graph.invoke({"foo": "bar"})
1 Like
2

hi @fanlei

LangGraph supports HITL inside subgraphs, but your snippet misses the required interrupt mechanics and is mixing up the interrupt payload vs the resume value.

Key fixes:

  • Add a checkpointer and thread_id. Interrupts only pause when graph state can be persisted and resumed. Without these, interrupt() won’t actually block.
  • Resume with Command(resume=…). The value you pass to Command(resume=…) is what interrupt() returns inside the node.
  • Don’t treat the prompt payload as the resume value. The prompt you passed to interrupt() shows up in the caller under __interrupt__.
  • Subgraphs invoked as functions re-run from the start on resume (both the parent node and the subgraph node), so keep side effects idempotent.
  • Graph topology: you currently connect both subgraph nodes from START. If you intended a sequence, connect node_1 → node_2 and add an END edge.

Also, inside your subgraph_node_2, user_answer will be the resume value (e.g., “Y”), so user_answer[“question”] will error. If you want the question text, read it from result[“__interrupt__”] in the caller.

Docs: Interrupts - Docs by LangChain

Example:

from typing_extensions import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class SubgraphState(TypedDict):
    bar: str
    answer: str


def subgraph_node_1(state: SubgraphState) -> SubgraphState:
    print(state["bar"])
    return {"bar": "hi! " + state["bar"]}


def subgraph_node_2(state: SubgraphState) -> SubgraphState:
    user_answer = interrupt({"question": "continue? (Y/N)"})
    if user_answer == "Y":
        print("user approved")
    else:
        print("user rejected")
    return {**state, **{"answer": user_answer}}


subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_node("subgraph_node_2", subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph_builder.add_edge("subgraph_node_2", END)


class State(TypedDict):
    foo: str
    i: int
    subgraph_answer: str


def call_subgraph(state: State) -> State:
    # Keep progress in state so re-running is safe after resume.
    current = state.get("i", 1)
    subgraph_output = subgraph.invoke({"bar": f"{state['foo']}_{current}"})
    return {"foo": subgraph_output["bar"], "i": current + 1, "subgraph_answer": subgraph_output["answer"]}


builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)


if __name__ == "__main__":
    checkpointer = MemorySaver()
    subgraph = subgraph_builder.compile(checkpointer=checkpointer)
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "demo-thread"}}
    initial_state: State = {"foo": "bar", "i": 1}

    while True:
        result = graph.invoke(initial_state, config=config)
        if "__interrupt__" in result:
            interrupt_payload = result["__interrupt__"][0].value
            print(f"INTERRUPT: {interrupt_payload}")
            user_input = input("Enter Y or N to resume: ").strip().upper()
            initial_state = Command(resume=user_input)
            continue
        print("FINAL STATE:", result)
        break
1 Like