Run and Monitor Industrial ML Experiments Through to Model Serving with MLRun and KServe

"""
Production implementation for running and monitoring industrial ML experiments using MLRun and KServe.
Provides secure, scalable operations for managing machine learning workflows.
"""
from typing import Dict, Any, List
import os
import time
import logging
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

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

class Config:
    mlrun_api_url: str = os.getenv('MLRUN_API_URL')
    kserve_api_url: str = os.getenv('KSERVE_API_URL')
    retry_attempts: int = 3
    retry_delay: int = 2  # seconds

class ExperimentData(BaseModel):
    experiment_id: str = Field(..., description="ID of the experiment")
    parameters: Dict[str, Any] = Field(..., description="Parameters for the experiment")

async def validate_input(data: ExperimentData) -> bool:
    """Validate experiment data.
    
    Args:
        data: ExperimentData instance to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not data.experiment_id:
        raise ValueError('Missing experiment_id')
    if not data.parameters:
        raise ValueError('Missing parameters')
    return True

async def fetch_experiment_results(experiment_id: str) -> Dict[str, Any]:
    """Fetch results from MLRun API.
    
    Args:
        experiment_id: ID of the experiment to fetch results for
    Returns:
        JSON response with results
    Raises:
        HTTPException: If API call fails
    """
    url = f"{Config.mlrun_api_url}/get/{experiment_id}"
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logger.error(f"Failed to fetch results: {e}")
        raise HTTPException(status_code=500, detail="Failed to fetch results")

async def save_experiment_to_db(data: ExperimentData) -> None:
    """Save experiment data to the database.
    
    Args:
        data: ExperimentData instance to save
    Raises:
        Exception: If saving fails
    """
    # Simulated database save
    logger.info(f"Saving experiment {data.experiment_id} to database...")
    # Here you would implement your DB logic

async def call_kserve_service(model_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
    """Call KServe service to make predictions.
    
    Args:
        model_name: Name of the model to call
        payload: Data to send for prediction
    Returns:
        Prediction results
    Raises:
        HTTPException: If API call fails
    """
    url = f"{Config.kserve_api_url}/{model_name}/predict"
    try:
        response = requests.post(url, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logger.error(f"Failed to call KServe: {e}")
        raise HTTPException(status_code=500, detail="Failed to call KServe")

async def monitor_experiment(experiment_id: str) -> Dict[str, Any]:
    """Monitor the status of an experiment.
    
    Args:
        experiment_id: ID of the experiment to monitor
    Returns:
        Status of the experiment
    Raises:
        Exception: If monitoring fails
    """
    logger.info(f"Monitoring experiment {experiment_id}...")
    for attempt in range(Config.retry_attempts):
        try:
            results = await fetch_experiment_results(experiment_id)
            return results
        except HTTPException:
            if attempt < Config.retry_attempts - 1:
                logger.warning(f"Retrying... {attempt + 1}/{Config.retry_attempts}")
                time.sleep(Config.retry_delay)
            else:
                logger.error(f"Failed to monitor experiment after {Config.retry_attempts} attempts")
                raise

app = FastAPI()

@app.post("/run-experiment/")
async def run_experiment(data: ExperimentData):
    """Run machine learning experiment.
    
    Args:
        data: ExperimentData instance containing the experiment details
    Returns:
        Status of the experiment
    Raises:
        HTTPException: If the experiment fails
    """
    await validate_input(data)  # Validate input
    await save_experiment_to_db(data)  # Save to DB
    # Simulated model call
    prediction = await call_kserve_service(data.experiment_id, data.parameters)
    logger.info(f"Experiment {data.experiment_id} completed with prediction: {prediction}")
    return prediction

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