MLOps – Modelle versionieren, deployen und überwachen

Fach: Data Science & Artificial Intelligence
Schulstufe: 12. Schulstufe – HTL Informatik
Voraussetzungen: Supervised Learning, Deep Learning, Python, grundlegende Web-Kenntnisse
Autor: HTL Pinkafeld – IF/IT

1. Was ist MLOps?

1.1 Das Problem ohne MLOps

Ein Data Scientist trainiert ein Modell auf seinem Laptop – es funktioniert großartig. Dann soll es in Produktion:

"It works on my machine!"

Probleme in der Praxis:
  ❌ Anderes Python / Bibliotheksversionen auf dem Server
  ❌ Keine Möglichkeit, Modell-Versionen zu vergleichen
  ❌ Niemand weiß, mit welchen Daten das Modell trainiert wurde
  ❌ Modell verschlechtert sich mit der Zeit – niemand merkt es
  ❌ Deployment dauert Wochen statt Stunden

1.2 Was ist MLOps?

MLOps (Machine Learning Operations) ist die Kombination aus ML-Entwicklung und DevOps-Praktiken. Ziel ist es, ML-Modelle reproduzierbar, skalierbar und zuverlässig in Produktion zu bringen und dort zu betreiben.

DevOps:  Development + Operations  →  Software schnell und stabil deployen
MLOps:   ML + Development + Ops    →  ML-Modelle schnell und stabil deployen

1.3 MLOps vs. klassisches ML

1.4 Kernprinzipien

1. Reproduzierbarkeit  → Jedes Experiment muss wiederholbar sein
2. Versionierung       → Code, Daten UND Modelle werden versioniert
3. Automatisierung     → Training, Tests, Deployment laufen automatisch
4. Monitoring          → Modellqualität wird laufend überwacht
5. Kollaboration       → Teams arbeiten gemeinsam an Pipelines

2. Der ML-Lifecycle

2.1 Übersicht

        ┌──────────────────────────────────────────────┐
        │                 ML-Lifecycle                  │
        │                                              │
        │   Daten         Training        Serving      │
        │  sammeln   →   & Experimente  → & Monitoring │
        │     ↑                                  │     │
        │     └──────────── Feedback ────────────┘     │
        └──────────────────────────────────────────────┘

2.2 Die drei Phasen im Detail

Phase 1 – Daten:

• Daten erfassen, bereinigen, labeln
• Feature Engineering
• Datensatz versionieren

Phase 2 – Training & Experimente:

• Experimente tracken (Hyperparameter, Metriken)
• Modell validieren
• Modell registrieren und versionieren

Phase 3 – Serving & Monitoring:

• Modell als API bereitstellen
• Predictions loggen
• Datendrift erkennen
• Bei Bedarf neu trainieren (Retraining)

2.3 Automatisierungsgrad

MLOps unterscheidet drei Reifegrade:

3. Experimente verwalten mit MLflow

3.1 Was ist MLflow?

MLflow ist ein Open-Source-Tool zur Verwaltung des gesamten ML-Lifecycles:

• Tracking: Experimente, Parameter, Metriken, Artefakte
• Models: Modell-Registry und Versionierung
• Projects: Reproduzierbare ML-Projekte
• Serving: Modelle als REST-Endpunkt bereitstellen

pip install mlflow scikit-learn

3.2 Erstes Experiment tracken

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

# Daten laden
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
)

# MLflow-Tracking-Server starten (lokal):
# > mlflow ui  →  http://localhost:5000

mlflow.set_experiment("Brustkrebs-Klassifikation")

# Verschiedene Hyperparameter ausprobieren
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}"):

            # ── Parameter loggen ─────────────────────────────────────────
            mlflow.log_param("n_estimators", n_estimators)
            mlflow.log_param("max_depth",    max_depth)
            mlflow.log_param("dataset",      "breast_cancer")

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

            # ── Metriken loggen ──────────────────────────────────────────
            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)

            # ── Modell speichern ─────────────────────────────────────────
            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

