model-serving-and-monitoring

Model Serving and Monitoring

Overview

A model in production fails in ways a notebook never shows: the API re-implements preprocessing slightly differently, the artifact loaded is not the one evaluated, inputs drift away from the training data, or an async endpoint blocks on CPU work and the service stalls. Serving code is ordinary backend code with typed inputs, tests and a health check — plus a version on every prediction.

Core principle: serve the exact fitted pipeline you evaluated, identified by an immutable version, and log enough with every prediction to notice when the world stops looking like the training data.

1. Batch or online?

Batch is simpler to operate, test and roll back; choose online only when the decision needs it.

2. Online endpoint (FastAPI)

class Features(BaseModel):
    model_config = ConfigDict(extra="forbid")
    days_since_order: int = Field(ge=0, le=3650)
    orders_90d: int = Field(ge=0)
    plan: Literal["free", "pro", "team"]

class ScoreRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    rows: list[Features] = Field(min_length=1, max_length=1000)

class ScoreResponse(BaseModel):
    model_version: str
    scores: list[float]

def create_app(settings: Settings) -> FastAPI:
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        app.state.model = load_model(settings.model_path, settings.trusted_types)
        app.state.model_version = settings.model_version
        yield

    app = FastAPI(lifespan=lifespan)

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/ready")
    def ready(request: Request) -> dict[str, str]:
        if getattr(request.app.state, "model", None) is None:
            raise HTTPException(status_code=503, detail="model not loaded")
        return {"status": "ready", "model_version": request.app.state.model_version}

    @app.post("/score")
    def score(body: ScoreRequest, request: Request) -> ScoreResponse:
        frame = pd.DataFrame([row.model_dump() for row in body.rows])
        proba = request.app.state.model.predict_proba(frame)[:, 1]
        return ScoreResponse(model_version=request.app.state.model_version, scores=proba.tolist())

    return app