LLMs are not perfect when calling tools. Models may attempt to call non-existent tools or fail to return parameters that match the requested schema. Strategies such as keeping the schema simple, reducing the number of tools passed at once, and using good names and descriptions can help mitigate this risk, but they are not foolproof.
This guide introduces methods for building error handling in graphs to alleviate these failure modes.
Automatic handling of tool nodes
We simulate some exceptions:
from langchain_core.tools import tool
@tool
def get_weather(location: str): """Call to get the current weather."""
if location == "shanghai": raise ValueError("Input queries must be proper nouns")
elif location == "Shanghai": return "It's 60 degrees and foggy."
else: raise ValueError("Invalid input.")
Then we build the same graph:
from typing import Literal
from langchain_ollama import ChatOllama
import base_conf
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([get_weather])
model_with_tools = (ChatOllama(base_url=base_conf.base_url, model=base_conf.model_name, temperature=base_conf.temperature) .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)
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()
The flowchart is as follows:
When attempting to call the tool, it will be found that the model uses incorrect input to call the tool, causing the tool to throw an error.
The pre-built ToolNode for executing the tool has some built-in error handling capabilities that can catch errors and return them to the model so that it can retry:
response = app.invoke( {"messages": [HumanMessage("what is the weather in shanghai?")]},)
for message in response["messages"]: string_representation = f"{message.type.upper()}: {message.content}\n" print(string_representation)
HUMAN: what is the weather in shanghai?
AI:
TOOL: Error: ValueError('Input queries must be proper nouns') Please fix your mistakes.
AI:
TOOL: It's 60 degrees and foggy.
AI: Currently in Shanghai, it is 60 degrees with foggy conditions. Please make sure to carry an umbrella if you plan on going outside!
As we can see, the tools wrapped by ToolNode handle context exceptions when encountering errors, prompting the LLM to re-infer, which is very convenient!
If you want to customize your strategy, is there a way? Yes, first you should not use ToolNode but instead mimic its approach. What is its approach?
The wrapping of ToolNode is actually the following code:
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}
In other words, we configure a tool call node in the graph ourselves, and when an exception occurs, we catch it ourselves, then construct a ToolMessage and add it to the state, and then add a routing function in the graph, similar to:
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"
When we detect an exception in the state, we specify the next node. By manually wrapping ToolMessage and using exception routing judgment in this way, we can achieve our most flexible error handling.