How to interrupt execution in Langgraph create_react_agent - stream/invoke?

[edisimo]

I am using the ReAct agent to control a robot via tools, i want to implement a stop button in my application, which should interrupt the agents query execution. I would like it to work like this:

1. human input for the agent
2. executing tools, answering and so on...
3. interrupt is received during execution
4. all tools stop, the agent stops execution, (i manually call a tool for robot to stop)
5. after stop human can input again and the agent wont try to complete the query before.

I am using create_react_agent() with MemorySaver() as checkpointer

This is my current implementation of query invocation:

def invoke(self, query: str) -> str:
        """
        Processes a user query and returns a complete response, but uses streaming internally.
        """
        messages = []
        messages.append(("human", query))
        response = ""
        total_tokens = 0
        
        try:
            # Use stream instead of invoke, but collect the entire result
            for chunk in self.__agent_executor.stream(
                {"messages": messages},
                self.__config,
                stream_mode="values",
            ):
                # Get the latest message content
                if "messages" in chunk and chunk["messages"]:
                    all_msgs = chunk["messages"]
                    new_msgs = []
                    found_last = False
                    # loop from the end to start
                    for i in range(len(all_msgs) - 1, -1, -1):
                        if all_msgs[i].id == self.last_message_id and self.last_message_id is not None:
                            found_last = True
                            new_msgs = all_msgs[i + 1:]
                            break
                    if not found_last:
                        new_msgs = all_msgs
                    
                    for msg in new_msgs:
                        print(msg.type)
                        if self.__show_token_usage and 'token_usage' in msg.response_metadata:
                            total_tokens += msg.response_metadata['token_usage']['total_tokens']



                    
                    self.last_message_id = new_msgs[-1].id

                    latest_msg = chunk["messages"][-1]
                    # For the first chunk, set response to the content
                    # For subsequent chunks, we'll get the full updated content each time
                    response = latest_msg.content
                    
                    # Store token usage if available and we're configured to show it

            # Display token usage information after streaming is complete
            if self.__show_token_usage :
                print(f"Token usage: {total_tokens}")
        except asyncio.CancelledError: 
            print("Task cancelled!")
            # Handle cancellation gracefully
            raise

        except Exception as e:

            return f"An error occurred: {str(e)}"
            
        return response

However when i try to run invoke with asyncio and then cancel the task, it doesnt seem to work, the agent prints an error, that the tool call has not been finished and then proceeds to finish from there

    Can someone help me? :)

[dosubot[bot]]

To implement a stop button that interrupts the agent's execution in LangChain, you can use asyncio to manage the task and handle cancellation gracefully. Here's a refined approach to achieve this:

1. Use asyncio.create_task to run the agent's execution asynchronously.
2. Implement a mechanism to cancel the task when the stop button is pressed.
3. Handle asyncio.CancelledError to stop the agent and any ongoing tool executions.

Here's an example of how you can modify your invoke method to support task cancellation:

import asyncio

async def invoke(self, query: str) -> str:
    messages = [("human", query)]
    response = ""
    total_tokens = 0

    try:
        # Use stream instead of invoke, but collect the entire result
        async for chunk in self.__agent_executor.stream(
            {"messages": messages},
            self.__config,
            stream_mode="values",
        ):
            if "messages" in chunk and chunk["messages"]:
                all_msgs = chunk["messages"]
                new_msgs = []
                found_last = False
                for i in range(len(all_msgs) - 1, -1, -1):
                    if all_msgs[i].id == self.last_message_id and self.last_message_id is not None:
                        found_last = True
                        new_msgs = all_msgs[i + 1:]
                        break
                if not found_last:
                    new_msgs = all_msgs

                for msg in new_msgs:
                    print(msg.type)
                    if self.__show_token_usage and 'token_usage' in msg.response_metadata:
                        total_tokens += msg.response_metadata['token_usage']['total_tokens']

                self.last_message_id = new_msgs[-1].id
                latest_msg = chunk["messages"][-1]
                response = latest_msg.content

        if self.__show_token_usage:
            print(f"Token usage: {total_tokens}")

    except asyncio.CancelledError:
        print("Task cancelled!")
        # Call a tool to stop the robot here
        await self.stop_robot_tool()
        return "Execution stopped by user."

    except Exception as e:
        return f"An error occurred: {str(e)}"

    return response

# Example usage
async def main():
    task = asyncio.create_task(invoke(query="Your query here"))
    # Simulate a stop button press after some time
    await asyncio.sleep(1)  # Adjust the sleep time as needed
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled successfully.")

# Run the main function
asyncio.run(main())

Key Points:

