Human-in-the-loop

There are certain tools that we don’t trust a model to execute on its own. One thing we can do in such situations is require human approval before the tool is invoked.

Setup

We’ll need to install the following packages:

%pip install --upgrade --quiet langchain langchain-openai

And set these environment variables:

import getpass
import os

os.environ["OPENAI_API_KEY"] = getpass.getpass()

# If you'd like to use LangSmith, uncomment the below:
# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()

Chain

Suppose we have the following (dummy) tools and tool-calling chain:

from operator import itemgetter

from langchain.output_parsers import JsonOutputToolsParser
from langchain_core.runnables import Runnable, RunnableLambda, RunnablePassthrough
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI


@tool
def count_emails(last_n_days: int) -> int:
    """Multiply two integers together."""
    return last_n_days * 2


@tool
def send_email(message: str, recipient: str) -> str:
    "Add two integers."
    return f"Successfully sent email to {recipient}."


tools = [count_emails, send_email]
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0).bind_tools(tools)


def call_tool(tool_invocation: dict) -> Runnable:
    """Function for dynamically constructing the end of the chain based on the model-selected tool."""
    tool_map = {tool.name: tool for tool in tools}
    tool = tool_map[tool_invocation["type"]]
    return RunnablePassthrough.assign(output=itemgetter("args") | tool)


# .map() allows us to apply a function to a list of inputs.
call_tool_list = RunnableLambda(call_tool).map()
chain = model | JsonOutputToolsParser() | call_tool_list
chain.invoke("how many emails did i get in the last 5 days?")

[{'type': 'count_emails', 'args': {'last_n_days': 5}, 'output': 10}]

Adding human approval

We can add a simple human approval step to our tool_chain function:

import json


def human_approval(tool_invocations: list) -> Runnable:
    tool_strs = "\n\n".join(
        json.dumps(tool_call, indent=2) for tool_call in tool_invocations
    )
    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."
    )
    resp = input(msg)
    if resp.lower() not in ("yes", "y"):
        raise ValueError(f"Tool invocations not approved:\n\n{tool_strs}")
    return tool_invocations

chain = model | JsonOutputToolsParser() | human_approval | call_tool_list
chain.invoke("how many emails did i get in the last 5 days?")

Do you approve of the following tool invocations

{
  "type": "count_emails",
  "args": {
    "last_n_days": 5
  }
}

Anything except 'Y'/'Yes' (case-insensitive) will be treated as a no. y

[{'type': 'count_emails', 'args': {'last_n_days': 5}, 'output': 10}]

chain.invoke("Send sally@gmail.com an email saying 'What's up homie'")

Do you approve of the following tool invocations

{
  "type": "send_email",
  "args": {
    "message": "What's up homie",
    "recipient": "sally@gmail.com"
  }
}

Anything except 'Y'/'Yes' (case-insensitive) will be treated as a no. no

ValueError: Tool invocations not approved:

{
  "type": "send_email",
  "args": {
    "message": "What's up homie",
    "recipient": "sally@gmail.com"
  }
}