Nach dem Ausführen zeigt mlflow ui (http://localhost:5000) eine Übersicht aller Runs mit allen Parametern und Metriken. Man kann Runs vergleichen, sortieren und das beste Modell auswählen.

3.4 Modell aus der Registry laden

import mlflow.sklearn

# Bestes Modell laden (nach Auswahl in der UI)
run_id = "abc123def456"   # aus MLflow UI kopieren
model  = mlflow.sklearn.load_model(f"runs:/{run_id}/random_forest_model")

# Oder aus der Model Registry (wenn registriert):
model = mlflow.sklearn.load_model("models:/BrustkrebsKlassifikator/Production")

4. Modelle speichern und laden

4.1 scikit-learn Modelle (joblib)

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

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

# ── Speichern ─────────────────────────────────────────────────────────────
joblib.dump(model, 'models/iris_classifier.joblib')
print("Modell gespeichert.")

# ── Laden und verwenden ───────────────────────────────────────────────────
loaded_model = joblib.load('models/iris_classifier.joblib')
pred = loaded_model.predict([[5.1, 3.5, 1.4, 0.2]])
print(f"Vorhersage: {iris.target_names[pred[0]]}")

# ── Wichtig: Scaler immer mitpersistieren! ────────────────────────────────
from sklearn.preprocessing import StandardScaler
import joblib

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

# Beide zusammen speichern
joblib.dump({'model': model, 'scaler': scaler}, 'models/pipeline.joblib')

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

# Neue Daten: erst skalieren, dann vorhersagen
X_neu = scaler_loaded.transform([[5.1, 3.5, 1.4, 0.2]])
print(model_loaded.predict(X_neu))

4.2 Keras / TensorFlow Modelle

import tensorflow as tf
from tensorflow import keras

# Modell erstellen und trainieren
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)

# ── Im neuen Keras-Format speichern (.keras) ──────────────────────────────
model.save('models/iris_nn.keras')

# ── Laden ─────────────────────────────────────────────────────────────────
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()}")

# ── Als SavedModel (TF Serving kompatibel) ────────────────────────────────
model.export('models/iris_savedmodel')

4.3 Versionierungsstrategie

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

Namenskonvention:  {name}_v{major}.{minor}.{patch}_{datum}.{ext}
  major: inkompatible Änderungen (neues Feature-Set)
  minor: Verbesserungen, abwärtskompatibel
  patch: Bugfixes, Retraining auf neuen Daten

5. Modelle als REST-API bereitstellen

5.1 Warum eine API?

Ein gespeichertes Modell ist noch nicht produktiv. Andere Systeme (Web-Apps, Mobile-Apps, andere Dienste) brauchen eine standardisierte Schnittstelle – eine REST-API.

Mobilapp          Web-Frontend       Anderer Dienst
    │                   │                  │
    └───────────────────┼──────────────────┘
                        │ HTTP POST /predict
                        ▼
                   ML-API Server
                        │
                   ML-Modell

5.2 API mit FastAPI

FastAPI ist ein modernes Python-Framework für REST-APIs – schnell, typsicher und mit automatischer Swagger-Dokumentation:

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

# ── Modell laden ──────────────────────────────────────────────────────────
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-Modelle ──────────────────────────────────────────────
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]

# ── Endpunkte ─────────────────────────────────────────────────────────────
@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))

# ── Server starten ─────────────────────────────────────────────────────────
# uvicorn app:app --host 0.0.0.0 --port 8000 --reload

5.3 API testen

import requests

# Endpunkt aufrufen
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%}")

# Oder mit 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. Containerisierung mit Docker

6.1 Warum Docker?

Docker verpackt die Anwendung zusammen mit allen Abhängigkeiten in einen Container. Dieser läuft überall gleich – auf dem Laptop, auf dem Server, in der Cloud.

Ohne Docker:          Mit Docker:
"Läuft bei mir!"      Container = App + Python 3.11 + alle Libraries
                      → Überall identisches Verhalten

6.2 Projektstruktur

