langgraph/examples/human_in_the_loop/wait-user-input.ipynb

How to wait for user input

One of the main human-in-the-loop interaction patterns is waiting for human input. A key use case involves asking the user clarifying questions. One way to accomplish this is simply go to the END node and exit the graph. Then, any user response comes back in as fresh invocation of the graph. This is basically just creating a chatbot architecture.

The issue with this is it is tough to resume back in a particular point in the graph. Often times the agent is halfway through some process, and just needs a bit of a user input. Although it is possible to design your graph in such a way where you have a conditional_entry_point to route user messages back to the right place, that is not super scalable (as it essentially involves having a routing function that can end up almost anywhere).

A separate way to do this is to have a node explicitly for getting user input. This is easy to implement in a notebook setting - you just put an input() call in the node. But that isn't exactly production ready.

Luckily, LangGraph makes it possible to do similar things in a production way. The basic idea is:

• Set up a node that represents human input. This can have specific incoming/outgoing edges (as you desire). There shouldn't actually be any logic inside this node.
• Add a breakpoint before the node. This will stop the graph before this node executes (which is good, because there's no real logic in it anyways)
• Use .update_state to update the state of the graph. Pass in whatever human response you get. The key here is to use the as_node parameter to apply this update as if you were that node. This will have the effect of making it so that when you resume execution next it resumes as if that node just acted, and not from the beginning.

Note: this requires passing in a checkpointer.

Below is a quick example.

Setup

First we need to install the packages required

In [1]:

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic

Next, we need to set API keys for Anthropic (the LLM we will use)

In [2]:

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

Optionally, we can set API key for LangSmith tracing, which will give us best-in-class observability.

In [3]:

os.environ["LANGCHAIN_TRACING_V2"] = "true"
_set_env("LANGCHAIN_API_KEY")

Build the agent

We can now build the agent. We will build a relatively simple ReAct-style agent that does tool calling. We will use Anthropic's models and a fake tool (just for demo purposes).

In [2]:

# Set up the state
from langgraph.graph import MessagesState

# Set up the tool
# We will have one real tool - a search tool
# We'll also have one "fake" tool - a "ask_human" tool
# Here we define any ACTUAL tools
from langchain_core.tools import tool
from langgraph.prebuilt import ToolExecutor

@tool
def search(query: str):
    """Call to surf the web."""
    # This is a placeholder for the actual implementation
    # Don't let the LLM know this though 😊
    return [
        f"I looked up: {query}. Result: It's sunny in San Francisco, but you better look out if you're a Gemini 😈."
    ]


tools = [search]
tool_executor = ToolExecutor(tools)

# Set up the model
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-sonnet-20240620")


# We are going "bind" all tools to the model
# We have the ACTUAL tools from above, but we also need a mock tool to ask a human
# Since `bind_tools` takes in tools but also just tool definitions,
# We can define a tool definition for `ask_human`

from langchain_core.pydantic_v1 import BaseModel

class AskHuman(BaseModel):
    """Ask the human a question"""
    question: str


model = model.bind_tools(tools + [AskHuman])

# Define nodes and conditional edges

from langchain_core.messages import ToolMessage

from langgraph.prebuilt import ToolInvocation


# Define the function that determines whether to continue or not
def should_continue(state):
    messages = state["messages"]
    last_message = messages[-1]
    # If there is no function call, then we finish
    if not last_message.tool_calls:
        return "end"
    # If tool call is asking Human, we return that node
    # You could also add logic here to let some system know that there's something that requires Human input
    # For example, send a slack message, etc
    elif last_message.tool_calls[0]['name'] == "AskHuman":
        return "ask_human"
    # Otherwise if there is, we continue
    else:
        return "continue"


# Define the function that calls the model
def call_model(state):
    messages = state["messages"]
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}


# Define the function to execute tools
def call_tool(state):
    messages = state["messages"]
    # Based on the continue condition
    # we know the last message involves a function call
    last_message = messages[-1]
    # We construct an ToolInvocation from the function_call
    tool_call = last_message.tool_calls[0]
    action = ToolInvocation(
        tool=tool_call["name"],
        tool_input=tool_call["args"],
    )
    # We call the tool_executor and get back a response
    response = tool_executor.invoke(action)
    # We use the response to create a ToolMessage
    tool_message = ToolMessage(
        content=str(response), name=action.tool, tool_call_id=tool_call["id"]
    )
    # We return a list, because this will get added to the existing list
    return {"messages": [tool_message]}

# We define a fake node to ask the human
def ask_human(state):
    pass

