Deploying ML models with Modal and Ouro

import modal
 
# Define container image with ML dependencies
ml_image = (
    modal.Image.debian_slim(python_version="3.12")
    .pip_install(
        "fastapi[standard]==0.115.4",
        "torch>=2.0.0",
        "torchvision",
        "transformers>=4.30.0",
        "scikit-learn>=1.3.0",
        "tensorflow>=2.13.0",
        "numpy>=1.24.0",
        "pandas>=2.0.0",
        "pillow>=10.0.0",
        "pydantic>=2.0.0"
    )
    .apt_install("git", "wget", "curl")
)
 
app = modal.App("ml-model-api", image=ml_image)

from fastapi import FastAPI, Header, HTTPException, Depends, Request
import os
import time
import logging
 
# Authentication utilities
async def validate_ouro_authentication(
    authorization: str = Header(None, alias="Authorization"),
    request: Request = None,
):
    """Validate Ouro platform authentication using `Authorization: Basic <token>` header"""
 
    # Required header: Authorization: Basic <token>
    if not authorization or not authorization.lower().startswith("basic "):
        client_host = getattr(getattr(request, "client", None), "host", "unknown")
        logging.warning(f"Missing or invalid Authorization header from {client_host}")
        raise HTTPException(status_code=401, detail="Missing Ouro auth header")
 
    token = authorization.split(" ", 1)[1].strip()
 
    # Validate against stored token
    expected_secret = os.environ.get("OURO_SERVICE_SECRET")
 
    if not expected_secret:
        logging.error("Missing env var OURO_SERVICE_SECRET for Ouro authentication")
        raise HTTPException(status_code=500, detail="Server configuration error")
 
    if token != expected_secret:
        logging.warning("Invalid Ouro token")
        raise HTTPException(status_code=401, detail="Invalid Ouro credentials")
 
    # Log successful authentication
    logging.info("Successful Ouro authentication")
 
    return {"platform": "ouro", "authenticated_at": time.time()}
 
# Apply authentication to endpoints
@web_app.post("/inference/ouro")
async def inference_for_ouro(
    request: PredictionInput,
    auth: dict = Depends(validate_ouro_authentication)
):
    """ML inference endpoint specifically for Ouro integration"""
    try:
        # Your ML processing logic
        result = await process_ml_inference(request)
 
        return {
            **result,
            "authenticated_via": auth["platform"],
            "request_id": f"ouro_{int(time.time())}"
        }
 
    except Exception as e:
        logging.error(f"Ouro inference error: {str(e)}")
        raise HTTPException(status_code=500, detail="Inference processing failed")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
import torch
import torch.nn.functional as F
import logging
 
# Request/Response schemas for OpenAPI generation
class PredictionInput(BaseModel):
    features: List[float] = Field(..., description="Input feature vector")
    model_version: Optional[str] = Field("v1.0", description="Model version to use")
 
    class Config:
        json_schema_extra = {
            "example": {
                "features": [1.0, 2.0, 3.0, 4.0],
                "model_version": "v1.0"
            }
        }
 
class PredictionOutput(BaseModel):
    prediction: float = Field(..., description="Model prediction")
    confidence: float = Field(..., description="Prediction confidence score")
    model_version: str = Field(..., description="Model version used")
    inference_time_ms: float = Field(..., description="Inference time in milliseconds")
 
@app.cls(
    image=ml_image,
    gpu="T4",  # Use T4 GPU for cost-effective inference
    memory=4096,  # 4GB RAM
    volumes={"/cache": modal.Volume.from_name("model-cache", create_if_missing=True)},
    secrets=[modal.Secret.from_name("ml-model-secrets")]
)
class PyTorchModelService:
    def __enter__(self):
        """Load model once when container starts"""
        import time
        start_time = time.time()
 
        # Load model from cache or download
        model_path = "/cache/pytorch_model.pt"
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
        try:
            self.model = torch.load(model_path, map_location=self.device)
        except FileNotFoundError:
            # Download model if not cached
            self.model = self._download_and_cache_model(model_path)
 
        self.model.eval()
        self.model_version = "v1.0"
 
        load_time = time.time() - start_time
        logging.info(f"Model loaded in {load_time:.2f}s on {self.device}")
 
    def _download_and_cache_model(self, cache_path: str):
        """Download model and save to cache"""
        # Implementation depends on your model source
        model = torch.nn.Linear(4, 1)  # Example model
        torch.save(model, cache_path)
        return model
 
    @modal.method()
    def predict(self, input_data: PredictionInput) -> PredictionOutput:
        import time
        start_time = time.time()
 
        try:
            # Prepare input tensor
            features = torch.FloatTensor([input_data.features]).to(self.device)
 
            # Perform inference
            with torch.no_grad():
                output = self.model(features)
                prediction = output.item()
 
                # Calculate confidence using sigmoid
                confidence = torch.sigmoid(output).item()
 
            inference_time = (time.time() - start_time) * 1000  # Convert to ms
 
            return PredictionOutput(
                prediction=prediction,
                confidence=confidence,
                model_version=self.model_version,
                inference_time_ms=inference_time
            )
 
        except Exception as e:
            logging.error(f"Prediction error: {str(e)}")
            raise HTTPException(status_code=500, detail=f"Inference failed: {str(e)}")
 
    @modal.fastapi_app()
    def create_fastapi_app(self):
        web_app = FastAPI(
            title="PyTorch ML Model API",
            description="Production PyTorch model serving with Modal",
            version="1.0.0",
            docs_url="/docs",
            openapi_url="/openapi.json"
        )
 
        @web_app.post("/predict", response_model=PredictionOutput)
        async def predict_endpoint(request: PredictionInput):
            return self.predict(request)
 
        @web_app.get("/health")
        async def health_check():
            return {"status": "healthy", "model_loaded": True, "device": str(self.device)}
 
        return web_app

