Logo of AI Product Engineer
AI Product Engineer
Tutorial Image: LangGraph Tutorial: Parallel Tool Execution State Management - Unit 2.3 Exercise 1

LangGraph Tutorial: Parallel Tool Execution State Management - Unit 2.3 Exercise 1

Learn how to manage state effectively for parallel tool execution in LangGraph. This tutorial explores creating a robust state structure for concurrent operations, tracking pending tools, managing results and errors, and ensuring type safety. Discover how to initialize and update the state to streamline multi-tool workflows and handle results seamlessly.

🎯 What You'll Learn Today

LangGraph Tutorial: Parallel Tool Execution State Management - Unit 2.3 Exercise 1

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 state management for parallel tool execution in LangGraph, enabling concurrent operations and result tracking.

Key Concepts Covered

  1. State Structure for Parallel Operations
  2. Pending Tool Management
  3. Result and Error Tracking
  4. Type Safety
from typing import Annotated, Any, TypedDict
!pip install langchain-core
!pip install langgraph
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph.message import add_messages

Step 1: State Definition for Parallel Execution

Define state structure to manage parallel tool operations.

Why This Matters

Parallel execution state is crucial because

  1. Enables concurrent tool operations
  2. Tracks execution status
  3. Manages results and errors

Debug Tips

  1. State Structure Issues:
  • Verify tool ID uniqueness
  • Check pending tool formatting
  • Monitor result/error maps
class State(TypedDict):
    """State container for parallel tool execution.

    Notes:
        - messages tracks conversation flow
        - pending_tools stores queued operations
        - results maps tool IDs to outputs
        - errors tracks failures by tool ID

    Attributes:
        messages: Conversation history with proper typing
        pending_tools: List of tool operations to execute
        results: Mapping of tool IDs to execution results
        errors: Mapping of tool IDs to error messages
    """

    messages: Annotated[list[BaseMessage], add_messages]
    pending_tools: list[dict]  # [{id, tool_name, args}]
    results: dict[str, Any]  # {tool_id: result}
    errors: dict[str, str]  # {tool_id: error_message}

Step 2: State Usage Examples

Demonstrate proper state initialization and usage patterns.

Why This Matters

Usage examples are essential because

  1. Show proper initialization
  2. Demonstrate field access
  3. Illustrate state updates

Debug Tips

  1. Usage Issues:
  • Check tool ID generation
  • Verify argument formatting
  • Monitor state updates
def demonstrate_parallel_state():
    """Demonstrate parallel state management."""
    # Initialize parallel execution state
    state = {
        "messages": [HumanMessage(content="Starting parallel execution...")],
        "pending_tools": [
            {
                "id": "search_1",
                "tool_name": "TavilySearchResults",
                "args": {"query": "capital of France"},
            },
            {
                "id": "search_2",
                "tool_name": "TavilySearchResults",
                "args": {"query": "largest city in Japan"},
            },
        ],
        "results": {},
        "errors": {},
    }

    # Example state operations
    print("Parallel State Management Example")
    print("================================")

    # Access pending tools
    print("\nPending Tools:")
    for tool in state["pending_tools"]:
        print(f"- Tool ID: {tool['id']}")
        print(f"  Name: {tool['tool_name']}")
        print(f"  Args: {tool['args']}")

    # Simulate result updates
    state["results"]["search_1"] = "Paris is the capital of France"
    state["results"]["search_2"] = "Tokyo is the largest city in Japan"

    # Show results
    print("\nExecution Results:")
    for tool_id, result in state["results"].items():
        print(f"- {tool_id}: {result}")

    return state

Common Pitfalls

  1. Non-unique tool IDs
  2. Missing required tool fields
  3. Improper type annotations
  4. Inconsistent state updates

Key Takeaways

  1. Tool IDs must be unique
  2. State structure enables parallel tracking
  3. Type safety ensures consistency
  4. Proper initialization is crucial

Next Steps

  1. Add tool execution tracking
  2. Implement result aggregation
  3. Add error recovery
  4. Enhance status monitoring

Expected Output

Parallel State Management Example

## Pending Tools

- Tool ID: search_1

 Name: TavilySearchResults
 Args: {'query': 'capital of France'}

- Tool ID: search_2

 Name: TavilySearchResults
 Args: {'query': 'largest city in Japan'}

## Execution Results

- search_1: Paris is the capital of France
- search_2: Tokyo is the largest city in Japan

if __name__ == "__main__":
    demonstrate_parallel_state()

Rod Rivera

🇬🇧 Chapter