Comprehensive comparison for Model Serving technology in AI applications

import asyncio
import json
import logging
from typing import Dict, List, Optional
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field
import tensorrt_llm
from tensorrt_llm.runtime import ModelRunner, SamplingConfig
import torch

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

app = FastAPI(title="TensorRT-LLM Model Serving API")

# Request/Response models
class GenerationRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4096)
    max_tokens: int = Field(default=256, ge=1, le=2048)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    top_p: float = Field(default=0.9, ge=0.0, le=1.0)
    stream: bool = Field(default=False)

class GenerationResponse(BaseModel):
    generated_text: str
    tokens_generated: int
    model_name: str

# Global model runner instance
model_runner: Optional[ModelRunner] = None
MODEL_PATH = "/models/llama-7b-trt"
MODEL_NAME = "llama-7b"

@app.on_event("startup")
async def load_model():
    """Load TensorRT-LLM model on startup"""
    global model_runner
    try:
        logger.info(f"Loading TensorRT-LLM model from {MODEL_PATH}")
        model_runner = ModelRunner.from_dir(
            engine_dir=MODEL_PATH,
            rank=0,
            debug_mode=False
        )
        logger.info("Model loaded successfully")
    except Exception as e:
        logger.error(f"Failed to load model: {str(e)}")
        raise RuntimeError(f"Model initialization failed: {str(e)}")

@app.on_event("shutdown")
async def cleanup():
    """Cleanup resources on shutdown"""
    global model_runner
    if model_runner:
        del model_runner
        torch.cuda.empty_cache()
        logger.info("Model resources cleaned up")

@app.post("/v1/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest):
    """Generate text using TensorRT-LLM model"""
    if model_runner is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    try:
        # Configure sampling parameters
        sampling_config = SamplingConfig(
            max_new_tokens=request.max_tokens,
            temperature=request.temperature,
            top_p=request.top_p,
            end_id=model_runner.tokenizer.eos_token_id,
            pad_id=model_runner.tokenizer.pad_token_id
        )
        
        # Tokenize input
        input_ids = model_runner.tokenizer.encode(
            request.prompt,
            add_special_tokens=True,
            return_tensors="pt"
        )
        
        # Generate output
        with torch.no_grad():
            outputs = model_runner.generate(
                input_ids,
                sampling_config=sampling_config
            )
        
        # Decode output
        generated_text = model_runner.tokenizer.decode(
            outputs[0],
            skip_special_tokens=True
        )
        
        # Remove input prompt from output
        generated_text = generated_text[len(request.prompt):].strip()
        tokens_generated = len(outputs[0]) - len(input_ids[0])
        
        logger.info(f"Generated {tokens_generated} tokens")
        
        return GenerationResponse(
            generated_text=generated_text,
            tokens_generated=tokens_generated,
            model_name=MODEL_NAME
        )
    
    except torch.cuda.OutOfMemoryError:
        logger.error("GPU out of memory")
        raise HTTPException(status_code=507, detail="GPU memory exhausted")
    except Exception as e:
        logger.error(f"Generation failed: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Generation error: {str(e)}")

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy" if model_runner else "unhealthy",
        "model": MODEL_NAME,
        "gpu_available": torch.cuda.is_available()
    }