Comprehensive comparison for Model Inference & Serving technology in AI / Machine Learning applications

import onnxruntime as ort
import numpy as np
from typing import Dict, List, Optional
import logging
from pathlib import Path
import json
from dataclasses import dataclass

@dataclass
class PredictionResult:
    """Structure for model prediction results"""
    prediction: str
    confidence: float
    all_scores: Dict[str, float]

class ProductCategoryClassifier:
    """Production-ready ONNX Runtime classifier for e-commerce product categorization"""
    
    def __init__(self, model_path: str, labels_path: str, providers: Optional[List[str]] = None):
        """
        Initialize the classifier with ONNX model
        
        Args:
            model_path: Path to ONNX model file
            labels_path: Path to JSON file containing category labels
            providers: List of execution providers (e.g., ['CUDAExecutionProvider', 'CPUExecutionProvider'])
        """
        self.logger = logging.getLogger(__name__)
        
        if not Path(model_path).exists():
            raise FileNotFoundError(f"Model file not found: {model_path}")
        
        if not Path(labels_path).exists():
            raise FileNotFoundError(f"Labels file not found: {labels_path}")
        
        try:
            # Set execution providers with fallback to CPU
            self.providers = providers or ['CPUExecutionProvider']
            
            # Create inference session with optimization
            sess_options = ort.SessionOptions()
            sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
            sess_options.intra_op_num_threads = 4
            
            self.session = ort.InferenceSession(
                model_path,
                sess_options=sess_options,
                providers=self.providers
            )
            
            # Load category labels
            with open(labels_path, 'r') as f:
                self.labels = json.load(f)
            
            # Get model metadata
            self.input_name = self.session.get_inputs()[0].name
            self.output_name = self.session.get_outputs()[0].name
            self.input_shape = self.session.get_inputs()[0].shape
            
            self.logger.info(f"Model loaded successfully with providers: {self.session.get_providers()}")
            
        except Exception as e:
            self.logger.error(f"Failed to initialize model: {str(e)}")
            raise
    
    def preprocess(self, features: np.ndarray) -> np.ndarray:
        """Preprocess input features for model inference"""
        if not isinstance(features, np.ndarray):
            features = np.array(features)
        
        # Ensure correct dtype
        features = features.astype(np.float32)
        
        # Validate input shape
        expected_features = self.input_shape[-1]
        if features.shape[-1] != expected_features:
            raise ValueError(f"Expected {expected_features} features, got {features.shape[-1]}")
        
        # Add batch dimension if needed
        if len(features.shape) == 1:
            features = features.reshape(1, -1)
        
        return features
    
    def predict(self, features: np.ndarray, threshold: float = 0.5) -> PredictionResult:
        """
        Predict product category from features
        
        Args:
            features: Input feature vector
            threshold: Minimum confidence threshold
            
        Returns:
            PredictionResult with prediction, confidence, and all scores
        """
        try:
            # Preprocess input
            processed_input = self.preprocess(features)
            
            # Run inference
            outputs = self.session.run(
                [self.output_name],
                {self.input_name: processed_input}
            )
            
            # Get probabilities (assuming softmax output)
            probabilities = outputs[0][0]
            
            # Get top prediction
            predicted_idx = int(np.argmax(probabilities))
            confidence = float(probabilities[predicted_idx])
            
            # Create scores dictionary
            all_scores = {
                self.labels[i]: float(probabilities[i])
                for i in range(len(self.labels))
            }
            
            # Check confidence threshold
            if confidence < threshold:
                self.logger.warning(f"Low confidence prediction: {confidence:.3f}")
            
            return PredictionResult(
                prediction=self.labels[predicted_idx],
                confidence=confidence,
                all_scores=all_scores
            )
            
        except Exception as e:
            self.logger.error(f"Prediction failed: {str(e)}")
            raise
    
    def batch_predict(self, features_batch: List[np.ndarray]) -> List[PredictionResult]:
        """Batch prediction for multiple inputs"""
        results = []
        for features in features_batch:
            try:
                result = self.predict(features)
                results.append(result)
            except Exception as e:
                self.logger.error(f"Batch prediction error: {str(e)}")
                results.append(None)
        return results