• Use asyncio.create_task to run the invoke method asynchronously.
• Use task.cancel() to cancel the task when the stop button is pressed.
• Handle asyncio.CancelledError to perform any necessary cleanup, such as stopping the robot.

This setup should allow you to interrupt the agent's execution and handle the stop action gracefully.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[edisimo]

@dosu What if I want to run the tasks in a cycle? I want the functionality of being able to write new queries after i interrupt

[dosubot[bot]]

To run tasks in a cycle and allow new queries after an interruption, you can set up a loop that continuously waits for new input and processes it. Here's how you can modify the approach to support this functionality:

1. Use a loop to continuously accept new queries.
2. Allow interruption and resumption of tasks within the loop.

Here's an example implementation:

import asyncio

async def invoke(self, query: str) -> str:
    messages = [("human", query)]
    response = ""
    total_tokens = 0

    try:
        async for chunk in self.__agent_executor.stream(
            {"messages": messages},
            self.__config,
            stream_mode="values",
        ):
            if "messages" in chunk and chunk["messages"]:
                all_msgs = chunk["messages"]
                new_msgs = []
                found_last = False
                for i in range(len(all_msgs) - 1, -1, -1):
                    if all_msgs[i].id == self.last_message_id and self.last_message_id is not None:
                        found_last = True
                        new_msgs = all_msgs[i + 1:]
                        break
                if not found_last:
                    new_msgs = all_msgs

                for msg in new_msgs:
                    print(msg.type)
                    if self.__show_token_usage and 'token_usage' in msg.response_metadata:
                        total_tokens += msg.response_metadata['total_tokens']

                self.last_message_id = new_msgs[-1].id
                latest_msg = chunk["messages"][-1]
                response = latest_msg.content

        if self.__show_token_usage:
            print(f"Token usage: {total_tokens}")

    except asyncio.CancelledError:
        print("Task cancelled!")
        await self.stop_robot_tool()
        return "Execution stopped by user."

    except Exception as e:
        return f"An error occurred: {str(e)}"

    return response

async def main():
    while True:
        query = input("Enter your query (or 'exit' to quit): ")
        if query.lower() == 'exit':
            break

        task = asyncio.create_task(invoke(query=query))
        try:
            await task
        except asyncio.CancelledError:
            print("Task was cancelled successfully.")

# Run the main function
asyncio.run(main())

Key Points:

• The main function runs in a loop, continuously asking for new queries.
• You can interrupt the current task by calling task.cancel(), and the loop will allow you to input a new query afterward.
• Use input() to get user input for new queries, and handle the "exit" command to break the loop and stop the program.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[edisimo]

@dosu This still doesnt seem to work. It raises the exception, however the execution continues

If i try to send anythign in the meantime i receive this : "Found AIMessages with tool_calls that do not have a corresponding ToolMessage. Here are the first few of those tool calls: [{'name': 'speak', 'args': {'text': 'Alright, here's more! After Fred enjoyed the cheese, he thought about his next meal. He wandered deeper into the forest and came across a wise old owl perched on a branch. Fred approached the owl and asked, 'What's the secret to being clever?' The owl replied, 'The secret lies in asking questions and listening closely.' Inspired, Fred decided to pay attention not only to what others said but also to the sounds of the forest. He learned to recognize the rustle of leaves, the whisper of the wind, and even the distant calls of other animals. This new knowledge helped him become even smarter and more resourceful in finding food. And from that day on, he never underestimated the importance of listening."'}, 'id': 'call_JYJ35dKaCTqz3edOne9LjcLV', 'type': 'tool_call'}].

Every tool call (LLM requesting to call a tool) in the message history MUST have a corresponding ToolMessage (result of a tool invocation to return to the LLM) - this is required by most LLM providers.
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/INVALID_CHAT_HISTORY"

[dosubot[bot]]

To resolve the issue of "AIMessages with tool_calls that do not have a corresponding ToolMessage," you need to ensure that every tool call in your message history has a corresponding ToolMessage. Here are the steps to address this:

1. Ensure Matching ToolMessages: For each tool call in the tool_calls array of the AIMessage, you must pass back exactly one ToolMessage with a matching tool_call_id. This ensures that the tool call is properly completed and acknowledged by the system ^[1].

2. Handle Interruptions: If execution is interrupted, make sure to provide matching ToolMessages for any existing tool calls. This might involve manually adding ToolMessages to the message history if the tool call was not completed due to the interruption ^[1].

3. Update Message History: You can modify the message list to either remove any unanswered tool calls or add the corresponding ToolMessages. This can be done by updating the state of the graph with the corrected message list using graph.update_state(config, {'messages': ...}) ^[1].

By ensuring that each tool call has a corresponding ToolMessage, you can prevent the "INVALID_CHAT_HISTORY" error and maintain a consistent state in your application.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}