Tools and Structured Output in LangGraph: Building Reliable AI Agents

Building reliable AI agents requires more than just chat completions. You need tools that can interact with external systems and structured output that ensures predictable, type-safe responses. LangGraph provides an elegant framework for orchestrating these complex interactions while maintaining full control over your agent's behavior.

In this guide, we'll explore how to leverage tools and structured output in LangGraph to build production-ready AI agents that can handle real-world tasks with confidence.

What is LangGraph?

LangGraph is a library for building stateful, multi-actor applications with LLMs, built on top of LangChain. Unlike simple chat interfaces, LangGraph allows you to create complex workflows where AI agents can use tools, make decisions, and maintain state across multiple interactions.

Think of it as a state machine for AI agents where each node represents a specific action or decision point, and edges define the flow between these states.

Setting Up Your First Tool

Let's start with a simple example that demonstrates how to integrate tools into a LangGraph workflow:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

@tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    # In a real implementation, you'd call a weather API
    return f"The weather in {location} is sunny with a temperature of 22°C"

@tool
def calculate_sum(a: int, b: int) -> int:
    """Calculate the sum of two numbers."""
    return a + b

# Initialize the LLM with tools
llm = ChatOpenAI(model="gpt-4")
tools = [get_weather, calculate_sum]
llm_with_tools = llm.bind_tools(tools)

