Trace LLM Inference Pipelines for Factory AI with Langfuse and BentoML

"""
Production implementation for tracing LLM inference pipelines for Factory AI.
Provides secure and scalable operations using Langfuse and BentoML.
"""

from typing import Dict, Any, List
import os
import logging
import time
import requests
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Configure logging for debugging and monitoring
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///factory_ai.db')
    langfuse_api_key: str = os.getenv('LANGFUSE_API_KEY')

# Create a database engine with connection pooling
engine = create_engine(Config.database_url, pool_size=20, max_overflow=0)
Session = sessionmaker(bind=engine)

# Helper function to validate input data
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
    """
    if 'input' not in data:
        raise ValueError('Missing required input field')  # Validation error
    return True  # Data is valid

# Helper function to sanitize input fields
def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields for security.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    return {key: str(value).strip() for key, value in data.items()}  # Remove leading/trailing spaces

# Helper function to fetch data from Langfuse API
async def fetch_data(endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """Fetch data from Langfuse API.
    
    Args:
        endpoint: API endpoint to call
        params: Query parameters
    Returns:
        Response data as dict
    Raises:
        Exception: If API call fails
    """
    try:
        response = requests.get(endpoint, headers={'Authorization': f'Bearer {Config.langfuse_api_key}'}, params=params)
        response.raise_for_status()  # Raise error for bad responses
        return response.json()  # Return JSON response
    except requests.RequestException as e:
        logger.error(f'Error fetching data from Langfuse: {e}')  # Log error
        raise RuntimeError('Failed to fetch data from Langfuse')  # Raise runtime error

# Helper function to process a batch of records
async def process_batch(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of input data.
    
    Args:
        data: List of records to process
    Returns:
        Processed data list
    """
    processed_data = []  # List to hold processed data
    for record in data:
        # Simulated processing (replace with actual logic)
        processed_data.append({'id': record['id'], 'status': 'processed'})  # Example transformation
    return processed_data  # Return processed data

# Helper function to log metrics
def log_metrics(metrics: Dict[str, Any]) -> None:
    """Log metrics for monitoring.
    
    Args:
        metrics: Dictionary of metrics to log
    """
    logger.info(f'Metrics: {metrics}')  # Log metrics at info level

# Main class to orchestrate inference pipeline
class InferencePipeline:
    def __init__(self):
        self.session = Session()  # Database session

    async def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Run the inference pipeline.
        
        Args:
            input_data: Input data for inference
        Returns:
            Results of the inference
        """
        try:
            await validate_input(input_data)  # Validate input
            sanitized_data = sanitize_fields(input_data)  # Sanitize input data
            langfuse_data = await fetch_data('https://api.langfuse.com/data', {'query': sanitized_data['input']})  # Fetch data from Langfuse
            processed_data = await process_batch(langfuse_data['results'])  # Process the data
            log_metrics({'total_processed': len(processed_data)})  # Log metrics
            return {'success': True, 'data': processed_data}  # Return processed results
        except ValueError as ve:
            logger.error(f'Validation error: {ve}')  # Log validation error
            return {'success': False, 'error': str(ve)}  # Return error response
        except RuntimeError as re:
            logger.error(f'Runtime error: {re}')  # Log runtime error
            return {'success': False, 'error': str(re)}  # Return error response
        except Exception as e:
            logger.error(f'Unexpected error: {e}')  # Log unexpected errors
            return {'success': False, 'error': 'An unexpected error occurred'}  # Return generic error

    def __enter__(self):
        return self  # Context manager entry

    def __exit__(self, exc_type, exc_value, traceback):
        self.session.close()  # Ensure session is closed

# Main block for example usage
if __name__ == '__main__':
    # Example input data
    example_input = {'input': 'Sample input for LLM inference.'}
    with InferencePipeline() as pipeline:
        result = pipeline.run(example_input)  # Run the inference pipeline
        logger.info(f'Result: {result}')  # Log the result