import modal
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
 
app = modal.App("ml-model-ouro")
web_app = FastAPI(
    title="Your ML Model API",
    description="Detailed description of your model's capabilities, use cases, and limitations",
    version="1.0.0",
    contact={
        "name": "Your Name",
        "email": "[email protected]"
    }
)
 
class PredictionRequest(BaseModel):
    input_data: List[float]
    model_parameters: Optional[dict] = None
 
    class Config:
        schema_extra = {
            "example": {
                "input_data": [1.0, 2.5, 3.2],
                "model_parameters": {"temperature": 0.7}
            }
        }
 
class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str
    processing_time_ms: float
 
@web_app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """
    Generate predictions using the ML model.
 
    This endpoint processes input data through our trained model
    and returns predictions with confidence scores.
    """
    try:
        # Your model inference logic here
        result = your_model_function(request.input_data)
 
        return PredictionResponse(
            prediction=result["prediction"],
            confidence=result["confidence"],
            model_version="1.0.0",
            processing_time_ms=result["processing_time"]
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Prediction failed: {str(e)}")
 
@app.function(
    image=modal.Image.debian_slim().pip_install(["fastapi", "your-ml-dependencies"]),
    container_idle_timeout=300
)
@modal.asgi_app()
def fastapi_app():
    return web_app

from fastapi import FastAPI, Header
from fastapi.openapi.utils import get_openapi
from ouro.utils import get_custom_openapi, ouro_field
from typing import Optional
 
web_app = FastAPI(
    title="Your ML Model API",
    description="Production ML model serving with Ouro integration",
    version="1.0.0"
)
 
# Apply Ouro's custom OpenAPI enhancements
web_app.openapi = get_custom_openapi(web_app, get_openapi)
 
@web_app.post("/predict")
@ouro_field(
    "x-ouro-input-assets",
    {
        "input_file": {
            "asset_type": "file",
            "input_file_extensions": ["xy", "xye"],
        }
    },
)
@ouro_field(
    "x-ouro-output-assets",
    {
        "prediction_file": {
            "asset_type": "file",
            "file_extensions": ["json"],
            "primary": True,
        }
    },
)
async def predict(
    request: PredictionRequest,
    # Ouro passes these headers to identify context
    ouro_route_id: Optional[str] = Header(None, alias="ouro-route-id"),
    ouro_route_org_id: Optional[str] = Header(None, alias="ouro-route-org-id"),
    ouro_route_team_id: Optional[str] = Header(None, alias="ouro-route-team-id"),
    ouro_action_id: Optional[str] = Header(None, alias="ouro-action-id"),
):
    """
    Process input and generate predictions.
 
    The Ouro headers provide context about:
    - ouro_route_id: The service route being accessed
    - ouro_route_org_id: Organization ID for asset creation context
    - ouro_route_team_id: Team ID for asset creation context
    - ouro_action_id: Unique action ID for logging and tracking
    """
    from ouro import Ouro
    import os
 
    # Initialize Ouro SDK with your API key
    ouro = Ouro(api_key=os.environ["OURO_API_KEY"])
 
    # Log progress to Ouro's action system
    if ouro_action_id:
        ouro.client.post(
            f"/actions/{ouro_action_id}/log",
            json={
                "message": "Starting model inference...",
                "asset_id": ouro_route_id,
                "level": "info",
            }
        )
 
    # Your model inference logic here
    result = your_model_function(request)
 
    # Return result with Ouro metadata for asset creation
    return {
        "prediction": result,
        "prediction_file": {  # This creates a file asset in Ouro
            "name": f"Prediction for {request.input_name}",
            "description": "Model prediction output",
            "filename": "prediction.json",
            "type": "application/json",
            "extension": "json",
            "base64": base64_encode(result),
            "org_id": ouro_route_org_id,  # Use Ouro context
            "team_id": ouro_route_team_id,
        }
    }

from fastapi import Request, HTTPException
 
ALLOWED_DOMAINS = ["api.ouro.foundation", "ouro.foundation"]
 
@web_app.middleware("http")
async def verify_origin(request: Request, call_next):
    origin = request.headers.get("origin", "")
    referer = request.headers.get("referer", "")
 
    # Allow direct API testing and Ouro platform access
    if any(domain in origin or domain in referer for domain in ALLOWED_DOMAINS):
        response = await call_next(request)
        return response
 
    # Allow requests without origin (direct API calls)
    if not origin and not referer:
        response = await call_next(request)
        return response
 
    raise HTTPException(status_code=403, detail="Access denied")