LangGraph Basics: Building Advanced AI Agents with Graph Architecture

Why LangGraph?

Before diving into installation, let’s understand why LangGraph is a powerful tool for building AI agents:

1. Graph-Based Architecture: LangGraph allows you to structure your agent’s logic as a graph, making complex flows more manageable and explicit.
2. State Management: It provides built-in state management, ensuring your agent maintains context throughout its operation.
3. Composability: Build complex agents by composing smaller, reusable components.
4. Integration with LangChain: Seamlessly works with LangChain’s components, leveraging its extensive ecosystem.
5. Flexibility: Supports various patterns from simple sequential flows to complex multi-agent systems.

Installation and Setup with uv

Please see our uv tutorial for instructions on installing and setting up uv. These commands will work with pip and a standard virtual environment as well.

Installation Steps

Here’s how to set up your environment using uv:

1. Create a virtual environment:
   uv venv .venv
2. Activate the virtual environment:
   # On macOS/Linux:
source .venv/bin/activate
# On Windows:
.venv\Scripts\activate
3. Install the required packages:
   uv pip install langgraph langchain langchain-openai

Code Walkthrough

## Simple verification that langgraph is installed correctly
from langgraph.graph import StateGraph

## Print the version of langgraph
import langgraph

## You can also verify other dependencies are installed
import langchain
import langchain_openai

print(f"LangChain version: {langchain.__version__}")

print("\nYour LangGraph environment is set up correctly!")

What This Code Does

1. Imports Key Modules: The code imports the essential components from LangGraph and related libraries.
2. Version Verification: It prints the versions of installed packages, which serves two purposes:
   • Confirms that the packages are installed correctly
   • Provides version information for troubleshooting or documentation
3. Environment Check: The successful execution of this script confirms that your Python environment is correctly set up for the tutorial.

Now that you have LangGraph installed, you’re ready to learn about the core concepts of nodes and edges in the next section. These are the fundamental building blocks that make LangGraph powerful for creating AI agents.

Understanding Nodes and Edges in LangGraph

In this section, we’ll learn:

• What nodes and edges are in LangGraph
• How to create and connect nodes to form a graph
• How data flows through a graph
• How to implement conditional logic in your graph

The Power of Graph-Based Agents

Graph-based agents offer several advantages over traditional sequential programming:

1. Explicit Flow Representation: Graphs visually represent your agent’s logic flow, making it easier to understand and reason about.
2. Modularity: Each node performs a specific function, allowing you to build complex systems from simple components.
3. Flexibility: Easily modify your agent’s behavior by adding, removing, or reconnecting nodes.
4. Conditional Routing: Direct your agent’s flow based on the current state, enabling dynamic decision-making.
5. Reusability: Create reusable node patterns that can be applied across different agents.

Nodes: The Building Blocks

In LangGraph, nodes are the fundamental units of computation. A node:

• Takes the current state as input
• Performs some operation (processing, decision-making, etc.)
• Returns updates to the state

Nodes can represent various operations:

• Processing nodes: Transform data or perform calculations
• LLM nodes: Generate text or make decisions using language models
• Tool nodes: Interact with external systems like databases or APIs
• Decision nodes: Determine the next steps in the graph

Edges: Defining the Flow

Edges connect nodes and define how your agent’s execution flows from one node to another. There are several types of edges in LangGraph:

• Sequential edges: Simple connections where execution always flows from one node to the next
• Conditional edges: Direct execution based on the output of a decision function
• Dynamic edges: Determine the next node at runtime based on complex logic

Code Walkthrough

# Understanding Nodes and Edges in LangGraph

"""
This example demonstrates the core concepts of nodes and edges in LangGraph.

Nodes are the building blocks of your graph-based agent, representing
individual processing steps, while edges define the flow between nodes.
"""

import json
from typing import TypedDict, Annotated, Literal, List

from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, END


# Define a simple state schema
class AgentState(TypedDict):
    messages: List[HumanMessage | AIMessage]
    current_step: str


# Define node functions
def process_input(state: AgentState) -> AgentState:
    """Process the user input"""
    print("Processing input...")
    # In a real application, you would process the input here
    return {"current_step": "thinking"}


def thinking(state: AgentState) -> AgentState:
    """Think about the problem"""
    print("Thinking about the problem...")
    # In a real application, you would use an LLM here
    return {"current_step": "thinking"}


def respond(state: AgentState) -> AgentState:
    """Generate a response"""
    print("Generating response...")
    # In a real application, you would generate a response here
    state["messages"].append(AIMessage(content="This is a response from the agent."))
    return {"current_step": "done"}


# Define a conditional edge function
def decide_next_step(state: AgentState) -> Literal["respond", "thinking"]:
    """Decide the next step based on the current state"""
    # Simple conditional logic
    if state["current_step"] == "thinking":
        return "respond"
    else:
        return "thinking"


# Create a graph
def build_graph():
    # Initialize the graph with our state schema
    workflow = StateGraph(AgentState)
    
    # Add nodes to the graph
    workflow.add_node("process_input", process_input)
    workflow.add_node("thinking", thinking)
    workflow.add_node("respond", respond)
    
    # Add edges to define the flow
    # Simple sequential flow
    workflow.add_edge("process_input", "thinking")
    
    # Conditional edge using the decide_next_step function
    workflow.add_conditional_edges(
        "thinking",
        decide_next_step,
        {
            "respond": "respond",
            "thinking": "thinking"  # This creates a potential loop
        }
    )
    
    # End the graph after responding
    workflow.add_edge("respond", END)
    
    # Set the entry point
    workflow.set_entry_point("process_input")
    
    return workflow.compile()


if __name__ == "__main__":
    
    # Build the graph
    graph = build_graph()
    
    # Run the graph with initial state
    initial_state = {
        "messages": [HumanMessage(content="Hello, agent!")],
        "current_step": "start"
    }
    
    result = graph.invoke(initial_state)
    
    print("\nFinal state:")
    print(json.dumps({
        "messages": [msg.content for msg in result["messages"]],
        "current_step": result["current_step"]
    }, indent=2))

1. Defining the State Schema

class AgentState(TypedDict):
    messages: List[HumanMessage | AIMessage]
    current_step: str

This defines the structure of our agent’s state, which includes a list of messages and a string indicating the current step.

2. Creating Node Functions

def process_input(state: AgentState) -> AgentState:
    """Process the user input"""
    print("Processing input...")
    # In a real application, you would process the input here
    return {"current_step": "thinking"}


def thinking(state: AgentState) -> AgentState:
    """Think about the problem"""
    print("Thinking about the problem...")
    # In a real application, you would use an LLM here
    return {"current_step": "thinking"}


def respond(state: AgentState) -> AgentState:
    """Generate a response"""
    print("Generating response...")
    # In a real application, you would generate a response here
    state["messages"].append(AIMessage(content="This is a response from the agent."))
    return {"current_step": "done"}

Each function represents a node in our graph. Note that:

• Each function takes the current state as input
• Each function returns a partial state update (only the changed fields)
• LangGraph automatically merges these updates into the full state

3. Defining a Conditional Edge Function

def decide_next_step(state: AgentState) -> Literal["respond", "thinking"]:
    """Decide the next step based on the current state"""
    # Simple conditional logic
    if state["current_step"] == "thinking":
        return "respond"
    else:
        return "thinking"

This function determines which node to execute next based on the current state.

4. Building the Graph

def build_graph():
    # Initialize the graph with our state schema
    workflow = StateGraph(AgentState)
    
    # Add nodes to the graph
    workflow.add_node("process_input", process_input)
    workflow.add_node("thinking", thinking)
    workflow.add_node("respond", respond)
    
    # Add edges to define the flow
    # Simple sequential flow
    workflow.add_edge("process_input", "thinking")
    
    # Conditional edge using the decide_next_step function
    workflow.add_conditional_edges(
        "thinking",
        decide_next_step,
        {
            "respond": "respond",
            "thinking": "thinking"  # This creates a potential loop
        }
    )
    
    # End the graph after responding
    workflow.add_edge("respond", END)
    
    # Set the entry point
    workflow.set_entry_point("process_input")
    
    return workflow.compile()

This function creates the graph structure by:

1. Initializing a graph with our state schema
2. Adding nodes to the graph
3. Adding regular edges for sequential flow
4. Adding conditional edges for dynamic routing
5. Setting the entry point (where execution begins)
6. Compiling the graph for execution

5. Running the Graph

## Build the graph
graph = build_graph()

## Run the graph with initial state
initial_state = {
    "messages": [HumanMessage(content="Hello, agent!")],
    "current_step": "start"
}

result = graph.invoke(initial_state)

This code initializes the graph with a starting state and runs it, returning the final state after execution completes.

