model-deployer

Files (2)

• SKILL.md
• cloud-deploy.md · references/

SKILL.md (first lines)

# Model Deployer

Take a trained model from notebook to production with serving code, containers, and cloud deployment.

---

## Workflow

```
1. Identify deployment pattern (real-time API vs batch)
2. Generate serving code
3. Create Docker container
4. Add monitoring and health checks
5. Generate cloud deployment config
```

---

## Step 1 -- Deployment Pattern

| Pattern | When | Framework |
|---|---|---|
| **Real-time API** | Low-latency predictions, user-facing | FastAPI |
| **Batch prediction** | Scheduled scoring of large datasets | Python script + scheduler |
| **Streaming** | Event-driven predictions | Kafka consumer + model |
| **Embedded** | Model runs in app process | Direct library import |
| **Serverless** | Infrequent, bursty traffic | AWS Lambda / GCP Cloud Functions |

**Default**: Real-time API with FastAPI.

---

## Step 2 -- Serving Code

### FastAPI Template

```python
"""Model serving API."""
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import logging
import time

# --- Config ---
MODEL_PATH = "model_artifacts/best_model.joblib"
APP_TITLE = "ML Model API"
APP_VERSION = "1.0.0"

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

app = FastAPI(title=APP_TITLE, version=APP_VERSION)

# Load model at startup
model = None

@app.on_event("startup")
def load_model():
    global model
    logger.info(f"Loading model from {MODEL_PATH}")
    model = joblib.load(MODEL_PATH)
    logger.info("Model loaded successfully")


# --- Schemas ---
class PredictionRequest(BaseModel):
    features: dict = Field(..., description="Feature name-value pairs", example={
        "age": 35, "income": 75000, "tenure_months": 24
    })

class PredictionResponse(BaseModel):
    prediction: float | int | str
    probability: Optional[list[float]] = None
    model_version: str = APP_VERSION
    latency_ms: float


# --- Endpoints ---
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    start = time.time()
    try:
        # Convert features to model input format
        feature_names = list(request.features.keys())
        feature_values = np.array([list(request.features.values())])

        # Handle pandas DataFrame input if model expects it
        import pandas as pd
        X = pd.DataFrame(feature_values, columns=feature_names)

        prediction = model.predict(X)[0]
        probability = None
        if hasattr(model, "predict_proba"):
            probability = model.predict_proba(X)[0].tolist()

        latency = (time.time() - start) * 1000
        return PredictionResponse(
            prediction=prediction,
            probability=probability,
            latency_ms=round(latency, 2),
        )
    except Exception as e:
        logger.error(f"Prediction error: {e}")
        raise HTTPException(status_code=422, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "healthy", "model_loaded": model is not None}


@app.get("/model-info")
async def model_info():
    info = {"version": APP_VERSION, "type": type(model).__name__}
    if hasattr(model, "feature_names_in_"):

From tezansahu/ai-garage, GPL-3.0.

More from tezansahu/ai-garage