langgraph/concepts/human_in_the_loop/

[giscus[bot]]

langgraph/concepts/human_in_the_loop/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

[phfifofum]

I have a backend, fast api and front end streamlit architecture.... struggling to get this working well.

I can edit and resume the graph fine from the back end and fast api using checkpointing.

What I cant do well (or cleanly) is trigger a graph update from streamlit front end so the events stream nicely into my existing graph invoke code and event/UI handling.

Any insights or code examples for this?

[idotr7]

i have this problem a well, it's not clear how to make it work

[DuncanRiv]

If you are solely using streamlit on the frontend, this might not be possible. (I don't have much experience with streamlit). You can probably pull this off by making successive API calls with additional python libraries, like trying using streamlit in tandem with the "requests" python library.

Regardless,
I always recommend using a javascript library or just plain javascript for the frontend and just establish a streaming or event based API connection. It gives you a lot more flexibility and your UI/UX will get improved! WIN-WIN.

If you do choose to go down this route, a nice way that I have found to accomplish this is by routing between "Initializing"/"Starting"/"Continuing" the graph WITHOUT human input and "Resuming the graph" WITH human input. This is because your graph invocation is going to look different for both cases. All it takes is just a slightly modified post request from your frontend for each case.

IMPORTANT On your frontend you need to keep track of your thread_id.

I will give you an example just using astream because we want our applications to be as close to production-ready as possible:

Step 1: On your frontend, establish some boolean to decide whether to resume or not. Call it resume = False initially.

Note: I am only going to include the necessary details below because this is a lot to type out lol

Step 2: Setup your post API endpoint in FASTAPI roughly like this:

  async def apicall(state):
  
         try:
               if state[resume]: ****ASSUMING TYPEDICT (state.resume IF USING PYDANTIC)
               
                   THIS WILL HANDLE THE RESUMING WITH HUMAN INPUT
                   
                async for chunk in graph.astream(resume= *insert variable needed to resume), config=thread_config, stream_mode=. 
                 ["updates","custom"]):
                 
                 print(chunk) etc.....
                 
                 
                 else:
                 THIS WILL HANDLE EVERYTHING ELSE
                 async for chunk in graph.astream({"messages":input_message},  config=thread_config, stream_mode=["updates","custom"]):
                 print(chunk) etc.....

  
  catch/Except:
       etc....

Step 3:

Setup the node you wish to interrupt in:

def One_node(state:ChatbotState, writer:StreamWriter) -> Command:
print("I am in One Node")

question = "What node would you like to go to next?"





next_node = interrupt(question)

return Command(goto=next_node)

I know the details are rough, but hopefully you get the point: Just route between a human graph invocation and a regular graph invocation using a boolean from your frontend and remember to use the same thread_id!

If still not clear, don't worry I am going to make a tutorial soon on using LangGraph in combination with frontends. I know how confusing the continuously changing docs can get. I might even do my own version of a LangGraph course on YouTube soon just to supplement and clarify confusions. I will post a link here if that happens :)

[cris-m]

I have try this and I am still getting HumanMessage instead of AIMessage after resuming the state.

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

def human_node(state: MessagesState):
    value = interrupt(f"What should I say in response to {state['messages']}")

    return {"messages": [{"role": "user", "content": value}]}

checkpointer = MemorySaver()
graph_builder = StateGraph(MessagesState)
graph_builder.add_node(human_node)
graph_builder.set_entry_point("human_node")
graph = graph_builder.compile(
    checkpointer=checkpointer
)

First call

thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke({"messages": [{ "role": "user", "content": "Hello!"}]}, config=thread_config)

I go the following out put
{'messages': [HumanMessage(content='Hello!', additional_kwargs={}, response_metadata={}, id='dce77926-559a-4900-bf80-63e3879aecd9')]}

Now resume the graph:

graph.invoke(Command(resume="Hello, How can i help you?"), config=thread_config)

The output is 2 HummanMessage:

{'messages': [HumanMessage(content='Hello!', additional_kwargs={}, response_metadata={}, id='dce77926-559a-4900-bf80-63e3879aecd9'),
  HumanMessage(content='Hello, How can i help you?', additional_kwargs={}, response_metadata={}, id='e20ad3e8-86d9-4bd1-a7b1-c17f88a8a421')]}

