đ¯ What You'll Learn Today
LangGraph Tutorial: Direct Tool Execution System - Unit 2.2 Exercise 5
Try It Yourself
đĸ Joint Initiative
This tutorial is part of a collaboration between
AI Product Engineer
and
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 foundif __name__ == "__main__":
demonstrate_tool_execution()