🎯 What You'll Learn Today
LangGraph Tutorial: Direct Tool Execution System - Unit 2.2 Exercise 5
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 how to implement and execute tools directly in LangGraph, focusing on safe execution, state management, and proper error handling.
Key Concepts Covered
- Tool Definition and Decoration
- Safe Mathematical Evaluation
- State Management
- Error Handling and Type Safety
import uuid
from typing import Annotated, Any, TypedDict
#!pip install langchain-core
#!pip install langgraph
import numexpr
from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph.message import add_messagesStep 1: State Definition
Define the state structure for tool execution tracking.
Why This Matters
Proper state structure is crucial because
- Tracks tool execution history
- Maintains conversation context
- Enables proper error handling
Debug Tips
-
State Structure Issues:
- Verify TypedDict fields initialization
- Check message list structure
- Monitor tool output tracking
class State(TypedDict):
"""State container for tool execution system.
Attributes:
messages: Conversation history
tool_name: Active tool identifier
tool_outputs: Results from executions
tool_input: Current tool input
"""
messages: Annotated[list[BaseMessage], add_messages]
tool_name: str | None
tool_outputs: list[str]
tool_input: str | NoneStep 2: Tool Implementation
Create safe, well-documented tools with proper decorators.
Why This Matters
Tool implementation is critical because
- Ensures safe execution
- Provides clear documentation
- Maintains type safety
Debug Tips
-
Tool Implementation Issues:
- Check error handling
- Verify input validation
- Monitor return types
@tool
def calculator(expression: str) -> str:
"""Safely evaluate mathematical expressions.
Args:
expression: Mathematical expression to evaluate
Returns:
String representation of result
Examples:
>>> calculator("2 + 2")
'4'
"""
try:
result = numexpr.evaluate(expression.strip()).item()
return str(result)
except Exception as e:
return f"Error evaluating expression: {e!s}"@tool
def weather_check(location: str) -> str:
"""Simulate weather information retrieval.
Args:
location: Location name
Returns:
Simulated weather data
"""
return f"Weather in {location}: 22°C, Sunny"Step 3: Tool Execution Engine
Implement the core tool execution logic with proper error handling.
Why This Matters
Execution engine is essential because
- Manages tool lifecycle
- Handles errors gracefully
- Updates state properly
Debug Tips
-
Execution Issues:
- Log tool selection
- Monitor state updates
- Track error handling
def execute_direct_tool(state: State, tools: dict[str, Any]) -> State:
"""Execute selected tool and update state.
Args:
state: Current system state
tools: Available tool mapping
Returns:
Updated state with execution results
"""
if not state.get("tool_name") or not state.get("tool_input"):
return {**state, "tool_outputs": []}
tool_name = state["tool_name"]
tool_input = state["tool_input"]
tool = tools.get(tool_name)
if not tool:
error_msg = f"Tool '{tool_name}' not found"
return {
**state,
"tool_outputs": [error_msg],
"messages": state.get("messages", []) + [AIMessage(content=error_msg)],
}
try:
output = tool.invoke(tool_input)
tool_message = ToolMessage(
content=output, tool_call_id=str(uuid.uuid4()), name=tool_name
)
ai_message = AIMessage(content=f"Tool execution result: {output}")
return {
**state,
"tool_outputs": [output],
"messages": state.get("messages", []) + [tool_message, ai_message],
}
except Exception as e:
error_msg = f"Error executing {tool_name}: {e!s}"
return {
**state,
"tool_outputs": [error_msg],
"messages": state.get("messages", []) + [AIMessage(content=error_msg)],
}Step 4: System Demonstration
Create comprehensive test cases for the tool execution system.
Why This Matters
System demonstration is valuable because
- Verifies functionality
- Shows error handling
- Provides usage examples
Debug Tips
-
Testing Issues:
- Verify edge cases
- Check error scenarios
- Monitor state transitions
def demonstrate_tool_execution():
"""Demonstrate tool execution with various test cases."""
tools = {"calculator": calculator, "weather": weather_check}
test_cases = [
{
"tool_name": "calculator",
"tool_input": "2 + 2",
"description": "Simple addition",
},
{
"tool_name": "calculator",
"tool_input": "3 * (4 + 5)",
"description": "Complex calculation",
},
{"tool_name": "weather", "tool_input": "Paris", "description": "Weather check"},
{
"tool_name": "unknown_tool",
"tool_input": "test",
"description": "Invalid tool",
},
]
print("Tool Execution Demonstration")
print("===========================")
for case in test_cases:
print(f"\nTest: {case['description']}")
state = {
"messages": [],
"tool_name": case["tool_name"],
"tool_input": case["tool_input"],
"tool_outputs": [],
}
result = execute_direct_tool(state, tools)
print(f"Tool: {case['tool_name']}")
print(f"Input: {case['tool_input']}")
print(
f"Output: {result['tool_outputs'][0] if result['tool_outputs'] else 'No output'}"
)
print("-" * 40)Common Pitfalls
- Unsafe expression evaluation
- Missing error handling
- Improper state updates
- Incomplete type checking
Key Takeaways
- Safe tool execution requires proper validation
- Error handling is crucial for stability
- State updates must be immutable
- Type safety prevents runtime errors
Next Steps
- Add tool execution history
- Implement tool rate limiting
- Add result caching
- Enhance error reporting
Expected Output
Tool Execution Demonstration
Test: Simple addition
Tool: calculator
Input: 2 + 2
## Output: 4
Test: Complex calculation
Tool: calculator
Input: 3 * (4 + 5)
## Output: 27
Test: Weather check
Tool: weather
Input: Paris
## Output: Weather in Paris: 22°C, Sunny
Test: Invalid tool
Tool: unknown_tool
Input: test
## Output: Tool 'unknown_tool' not found
if __name__ == "__main__":
demonstrate_tool_execution()
🇬🇧 Chapter