[eyurtsev]

I have try this and I am still getting HumanMessage instead of AIMessage after resuming the state.

The code that you shared is expected to be doing that. There's no node for a chat model to add an AIMessage.

What you're doing is:

1. graph.invoke({"messages": [{ "role": "user", "content": "Hello!"}]}, config=thread_config) writing a human message to the chat history
2. value = interrupt(f"What should I say in response to {state['messages']}") pausing immediately to collect human input
3. graph.invoke(Command(resume="Hello, How can i help you?"), config=thread_config) followed by: return {"messages": [{"role": "user", "content": value}]}
   writing another human message to the chat history
4. ending the graph execution.

---

I recommend starting from reading the langgraph tutorial before doing any of the concept pages or how-to guides: https://langchain-ai.github.io/langgraph/tutorials/introduction/

HIL isn't needed if the graph only contains a single agent. (In the shared code there isn't even a single agent / llm call.)

[vigneshmj1997]

Hey i am using interrupt inside sub graph and when i saw the writes variable of the interrupt
it was only half filled Writes [('00000000-0000-0000-0000-000000000000', 'resume', '3')]
any idea why ?

[vigneshmj1997]

Hey My interrupt is going in a infinite loop any solutions ?

[eyurtsev]

Start here if you're looking for help: https://stackoverflow.com/help/minimal-reproducible-example

[mnuryar]

it is said that "Resuming from an interrupt is different from Python's input() function, where execution resumes from the exact point where the input() function was called.".

Is there a way to let execution resume from the exact point where interrupt() function was called, just like input() function?

[mnuryar]

Let's say there is a graph with multiple human in the loop nodes
Below is the order of nodes in the graph:

step 1 node
human input node 1
step 2 node
human input node 2

I am streaming the graph with following:

for event in graph.stream(input, config):
	if '__interrupt__' in event.keys():
		human_input = "..."
		graph.stram(Comman(resume=human_input), config):	
	else:
		for value in event.values()
			print(agent_response)

When 'human input node 1' hits the interrupt function, it stops the graph and resumes the graph execution with the 'human_input" value.
After this, instead of going to 'step 2 node, it starts all over again from the very beginning of the graph (i.e., step 1 node). Basically, the 'interrupt' functions breaks the
for loop at the top, and remaining nodes ('step 2 node' and 'human input node 2') don't get executed.

The documentation says the interrupt function resumes from the node where it is applied, and I expected it to finish executing all the nodes in the graph.
Why does it exit the for loop?

[avfranco-br]

It seems not possible to interrupt and resume on the same streaming. I'm struggling to make this work. In my use case, the workflow (super_graph, sub_graph, Gradio) is working when the graph is tested from a Jupyter notebook making a first call to the graph, interrupt, next calling the Gradio app, then a new call to resume with the user input as code below. What I am trying to achieve, don't know if possible, is to trigger the call to _get_user_input() and resume the graph directly from the graph structure (nodes). At this way, to use the graph it would be just a one call to the super_graph but not working.

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

user_input = await _get_user_input()

async for chunk in super_graph.astream(
        Command(resume=user_input),
        config=thread_config,
        subgraphs=True,
    ):
        print(chunk)

