Serve Concurrent LLM Requests on Factory Edge with SGLang and llama.cpp

"""
Production implementation for serving concurrent LLM requests on factory edge.
Utilizes SGLang and llama.cpp for efficient handling.
"""
from typing import Dict, Any, List
import os
import logging
import time
import asyncio
from aiohttp import ClientSession, ClientTimeout

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

class Config:
    """
    Configuration class to manage environment variables.
    """
    llama_endpoint: str = os.getenv('LLAMA_ENDPOINT', 'http://localhost:5000')
    timeout: int = int(os.getenv('REQUEST_TIMEOUT', 5))

async def validate_input(data: Dict[str, Any]) -> bool:
    """
    Validate the input data for the LLM request.

    Args:
        data: Input dictionary that contains request parameters.
    Returns:
        bool: True if valid, otherwise raises ValueError.
    Raises:
        ValueError: If validation fails.
    """
    if 'prompt' not in data:
        raise ValueError('Missing prompt in input data.')  # Ensure prompt is present
    if not isinstance(data['prompt'], str):
        raise ValueError('Prompt must be a string.')  # Validate prompt type
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sanitize input fields to prevent injections.

    Args:
        data: Input data dictionary.
    Returns:
        Dict[str, Any]: Sanitized data.
    """
    sanitized_data = {k: str(v).strip() for k, v in data.items()}  # Strip whitespace
    logger.info('Sanitized input data.')  # Log sanitization
    return sanitized_data

async def fetch_data(session: ClientSession, prompt: str) -> Dict[str, Any]:
    """
    Fetch data from the LLM service.

    Args:
        session: An aiohttp ClientSession object.
        prompt: The prompt to send to the LLM.
    Returns:
        Dict[str, Any]: Response from the LLM service.
    Raises:
        Exception: If fetching data fails.
    """
    try:
        async with session.post(Config.llama_endpoint, json={'prompt': prompt}) as response:
            response.raise_for_status()  # Raise error for bad responses
            data = await response.json()  # Parse JSON response
            logger.info('Data fetched from LLM service.')  # Log successful fetch
            return data
    except Exception as e:
        logger.error(f'Error fetching data: {e}')  # Log error
        raise

async def process_batch(prompts: List[str]) -> List[Dict[str, Any]]:
    """
    Process a batch of prompts concurrently.

    Args:
        prompts: List of prompts to process.
    Returns:
        List[Dict[str, Any]]: List of responses from the LLM service.
    """
    async with ClientSession(timeout=ClientTimeout(total=Config.timeout)) as session:
        tasks = [fetch_data(session, prompt) for prompt in prompts]  # Create fetch tasks
        results = await asyncio.gather(*tasks)  # Run tasks concurrently
        logger.info('Processed batch of prompts.')  # Log batch processing
        return results

async def handle_requests(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Handle incoming LLM requests.

    Args:
        data: Incoming request data.
    Returns:
        Dict[str, Any]: Response data.
    """
    try:
        await validate_input(data)  # Validate input
        sanitized_data = await sanitize_fields(data)  # Sanitize input
        responses = await process_batch([sanitized_data['prompt']])  # Process request
        return {'responses': responses}  # Return responses
    except ValueError as ve:
        logger.warning(f'Validation error: {ve}')  # Log validation issues
        return {'error': str(ve)}  # Return error message
    except Exception as e:
        logger.error(f'Error processing request: {e}')  # Log unexpected errors
        return {'error': 'Internal server error.'}  # Return generic error

if __name__ == '__main__':
    # Example usage of the async functions
    example_data = {'prompt': 'What is the capital of France?'}
    loop = asyncio.get_event_loop()  # Get the event loop
    result = loop.run_until_complete(handle_requests(example_data))  # Handle request
    print(result)  # Print the result