Serve Vision-Language Model APIs for Factory Robots with TensorRT Edge-LLM and FastAPI

"""
Production implementation for serving vision-language model APIs for factory robots using TensorRT Edge-LLM and FastAPI.
Provides secure, scalable operations.
"""

from typing import Dict, Any, List
import os
import logging
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ValidationError

# Logger setup for monitoring application behavior
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables
class Config:
    api_url: str = os.getenv('API_URL', 'http://localhost:8000/api')
    db_url: str = os.getenv('DATABASE_URL', 'sqlite:///./test.db')

# Define the data model for input requests
class VisionLanguageRequest(BaseModel):
    image: str  # Base64 encoded image
    query: str  # Text query for the model

# Helper function to validate input data
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'image' not in data or 'query' not in data:
        raise ValueError('Both image and query are required.')
    return True

# Function to transform the input data for processing
def transform_records(data: VisionLanguageRequest) -> Dict[str, Any]:
    """Transform input records for model processing.
    
    Args:
        data: VisionLanguageRequest model instance
    Returns:
        Transformed data as dictionary
    """
    return {
        'image_data': data.image,
        'query_text': data.query
    }

# Function to process a batch of requests
async def process_batch(requests: List[VisionLanguageRequest]) -> List[Dict[str, Any]]:
    """Process a batch of requests to the model.
    
    Args:
        requests: List of VisionLanguageRequest instances
    Returns:
        Results of processing as list of dictionaries
    """
    results = []
    for request in requests:
        try:
            transformed_data = transform_records(request)
            # Here we simulate a call to the model API
            response = await call_api(transformed_data)
            results.append(response)
        except Exception as e:
            logger.error(f'Error processing request: {e}')
            results.append({'error': str(e)})
    return results

# Function to call the external API
async def call_api(data: Dict[str, Any]) -> Dict[str, Any]:
    """Call the external model API.
    
    Args:
        data: Transformed data to send
    Returns:
        API response as dictionary
    Raises:
        HTTPException: If API call fails
    """
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(Config.api_url, json=data)
            response.raise_for_status()  # Raise error for bad responses
            return response.json()  # Return the JSON response
        except httpx.HTTPStatusError as http_err:
            logger.error(f'HTTP error occurred: {http_err}')
            raise HTTPException(status_code=http_err.response.status_code, detail=str(http_err))
        except Exception as err:
            logger.error(f'Other error occurred: {err}')
            raise HTTPException(status_code=500, detail=str(err))

# FastAPI app initialization
app = FastAPI()

@app.post('/process', response_model=List[Dict[str, Any]])
async def process_requests(requests: List[VisionLanguageRequest]) -> List[Dict[str, Any]]:
    """Process incoming requests for vision-language model.
    
    Args:
        requests: List of VisionLanguageRequest
    Returns:
        List of processed results
    Raises:
        HTTPException: If validation or processing fails
    """
    try:
        # Validate each request
        for request in requests:
            await validate_input(request.dict())
        # Process the validated requests
        results = await process_batch(requests)
        return results
    except ValidationError as ve:
        logger.error(f'Validation error: {ve}')
        raise HTTPException(status_code=422, detail=str(ve))
    except Exception as e:
        logger.error(f'Unexpected error: {e}')
        raise HTTPException(status_code=500, detail='Internal Server Error')

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