`

[mnuryar]

essentially, I am trying to get the same thing as you do. The scenario I described is for my front end application using FastAPI and Websocket ( equivalent of Gradio in your case).

The examples in the documentation work great in Jupyter note book where the human in the loop has been implemented on graphs with very simple structure. As others were asking, the LangChain team hasn't shown an example where this can be implemented seamlessly in a production environment, both from backend and front end perspective.

[avfranco-br]

essentially, I am trying to get the same thing as you do. The scenario I described is for my front end application using FastAPI and Websocket ( equivalent of Gradio in your case).

The examples in the documentation work great in Jupyter note book where the human in the loop has been implemented on graphs with very simple structure. As others were asking, the LangChain team hasn't shown an example where this can be implemented seamlessly in a production environment, both from backend and front end perspective.

Forgot to mention one thing that may help in your case. The first option I tried was to override the Interrupt class with a custom_async_interrupt function and class but as I'm not an expert handling async, multi-thread I got stuck and pivoted for now. However maybe something you can try it out.

[mnuryar]

Let's say there is a graph with multiple human in the loop nodes Below is the order of nodes in the graph:

step 1 node human input node 1 step 2 node human input node 2

I am streaming the graph with following:

for event in graph.stream(input, config):
	if '__interrupt__' in event.keys():
		human_input = "..."
		graph.stram(Comman(resume=human_input), config):	
	else:
		for value in event.values()
			print(agent_response)

When 'human input node 1' hits the interrupt function, it stops the graph and resumes the graph execution with the 'human_input" value. After this, instead of going to 'step 2 node, it starts all over again from the very beginning of the graph (i.e., step 1 node). Basically, the 'interrupt' functions breaks the for loop at the top, and remaining nodes ('step 2 node' and 'human input node 2') don't get executed.

The documentation says the interrupt function resumes from the node where it is applied, and I expected it to finish executing all the nodes in the graph. Why does it exit the for loop?

I was able to figure out why it exists the for loop. The reason was due to anther while loop came before this.

But I got another problem.
The for loop was able to handle the 'human input node 1'. But when the execution reaches to the 'human input node 2', the code breaks because
graph.stram(Comman(resume=human_input), config)
wants to execute the rest of the graph till the end. I want the graph execute only 'human input 1' node with the line above, and go back to the for loop to execute the rest of the nodes.

Any ideas how I can do that?

[Gonzalob90]

When multiple nodes raise an interrupt, it looks like only one interrupt can be pending at a time. Is there a way to handle multiple interrupts independently within the same graph?

[azmathmoosa]

I have a situation where I have a chat interface and a ticketing tool that needs to confirm with user. I want to know who caused the interrupt so I can prompt the user outside the graph with appropriate questions.

My ticketing tool has this code

print("Would you like to raise ticket with below description?")  # i want to move this outside graph 
print(f"{issue_description}")   # it prints twice
print(f"Yes/No (default: No):")
waiting_for_user_flag.set()
user_confirmation = interrupt("ticket_confirmation")
if "yes" in user_confirmation.lower():
    user_confirmation = True 
else:
    user_confirmation = False

and my user_node code is

def node_user(state: State):
    # print("NODE:: node_user")
    waiting_for_user_flag.set()
    user = interrupt("get user input")
    return {"messages": user}

and this is how I am trying to loop

config = {"configurable":{"thread_id": 20}}
while not goodbye_flag.is_set():
    if waiting_for_user_flag.is_set():  #who caused the interrupt? was it tool or was it normal chat flow
        user = input("User: ")
        waiting_for_user_flag.clear()        
        start_state = Command(resume=user)    
    else:
        start_state = initial_state

    stream = app.stream(start_state, config=config, stream_mode="messages")
    for msg, metadata in stream:
        print(msg.content, end="")
        try:
            if msg.response_metadata['done']:
                print()
        except KeyError:
            pass

My problem is that I am not able to differentiate between who caused the interrupt? Since the tool node is called again when i resume, i am seeing it print two times the confirmation question. I was hoping that this would be a simple use case but its convoluted like crazy for no reason.

Can someone guide me to some code that handles similar case?

[udaylunawat]

@phfifofum @DuncanRiv @mnuryar Can anyone of you provide with a minimal example of how you've implemented FastAPI with HITL using Interrupt? Or just point me to any resource. I'd be of great help!

[willjrho]

This is what i am looking for as well

[TillSimon]

You can also have a look at the example Sales Agent here:
https://github.com/gotohuman/examples-langgraph-py
It's a FastAPI endpoint using the interrupt to wait for human approval/feedback and resumes with

await graph.ainvoke(Command(resume={ "response": approval, "reviewed_email": email_text, "comment": retry_comment }), config=thread_config)

Might also help you @udaylunawat and @mnuryar @azmathmoosa (see how I pass values from and to the interrupt)

[antoniowhbh]

I have a problem, I'm using the interrupt and defined it in a separate node in my graph to use it with the agent inbox ui and hosted my graph locally using "langgraph dev". The interrupt actions are only functioning when the interrupt is on the first message of the thread, when I try to accept or edit or do any other action on the interrupt that is not on the first message on the thread it does nothing. if the I did this exactly the same way define in the agent inbox repo and I tried it with the examples that creates and interrupts them and confirms the jokes using the ui. I am also facing this exact problem with example graph in this repository https://github.com/langchain-ai/agent-inbox-langgraph-example I don't know if the problem is in the graph itself or how the agent inbox ui processes and submits the interrupts. I tried looking at the source code of agent inbox but I couldn't find the issue

[willjrho]

having a similar issue with interrupt exposed via an api. the graph crashes without an opportunity to get a state update from a second endpoint to resume. the intended behavior is:

-get to the interupt node in the graph
-return info to the user
-user confirms via second endpoint
-graph state updates and either continues to the next node or ends

[anjijava16]

@phfifofum use as Websocket apporach Front end (.js will call FastAPI websocket) internally Fastapi websocket will call to langgraph human in the loop approach

[junbo2001]

我在使用人机交互，想实现一个用户输入的断点，以下是我的测试代码：

     def node_1(state: States):
        print('经过：node_1')
        return {"status": 'node_1'}
    def node_2(state: States):
        print('经过：node_2')
        return {"status": 'node_2'}
    def human_approval(state: States):
        print('断电之前')
        value = interrupt(
            {
                "feedback": "验证你是否是真人，请输入1+1的答案"
            }
        )
        print('断电之后')
        return {"some_text": value}

    def node_3(state: States):
        print('经过：node_3')
        return {"status": 'node_3'}

    graph_builder = StateGraph(States)
    graph_builder.add_node("node_1", node_1)
    graph_builder.add_node("node_2", node_2)
    graph_builder.add_node("human_approval", human_approval)
    graph_builder.add_node("node_3", node_3)

    graph_builder.set_entry_point("node_1")
    graph_builder.add_edge("node_1", "node_2")
    graph_builder.add_edge("node_2", "human_approval")
    graph_builder.add_edge("human_approval", "node_3")
    graph_builder.add_edge("node_3", END)
    graph = graph_builder.compile(checkpointer=memory)

    thread_config = {"configurable": {"thread_id": "some_id"}}
    if user_id == '1151':
        for chunk in graph.stream({"issue": "怎么注册账号？"}, config=thread_config):
            print(chunk)
    else:
        for chunk in graph.stream(Command(resume="2"), config=thread_config):
            print(chunk)

第一次请求我会携带user_id=1151的参数对接口发起第一次请求，然后他会打出断点信息如下：

经过：node_1
{'node_1': {'status': 'node_1'}}

经过：node_2
{'node_2': {'status': 'node_2'}}

断电之前

{'__interrupt__': (Interrupt(value={'feedback': '验证你是否是真人，请输入1+1的答案'}, resumable=True, ns=['human_approval:da68a24d-408a-2eca-24eb-2e1ba5a96819'], when='during'),)}

第二次我会携带user_id=1152（user_id不等于1151）的参数模拟第二遍请求（即用户的回答）

而在我的实际场景中，我第一遍请求时候我该怎么把反馈内容"验证你是否是真人，请输入1+1的答案"这个反馈展现到用户端？能让用户清楚的知道在输入框中输入正确值？

[TaisukeIto]

Please make the implementation clearly show the meaning of "Human-in-the-loop."
The current example feels like the developer’s self-satisfaction—it’s not clear what the code is even supposed to do.
The process doesn’t pause, and there’s no indication of what the program is capable of.
It finishes execution without ever requiring any human input. I don’t get it.

[junbo2001]

我是想实现系统中一个真人校验的功能，我感觉应该会使用到动态断点的功能，当一名用户请求我的图，我会验证这名用户的消息如果超过10轮对话，系统需要对他发起一次真人校验，使用 value = interrupt({ "feedback": "验证你是否是真人，请输入1+1的答案"} )打断，现在的问题是，我该怎么把断点信息这句话：“验证你是否是真人，请输入1+1的答案”，怎么返回给ui前端 ？并且让前端知道这是断点信息，需要用户根据指令输入内容，当用户输入正确答案，继续图的流程？

[TaisukeIto]

sample

while True:
for event in graph.stream(cmd, config=thread_config, stream_mode="updates"):
if "interrupt" in event:
user_input = input("yes or no")
cmd = Command(resume=user_input)
break
else:
break