Interrupt and Gradio Integration (HIL)

[avfranco-br]

Hello,

Has anyone successfully override the langgraph.types.interrupt with a custom_async_interrupt to get a user input using Gradio? The integration with Gradio seems to work fine, user input captured OK, but when trying to resume the graph getting the error Error resuming graph: 'Button' object has no attribute '_id' - During task with name 'ask_human'. ask_human is the node triggering the custom interrupt function.

This is the code generating the error:
`async def resume_graph(self, user_input):
"""
Async method to resume the graph with user input
"""
if not self.graph or not self.thread_config:
print("Graph or thread configuration not set")
return None

    print("Trying to resume the Graph!")
    try:
        # Use Command to resume the graph
        resume_result = await self.graph.ainvoke(
            Command(update={"user_feedback": user_input},resume=user_input),
            config=self.thread_config
        )
        return resume_result
    except Exception as e:
        print(f"Error resuming graph: {e}")
        return None`

It seems to be something related to correctly managing asynchronous but couldn't find the error.

Thanks in advance for any input, working example.

Regards,
Alexandre

[avfranco-br]

Hi,
I've been trying different options, the error has changed but the problem not solved. The graph is being interrupt and the user-input captured but the graph is not being resumed after the Gradio input is captured.

I am stuck on this for days, couldn't find any similar example and any help would be really appreciated. I've updated this comment with a minimal reproducible code.

Many thanks in advance.

app-ea4ll-human-in-the-loop.ipynb

`

import uuid
from app_ea4all_hiit import super_graph
user_input = "Build an architecture bleprint."

thread_config = {"configurable": {"thread_id": uuid.uuid4()}}

async for event in super_graph.astream(
    {"messages": [{"role": "user", "content": user_input}]},
    config=thread_config,
    stream_mode="values",
    debug=True,
    subgraphs=True
): 
    print(f"{event}")`

app_ea4all_hiit.py