# Build the graph

from langgraph.graph import END, StateGraph

# Define a new graph
workflow = StateGraph(MessagesState)

# Define the three nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)
workflow.add_node("ask_human", ask_human)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    # Finally we pass in a mapping.
    # The keys are strings, and the values are other nodes.
    # END is a special node marking that the graph should finish.
    # What will happen is we will call `should_continue`, and then the output of that
    # will be matched against the keys in this mapping.
    # Based on which one it matches, that node will then be called.
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # We may ask the human
        "ask_human": "ask_human",
        # Otherwise we finish.
        "end": END,
    },
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")

# After we get back the human response, we go back to the agent
workflow.add_edge("ask_human", "agent")

# Set up memory
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
# We add a breakpoint BEFORE the `ask_human` node so it never executes
app = workflow.compile(checkpointer=memory, interrupt_before=['ask_human'])

Interacting with the Agent

We can now interact with the agent. Let's ask it to ask the user where they are, then tell them the weather. This should make it use the ask_human tool first, then use the normal tool.

In [18]:

from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(content="Use the search tool to ask the user where they are, then look up the weather there")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
    event["messages"][-1].pretty_print()

================================ Human Message =================================

Use the search tool to ask the user where they are, then look up the weather there
================================== Ai Message ==================================

[{'text': "Certainly! I'll use the AskHuman function to ask the user where they are, and then I'll use the search function to look up the weather for that location. Let's start by asking the user about their location.", 'type': 'text'}, {'id': 'toolu_01VbhJQTHN44bzfWHwPh2KK6', 'input': {'question': 'Where are you currently located?'}, 'name': 'AskHuman', 'type': 'tool_use'}]
Tool Calls:
  AskHuman (toolu_01VbhJQTHN44bzfWHwPh2KK6)
 Call ID: toolu_01VbhJQTHN44bzfWHwPh2KK6
  Args:
    question: Where are you currently located?

We now want to update this thread with a response from the user. We then can kick off another run.

Because we are treating this as a tool call, we will need to update the state as if it is a response from a tool call. In order to do this, we will need to check the state to get the ID of the tool call.

In [19]:

tool_call_id = app.get_state(config).values['messages'][-1].tool_calls[0]['id']

# We now create the tool call with the id and the response we want
tool_message = [{"tool_call_id": tool_call_id, "type": "tool", "content": "san francisco"}]

# # This is equivalent to the below, either one works
# from langchain_core.messages import ToolMessage
# tool_message = [ToolMessage(tool_call_id=tool_call_id, content="san francisco")]

# We now update the state
# Notice that we are also specifying `as_node="ask_human"`
# This will apply this update as this node,
# which will make it so that afterwards it continues as normal
app.update_state(config, {"messages": tool_message}, as_node="ask_human")

# We can check the state
# We can see that the state currently has the `agent` node next
# This is based on how we define our graph, 
# where after the `ask_human` node goes (which we just triggered)
# there is an edge to the `agent` node
app.get_state(config).next

Out [19]:

('agent',)

We can now tell the agent to continue. We can just pass in None as the input to the graph, since no additional input is needed

In [20]:

for event in app.stream(None, config, stream_mode="values"):
    event["messages"][-1].pretty_print()

================================== Ai Message ==================================

[{'text': "Thank you for letting me know that you're in San Francisco. Now, I'll use the search function to look up the weather in San Francisco.", 'type': 'text'}, {'id': 'toolu_013cvwkiiDBrHV9w1e97SvqB', 'input': {'query': 'current weather in San Francisco'}, 'name': 'search', 'type': 'tool_use'}]
Tool Calls:
  search (toolu_013cvwkiiDBrHV9w1e97SvqB)
 Call ID: toolu_013cvwkiiDBrHV9w1e97SvqB
  Args:
    query: current weather in San Francisco
================================= Tool Message =================================
Name: search

["I looked up: current weather in San Francisco. Result: It's sunny in San Francisco, but you better look out if you're a Gemini 😈."]
================================== Ai Message ==================================

Based on the search results, I can provide you with information about the current weather in San Francisco:

The weather in San Francisco is currently sunny. This is great news for outdoor activities and enjoying the city's beautiful scenery.

However, there was an unusual and somewhat humorous addition to the weather report, mentioning something about Geminis. This appears to be a joke or possibly a reference to an astrological forecast mixed in with the weather information. It's not typical for weather reports to include astrological references, so we should focus on the factual weather information, which is that it's sunny in San Francisco.

Is there anything else you'd like to know about the weather or San Francisco in general?

In [ ]: