Route Multi-Model Edge Inference Requests with vLLM and Triton Inference Server

"""
Production implementation for routing multi-model edge inference requests with vLLM and Triton Inference Server.
Provides secure, scalable operations.
"""
from typing import Dict, Any, List
import os
import logging
import requests
from pydantic import BaseModel, ValidationError
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from functools import wraps
from time import sleep

# Logger setup with appropriate levels
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    triton_url: str = os.getenv('TRITON_URL', 'http://localhost:8000')
    retry_attempts: int = int(os.getenv('RETRY_ATTEMPTS', 3))
    retry_delay: float = float(os.getenv('RETRY_DELAY', 1.0))

app = FastAPI()

def retry(func):
    """Decorator for retrying a function with exponential backoff.
    """  
    @wraps(func)
    def wrapper(*args, **kwargs):
        attempts = 0
        while attempts < Config.retry_attempts:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.warning(f'Attempt {attempts + 1} failed: {e}')
                attempts += 1
                sleep(Config.retry_delay * (2 ** attempts))  # Exponential backoff
        raise Exception('Max retry attempts exceeded')
    return wrapper

class InferenceRequest(BaseModel):
    model_name: str
    input_data: List[Dict[str, Any]]

def validate_input(data: InferenceRequest) -> None:
    """Validate the input data for inference requests.
    
    Args:
        data: InferenceRequest object containing model name and input data
    Raises:
        ValueError: If validation fails
    """
    if not data.model_name:
        raise ValueError('Model name is required')
    if not data.input_data:
        raise ValueError('Input data is required')

@retry
def call_triton_inference(data: InferenceRequest) -> Dict[str, Any]:
    """Call the Triton Inference Server for model inference.
    
    Args:
        data: InferenceRequest object
    Returns:
        Response from Triton server
    Raises:
        HTTPException: If the request fails
    """  
    logger.info('Calling Triton Inference Server')
    response = requests.post(f'{Config.triton_url}/v2/models/{data.model_name}/infer', json=data.dict())
    if response.status_code != 200:
        raise HTTPException(status_code=response.status_code, detail=response.text)
    return response.json()

@app.post('/inference', response_model=Dict[str, Any])
async def infer(request: Request, data: InferenceRequest):
    """Handle inference requests.
    
    Args:
        request: FastAPI request object
        data: InferenceRequest object
    Returns:
        JSON response with inference results
    Raises:
        HTTPException: On validation or inference errors
    """  
    try:
        validate_input(data)
        results = call_triton_inference(data)
        return JSONResponse(content=results)
    except ValidationError as e:
        logger.error(f'Validation error: {e}')
        raise HTTPException(status_code=422, detail=e.errors())
    except Exception as e:
        logger.error(f'Inference error: {e}')
        raise HTTPException(status_code=500, detail='Inference failed')

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