如何审查工具调用¶
人机交互（HIL）对于代理系统至关重要。一个常见的模式是在某些工具调用后添加一个步骤，让人类参与其中。这些工具调用通常会导致函数调用或保存某些信息。例如：
- 调用一个执行SQL的工具，然后由工具执行
- 调用一个生成摘要的工具，然后将摘要保存到图的状态中
请注意，使用工具调用是否实际调用工具都是常见的做法。
在这里，您通常可能希望进行以下几种不同的交互：
- 批准工具调用并继续执行
- 手动修改工具调用然后继续执行
- 给出自然语言反馈，然后将反馈传递给代理
我们可以在LangGraph中使用interrupt()函数来实现这些功能。interrupt允许我们在收集用户输入后停止图的执行，并继续执行收集到的输入：
def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
# 这是我们将通过Command(resume=<human_review>)提供的值
human_review = interrupt(
{
"question": "这是正确的吗？",
# 提供工具调用以供审查
"tool_call": tool_call
}
)
review_action, review_data = human_review
# 批准工具调用并继续执行
if review_action == "continue":
return Command(goto="run_tool")
# 手动修改工具调用然后继续执行
elif review_action == "update":
...
updated_msg = get_updated_msg(review_data)
return Command(goto="run_tool", update={"messages": [updated_message]})
# 给出自然语言反馈，然后将反馈传递给代理
elif review_action == "feedback":
...
feedback_msg = get_feedback_msg(review_data)
return Command(goto="call_llm", update={"messages": [feedback_msg]})
环境搭建¶
首先我们需要安装所需的包
接下来，我们需要为Anthropic（我们将使用的大型语言模型）设置API密钥
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")
为LangGraph开发设置LangSmith
注册LangSmith，可以快速发现并解决您的LangGraph项目中的问题，提高性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅此处。
简单用法¶
让我们设置一个非常简单的图来实现这一点。 首先，我们将有一个LLM调用，它决定采取什么行动。 然后我们进入一个人类节点。这个节点实际上不做任何事情——想法是在到达这个节点之前进行中断，然后对状态进行任何更新。 之后，我们检查状态，并根据需要将流程路由回LLM或正确的工具。
让我们看看这个过程的实际操作！
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import AIMessage
from IPython.display import Image, display
@tool
def weather_search(city: str):
"""Search for the weather"""
print("----")
print(f"Searching for: {city}")
print("----")
return "Sunny!"
model = ChatAnthropic(model_name="claude-3-5-sonnet-latest").bind_tools(
[weather_search]
)
class State(MessagesState):
"""Simple state."""
def call_llm(state):
return {"messages": [model.invoke(state["messages"])]}
def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
last_message = state["messages"][-1]
tool_call = last_message.tool_calls[-1]
# 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,
}
)
review_action = human_review["action"]
review_data = human_review.get("data")
# if approved, call the tool
if review_action == "continue":
return Command(goto="run_tool")
# update the AI message AND call tools
elif review_action == "update":
updated_message = {
"role": "ai",
"content": last_message.content,
"tool_calls": [
{
"id": tool_call["id"],
"name": tool_call["name"],
# This the update provided by the human
"args": review_data,
}
],
# This is important - this needs to be the same as the message you replacing!
# Otherwise, it will show up as a separate message
"id": last_message.id,
}
return Command(goto="run_tool", update={"messages": [updated_message]})
# provide feedback to LLM
elif review_action == "feedback":
# NOTE: we're adding feedback message as a ToolMessage
# to preserve the correct order in the message history
# (AI messages with tool calls need to be followed by tool call messages)
tool_message = {
"role": "tool",
# This is our natural language feedback
"content": review_data,
"name": tool_call["name"],
"tool_call_id": tool_call["id"],
}
return Command(goto="call_llm", update={"messages": [tool_message]})
def run_tool(state):
new_messages = []
tools = {"weather_search": weather_search}
tool_calls = state["messages"][-1].tool_calls
for tool_call in tool_calls:
tool = tools[tool_call["name"]]
result = tool.invoke(tool_call["args"])
new_messages.append(
{
"role": "tool",
"name": tool_call["name"],
"content": result,
"tool_call_id": tool_call["id"],
}
)
return {"messages": new_messages}
def route_after_llm(state) -> Literal[END, "human_review_node"]:
if len(state["messages"][-1].tool_calls) == 0:
return END
else:
return "human_review_node"
builder = StateGraph(State)
builder.add_node(call_llm)
builder.add_node(run_tool)
builder.add_node(human_review_node)
builder.add_edge(START, "call_llm")
builder.add_conditional_edges("call_llm", route_after_llm)
builder.add_edge("run_tool", "call_llm")
# Set up memory
memory = MemorySaver()
# Add
graph = builder.compile(checkpointer=memory)
# View
display(Image(graph.get_graph().draw_mermaid_png()))
API Reference: StateGraph | START | END | MemorySaver | Command | interrupt | ChatAnthropic | tool | AIMessage
无需审查的示例¶
让我们来看一个不需要审查的示例（因为没有调用任何工具）
# Input
initial_input = {"messages": [{"role": "user", "content": "hi!"}]}
# Thread
thread = {"configurable": {"thread_id": "1"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
print(event)
print("\n")
{'call_llm': {'messages': [AIMessage(content="Hello! I'm here to help you. I can assist you with checking the weather in different cities using the weather search tool. Would you like to know the weather for a specific city? Just let me know which city you're interested in!", additional_kwargs={}, response_metadata={'id': 'msg_01XHvA3ZWpsq4PdyiruWFLBs', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 374, 'output_tokens': 52}}, id='run-c3ff5fea-0135-4d66-8ec1-f8ed6a88356b-0', usage_metadata={'input_tokens': 374, 'output_tokens': 52, 'total_tokens': 426, 'input_token_details': {}})]}}
工具批准示例¶
现在让我们看看批准工具调用是什么样子的
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in sf?"}]}
# Thread
thread = {"configurable": {"thread_id": "2"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
print(event)
print("\n")
{'call_llm': {'messages': [AIMessage(content=[{'text': "I'll help you check the weather in San Francisco.", 'type': 'text'}, {'id': 'toolu_01Kn67GmQAA3BEF1cfYdNW3c', 'input': {'city': 'sf'}, 'name': 'weather_search', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_013eJXUAEA2ANvYLkDUQFRPo', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 379, 'output_tokens': 65}}, id='run-e8174b94-f681-4688-967f-a32295412f91-0', tool_calls=[{'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_01Kn67GmQAA3BEF1cfYdNW3c', 'type': 'tool_call'}], usage_metadata={'input_tokens': 379, 'output_tokens': 65, 'total_tokens': 444, 'input_token_details': {}})]}}
{'__interrupt__': (Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_01Kn67GmQAA3BEF1cfYdNW3c', 'type': 'tool_call'}}, resumable=True, ns=['human_review_node:be252162-5b29-0a98-1ed2-c807c1fc64c6'], when='during'),)}
human_review_node知道我们定义的human_review变量的值。我们可以通过使用Command(resume=<human_review>)输入来调用图以提供此值。由于我们正在批准工具调用，我们将提供resume值为{"action": "continue"}以导航到run_tool节点。
for event in graph.stream(
# provide value
Command(resume={"action": "continue"}),
thread,
stream_mode="updates",
):
print(event)
print("\n")
{'human_review_node': None}
----
Searching for: sf
----
{'run_tool': {'messages': [{'role': 'tool', 'name': 'weather_search', 'content': 'Sunny!', 'tool_call_id': 'toolu_01Kn67GmQAA3BEF1cfYdNW3c'}]}}
{'call_llm': {'messages': [AIMessage(content="According to the search, it's sunny in San Francisco today!", additional_kwargs={}, response_metadata={'id': 'msg_01FJTbC8oK5fkD73rUBmAtUx', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 457, 'output_tokens': 17}}, id='run-c21af72d-3cc5-4b74-bb7c-fbeb8f88bd6d-0', usage_metadata={'input_tokens': 457, 'output_tokens': 17, 'total_tokens': 474, 'input_token_details': {}})]}}
编辑工具调用¶
现在假设我们想要编辑工具调用。例如，更改一些参数（甚至更改调用的工具！），然后执行该工具。
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in sf?"}]}
# Thread
thread = {"configurable": {"thread_id": "3"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
print(event)
print("\n")
{'call_llm': {'messages': [AIMessage(content=[{'text': "I'll help you check the weather in San Francisco.", 'type': 'text'}, {'id': 'toolu_013eUXow3jwM6eekcDJdrjDa', 'input': {'city': 'sf'}, 'name': 'weather_search', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_013ruFpCRNZKX3cDeBAH8rEb', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 379, 'output_tokens': 65}}, id='run-13df3982-ce6d-4fe2-9e5c-ea6ce30a63e4-0', tool_calls=[{'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_013eUXow3jwM6eekcDJdrjDa', 'type': 'tool_call'}], usage_metadata={'input_tokens': 379, 'output_tokens': 65, 'total_tokens': 444, 'input_token_details': {}})]}}
{'__interrupt__': (Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_013eUXow3jwM6eekcDJdrjDa', 'type': 'tool_call'}}, resumable=True, ns=['human_review_node:da717c23-60a0-2a1a-45de-cac5cff308bb'], when='during'),)}
Command，其恢复值为{"action": "update", "data": <tool call args>}。这将执行以下操作：
- 将现有工具调用与用户提供的工具调用参数相结合，并用新的工具调用更新现有的AI消息
- 导航到具有更新后的AI消息的
run_tool节点，并继续执行
# Let's now continue executing from here
for event in graph.stream(
Command(resume={"action": "update", "data": {"city": "San Francisco, USA"}}),
thread,
stream_mode="updates",
):
print(event)
print("\n")
{'human_review_node': {'messages': [{'role': 'ai', 'content': [{'text': "I'll help you check the weather in San Francisco.", 'type': 'text'}, {'id': 'toolu_013eUXow3jwM6eekcDJdrjDa', 'input': {'city': 'sf'}, 'name': 'weather_search', 'type': 'tool_use'}], 'tool_calls': [{'id': 'toolu_013eUXow3jwM6eekcDJdrjDa', 'name': 'weather_search', 'args': {'city': 'San Francisco, USA'}}], 'id': 'run-13df3982-ce6d-4fe2-9e5c-ea6ce30a63e4-0'}]}}
----
Searching for: San Francisco, USA
----
{'run_tool': {'messages': [{'role': 'tool', 'name': 'weather_search', 'content': 'Sunny!', 'tool_call_id': 'toolu_013eUXow3jwM6eekcDJdrjDa'}]}}
{'call_llm': {'messages': [AIMessage(content="According to the search, it's sunny in San Francisco right now!", additional_kwargs={}, response_metadata={'id': 'msg_01QssVtxXPqr8NWjYjTaiHqN', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 460, 'output_tokens': 18}}, id='run-8ab865c8-cc9e-4300-8e1d-9eb673e8445c-0', usage_metadata={'input_tokens': 460, 'output_tokens': 18, 'total_tokens': 478, 'input_token_details': {}})]}}
给工具调用提供反馈¶
有时，你可能不想执行一个工具调用，但也不想让用户手动修改工具调用。在这种情况下，最好从用户那里获取自然语言反馈。你可以将此反馈插入为工具调用的模拟 RESULT。
有多种方法可以做到这一点：
- 可以向状态中添加一条新消息（代表工具调用的“结果”）
- 可以向状态中添加两条新消息——一条代表工具调用的“错误”，另一条代表用户的反馈
这两种方法都涉及向状态中添加消息，主要区别在于 human_review_node 之后的逻辑以及如何处理不同类型的消息。
在这个示例中，我们将只添加一个代表反馈的工具调用（参见 human_review_node 实现）。让我们看看实际操作吧！
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in sf?"}]}
# Thread
thread = {"configurable": {"thread_id": "4"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
print(event)
print("\n")
{'call_llm': {'messages': [AIMessage(content=[{'text': "I'll help you check the weather in San Francisco.", 'type': 'text'}, {'id': 'toolu_01QxXNTCasnNLQCGAiVoNUBe', 'input': {'city': 'sf'}, 'name': 'weather_search', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01DjwkVxgfqT2K329rGkycx6', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 379, 'output_tokens': 65}}, id='run-c57bee36-9f5f-4d2e-85df-758b56d3cc05-0', tool_calls=[{'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_01QxXNTCasnNLQCGAiVoNUBe', 'type': 'tool_call'}], usage_metadata={'input_tokens': 379, 'output_tokens': 65, 'total_tokens': 444, 'input_token_details': {}})]}}
{'__interrupt__': (Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'weather_search', 'args': {'city': 'sf'}, 'id': 'toolu_01QxXNTCasnNLQCGAiVoNUBe', 'type': 'tool_call'}}, resumable=True, ns=['human_review_node:47a3f541-b630-5f8a-32d7-5a44826d99da'], when='during'),)}
Command，其恢复值为{"action": "feedback", "data": <feedback string>}。这将执行以下操作：
- 创建一个新的工具消息，该消息结合了来自LLM的现有工具调用和用户提供的反馈内容
- 导航到
call_llm节点，并使用更新的工具消息继续执行
# Let's now continue executing from here
for event in graph.stream(
# provide our natural language feedback!
Command(
resume={
"action": "feedback",
"data": "User requested changes: use <city, country> format for location",
}
),
thread,
stream_mode="updates",
):
print(event)
print("\n")
{'human_review_node': {'messages': [{'role': 'tool', 'content': 'User requested changes: use <city, country> format for location', 'name': 'weather_search', 'tool_call_id': 'toolu_01QxXNTCasnNLQCGAiVoNUBe'}]}}
{'call_llm': {'messages': [AIMessage(content=[{'text': 'Let me try again with the full city name.', 'type': 'text'}, {'id': 'toolu_01WBGTKBWusaPNZYJi5LKmeQ', 'input': {'city': 'San Francisco, USA'}, 'name': 'weather_search', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_0141KCdx6KhJmWXyYwAYGvmj', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 468, 'output_tokens': 68}}, id='run-60c8267a-52c7-4b6e-87ca-16aa3bd6266b-0', tool_calls=[{'name': 'weather_search', 'args': {'city': 'San Francisco, USA'}, 'id': 'toolu_01WBGTKBWusaPNZYJi5LKmeQ', 'type': 'tool_call'}], usage_metadata={'input_tokens': 468, 'output_tokens': 68, 'total_tokens': 536, 'input_token_details': {}})]}}
{'__interrupt__': (Interrupt(value={'question': 'Is this correct?', 'tool_call': {'name': 'weather_search', 'args': {'city': 'San Francisco, USA'}, 'id': 'toolu_01WBGTKBWusaPNZYJi5LKmeQ', 'type': 'tool_call'}}, resumable=True, ns=['human_review_node:621fc4a9-bbf1-9a99-f50b-3bf91675234e'], when='during'),)}
for event in graph.stream(
Command(resume={"action": "continue"}), thread, stream_mode="updates"
):
print(event)
print("\n")
{'human_review_node': None}
----
Searching for: San Francisco, USA
----
{'run_tool': {'messages': [{'role': 'tool', 'name': 'weather_search', 'content': 'Sunny!', 'tool_call_id': 'toolu_01WBGTKBWusaPNZYJi5LKmeQ'}]}}
{'call_llm': {'messages': [AIMessage(content='The weather in San Francisco is sunny!', additional_kwargs={}, response_metadata={'id': 'msg_01JrfZd8SYyH51Q8rhZuaC3W', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 549, 'output_tokens': 12}}, id='run-09a198b2-79fa-484d-9d9d-f12432978488-0', usage_metadata={'input_tokens': 549, 'output_tokens': 12, 'total_tokens': 561, 'input_token_details': {}})]}}