Serve INT4-Quantized Factory Classification Models with torchao and Triton Inference Server

"""
Production implementation for serving INT4-Quantized factory classification models with TorchAO and Triton Inference Server.
Provides secure, scalable operations for model inference and data processing.
"""

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, validator
import logging
import os
import json
import requests
import time
from typing import Dict, Any, List, Tuple

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

# Configuration class to manage environment variables
class Config:
    TRITON_SERVER_URL: str = os.getenv('TRITON_SERVER_URL', 'http://localhost:8000')

# Input model for request validation
class ClassificationRequest(BaseModel):
    data: List[List[float]]  # Expecting a list of feature vectors

    @validator('data')
    def validate_data(cls, v):
        if not v:
            raise ValueError('Input data cannot be empty.')
        return v

# Helper function to send inference requests to Triton
def send_inference_request(data: List[List[float]]) -> Dict[str, Any]:
    """Send inference request to Triton Inference Server.
    
    Args:
        data: List of feature vectors for classification
    Returns:
        Inference result from Triton
    Raises:
        Exception: If request fails
    """
    url = f"{Config.TRITON_SERVER_URL}/v2/models/factory_model/infer"
    payload = {'inputs': [{'name': 'input_tensor', 'shape': [len(data), len(data[0])], 'datatype': 'FP32', 'data': data}]}
    try:
        response = requests.post(url, json=payload)
        response.raise_for_status()  # Raise an error for bad responses
        logger.info('Inference request successful.')
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'Error during inference: {e}')
        raise Exception('Inference request failed.')

# FastAPI application setup
app = FastAPI()

@app.post('/classify', response_model=Dict[str, Any])
async def classify(request: ClassificationRequest):
    """Endpoint for classifying input data.
    
    Args:
        request: Request object containing input data
    Returns:
        Classification result
    Raises:
        HTTPException: If input validation fails
    """
    try:
        data = request.data  # Extracting validated data
        inference_result = send_inference_request(data)
        return {'result': inference_result}
    except ValueError as ve:
        logger.warning(f'Validation error: {ve}')
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        logger.error(f'Error during classification: {e}')
        raise HTTPException(status_code=500, detail='Internal Server Error')

def retry_request(func, retries: int = 3, backoff: float = 2.0):
    """Retry logic for request handling.
    
    Args:
        func: Function to execute with retries
        retries: Number of retries
        backoff: Exponential backoff factor
    Returns:
        Result of the function call
    """
    for attempt in range(retries):
        try:
            return func()  # Call the function
        except Exception as e:
            logger.warning(f'Attempt {attempt + 1} failed: {e}')
            time.sleep(backoff * (2 ** attempt))  # Exponential backoff
    raise Exception('Max retries exceeded')

if __name__ == '__main__':
    # Example usage
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)