๐ฏ What You'll Learn Today
LangGraph Tutorial: Mastering ToolNode Implementation - Unit 2.2 Exercise 6
This tutorial is also available in Google Colab here or for download here
Joint Initiative: This tutorial is part of a collaboration between AI Product Engineer and the Nebius Academy.
This tutorial demonstrates critical patterns and best practices for using ToolNode in LangGraph, based on hands-on implementation experience.
Key Concepts Covered
- Required Message Structure for ToolNode
- Proper Tool Implementation
- Execution Patterns and State Management
- Error Handling Best Practices
import uuid
from datetime import datetime
from typing import Annotated, Any, TypedDict
!pip install langchain-core
!pip install langgraph
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
Step 1: State Definition
Define our state structure for proper tool execution.
Why This Matters
Correct state structure is essential because
- ToolNode requires specific message formats
- Tool execution needs context
- Results must be tracked consistently
Debug Tips
-
State Structure Issues:
- Initialize all TypedDict fields
- Ensure message list is never None
- Track both tool_name and outputs
class State(TypedDict):
"""State container for ToolNode execution.
Notes:
- messages must be initialized as empty list
- tool_name can be None but field must exist
- tool_outputs tracks execution results
"""
messages: Annotated[list[BaseMessage], add_messages]
tool_name: str | None
tool_outputs: list[str]
Step 2: Tool Implementation
Create tools with proper error handling and consistent interfaces.
Why This Matters
Tool implementation affects ToolNode because
- Args must match ToolNode expectations
- Return types must be consistent
- Error handling prevents execution failures
Debug Tips
-
Tool Issues:
- Check parameter names match ToolNode
- Verify error handling is complete
- Ensure consistent return types
@tool
def calculator(query: str) -> str:
"""Calculate mathematical expressions safely.
Note:
- query parameter name must match ToolNode args
- Always return string for consistency
- Handle all potential errors
"""
try:
expression = query.replace("what is ", "").replace("calculate ", "")
result = eval(expression) # In production, use numexpr
return f"{expression} = {result}"
except Exception as e:
return f"Error in calculation: {e!s}"
@tool
def check_weather(query: str) -> str:
"""Get simulated weather information.
Note:
- query parameter matches ToolNode expectation
- Return format is consistent
- All errors are caught and handled
"""
try:
location = (
query.lower()
.replace("what's the weather in ", "")
.replace("weather in ", "")
)
current_time = datetime.now().strftime("%H:%M")
return (
f"Weather in {location.title()} at {current_time}:\n"
f"Temperature: 22ยฐC\n"
f"Condition: Sunny\n"
f"Humidity: 60%"
)
except Exception as e:
return f"Error getting weather: {e!s}"
Step 3: ToolNode Execution
Execute tools with proper message formatting and result handling.
Why This Matters
Execution structure is critical because
- ToolNode requires specific message format
- Result processing needs careful type checking
- Error handling must be comprehensive
Debug Tips
-
Execution Issues:
- Verify tool_calls structure
- Check result message types
- Monitor state updates
def execute_with_tool_node(state: State, tools: list[Any]) -> State:
"""Execute tools using ToolNode with proper message structure.
Note:
- AIMessage must have empty content
- tool_calls requires specific structure
- Result processing needs type verification
"""
if not state.get("tool_name"):
return {**state, "tool_outputs": []}
try:
tool_node = ToolNode(tools)
message = state["messages"][-1].content if state["messages"] else ""
# Required ToolNode message structure
tool_message = AIMessage(
content="", # Must be empty string
tool_calls=[
{
"name": state["tool_name"],
"args": {"query": message}, # args must match tool parameters
"id": str(uuid.uuid4()),
"type": "tool_call", # Required by ToolNode
}
],
)
# Simple invoke pattern - no extra parameters
result = tool_node.invoke({"messages": [tool_message]})
# Careful result processing with type checks
if (
isinstance(result, dict)
and "messages" in result
and result["messages"]
and isinstance(result["messages"][0], ToolMessage)
):
tool_output = result["messages"][0].content
else:
tool_output = "Tool execution failed: Invalid response format"
tool_result = ToolMessage(
content=tool_output, tool_call_id=str(uuid.uuid4()), name=state["tool_name"]
)
ai_response = AIMessage(content=f"Here's what I found: {tool_output}")
return {
**state,
"tool_outputs": [tool_output],
"messages": state["messages"] + [tool_result, ai_response],
}
except Exception as e:
error_msg = f"Error executing tool: {e!s}"
print(f"Debug - Error details: {type(e).__name__}: {e!s}")
return {
**state,
"tool_outputs": [error_msg],
"messages": state["messages"] + [AIMessage(content=error_msg)],
}
Step 4: System Demonstration
Test the implementation with various scenarios.
Why This Matters
Testing verifies
- Message formatting works
- Tool execution succeeds
- Error handling functions
def demonstrate_toolnode():
"""Demonstrate ToolNode execution with various examples."""
tools = [calculator, check_weather]
test_cases = [
{
"tool_name": "calculator",
"message": "what is 5 * 3",
"description": "Basic calculation",
},
{
"tool_name": "check_weather",
"message": "what's the weather in Tokyo",
"description": "Weather check",
},
]
print("ToolNode Execution Demonstration")
print("===============================")
for case in test_cases:
print(f"\nTest: {case['description']}")
state = {
"messages": [HumanMessage(content=case["message"])],
"tool_name": case["tool_name"],
"tool_outputs": [],
}
result = execute_with_tool_node(state, tools)
print(f"Query: {case['message']}")
print(f"Tool: {case['tool_name']}")
if result["tool_outputs"]:
print(f"Result: {result['tool_outputs'][0]}")
print("\nConversation:")
for msg in result["messages"]:
prefix = {
HumanMessage: "Human",
AIMessage: "Assistant",
ToolMessage: "Tool",
}.get(type(msg), "Unknown")
print(f"{prefix}: {msg.content}")
print("-" * 50)
if __name__ == "__main__":
demonstrate_toolnode()
๐ฌ๐ง Chapter