Run Edge LLMs on IoT Devices with Ollama and llama.cpp

                      
                     
"""
Production implementation for running edge LLMs on IoT devices using Ollama and llama.cpp.
Enables efficient, secure inference on constrained hardware.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import time

# Set up logging for the application
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Holds configuration variables.
    Loads values from environment variables for flexibility.
    """
    model_path: str = os.getenv('MODEL_PATH', 'models/llama.cpp')
    api_endpoint: str = os.getenv('API_ENDPOINT', 'http://localhost:8000/api')

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate the input data for the model.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not isinstance(data, dict):
        raise ValueError('Input must be a dictionary.')
    if 'text' not in data:
        raise ValueError('Missing required field: text')
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent security issues.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized input data
    """
    sanitized_data = {k: str(v).strip() for k, v in data.items()}
    return sanitized_data

def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from the specified API endpoint.
    
    Args:
        api_url: The API endpoint to fetch data from
    Returns:
        List of records fetched from the API
    Raises:
        ConnectionError: If the API call fails
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as e:
        logger.error(f'HTTP error occurred: {e}')
        raise ConnectionError('Failed to fetch data from API.')

def call_api(data: Dict[str, Any]) -> Dict[str, Any]:
    """Call the API with the provided data.
    
    Args:
        data: Input data to send to the API
    Returns:
        API response as a dictionary
    Raises:
        RuntimeError: If API call fails
    """
    try:
        response = requests.post(Config.api_endpoint, json=data)
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as e:
        logger.error(f'Error calling API: {e}')
        raise RuntimeError('API call failed.')

def process_batch(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of data records.
    
    Args:
        data: List of data records to process
    Returns:
        List of processed results
    """
    results = []
    for record in data:
        sanitized = sanitize_fields(record)
        if validate_input(sanitized):
            result = call_api(sanitized)
            results.append(result)
    return results

def aggregate_metrics(results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics from the results.
    
    Args:
        results: The processed results from API call
    Returns:
        Aggregated metrics as a dictionary
    """
    metrics = {'success': 0, 'failure': 0}
    for result in results:
        if result.get('status') == 'success':
            metrics['success'] += 1
        else:
            metrics['failure'] += 1
    return metrics

def format_output(metrics: Dict[str, Any]) -> str:
    """Format the output metrics for display.
    
    Args:
        metrics: The metrics to format
    Returns:
        Formatted string of metrics
    """
    return f"Success: {metrics['success']}, Failure: {metrics['failure']}"

def handle_errors(func):
    """Decorator for handling errors in functions.
    
    Args:
        func: The function to wrap
    Returns:
        Wrapped function with error handling
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')
            return None
    return wrapper

class EdgeLLMService:
    """Main orchestrator class for the Edge LLM service.
    Handles the workflow of processing inputs and generating outputs.
    """
    def __init__(self, config: Config):
        self.config = config

    @handle_errors
    def run(self, input_data: List[Dict[str, Any]]) -> None:
        """Run the main workflow of the Edge LLM service.
        
        Args:
            input_data: The data to process
        """
        results = process_batch(input_data)
        metrics = aggregate_metrics(results)
        output = format_output(metrics)
        logger.info(output)

if __name__ == '__main__':
    # Example usage
    config = Config()
    service = EdgeLLMService(config)
    data_to_process = [{'text': 'Hello, world!'}, {'text': 'How are you?'}]
    service.run(data_to_process)