ml-api/
  ├── app.py               ← FastAPI-Anwendung
  ├── models/
  │   └── pipeline.joblib  ← Gespeichertes Modell
  ├── requirements.txt     ← Python-Abhängigkeiten
  └── Dockerfile           ← Container-Bauanleitung

6.3 Dockerfile

# Basis-Image: Python 3.11 (schlank)
FROM python:3.11-slim

# Arbeitsverzeichnis im Container
WORKDIR /app

# Abhängigkeiten zuerst kopieren und installieren (Docker Cache nutzen)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Anwendungscode kopieren
COPY app.py .
COPY models/ models/

# Port freigeben
EXPOSE 8000

# Startbefehl
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-Befehle

# Image bauen
docker build -t ml-api:1.0 .

# Container starten
docker run -d -p 8000:8000 --name iris-api ml-api:1.0

# Status prüfen
docker ps
docker logs iris-api

# Container stoppen
docker stop iris-api

# Image in Registry veröffentlichen (z.B. Docker Hub)
docker tag ml-api:1.0 meinbenutzername/ml-api:1.0
docker push meinbenutzername/ml-api:1.0

6.6 Docker Compose (mehrere 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 für ML-Projekte

7.1 Was ist CI/CD?

CI (Continuous Integration): Bei jedem Code-Push werden automatisch Tests ausgeführt.

CD (Continuous Deployment): Bei erfolgreichem CI wird automatisch deployed.

Developer pusht Code
        │
        ▼
   GitHub Actions
        │
  ┌─────┴────────┐
  │              │
Tests           Model Training
  │              │
  ▼              ▼
Linting       Evaluation
  │              │
  └─────┬────────┘
        │ Alle Tests OK?
        ▼
   Docker Build
        │
        ▼
   Deploy zu 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: Abhängigkeiten installieren
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov

      - name: Tests ausführen
        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: Abhängigkeiten installieren
        run: pip install -r requirements.txt

      - name: Modell trainieren
        run: python train.py

      - name: Modell evaluieren
        run: python evaluate.py --min-accuracy 0.90

      - name: Modell als Artefakt speichern
        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: Modell herunterladen
        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 Modell-Tests schreiben

# 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. Modell-Monitoring und Datendrift

8.1 Warum Monitoring?

Ein Modell, das heute gut funktioniert, kann morgen versagen – weil sich die Welt ändert:

Beispiele für Datendrift:
  → Empfehlungsalgorithmus: Nutzerverhalten ändert sich
  → Betrugserkennung: Betrüger passen ihre Methoden an
  → Preisvorhersage: Inflation verändert alle Preise
  → Medizin: Neue Krankheitsvarianten entstehen

8.2 Arten von Drift

8.3 Drift erkennen mit 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

# ── Referenz-Daten (Trainingsdaten) ──────────────────────────────────────
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),
    'ausgaben':   np.random.normal(1500, 400, n).clip(100),
    'klasse':     np.random.choice([0, 1], n, p=[0.7, 0.3])
})

# ── Aktuelle Daten (Produktionsdaten, simulierter Drift) ──────────────────
aktuell = pd.DataFrame({
    'alter':     np.random.normal(42, 12, n).clip(18, 80),   # älter geworden!
    'einkommen': np.random.normal(3500, 1000, n).clip(1000), # mehr Einkommen
    'ausgaben':  np.random.normal(1500, 400, n).clip(100),   # unverändert
    'klasse':    np.random.choice([0, 1], n, p=[0.5, 0.5])   # mehr Klasse 1
})

# ── Evidently Report erstellen ────────────────────────────────────────────
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-Entscheidung ────────────────────────────────────────────────────
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 Predictions loggen

import logging
import json
from datetime import datetime

