Tutorial Image: LangGraph Tutorial: Tool Result Processing - Unit 2.2 Exercise 7

LangGraph Tutorial: Tool Result Processing - Unit 2.2 Exercise 7

Learn how to process and format tool execution results in LangGraph. This tutorial covers result state management, error handling, message generation, and maintaining state immutability to ensure consistent and robust conversation flow.

๐ŸŽฏ What You'll Learn Today

LangGraph Tutorial: Tool Result Processing - Unit 2.2 Exercise 7

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 properly process and format tool execution results in LangGraph applications.

Key Concepts Covered

  1. Result State Management
  2. Message Generation
  3. Error Handling and Formatting
  4. State Immutability
from typing import Annotated, TypedDict
!pip install langchain-core
!pip install langgraph
from langchain_core.messages import AIMessage, BaseMessage
from langgraph.graph.message import add_messages

Step 1: State Definition for Results

Define the state structure for tool result processing.

Why This Matters

Result state structure is crucial because

  1. Tracks execution outputs consistently
  2. Maintains message history
  3. Enables proper error handling

Debug Tips

  1. State Structure Issues:
  • Verify tool_outputs initialization
  • Check message list structure
  • Monitor state immutability
class State(TypedDict):
    """State container for result processing.

    Notes:
        - messages tracks conversation history
        - tool_outputs stores execution results
        - All fields must be properly initialized

    Attributes:
        messages: Conversation history with proper typing
        tool_outputs: List of tool execution results
    """

    messages: Annotated[list[BaseMessage], add_messages]
    tool_outputs: list[str]

Step 2: Result Processing Implementation

Process tool execution results and generate appropriate messages.

Why This Matters

Result processing is critical because

  1. Formats outputs consistently
  2. Handles errors gracefully
  3. Maintains conversation flow

Debug Tips

  1. Processing Issues:
  • Check output existence
  • Verify error detection
  • Monitor message formatting
def result_processor(state: State) -> State:
    """Process and format tool execution results.

    Notes:
        - Checks for tool_outputs existence
        - Handles error messages specially
        - Creates appropriate AI responses
        - Maintains state immutability

    Args:
        state: Current conversation state

    Returns:
        Updated state with formatted results
    """
    # Early return if no results to process
    if not state.get("tool_outputs"):
        return state

    # Get latest tool output
    tool_output = state["tool_outputs"][-1]

    # Format output with error handling
    formatted_output = str(tool_output)
    if "error" in formatted_output.lower():
        formatted_output = f"Sorry, there was an error: {formatted_output}"

    # Update state immutably with new message
    return {
        **state,  # Maintain immutability
        "messages": state.get("messages", []) + [AIMessage(content=formatted_output)],
    }

Step 3: System Demonstration

Test the result processing system with various scenarios.

Why This Matters

Testing verifies

  1. Error handling works
  2. Message formatting is correct
  3. State updates properly
def demonstrate_result_processing():
    """Demonstrate result processing with various test cases."""
    # Test cases
    test_cases = [
        {
            "name": "Successful Result",
            "output": "Temperature is 72ยฐF",
            "description": "Processing normal output",
        },
        {
            "name": "Error Result",
            "output": "Error: Invalid temperature reading",
            "description": "Processing error message",
        },
        {
            "name": "Empty Result",
            "output": "",
            "description": "Processing empty output",
        },
    ]

    print("Result Processing Demonstration")
    print("==============================")

    for case in test_cases:
        print(f"\nTest: {case['name']}")
        print(f"Description: {case['description']}")

        # Initialize state with test output
        state = {"messages": [], "tool_outputs": [case["output"]]}

        # Process result
        result = result_processor(state)

        # Print results
        print(f"Input: {case['output']}")
        if result["messages"]:
            print(f"Processed Output: {result['messages'][-1].content}")
        else:
            print("No messages generated")

        print("-" * 50)

Common Pitfalls

  1. Not checking for empty outputs
  2. Mutating state directly
  3. Missing error handling
  4. Improper message formatting

Key Takeaways

  1. Always verify tool outputs exist
  2. Handle errors explicitly
  3. Maintain state immutability
  4. Format messages consistently

Expected Output

Result Processing Demonstration
Test: Successful Result
Description: Processing normal output
Input: Temperature is 72ยฐF

## Processed Output: Temperature is 72ยฐF

Test: Error Result
Description: Processing error message
Input: Error: Invalid temperature reading

## Processed Output: Sorry, there was an error: Error: Invalid temperature reading

Test: Empty Result
Description: Processing empty output

if __name__ == "__main__":
    demonstrate_result_processing()

Rod Rivera

๐Ÿ‡ฌ๐Ÿ‡ง Chapter