MLOps Cheatsheet

MLflow Tracking

Logging Runs

import mlflow
mlflow.set_experiment("exp-name")
mlflow.autolog()                     # auto params + metrics

with mlflow.start_run(run_name="v1"):
    mlflow.log_param("lr", 0.01)
    mlflow.log_params({"depth": 8})
    mlflow.log_metric("acc", 0.91)
    mlflow.log_artifact("plot.png")
    mlflow.sklearn.log_model(model, "model")

mlflow ui                            # launch UI :5000
mlflow models serve -m runs:/ID/model -p 5001

Model Registry

mlflow.register_model("runs:/ID/model", "my-model")
client.set_registered_model_alias("my-model","production",v)
mlflow.pyfunc.load_model("models:/my-model@production")

DVC Commands

dvc init
dvc add data/train.csv          # track large file
dvc remote add -d storage s3://bucket/dvc
dvc push                        # upload data
dvc pull                        # download data
dvc repro                       # run pipeline DAG
dvc exp run                     # run experiment
dvc exp show                    # compare experiments
dvc metrics diff                # metric changes
dvc checkout                    # restore data for commit

dvc.yaml Stage

stages:
  train:
    cmd: python train.py
    deps: [data/clean.csv, train.py]
    params: [train.lr]
    outs: [model.pkl]
    metrics: [metrics.json]

FastAPI Serving

from fastapi import FastAPI
from pydantic import BaseModel
import joblib

app = FastAPI()
model = joblib.load("model.pkl")

class In(BaseModel):
    x1: float
    x2: float

@app.post("/predict")
def predict(f: In):
    p = model.predict([[f.x1, f.x2]])[0]
    return {"prediction": int(p)}

@app.get("/health")
def health(): return {"status": "ok"}

# run:
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

Docker

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn","app:app","--host","0.0.0.0","--port","8000"]

docker build -t api:1.0 .
docker run -p 8000:8000 api:1.0
docker push registry/api:1.0
docker compose up -d

Orchestration (Prefect)

from prefect import flow, task

@task(retries=2, retry_delay_seconds=30)
def step(): ...

@flow(name="pipeline")
def run():
    step()

run.serve(cron="0 2 * * *")      # schedule daily 2am

Monitoring & Drift

from evidently import Report
from evidently.presets import DataDriftPreset

r = Report(metrics=[DataDriftPreset()])
res = r.run(reference_data=ref, current_data=live)
res.save_html("drift.html")

CI/CD (GitHub Actions)

on: [push]
jobs:
  ci:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -r requirements.txt
      - run: pytest tests/
      - run: dvc pull && dvc repro
      - run: python evaluate.py --min-accuracy 0.85