MLOps – versioning, deploying and monitoring models

Subject: Data Science & Artificial Intelligence
School level: 12th grade – HTL Informatik
Prerequisites: Supervised Learning, Deep Learning, Python, basic web knowledge
Author: HTL Pinkafeld – IF/IT

1. What is MLOps?

1.1 The problem without MLOps

A data scientist trains a model on his laptop - it works great. Then it should go into production:

"It works on my machine!"

Problems in practice:
❌ Other Python/library versions on the server
❌ No way to compare model versions
❌ Nobody knows what data the model was trained with
❌ Model deteriorates over time - no one notices
❌ Deployment takes weeks instead of hours

1.2 What is MLOps?

MLOps (Machine Learning Operations) is the combination of ML development and DevOps practices. The goal is to bring ML models into production and operate them in a reproducible, scalable and reliable manner.

DevOps: Development + Operations → Deploy software quickly and stably
MLOps: ML + Development + Ops → Deploy ML models quickly and stably

1.3 MLOps vs. classic ML

1.4 Core Principles

1. Reproducibility → Every experiment must be repeatable
2. Versioning → Code, data AND models are versioned
3. Automation → Training, testing, deployment run automatically
4. Monitoring → Model quality is continuously monitored
5. Collaboration → Teams work together on pipelines

2. The ML lifecycle

2.1 Overview

┌─────────────────────── ───────────────────────┐
│ ML Lifecycle │
│ │
│ Data Training Serving │
│ collecting → & experiments → & monitoring │
│ ↑ │ │
│ └──────────── Feedback ────────────┘ │
└─────────────────────── ───────────────────────┘

2.2 The three phases in detail

Phase 1 – Data:

• Collect, clean, label data
• Feature engineering
• Version data set

Phase 2 – Training & Experiments:

• Track experiments (hyperparameters, metrics)
• Validate model
• Register and version model

Phase 3 – Serving & Monitoring:

• Provide model as API
• Log predictions
• Detect data drift
• Retrain if necessary (retraining)

2.3 Level of automation

MLOps distinguishes between three levels of maturity:

3. Manage experiments with MLflow

3.1 What is MLflow?

MLflow is an open source tool for managing the entire ML lifecycle:

• Tracking: Experiments, parameters, metrics, artifacts
• Models: Model registry and versioning
• Projects: Reproducible ML projects
• Serving: Serve models as a REST endpoint

pip install mlflow scikit-learn

3.2 Track first experiment

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score, roc_auc_score
import numpy as np

# Load data
data = load_breast_cancer()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.2, random_state=42
)

# Start MLflow tracking server (local):
# > mlflow ui → http://localhost:5000

mlflow.set_experiment("Breast Cancer Classification")

# Try different hyperparameters
for n_estimators in [50, 100, 200]:
    for max_depth in [3, 5, None]:

        with mlflow.start_run(run_name=f"RF_n{n_estimators}_d{max_depth}"):

            # ── Log parameters ──────────────────── ─────────────────────
            mlflow.log_param("n_estimators", n_estimators)
            mlflow.log_param("max_depth",    max_depth)
            mlflow.log_param("dataset",      "breast_cancer")

            # ── Train model ──────────────────── ────────────────────
            model = RandomForestClassifier(
                n_estimators=n_estimators,
                max_depth=max_depth,
                random_state=42
            )
            model.fit(X_train, y_train)

            # ── Log metrics ───────────────────── ─────────────────────
            y_pred  = model.predict(X_test)
            y_proba = model.predict_proba(X_test)[:, 1]

            mlflow.log_metric("accuracy", accuracy_score(y_test, y_pred))
            mlflow.log_metric("f1_score", f1_score(y_test, y_pred))
            mlflow.log_metric("roc_auc",  roc_auc_score(y_test, y_proba))
            mlflow.log_metric("oob_score", model.oob_score_
                              if hasattr(model, 'oob_score_') else 0)

            # ── Save model ──────────────────── ─────────────────────
            mlflow.sklearn.log_model(model, "random_forest_model")

            print(f"n={n_estimators}, depth={max_depth}: "
                  f"Acc={accuracy_score(y_test, y_pred):.3f}")

