如何审查工具调用¶
人机交互（HIL）对于智能体系统至关重要。常见的模式是在某些工具调用后添加一些人机交互步骤。这些工具调用通常会导致函数调用或保存某些信息。例如包括：
- 调用一个执行 SQL 的工具，之后该工具将运行此 SQL
- 调用一个生成摘要的工具，之后该摘要将被保存到图的状态中
请注意，使用工具调用是常见的 无论是否实际调用工具。
在此处，您可能想要进行几种不同的交互：
- 批准工具调用并继续
- 手动修改工具调用，然后继续
- 提供自然语言反馈，然后将反馈传递回代理
我们可以在 LangGraph 中使用 [interrupt()][langgraph.types.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（我们将使用的LLM）设置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 或正确的工具。
让我们实际看一下！
API Reference: StateGraph | START | END | MemorySaver | Command | interrupt | ChatAnthropic | tool | AIMessage
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()))
无需审查的示例¶
让我们来看一个不需要审查的示例（因为没有调用任何工具）
# 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 并采用不同的 resume 值 {"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 插入。
实现这一目标有多种方式：
- 你可以向状态中添加一条新消息（代表工具调用的“结果”）。
- 你可以向状态中添加两条新消息：一条表示工具调用的“错误”，另一条是 HumanMessage 表示反馈。
这两种方法在本质上相似，都涉及向状态中添加消息。主要区别在于 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，并使用不同的 resume 值 {"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': {}})]}}