Comprehensive comparison for Model Serving technology in AI applications

import bentoml
import numpy as np
from bentoml.io import JSON, NumpyNdarray
from pydantic import BaseModel, Field, validator
from typing import List, Optional
import logging

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

# Define input/output schemas
class ProductRecommendationInput(BaseModel):
    user_id: str = Field(..., description="Unique user identifier")
    product_history: List[int] = Field(..., min_items=1, max_items=50)
    num_recommendations: Optional[int] = Field(default=5, ge=1, le=20)
    
    @validator('product_history')
    def validate_product_ids(cls, v):
        if not all(pid > 0 for pid in v):
            raise ValueError("Product IDs must be positive integers")
        return v

class ProductRecommendationOutput(BaseModel):
    user_id: str
    recommended_products: List[int]
    confidence_scores: List[float]
    model_version: str

# Load the pre-trained recommendation model
try:
    recommendation_model = bentoml.pytorch.get("product_recommender:latest")
    logger.info(f"Loaded model: {recommendation_model.tag}")
except bentoml.exceptions.NotFound:
    logger.error("Model not found. Please ensure the model is saved to BentoML store.")
    raise

# Create BentoML service
svc = bentoml.Service("product_recommendation_service", runners=[recommendation_model.to_runner()])

@svc.api(input=JSON(pydantic_model=ProductRecommendationInput), output=JSON())
async def recommend_products(input_data: ProductRecommendationInput) -> dict:
    """
    Generate personalized product recommendations based on user purchase history.
    
    Args:
        input_data: User information and purchase history
        
    Returns:
        Recommended products with confidence scores
    """
    try:
        logger.info(f"Processing recommendation request for user: {input_data.user_id}")
        
        # Prepare input features
        product_history = np.array(input_data.product_history, dtype=np.int32)
        
        # Create feature vector (simplified example)
        feature_vector = np.zeros(1000, dtype=np.float32)
        for pid in product_history:
            if pid < 1000:
                feature_vector[pid] = 1.0
        
        # Reshape for model input
        feature_vector = feature_vector.reshape(1, -1)
        
        # Run inference
        predictions = await recommendation_model.async_run(feature_vector)
        
        # Get top N recommendations
        top_indices = np.argsort(predictions[0])[::-1][:input_data.num_recommendations]
        recommended_products = top_indices.tolist()
        confidence_scores = predictions[0][top_indices].tolist()
        
        # Filter out products already in history
        filtered_recommendations = []
        filtered_scores = []
        for prod_id, score in zip(recommended_products, confidence_scores):
            if prod_id not in input_data.product_history:
                filtered_recommendations.append(prod_id)
                filtered_scores.append(float(score))
        
        # Prepare response
        output = ProductRecommendationOutput(
            user_id=input_data.user_id,
            recommended_products=filtered_recommendations[:input_data.num_recommendations],
            confidence_scores=filtered_scores[:input_data.num_recommendations],
            model_version=str(recommendation_model.tag)
        )
        
        logger.info(f"Successfully generated {len(filtered_recommendations)} recommendations")
        return output.dict()
        
    except ValueError as ve:
        logger.error(f"Validation error: {str(ve)}")
        return {"error": "Invalid input data", "details": str(ve)}
    except Exception as e:
        logger.error(f"Unexpected error during inference: {str(e)}")
        return {"error": "Internal server error", "details": "Failed to generate recommendations"}

@svc.api(input=JSON(), output=JSON())
def health_check() -> dict:
    """Health check endpoint for monitoring."""
    return {
        "status": "healthy",
        "model_loaded": True,
        "model_version": str(recommendation_model.tag)
    }