`

from langgraph.types import Command
from langchain_core.messages import (
    HumanMessage,
    AIMessage
)
from langgraph.graph import (
    START, 
    END,
    StateGraph,
)
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState

##langsmith (run_id, finish tracing log, traceable)
from langchain.callbacks.tracers import LangChainTracer

import gradio as gr

import uuid
from typing import Any, Optional
from typing_extensions import Literal
import asyncio
import threading

import os

#Set OpenAI Key
openai_api_key = os.environ['OPENAI_API_KEY']  

#Set environment variables for build deployment (local run)
ea4all_stage = os.environ["EA4ALL_ENV"]
if ea4all_stage in ('build', 'experimentation'):
    project_name = "ea4all-agent-gradio"
    runname = "ea4all-HIL-experiments"
    os.environ["LANGCHAIN_PROJECT"] = project_name  # Optional: "default" is used if not set
    os.environ['LANGCHAIN_RUNNAME'] = runname
    os.environ['EA4ALL_ENV'] = 'build'

#Set LangSmith project
tracer = LangChainTracer(project_name=os.environ['LANGCHAIN_PROJECT'])
runname = os.environ['LANGCHAIN_RUNNAME']
thread_id = str(uuid.uuid4())

#Custom Interrupt
class AsyncInterruptHandler:
    def __init__(self, graph=None, thread_config=None):
        self.input_queue = asyncio.Queue()
        self.event = asyncio.Event()
        self.user_feedback = None
        self.graph = graph
        self.thread_config = thread_config
        self.interface = None
        self.interface_thread = None

async def resume_graph(self, user_feedback, interrupt_details: Optional[dict] = None):
    """
    Resume the graph with the user input using Command
    """
    try:
        # Use Command to resume the graph
        result = await self.graph.ainvoke(
            Command(
                resume=user_feedback,
                update={
                    "messages": AIMessage(content=str(interrupt_details), name="blueprint_team"),
                    "next": "call_togaf_agentic"
                }
            ),
            config=self.thread_config
        )
        return result
    
    except Exception as e:
        print(f"Error resuming graph: {e}")
        return None

def launch_gradio_interface(self, interrupt_details: Optional[dict] = None):
    """
    Launch Gradio interface in a separate thread
    """
    def run_interface():
        # Create event loop for this thread
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        async def async_launch():
            return await self._create_and_launch_interface(interrupt_details)

        loop.run_until_complete(async_launch())

    #Create and start the thread
    self.interface_thread = threading.Thread(target=run_interface, daemon=True)
    self.interface_thread.start()

async def _create_and_launch_interface(self, interrupt_details: Optional[dict] = None):
    """
    Async method to create and launch Gradio interface
    """
    #Dynamically create Gradio interface based on interrupt type
    if isinstance(interrupt_details, dict):
        # If interrupt provides specific details
        title = interrupt_details.get('task', 'User Input')
        description = interrupt_details.get('content', 'Please provide input')
    else:
        title = 'User Input'
        description = 'Please provide input'

    async def submit_input(user_feedback):
        await self.input_queue.put(user_feedback)
        self.event.set()

        return {"user_feedback": user_feedback}

    #Dynamically create interface based on interrupt details
    with gr.Blocks() as demo:
        gr.Markdown(f"### {title}")
        gr.Markdown(f"**{description}")

        input_component = gr.Textbox(label="Your input")
        submit_btn = gr.Button("Submit")
        output = gr.Textbox(label="Status")

        submit_btn.click(
            submit_input, 
            inputs=input_component, 
            outputs=output
        )

    #Reset the event
    self.event = asyncio.Event()
    self.event.clear()
    self.user_feedback = None

    #Store the interface
    self.interface = demo

    #Launch the interface
    self.interface.launch(prevent_thread_lock=True, debug=True, inbrowser=True)

async def handle_interrupt(self, interrupt_details: Optional[dict] = None):
    """
    Handle interrupt by launching Gradio interface and waiting for input
    """
    # Launch the interface in a separate thread
    self.launch_gradio_interface(interrupt_details)

    try:
        #Retrieve the input from the queue
        self.user_feedback = await self.input_queue.get()

        #Resume the graph with the user input // NOT WORKING
        return await self.resume_graph(self.user_feedback, interrupt_details)

    except asyncio.TimeoutError:
        print("Gradio interface timed out")
        return None
    except Exception as e:
        print(f"Error in handle_interrupt: {e}")
        return        
    finally:
        #Close the interface
        self.interface.close()

#Global interrupt handler
global_interrupt_handler = AsyncInterruptHandler()

#Custom interrupt function
async def custom_async_interrupt(details=None, graph=None, thread_config=None):

global global_interrupt_handler
global_interrupt_handler.graph = graph
#'global_interrupt_handler.thread_config = thread_config

return await global_interrupt_handler.handle_interrupt(details)

#Override the default interrupt
import langgraph.types
langgraph.types.interrupt = custom_async_interrupt

class State(MessagesState):
    next: Optional[str] = None
    user_feedback: Optional[str] = None

global super_graph

thread_config = {"configurable": {"thread_id": uuid.uuid4()}}

#CAPTURE business requirement asking for USER input & call togad_agentic workflow
async def call_togaf_agentic(state: State) -> Command[Literal["__end__"]]:

    business_query = state["messages"][-1].content

    print("CALLING togaf_graph!!!")

    return Command(
        update={
            "messages": [
                HumanMessage(
                    content=business_query, name="blueprint_team"
                )
            ],
        },
        goto=END,
    )

def blueprint_team(state: State) -> Command:
    print("---Blueprint team got a request---")

async def togaf_ask_human(state: State, graph=None) -> dict:
    global super_graph

    print("---Ask User for Input---")
    user_feedback = await langgraph.types.interrupt(
        {
            "task": state['messages'][-1].content,
            "content": "Please provide your business requirement in the form of document/file name or past the content:",
            "optional": False
        },
        graph=super_graph,
        #'thread_config=thread_config
    )

    return Command(
        update={
            "messages": [
                HumanMessage(
                    content=f"{state['messages'][-1].content} User_Input: {user_feedback}", name="landscape_agentic"
                )
            ]
        },
        goto=call_togaf_agentic,
    )

#Define the graph.
super_builder = StateGraph(State)
super_builder.add_node("blueprint_team", blueprint_team)
super_builder.add_node("generate_togaf", call_togaf_agentic)
super_builder.add_node("ask_user_input", togaf_ask_human)

super_builder.add_edge(START, "blueprint_team")
super_builder.add_edge("blueprint_team", "ask_user_input")
super_builder.add_edge("ask_user_input", "generate_togaf")
super_builder.add_edge("generate_togaf", END)

memory = MemorySaver()
super_graph = super_builder.compile(checkpointer=memory)
super_graph.name = "EA4ALL Agentic Workflow Graph"

`

