WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-10-01 22:15:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

docs: HITL consolidation (#5192)

Browse Source 
* HITL consolidation, minus server

* Fix links

* Fix server page

* remove extra page

* nits

* updates based on feedback

* Update docs/docs/concepts/human_in_the_loop.md

Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>

* edits based on feedback

---------

Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
This commit is contained in:
Lauren Hirata Singh
2025-06-25 14:16:46 -04:00
committed by GitHub
co-authored by Sydney Runkle
parent 5b8edf3c72
commit 0aefe68a5f
16 changed files with 430 additions and 2944 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md
+247 -139
View File
@@ -9,13 +9,20 @@ hide:
- tags
---
# Add human-in-the-loop
# Enable human intervention
## `interrupt`
To review, edit, and approve tool calls in an agent or workflow, use LangGraph's [human-in-the-loop](../../concepts/human_in_the_loop.md) features.
## Pause using `interrupt`
The [`interrupt` function][langgraph.types.interrupt] in LangGraph enables human-in-the-loop workflows by pausing the graph at a specific node, presenting information to a human, and resuming the graph with their input. It's useful for tasks like approvals, edits, or gathering additional context.
The graph is resumed using a [`Command`][langgraph.types.Command] object that provides the human's response.
To use `interrupt` in your graph, you need to:
1. [**Specify a checkpointer**](../../concepts/persistence.md#checkpoints) to save the graph state after each step.
2. **Call `interrupt()`** in the appropriate place. See the [Common Patterns](#common-patterns) section for examples.
3. **Run the graph** with a [**thread ID**](../../concepts/persistence.md#threads) until the `interrupt` is hit.
4. **Resume execution** using `invoke`/`ainvoke`/`stream`/`astream` (see [**The `Command` primitive**](#resume-using-the-command-primitive)).
```python
# highlight-next-line
@@ -125,34 +132,48 @@ print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
7. The graph is resumed with a `Command(resume=...)`, injecting the human's input and continuing execution.
!!! tip "New in 0.4.0"
`__interrupt__` is a special key that will be returned when running the graph if the graph is interrupted. Support for `__interrupt__` in `invoke` and `ainvoke` has been added in version 0.4.0. If you're on an older version, you will only see `__interrupt__` in the result if you use `stream` or `astream`. You can also use `graph.get_state(thread_id)` to get the interrupt value.
!!! warning
Interrupts are both powerful and ergonomic. However, while they may resemble Python's input() function in terms of developer experience, it's important to note that they do not automatically resume execution from the interruption point. Instead, they rerun the entire node where the interrupt was used.
For this reason, interrupts are typically best placed at the start of a node or in a dedicated node. Please read the [resuming from an interrupt](#how-does-resuming-from-an-interrupt-work) section for more details.
Interrupts are both powerful and ergonomic. However, while they may resemble Python's input() function in terms of developer experience, it's important to note that they do not automatically resume execution from the interruption point. Instead, they rerun the entire node where the interrupt was used. For this reason, interrupts are typically best placed at the start of a node or in a dedicated node.
## Requirements
To use `interrupt` in your graph, you need to:
## Resume using the `Command` primitive
1. [**Specify a checkpointer**](../../concepts/persistence.md#checkpoints) to save the graph state after each step.
2. **Call `interrupt()`** in the appropriate place. See the [Design Patterns](#design-patterns) section for examples.
3. **Run the graph** with a [**thread ID**](../../concepts/persistence.md#threads) until the `interrupt` is hit.
4. **Resume execution** using `invoke`/`ainvoke`/`stream`/`astream` (see [**The `Command` primitive**](#resume-using-the-command-primitive)).
!!! warning
## Design patterns
Resuming from an `interrupt` is different from Python's `input()` function, where execution resumes from the exact point where the `input()` function was called.
There are typically three different **actions** that you can do with a human-in-the-loop workflow:
When the `interrupt` function is used within a graph, execution pauses at that point and awaits user input.
1. **Approve or Reject**: Pause the graph before a critical step, such as an API call, to review and approve the action. If the action is rejected, you can prevent the graph from executing the step, and potentially take an alternative action. This pattern often involve **routing** the graph based on the human's input.
2. **Edit Graph State**: Pause the graph to review and edit the graph state. This is useful for correcting mistakes or updating the state with additional information. This pattern often involves **updating** the state with the human's input.
3. **Get Input**: Explicitly request human input at a particular step in the graph. This is useful for collecting additional information or context to inform the agent's decision-making process.
To resume execution, use the [`Command`][langgraph.types.Command] primitive, which can be supplied via the `invoke`, `ainvoke`, `stream`, or `astream` methods. The graph resumes execution from the beginning of the node where `interrupt(...)` was initially called. This time, the `interrupt` function will return the value provided in `Command(resume=value)` rather than pausing again. All code from the beginning of the node to the `interrupt` will be re-executed.
Below we show different design patterns that can be implemented using these **actions**.
```python
# Resume graph execution by providing the user's input.
graph.invoke(Command(resume={"age": "25"}), thread_config)
```
### Resume multiple interrupts with one invocation
If you have multiple interrupts in the task queue, you can use `Command.resume` with a dictionary mapping of interrupt ids to resume with a single `invoke` / `stream` call.
For example, once your graph has been interrupted (multiple times, theoretically) and is stalled:
```python
resume_map = {
i.interrupt_id: f"human input for prompt {i.value}"
for i in parent.get_state(thread_config).interrupts
}
parent_graph.invoke(Command(resume=resume_map), config=thread_config)
```
## Common patterns
Below we show different design patterns that can be implemented using `interrupt` and `Command`.
### Approve or reject
@@ -263,9 +284,7 @@ graph.invoke(Command(resume=True), config=thread_config)
print(final_result)
```
See [how to review tool calls](./review-tool-calls.ipynb) for a more detailed example.
### Review & edit state
### Review and edit state
<figure markdown="1">
![image](../../concepts/img/human_in_the_loop/edit-graph-state-simple.png){: style="max-height:400px"}
@@ -393,41 +412,209 @@ critical in applications where the tool calls requested by the LLM may be sensit
</figcaption>
</figure>
To add a human approval step to a tool:
1. Use `interrupt()` in the tool to pause execution.
2. Resume with a `Command(resume=...)` to continue based on human input.
```python
def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
# This is the value we'll be providing via Command(resume=<human_review>)
human_review = interrupt(
{
"question": "Is this correct?",
# Surface tool calls for review
"tool_call": tool_call
}
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent
# An example of a sensitive tool that requires human review / approval
def book_hotel(hotel_name: str):
"""Book a hotel"""
# highlight-next-line
response = interrupt( # (1)!
f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
"Please approve or suggest edits."
)
if response["type"] == "accept":
pass
elif response["type"] == "edit":
hotel_name = response["args"]["hotel_name"]
else:
raise ValueError(f"Unknown response type: {response['type']}")
return f"Successfully booked a stay at {hotel_name}."
review_action, review_data = human_review
# highlight-next-line
checkpointer = InMemorySaver() # (2)!
# Approve the tool call and continue
if review_action == "continue":
return Command(goto="run_tool")
# Modify the tool call manually and then continue
elif review_action == "update":
...
updated_msg = get_updated_msg(review_data)
# Remember that to modify an existing message you will need
# to pass the message with a matching ID.
return Command(goto="run_tool", update={"messages": [updated_message]})
# Give natural language feedback, and then pass that back to the agent
elif review_action == "feedback":
...
feedback_msg = get_feedback_msg(review_data)
return Command(goto="call_llm", update={"messages": [feedback_msg]})
agent = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
tools=[book_hotel],
# highlight-next-line
checkpointer=checkpointer, # (3)!
)
```
See [how to review tool calls](./review-tool-calls.ipynb) for a more detailed example.
1. The [`interrupt` function][langgraph.types.interrupt] pauses the agent graph at a specific node. In this case, we call `interrupt()` at the beginning of the tool function, which pauses the graph at the node that executes the tool. The information inside `interrupt()` (e.g., tool calls) can be presented to a human, and the graph can be resumed with the user input (tool call approval, edit or feedback).
2. The `InMemorySaver` is used to store the agent state at every step in the tool calling loop. This enables [short-term memory](../memory/add-memory.md#add-short-term-memory) and [human-in-the-loop](../../concepts/human_in_the_loop.md) capabilities. In this example, we use `InMemorySaver` to store the agent state in memory. In a production application, the agent state will be stored in a database.
3. Initialize the agent with the `checkpointer`.
### Validating human input
Run the agent with the `stream()` method, passing the `config` object to specify the thread ID. This allows the agent to resume the same conversation on future invocations.
```python
config = {
"configurable": {
# highlight-next-line
"thread_id": "1"
}
}
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
# highlight-next-line
config
):
print(chunk)
print("\n")
```
> You should see that the agent runs until it reaches the `interrupt()` call, at which point it pauses and waits for human input.
Resume the agent with a `Command(resume=...)` to continue based on human input.
```python
from langgraph.types import Command
for chunk in agent.stream(
# highlight-next-line
Command(resume={"type": "accept"}), # (1)!
# Command(resume={"type": "edit", "args": {"hotel_name": "McKittrick Hotel"}}),
config
):
print(chunk)
print("\n")
```
1. The [`interrupt` function][langgraph.types.interrupt] is used in conjunction with the [`Command`][langgraph.types.Command] object to resume the graph with a value provided by the human.
### Add interrupts to any tool
You can create a wrapper to add interrupts to *any* tool. The example below provides a reference implementation compatible with [Agent Inbox UI](https://github.com/langchain-ai/agent-inbox) and [Agent Chat UI](https://github.com/langchain-ai/agent-chat-ui).
```python title="Wrapper that adds human-in-the-loop to any tool"
from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt
def add_human_in_the_loop(
tool: Callable | BaseTool,
*,
interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
"""Wrap a tool to support human-in-the-loop review."""
if not isinstance(tool, BaseTool):
tool = create_tool(tool)
if interrupt_config is None:
interrupt_config = {
"allow_accept": True,
"allow_edit": True,
"allow_respond": True,
}
@create_tool( # (1)!
tool.name,
description=tool.description,
args_schema=tool.args_schema
)
def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
request: HumanInterrupt = {
"action_request": {
"action": tool.name,
"args": tool_input
},
"config": interrupt_config,
"description": "Please review the tool call"
}
# highlight-next-line
response = interrupt([request])[0] # (2)!
# approve the tool call
if response["type"] == "accept":
tool_response = tool.invoke(tool_input, config)
# update tool call args
elif response["type"] == "edit":
tool_input = response["args"]["args"]
tool_response = tool.invoke(tool_input, config)
# respond to the LLM with user feedback
elif response["type"] == "response":
user_feedback = response["args"]
tool_response = user_feedback
else:
raise ValueError(f"Unsupported interrupt response type: {response['type']}")
return tool_response
return call_tool_with_interrupt
```
1. This wrapper creates a new tool that calls `interrupt()` **before** executing the wrapped tool.
2. `interrupt()` is using special input and output format that's expected by [Agent Inbox UI](https://github.com/langchain-ai/agent-inbox):
- a list of [`HumanInterrupt`][langgraph.prebuilt.interrupt.HumanInterrupt] objects is sent to `AgentInbox` render interrupt information to the end user
- resume value is provided by `AgentInbox` as a list (i.e., `Command(resume=[...])`)
You can use the `add_human_in_the_loop` wrapper to add `interrupt()` to any tool without having to add it *inside* the tool:
```python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
# highlight-next-line
checkpointer = InMemorySaver()
def book_hotel(hotel_name: str):
"""Book a hotel"""
return f"Successfully booked a stay at {hotel_name}."
agent = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
tools=[
# highlight-next-line
add_human_in_the_loop(book_hotel), # (1)!
],
# highlight-next-line
checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "1"}}
# Run the agent
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
# highlight-next-line
config
):
print(chunk)
print("\n")
```
1. The `add_human_in_the_loop` wrapper is used to add `interrupt()` to the tool. This allows the agent to pause execution and wait for human input before proceeding with the tool call.
> You should see that the agent runs until it reaches the `interrupt()` call,
> at which point it pauses and waits for human input.
Resume the agent with a `Command(resume=...)` to continue based on human input.
```python
from langgraph.types import Command
for chunk in agent.stream(
# highlight-next-line
Command(resume=[{"type": "accept"}]),
# Command(resume=[{"type": "edit", "args": {"args": {"hotel_name": "McKittrick Hotel"}}}]),
config
):
print(chunk)
print("\n")
```
### Validate human input
If you need to validate the input provided by the human within the graph itself (rather than on the client side), you can achieve this by using multiple interrupt calls within a single node.
@@ -525,91 +712,15 @@ def human_node(state: State):
print(final_result) # Should include the valid age
```
## Considerations
## Resume using the `Command` primitive
When using human-in-the-loop, there are some considerations to keep in mind.
When the `interrupt` function is used within a graph, execution pauses at that point and awaits user input.
### Using with code with side-effects
To resume execution, use the [`Command`][langgraph.types.Command] primitive, which can be supplied via the `invoke`, `ainvoke`, `stream`, or `astream` methods.
Place code with side effects, such as API calls, after the `interrupt` or in a separate node to avoid duplication, as these are re-triggered every time the node is resumed.
**Providing a response to the `interrupt`:**
To continue execution, pass the user's input using `Command(resume=value)`. The graph resumes execution from the beginning of the node where `interrupt(...)` was initially called. This time, the `interrupt` function will return the value provided in `Command(resume=value)` rather than pausing again.
```python
# Resume graph execution by providing the user's input.
graph.invoke(Command(resume={"age": "25"}), thread_config)
```
## How does resuming from an interrupt work?
!!! warning
Resuming from an `interrupt` is **different** from Python's `input()` function, where execution resumes from the exact point where the `input()` function was called.
A critical aspect of using `interrupt` is understanding how resuming works. When you resume execution after an `interrupt`, graph execution starts from the **beginning** of the **graph node** where the last `interrupt` was triggered.
**All** code from the beginning of the node to the `interrupt` will be re-executed.
```python
counter = 0
def node(state: State):
# All the code from the beginning of the node to the interrupt will be re-executed
# when the graph resumes.
global counter
counter += 1
print(f"> Entered the node: {counter} # of times")
# Pause the graph and wait for user input.
answer = interrupt()
print("The value of counter is:", counter)
...
```
Upon **resuming** the graph, the counter will be incremented a second time, resulting in the following output:
```pycon
> Entered the node: 2 # of times
The value of counter is: 2
```
### Resuming multiple interrupts with one invocation
If you have multiple interrupts in the task queue, you can use `Command.resume` with a dictionary mapping
of interrupt ids to resume values to resume multiple interrupts with a single `invoke` / `stream` call.
For example, once your graph has been interrupted (multiple times, theoretically) and is stalled:
```python
resume_map = {
i.interrupt_id: f"human input for prompt {i.value}"
for i in parent.get_state(thread_config).interrupts
}
parent_graph.invoke(Command(resume=resume_map), config=thread_config)
```
## Common pitfalls
### Side-effects
Place code with side effects, such as API calls, **after** the `interrupt` to avoid duplication, as these are re-triggered every time the node is resumed.
=== "Side effects before interrupt (BAD)"
This code will re-execute the API call another time when the node is resumed from
the `interrupt`.
This can be problematic if the API call is not idempotent or is just expensive.
```python
from langgraph.types import interrupt
def human_node(state: State):
"""Human node with validation."""
api_call(...) # This code will be re-executed when the node is resumed.
answer = interrupt(question)
```
=== "Side effects after interrupt (OK)"
=== "Side effects after interrupt"
```python
from langgraph.types import interrupt
@@ -622,7 +733,7 @@ Place code with side effects, such as API calls, **after** the `interrupt` to av
api_call(answer) # OK as it's after the interrupt
```
=== "Side effects in a separate node (OK)"
=== "Side effects in a separate node"
```python
from langgraph.types import interrupt
@@ -640,11 +751,9 @@ Place code with side effects, such as API calls, **after** the `interrupt` to av
api_call(...) # OK as it's in a separate node
```
### Subgraphs called as functions
### Using with subgraphs called as functions
When invoking a subgraph [as a function](../../how-tos/subgraph.ipynb#different-state-schemas), the **parent graph** will resume execution from the **beginning of the node** where the subgraph was invoked (and where an `interrupt` was triggered). Similarly, the **subgraph**, will resume from the **beginning of the node** where the `interrupt()` function was called.
For example,
When invoking a subgraph as a function, the parent graph will resume execution from the **beginning of the node** where the subgraph was invoked where the `interrupt` was triggered. Similarly, the **subgraph** will resume from the **beginning of the node** where the `interrupt()` function was called.
```python
def node_in_parent_graph(state: State):
@@ -772,11 +881,9 @@ def node_in_parent_graph(state: State):
{'parent_node': {'state_counter': 1}}
```
### Using multiple interrupts
Using multiple interrupts within a **single** node can be helpful for patterns like [validating human input](#validating-human-input). However, using multiple interrupts in the same node can lead to unexpected behavior if not handled carefully.
Using multiple interrupts within a **single** node can be helpful for patterns like [validating human input](../how-tos/human_in_the_loop/add-human-in-the-loop.md#validate-human-input). However, using multiple interrupts in the same node can lead to unexpected behavior if not handled carefully.
When a node contains multiple interrupt calls, LangGraph keeps a list of resume values specific to the task executing the node. Whenever execution resumes, it starts at the beginning of the node. For each interrupt encountered, LangGraph checks if a matching value exists in the task's resume list. Matching is **strictly index-based**, so the order of interrupt calls within the node is critical.
@@ -845,4 +952,5 @@ To avoid issues, refrain from dynamically changing the node's structure between
{'__interrupt__': (Interrupt(value='what is your name?', resumable=True, ns=['human_node:3a007ef9-c30d-c357-1ec1-86a1a70d8fba'], when='during'),)}
Name: N/A. Age: John
{'human_node': {'age': 'John', 'name': 'N/A'}}
```
```
docs/docs/how-tos/human_in_the_loop/review-tool-calls.ipynb
-773
View File
File diff suppressed because one or more lines are too long
docs/docs/how-tos/memory/add-memory.md
+1 -1
View File
@@ -1343,7 +1343,7 @@ def delete_messages(state):
The problem with trimming or removing messages, as shown above, is that you may lose information from culling of the message queue. Because of this, some applications benefit from a more sophisticated approach of summarizing the message history using a chat model.
![](img/memory/summary.png)
![](../../concepts/img/memory/summary.png)
=== "In an agent"
docs/docs/how-tos/review-tool-calls-functional.ipynb
-627
View File
@@ -1,627 +0,0 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# How to review tool calls (Functional API)\n",
"\n",
"!!! info \"Prerequisites\"\n",
" This guide assumes familiarity with the following:\n",
"\n",
" - Implementing [human-in-the-loop](../../concepts/human_in_the_loop) workflows with [interrupt](../../concepts/human_in_the_loop/#interrupt)\n",
" - [How to create a ReAct agent using the Functional API](../../how-tos/react-agent-from-scratch-functional)\n",
"\n",
"This guide demonstrates how to implement human-in-the-loop workflows in a ReAct agent using the LangGraph [Functional API](../../concepts/functional_api).\n",
"\n",
"We will build off of the agent created in the [How to create a ReAct agent using the Functional API](../../how-tos/react-agent-from-scratch-functional) guide.\n",
"\n",
"Specifically, we will demonstrate how to review [tool calls](https://python.langchain.com/docs/concepts/tool_calling/) generated by a [chat model](https://python.langchain.com/docs/concepts/chat_models/) prior to their execution. This can be accomplished through use of the [interrupt](../../concepts/human_in_the_loop/#interrupt) function at key points in our application.\n",
"\n",
"**Preview**:\n",
"\n",
"We will implement a simple function that reviews tool calls generated from our chat model and call it from inside our application's [entrypoint](../../concepts/functional_api/#entrypoint):\n",
"\n",
"```python\n",
"def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:\n",
" \"\"\"Review a tool call, returning a validated version.\"\"\"\n",
" human_review = interrupt(\n",
" {\n",
" \"question\": \"Is this correct?\",\n",
" \"tool_call\": tool_call,\n",
" }\n",
" )\n",
" review_action = human_review[\"action\"]\n",
" review_data = human_review.get(\"data\")\n",
" if review_action == \"continue\":\n",
" return tool_call\n",
" elif review_action == \"update\":\n",
" updated_tool_call = {**tool_call, **{\"args\": review_data}}\n",
" return updated_tool_call\n",
" elif review_action == \"feedback\":\n",
" return ToolMessage(\n",
" content=review_data, name=tool_call[\"name\"], tool_call_id=tool_call[\"id\"]\n",
" )\n",
"```\n",
"\n",
"## Setup\n",
"\n",
"First, let's install the required packages and set our API keys:"
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install -U langgraph langchain-openai"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"import getpass\n",
"import os\n",
"\n",
"\n",
"def _set_env(var: str):\n",
" if not os.environ.get(var):\n",
" os.environ[var] = getpass.getpass(f\"{var}: \")\n",
"\n",
"\n",
"_set_env(\"OPENAI_API_KEY\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<div class=\"admonition tip\">\n",
" <p class=\"admonition-title\">Set up <a href=\"https://smith.langchain.com\">LangSmith</a> for better debugging</p>\n",
" <p style=\"padding-top: 5px;\">\n",
" Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM aps built with LangGraph — read more about how to get started in the <a href=\"https://docs.smith.langchain.com\">docs</a>. \n",
" </p>\n",
" </div>"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Define model and tools\n",
"\n",
"Let's first define the tools and model we will use for our example. As in the [ReAct agent guide](../../how-tos/react-agent-from-scratch-functional), we will use a single place-holder tool that gets a description of the weather for a location.\n",
"\n",
"We will use an [OpenAI](https://python.langchain.com/docs/integrations/providers/openai/) chat model for this example, but any model [supporting tool-calling](https://python.langchain.com/docs/integrations/chat/) will suffice."
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.tools import tool\n",
"\n",
"model = ChatOpenAI(model=\"gpt-4o-mini\")\n",
"\n",
"\n",
"@tool\n",
"def get_weather(location: str):\n",
" \"\"\"Call to get the weather from a specific location.\"\"\"\n",
" # This is a placeholder for the actual implementation\n",
" if any([city in location.lower() for city in [\"sf\", \"san francisco\"]]):\n",
" return \"It's sunny!\"\n",
" elif \"boston\" in location.lower():\n",
" return \"It's rainy!\"\n",
" else:\n",
" return f\"I am not sure what the weather is in {location}\"\n",
"\n",
"\n",
"tools = [get_weather]"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Define tasks\n",
"\n",
"Our [tasks](../../concepts/functional_api/#task) are unchanged from the [ReAct agent guide](../../how-tos/react-agent-from-scratch-functional):\n",
"\n",
"1. **Call model**: We want to query our chat model with a list of messages.\n",
"2. **Call tool**: If our model generates tool calls, we want to execute them."
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"from langchain_core.messages import ToolCall, ToolMessage\n",
"from langgraph.func import entrypoint, task\n",
"\n",
"\n",
"tools_by_name = {tool.name: tool for tool in tools}\n",
"\n",
"\n",
"@task\n",
"def call_model(messages):\n",
" \"\"\"Call model with a sequence of messages.\"\"\"\n",
" response = model.bind_tools(tools).invoke(messages)\n",
" return response\n",
"\n",
"\n",
"@task\n",
"def call_tool(tool_call):\n",
" tool = tools_by_name[tool_call[\"name\"]]\n",
" observation = tool.invoke(tool_call[\"args\"])\n",
" return ToolMessage(content=observation, tool_call_id=tool_call[\"id\"])"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Define entrypoint\n",
"\n",
"To review tool calls before execution, we add a `review_tool_call` function that calls [interrupt](../../concepts/human_in_the_loop/#interrupt). When this function is called, execution will be paused until we issue a command to resume it.\n",
"\n",
"Given a tool call, our function will `interrupt` for human review. At that point we can either:\n",
"\n",
"- Accept the tool call;\n",
"- Revise the tool call and continue;\n",
"- Generate a custom tool message (e.g., instructing the model to re-format its tool call).\n",
"\n",
"We will demonstrate these three cases in the [usage examples](#usage) below."
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [],
"source": [
"from typing import Union\n",
"\n",
"\n",
"def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:\n",
" \"\"\"Review a tool call, returning a validated version.\"\"\"\n",
" human_review = interrupt(\n",
" {\n",
" \"question\": \"Is this correct?\",\n",
" \"tool_call\": tool_call,\n",
" }\n",
" )\n",
" review_action = human_review[\"action\"]\n",
" review_data = human_review.get(\"data\")\n",
" if review_action == \"continue\":\n",
" return tool_call\n",
" elif review_action == \"update\":\n",
" updated_tool_call = {**tool_call, **{\"args\": review_data}}\n",
" return updated_tool_call\n",
" elif review_action == \"feedback\":\n",
" return ToolMessage(\n",
" content=review_data, name=tool_call[\"name\"], tool_call_id=tool_call[\"id\"]\n",
" )"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can now update our [entrypoint](../../concepts/functional_api/#entrypoint) to review the generated tool calls. If a tool call is accepted or revised, we execute in the same way as before. Otherwise, we just append the `ToolMessage` supplied by the human.\n",
"\n",
"!!! tip\n",
"\n",
" The results of prior tasks — in this case the initial model call — are persisted, so that they are not run again following the `interrupt`."
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"from langgraph.checkpoint.memory import MemorySaver\n",
"from langgraph.graph.message import add_messages\n",
"from langgraph.types import Command, interrupt\n",
"\n",
"\n",
"checkpointer = MemorySaver()\n",
"\n",
"\n",
"@entrypoint(checkpointer=checkpointer)\n",
"def agent(messages, previous):\n",
" if previous is not None:\n",
" messages = add_messages(previous, messages)\n",
"\n",
" llm_response = call_model(messages).result()\n",
" while True:\n",
" if not llm_response.tool_calls:\n",
" break\n",
"\n",
" # Review tool calls\n",
" tool_results = []\n",
" tool_calls = []\n",
" for i, tool_call in enumerate(llm_response.tool_calls):\n",
" review = review_tool_call(tool_call)\n",
" if isinstance(review, ToolMessage):\n",
" tool_results.append(review)\n",
" else: # is a validated tool call\n",
" tool_calls.append(review)\n",
" if review != tool_call:\n",
" llm_response.tool_calls[i] = review # update message\n",
"\n",
" # Execute remaining tool calls\n",
" tool_result_futures = [call_tool(tool_call) for tool_call in tool_calls]\n",
" remaining_tool_results = [fut.result() for fut in tool_result_futures]\n",
"\n",
" # Append to message list\n",
" messages = add_messages(\n",
" messages,\n",
" [llm_response, *tool_results, *remaining_tool_results],\n",
" )\n",
"\n",
" # Call model again\n",
" llm_response = call_model(messages).result()\n",
"\n",
" # Generate final response\n",
" messages = add_messages(messages, llm_response)\n",
" return entrypoint.final(value=llm_response, save=messages)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Usage\n",
"\n",
"Let's demonstrate some scenarios."
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [],
"source": [
"def _print_step(step: dict) -> None:\n",
" for task_name, result in step.items():\n",
" if task_name == \"agent\":\n",
" continue # just stream from tasks\n",
" print(f\"\\n{task_name}:\")\n",
" if task_name in (\"__interrupt__\", \"review_tool_call\"):\n",
" print(result)\n",
" else:\n",
" result.pretty_print()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Accept a tool call\n",
"\n",
"To accept a tool call, we just indicate in the data we provide in the `Command` that the tool call should pass through."
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [],
"source": [
"config = {\"configurable\": {\"thread_id\": \"1\"}}"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'role': 'user', 'content': \"What's the weather in san francisco?\"}\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"Tool Calls:\n",
" get_weather (call_Bh5cSwMqCpCxTjx7AjdrQTPd)\n",
" Call ID: call_Bh5cSwMqCpCxTjx7AjdrQTPd\n",
" Args:\n",
" location: San Francisco\n",
"\n",
"__interrupt__:\n",
"(Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'get_weather', 'args': {'location': 'San Francisco'}, 'id': 'call_Bh5cSwMqCpCxTjx7AjdrQTPd', 'type': 'tool_call'}}, resumable=True, ns=['agent:22fcc9cd-3573-b39b-eea7-272a025903e2'], when='during'),)\n"
]
}
],
"source": [
"user_message = {\"role\": \"user\", \"content\": \"What's the weather in san francisco?\"}\n",
"print(user_message)\n",
"\n",
"for step in agent.stream([user_message], config):\n",
" _print_step(step)"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\n",
"call_tool:\n",
"=================================\u001b[1m Tool Message \u001b[0m=================================\n",
"\n",
"It's sunny!\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"\n",
"The weather in San Francisco is sunny!\n"
]
}
],
"source": [
"# highlight-next-line\n",
"human_input = Command(resume={\"action\": \"continue\"})\n",
"\n",
"for step in agent.stream(human_input, config):\n",
" _print_step(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Revise a tool call\n",
"\n",
"To revise a tool call, we can supply updated arguments."
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [],
"source": [
"config = {\"configurable\": {\"thread_id\": \"2\"}}"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'role': 'user', 'content': \"What's the weather in san francisco?\"}\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"Tool Calls:\n",
" get_weather (call_b9h8e18FqH0IQm3NMoeYKz6N)\n",
" Call ID: call_b9h8e18FqH0IQm3NMoeYKz6N\n",
" Args:\n",
" location: san francisco\n",
"\n",
"__interrupt__:\n",
"(Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'get_weather', 'args': {'location': 'san francisco'}, 'id': 'call_b9h8e18FqH0IQm3NMoeYKz6N', 'type': 'tool_call'}}, resumable=True, ns=['agent:9559a81d-5720-dc19-a457-457bac7bdd83'], when='during'),)\n"
]
}
],
"source": [
"user_message = {\"role\": \"user\", \"content\": \"What's the weather in san francisco?\"}\n",
"print(user_message)\n",
"\n",
"for step in agent.stream([user_message], config):\n",
" _print_step(step)"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\n",
"call_tool:\n",
"=================================\u001b[1m Tool Message \u001b[0m=================================\n",
"\n",
"It's sunny!\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"\n",
"The weather in San Francisco is sunny!\n"
]
}
],
"source": [
"# highlight-next-line\n",
"human_input = Command(resume={\"action\": \"update\", \"data\": {\"location\": \"SF, CA\"}})\n",
"\n",
"for step in agent.stream(human_input, config):\n",
" _print_step(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"The LangSmith traces for this run are particularly informative:\n",
"\n",
"- In the trace [before the interrupt](https://smith.langchain.com/public/c8b07579-5cf4-4adb-a849-282163bc9d99/r/b5b128d6-e715-480b-b58d-59e64f724275), we generate a tool call for location `\"San Francisco\"`.\n",
"- In the trace [after resuming](https://smith.langchain.com/public/b28b92e5-a555-482d-aa4d-c675a19f0eb5/r), we see that the tool call in the message has been updated to `\"SF, CA\"`."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Generate a custom ToolMessage\n",
"\n",
"To Generate a custom `ToolMessage`, we supply the content of the message. In this case we will ask the model to reformat its tool call."
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"outputs": [],
"source": [
"config = {\"configurable\": {\"thread_id\": \"3\"}}"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'role': 'user', 'content': \"What's the weather in san francisco?\"}\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"Tool Calls:\n",
" get_weather (call_VqGjKE7uu8HdWs9XuY1kMV18)\n",
" Call ID: call_VqGjKE7uu8HdWs9XuY1kMV18\n",
" Args:\n",
" location: San Francisco\n",
"\n",
"__interrupt__:\n",
"(Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'get_weather', 'args': {'location': 'San Francisco'}, 'id': 'call_VqGjKE7uu8HdWs9XuY1kMV18', 'type': 'tool_call'}}, resumable=True, ns=['agent:4b3b372b-9da3-70be-5c68-3d9317346070'], when='during'),)\n"
]
}
],
"source": [
"user_message = {\"role\": \"user\", \"content\": \"What's the weather in san francisco?\"}\n",
"print(user_message)\n",
"\n",
"for step in agent.stream([user_message], config):\n",
" _print_step(step)"
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"Tool Calls:\n",
" get_weather (call_xoXkK8Cz0zIpvWs78qnXpvYp)\n",
" Call ID: call_xoXkK8Cz0zIpvWs78qnXpvYp\n",
" Args:\n",
" location: San Francisco, CA\n",
"\n",
"__interrupt__:\n",
"(Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'get_weather', 'args': {'location': 'San Francisco, CA'}, 'id': 'call_xoXkK8Cz0zIpvWs78qnXpvYp', 'type': 'tool_call'}}, resumable=True, ns=['agent:4b3b372b-9da3-70be-5c68-3d9317346070'], when='during'),)\n"
]
}
],
"source": [
"# highlight-next-line\n",
"human_input = Command(\n",
" # highlight-next-line\n",
" resume={\n",
" # highlight-next-line\n",
" \"action\": \"feedback\",\n",
" # highlight-next-line\n",
" \"data\": \"Please format as <City>, <State>.\",\n",
" # highlight-next-line\n",
" },\n",
" # highlight-next-line\n",
")\n",
"\n",
"for step in agent.stream(human_input, config):\n",
" _print_step(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Once it is re-formatted, we can accept it:"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\n",
"call_tool:\n",
"=================================\u001b[1m Tool Message \u001b[0m=================================\n",
"\n",
"It's sunny!\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"\n",
"The weather in San Francisco, CA is sunny!\n"
]
}
],
"source": [
"# highlight-next-line\n",
"human_input = Command(resume={\"action\": \"continue\"})\n",
"\n",
"for step in agent.stream(human_input, config):\n",
" _print_step(step)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.3"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
docs/docs/how-tos/use-functional-api.md
+154 -3
View File
@@ -462,10 +462,161 @@ main.invoke(None, config=config)
The functional API supports [human-in-the-loop](../concepts/human_in_the_loop.md) workflows using the `interrupt` function and the `Command` primitive.
Please see the following examples for more details:
### Basic human-in-the-loop workflow
* [How to wait for user input (Functional API)](./wait-user-input-functional.ipynb): Shows how to implement a simple human-in-the-loop workflow using the functional API.
* [How to review tool calls (Functional API)](./review-tool-calls-functional.ipynb): Guide demonstrates how to implement human-in-the-loop workflows in a ReAct agent using the LangGraph Functional API.
We will create three [tasks](../concepts/functional_api.md#task):
1. Append `"bar"`.
2. Pause for human input. When resuming, append human input.
3. Append `"qux"`.
```python
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt
@task
def step_1(input_query):
"""Append bar."""
return f"{input_query} bar"
@task
def human_feedback(input_query):
"""Append user input."""
feedback = interrupt(f"Please provide feedback: {input_query}")
return f"{input_query} {feedback}"
@task
def step_3(input_query):
"""Append qux."""
return f"{input_query} qux"
```
We can now compose these tasks in an [entrypoint](../concepts/functional_api.md#entrypoint):
```python
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
@entrypoint(checkpointer=checkpointer)
def graph(input_query):
result_1 = step_1(input_query).result()
result_2 = human_feedback(result_1).result()
result_3 = step_3(result_2).result()
return result_3
```
[interrupt()](../how-tos/human_in_the_loop/add-human-in-the-loop.md#pause-using-interrupt) is called inside a task, enabling a human to review and edit the output of the previous task. The results of prior tasks-- in this case `step_1`-- are persisted, so that they are not run again following the `interrupt`.
Let's send in a query string:
```python
config = {"configurable": {"thread_id": "1"}}
for event in graph.stream("foo", config):
print(event)
print("\n")
```
Note that we've paused with an `interrupt` after `step_1`. The interrupt provides instructions to resume the run. To resume, we issue a [Command](../how-tos/human_in_the_loop/add-human-in-the-loop.md#resume-using-the-command-primitive) containing the data expected by the `human_feedback` task.
```python
# Continue execution
for event in graph.stream(Command(resume="baz"), config):
print(event)
print("\n")
```
After resuming, the run proceeds through the remaining step and terminates as expected.
### Review tool calls
To review tool calls before execution, we add a `review_tool_call` function that calls [`interrupt`](../how-tos/human_in_the_loop/add-human-in-the-loop.md#pause-using-interrupt). When this function is called, execution will be paused until we issue a command to resume it.
Given a tool call, our function will `interrupt` for human review. At that point we can either:
- Accept the tool call
- Revise the tool call and continue
- Generate a custom tool message (e.g., instructing the model to re-format its tool call)
```python
from typing import Union
def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:
"""Review a tool call, returning a validated version."""
human_review = interrupt(
{
"question": "Is this correct?",
"tool_call": tool_call,
}
)
review_action = human_review["action"]
review_data = human_review.get("data")
if review_action == "continue":
return tool_call
elif review_action == "update":
updated_tool_call = {**tool_call, **{"args": review_data}}
return updated_tool_call
elif review_action == "feedback":
return ToolMessage(
content=review_data, name=tool_call["name"], tool_call_id=tool_call["id"]
)
```
We can now update our [entrypoint](../concepts/functional_api.md#entrypoint) to review the generated tool calls. If a tool call is accepted or revised, we execute in the same way as before. Otherwise, we just append the `ToolMessage` supplied by the human. The results of prior tasks — in this case the initial model call — are persisted, so that they are not run again following the `interrupt`.
```python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt
checkpointer = MemorySaver()
@entrypoint(checkpointer=checkpointer)
def agent(messages, previous):
if previous is not None:
messages = add_messages(previous, messages)
llm_response = call_model(messages).result()
while True:
if not llm_response.tool_calls:
break
# Review tool calls
tool_results = []
tool_calls = []
for i, tool_call in enumerate(llm_response.tool_calls):
review = review_tool_call(tool_call)
if isinstance(review, ToolMessage):
tool_results.append(review)
else: # is a validated tool call
tool_calls.append(review)
if review != tool_call:
llm_response.tool_calls[i] = review # update message
# Execute remaining tool calls
tool_result_futures = [call_tool(tool_call) for tool_call in tool_calls]
remaining_tool_results = [fut.result() for fut in tool_result_futures]
# Append to message list
messages = add_messages(
messages,
[llm_response, *tool_results, *remaining_tool_results],
)
# Call model again
llm_response = call_model(messages).result()
# Generate final response
messages = add_messages(messages, llm_response)
return entrypoint.final(value=llm_response, save=messages)
```
## Short-term memory
docs/docs/how-tos/wait-user-input-functional.ipynb
-561
View File
@@ -1,561 +0,0 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# How to wait for user input (Functional API)\n",
"\n",
"!!! info \"Prerequisites\"\n",
" This guide assumes familiarity with the following:\n",
"\n",
" - Implementing [human-in-the-loop](../../concepts/human_in_the_loop) workflows with [interrupt](../../concepts/human_in_the_loop/#interrupt)\n",
" - [How to create a ReAct agent using the Functional API](../../how-tos/react-agent-from-scratch-functional)\n",
"\n",
"**Human-in-the-loop (HIL)** interactions are crucial for [agentic systems](../../concepts/agentic_concepts/#human-in-the-loop). Waiting for human input is a common HIL interaction pattern, allowing the agent to ask the user clarifying questions and await input before proceeding. \n",
"\n",
"We can implement this in LangGraph using the [interrupt()][langgraph.types.interrupt] function. `interrupt` allows us to stop graph execution to collect input from a user and continue execution with collected input.\n",
"\n",
"This guide demonstrates how to implement human-in-the-loop workflows using LangGraph's [Functional API](../../concepts/functional_api). Specifically, we will demonstrate:\n",
"\n",
"1. [A simple usage example](#simple-usage)\n",
"2. [How to use with a ReAct agent](#agent)\n",
"\n",
"## Setup\n",
"\n",
"First, let's install the required packages and set our API keys:"
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install -U langgraph langchain-openai"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"import getpass\n",
"import os\n",
"\n",
"\n",
"def _set_env(var: str):\n",
" if not os.environ.get(var):\n",
" os.environ[var] = getpass.getpass(f\"{var}: \")\n",
"\n",
"\n",
"_set_env(\"OPENAI_API_KEY\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<div class=\"admonition tip\">\n",
" <p class=\"admonition-title\">Set up <a href=\"https://smith.langchain.com\">LangSmith</a> for better debugging</p>\n",
" <p style=\"padding-top: 5px;\">\n",
" Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM aps built with LangGraph — read more about how to get started in the <a href=\"https://docs.smith.langchain.com\">docs</a>. \n",
" </p>\n",
" </div>"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Simple usage\n",
"\n",
"Let's demonstrate a simple usage example. We will create three [tasks](../../concepts/functional_api/#task):\n",
"\n",
"1. Append `\"bar\"`.\n",
"2. Pause for human input. When resuming, append human input.\n",
"3. Append `\"qux\"`."
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"from langgraph.func import entrypoint, task\n",
"from langgraph.types import Command, interrupt\n",
"\n",
"\n",
"@task\n",
"def step_1(input_query):\n",
" \"\"\"Append bar.\"\"\"\n",
" return f\"{input_query} bar\"\n",
"\n",
"\n",
"@task\n",
"def human_feedback(input_query):\n",
" \"\"\"Append user input.\"\"\"\n",
" feedback = interrupt(f\"Please provide feedback: {input_query}\")\n",
" return f\"{input_query} {feedback}\"\n",
"\n",
"\n",
"@task\n",
"def step_3(input_query):\n",
" \"\"\"Append qux.\"\"\"\n",
" return f\"{input_query} qux\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can now compose these tasks in a simple [entrypoint](../../concepts/functional_api/#entrypoint):"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [],
"source": [
"from langgraph.checkpoint.memory import MemorySaver\n",
"\n",
"checkpointer = MemorySaver()\n",
"\n",
"\n",
"@entrypoint(checkpointer=checkpointer)\n",
"def graph(input_query):\n",
" result_1 = step_1(input_query).result()\n",
" result_2 = human_feedback(result_1).result()\n",
" result_3 = step_3(result_2).result()\n",
"\n",
" return result_3"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"All we have done to enable human-in-the-loop workflows is called [interrupt()](../../concepts/human_in_the_loop/#interrupt) inside a task.\n",
"\n",
"!!! tip\n",
"\n",
" The results of prior tasks-- in this case `step_1`-- are persisted, so that they are not run again following the `interrupt`.\n",
"\n",
"\n",
"Let's send in a query string:"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"config = {\"configurable\": {\"thread_id\": \"1\"}}"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'step_1': 'foo bar'}\n",
"\n",
"\n",
"{'__interrupt__': (Interrupt(value='Please provide feedback: foo bar', resumable=True, ns=['graph:d66b2e35-0ee3-d8d6-1a22-aec9d58f13b9', 'human_feedback:e0cd4ee2-b874-e1d2-8bc4-3f7ddc06bcc2'], when='during'),)}\n",
"\n",
"\n"
]
}
],
"source": [
"for event in graph.stream(\"foo\", config):\n",
" print(event)\n",
" print(\"\\n\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Note that we've paused with an `interrupt` after `step_1`. The interrupt provides instructions to resume the run. To resume, we issue a [Command](../../concepts/human_in_the_loop/#the-command-primitive) containing the data expected by the `human_feedback` task."
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'human_feedback': 'foo bar baz'}\n",
"\n",
"\n",
"{'step_3': 'foo bar baz qux'}\n",
"\n",
"\n",
"{'graph': 'foo bar baz qux'}\n",
"\n",
"\n"
]
}
],
"source": [
"# Continue execution\n",
"for event in graph.stream(Command(resume=\"baz\"), config):\n",
" print(event)\n",
" print(\"\\n\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"After resuming, the run proceeds through the remaining step and terminates as expected."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Agent\n",
"\n",
"We will build off of the agent created in the [How to create a ReAct agent using the Functional API](../../how-tos/react-agent-from-scratch-functional) guide.\n",
"\n",
"Here we will extend the agent by allowing it to reach out to a human for assistance when needed.\n",
"\n",
"### Define model and tools\n",
"\n",
"Let's first define the tools and model we will use for our example. As in the [ReAct agent guide](../../how-tos/react-agent-from-scratch-functional), we will use a single place-holder tool that gets a description of the weather for a location.\n",
"\n",
"We will use an [OpenAI](https://python.langchain.com/docs/integrations/providers/openai/) chat model for this example, but any model [supporting tool-calling](https://python.langchain.com/docs/integrations/chat/) will suffice."
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.tools import tool\n",
"\n",
"model = ChatOpenAI(model=\"gpt-4o-mini\")\n",
"\n",
"\n",
"@tool\n",
"def get_weather(location: str):\n",
" \"\"\"Call to get the weather from a specific location.\"\"\"\n",
" # This is a placeholder for the actual implementation\n",
" if any([city in location.lower() for city in [\"sf\", \"san francisco\"]]):\n",
" return \"It's sunny!\"\n",
" elif \"boston\" in location.lower():\n",
" return \"It's rainy!\"\n",
" else:\n",
" return f\"I am not sure what the weather is in {location}\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"To reach out to a human for assistance, we can simply add a tool that calls [interrupt](../../concepts/human_in_the_loop/#interrupt):"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [],
"source": [
"from langgraph.types import Command, interrupt\n",
"\n",
"\n",
"@tool\n",
"def human_assistance(query: str) -> str:\n",
" \"\"\"Request assistance from a human.\"\"\"\n",
" human_response = interrupt({\"query\": query})\n",
" return human_response[\"data\"]\n",
"\n",
"\n",
"tools = [get_weather, human_assistance]"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Define tasks\n",
"\n",
"Our tasks are otherwise unchanged from the [ReAct agent guide](../../how-tos/react-agent-from-scratch-functional):\n",
"\n",
"1. **Call model**: We want to query our chat model with a list of messages.\n",
"2. **Call tool**: If our model generates tool calls, we want to execute them.\n",
"\n",
"We just have one more tool accessible to the model."
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [],
"source": [
"from langchain_core.messages import ToolMessage\n",
"from langgraph.func import entrypoint, task\n",
"\n",
"tools_by_name = {tool.name: tool for tool in tools}\n",
"\n",
"\n",
"@task\n",
"def call_model(messages):\n",
" \"\"\"Call model with a sequence of messages.\"\"\"\n",
" response = model.bind_tools(tools).invoke(messages)\n",
" return response\n",
"\n",
"\n",
"@task\n",
"def call_tool(tool_call):\n",
" tool = tools_by_name[tool_call[\"name\"]]\n",
" observation = tool.invoke(tool_call)\n",
" return ToolMessage(content=observation, tool_call_id=tool_call[\"id\"])"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Define entrypoint\n",
"\n",
"Our [entrypoint](../../concepts/functional_api/#entrypoint) is also unchanged from the [ReAct agent guide](../../how-tos/react-agent-from-scratch-functional):"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {},
"outputs": [],
"source": [
"from langgraph.checkpoint.memory import MemorySaver\n",
"from langgraph.graph.message import add_messages\n",
"\n",
"checkpointer = MemorySaver()\n",
"\n",
"\n",
"@entrypoint(checkpointer=checkpointer)\n",
"def agent(messages, previous):\n",
" if previous is not None:\n",
" messages = add_messages(previous, messages)\n",
"\n",
" llm_response = call_model(messages).result()\n",
" while True:\n",
" if not llm_response.tool_calls:\n",
" break\n",
"\n",
" # Execute tools\n",
" tool_result_futures = [\n",
" call_tool(tool_call) for tool_call in llm_response.tool_calls\n",
" ]\n",
" tool_results = [fut.result() for fut in tool_result_futures]\n",
"\n",
" # Append to message list\n",
" messages = add_messages(messages, [llm_response, *tool_results])\n",
"\n",
" # Call model again\n",
" llm_response = call_model(messages).result()\n",
"\n",
" # Generate final response\n",
" messages = add_messages(messages, llm_response)\n",
" return entrypoint.final(value=llm_response, save=messages)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Usage\n",
"\n",
"Let's invoke our model with a question that requires human assistance. Our question will also require an invocation of the `get_weather` tool:"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"def _print_step(step: dict) -> None:\n",
" for task_name, result in step.items():\n",
" if task_name == \"agent\":\n",
" continue # just stream from tasks\n",
" print(f\"\\n{task_name}:\")\n",
" if task_name == \"__interrupt__\":\n",
" print(result)\n",
" else:\n",
" result.pretty_print()"
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"outputs": [],
"source": [
"config = {\"configurable\": {\"thread_id\": \"1\"}}"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'role': 'user', 'content': 'Can you reach out for human assistance: what should I feed my cat? Separately, can you check the weather in San Francisco?'}\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"Tool Calls:\n",
" human_assistance (call_joAEBVX7Abfm7TsZ0k95ZkVx)\n",
" Call ID: call_joAEBVX7Abfm7TsZ0k95ZkVx\n",
" Args:\n",
" query: What should I feed my cat?\n",
" get_weather (call_ut7zfHFCcms63BOZLrRHszGH)\n",
" Call ID: call_ut7zfHFCcms63BOZLrRHszGH\n",
" Args:\n",
" location: San Francisco\n",
"\n",
"call_tool:\n",
"=================================\u001b[1m Tool Message \u001b[0m=================================\n",
"\n",
"content=\"It's sunny!\" name='get_weather' tool_call_id='call_ut7zfHFCcms63BOZLrRHszGH'\n",
"\n",
"__interrupt__:\n",
"(Interrupt(value={'query': 'What should I feed my cat?'}, resumable=True, ns=['agent:aa676ccc-b038-25e3-9c8a-18e81d4e1372', 'call_tool:059d53d2-3344-13bc-e170-48b632c2dd97'], when='during'),)\n"
]
}
],
"source": [
"user_message = {\n",
" \"role\": \"user\",\n",
" \"content\": (\n",
" \"Can you reach out for human assistance: what should I feed my cat? \"\n",
" \"Separately, can you check the weather in San Francisco?\"\n",
" ),\n",
"}\n",
"print(user_message)\n",
"\n",
"for step in agent.stream([user_message], config):\n",
" _print_step(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Note that we generate two tool calls, and although our run is interrupted, we did not block the execution of the `get_weather` tool.\n",
"\n",
"Let's inspect where we're interrupted:"
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'__interrupt__': (Interrupt(value={'query': 'What should I feed my cat?'}, resumable=True, ns=['agent:aa676ccc-b038-25e3-9c8a-18e81d4e1372', 'call_tool:059d53d2-3344-13bc-e170-48b632c2dd97'], when='during'),)}\n"
]
}
],
"source": [
"print(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can resume execution by issuing a [Command](../../concepts/human_in_the_loop/#the-command-primitive). Note that the data we supply in the `Command` can be customized to your needs based on the implementation of `human_assistance`."
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\n",
"call_tool:\n",
"=================================\u001b[1m Tool Message \u001b[0m=================================\n",
"\n",
"content='You should feed your cat a fish.' name='human_assistance' tool_call_id='call_joAEBVX7Abfm7TsZ0k95ZkVx'\n",
"\n",
"call_model:\n",
"==================================\u001b[1m Ai Message \u001b[0m==================================\n",
"\n",
"For human assistance, you should feed your cat fish. \n",
"\n",
"Regarding the weather in San Francisco, it's sunny!\n"
]
}
],
"source": [
"human_response = \"You should feed your cat a fish.\"\n",
"human_command = Command(resume={\"data\": human_response})\n",
"\n",
"for step in agent.stream(human_command, config):\n",
" _print_step(step)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Above, when we resume we provide the final tool message, allowing the model to generate its response. Check out the LangSmith traces to see a full breakdown of the runs:\n",
"\n",
"1. [Trace from initial query](https://smith.langchain.com/public/c3d8879d-4d01-41be-807e-6d9eed15df99/r)\n",
"2. [Trace after resuming](https://smith.langchain.com/public/97c05ef9-8b4c-428e-8826-3fd417c8c75f/r)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.3"
}
},
"nbformat": 4,
"nbformat_minor": 4
}