Is it possible to support human in the middle from multiple concurrent subgraphs?

[mattcoulter7]

Hey guys,

I am trying to combine the Human in the Loop with Parellel Execution of a subgraph. My use case is that within each subgraph, there is a a human review stage.

I originally expected that I would be able to call interrupt from within a subgraph. However, it seems that interrupt call from one worker will interrupt all of the other workers. Then, upon resume, all of the workers pickup the same user input.

See the minimal example below (sorry I could not get the entire graph to show as child graph as a single image... just appears as a node...)

Parent Graph:

Child Graph:

Minimal Implementation

import operator
import uuid

from typing import List, Annotated, Optional
from pydantic import BaseModel, Field
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Send, Interrupt, Command


# --- CHILD GRAPH ---

class ChildState(BaseModel):
    prompt: str = Field(
        ...,
        description="What is going to be asked to the user?"
    )
    human_input: Optional[str] = Field(
        None,
        description="What the human said"
    )
    human_inputs: Annotated[List[str], operator.add] = Field(
        default_factory=list,
        description="All of my messages"
    )


def get_human_input(state: ChildState):
    human_input = interrupt(state.prompt)

    return dict(
        human_input=human_input,  # update child state
        human_inputs=[human_input],  # update parent state
    )

child_graph_builder = StateGraph(ChildState)
child_graph_builder.add_node("get_human_input", get_human_input)
child_graph_builder.add_edge(START, "get_human_input")
child_graph_builder.add_edge("get_human_input", END)
child_graph = child_graph_builder.compile(checkpointer=True)

# --- PARENT GRAPH ---

class ParentState(BaseModel):
    prompts: List[str] = Field(
        ...,
        description="What is going to be asked to the user?"
    )
    human_inputs: Annotated[List[str], operator.add] = Field(
        default_factory=list,
        description="All of my messages"
    )

def assign_workers(state: ParentState):
    return [
        Send(
            "child_graph", 
            dict(
                prompt=prompt,
            )
        )
        for prompt in state.prompts
    ]

def cleanup(state: ParentState):
    assert len(state.human_inputs) == len(state.prompts)

parent_graph_builder = StateGraph(ParentState)
parent_graph_builder.add_node("child_graph", child_graph)
parent_graph_builder.add_node("cleanup", cleanup)

parent_graph_builder.add_conditional_edges(START, assign_workers, ["child_graph"])
parent_graph_builder.add_edge("child_graph", "cleanup")
parent_graph_builder.add_edge("cleanup", END)

parent_graph = parent_graph_builder.compile(checkpointer=MemorySaver())


# --- CLIENT INVOCATION ---

if __name__ == "__main__":
    thread_config = dict(
        configurable=dict(
            thread_id=uuid.uuid4(),
        )
    )
    current_input = dict(
        prompts=['a', 'b'],
    )

    done = False
    while not done:
        # reset interrupt
        current_interrupt: Optional[Interrupt] = None

        # start / resume the graph
        for event in parent_graph.stream(
            input=current_input,
            config=thread_config,
            stream_mode="updates",
        ):
            print(event)
            # handle the interrupt
            if "__interrupt__" in event:
                current_interrupt: Interrupt = event["__interrupt__"][0]
                # assume that it breaks here, because it is an interrupt

        # get human input and resume
        if (
            current_interrupt is not None
            and current_interrupt.resumable is True  # make sure it is resumable
        ):
            response = input(f"Input for prompt \"{current_interrupt.value}\": ")
            current_input = Command(resume=response)

        # not more human input required, must be completed
        else:
            done = True

    exit(0)

So I expect to see something like this:

{'__interrupt__': (Interrupt(value='a', resumable=True, ns=['child_graph', 'get_human_input:...'], when='during'),)}
Input for prompt "a": 1  // <-- use input capture #1
{'__interrupt__': (Interrupt(value='b', resumable=True, ns=['child_graph', 'get_human_input:...'], when='during'),)}
Input for prompt "b": 2  // <-- use input capture #2
{'child_graph': {'human_inputs': ['1']}}
{'child_graph': {'human_inputs': ['2']}}
{'cleanup': None}

However, the actual output is this:

{'__interrupt__': (Interrupt(value='b', resumable=True, ns=['child_graph', 'get_human_input:49611ec9-bf1f-bdf7-41c0-1bc2cb7ec5ab'], when='during'),)}
Input for prompt "b": 1    // <-- only single user input capture
{'child_graph': {'human_inputs': ['1']}}  // <-- applied to worker 1
{'child_graph': {'human_inputs': ['1']}}  // <-- and also, applied to worker 1 XD
{'cleanup': None}

Current workaround

In the meantime, I can work around the issue by not using concurrency, and instead just use a synchronous workflow to capture input for each element. But, that obviously comes with its performance limitations.

Furthermore, since my application is just CLI, I can call input(...) or typer.prompt(...) directly in the node, but that is just temporary as I don't want to couple the application logic with how it is interfaced.

[mattcoulter7]

This was my implementation of a synchronous human in the middle flow, combined with executing the sub graph.

Strangely enough, this approach doesn't even execute the child graph on the second time, it just skips over it... Am I missing something?

# --- PARENT GRAPH ---

class ParentState(BaseModel):
    prompts: List[str] = Field(
        ...,
        description="What is going to be asked to the user?"
    )
    human_inputs: Annotated[List[str], operator.add] = Field(
        default_factory=list,
        description="All of my messages"
    )

    prompt: Optional[str] = Field(
        None,
        description="The current prompt that is being worked on"
    )
    prompt_index: int = Field(
        0,
        description="The index of the current prompt we are working on"
    )


def router(state: ParentState):
    return str(bool(state.prompt_index < len(state.prompts)))


def placeholder(state: ParentState):
    return dict()


def set_current_prompt(state: ParentState):
    return dict(
        prompt=state.prompts[state.prompt_index],
    )


def increment_prompt_index(state: ParentState):
    return dict(
        prompt_index=state.prompt_index + 1,
    )


parent_graph_builder = StateGraph(ParentState)
parent_graph_builder.add_node("placeholder", placeholder)
parent_graph_builder.add_node("set_current_prompt", set_current_prompt)
parent_graph_builder.add_node("child_graph", child_graph)
parent_graph_builder.add_node("increment_prompt_index", increment_prompt_index)

parent_graph_builder.add_edge(START, "placeholder")
parent_graph_builder.add_conditional_edges(
    "placeholder",
    router,
    {
        str(True): "set_current_prompt",
        str(False): END,
    }
)
parent_graph_builder.add_edge("set_current_prompt", "child_graph")
parent_graph_builder.add_edge("child_graph", "increment_prompt_index")
parent_graph_builder.add_edge("increment_prompt_index", "placeholder")  # would route back to START, but that is illegal lol

parent_graph = parent_graph_builder.compile(checkpointer=MemorySaver())

Output:

{'placeholder': None}
{'set_current_prompt': {'prompt': 'a'}}
{'__interrupt__': (Interrupt(value='a', resumable=True, ns=['child_graph', 'get_human_input:b2898a55-4798-14b5-3273-ddbae1e75171'], when='during'),)}
Input for prompt "a": 1
{'child_graph': {'prompt': 'a', 'human_inputs': ['1']}}
{'increment_prompt_index': {'prompt_index': 1}}
{'placeholder': None}
{'set_current_prompt': {'prompt': 'b'}}  // <-- updated `prompt` in state to next in sequence
{'child_graph': {'prompt': 'a', 'human_inputs': ['1']}}  // <-- that this still references prompt `a`
{'increment_prompt_index': {'prompt_index': 2}}
{'placeholder': None}