[avfranco-br]

Hi, I've been trying different options, the error has changed but the problem not solved. The graph is being interrupt and the user-input captured but the graph is not being resumed after the Gradio input is captured.

I am stuck on this for days, couldn't find any similar example and any help would be really appreciated. I've updated this comment with a minimal reproducible code.

Many thanks in advance.

app-ea4ll-human-in-the-loop.ipynb

`

import uuid
from app_ea4all_hiit import super_graph
user_input = "Build an architecture bleprint."

thread_config = {"configurable": {"thread_id": uuid.uuid4()}}

async for event in super_graph.astream(
    {"messages": [{"role": "user", "content": user_input}]},
    config=thread_config,
    stream_mode="values",
    debug=True,
    subgraphs=True
): 
    print(f"{event}")`

app_ea4all_hiit.py

`

from langgraph.types import Command
from langchain_core.messages import (
    HumanMessage,
    AIMessage
)
from langgraph.graph import (
    START, 
    END,
    StateGraph,
)
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState

##langsmith (run_id, finish tracing log, traceable)
from langchain.callbacks.tracers import LangChainTracer

import gradio as gr

import uuid
from typing import Any, Optional
from typing_extensions import Literal
import asyncio
import threading

import os

#Set OpenAI Key
openai_api_key = os.environ['OPENAI_API_KEY']  

#Set environment variables for build deployment (local run)
ea4all_stage = os.environ["EA4ALL_ENV"]
if ea4all_stage in ('build', 'experimentation'):
    project_name = "ea4all-agent-gradio"
    runname = "ea4all-HIL-experiments"
    os.environ["LANGCHAIN_PROJECT"] = project_name  # Optional: "default" is used if not set
    os.environ['LANGCHAIN_RUNNAME'] = runname
    os.environ['EA4ALL_ENV'] = 'build'

#Set LangSmith project
tracer = LangChainTracer(project_name=os.environ['LANGCHAIN_PROJECT'])
runname = os.environ['LANGCHAIN_RUNNAME']
thread_id = str(uuid.uuid4())

#Custom Interrupt
class AsyncInterruptHandler:
    def __init__(self, graph=None, thread_config=None):
        self.input_queue = asyncio.Queue()
        self.event = asyncio.Event()
        self.user_feedback = None
        self.graph = graph
        self.thread_config = thread_config
        self.interface = None
        self.interface_thread = None

async def resume_graph(self, user_feedback, interrupt_details: Optional[dict] = None):
    """
    Resume the graph with the user input using Command
    """
    try:
        # Use Command to resume the graph
        result = await self.graph.ainvoke(
            Command(
                resume=user_feedback,
                update={
                    "messages": AIMessage(content=str(interrupt_details), name="blueprint_team"),
                    "next": "call_togaf_agentic"
                }
            ),
            config=self.thread_config
        )
        return result
    
    except Exception as e:
        print(f"Error resuming graph: {e}")
        return None

def launch_gradio_interface(self, interrupt_details: Optional[dict] = None):
    """
    Launch Gradio interface in a separate thread
    """
    def run_interface():
        # Create event loop for this thread
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        async def async_launch():
            return await self._create_and_launch_interface(interrupt_details)

        loop.run_until_complete(async_launch())

    #Create and start the thread
    self.interface_thread = threading.Thread(target=run_interface, daemon=True)
    self.interface_thread.start()

async def _create_and_launch_interface(self, interrupt_details: Optional[dict] = None):
    """
    Async method to create and launch Gradio interface
    """
    #Dynamically create Gradio interface based on interrupt type
    if isinstance(interrupt_details, dict):
        # If interrupt provides specific details
        title = interrupt_details.get('task', 'User Input')
        description = interrupt_details.get('content', 'Please provide input')
    else:
        title = 'User Input'
        description = 'Please provide input'

    async def submit_input(user_feedback):
        await self.input_queue.put(user_feedback)
        self.event.set()

        return {"user_feedback": user_feedback}

    #Dynamically create interface based on interrupt details
    with gr.Blocks() as demo:
        gr.Markdown(f"### {title}")
        gr.Markdown(f"**{description}")

        input_component = gr.Textbox(label="Your input")
        submit_btn = gr.Button("Submit")
        output = gr.Textbox(label="Status")

        submit_btn.click(
            submit_input, 
            inputs=input_component, 
            outputs=output
        )

    #Reset the event
    self.event = asyncio.Event()
    self.event.clear()
    self.user_feedback = None

    #Store the interface
    self.interface = demo

    #Launch the interface
    self.interface.launch(prevent_thread_lock=True, debug=True, inbrowser=True)

async def handle_interrupt(self, interrupt_details: Optional[dict] = None):
    """
    Handle interrupt by launching Gradio interface and waiting for input
    """
    # Launch the interface in a separate thread
    self.launch_gradio_interface(interrupt_details)

    try:
        #Retrieve the input from the queue
        self.user_feedback = await self.input_queue.get()

        #Resume the graph with the user input // NOT WORKING
        return await self.resume_graph(self.user_feedback, interrupt_details)

    except asyncio.TimeoutError:
        print("Gradio interface timed out")
        return None
    except Exception as e:
        print(f"Error in handle_interrupt: {e}")
        return        
    finally:
        #Close the interface
        self.interface.close()

#Global interrupt handler
global_interrupt_handler = AsyncInterruptHandler()

#Custom interrupt function
async def custom_async_interrupt(details=None, graph=None, thread_config=None):

global global_interrupt_handler
global_interrupt_handler.graph = graph
#'global_interrupt_handler.thread_config = thread_config

return await global_interrupt_handler.handle_interrupt(details)

#Override the default interrupt
import langgraph.types
langgraph.types.interrupt = custom_async_interrupt

class State(MessagesState):
    next: Optional[str] = None
    user_feedback: Optional[str] = None

global super_graph

thread_config = {"configurable": {"thread_id": uuid.uuid4()}}

#CAPTURE business requirement asking for USER input & call togad_agentic workflow
async def call_togaf_agentic(state: State) -> Command[Literal["__end__"]]:

    business_query = state["messages"][-1].content

    print("CALLING togaf_graph!!!")

    return Command(
        update={
            "messages": [
                HumanMessage(
                    content=business_query, name="blueprint_team"
                )
            ],
        },
        goto=END,
    )

def blueprint_team(state: State) -> Command:
    print("---Blueprint team got a request---")

async def togaf_ask_human(state: State, graph=None) -> dict:
    global super_graph

    print("---Ask User for Input---")
    user_feedback = await langgraph.types.interrupt(
        {
            "task": state['messages'][-1].content,
            "content": "Please provide your business requirement in the form of document/file name or past the content:",
            "optional": False
        },
        graph=super_graph,
        #'thread_config=thread_config
    )

    return Command(
        update={
            "messages": [
                HumanMessage(
                    content=f"{state['messages'][-1].content} User_Input: {user_feedback}", name="landscape_agentic"
                )
            ]
        },
        goto=call_togaf_agentic,
    )

#Define the graph.
super_builder = StateGraph(State)
super_builder.add_node("blueprint_team", blueprint_team)
super_builder.add_node("generate_togaf", call_togaf_agentic)
super_builder.add_node("ask_user_input", togaf_ask_human)

super_builder.add_edge(START, "blueprint_team")
super_builder.add_edge("blueprint_team", "ask_user_input")
super_builder.add_edge("ask_user_input", "generate_togaf")
super_builder.add_edge("generate_togaf", END)

memory = MemorySaver()
super_graph = super_builder.compile(checkpointer=memory)
super_graph.name = "EA4ALL Agentic Workflow Graph"

`

@hinthornw , @vbarda , apologies in advance for directly mentioning you both in but haven't been able to make this work. If you may have a look at, it'd be greatly appreciated. Thanks, Alexandre