如何审查工具调用¶
人工参与 (HIL) 交互对于 代理系统 至关重要。常见的模式是在某些工具调用之后添加一些人工参与步骤。这些工具调用通常会导致函数调用或保存某些信息。示例包括
- 执行 SQL 的工具调用，然后将由该工具运行
- 生成摘要的工具调用，然后将摘要保存到图的状态
请注意，使用工具调用很常见，无论是否实际调用工具。
通常有几种不同的交互您可能想在此处执行
- 批准工具调用并继续
- 手动修改工具调用，然后继续
- 提供自然语言反馈，然后将其传递回代理
我们可以使用 LangGraph 中的 interrupt() 函数来实现这些。interrupt 允许我们停止图的执行，以收集用户的输入，并使用收集到的输入继续执行
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
}
)
review_action, review_data = human_review
# 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)
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]})
设置¶
首先，我们需要安装所需的软件包
接下来，我们需要为 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")
设置 LangSmith 以进行 LangGraph 开发
注册 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 参考：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，其不同的 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': {}})]}}
向工具调用提供反馈¶
有时，您可能不想执行工具调用，但也可能不想要求用户手动修改工具调用。在这种情况下，最好从用户那里获得自然语言反馈。然后，您可以将此反馈作为工具调用的模拟结果插入。
有多种方法可以做到这一点
- 您可以向状态添加新消息（表示工具调用的“结果”）
- 您可以向状态添加两条新消息 - 一条表示工具调用的“错误”，另一条 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': {}})]}}