如何处理工具调用错误¶
LLMs在调用工具时并不总是完美的。模型可能会尝试调用一个不存在的工具，或者无法返回与请求模式匹配的参数。保持模式简单、减少一次传递的工具数量、以及拥有良好的名称和描述可以帮助缓解这些风险，但这些方法并不是万无一失的。
本指南将介绍一些方法，帮助您在图中构建错误处理机制，以缓解这些失败模式。
设置¶
首先，让我们安装所需的包并设置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应用程序——更多关于如何开始的信息，请参阅这里。
使用预构建的 ToolNode¶
首先，定义一个模拟天气工具，该工具对输入查询有一些隐含的限制。这里的意图是模拟一个真实世界的案例，其中模型未能正确调用工具：
from langchain_core.tools import tool
@tool
def get_weather(location: str):
"""Call to get the current weather."""
if location == "san francisco":
raise ValueError("Input queries must be proper nouns")
elif location == "San Francisco":
return "It's 60 degrees and foggy."
else:
raise ValueError("Invalid input.")
API Reference: tool
接下来，设置ReAct代理的图实现。该代理接收某个查询作为输入，然后反复调用工具，直到它拥有足够的信息来解决该查询。我们将使用预构建的ToolNode来执行调用的工具，并使用由Anthropic提供支持的小型快速模型：
from typing import Literal
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([get_weather])
model_with_tools = ChatAnthropic(
model="claude-3-haiku-20240307", temperature=0
).bind_tools([get_weather])
def should_continue(state: MessagesState):
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: MessagesState):
messages = state["messages"]
response = model_with_tools.invoke(messages)
return {"messages": [response]}
workflow = StateGraph(MessagesState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent")
app = workflow.compile()
API Reference: ChatAnthropic | StateGraph | START | END | ToolNode
from IPython.display import Image, display
try:
display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
当你尝试调用该工具时，可以看到模型使用了错误的输入来调用工具，导致工具抛出错误。预构建的ToolNode负责执行工具，并且内置了一些错误处理机制，用于捕获错误并将错误回传给模型，以便模型能够再次尝试。
response = app.invoke(
{"messages": [("human", "what is the weather in san francisco?")]},
)
for message in response["messages"]:
string_representation = f"{message.type.upper()}: {message.content}\n"
print(string_representation)
HUMAN: what is the weather in san francisco?
AI: [{'id': 'toolu_01K5tXKVRbETcs7Q8U9PHy96', 'input': {'location': 'san francisco'}, 'name': 'get_weather', 'type': 'tool_use'}]
TOOL: Error: ValueError('Input queries must be proper nouns')
Please fix your mistakes.
AI: [{'text': 'Apologies, it looks like there was an issue with the weather lookup. Let me try that again with the proper format:', 'type': 'text'}, {'id': 'toolu_01KSCsme3Du2NBazSJQ1af4b', 'input': {'location': 'San Francisco'}, 'name': 'get_weather', 'type': 'tool_use'}]
TOOL: It's 60 degrees and foggy.
AI: The current weather in San Francisco is 60 degrees and foggy.
自定义策略¶
这种情况下的默认设置在许多情况下都是合适的，但在某些情况下，自定义回退策略可能更好。
例如，下面的工具需要输入一组特定长度的元素——这对小型模型来说可能比较棘手！我们还将故意避免将 topic 多复数化，以误导模型认为它应该传递一个字符串：
from langchain_core.output_parsers import StrOutputParser
from pydantic import BaseModel, Field
class HaikuRequest(BaseModel):
topic: list[str] = Field(
max_length=3,
min_length=3,
)
@tool
def master_haiku_generator(request: HaikuRequest):
"""Generates a haiku based on the provided topics."""
model = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)
chain = model | StrOutputParser()
topics = ", ".join(request.topic)
haiku = chain.invoke(f"Write a haiku about {topics}")
return haiku
tool_node = ToolNode([master_haiku_generator])
model = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)
model_with_tools = model.bind_tools([master_haiku_generator])
def should_continue(state: MessagesState):
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: MessagesState):
messages = state["messages"]
response = model_with_tools.invoke(messages)
return {"messages": [response]}
workflow = StateGraph(MessagesState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent")
app = workflow.compile()
response = app.invoke(
{"messages": [("human", "Write me an incredible haiku about water.")]},
{"recursion_limit": 10},
)
for message in response["messages"]:
string_representation = f"{message.type.upper()}: {message.content}\n"
print(string_representation)
API Reference: StrOutputParser
HUMAN: Write me an incredible haiku about water.
AI: [{'text': 'Here is a haiku about water:', 'type': 'text'}, {'id': 'toolu_01L13Z3Gtaym5KKgPXVyZhYn', 'input': {'topic': ['water']}, 'name': 'master_haiku_generator', 'type': 'tool_use'}]
TOOL: Error: 1 validation error for master_haiku_generator
request
Field required [type=missing, input_value={'topic': ['water']}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.7/v/missing
Please fix your mistakes.
AI: [{'text': 'Oops, my apologies. Let me try that again with the correct format:', 'type': 'text'}, {'id': 'toolu_01HCQ5uXr5kXQHBQ3FyQ1Ysk', 'input': {'topic': ['water']}, 'name': 'master_haiku_generator', 'type': 'tool_use'}]
TOOL: Error: 1 validation error for master_haiku_generator
request
Field required [type=missing, input_value={'topic': ['water']}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.7/v/missing
Please fix your mistakes.
AI: [{'text': 'Hmm, it seems there was an issue with the input format. Let me try a different approach:', 'type': 'text'}, {'id': 'toolu_01RF96nruwr4nMqhLBRsbfE5', 'input': {'request': {'topic': ['water']}}, 'name': 'master_haiku_generator', 'type': 'tool_use'}]
TOOL: Error: 1 validation error for master_haiku_generator
request.topic
List should have at least 3 items after validation, not 1 [type=too_short, input_value=['water'], input_type=list]
For further information visit https://errors.pydantic.dev/2.7/v/too_short
Please fix your mistakes.
AI: [{'text': 'Ah I see, the haiku generator requires at least 3 topics. Let me provide 3 topics related to water:', 'type': 'text'}, {'id': 'toolu_011jcgHuG2Kyr87By459huqQ', 'input': {'request': {'topic': ['ocean', 'rain', 'river']}}, 'name': 'master_haiku_generator', 'type': 'tool_use'}]
TOOL: Here is a haiku about ocean, rain, and river:
Vast ocean's embrace,
Raindrops caress the river,
Nature's symphony.
AI: I hope this haiku about water captures the essence you were looking for! Let me know if you would like me to generate another one.
更好的策略可能是修剪失败的尝试以减少干扰，然后退回到更高级的模型。这里有一个例子。我们使用自建的节点来调用我们的工具，而不是使用预构建的ToolNode：
import json
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.messages.modifier import RemoveMessage
@tool
def master_haiku_generator(request: HaikuRequest):
"""Generates a haiku based on the provided topics."""
model = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)
chain = model | StrOutputParser()
topics = ", ".join(request.topic)
haiku = chain.invoke(f"Write a haiku about {topics}")
return haiku
def call_tool(state: MessagesState):
tools_by_name = {master_haiku_generator.name: master_haiku_generator}
messages = state["messages"]
last_message = messages[-1]
output_messages = []
for tool_call in last_message.tool_calls:
try:
tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
output_messages.append(
ToolMessage(
content=json.dumps(tool_result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
except Exception as e:
# Return the error if the tool call fails
output_messages.append(
ToolMessage(
content="",
name=tool_call["name"],
tool_call_id=tool_call["id"],
additional_kwargs={"error": e},
)
)
return {"messages": output_messages}
model = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)
model_with_tools = model.bind_tools([master_haiku_generator])
better_model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0)
better_model_with_tools = better_model.bind_tools([master_haiku_generator])
def should_continue(state: MessagesState):
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def should_fallback(
state: MessagesState,
) -> Literal["agent", "remove_failed_tool_call_attempt"]:
messages = state["messages"]
failed_tool_messages = [
msg
for msg in messages
if isinstance(msg, ToolMessage)
and msg.additional_kwargs.get("error") is not None
]
if failed_tool_messages:
return "remove_failed_tool_call_attempt"
return "agent"
def call_model(state: MessagesState):
messages = state["messages"]
response = model_with_tools.invoke(messages)
return {"messages": [response]}
def remove_failed_tool_call_attempt(state: MessagesState):
messages = state["messages"]
# Remove all messages from the most recent
# instance of AIMessage onwards.
last_ai_message_index = next(
i
for i, msg in reversed(list(enumerate(messages)))
if isinstance(msg, AIMessage)
)
messages_to_remove = messages[last_ai_message_index:]
return {"messages": [RemoveMessage(id=m.id) for m in messages_to_remove]}
# Fallback to a better model if a tool call fails
def call_fallback_model(state: MessagesState):
messages = state["messages"]
response = better_model_with_tools.invoke(messages)
return {"messages": [response]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)
workflow.add_node("remove_failed_tool_call_attempt", remove_failed_tool_call_attempt)
workflow.add_node("fallback_agent", call_fallback_model)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_conditional_edges("tools", should_fallback)
workflow.add_edge("remove_failed_tool_call_attempt", "fallback_agent")
workflow.add_edge("fallback_agent", "tools")
app = workflow.compile()
API Reference: AIMessage | ToolMessage | RemoveMessage
tools 节点现在会在工具调用失败时返回带有 additional_kwargs 中 error 字段的 ToolMessage。如果发生这种情况，它将转到另一个节点，该节点会移除失败的工具消息，并让更好的模型重试工具调用生成。
下图显示了这一过程：
try:
display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
让我们试一试。为了强调移除步骤，我们将从模型中“流式传输”响应，以便我们可以看到每个已执行的节点：
stream = app.stream(
{"messages": [("human", "Write me an incredible haiku about water.")]},
{"recursion_limit": 10},
)
for chunk in stream:
print(chunk)
{'agent': {'messages': [AIMessage(content=[{'text': 'Here is a haiku about water:', 'type': 'text'}, {'id': 'toolu_019mY8NX4t7YkJBWeHG6jE4T', 'input': {'topic': ['water']}, 'name': 'master_haiku_generator', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01RmoaLh38DnRX2fv7E8vCFh', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 384, 'output_tokens': 67}}, id='run-a1511215-1a62-49b5-b5b3-b2c8f8c7920e-0', tool_calls=[{'name': 'master_haiku_generator', 'args': {'topic': ['water']}, 'id': 'toolu_019mY8NX4t7YkJBWeHG6jE4T', 'type': 'tool_call'}], usage_metadata={'input_tokens': 384, 'output_tokens': 67, 'total_tokens': 451})]}}
{'tools': {'messages': [ToolMessage(content='', name='master_haiku_generator', id='69f85339-dbc2-4341-8c4d-26300dfe31a5', tool_call_id='toolu_019mY8NX4t7YkJBWeHG6jE4T')]}}
{'remove_failed_tool_call_attempt': {'messages': [RemoveMessage(content='', additional_kwargs={}, response_metadata={}, id='run-a1511215-1a62-49b5-b5b3-b2c8f8c7920e-0'), RemoveMessage(content='', additional_kwargs={}, response_metadata={}, id='69f85339-dbc2-4341-8c4d-26300dfe31a5')]}}
{'fallback_agent': {'messages': [AIMessage(content=[{'text': 'Certainly! I\'d be happy to help you create an incredible haiku about water. To do this, I\'ll use the master_haiku_generator function, which requires three topics. Since you\'ve specified water as the main theme, I\'ll add two related concepts to create a more vivid and interesting haiku. Let\'s use "water," "flow," and "reflection" as our three topics.', 'type': 'text'}, {'id': 'toolu_01FxSxy8LeQ5PjdNYq8vLFTd', 'input': {'request': {'topic': ['water', 'flow', 'reflection']}}, 'name': 'master_haiku_generator', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01U5HV3pt1NVm6syGbxx29no', 'model': 'claude-3-5-sonnet-20240620', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 414, 'output_tokens': 158}}, id='run-3eb746c7-b607-4ad3-881a-1c11a7638af7-0', tool_calls=[{'name': 'master_haiku_generator', 'args': {'request': {'topic': ['water', 'flow', 'reflection']}}, 'id': 'toolu_01FxSxy8LeQ5PjdNYq8vLFTd', 'type': 'tool_call'}], usage_metadata={'input_tokens': 414, 'output_tokens': 158, 'total_tokens': 572})]}}
{'tools': {'messages': [ToolMessage(content='"Here is a haiku about water, flow, and reflection:\\n\\nRippling waters flow,\\nMirroring the sky above,\\nTranquil reflection."', name='master_haiku_generator', id='fdfc497d-939a-42c0-8748-31371b98a3a7', tool_call_id='toolu_01FxSxy8LeQ5PjdNYq8vLFTd')]}}
{'agent': {'messages': [AIMessage(content='I hope you enjoy this haiku about the beauty and serenity of water. Please let me know if you would like me to generate another one.', additional_kwargs={}, response_metadata={'id': 'msg_012rXWHapc8tPfBPEonpAT6W', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 587, 'output_tokens': 35}}, id='run-ab6d412d-9374-4a4b-950d-6dcc43d87cf5-0', usage_metadata={'input_tokens': 587, 'output_tokens': 35, 'total_tokens': 622})]}}
你也可以检查这个 LangSmith 跟踪，它显示了较小模型的初始调用失败。
下一步¶
你现在已经看到了如何实现一些处理工具调用错误的策略。
接下来，查看一些 其他 LangGraph 如何操作指南。