# Create the agent node
def agent(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# Create the tool node
tool_node = ToolNode(tools)

# Build the graph
workflow = StateGraph(MessagesState)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")

# Add conditional edges
workflow.add_conditional_edges(
    "agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "__end__",
    {"tools": "tools", "__end__": "__end__"}
)
workflow.add_edge("tools", "agent")

app = workflow.compile()

This basic setup creates an agent that can use tools when needed and continues the conversation based on the results.

Structured Output with Pydantic

While tools handle external interactions, structured output ensures your agent returns data in a predictable format. Here's how to combine both:

from pydantic import BaseModel, Field
from typing import List, Optional
from langchain_core.messages import HumanMessage

class WeatherReport(BaseModel):
    """Structured weather report"""
    location: str = Field(description="The location queried")
    temperature: int = Field(description="Temperature in Celsius")
    conditions: str = Field(description="Weather conditions")
    recommendations: List[str] = Field(description="Clothing or activity recommendations")

class TravelPlanner(BaseModel):
    """Complete travel planning response"""
    destination: str
    weather_info: WeatherReport
    recommended_activities: List[str]
    estimated_budget: Optional[int] = None

@tool
def get_detailed_weather(location: str) -> WeatherReport:
    """Get detailed weather information with recommendations."""
    return WeatherReport(
        location=location,
        temperature=22,
        conditions="sunny",
        recommendations=["Light jacket for evening", "Sunglasses recommended"]
    )

# Use structured output with the LLM
def planning_agent(state: MessagesState):
    response = llm_with_tools.with_structured_output(TravelPlanner).invoke(
        state["messages"]
    )
    return {"messages": [response]}

Building a Complex Multi-Tool Workflow

Real-world agents often need to orchestrate multiple tools and make decisions based on intermediate results. Here's a more sophisticated example:

from enum import Enum
from datetime import datetime

class TaskType(str, Enum):
    RESEARCH = "research"
    CALCULATION = "calculation"
    COMMUNICATION = "communication"

class TaskResult(BaseModel):
    task_type: TaskType
    success: bool
    result: str
    confidence: float = Field(ge=0.0, le=1.0)
    next_steps: Optional[List[str]] = None

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    # Mock implementation
    return f"Search results for '{query}': Found 5 relevant articles about the topic."

@tool
def send_email(recipient: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    # Mock implementation
    return f"Email sent to {recipient} with subject '{subject}'"

@tool
def analyze_data(data: str) -> dict:
    """Analyze data and return insights."""
    # Mock implementation
    return {
        "summary": "Data shows positive trend",
        "key_metrics": {"growth": 15.2, "accuracy": 0.94},
        "recommendations": ["Continue current strategy", "Monitor weekly"]
    }

class ResearchAgent:
    def __init__(self):
        self.tools = [web_search, analyze_data, send_email]
        self.llm = ChatOpenAI(model="gpt-4").bind_tools(self.tools)

    def create_workflow(self):
        workflow = StateGraph(MessagesState)

        # Add nodes for different stages
        workflow.add_node("planner", self.planner_node)
        workflow.add_node("researcher", self.researcher_node)
        workflow.add_node("analyzer", self.analyzer_node)
        workflow.add_node("communicator", self.communicator_node)
        workflow.add_node("tools", ToolNode(self.tools))

        # Define the flow
        workflow.set_entry_point("planner")
        workflow.add_conditional_edges(
            "planner",
            self.route_next_step,
            {
                "research": "researcher",
                "analyze": "analyzer",
                "communicate": "communicator",
                "__end__": "__end__"
            }
        )

        # Add tool integration for each node
        for node in ["researcher", "analyzer", "communicator"]:
            workflow.add_conditional_edges(
                node,
                self.should_use_tools,
                {"tools": "tools", "continue": "planner"}
            )

        workflow.add_edge("tools", "planner")

        return workflow.compile()

    def planner_node(self, state: MessagesState):
        """Determine the next action to take."""
        system_prompt = """You are a research planning agent. Analyze the conversation
        and determine what needs to be done next. Return a structured response indicating
        the next step."""

        messages = [{"role": "system", "content": system_prompt}] + state["messages"]
        response = self.llm.with_structured_output(TaskResult).invoke(messages)
        return {"messages": [response]}

Error Handling and Validation

Production agents need robust error handling. Here's how to implement it:

from typing import Union
from langchain_core.messages import AIMessage

class ErrorResponse(BaseModel):
    error_type: str
    message: str
    retry_possible: bool
    suggested_action: Optional[str] = None

class SuccessResponse(BaseModel):
    data: dict
    timestamp: datetime
    confidence: float

@tool
def safe_api_call(endpoint: str, params: dict) -> Union[SuccessResponse, ErrorResponse]:
    """Make a safe API call with error handling."""
    try:
        # Simulate API call
        if not endpoint.startswith("https://"):
            raise ValueError("Invalid endpoint URL")

        result = {"status": "success", "data": params}
        return SuccessResponse(
            data=result,
            timestamp=datetime.now(),
            confidence=0.95
        )
    except Exception as e:
        return ErrorResponse(
            error_type=type(e).__name__,
            message=str(e),
            retry_possible=True,
            suggested_action="Check endpoint URL format"
        )

def error_handling_agent(state: MessagesState):
    """Agent with built-in error handling."""
    try:
        response = llm_with_tools.invoke(state["messages"])

        # Check if the response contains tool calls
        if response.tool_calls:
            # Validate tool calls before execution
            for tool_call in response.tool_calls:
                if not tool_call.get("args"):
                    return {"messages": [AIMessage(
                        content="Tool call missing required arguments",
                        additional_kwargs={"error": True}
                    )]}

        return {"messages": [response]}

    except Exception as e:
        error_msg = ErrorResponse(
            error_type=type(e).__name__,
            message=str(e),
            retry_possible=True,
            suggested_action="Please rephrase your request"
        )
        return {"messages": [AIMessage(content=error_msg.json())]}

Best Practices for Production

When building production-ready agents with LangGraph, keep these principles in mind:

1. Type Safety First

Always use Pydantic models for structured output. This ensures your agent's responses are predictable and can be safely consumed by other systems:

class ProductionResponse(BaseModel):
    """Production-ready response format"""
    success: bool
    data: Optional[dict] = None
    errors: List[str] = Field(default_factory=list)
    metadata: dict = Field(default_factory=dict)

    class Config:
        extra = "forbid"  # Prevent unexpected fields

2. Implement Circuit Breakers

Protect your system from cascading failures:

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half-open"
            else:
                raise Exception("Circuit breaker is open")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise e

3. Observability and Monitoring

Add comprehensive logging and metrics:

import logging
from functools import wraps

def monitor_tool_calls(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start_time
            logging.info(f"Tool {func.__name__} succeeded in {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start_time
            logging.error(f"Tool {func.__name__} failed after {duration:.2f}s: {e}")
            raise
    return wrapper

@monitor_tool_calls
@tool
def monitored_api_call(endpoint: str) -> dict:
    """API call with monitoring."""
    # Implementation here
    pass

Testing Your LangGraph Agents

Testing AI agents requires special considerations:

import pytest
from unittest.mock import Mock, patch

class TestWeatherAgent:
    @pytest.fixture
    def agent(self):
        return WeatherAgent()

    def test_successful_weather_query(self, agent):
        """Test successful weather information retrieval."""
        with patch('your_module.get_weather') as mock_weather:
            mock_weather.return_value = WeatherReport(
                location="London",
                temperature=15,
                conditions="rainy",
                recommendations=["Bring an umbrella"]
            )

            result = agent.process_query("What's the weather in London?")
            assert result.success
            assert result.data.location == "London"
            mock_weather.assert_called_once_with("London")

    def test_error_handling(self, agent):
        """Test error handling when tools fail."""
        with patch('your_module.get_weather') as mock_weather:
            mock_weather.side_effect = Exception("API unavailable")

            result = agent.process_query("What's the weather in London?")
            assert not result.success
            assert "API unavailable" in result.errors

Advanced Patterns

Conditional Tool Selection

Sometimes you need to dynamically choose which tools to make available:

def get_available_tools(user_role: str, context: dict) -> List:
    """Return tools based on user permissions and context."""
    base_tools = [get_weather, calculate_sum]

    if user_role == "admin":
        base_tools.extend([send_email, delete_data])

    if context.get("location") == "internal":
        base_tools.append(access_internal_db)

    return base_tools

def dynamic_agent(state: MessagesState):
    user_context = state.get("user_context", {})
    available_tools = get_available_tools(
        user_context.get("role", "user"),
        user_context
    )

    llm_with_dynamic_tools = llm.bind_tools(available_tools)
    return {"messages": [llm_with_dynamic_tools.invoke(state["messages"])]}

Streaming and Real-time Updates

For long-running tasks, implement streaming responses:

async def streaming_agent(state: MessagesState):
    """Agent that streams responses for long-running operations."""
    async for chunk in llm_with_tools.astream(state["messages"]):
        if chunk.content:
            yield {"messages": [chunk]}
        elif chunk.tool_calls:
            # Handle tool calls
            tool_results = await execute_tools_async(chunk.tool_calls)
            yield {"messages": tool_results}

LangGraph's combination of tools and structured output provides a powerful foundation for building AI agents that are both capable and reliable. By following these patterns and best practices, you can create production-ready systems that handle complex workflows while maintaining the predictability and safety that real applications demand.

The key is to start simple, add complexity gradually, and always prioritize type safety and error handling. Your users will thank you for building agents that work reliably, and your future self will thank you for building systems that are easy to maintain and extend.