Skip to main content

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"
}
}