# Einfaches 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):
    """Vorhersage mit automatischem 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

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

9. Feature Stores und Datenpipelines

9.1 Das Problem ohne Feature Store

Team A (Empfehlungen):  berechnet "Kundenwert" auf eigene Art
Team B (Churn):         berechnet "Kundenwert" anders
Team C (Marketing):     berechnet "Kundenwert" nochmals anders

→ Inkonsistenz, doppelte Arbeit, Training-Serving-Skew!

Training-Serving-Skew: Features werden beim Training anders berechnet als in Produktion → Modell verhält sich anders als erwartet.

9.2 Was ist ein Feature Store?

Ein Feature Store ist eine zentrale Plattform zum Speichern, Verwalten und Bereitstellen von Features:

Rohdaten
    │
    ▼
Feature Engineering (einmalig, zentral)
    │
    ▼
Feature Store
    │
    ├── Training:   historische Features abrufen
    └── Serving:    aktuelle Features in Echtzeit

9.3 Einfache Datenpipeline mit 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:
    """Schritt 1: Rohdaten laden."""
    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:
    """Schritt 2: Bereinigung."""
    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:
    """Schritt 3: Feature Engineering."""
    df['ausgaben_ratio']   = df['ausgaben'] / 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):
    """Schritt 4: Skalierung."""
    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):
    """Vollständige Pipeline ausführen."""
    feature_cols = ['alter', 'einkommen', 'ausgaben',
                    'ausgaben_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

# Ausführen
# run_pipeline('data/raw.csv', 'data/features.parquet', 'models/scaler.joblib')

10. MLOps-Reifegrad und Werkzeuge

10.1 Werkzeuge im Überblick

MLOps-Stack
│
├── Experiment Tracking
│     ├── MLflow          (Open Source, lokal + Cloud)
│     ├── Weights & Biases (SaaS, sehr beliebt)
│     └── Neptune.ai      (SaaS)
│
├── Datenversionierung
│     ├── DVC             (Data Version Control, Git-ähnlich)
│     └── Delta Lake      (für große Datenmengen)
│
├── Pipeline-Orchestrierung
│     ├── Apache Airflow  (weit verbreitet, komplex)
│     ├── Prefect         (einfacher als Airflow)
│     └── Kubeflow        (Kubernetes-nativ)
│
├── Model Serving
│     ├── FastAPI         (einfach, flexibel)
│     ├── TorchServe      (PyTorch-Modelle)
│     ├── TF Serving      (TensorFlow-Modelle)
│     └── BentoML         (framework-agnostisch)
│
├── Monitoring
│     ├── Evidently AI    (Datendrift, Modellqualität)
│     ├── Prometheus      (Metriken)
│     └── Grafana         (Visualisierung)
│
└── Infrastruktur
      ├── Docker          (Containerisierung)
      ├── Kubernetes      (Container-Orchestrierung)
      └── Cloud (AWS SageMaker, GCP Vertex AI, Azure ML)

10.2 Einstieg-Empfehlung

Phase 1 – Grundlagen (ab sofort):
  ✅ MLflow für Experiment-Tracking
  ✅ joblib/keras für Modell-Persistenz
  ✅ FastAPI für Modell-Serving
  ✅ Git für Code-Versionierung

Phase 2 – Professionell:
  ✅ Docker für Containerisierung
  ✅ GitHub Actions für CI/CD
  ✅ Evidently für Drift-Monitoring
  ✅ DVC für Datenversionierung

Phase 3 – Enterprise:
  ✅ Kubernetes für skaliertes Serving
  ✅ Airflow/Prefect für Pipelines
  ✅ Cloud ML-Platform (SageMaker, Vertex AI)

11. Praktisches Projekt: End-to-End MLOps-Pipeline

11.1 Projektstruktur

ml-projekt/
  ├── data/
  │   ├── raw/           ← Rohdaten
  │   └── processed/     ← Verarbeitete Features
  ├── models/            ← Gespeicherte Modelle
  ├── src/
  │   ├── pipeline.py    ← Datenpipeline
  │   ├── train.py       ← Training
  │   └── predict.py     ← Vorhersage
  ├── tests/
  │   └── test_model.py  ← Modell-Tests
  ├── app.py             ← FastAPI-App
  ├── Dockerfile         ← Container
  ├── requirements.txt
  └── README.md

11.2 train.py – Reproduzierbares 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
):
    """Vollständig reproduzierbares Training mit MLflow-Tracking."""

    mlflow.set_experiment("Produktion-Brustkrebs")

    with mlflow.start_run():

        # ── 1. Daten laden ────────────────────────────────────────────────
        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. Parameter loggen ───────────────────────────────────────────
        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. Modell trainieren ──────────────────────────────────────────
        model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            oob_score=True,
            random_state=random_state
        )
        model.fit(X_train, y_train)

        # ── 4. Metriken berechnen ─────────────────────────────────────────
        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. Modell speichern ───────────────────────────────────────────
        joblib.dump(model, 'models/model.joblib')
        mlflow.sklearn.log_model(model, "model")

        # Metadaten speichern
        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"✅ Modell trainiert: Accuracy={acc:.2%}, CV={cv:.2%}")
        return model, acc

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

11.3 Minimales 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."""

    # Log-Datei einlesen
    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)

    # Nur letztes Zeitfenster
    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%})")

