Build Tool-Calling Warehouse Agents with PydanticAI and LangGraph

"""
Production implementation for Building Tool-Calling Warehouse Agents.
Provides secure, scalable operations using PydanticAI and LangGraph.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import time
from pydantic import BaseModel, ValidationError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class for environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL')
    api_key: str = os.getenv('API_KEY')

class InputData(BaseModel):
    """
    Data model for input validation using PydanticAI.
    """
    id: str
    name: str
    warehouse_id: str

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    try:
        InputData(**data)  # Validate using Pydantic
        return True
    except ValidationError as e:
        logger.error(f'Validation error: {e}')  # Log validation errors
        raise ValueError('Invalid input data')

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize fields in input data.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    sanitized = {key: value.strip() for key, value in data.items()}
    return sanitized  # Strip whitespace from all fields

async def fetch_data(warehouse_id: str) -> Dict[str, Any]:
    """Fetch data from the warehouse.
    
    Args:
        warehouse_id: ID of the warehouse
    Returns:
        Fetched data as a dictionary
    Raises:
        RuntimeError: If API call fails
    """
    url = f'https://api.example.com/warehouses/{warehouse_id}'
    headers = {'Authorization': f'Bearer {Config.api_key}'}
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        logger.error('Failed to fetch data')
        raise RuntimeError('API request failed')
    return response.json()  # Return fetched data

async def process_batch(data: List[Dict[str, Any]]) -> None:
    """Process a batch of data records.
    
    Args:
        data: List of records to process
    """
    for record in data:
        logger.info(f'Processing record: {record}')  # Log each record processing
        # Add processing logic here (e.g., transformation, saving to DB)

def retry_request(func, retries: int = 3, delay: int = 2) -> Any:
    """Retry a function call with exponential backoff.
    
    Args:
        func: Function to call
        retries: Number of retry attempts
        delay: Initial delay in seconds
    Returns:
        Result of the function call
    """
    for attempt in range(retries):
        try:
            return func()  # Call the function
        except Exception as e:
            logger.warning(f'Attempt {attempt + 1} failed: {e}')
            time.sleep(delay)
            delay *= 2  # Exponential backoff
    raise RuntimeError('Max retries exceeded')  # Raise error if all attempts fail

async def save_to_db(data: Dict[str, Any]) -> None:
    """Save processed data to the database.
    
    Args:
        data: Data to save
    """
    # Database saving logic goes here
    logger.info('Data saved to database')  # Log successful save

async def format_output(data: Dict[str, Any]) -> str:
    """Format output data for presentation.
    
    Args:
        data: Data to format
    Returns:
        Formatted string output
    """
    return f"Record: {data['name']} (ID: {data['id']})"  # Simple formatting

async def handle_errors(func):
    """Error handling decorator.
    
    Args:
        func: Function to decorate
    Returns:
        Wrapped function
    """
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')  # Log error
            return None  # Return None on error
    return wrapper

class WarehouseAgent:
    """Main orchestrator for warehouse operations.
    """
    async def execute(self, input_data: Dict[str, Any]) -> None:
        """Main workflow for processing warehouse data.
        
        Args:
            input_data: Input data to process
        """
        logger.info('Starting execution')
        try:
            validated_data = await validate_input(input_data)  # Validate input
            sanitized_data = await sanitize_fields(validated_data)  # Sanitize input
            warehouse_data = await retry_request(lambda: fetch_data(sanitized_data['warehouse_id']))  # Fetch data
            await process_batch([warehouse_data])  # Process fetched data
            await save_to_db(warehouse_data)  # Save processed data
            output = await format_output(warehouse_data)  # Format output
            logger.info(f'Output: {output}')  # Log output
        except Exception as e:
            logger.error(f'Execution failed: {e}')  # Log execution failure

if __name__ == '__main__':
    # Example usage
    agent = WarehouseAgent()
    example_data = {'id': '1', 'name': 'Example Item', 'warehouse_id': 'W1'}
    import asyncio
    asyncio.run(agent.execute(example_data))