Run Hybrid LLM and ML Pipelines on Edge Gateways with Ollama and ONNX Runtime

                      
                     
"""
Production implementation for running Hybrid LLM and ML pipelines on edge gateways using Ollama and ONNX Runtime.
Provides secure, scalable operations with efficient data handling.
"""

from typing import Dict, Any, List
import os
import logging
import time
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, constr

# Logger setup to track application flow and errors
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """Configuration class to manage environment variables."""
    database_url: str = os.getenv('DATABASE_URL')
    ollama_api_url: str = os.getenv('OLLAMA_API_URL')

class InputData(BaseModel):
    """Model for input data validation using Pydantic."""
    id: constr(min_length=1)
    data: List[Dict[str, Any]]

async def validate_input(data: InputData) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not data.data:
        raise ValueError('Data cannot be empty')
    return True

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

async def call_ollama_api(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Call the Ollama API and return the response.
    
    Args:
        payload: The data to send to the API
    Returns:
        The response from the API
    Raises:
        HTTPException: If API call fails
    """
    try:
        response = requests.post(Config.ollama_api_url, json=payload)
        response.raise_for_status()  # Raise error for bad responses
        logger.info('Ollama API called successfully')
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'API call failed: {e}')
        raise HTTPException(status_code=500, detail='API call failed')

async def process_batch(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of data through the ML pipeline.
    
    Args:
        data: List of records to process
    Returns:
        List of processed records
    """
    results = []
    for record in data:
        sanitized_record = await sanitize_fields(record)  # Sanitize inputs
        result = await call_ollama_api(sanitized_record)  # Call API
        results.append(result)  # Collect results
    logger.info('Batch processed successfully')
    return results

async def aggregate_metrics(results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics from the processed results.
    
    Args:
        results: List of results from the API
    Returns:
        Aggregated metrics
    """
    metrics = {'success_count': 0, 'failure_count': 0}
    for result in results:
        if result.get('success'):
            metrics['success_count'] += 1
        else:
            metrics['failure_count'] += 1
    logger.info('Metrics aggregated')
    return metrics

app = FastAPI()

@app.post('/process')
async def process_data(input_data: InputData):
    """Endpoint to process data.
    
    Args:
        input_data: Input data model
    Returns:
        Processed results and metrics
    Raises:
        HTTPException: If validation or processing fails
    """
    try:
        await validate_input(input_data)  # Validate input
        results = await process_batch(input_data.data)  # Process data
        metrics = await aggregate_metrics(results)  # Aggregate results
    except ValueError as e:
        logger.error(f'Validation error: {e}')
        raise HTTPException(status_code=400, detail=str(e))  # Bad Input
    except Exception as e:
        logger.error(f'Processing error: {e}')
        raise HTTPException(status_code=500, detail='Processing failed')  # Internal Server Error
    return {'results': results, 'metrics': metrics}  # Return results

if __name__ == '__main__':
    # Example usage
    # Run your FastAPI app with: uvicorn main:app --reload
    logger.info('Starting the FastAPI application...')
    pass