# analysiere_predictions('predictions.log')

12. Zusammenfassung und Ausblick

Überblick

MLOps-Stack
│
├── Experiment Tracking   → MLflow: Parameter, Metriken, Artefakte
├── Modell-Persistenz     → joblib / keras.save
├── Serving               → FastAPI + REST-Endpunkt
├── Containerisierung     → Docker + Dockerfile
├── CI/CD                 → GitHub Actions
├── Monitoring            → Evidently (Drift), Logging (Predictions)
└── Datenpipeline         → pandas, parquet, Feature Store

Die goldenen MLOps-Regeln

✅ Alles versionieren: Code (Git), Daten (DVC), Modelle (MLflow)
✅ Experimente immer tracken – nie manuell optimieren ohne Logging
✅ Quality Gates einbauen: Modell nur deployen wenn Accuracy > X%
✅ Scaler/Preprocessor immer zusammen mit Modell speichern
✅ Predictions loggen – Grundlage für Monitoring und Retraining
✅ Tests schreiben: Modell-Tests sind genauso wichtig wie Code-Tests
✅ Docker nutzen: "works on my machine" ist keine Lösung
✅ Drift regelmäßig prüfen – Modelle veralten still und leise

Checkliste für ein Produktionsprojekt

Training:
  ✅ MLflow Experiment erstellt
  ✅ Alle Hyperparameter geloggt
  ✅ Alle Metriken geloggt
  ✅ Quality Gate (min. Accuracy) definiert
  ✅ Modell + Scaler gespeichert
  ✅ Metadaten (Feature-Namen, Klassen) gespeichert

Serving:
  ✅ FastAPI Endpunkt mit Input-Validierung
  ✅ /health Endpunkt vorhanden
  ✅ Predictions werden geloggt
  ✅ Fehlerbehandlung (try/except)

Deployment:
  ✅ Dockerfile erstellt
  ✅ requirements.txt vollständig
  ✅ GitHub Actions Workflow eingerichtet
  ✅ Modell-Tests vorhanden (pytest)

Monitoring:
  ✅ Drift-Monitoring eingerichtet
  ✅ Alerts bei schlechter Performance
  ✅ Retraining-Strategie definiert

Ausblick

Mit den abgeschlossenen Themen dieser Einheit hast du den vollständigen Data Science & AI Entwicklungszyklus kennengelernt:

Daten verstehen   → Statistik & Visualisierung
Daten modellieren → Supervised Learning, Ensemble, Unsupervised
Bilder & Text     → Deep Learning, CNNs, NLP, Transformer
Produktion        → MLOps, Deployment, Monitoring

Weiterführende Themen:

• Retrieval-Augmented Generation (RAG): LLMs mit eigenen Daten verbinden
• Reinforcement Learning: Agenten trainieren, die in Umgebungen handeln
• Federated Learning: Datenschutzfreundliches Training auf verteilten Geräten
• AutoML: Automatisierte Modellauswahl und Hyperparameter-Optimierung

Anhang: Schnellreferenz 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")

# ── Modell speichern / laden ──────────────────────────────────────────────
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 Modell-Test ────────────────────────────────────────────────────
# pytest tests/ -v

HTL Pinkafeld – IF/IT | Data Science & AI | 12. Schulstufe