Key Concepts Explained

State Updates

In LangGraph, nodes only need to return the parts of the state they modify. The framework handles merging these partial updates into the complete state. This makes node functions cleaner and more focused.

Conditional Routing

The add_conditional_edges method allows you to dynamically choose the next node based on the current state. This enables complex decision-making within your graph.

The END Constant

The special END constant marks the termination point of your graph. When execution reaches a node connected to END, the graph completes and returns the final state.

Practical Applications

This pattern of nodes and edges is powerful for many AI agent scenarios:

• Conversational Agents: Process user input, generate responses, and manage dialog state
• Task Automation: Break complex tasks into discrete steps with conditional logic
• Decision Trees: Implement sophisticated decision-making based on various inputs
• Multi-Step Reasoning: Guide LLMs through structured reasoning processes

Now that you understand nodes and edges, the next section will explore how to work with schemas in LangGraph, which will help you define and validate the structure of your agent’s state.

Working with Schemas in LangGraph

Why Schemas Matter in LangGraph

Schemas are the foundation of reliable agent development in LangGraph. They provide several crucial benefits:

1. Type Safety: Catch errors at development time rather than runtime
2. Documentation: Self-documenting code that clearly shows the structure of your agent’s state
3. Validation: Ensure that data flowing through your graph maintains the expected structure
4. Maintainability: Make your code easier to understand and modify

Schema Options in LangGraph

LangGraph supports two primary ways to define schemas:

1. TypedDict

A lightweight approach using Python’s built-in typing system:

from typing import TypedDict, List, Dict, Any

class SimpleState(TypedDict):
    counter: int
    messages: List[str]
    metadata: Dict[str, Any]

2. Pydantic Models

A more robust approach with runtime validation and additional features:

from pydantic import BaseModel, Field

class AdvancedState(BaseModel):
    counter: int = Field(default=0, description="A simple counter")
    messages: List[str] = Field(default_factory=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)

Code Walkthrough

# Working with Schemas in LangGraph

"""
This example demonstrates how to work with schemas in LangGraph.

Schemas define the structure of your agent's state and help with validation.
They ensure that data flowing through your graph maintains the expected structure.
"""

from typing import TypedDict, List, Dict, Any, Optional, Union, Annotated
from pydantic import BaseModel, Field

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END


# Basic schema using TypedDict
class SimpleState(TypedDict):
    """A simple state definition using TypedDict"""
    counter: int
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    metadata: Dict[str, Any]


# More advanced schema using Pydantic
class AgentMemory(BaseModel):
    """Agent's memory component"""
    short_term: List[str] = Field(default_factory=list, description="Short-term memory items")
    long_term: Dict[str, Any] = Field(default_factory=dict, description="Long-term memory storage")


class AgentTools(BaseModel):
    """Available tools for the agent"""
    active_tools: List[str] = Field(default_factory=list)
    tool_results: Dict[str, Any] = Field(default_factory=dict)


class AdvancedAgentState(BaseModel):
    """A more complex state definition using Pydantic"""
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]] = Field(default_factory=list)
    memory: AgentMemory = Field(default_factory=AgentMemory)
    tools: AgentTools = Field(default_factory=AgentTools)
    current_step: str = Field(default="start")
    error: Optional[str] = Field(default=None)


# Example node functions that work with these schemas
def update_counter(state: SimpleState) -> Dict[str, Any]:
    """Increment the counter in a simple state"""
    return {"counter": state["counter"] + 1}


def add_to_memory(state: dict) -> dict:
    """Add an item to the agent's short-term memory"""
    # Convert dict to Pydantic model for better IDE support and validation
    s = AdvancedAgentState.model_validate(state)
    
    # Add a new memory item
    s.memory.short_term.append(f"Memory item at step {s.current_step}")
    
    # Return only the changed parts as a dict
    return {"memory": s.memory.model_dump()}


def demonstrate_schema_usage():
    """Demonstrate how to use both schema types"""
    print("\n1. Using SimpleState (TypedDict):")
    # Initialize a state that conforms to SimpleState
    simple_state: SimpleState = {
        "counter": 0,
        "messages": [HumanMessage(content="Hello")],
        "metadata": {"session_id": "12345"}
    }
    
    # Update the state
    updated = update_counter(simple_state)
    simple_state = {**simple_state, **updated}
    
    print(f"  Counter updated: {simple_state['counter']}")
    
    print("\n2. Using AdvancedAgentState (Pydantic):")
    # Initialize using the Pydantic model
    advanced_state = AdvancedAgentState(
        messages=[SystemMessage(content="You are a helpful assistant")],
        current_step="processing"
    )
    
    # Convert to dict for processing
    state_dict = advanced_state.model_dump()
    
    # Update the state
    updated = add_to_memory(state_dict)
    
    # Merge updates back
    for key, value in updated.items():
        state_dict[key] = value
    
    # Convert back to Pydantic model
    updated_state = AdvancedAgentState.model_validate(state_dict)
    
    print(f"  Memory items: {updated_state.memory.short_term}")


# Example of creating a graph with a Pydantic schema
def create_graph_with_pydantic_schema():
    """Create a graph using a Pydantic model as the state schema"""
    # The graph will convert the Pydantic model to a TypedDict internally
    graph = StateGraph(AdvancedAgentState)
    
    # Add a node that works with our schema
    graph.add_node("memory_node", add_to_memory)
    
    # For demonstration purposes, we'll create a simple linear flow
    graph.add_edge("memory_node", END)
    graph.set_entry_point("memory_node")
    
    return graph.compile()


if __name__ == "__main__":
    # Demonstrate schema usage
    demonstrate_schema_usage()
    
    # Create and run a graph with Pydantic schema
    print("\n3. Running a graph with Pydantic schema:")
    graph = create_graph_with_pydantic_schema()
    
    # Create initial state using the Pydantic model
    initial_state = AdvancedAgentState(
        messages=[HumanMessage(content="Hello, agent!")],
        current_step="start"
    )
    
    # Convert to dict for the graph
    result = graph.invoke(initial_state.model_dump())
    
    # Convert result back to Pydantic for better access
    final_state = AdvancedAgentState.model_validate(result)
    
    print(f"  Final memory items: {final_state.memory.short_term}")

1. Basic Schema with TypedDict

class SimpleState(TypedDict):
    """A simple state definition using TypedDict"""
    counter: int
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    metadata: Dict[str, Any]

This defines a basic state schema with three fields: a counter, a list of messages, and a metadata dictionary.

2. Advanced Schema with Pydantic

class AgentMemory(BaseModel):
    """Agent's memory component"""
    short_term: List[str] = Field(default_factory=list, description="Short-term memory items")
    long_term: Dict[str, Any] = Field(default_factory=dict, description="Long-term memory storage")


class AgentTools(BaseModel):
    """Available tools for the agent"""
    active_tools: List[str] = Field(default_factory=list)
    tool_results: Dict[str, Any] = Field(default_factory=dict)


class AdvancedAgentState(BaseModel):
    """A more complex state definition using Pydantic"""
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]] = Field(default_factory=list)
    memory: AgentMemory = Field(default_factory=AgentMemory)
    tools: AgentTools = Field(default_factory=AgentTools)
    current_step: str = Field(default="start")
    error: Optional[str] = Field(default=None)

This more advanced schema uses Pydantic to define a nested structure with:

• Default values for fields
• Nested models for organization
• Field descriptions for documentation
• Optional fields

3. Working with TypedDict Schemas

def update_counter(state: SimpleState) -> Dict[str, Any]:
    """Increment the counter in a simple state"""
    return {"counter": state["counter"] + 1}

This function shows how to work with a TypedDict schema:

1. The function takes a state that conforms to SimpleState
2. It returns a partial update (only the counter field)
3. LangGraph will merge this update into the full state

4. Working with Pydantic Schemas

def add_to_memory(state: dict) -> dict:
    """Add an item to the agent's short-term memory"""
    # Convert dict to Pydantic model for better IDE support and validation
    s = AdvancedAgentState.model_validate(state)
    
    # Add a new memory item
    s.memory.short_term.append(f"Memory item at step {s.current_step}")
    
    # Return only the changed parts as a dict
    return {"memory": s.memory.model_dump()}

This function demonstrates working with Pydantic schemas:

1. Convert the incoming dict to a Pydantic model for validation and IDE support
2. Modify the model using dot notation and proper types
3. Convert the modified parts back to a dict for the state update

5. Creating a Graph with Pydantic Schema

