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

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 [4]:

# 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 ToolNode

@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_node = ToolNode(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]}



# 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", tool_node)
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 [5]:

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()

================================[1m Human Message [0m=================================

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

[{'text': "Certainly! I'll use the AskHuman function to ask the user about their location, and then use the search function to look up the weather. Let's start by asking the user where they are.", 'type': 'text'}, {'id': 'toolu_01HgfTkwJ2pZhPuCuunBYUq4', 'input': {'question': 'Where are you currently located?'}, 'name': 'AskHuman', 'type': 'tool_use'}]
Tool Calls:
  AskHuman (toolu_01HgfTkwJ2pZhPuCuunBYUq4)
 Call ID: toolu_01HgfTkwJ2pZhPuCuunBYUq4
  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 [6]:

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 [6]:

('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 [7]:

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

==================================[1m Ai Message [0m==================================

[{'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_01BKzfQfHBSUcgTbz7c4gs8w', 'input': {'query': 'current weather in San Francisco'}, 'name': 'search', 'type': 'tool_use'}]
Tool Calls:
  search (toolu_01BKzfQfHBSUcgTbz7c4gs8w)
 Call ID: toolu_01BKzfQfHBSUcgTbz7c4gs8w
  Args:
    query: current weather in San Francisco
=================================[1m Tool Message [0m=================================
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 \ud83d\ude08."]
==================================[1m Ai Message [0m==================================

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. 

It's worth noting that the search result included a playful comment about Geminis, but that's not directly related to the weather information you requested. If you need more specific details about the temperature, humidity, or forecast, please let me know, and I'd be happy to search for more detailed weather information for San Francisco.

In [ ]:

16 KiB Raw Blame History

How to wait for user input

Setup

Build the agent

Interacting with the Agent

16 KiB

Raw Blame History