Tutorial Image: LangGraph Tutorial: Mastering ToolNode Implementation - Unit 2.2 Exercise 6

LangGraph Tutorial: Mastering ToolNode Implementation - Unit 2.2 Exercise 6

Master the implementation of ToolNode in LangGraph with this comprehensive tutorial. Learn how to structure state management, create robust tools, and execute them efficiently with proper message formatting, result handling, and error management. This exercise demonstrates critical best practices to build scalable and reliable tool execution workflows.

๐ŸŽฏ 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

  1. Required Message Structure for ToolNode
  2. Proper Tool Implementation
  3. Execution Patterns and State Management
  4. 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

  1. ToolNode requires specific message formats
  2. Tool execution needs context
  3. Results must be tracked consistently

Debug Tips

  1. 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

  1. Args must match ToolNode expectations
  2. Return types must be consistent
  3. Error handling prevents execution failures

Debug Tips

  1. 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

  1. ToolNode requires specific message format
  2. Result processing needs careful type checking
  3. Error handling must be comprehensive

Debug Tips

  1. 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

  1. Message formatting works
  2. Tool execution succeeds
  3. 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()

Rod Rivera

๐Ÿ‡ฌ๐Ÿ‡ง Chapter