def create_graph_with_pydantic_schema():
    """Create a graph using a Pydantic model as the state schema"""
    # The graph will convert the Pydantic model to a TypedDict internally
    graph = StateGraph(AdvancedAgentState)
    
    # Add a node that works with our schema
    graph.add_node("memory_node", add_to_memory)
    
    # For demonstration purposes, we'll create a simple linear flow
    graph.add_edge("memory_node", END)
    graph.set_entry_point("memory_node")
    
    return graph.compile()

This shows how to use a Pydantic model as the schema for a LangGraph.

Schema Best Practices

When to Use TypedDict

• For simpler agents with straightforward state
• When you want minimal dependencies
• For better performance in high-throughput scenarios

When to Use Pydantic

• For complex state structures with nested data
• When you need runtime validation
• When you want self-documenting schemas with descriptions
• For agents that interface with external APIs

General Tips

1. Start Simple: Begin with a minimal schema and expand as needed
2. Group Related Fields: Use nested models to organize related data
3. Use Descriptive Names: Make field names clear and self-explanatory
4. Document Your Fields: Add descriptions to explain the purpose of each field
5. Consider Defaults: Provide sensible default values to avoid initialization errors

Common Schema Patterns

Conversation State

class ConversationState(TypedDict):
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    current_topic: Optional[str]
    user_info: Dict[str, Any]

Tool-Using Agent State

class ToolState(TypedDict):
    messages: List[Union[HumanMessage, AIMessage]]
    available_tools: List[str]
    tool_calls: List[Dict[str, Any]]
    tool_results: List[Dict[str, Any]]

Task Processing State

class TaskState(TypedDict):
    task_description: str
    status: Literal["not_started", "in_progress", "completed", "failed"]
    steps_completed: List[str]
    result: Optional[Dict[str, Any]]

Now that you understand how to define and work with schemas, the next section will show you how to build complete graphs that manipulate state data between nodes, creating functional agents with LangGraph.

Building Your First LangGraph with Schema Manipulation

• How to build a complete functional graph in LangGraph
• How to transform and manipulate data between nodes
• How state flows through a graph during execution
• How to implement decision logic for dynamic graph execution

The Power of State Manipulation

One of LangGraph’s key strengths is its ability to maintain and transform state throughout the execution of your agent. This provides several advantages:

1. Persistent Context: Maintain information across multiple steps of processing
2. Incremental Updates: Each node can focus on updating only what it needs to
3. Clean Separation: Keep your agent’s logic modular and focused
4. Traceability: Easily track how data changes throughout execution

Code Walkthrough

# Building Your First LangGraph with Schema Manipulation

"""
This example demonstrates how to build a simple LangGraph and manipulate schema data
between nodes. It shows how to transform data as it flows through the graph.
"""

import json
from typing import TypedDict, List, Dict, Any, Annotated
from enum import Enum

from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END


