๐ฏ What You'll Learn Today
LangGraph Tutorial: Parallel Tool Execution State Management - Unit 2.3 Exercise 1
Try It Yourself
๐ข Joint Initiative
This tutorial is part of a collaboration between AIPE and 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
- State Structure for Parallel Operations
- Pending Tool Management
- Result and Error Tracking
- 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
- Enables concurrent tool operations
- Tracks execution status
- Manages results and errors
Debug Tips
- 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
- Show proper initialization
- Demonstrate field access
- Illustrate state updates
Debug Tips
- 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
- Non-unique tool IDs
- Missing required tool fields
- Improper type annotations
- Inconsistent state updates
Key Takeaways
- Tool IDs must be unique
- State structure enables parallel tracking
- Type safety ensures consistency
- Proper initialization is crucial
Next Steps
- Add tool execution tracking
- Implement result aggregation
- Add error recovery
- 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()