Using interrupts to approve upcoming tool calls

[rflx123]

@dosu
I have a tool that is invoked by an agent, and it may happen that the agent makes several tool calls in sequence. Before each individual tool call, the user should verify the upcoming tool call. The user should also have the option to reject a tool call and provide a description explaining the reason for the rejection.

Rejection should only affect the specific tool call in question, without impacting the previous or subsequent tool calls.

At the end, the LLM agent should receive the tool messages from the successful tool calls, as well as the tool messages from the rejected ones, including the respective reasons for their rejection.

[dosubot[bot]]

Hello @rflx123! I'm here to help you with any bugs, questions, or contributions you may have. Let's work on this together!

To implement a system where users can approve or reject tool calls with reasons, you can follow these steps:

1. Intercept Tool Calls: Before executing each tool call, present the call details to the user for approval. This can be done by pausing the execution and waiting for user input.

2. User Decision: Allow the user to either approve or reject the tool call. If rejected, prompt the user to provide a reason for the rejection.

3. Execute or Record:

   • If approved, execute the tool call and capture the result.
   • If rejected, create a ToolMessage that includes the rejection reason.

4. Communicate Results: After processing all tool calls, compile the results:

   • For approved calls, include the successful tool outputs.
   • For rejected calls, include the rejection messages with reasons.

5. Return to LLM Agent: Send all ToolMessages back to the LLM agent, ensuring each message includes the tool_call_id to match it with the original tool call.

This approach ensures that each tool call is independently verified and processed, allowing for a clear communication of both successful and rejected tool calls to the LLM agent.

To continue talking to Dosu, mention @dosu.

[rflx123]

@dosu Can you give me an example with LangGraph in Python?

[dosubot[bot]]

Certainly! Here's an example of how you might implement this using LangGraph in Python:

class ToolCall:
    def __init__(self, tool_call_id, details):
        self.tool_call_id = tool_call_id
        self.details = details
        self.approved = None
        self.rejection_reason = None

    def request_approval(self):
        print(f"Tool Call ID: {self.tool_call_id}")
        print(f"Details: {self.details}")
        decision = input("Approve this tool call? (yes/no): ").strip().lower()
        if decision == 'yes':
            self.approved = True
        else:
            self.approved = False
            self.rejection_reason = input("Please provide a reason for rejection: ")

    def execute(self):
        if self.approved:
            # Simulate tool execution
            return f"Result of tool call {self.tool_call_id}"
        else:
            return f"Rejected tool call {self.tool_call_id}: {self.rejection_reason}"

def process_tool_calls(tool_calls):
    results = []
    for tool_call in tool_calls:
        tool_call.request_approval()
        result = tool_call.execute()
        results.append({
            'tool_call_id': tool_call.tool_call_id,
            'result': result
        })
    return results

# Example usage
tool_calls = [
    ToolCall(1, "Tool call details for ID 1"),
    ToolCall(2, "Tool call details for ID 2"),
    ToolCall(3, "Tool call details for ID 3")
]

results = process_tool_calls(tool_calls)

# Send results back to the LLM agent
for result in results:
    print(result)

In this example:

• Each ToolCall object represents a tool call with a unique ID and details.
• The request_approval method prompts the user to approve or reject the call and, if rejected, asks for a reason.
• The execute method simulates the execution of the tool call or records the rejection reason.
• The process_tool_calls function processes a list of tool calls, collecting results for each one.
• Finally, the results are printed, which you can replace with sending them back to the LLM agent.

This setup allows each tool call to be independently approved or rejected, with reasons for rejections being recorded and communicated.

To continue talking to Dosu, mention @dosu.

[rflx123]

@dosu Thanks, but give me an example with a Graph, where the tool that will be executed is a ToolNode. Also I want to do this with Human-in-the-Loop interrupts.

[dosubot[bot]]

Here's an example using a Graph in Python with a ToolNode, incorporating Human-in-the-Loop interrupts for approval or rejection:

import json
from typing import Dict, List
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable, RunnablePassthrough
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langchain.callbacks import HumanApprovalCallbackHandler, HumanRejectedException

# Define a custom exception for not approved actions
class NotApproved(Exception):
    """Custom exception."""

# Define dummy tools
@tool
def count_emails(last_n_days: int) -> int:
    """Dummy function to count number of e-mails. Returns 2 * last_n_days."""
    return last_n_days * 2

@tool
def send_email(message: str, recipient: str) -> str:
    """Dummy function for sending an e-mail."""
    return f"Successfully sent email to {recipient}."

# Initialize the language model and bind tools
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=0)
tools = [count_emails, send_email]
llm_with_tools = llm.bind_tools(tools)

# Define a function to call tools
def call_tools(msg: AIMessage) -> List[Dict]:
    """Simple sequential tool calling helper."""
    tool_map = {tool.name: tool for tool in tools}
    tool_calls = msg.tool_calls.copy()
    for tool_call in tool_calls:
        tool_call["output"] = tool_map[tool_call["name"]].invoke(tool_call["args"])
    return tool_calls

# Define a function for human approval using HumanApprovalCallbackHandler
def human_approval(msg: AIMessage) -> AIMessage:
    """Responsible for passing through its input or raising an exception.

    Args:
        msg: output from the chat model

    Returns:
        msg: original output from the msg
    """
    tool_strs = "\n\n".join(
        json.dumps(tool_call, indent=2) for tool_call in msg.tool_calls
    )
    input_msg = (
        f"Do you approve of the following tool invocations\n\n{tool_strs}\n\n"
        "Anything except 'Y'/'Yes' (case-insensitive) will be treated as a no.\n >>>"
    )
    resp = input(input_msg)
    if resp.lower() not in ("yes", "y"):
        raise NotApproved(f"Tool invocations not approved:\n\n{tool_strs}")
    return msg

# Create a chain with human approval
chain = llm_with_tools | human_approval | call_tools

# Example invocation
try:
    chain.invoke("Send sally@gmail.com an email saying 'What's up homie'")
except NotApproved as e:
    print()
    print(e)

This example demonstrates how to incorporate a human-in-the-loop interrupt for tool execution using a ToolNode in a Python graph. The human_approval function asks for human approval before executing the tool, and raises a NotApproved exception if the approval is not granted ^[1][2][3].

To continue talking to Dosu, mention @dosu.