🎯 What You'll Learn Today

import json
import uuid
from typing import Annotated, Any, TypedDict
#!pip install langchain-core
#!pip install langgraph
from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

class State(TypedDict):
    """State container for result processing.

    This state implementation tracks:
    1. Message history with special handling
    2. Tool call specifications
    3. Raw tool outputs

    Attributes:
        messages: List of conversation messages
        tool_calls: List of pending tool calls
        tool_outputs: List of raw tool execution results
    """

    messages: Annotated[list[BaseMessage], add_messages]
    tool_calls: list[dict]
    tool_outputs: list[Any]

def get_last_message_by_type(
    messages: list[BaseMessage], message_type
) -> BaseMessage | None:
    """Find the last message of a specific type.

    Args:
        messages: List of messages to search
        message_type: Type of message to find

    Returns:
        Last message of specified type or None
    """
    for message in reversed(messages):
        if isinstance(message, message_type):
            return message
    return None

def process_tool_output(tool_output: Any) -> tuple[str, Any]:
    """Process raw tool output into usable format.

    Args:
        tool_output: Raw output to process

    Returns:
        Tuple of (processed_content, raw_data)
    """
    try:
        data = json.loads(tool_output) if isinstance(tool_output, str) else tool_output
        content = data.get("result", str(data))
        return content, data
    except json.JSONDecodeError:
        return str(tool_output), tool_output

def result_processor(state: State) -> State:
    """Process tool results into appropriate message types.

    This function demonstrates:
    1. Tool output processing
    2. Message type conversion
    3. Conversation structuring

    Args:
        state: Current conversation state

    Returns:
        Updated state with new messages
    """
    if not state.get("tool_outputs"):
        return {"messages": [], "tool_calls": [], "tool_outputs": []}

    tool_output = state["tool_outputs"][-1]
    processed_content, raw_data = process_tool_output(tool_output)

    tool_message = ToolMessage(
        content=str(raw_data),
        tool_call_id=str(uuid.uuid4()),
        name="search_tool",
    )

    ai_message = AIMessage(content=f"Here's what I found: {processed_content}")

    return {
        "messages": [tool_message, ai_message],
        "tool_calls": [],
        "tool_outputs": [],
    }

def should_end(state: State) -> bool:
    """Determine if conversation should end.

    Args:
        state: Current conversation state

    Returns:
        Boolean indicating whether to end
    """
    if not state.get("messages"):
        return False

    last_ai = get_last_message_by_type(state["messages"], AIMessage)
    if not last_ai:
        return False

    return "Here's what I found:" in last_ai.content

def create_processing_graph() -> StateGraph:
    """Create and configure the processing graph.

    Returns:
        Compiled StateGraph
    """
    graph = StateGraph(State)
    graph.add_node("processor", result_processor)
    graph.add_edge(START, "processor")
    graph.add_conditional_edges(
        "processor", should_end, {True: END, False: "processor"}
    )
    return graph.compile()

def main():
    """Demonstrate result processing functionality."""
    initial_state = {
        "tool_outputs": [json.dumps({"result": "Paris is the capital of France"})],
        "messages": [],
        "tool_calls": [],
    }

    chain = create_processing_graph()
    result = chain.invoke(initial_state)

    print("\nFinal conversation state:")
    for message in result["messages"]:
        if isinstance(message, ToolMessage):
            print(f"\nTool Output ({message.name}): {message.content}")
        elif isinstance(message, AIMessage):
            print(f"\nAssistant: {message.content}")
        else:
            print(f"\n{message.__class__.__name__}: {message.content}")

if __name__ == "__main__":
    main()

## Final conversation state

Tool Output (search_tool): {"result": "Paris is the capital of France"}
Assistant: Here's what I found: Paris is the capital of France