# Define our state schema
class TaskStatus(str, Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


class TaskState(TypedDict):
    """State for a task processing agent"""
    task_description: str
    status: TaskStatus
    messages: List[HumanMessage | AIMessage]
    steps_taken: List[str]
    result: Dict[str, Any]


# Define node functions that manipulate the schema
def initialize_task(state: TaskState) -> Dict[str, Any]:
    """Initialize the task processing"""
    print(f"Initializing task: {state['task_description']}")
    
    # Update the state with initialization data
    return {
        "status": TaskStatus.IN_PROGRESS,
        "steps_taken": ["Task initialized"],
        "result": {"progress": 0}
    }


def process_task(state: TaskState) -> Dict[str, Any]:
    """Process the task"""
    print(f"Processing task. Current status: {state['status']}")
    
    # In a real application, you might use an LLM here
    # For demonstration, we'll simulate processing
    
    # Add to steps taken
    steps = state["steps_taken"].copy()
    steps.append("Task processed")
    
    # Update the result
    result = state["result"].copy()
    result["progress"] = 50
    result["details"] = "Halfway through processing"
    
    # Return only the changed parts
    return {
        "steps_taken": steps,
        "result": result
    }


def analyze_results(state: TaskState) -> Dict[str, Any]:
    """Analyze the results of the task"""
    print(f"Analyzing results. Progress: {state['result']['progress']}%")
    
    # Add to steps taken
    steps = state["steps_taken"].copy()
    steps.append("Results analyzed")
    
    # Update the result
    result = state["result"].copy()
    result["progress"] = 75
    result["analysis"] = "Task analysis complete"
    
    # Add a message summarizing the analysis
    messages = state["messages"].copy()
    messages.append(AIMessage(content=f"Analysis complete with {result['progress']}% progress."))
    
    return {
        "steps_taken": steps,
        "result": result,
        "messages": messages
    }


def complete_task(state: TaskState) -> Dict[str, Any]:
    """Mark the task as completed"""
    print("Completing task")
    
    # Add to steps taken
    steps = state["steps_taken"].copy()
    steps.append("Task completed")
    
    # Update the result
    result = state["result"].copy()
    result["progress"] = 100
    result["outcome"] = "Success"
    
    # Add a completion message
    messages = state["messages"].copy()
    messages.append(AIMessage(content="Task has been successfully completed!"))
    
    return {
        "status": TaskStatus.COMPLETED,
        "steps_taken": steps,
        "result": result,
        "messages": messages
    }


# Decision function to determine if we need more processing
def check_if_complete(state: TaskState) -> str:
    """Check if the task needs more processing or is complete"""
    if state["result"]["progress"] >= 75:
        return "complete"
    else:
        return "process_more"


def build_task_graph():
    """Build a graph for task processing"""
    # Initialize the graph with our state schema
    workflow = StateGraph(TaskState)
    
    # Add nodes
    workflow.add_node("initialize", initialize_task)
    workflow.add_node("process", process_task)
    workflow.add_node("analyze", analyze_results)
    workflow.add_node("complete", complete_task)
    
    # Add edges
    workflow.add_edge("initialize", "process")
    workflow.add_edge("process", "analyze")
    
    # Add conditional edge based on completion status
    workflow.add_conditional_edges(
        "analyze",
        check_if_complete,
        {
            "complete": "complete",
            "process_more": "process"  # Loop back for more processing if needed
        }
    )
    
    # End the graph
    workflow.add_edge("complete", END)
    
    # Set the entry point
    workflow.set_entry_point("initialize")
    
    return workflow.compile()


if __name__ == "__main__":
    # Build the graph
    task_graph = build_task_graph()
    
    # Create initial state
    initial_state: TaskState = {
        "task_description": "Analyze the quarterly sales data",
        "status": TaskStatus.NOT_STARTED,
        "messages": [HumanMessage(content="Please analyze our quarterly sales data.")],
        "steps_taken": [],
        "result": {}
    }
    
    # Run the graph
    print("\nRunning task graph...")
    final_state = task_graph.invoke(initial_state)
    
    # Display the final state
    print("\nFinal State:")
    print(json.dumps({
        "task_description": final_state["task_description"],
        "status": final_state["status"],
        "steps_taken": final_state["steps_taken"],
        "result": final_state["result"],
        "messages": [msg.content for msg in final_state["messages"]]
    }, indent=2))

1. Defining the State Schema

class TaskStatus(str, Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


class TaskState(TypedDict):
    """State for a task processing agent"""
    task_description: str
    status: TaskStatus
    messages: List[HumanMessage | AIMessage]
    steps_taken: List[str]
    result: Dict[str, Any]

This schema defines the state for a task processing agent, including:

• A task description
• The current status (using an Enum for type safety)
• A history of messages
• Steps that have been taken
• Results accumulated during processing

2. Node Functions for State Manipulation

def initialize_task(state: TaskState) -> Dict[str, Any]:
    """Initialize the task processing"""
    print(f"Initializing task: {state['task_description']}")
    
    # Update the state with initialization data
    return {
        "status": TaskStatus.IN_PROGRESS,
        "steps_taken": ["Task initialized"],
        "result": {"progress": 0}
    }

This first node takes the initial state and updates it with new values. Note that it only returns the fields it wants to modify.

def process_task(state: TaskState) -> Dict[str, Any]:
    """Process the task"""
    print(f"Processing task. Current status: {state['status']}")
    
    # Add to steps taken
    steps = state["steps_taken"].copy()
    steps.append("Task processed")
    
    # Update the result
    result = state["result"].copy()
    result["progress"] = 50
    result["details"] = "Halfway through processing"
    
    # Return only the changed parts
    return {
        "steps_taken": steps,
        "result": result
    }

The processing node demonstrates how to:

1. Copy and modify list fields (to avoid modifying the original state)
2. Copy and update nested dictionaries
3. Return only the modified portions of the state

3. Decision Function for Conditional Routing

def check_if_complete(state: TaskState) -> str:
    """Check if the task needs more processing or is complete"""
    if state["result"]["progress"] >= 75:
        return "complete"
    else:
        return "process_more"

This function examines the current state and decides what path to take next, enabling dynamic routing in the graph.

4. Building the Graph

def build_task_graph():
    """Build a graph for task processing"""
    # Initialize the graph with our state schema
    workflow = StateGraph(TaskState)
    
    # Add nodes
    workflow.add_node("initialize", initialize_task)
    workflow.add_node("process", process_task)
    workflow.add_node("analyze", analyze_results)
    workflow.add_node("complete", complete_task)
    
    # Add edges
    workflow.add_edge("initialize", "process")
    workflow.add_edge("process", "analyze")
    
    # Add conditional edge based on completion status
    workflow.add_conditional_edges(
        "analyze",
        check_if_complete,
        {
            "complete": "complete",
            "process_more": "process"  # Loop back for more processing if needed
        }
    )
    
    # End the graph
    workflow.add_edge("complete", END)
    
    # Set the entry point
    workflow.set_entry_point("initialize")
    
    return workflow.compile()

This function builds a complete graph with:

1. Initialization node
2. Processing node
3. Analysis node
4. Completion node
5. Conditional routing based on the state

The flow includes a potential loop, where the task can go back to processing if it’s not yet complete.

5. Running the Graph

## Build the graph
task_graph = build_task_graph()

## Create initial state
initial_state: TaskState = {
    "task_description": "Analyze the quarterly sales data",
    "status": TaskStatus.NOT_STARTED,
    "messages": [HumanMessage(content="Please analyze our quarterly sales data.")],
    "steps_taken": [],
    "result": {}
}

## Run the graph
print("\nRunning task graph...")
final_state = task_graph.invoke(initial_state)

This code initializes the graph with a starting state and runs it, returning the final state after completion.

Key LangGraph State Concepts

Immutable State Updates

LangGraph treats state as immutable, meaning you shouldn’t modify the original state objects. Instead:

1. Copy any collections you need to modify (lists, dicts)
2. Make changes to the copies
3. Return the modified copies as part of your state update

This pattern ensures that state changes are explicit and traceable.

Partial State Updates

Nodes only need to return the parts of the state they modify. This has several benefits:

1. Clarity: Makes it obvious what each node is changing
2. Efficiency: Avoids unnecessary copying of unchanged data
3. Isolation: Reduces the chance of nodes interfering with each other

Conditional Routing

The check_if_complete function demonstrates how to implement conditional logic in your graph:

1. Examine the current state
2. Return a string identifier
3. LangGraph uses this identifier to choose the next node

This allows for dynamic, state-dependent execution paths.

Common Patterns for Schema Manipulation

Updating Collections

When updating collections like lists or dictionaries, always create a copy first:

## For lists
items = state["items"].copy()
items.append(new_item)
return {"items": items}

## For dictionaries
config = state["config"].copy()
config["setting"] = new_value
return {"config": config}

Transforming Data

Nodes often need to transform data from one format to another:

def transform_data(state):
    input_data = state["raw_data"]
    processed = some_processing_function(input_data)
    return {"processed_data": processed}

Accumulating Results

Many agents need to accumulate results over multiple steps:

def add_result(state):
    results = state["results"].copy()
    results.append(new_result)
    return {"results": results}

Practical Applications

The patterns demonstrated in this example can be applied to many real-world scenarios:

• Data Processing Pipelines: Process data through multiple stages with conditional logic
• Multi-Step Reasoning: Break complex reasoning into discrete steps with state tracking
• Conversational Agents: Maintain conversation state and history across interactions
• Task Automation: Track progress of multi-stage tasks with detailed status reporting

Next Steps

Now that you’ve built a simple functional graph with state manipulation, the next section will explore more advanced graph patterns including branching, loops, and error handling.

Advanced Graph Patterns in LangGraph

As your agents become more sophisticated, you’ll need more complex control flow patterns:

1. Branching allows your agent to take different actions based on context
2. Loops enable iterative refinement and processing
3. Error handling makes your agents robust in real-world scenarios
4. Complex routing supports sophisticated decision-making

LangGraph makes these advanced patterns explicit and manageable, unlike traditional programming where complex control flow can become difficult to follow.

Code Walkthrough

# Advanced Graph Patterns in LangGraph

"""
This example demonstrates advanced graph patterns in LangGraph including:
- Branching and conditionals
- Loops and recursion
- Error handling
- Parallel processing
"""

import json
import random
from typing import TypedDict, List, Dict, Any, Literal, Optional, Union, Annotated
from enum import Enum

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END


# Define our state schema
class ProcessingStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCESS = "success"
    ERROR = "error"


class AdvancedState(TypedDict):
    """State for demonstrating advanced patterns"""
    query: str
    status: ProcessingStatus
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    results: Dict[str, Any]
    error: Optional[str]
    processing_path: str
    iteration_count: int


# Node functions
def initialize(state: AdvancedState) -> Dict[str, Any]:
    """Initialize the processing state"""
    print(f"Initializing with query: {state['query']}")
    return {
        "status": ProcessingStatus.PROCESSING,
        "results": {},
        "error": None,
        "iteration_count": 0
    }


# Branching example: Different processing paths
def determine_processing_path(state: AdvancedState) -> Literal["simple", "complex", "error_path"]:
    """Determine which processing path to take based on the query"""
    query = state["query"].lower()
    
    # Simulate different paths based on the query
    if "error" in query:
        print("Taking error path")
        return "error_path"
    elif len(query) > 20 or "complex" in query:
        print("Taking complex processing path")
        return "complex"
    else:
        print("Taking simple processing path")
        return "simple"


# Simple processing path
def simple_processing(state: AdvancedState) -> Dict[str, Any]:
    """Handle simple queries"""
    print("Performing simple processing")
    
    # Update results
    results = state["results"].copy()
    results["simple_result"] = f"Processed '{state['query']}' with simple algorithm"
    results["complexity"] = "low"
    
    return {
        "results": results,
        "processing_path": "simple",
        "status": ProcessingStatus.SUCCESS
    }


# Complex processing path with potential for recursion/loops
def complex_processing(state: AdvancedState) -> Dict[str, Any]:
    """Handle complex queries with potential for multiple iterations"""
    iteration = state["iteration_count"]
    print(f"Performing complex processing (iteration {iteration + 1})")
    
    # Update results
    results = state["results"].copy()
    
    if "complex_steps" not in results:
        results["complex_steps"] = []
    
    results["complex_steps"].append(f"Complex processing step {iteration + 1}")
    results["complexity"] = "high"
    
    # Increment iteration counter
    iteration_count = iteration + 1
    
    # Determine if we need more processing
    status = (
        ProcessingStatus.SUCCESS if iteration_count >= 3 else ProcessingStatus.PROCESSING
    )
    
    return {
        "results": results,
        "processing_path": "complex",
        "iteration_count": iteration_count,
        "status": status
    }


# Error handling path
def error_processing(state: AdvancedState) -> Dict[str, Any]:
    """Demonstrate error handling"""
    print("Error detected in query, handling error case")
    
    return {
        "status": ProcessingStatus.ERROR,
        "error": f"Error processing query: '{state['query']}' contains invalid parameters",
        "processing_path": "error"
    }


# Check if we need more iterations for complex processing
def check_iteration_needed(state: AdvancedState) -> Literal["continue", "complete"]:
    """Determine if more iterations are needed"""
    if state["status"] == ProcessingStatus.PROCESSING:
        return "continue"
    else:
        return "complete"


# Error recovery attempt
def attempt_recovery(state: AdvancedState) -> Dict[str, Any]:
    """Try to recover from errors"""
    print(f"Attempting to recover from error: {state['error']}")
    
    # Simulate a recovery attempt with 50% success rate
    if random.random() > 0.5:
        print("Recovery successful")
        return {
            "status": ProcessingStatus.SUCCESS,
            "error": None,
            "results": {"recovered": True, "original_error": state["error"]}
        }
    else:
        print("Recovery failed")
        return {"status": ProcessingStatus.ERROR}  # Error persists


# Final processing based on status
def finalize_processing(state: AdvancedState) -> Dict[str, Any]:
    """Finalize the processing and generate a summary"""
    status = state["status"]
    print(f"Finalizing processing with status: {status}")
    
    messages = state["messages"].copy()
    
    if status == ProcessingStatus.SUCCESS:
        path = state["processing_path"]
        iterations = state["iteration_count"]
        
        summary = (
            f"Successfully processed query via {path} path "
            f"with {iterations} iteration(s)."
        )
        
        messages.append(AIMessage(content=summary))
    elif status == ProcessingStatus.ERROR:
        messages.append(AIMessage(content=f"Processing failed: {state['error']}"))
    
    return {"messages": messages}


# Build the advanced graph
def build_advanced_graph():
    """Build a graph demonstrating advanced patterns"""
    workflow = StateGraph(AdvancedState)
    
    # Add all nodes
    workflow.add_node("initialize", initialize)
    workflow.add_node("simple_processing", simple_processing)
    workflow.add_node("complex_processing", complex_processing)
    workflow.add_node("error_processing", error_processing)
    workflow.add_node("attempt_recovery", attempt_recovery)
    workflow.add_node("finalize", finalize_processing)
    
    # Start with initialization
    workflow.set_entry_point("initialize")
    
    # Branch based on the query type
    workflow.add_conditional_edges(
        "initialize",
        determine_processing_path,
        {
            "simple": "simple_processing",
            "complex": "complex_processing",
            "error_path": "error_processing"
        }
    )
    
    # For complex processing, we might need multiple iterations
    workflow.add_conditional_edges(
        "complex_processing",
        check_iteration_needed,
        {
            "continue": "complex_processing",  # Loop back for more processing
            "complete": "finalize"  # Move to finalization
        }
    )
    
    # Simple processing goes directly to finalization
    workflow.add_edge("simple_processing", "finalize")
    
    # For errors, try recovery
    workflow.add_edge("error_processing", "attempt_recovery")
    
    # After recovery attempt, always go to finalization
    workflow.add_edge("attempt_recovery", "finalize")
    
    # End after finalization
    workflow.add_edge("finalize", END)
    
    return workflow.compile()


def run_example(query: str):
    """Run the example with a given query"""
    graph = build_advanced_graph()
    
    # Create initial state
    initial_state: AdvancedState = {
        "query": query,
        "status": ProcessingStatus.PENDING,
        "messages": [HumanMessage(content=f"Process this query: {query}")],
        "results": {},
        "error": None,
        "processing_path": "",
        "iteration_count": 0
    }
    
    # Run the graph
    print(f"\nProcessing query: '{query}'")
    print("-" * 50)
    final_state = graph.invoke(initial_state)
    
    # Display the final state
    print("\nFinal State:")
    print(json.dumps({
        "query": final_state["query"],
        "status": final_state["status"],
        "processing_path": final_state["processing_path"],
        "iteration_count": final_state["iteration_count"],
        "results": final_state["results"],
        "error": final_state["error"],
        "messages": [msg.content for msg in final_state["messages"]]
    }, indent=2))
    print("-" * 50)


if __name__ == "__main__":
    # Run with different queries to demonstrate different paths
    run_example("simple query")
    run_example("complex analysis with multiple steps")
    run_example("query with error condition")

1. State Schema for Advanced Patterns

class ProcessingStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCESS = "success"
    ERROR = "error"


class AdvancedState(TypedDict):
    """State for demonstrating advanced patterns"""
    query: str
    status: ProcessingStatus
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    results: Dict[str, Any]
    error: Optional[str]
    processing_path: str
    iteration_count: int

This schema includes fields for tracking status, errors, processing path, and iteration count—all essential for implementing advanced patterns.

2. Branching Based on Input

def determine_processing_path(state: AdvancedState) -> Literal["simple", "complex", "error_path"]:
    """Determine which processing path to take based on the query"""
    query = state["query"].lower()
    
    # Simulate different paths based on the query
    if "error" in query:
        print("Taking error path")
        return "error_path"
    elif len(query) > 20 or "complex" in query:
        print("Taking complex processing path")
        return "complex"
    else:
        print("Taking simple processing path")
        return "simple"

This function examines the input query and decides which processing path to take. It returns a literal string that LangGraph will use to route execution to the appropriate node.

3. Implementing Loops and Iteration

def complex_processing(state: AdvancedState) -> Dict[str, Any]:
    """Handle complex queries with potential for multiple iterations"""
    iteration = state["iteration_count"]
    print(f"Performing complex processing (iteration {iteration + 1})")
    
    # Update results
    results = state["results"].copy()
    
    if "complex_steps" not in results:
        results["complex_steps"] = []
    
    results["complex_steps"].append(f"Complex processing step {iteration + 1}")
    results["complexity"] = "high"
    
    # Increment iteration counter
    iteration_count = iteration + 1
    
    # Determine if we need more processing
    status = (
        ProcessingStatus.SUCCESS if iteration_count >= 3 else ProcessingStatus.PROCESSING
    )
    
    return {
        "results": results,
        "processing_path": "complex",
        "iteration_count": iteration_count,
        "status": status
    }


def check_iteration_needed(state: AdvancedState) -> Literal["continue", "complete"]:
    """Determine if more iterations are needed"""
    if state["status"] == ProcessingStatus.PROCESSING:
        return "continue"
    else:
        return "complete"

These functions implement an iterative processing loop:

1. complex_processing increments a counter and adds processing steps
2. It decides whether more processing is needed based on the iteration count
3. check_iteration_needed determines whether to continue looping or move to completion

4. Error Handling and Recovery

def error_processing(state: AdvancedState) -> Dict[str, Any]:
    """Demonstrate error handling"""
    print("Error detected in query, handling error case")
    
    return {
        "status": ProcessingStatus.ERROR,
        "error": f"Error processing query: '{state['query']}' contains invalid parameters",
        "processing_path": "error"
    }


def attempt_recovery(state: AdvancedState) -> Dict[str, Any]:
    """Try to recover from errors"""
    print(f"Attempting to recover from error: {state['error']}")
    
    # Simulate a recovery attempt with 50% success rate
    if random.random() > 0.5:
        print("Recovery successful")
        return {
            "status": ProcessingStatus.SUCCESS,
            "error": None,
            "results": {"recovered": True, "original_error": state["error"]}
        }
    else:
        print("Recovery failed")
        return {"status": ProcessingStatus.ERROR}  # Error persists

These functions demonstrate error handling:

1. error_processing sets an error state and message
2. attempt_recovery tries to recover from the error with some probability of success

5. Building the Advanced Graph

def build_advanced_graph():
    """Build a graph demonstrating advanced patterns"""
    workflow = StateGraph(AdvancedState)
    
    # Add all nodes
    workflow.add_node("initialize", initialize)
    workflow.add_node("simple_processing", simple_processing)
    workflow.add_node("complex_processing", complex_processing)
    workflow.add_node("error_processing", error_processing)
    workflow.add_node("attempt_recovery", attempt_recovery)
    workflow.add_node("finalize", finalize_processing)
    
    # Start with initialization
    workflow.set_entry_point("initialize")
    
    # Branch based on the query type
    workflow.add_conditional_edges(
        "initialize",
        determine_processing_path,
        {
            "simple": "simple_processing",
            "complex": "complex_processing",
            "error_path": "error_processing"
        }
    )
    
    # For complex processing, we might need multiple iterations
    workflow.add_conditional_edges(
        "complex_processing",
        check_iteration_needed,
        {
            "continue": "complex_processing",  # Loop back for more processing
            "complete": "finalize"  # Move to finalization
        }
    )
    
    # Simple processing goes directly to finalization
    workflow.add_edge("simple_processing", "finalize")
    
    # For errors, try recovery
    workflow.add_edge("error_processing", "attempt_recovery")
    
    # After recovery attempt, always go to finalization
    workflow.add_edge("attempt_recovery", "finalize")
    
    # End after finalization
    workflow.add_edge("finalize", END)
    
    return workflow.compile()

This graph implements multiple advanced patterns:

1. Branching: Routes to different processing paths based on input
2. Looping: Complex processing can loop back to itself multiple times
3. Error Handling: Dedicated error path with recovery attempt
4. Multiple Pathways: Different execution paths converge at finalization

Key Advanced Patterns Explained

Pattern 1: Branching and Conditionals

Branching allows your agent to take different actions based on context:

workflow.add_conditional_edges(
    "source_node",
    decision_function,
    {
        "option1": "target_node_1",
        "option2": "target_node_2",
        "option3": "target_node_3"
    }
)

The decision function examines the state and returns a string key, which LangGraph uses to select the next node from the mapping dictionary.

Pattern 2: Loops and Recursion

Loops enable iterative processing or refinement:

workflow.add_conditional_edges(
    "processing_node",
    check_if_done,
    {
        "continue": "processing_node",  # Loop back to the same node
        "complete": "next_node"        # Move forward when done
    }
)

By routing back to the same node, you create a loop that continues until some condition is met.

Pattern 3: Error Handling

Robust agents need error handling:

try:
    # Attempt some operation
    result = some_operation(state)
    return {"status": "success", "result": result}
except Exception as e:
    # Handle the error
    return {"status": "error", "error": str(e)}

Combine this with conditional edges to create error recovery paths:

workflow.add_conditional_edges(
    "operation_node",
    check_status,
    {
        "success": "success_node",
        "error": "error_recovery_node"
    }
)

Pattern 4: Parallel Processing

For more advanced use cases, LangGraph also supports parallel processing:

## This is a conceptual example, not shown in the code
workflow.add_node("parallel_processing", parallel_processing_function)

In a parallel processing node, you could:

1. Split work into multiple tasks
2. Process them concurrently
3. Combine results back into the state

Practical Applications

These advanced patterns enable sophisticated agent behaviors:

• Multi-Step Reasoning: Implement chain-of-thought or tree-of-thought reasoning
• Iterative Refinement: Progressively improve outputs through multiple passes
• Fallback Strategies: Try alternative approaches when the primary method fails
• Dynamic Workflows: Adapt processing based on input complexity or user needs

Best Practices

1. Keep Decision Functions Simple: Decision functions should be clear and focused on a single routing decision
2. State-Based Decisions: Base routing decisions on the state, not external factors
3. Avoid Deep Nesting: If your graph has many nested conditions, consider restructuring
4. Limit Loop Iterations: Implement safeguards to prevent infinite loops
5. Comprehensive Error Handling: Plan for failures and provide meaningful error messages

Now that you understand advanced graph patterns, the next section will show you how to add tool support to your LangGraph agents, enabling them to interact with external systems and APIs.

Adding LLM-Powered Tool Support to Your LangGraph Agent

Why LLM-Powered Tool Selection Matters

While basic pattern matching can work for simple tool selection, LLMs offer significant advantages:

1. Natural Language Understanding: LLMs can understand complex, nuanced requests
2. Contextual Awareness: LLMs can consider the full conversation history
3. Parameter Extraction: LLMs can extract relevant parameters from natural language
4. Multi-Tool Reasoning: LLMs can decide when multiple tools need to be used together
5. Fallback Intelligence: LLMs can gracefully handle cases where no tool is appropriate

Setting Up OpenAI API Access

To use OpenAI models for tool selection, you need to set up your API key first:

Environment Variable Setup

## On macOS/Linux
export OPENAI_API_KEY="your-api-key-here"

## On Windows (Command Prompt)
set OPENAI_API_KEY=your-api-key-here

## On Windows (PowerShell)
$env:OPENAI_API_KEY="your-api-key-here"

Using a .env File

Alternatively, create a .env file in your project root:

OPENAI_API_KEY=your-api-key-here

And load it in your code:

from dotenv import load_dotenv
load_dotenv()

Code Walkthrough

Let’s examine the key components of our LLM-powered agent:

# Adding Tool Support with LLM Integration to Your LangGraph Agent

"""
This example demonstrates how to integrate tools with your LangGraph agent
and use LLMs for intelligent tool selection and response generation.
"""

import json
import os
import re
import datetime
from typing import TypedDict, List, Dict, Any, Optional, Union, Literal

try:
    from dotenv import load_dotenv
    load_dotenv()  # Load environment variables from .env file if present
except ImportError:
    print("Note: python-dotenv not installed. Using environment variables directly.")

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, FunctionMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END


# Define our state schema
class AgentState(TypedDict):
    """State for an agent with tool support"""
    messages: List[Union[HumanMessage, AIMessage, SystemMessage, FunctionMessage]]
    tool_calls: List[Dict[str, Any]]
    tool_results: List[Dict[str, Any]]
    current_step: Literal["user_input", "thinking", "tool_calling", "tool_processing", "response", "done"]


# Define some example tools
@tool
def search_web(query: str) -> str:
    """Search the web for information. Use this for general knowledge questions."""
    # In a real implementation, this would call a search API
    print(f"Searching the web for: {query}")
    return f"Search results for '{query}': This is simulated web search content about {query}."


@tool
def get_current_time(dummy_input: str = "") -> str:
    """Get the current time and date"""
    now = datetime.datetime.now()
    return f"Current time: {now.strftime('%Y-%m-%d %H:%M:%S')}"


@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression"""
    # Simple calculator for demonstration
    try:
        # Remove any non-math characters for safety
        sanitized = re.sub(r'[^0-9+\-*/().\s]', '', expression)
        result = eval(sanitized)
        return f"Result: {result}"
    except Exception as e:
        return f"Error calculating '{expression}': {str(e)}"


# Available tools collection
TOOLS = [search_web, get_current_time, calculate]


# Build an LLM-powered agent graph
def build_llm_agent_graph():
    """Build a graph for an agent with LLM-powered tool selection"""
    # Check for API key
    if not os.environ.get("OPENAI_API_KEY"):
        print("WARNING: OPENAI_API_KEY not found in environment variables.")
        print("Either set it with 'export OPENAI_API_KEY=your-key' or add it to a .env file.")
    
    # Initialize the graph
    workflow = StateGraph(AgentState)
    
    # Initialize the LLM
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    # Generate tool descriptions for the LLM
    tool_descriptions = "\n".join([f"- {tool.name}: {tool.description}" for tool in TOOLS])
    
    # Define node functions
    def process_user_input(state: AgentState) -> Dict[str, Any]:
        """Process the initial user input"""
        print("Processing user input...")
        return {"current_step": "thinking"}
    
    def llm_thinking(state: AgentState) -> Dict[str, Any]:
        """Use LLM to decide whether to use tools"""
        print("LLM thinking about the request...")
        messages = state["messages"]
        last_message = messages[-1].content if messages else ""
        
        # Get a list of tool names and their parameters
        tool_info = []
        for tool in TOOLS:
            params = []
            if tool.name == "search_web":
                params.append({"name": "query", "type": "string", "description": "The search query"})
            elif tool.name == "calculate":
                params.append({"name": "expression", "type": "string", "description": "The math expression to evaluate"})
            
            tool_info.append({
                "name": tool.name,
                "description": tool.description,
                "parameters": params
            })
        
        # Create a system prompt that explains available tools with their parameters
        tools_desc = ""
        for tool in tool_info:
            tools_desc += f"- {tool['name']}: {tool['description']}\n"
            if tool['parameters']:
                tools_desc += "  Parameters:\n"
                for param in tool['parameters']:
                    tools_desc += f"    - {param['name']} ({param['type']}): {param['description']}\n"
        
        system_prompt = f"""You are a helpful assistant with access to the following tools:

{tools_desc}

Analyze the user's request and decide if you should use a tool.
If a tool is appropriate, respond with a JSON object in this format:
{{
  "tool": "tool_name",
  "arguments": {{
    "parameter_name": "value"
  }}
}}

Make sure to use the exact parameter names specified for each tool.

If no tool is needed, respond with: {{
  "tool": null,
  "reasoning": "Explanation of why no tool is needed"
}}"""
        
        # Create a prompt for the LLM
        messages_for_llm = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": last_message}
        ]
        
        # Get the LLM's decision
        try:
            llm_response = llm.invoke(messages_for_llm)
            decision_text = llm_response.content
            print(f"LLM decision: {decision_text}")
            
            decision = json.loads(decision_text)
            
            if decision.get("tool"):
                # Tool was selected
                tool_name = decision["tool"]
                print(f"LLM decided to use tool: {tool_name}")
                return {
                    "current_step": "tool_calling",
                    "tool_calls": [{
                        "name": tool_name,
                        "arguments": decision.get("arguments", {})
                    }]
                }
            else:
                # No tool needed
                print("LLM decided no tool is needed")
                return {"current_step": "response"}
        except Exception as e:
            print(f"Error in LLM thinking: {str(e)}")
            return {"current_step": "response"}
    
    def call_tools(state: AgentState) -> Dict[str, Any]:
        """Call the tools based on the agent's decisions"""
        tool_calls = state.get("tool_calls", [])
        tool_results = []
        
        for call in tool_calls:
            tool_name = call.get("name")
            arguments = call.get("arguments", {})
            
            print(f"Calling tool: {tool_name} with arguments: {arguments}")
            
            # Manual tool dispatch for better control and error handling
            try:
                if tool_name == "search_web" and "query" in arguments:
                    result = search_web.invoke(arguments["query"])
                elif tool_name == "get_current_time":
                    result = get_current_time.invoke("")
                elif tool_name == "calculate" and "expression" in arguments:
                    result = calculate.invoke(arguments["expression"])
                else:
                    raise ValueError(f"Unknown tool or missing required arguments: {tool_name}")
                    
                tool_results.append({"name": tool_name, "result": result})
            except Exception as e:
                print(f"Error calling tool {tool_name}: {str(e)}")
                tool_results.append({"name": tool_name, "error": str(e)})
        
        # Add function messages to the conversation
        messages = state["messages"].copy()
        for result in tool_results:
            messages.append(FunctionMessage(
                content=result.get("result", f"Error: {result.get('error')}"),
                name=result.get("name")
            ))
        
        return {
            "tool_results": tool_results,
            "messages": messages,
            "current_step": "response"
        }
    
    def llm_response_generation(state: AgentState) -> Dict[str, Any]:
        """Generate a response using an LLM that incorporates tool results"""
        print("Generating LLM response...")
        messages = state["messages"].copy()
        
        # Create a system message
        system_message = "You are a helpful assistant. Generate a response to the user based on the conversation history and any tool results provided."
        
        # Convert our messages to the format expected by the LLM
        messages_for_llm = [
            {"role": "system", "content": system_message}
        ]
        
        for msg in messages:
            if isinstance(msg, HumanMessage):
                messages_for_llm.append({"role": "user", "content": msg.content})
            elif isinstance(msg, AIMessage):
                messages_for_llm.append({"role": "assistant", "content": msg.content})
            elif isinstance(msg, FunctionMessage):
                messages_for_llm.append({"role": "function", "name": msg.name, "content": msg.content})
        
        # Get the LLM's response
        llm_response = llm.invoke(messages_for_llm)
        
        # Add the AI's response to the messages
        messages.append(AIMessage(content=llm_response.content))
        
        return {
            "messages": messages,
            "current_step": "done"
        }
    
    # Add nodes to the graph
    workflow.add_node("process_input", process_user_input)
    workflow.add_node("thinking", llm_thinking)
    workflow.add_node("tool_calling", call_tools)
    workflow.add_node("response", llm_response_generation)
    
    # Add edges
    workflow.add_edge("process_input", "thinking")
    
    # Conditionally call tools or go straight to response
    workflow.add_conditional_edges(
        "thinking",
        lambda state: "tool_calling" if state["current_step"] == "tool_calling" else "response",
        {
            "tool_calling": "tool_calling",
            "response": "response"
        }
    )
    
    # After tool calling, generate a response
    workflow.add_edge("tool_calling", "response")
    
    # End after response
    workflow.add_edge("response", END)
    
    # Set the entry point
    workflow.set_entry_point("process_input")
    
    return workflow.compile()


def run_llm_agent(user_input: str):
    """Run the LLM-powered agent with a user input"""
    # Build the graph
    agent = build_llm_agent_graph()
    
    # Create initial state
    initial_state: AgentState = {
        "messages": [HumanMessage(content=user_input)],
        "tool_calls": [],
        "tool_results": [],
        "current_step": "user_input"
    }
    
    # Run the graph
    print(f"\nProcessing with LLM agent: '{user_input}'")
    print("-" * 50)
    final_state = agent.invoke(initial_state)
    
    # Display the final conversation
    print("\nConversation:")
    for msg in final_state["messages"]:
        sender = type(msg).__name__.replace("Message", "")
        if isinstance(msg, FunctionMessage):
            print(f"[{sender}: {msg.name}] {msg.content}")
        else:
            print(f"[{sender}] {msg.content}")
    print("-" * 50)
    
    return final_state


if __name__ == "__main__":
    # Check for API key
    if not os.environ.get("OPENAI_API_KEY"):
        print("\nWARNING: OPENAI_API_KEY not found in environment variables.")
        print("To run this example with LLM integration, set your API key with:")
        print("  export OPENAI_API_KEY=your-key-here  # On macOS/Linux")
        print("  set OPENAI_API_KEY=your-key-here     # On Windows cmd")
        print("  $env:OPENAI_API_KEY='your-key-here'  # On Windows PowerShell")
        print("\nAlternatively, you can create a .env file with OPENAI_API_KEY=your-key-here")
        print("\nRunning with basic pattern matching instead...")
        # Define a simple pattern-matching based agent when API key is not available
        def run_basic_agent(user_input: str):
            """Run a simple pattern-matching based agent as fallback"""
            print(f"\nProcessing with basic agent: '{user_input}'")
            print("-" * 50)
            
            # Simple response based on keywords
            if any(word in user_input.lower() for word in ["time", "date", "now"]):
                result = f"Current time: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
                print(f"[Basic Agent] {result}")
            elif any(word in user_input.lower() for word in ["search", "find", "information"]):
                print(f"[Basic Agent] Here's some basic information about {user_input}")
            else:
                print(f"[Basic Agent] I can help with that, but I don't have access to advanced capabilities right now.")
            
            print("-" * 50)
        
        # Run the basic agent with sample queries
        run_basic_agent("What time is it now?")
        run_basic_agent("Can you search for information about Python programming?")
    else:
        # Run the LLM-powered agent with different queries
        run_llm_agent("What time is it now?")
        run_llm_agent("Can you search for information about Python programming?")
        run_llm_agent("Calculate 23 * 45 for me")
        run_llm_agent("Tell me a joke about programming")

1. Tool Definitions

When defining tools for LLM usage, it’s important to include proper parameter definitions:

@tool
def search_web(query: str) -> str:
    """Search the web for information. Use this for general knowledge questions."""
    # In a real implementation, this would call a search API
    print(f"Searching the web for: {query}")
    return f"Search results for '{query}': This is simulated web search content about {query}."

@tool
def get_current_time(dummy_input: str = "") -> str:
    """Get the current time and date"""
    # Note: We include a dummy parameter to ensure compatibility with tool calling
    now = datetime.datetime.now()
    return f"Current time: {now.strftime('%Y-%m-%d %H:%M:%S')}"

## Note: In newer versions of langchain-core (0.1.47+), you should use the .invoke() method instead of calling the tool directly:
## result = get_current_time.invoke("")  # Instead of get_current_time("")

Note that even for tools like get_current_time that don’t need inputs, we include a dummy parameter to ensure compatibility with the tool calling mechanism.

2. LLM-Powered Tool Selection with Parameter Documentation

The key to reliable tool selection is providing explicit parameter information to the LLM:

## Get a list of tool names and their parameters
tool_info = []
for tool in TOOLS:
    params = []
    if tool.name == "search_web":
        params.append({"name": "query", "type": "string", "description": "The search query"})
    elif tool.name == "calculate":
        params.append({"name": "expression", "type": "string", "description": "The math expression to evaluate"})
    
    tool_info.append({
        "name": tool.name,
        "description": tool.description,
        "parameters": params
    })

## Create a system prompt that explains available tools with their parameters
tools_desc = ""
for tool in tool_info:
    tools_desc += f"- {tool['name']}: {tool['description']}\n"
    if tool['parameters']:
        tools_desc += "  Parameters:\n"
        for param in tool['parameters']:
            tools_desc += f"    - {param['name']} ({param['type']}): {param['description']}\n"

system_prompt = f"""You are a helpful assistant with access to the following tools:

{tools_desc}

Analyze the user's request and decide if you should use a tool.
If a tool is appropriate, respond with a JSON object in this format:
{{
  "tool": "tool_name",
  "arguments": {{
    "parameter_name": "value"
  }}
}}

Make sure to use the exact parameter names specified for each tool.

If no tool is needed, respond with: {{
  "tool": null,
  "reasoning": "Explanation of why no tool is needed"
}}"""

This approach explicitly documents each tool’s parameters, making it clear to the LLM what parameter names to use.

3. Reliable Tool Calling Implementation

For reliable tool calling, use direct function calls rather than dynamic dispatch:

def call_tools(state: AgentState) -> Dict[str, Any]:
    """Call the tools based on the agent's decisions"""
    tool_calls = state.get("tool_calls", [])
    tool_results = []
    
    for call in tool_calls:
        tool_name = call.get("name")
        arguments = call.get("arguments", {})
        
        print(f"Calling tool: {tool_name} with arguments: {arguments}")
        
        # Manual tool dispatch for better control and error handling
        try:
            if tool_name == "search_web" and "query" in arguments:
                # Use .invoke() method for newer versions of langchain-core
                result = search_web.invoke(arguments["query"])
            elif tool_name == "get_current_time":
                # Use .invoke() method with empty string parameter
                result = get_current_time.invoke("") 
            elif tool_name == "calculate" and "expression" in arguments:
                result = calculate.invoke(arguments["expression"])
            else:
                raise ValueError(f"Unknown tool or missing required arguments: {tool_name}")
                
            tool_results.append({"name": tool_name, "result": result})
        except Exception as e:
            print(f"Error calling tool {tool_name}: {str(e)}")
            tool_results.append({"name": tool_name, "error": str(e)})
    
    # Add function messages to the conversation
    messages = state["messages"].copy()
    for result in tool_results:
        messages.append(FunctionMessage(
            content=result.get("result", f"Error: {result.get('error')}"),
            name=result.get("name")
        ))
    
    return {
        "tool_results": tool_results,
        "messages": messages,
        "current_step": "response"
    }

This approach:

1. Uses explicit conditional checks for each tool type
2. Ensures correct parameter passing for each tool
3. Includes proper error handling
4. Avoids issues with dynamic function calls

4. LLM-Powered Response Generation

After calling tools, we use the LLM again to generate a coherent response:

def llm_response_generation(state: AgentState) -> Dict[str, Any]:
    """Generate a response using an LLM that incorporates tool results"""
    print("Generating LLM response...")
    messages = state["messages"].copy()
    
    # Create a system message
    system_message = "You are a helpful assistant. Generate a response to the user based on the conversation history and any tool results provided."
    
    # Convert our messages to the format expected by the LLM
    messages_for_llm = [
        {"role": "system", "content": system_message}
    ]
    
    for msg in messages:
        if isinstance(msg, HumanMessage):
            messages_for_llm.append({"role": "user", "content": msg.content})
        elif isinstance(msg, AIMessage):
            messages_for_llm.append({"role": "assistant", "content": msg.content})
        elif isinstance(msg, FunctionMessage):
            messages_for_llm.append({"role": "function", "name": msg.name, "content": msg.content})
    
    # Get the LLM's response
    llm_response = llm.invoke(messages_for_llm)
    
    # Add the AI's response to the messages
    messages.append(AIMessage(content=llm_response.content))
    
    return {
        "messages": messages,
        "current_step": "done"
    }

Key Advantages of LLM-Powered Tool Calling

1. Natural Parameter Extraction

The LLM can extract parameters from natural language without explicit regex or parsing:

User: "Calculate the square root of 144"

LLM Decision: {
  "tool": "calculate",
  "arguments": {
    "expression": "sqrt(144)"
  }
}

2. Contextual Understanding

The LLM understands when a tool is appropriate based on context:

User: "I need to know what time it is in Tokyo"

LLM Decision: {
  "tool": "search_web",
  "arguments": {
    "query": "current time in Tokyo Japan"
  }
}

3. Intelligent Response Composition

The LLM can compose responses that naturally incorporate tool results:

User: "What's 25 times 16?"

Tool Result: "Result: 400"

LLM Response: "The calculation of 25 times 16 equals 400."

Best Practices for LLM Tool Integration

1. Structured Output

Always request structured output (like JSON) from the LLM for tool selection. This makes parsing easier and more reliable.

2. Explicit Parameter Documentation

Clearly document parameter names and types for each tool in your LLM prompts:

## In your system prompt
tools_desc += f"- {tool['name']}: {tool['description']}\n"
if tool['parameters']:
    tools_desc += "  Parameters:\n"
    for param in tool['parameters']:
        tools_desc += f"    - {param['name']} ({param['type']}): {param['description']}\n"

3. Tool Function Design

Design tool functions with appropriate parameters, even if they don’t need inputs:

@tool
def get_current_time(dummy_input: str = "") -> str:
    """Get the current time and date"""
    # Dummy parameter ensures compatibility
    ...

4. Use .invoke() Method for Tool Calling

In newer versions of langchain-core (0.1.47+), use the .invoke() method instead of calling tools directly:

## Deprecated approach
result = get_current_time("")

## Recommended approach
result = get_current_time.invoke("")

5. Direct Tool Dispatch

Use explicit function calls rather than dynamic dispatch for reliability:

## More reliable approach
if tool_name == "search_web" and "query" in arguments:
    result = search_web.invoke(arguments["query"])
elif tool_name == "get_current_time":
    result = get_current_time.invoke("")

6. Error Handling

Implement robust error handling for both LLM calls and tool execution:

try:
    llm_response = llm.invoke(messages_for_llm)
    # Process response
except Exception as e:
    print(f"Error calling LLM: {str(e)}")
    # Fallback behavior

7. API Key Management

Never hardcode API keys. Use environment variables or secure vaults for production applications.

8. Fallback Mechanisms

Implement fallbacks for when the LLM or tools fail:

if not os.environ.get("OPENAI_API_KEY"):
    print("API key not found. Using pattern matching fallback...")
    return pattern_matching_tool_selection(state)

Advanced LLM Tool Integration

Multi-Tool Sequences

For more complex scenarios, you can enhance the system prompt to allow the LLM to call multiple tools in sequence:

system_prompt = """You can use multiple tools if needed. Respond with a JSON array:
[
  {"tool": "tool1", "arguments": {...}},
  {"tool": "tool2", "arguments": {...}}
]
"""

Tool Selection with Reasoning

You can ask the LLM to explain its tool selection reasoning:

system_prompt = """Respond with JSON:
{
  "reasoning": "Your step-by-step thought process",
  "tool": "selected_tool",
  "arguments": {...}
}
"""

Model Selection

Different models have different capabilities for tool use:

• GPT-4: Best for complex tool selection and parameter extraction
• GPT-3.5-Turbo: Good for simpler tools, more cost-effective
• Open source models: Capabilities vary; may require more structured prompting

Tool Use Summary

LLM-powered tool selection takes your agents to the next level by enabling them to intelligently choose and use tools based on natural language input. By combining the reasoning capabilities of LLMs with the specific functionality of tools, you can create agents that are both flexible in understanding requests and precise in taking actions.

In the next section, we’ll conclude our tutorial and explore next steps for continuing your LangGraph journey.

Conclusion and Next Steps

What We’ve Learned

Congratulations on completing the LangGraph Basics tutorial! Throughout this journey, we’ve covered the essential components of building AI agents with LangGraph:

1. Installation and Setup: We started by setting up our environment with uv, a modern Python package manager that provides faster and more reliable dependency management.
2. Nodes and Edges: We learned that nodes are the fundamental building blocks in LangGraph, representing discrete processing steps, while edges define the flow between nodes.
3. Schema Management: We explored how to define, validate, and transform state using both TypedDict and Pydantic schemas, ensuring type safety and clear data structures.
4. Building Simple Graphs: We built functional graphs that demonstrate how to manipulate state data between nodes, creating agents that maintain context throughout execution.
5. Advanced Patterns: We implemented sophisticated control flows including branching, loops, and error handling, enabling complex agent behaviors.
6. Tool Support: We added external capabilities to our agents by integrating tools, allowing them to interact with external systems and APIs.

The Power of LangGraph

LangGraph provides several key advantages for building AI agents:

1. Explicit Flow Control: Graph-based architecture makes agent logic visible and maintainable.
2. State Management: Built-in state handling ensures consistent data flow throughout your agent.
3. Modularity: Discrete nodes enable easy testing, reuse, and composition.
4. Flexibility: Support for various patterns from simple sequences to complex multi-agent systems.
5. Integration: Seamless compatibility with the broader LangChain ecosystem.

Where to Go Next

Your journey with LangGraph is just beginning. Here are some advanced topics to explore:

1. Advanced Agent Architectures

• ReAct Agents: Implement the Reasoning and Acting pattern using LangGraph
• Multi-Agent Systems: Create collaborative agent networks where multiple agents work together
• Memory Systems: Add sophisticated memory mechanisms to your agents
• Planning Agents: Build agents that can formulate and execute plans

2. Integration with LLMs

• Prompt Engineering: Develop effective prompts for different agent nodes
• Model Selection: Choose appropriate models for different agent tasks
• Output Parsing: Implement robust parsing of LLM outputs
• Chain of Thought: Guide LLMs through complex reasoning processes

3. Tool Ecosystems

• Custom Tool Development: Build specialized tools for your domain
• Tool Orchestration: Manage complex tool interactions
• API Integration: Connect to external services and platforms
• Database Tools: Enable agents to query and update databases

4. Production Deployment

• Monitoring: Track agent performance and behavior
• Scaling: Handle increased load and concurrent users
• Security: Ensure proper permissions and data handling
• Versioning: Manage agent versions and updates

5. Specialized Applications

• RAG Systems: Build retrieval-augmented generation systems
• Autonomous Agents: Create agents that operate independently
• Domain-Specific Agents: Tailor agents for specific industries or use cases
• Conversational Systems: Build advanced dialogue managers

Final Thoughts

LangGraph represents a significant step forward in agent development, providing a structured, maintainable approach to building complex AI systems. The graph-based paradigm aligns perfectly with how we conceptualize agent behavior—as a series of interconnected steps with dynamic decision-making.

As LLMs continue to evolve, the ability to orchestrate their capabilities through well-designed agent architectures becomes increasingly valuable. LangGraph gives you the tools to build these architectures in a way that’s both powerful and approachable.

Remember that effective agent development is an iterative process. Start simple, test thoroughly, and gradually add complexity as needed. The patterns and practices you’ve learned in this tutorial provide a solid foundation for building sophisticated AI agents that can solve real-world problems.