3.3 MLflow UI

After execution, mlflow ui (http://localhost:5000) shows an overview of all runs with all parameters and metrics. You can compare runs, sort them and choose the best model.

3.4 Load model from registry

import mlflow.sklearn

# Load best model (after selection in the UI)
run_id = "abc123def456"   # aus MLflow UI kopieren
model  = mlflow.sklearn.load_model(f"runs:/{run_id}/random_forest_model")

# Or from the Model Registry (if registered):
model = mlflow.sklearn.load_model("models:/BrustkrebsKlassifikator/Production")

4. Save and load models

4.1 scikit-learn models (joblib)

import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris

# Train model
iris  = load_iris()
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(iris.data, iris.target)

# ── Save ────────────────────────────── ───────────────────────────────
joblib.dump(model, 'models/iris_classifier.joblib')
print("Model saved.")

# ── Download and use ───────────────────────── ──────────────────────────
loaded_model = joblib.load('models/iris_classifier.joblib')
pred = loaded_model.predict([[5.1, 3.5, 1.4, 0.2]])
print(f"Prediction: {iris.target_names[pred[0]]}")

# ── Important: Always persist with scalers! ────────────────────────────────
from sklearn.preprocessing import StandardScaler
import joblib

scaler = StandardScaler()
X_scaled = scaler.fit_transform(iris.data)
model.fit(X_scaled, iris.target)

# Save both together
joblib.dump({'model': model, 'scaler': scaler}, 'models/pipeline.joblib')

# Load
artefakte    = joblib.load('models/pipeline.joblib')
model_loaded  = artefakte['model']
scaler_loaded = artefakte['scaler']

# New data: first scale, then predict
X_neu = scaler_loaded.transform([[5.1, 3.5, 1.4, 0.2]])
print(model_loaded.predict(X_neu))

4.2 Keras / TensorFlow models

import tensorflow as tf
from tensorflow import keras

# Create and train a model
model = keras.Sequential([
    keras.layers.Dense(64, activation='relu', input_shape=(4,)),
    keras.layers.Dense(3, activation='softmax')
])
model.compile('adam', 'sparse_categorical_crossentropy', ['accuracy'])
model.fit(iris.data, iris.target, epochs=20, verbose=0)

# ── Save in the new Keras format (.keras) ──────────────────────────────
model.save('models/iris_nn.keras')

# ── Loading ──────────────────────────────── ─────────────────────────────────
loaded = keras.models.load_model('models/iris_nn.keras')
pred   = loaded.predict([[5.1, 3.5, 1.4, 0.2]])
print(f"Klasse: {pred.argmax()}")

# ── As SavedModel (TF Serving compatible) ────────────────────────────────
model.export('models/iris_savedmodel')

4.3 Versioning strategy

models/
├── iris_classifier_v1.0.0_2024-01-15.joblib
├── iris_classifier_v1.1.0_2024-02-03.joblib
└── iris_classifier_v2.0.0_2024-03-20.joblib

Naming convention: {name}_v{major}.{minor}.{patch}_{date}.{ext}
major: incompatible changes (new feature set)
minor: improvements, backwards compatible
Patch: Bug fixes, retraining on new data

5. Deploy models as REST API

5.1 Why an API?

A saved model is not yet productive. Other systems (web apps, mobile apps, other services) need a standardized interface - a REST API.

Mobile app Web frontend Other service
│ │ │
└─────────────────── ┼──────────────────┘
│ HTTP POST /predict
▼
ML API Server
│
ML model

5.2 API with FastAPI

FastAPI is a modern Python framework for REST APIs - fast, type-safe and with automatic Swagger documentation:

pip install fastapi uvicorn joblib scikit-learn

# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
from typing import List

# ── Load model ───────────────────────────── ─────────────────────────────
app        = FastAPI(title="Iris Klassifikator API", version="1.0")
artefakte  = joblib.load('models/pipeline.joblib')
model      = artefakte['model']
scaler     = artefakte['scaler']
klassen    = ['setosa', 'versicolor', 'virginica']

# ── Request/Response models ─────────────────────── ───────────────────────
class IrisInput(BaseModel):
    sepal_length: float
    sepal_width:  float
    petal_length: float
    petal_width:  float

    class Config:
        json_schema_extra = {
            "example": {
                "sepal_length": 5.1,
                "sepal_width":  3.5,
                "petal_length": 1.4,
                "petal_width":  0.2
            }
        }

class PredictionResponse(BaseModel):
    klasse:          str
    klassen_id:      int
    wahrscheinlichkeit: float
    alle_wahrscheinlichkeiten: List[float]

# ── Endpoints ────────────────────────────── ───────────────────────────────
@app.get("/")
def root():
    return {"status": "online", "modell": "Iris Klassifikator v1.0"}

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

@app.post("/predict", response_model=PredictionResponse)
def predict(data: IrisInput):
    try:
        X = np.array([[
            data.sepal_length, data.sepal_width,
            data.petal_length, data.petal_width
        ]])
        X_scaled = scaler.transform(X)
        pred      = model.predict(X_scaled)[0]
        proba     = model.predict_proba(X_scaled)[0]

        return PredictionResponse(
            klasse=klassen[pred],
            klassen_id=int(pred),
            wahrscheinlichkeit=float(proba[pred]),
            alle_wahrscheinlichkeiten=proba.tolist()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ── Start server ──────────────────────────── ─────────────────────────────
# uvicorn app:app --host 0.0.0.0 --port 8000 --reload

5.3 Test API

import requests

# Call endpoint
response = requests.post(
    "http://localhost:8000/predict",
    json={
        "sepal_length": 5.1,
        "sepal_width":  3.5,
        "petal_length": 1.4,
        "petal_width":  0.2
    }
)

result = response.json()
print(f"Klasse: {result['klasse']}")
print(f"Konfidenz: {result['wahrscheinlichkeit']:.1%}")

# Or with curl:
curl -X POST "http://localhost:8000/predict" \
     -H "Content-Type: application/json" \
     -d '{"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}'

6. Containerization with Docker

6.1 Why Docker?

Docker packages the application together with all dependencies into a container. This runs the same everywhere – on the laptop, on the server, in the cloud.

Without Docker: With Docker:
"Works for me!"      Container = App + Python 3.11 + all libraries
→ Identical behavior everywhere

6.2 Project structure

ml-api/
├── app.py ← FastAPI application
├── models/
│ └── pipeline.joblib ← Saved model
├── requirements.txt ← Python dependencies
└── Dockerfile ← Container building instructions

6.3 Dockerfile

# Base image: Python 3.11 (slim)
FROM python:3.11-slim

# Working directory in the container
WORKDIR /app

# Copy and install dependencies first (use Docker Cache)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY app.py .
COPY models/ models/

# Share port
EXPOSE 8000

# Start command
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

6.4 requirements.txt

fastapi==0.104.1
uvicorn==0.24.0
scikit-learn==1.3.2
joblib==1.3.2
numpy==1.26.0
pydantic==2.5.0

6.5 Docker Commands

# Build image
docker build -t ml-api:1.0 .

# Start container
docker run -d -p 8000:8000 --name iris-api ml-api:1.0

# Check status
docker ps
docker logs iris-api

# Stop containers
docker stop iris-api

# Publish image to registry (e.g. Docker Hub)
docker tag ml-api:1.0 meinbenutzername/ml-api:1.0
docker push meinbenutzername/ml-api:1.0

6.6 Docker Compose (multiple services)

# docker-compose.yml
version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - mlflow
    environment:
      - MLFLOW_TRACKING_URI=http://mlflow:5000

  mlflow:
    image: ghcr.io/mlflow/mlflow:latest
    ports:
      - "5000:5000"
    volumes:
      - mlflow_data:/mlflow
    command: mlflow server --host 0.0.0.0

volumes:
  mlflow_data:

docker compose up -d       # Alle Services starten
docker compose down        # Alle Services stoppen
docker compose logs -f api # Logs verfolgen

7. CI/CD for ML projects

7.1 What is CI/CD?

CI (Continuous Integration): Tests are automatically executed with every code push.

CD (Continuous Deployment): If the CI is successful, it is deployed automatically.

Developer pushes code
│
▼
GitHub Actions
│
┌─────┴────────┐
│ │
Tests model training
│ │
▼ ▼
Linting evaluation
│ │
└─────┬────────┘
│ All tests OK?
▼
Docker build
│
▼
Deploy to server

7.2 GitHub Actions Workflow

# .github/workflows/ml-pipeline.yml
name: ML Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Python einrichten
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov

      - name: Run tests
        run: pytest tests/ --cov=app --cov-report=xml

  train-and-evaluate:
    runs-on: ubuntu-latest
    needs: test

    steps:
      - uses: actions/checkout@v4

      - name: Python einrichten
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Train model
        run: python train.py

      - name: Evaluate model
        run: python evaluate.py --min-accuracy 0.90

      - name: Store model as artifact
        uses: actions/upload-artifact@v4
        with:
          name: trained-model
          path: models/

  deploy:
    runs-on: ubuntu-latest
    needs: train-and-evaluate
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model
          path: models/

      - name: Docker Image bauen und pushen
        run: |
          docker build -t ${{ secrets.REGISTRY }}/ml-api:${{ github.sha }} .
          docker push ${{ secrets.REGISTRY }}/ml-api:${{ github.sha }}

7.3 Write model tests

# tests/test_model.py
import pytest
import joblib
import numpy as np
from sklearn.metrics import accuracy_score
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

@pytest.fixture
def model_und_scaler():
    artefakte = joblib.load('models/pipeline.joblib')
    return artefakte['model'], artefakte['scaler']

@pytest.fixture
def testdaten():
    iris = load_iris()
    _, X_test, _, y_test = train_test_split(
        iris.data, iris.target, test_size=0.2, random_state=42
    )
    return X_test, y_test

def test_modell_laedt(model_und_scaler):
    model, scaler = model_und_scaler
    assert model is not None
    assert scaler is not None

def test_vorhersage_korrekte_form(model_und_scaler):
    model, scaler = model_und_scaler
    X = np.array([[5.1, 3.5, 1.4, 0.2]])
    X_scaled = scaler.transform(X)
    pred = model.predict(X_scaled)
    assert pred.shape == (1,)
    assert pred[0] in [0, 1, 2]

def test_accuracy_ueber_schwellwert(model_und_scaler, testdaten):
    model, scaler = model_und_scaler
    X_test, y_test = testdaten
    X_scaled = scaler.transform(X_test)
    acc = accuracy_score(y_test, model.predict(X_scaled))
    assert acc >= 0.90, f"Accuracy {acc:.2%} unter Mindestwert 90%"

def test_wahrscheinlichkeiten(model_und_scaler):
    model, scaler = model_und_scaler
    X = np.array([[5.1, 3.5, 1.4, 0.2]])
    X_scaled = scaler.transform(X)
    proba = model.predict_proba(X_scaled)
    assert proba.shape == (1, 3)
    assert abs(proba.sum() - 1.0) < 1e-6   # Summe muss 1 sein

8. Model monitoring and data drift

8.1 Why monitoring?

A model that works well today may fail tomorrow - because the world is changing:

Examples of data drift:
→ Recommendation algorithm: User behavior changes
→ Fraud detection: Fraudsters adapt their methods
→ Price prediction: Inflation changes all prices
→ Medicine: New disease variants emerge

8.2 Types of Drift

8.3 Detect drift with Evidently

pip install evidently

import pandas as pd
import numpy as np
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import DatasetDriftMetric, ColumnDriftMetric

# ── Reference data (training data) ──────────────────────────────────────
np.random.seed(42)
n = 500

referenz = pd.DataFrame({
    'alter':      np.random.normal(35, 10, n).clip(18, 80),
    'einkommen':  np.random.normal(3000, 800, n).clip(1000),
    'spending':   np.random.normal(1500, 400, n).clip(100),
    'klasse':     np.random.choice([0, 1], n, p=[0.7, 0.3])
})

# ── Current data (production data, simulated drift) ──────────────────
aktuell = pd.DataFrame({
    'alter':     np.random.normal(42, 12, n).clip(18, 80),   # older!
    'einkommen': np.random.normal(3500, 1000, n).clip(1000), # mehr Einkommen
    'spending':  np.random.normal(1500, 400, n).clip(100),   # unchanged
    'klasse':    np.random.choice([0, 1], n, p=[0.5, 0.5])   # mehr Klasse 1
})

# ── Create Evidently Report ────────────────────── ──────────────────────
report = Report(metrics=[
    DataDriftPreset(),
    DataQualityPreset(),
])

report.run(reference_data=referenz, current_data=aktuell)
report.save_html("drift_report.html")
print("Report gespeichert: drift_report.html")

# ── Drift decision ────────────────────────── ──────────────────────────
result = report.as_dict()
drift_detected = result['metrics'][0]['result']['dataset_drift']
print(f"Datendrift erkannt: {drift_detected}")

if drift_detected:
    print("⚠️ Retraining empfohlen!")

8.4 Log predictions

import logging
import json
from datetime import datetime

# Simple predictions logging
logging.basicConfig(
    filename='predictions.log',
    level=logging.INFO,
    format='%(asctime)s %(message)s'
)

def predict_and_log(model, scaler, X_input, request_id=None):
    """Prediction with automatic logging."""
    X_scaled = scaler.transform([X_input])
    pred     = model.predict(X_scaled)[0]
    proba    = model.predict_proba(X_scaled)[0]

    log_entry = {
        "timestamp":  datetime.now().isoformat(),
        "request_id": request_id,
        "input":      X_input,
        "prediction": int(pred),
        "confidence": float(proba.max())
    }
    logging.info(json.dumps(log_entry))

    return pred, proba

# use
pred, proba = predict_and_log(model, scaler,
                               [5.1, 3.5, 1.4, 0.2],
                               request_id="req-001")

9. Feature stores and data pipelines

9.1 The problem without a feature store

Team A (Recommendations): calculates “Customer Value” in their own way
Team B (Churn): calculates “Customer Value” differently
Team C (Marketing): calculates “customer value” again differently

→ Inconsistency, duplicate work, training-serving skew!

Training Serving Skew: Features are calculated differently during training than in production → Model behaves differently than expected.

9.2 What is a feature store?

A Feature Store is a central platform for storing, managing and deploying features:

Raw data
│
▼
Feature engineering (one-off, central)
│
▼
Feature Store
│
├── Training: retrieve historical features
└── Serving: current features in real time

9.3 Simple data pipeline with pandas

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
import joblib
import os

def lade_rohdaten(pfad: str) -> pd.DataFrame:
    """Step 1: Load raw data."""
    df = pd.read_csv(pfad)
    print(f"Geladen: {len(df)} Zeilen, {df.shape[1]} Spalten")
    return df

def bereinige_daten(df: pd.DataFrame) -> pd.DataFrame:
    """Step 2: Cleaning."""
    df = df.dropna(subset=['target'])
    df['alter'] = df['alter'].fillna(df['alter'].median())
    df = df[df['alter'].between(0, 120)]
    df = df.drop_duplicates()
    print(f"Nach Bereinigung: {len(df)} Zeilen")
    return df

def erstelle_features(df: pd.DataFrame) -> pd.DataFrame:
    """Step 3: Feature engineering."""
    df['spending_ratio']   = df['spending'] / df['einkommen'].clip(1)
    df['high_earner']      = (df['einkommen'] > df['einkommen'].quantile(0.75)).astype(int)
    df['age_group']        = pd.cut(df['alter'],
                                     bins=[0, 25, 40, 60, 120],
                                     labels=[0, 1, 2, 3]).astype(int)
    return df

def skaliere_features(df: pd.DataFrame, feature_cols: list,
                       scaler_pfad: str, training: bool = True):
    """Step 4: Scaling."""
    if training:
        scaler = StandardScaler()
        df[feature_cols] = scaler.fit_transform(df[feature_cols])
        joblib.dump(scaler, scaler_pfad)
        print(f"Scaler gespeichert: {scaler_pfad}")
    else:
        scaler = joblib.load(scaler_pfad)
        df[feature_cols] = scaler.transform(df[feature_cols])
    return df

def run_pipeline(rohdaten_pfad: str, output_pfad: str,
                 scaler_pfad: str, training: bool = True):
    """Run full pipeline."""
    feature_cols = ['alter', 'einkommen', 'spending',
                    'spending_ratio', 'high_earner', 'age_group']

    df = lade_rohdaten(rohdaten_pfad)
    df = bereinige_daten(df)
    df = erstelle_features(df)
    df = skaliere_features(df, feature_cols, scaler_pfad, training)
    df.to_parquet(output_pfad, index=False)
    print(f"Features gespeichert: {output_pfad}")
    return df

# Carry out
# run_pipeline('data/raw.csv', 'data/features.parquet', 'models/scaler.joblib')

10. MLOps maturity and tools

10.1 Tools at a glance

MLOps stack
│
├── Experiment tracking
│ ├── MLflow (open source, local + cloud)
│ ├── Weights & Biases (SaaS, very popular)
│ └── Neptune.ai (SaaS)
│
├── Data versioning
│ ├── DVC (Data Version Control, Git-like)
│ └── Delta Lake (for large amounts of data)
│
├── Pipeline orchestration
│ ├── Apache Airflow (widespread, complex)
│ ├── Prefect (easier than Airflow)
│ └── Kubeflow (Kubernetes native)
│
├── Model Serving
│ ├── FastAPI (simple, flexible)
│ ├── TorchServe (PyTorch models)
│ ├── TF Serving (TensorFlow models)
│ └── BentoML (framework-agnostic)
│
├── Monitoring
│ ├── Evidently AI (data drift, model quality)
│ ├── Prometheus (metrics)
│ └── Grafana (visualization)
│
└── Infrastructure
├── Docker (containerization)
├── Kubernetes (container orchestration)
└── Cloud (AWS SageMaker, GCP Vertex AI, Azure ML)

10.2 Getting started recommendation

Phase 1 – Basics (from now on):
✅ MLflow for experiment tracking
✅ joblib/keras for model persistence
✅ FastAPI for model serving
✅ Git for code versioning

Phase 2 – Professional:
✅ Docker for containerization
✅ GitHub Actions for CI/CD
✅ Evidently for drift monitoring
✅ DVC for data versioning

Phase 3 – Enterprise:
✅ Kubernetes for scaled serving
✅ Airflow/Prefect for pipelines
✅ Cloud ML platform (SageMaker, Vertex AI)

11. Practical Project: End-to-End MLOps Pipeline

11.1 Project structure

ml project/
├── data/
│ ├── raw/ ← raw data
│ └── processed/ ← Processed features
├── models/ ← Saved models
├── src/
│ ├── pipeline.py ← Data pipeline
│ ├── train.py ← Training
│ └── predict.py ← prediction
├── tests/
│ └── test_model.py ← Model tests
├── app.py ← FastAPI app
├── Dockerfile ← Container
├── requirements.txt
└── README.md

11.2 train.py – Reproducible training

# src/train.py
import mlflow
import mlflow.sklearn
import joblib
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import accuracy_score, classification_report
from sklearn.datasets import load_breast_cancer
import json
from datetime import datetime

def trainiere_modell(
    n_estimators: int = 100,
    max_depth: int    = None,
    test_size: float  = 0.2,
    random_state: int = 42,
    min_accuracy: float = 0.90
):
    """Fully reproducible training with MLflow tracking."""

    mlflow.set_experiment("Produktion-Brustkrebs")

    with mlflow.start_run():

        # ── 1. Load data ──────────────────────── ────────────────────────
        data = load_breast_cancer()
        X_train, X_test, y_train, y_test = train_test_split(
            data.data, data.target,
            test_size=test_size,
            random_state=random_state
        )

        # ── 2. Log parameters ───────────────────── ──────────────────────
        params = {
            "n_estimators":  n_estimators,
            "max_depth":     str(max_depth),
            "test_size":     test_size,
            "random_state":  random_state,
            "trainiert_am":  datetime.now().isoformat()
        }
        mlflow.log_params(params)

        # ── 3. Train the model ───────────────────── ─────────────────────
        model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            oob_score=True,
            random_state=random_state
        )
        model.fit(X_train, y_train)

        # ── 4. Calculate metrics ──────────────────── ─────────────────────
        y_pred = model.predict(X_test)
        acc    = accuracy_score(y_test, y_pred)
        cv     = cross_val_score(model, data.data, data.target, cv=5).mean()

        mlflow.log_metric("accuracy",   acc)
        mlflow.log_metric("cv_accuracy", cv)
        mlflow.log_metric("oob_score",  model.oob_score_)

        # ── 5. Quality Gate ─────────────────────── ────────────────────────
        if acc < min_accuracy:
            raise ValueError(
                f"Accuracy {acc:.2%} unter Mindestwert {min_accuracy:.0%}! "
                "Training abgebrochen."
            )

        # ── 6. Save model ───────────────────── ──────────────────────
        joblib.dump(model, 'models/model.joblib')
        mlflow.sklearn.log_model(model, "model")

        # Save metadata
        with open('models/metadata.json', 'w') as f:
            json.dump({
                **params,
                "accuracy": acc,
                "cv_accuracy": cv,
                "feature_names": list(data.feature_names),
                "klassen": list(data.target_names)
            }, f, indent=2)

        print(f"✅ Model trained: Accuracy={acc:.2%}, CV={cv:.2%}")
        return model, acc

if __name__ == "__main__":
    trainiere_modell(n_estimators=200, max_depth=10)

11.3 Minimal monitoring dashboard

# monitor.py
import json
import pandas as pd
from collections import Counter
from datetime import datetime, timedelta

def analysiere_predictions(log_pfad: str, zeitfenster_stunden: int = 24):
    """Einfaches Monitoring der letzten Predictions."""

    # Read log file
    eintraege = []
    with open(log_pfad) as f:
        for zeile in f:
            try:
                # Format: "2024-01-15 10:30:00 {json}"
                teile = zeile.strip().split(' ', 2)
                if len(teile) == 3:
                    eintrag = json.loads(teile[2])
                    eintrag['timestamp'] = datetime.fromisoformat(eintrag['timestamp'])
                    eintraege.append(eintrag)
            except Exception:
                continue

    if not eintraege:
        print("Keine Predictions gefunden.")
        return

    df = pd.DataFrame(eintraege)

    # Last time slot only
    grenze = datetime.now() - timedelta(hours=zeitfenster_stunden)
    df     = df[df['timestamp'] > grenze]

    print(f"\n{'='*40}")
    print(f"Monitoring-Report (letzte {zeitfenster_stunden}h)")
    print(f"{'='*40}")
    print(f"Anzahl Predictions:    {len(df)}")
    print(f"Ø Konfidenz:           {df['confidence'].mean():.1%}")
    print(f"Min. Konfidenz:        {df['confidence'].min():.1%}")
    print(f"Predictions < 60%:     {(df['confidence'] < 0.6).sum()}")
    print(f"\nKlassenverteilung:")
    for klasse, anzahl in Counter(df['prediction']).items():
        print(f"  Klasse {klasse}: {anzahl} ({anzahl/len(df):.0%})")

# analyze_predictions('predictions.log')

12. Summary and outlook

Overview

MLOps stack
│
├── Experiment Tracking → MLflow: Parameters, metrics, artifacts
├── Model persistence → joblib / keras.save
├── Serving → FastAPI + REST endpoint
├── Containerization → Docker + Dockerfile
├── CI/CD → GitHub Actions
├── Monitoring → Evidently (Drift), Logging (Predictions)
└── Data pipeline → pandas, parquet, feature store

The golden rules of MLOps

✅ Version everything: code (Git), data (DVC), models (MLflow)
✅ Always track experiments – never optimize manually without logging
✅ Install quality gates: Only deploy the model if Accuracy > X%
✅ Always save the scaler/preprocessor together with the model
✅ Log predictions – basis for monitoring and retraining
✅ Write tests: Model tests are just as important as code tests
✅ Use Docker: “works on my machine” is not a solution
✅ Check drift regularly – models quietly become outdated

Checklist for a production project

Training:
✅ MLflow experiment created
✅ All hyperparameters logged
✅ All metrics logged
✅ Quality Gate (min. Accuracy) defined
✅ Model + Scaler saved
✅ Metadata (feature names, classes) saved

Serving:
✅ FastAPI endpoint with input validation
✅ /health endpoint available
✅ Predictions are logged
✅ Error handling (try/except)

Deployment:
✅ Dockerfile created
✅ requirements.txt complete
✅ GitHub Actions workflow set up
✅ Model tests available (pytest)

Monitoring:
✅ Drift monitoring set up
✅ Alerts for poor performance
✅ Retraining strategy defined

Outlook

With the completed topics in this unit, you have learned about the complete Data Science & AI development cycle:

Understanding data → Statistics & Visualization
Modeling data → Supervised Learning, Ensemble, Unsupervised
Images & Text → Deep Learning, CNNs, NLP, Transformer
Production → MLOps, deployment, monitoring

Related topics:

• Retrieval-Augmented Generation (RAG): Connect LLMs with your own data
• Reinforcement Learning: Train agents who act in environments
• Federated Learning: Privacy-friendly training on distributed devices
• AutoML: Automated model selection and hyperparameter optimization

Appendix: Quick Reference MLOps

# ── MLflow ──────────────────────────────── ────────────────────────────────
import mlflow, mlflow.sklearn

mlflow.set_experiment("mein-experiment")
with mlflow.start_run():
    mlflow.log_param("n_estimators", 100)
    mlflow.log_metric("accuracy", 0.95)
    mlflow.sklearn.log_model(model, "model")

# ── Save/load model ─────────────────────── ───────────────────────
import joblib
joblib.dump({'model': model, 'scaler': scaler}, 'models/pipeline.joblib')
artefakte = joblib.load('models/pipeline.joblib')

# ── FastAPI ─────────────────────────────── ────────────────────────────────
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Input(BaseModel):
    feature1: float
    feature2: float

@app.post("/predict")
def predict(data: Input):
    X = [[data.feature1, data.feature2]]
    return {"prediction": int(model.predict(X)[0])}

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

# ── Docker ──────────────────────────────── ────────────────────────────────
# docker build -t ml-api:1.0 .
# docker run -d -p 8000:8000 ml-api:1.0

# ── Evidently drift check ──────────────────────── ─────────────────────────
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=referenz_df, current_data=aktuell_df)
report.save_html("drift_report.html")

# ── pytest model test ────────────────────────── ──────────────────────────
# pytest tests/ -v

HTL Pinkafeld – IF/IT | Data Science & AI | 12th grade