1. Projektordnerstruktur

Organisieren Sie Ihr Projekt wie folgt. Die Skripte erstellen automatisch die fehlenden Verzeichnisse (runs, dataset_split, onnx_models).

my_project/
│
├── data_raw/               # Quellordner (hier Ihre Bilder einfügen)
│   ├── class_0/            # Hier ist ein Foto der ersten Klasse
│   └── class_1/            # Hier ist ein Foto der zweiten Klasse
│
├── model.py                # Netzwerkarchitektur
├── prepare_data.py         # Datenaufteilungsskript (Trainings-/Validierungs-/Testdaten)
├── train.py                # Skript für das Training mit realen Bildern
├── export_onnx.py          # ONNX-Exportskript
├── predict_onnx.py         # Skript zum Überprüfen eines fertigen ONNX-Modells
└── requirements.txt        # Paketabhängigkeiten (Bibliotheken und Versionen)
                    

2. Erstellt eine virtuelle Umgebung

Wir werden mit venv arbeiten, Sie können aber natürlich auch conda, miniconda usw. verwenden.

python3 -m venv .venv
source .venv/bin/activate
                    

3. Installation der notwendigen Komponenten (requirements.txt)

Am besten verwendet man Versionen, mit denen alles kompatibel ist. Da sich Versionen jedoch ständig ändern, empfiehlt es sich, alles mit den neuesten Versionen zu erstellen.

absl-py==2.4.0
cuda-bindings==13.2.0
cuda-pathfinder==1.5.4
cuda-toolkit==13.0.2
filelock==3.29.0
flatbuffers==25.12.19
fsspec==2026.4.0
grpcio==1.80.0
Jinja2==3.1.6
Markdown==3.10.2
MarkupSafe==3.0.3
ml_dtypes==0.5.4
mpmath==1.3.0
networkx==3.6.1
numpy==2.4.6
nvidia-cublas==13.1.1.3
nvidia-cuda-cupti==13.0.85
nvidia-cuda-nvrtc==13.0.88
nvidia-cuda-runtime==13.0.96
nvidia-cudnn-cu13==9.20.0.48
nvidia-cufft==12.0.0.61
nvidia-cufile==1.15.1.6
nvidia-curand==10.4.0.35
nvidia-cusolver==12.0.4.66
nvidia-cusparse==12.6.3.3
nvidia-cusparselt-cu13==0.8.1
nvidia-nccl-cu13==2.29.7
nvidia-nvjitlink==13.0.88
nvidia-nvshmem-cu13==3.4.5
nvidia-nvtx==13.0.85
onnx==1.21.0
onnx-ir==0.2.1
onnxruntime==1.26.0
onnxscript==0.7.0
packaging==26.2
pillow==12.2.0
protobuf==7.34.1
setuptools==81.0.0
sympy==1.14.0
tensorboard==2.20.0
tensorboard-data-server==0.7.2
torch==2.12.0
torchvision==0.27.0
triton==3.7.0
typing_extensions==4.15.0
Werkzeug==3.1.8
                    

pip install -r requirements.txt
                    

4. Datenaufteilungsskript (prepare_data.py)

Dieses Skript nimmt Bilder aus data_raw/, mischt sie zufällig und teilt sie in drei isolierte Stichproben auf: Train (70%), Val (15%), Test (15%), wobei die Klassenstruktur erhalten bleibt.

import os
import shutil
import random

def split_dataset(src_dir="data_raw", dest_dir="dataset_split", train_p=0.7, val_p=0.15):
    if not os.path.exists(src_dir):
        os.makedirs(src_dir)
        print(f"Der Ordner '{src_dir}' wurde erstellt.")
        print("Bitte platzieren Sie Ihre Bilder in den Unterordnern der Klassen dort und führen Sie das Skript erneut aus.")
        return

    modes = ['train', 'val', 'test']
    classes = [d for d in os.listdir(src_dir) if os.path.isdir(os.path.join(src_dir, d))]

    for mode in modes:
        for cls in classes:
            os.makedirs(os.path.join(dest_dir, mode, cls), exist_ok=True)

    for cls in classes:
        cls_dir = os.path.join(src_dir, cls)
        images = [f for f in os.listdir(cls_dir) if os.path.isfile(os.path.join(cls_dir, f))]
        random.shuffle(images)

        num_img = len(images)
        tr_end = int(num_img * train_p)
        val_end = tr_end + int(num_img * val_p)

        splits = {
            'train': images[:tr_end],
            'val': images[tr_end:val_end],
            'test': images[val_end:]
        }

        for mode, file_list in splits.items():
            for file in file_list:
                src_file = os.path.join(src_dir, cls, file)
                dest_file = os.path.join(dest_dir, mode, cls, file)
                shutil.copy(src_file, dest_file)

        print(f"Klasse '{cls}': {num_img} Fotoaufteilung (Train: {len(splits['train'])}, Val: {len(splits['val'])}, Test: {len(splits['test'])})")

if __name__ == "__main__":
    random.seed(42)  # Wir fixieren den Seed für Reproduzierbarkeit.
    split_dataset()
                    

5. Netzwerkarchitektur (model.py)

Die Hauptnetzwerkarchitektur des AdaptiveFractalNet

import torch
import torch.nn as nn
import torch.nn.functional as F

class MandelbrotRouter(nn.Module):
    """
    Fraktaler Router auf Basis der Mandelbrot-Menge.

    Anstelle eines klassischen gelernten Softmax-Gewichts berechnen wir die
    Pfad-Gewichte über die Iteration z_{n+1} = z_n^2 + c. Jedes Bild im Batch
    wird auf zwei komplexe Zahlen (c, z0) projiziert; die Anzahl der
    Iterationen bis zur Divergenz (|z| > 2) bestimmt das fraktale Gewicht.
    """

    def __init__(self, in_channels, max_iter=16, escape_radius=2.0):
        super().__init__()
        self.max_iter = max_iter
        self.escape_radius_sq = escape_radius ** 2

        # Wir projizieren das Bild auf 4 reelle Zahlen:
        # (Re(c), Im(c), Re(z0), Im(z0)) — pro Bild im Batch.
        self.projector = nn.Sequential(
            nn.AdaptiveAvgPool2d((1, 1)),
            nn.Flatten(),
            nn.Linear(in_channels, 4),
            nn.Tanh(),   # begrenzt c und z0 auf [-1, 1] — relevanter Bereich der Menge
        )

    def forward(self, x):
        """
        x: [B, C, H, W]
        return: weights [B, 2] — normalisierte Gewichte (path1, path2).
        """
        batch_size = x.size(0)
        params = self.projector(x)  # [B, 4]

        # c und z0 als komplexe Werte, skaliert auf den interessanten Bereich
        # der Mandelbrot-Menge: Re(c) in [-2, 1], Im(c) in [-1.5, 1.5].
        c_re = params[:, 0] * 1.5 - 0.5   # [B]
        c_im = params[:, 1] * 1.5         # [B]
        z_re = params[:, 2] * 0.5         # [B]
        z_im = params[:, 3] * 0.5         # [B]

        # Wir verfolgen, in welcher Iteration der Punkt divergiert ist.
        # Anfangs: noch niemand divergiert, escape_iter = max_iter.
        escape_iter = torch.full(
            (batch_size,), float(self.max_iter),
            device=x.device, dtype=x.dtype,
        )
        diverged = torch.zeros(batch_size, dtype=torch.bool, device=x.device)

        for i in range(self.max_iter):
            # z = z^2 + c  (komplexe Multiplikation manuell)
            z_re_new = z_re * z_re - z_im * z_im + c_re
            z_im_new = 2.0 * z_re * z_im + c_im
            z_re, z_im = z_re_new, z_im_new

            magnitude_sq = z_re * z_re + z_im * z_im
            newly_diverged = (magnitude_sq > self.escape_radius_sq) & (~diverged)

            # *** Hier kommt torch.where zum Einsatz: ***
            # Falls der Punkt gerade jetzt divergiert ist, schreiben wir
            # die aktuelle Iterationsnummer; sonst behalten wir den alten Wert.
            escape_iter = torch.where(
                newly_diverged,
                torch.full_like(escape_iter, float(i + 1)),
                escape_iter,
            )
            diverged = diverged | newly_diverged

            # Für divergierte Punkte friert z ein, damit Overflow vermieden wird.
            z_re = torch.where(diverged, torch.zeros_like(z_re), z_re)
            z_im = torch.where(diverged, torch.zeros_like(z_im), z_im)

        # Normalisierte Escape-Zeit in [0, 1]:
        # - Punkt INNERHALB der Menge (nie divergiert) -> escape_iter = max_iter -> mu = 1.0
        # - Punkt schnell divergiert                   -> mu nahe 0.0
        mu = escape_iter / float(self.max_iter)  # [B]

        # Fraktales Gewicht: tiefer Pfad bekommt mehr Gewicht für Punkte
        # INNERHALB der Mandelbrot-Menge (komplexere Strukturen ⇒ tiefere
        # Verarbeitung). Einfacher Pfad bekommt mehr Gewicht für Punkte
        # AUSSERHALB (einfache Strukturen ⇒ flache Verarbeitung).
        w2 = mu                  # tiefer Pfad
        w1 = 1.0 - mu            # flacher Pfad
        weights = torch.stack([w1, w2], dim=1)  # [B, 2]
        return weights

class AdaptiveFractalBlock(nn.Module):
    """
    Fraktaler Block mit zwei Pfaden und Mandelbrot-basiertem Router.
    """
    def __init__(self, in_channels, out_channels, max_iter=16):
        super().__init__()

        # Pfad 1: einfache Standard-Faltung
        self.path1 = nn.Sequential(
            nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True),
        )

        # Pfad 2: tiefere fraktale Struktur (zwei aufeinanderfolgende Faltungen)
        self.path2 = nn.Sequential(
            nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True),
            nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True),
        )

        # Fraktaler Router auf Basis der Mandelbrot-Menge
        self.router = MandelbrotRouter(in_channels, max_iter=max_iter)

    def forward(self, x):
        batch_size = x.size(0)

        out_path1 = self.path1(x)
        out_path2 = self.path2(x)

        # Mandelbrot-basierte Gewichte
        weights = self.router(x)  # [B, 2]

        w1 = weights[:, 0].view(batch_size, 1, 1, 1)
        w2 = weights[:, 1].view(batch_size, 1, 1, 1)

        combined_output = w1 * out_path1 + w2 * out_path2
        return combined_output, weights


class AdaptiveFractalNet(nn.Module):
    """
    Hauptnetzwerkarchitektur des AdaptiveFractalNet mit Mandelbrot-Routing.
    """
    def __init__(self, num_classes=2):
        super().__init__()

        # Initiale Feature-Extraktion (Eingabe: 3 Farbkanäle)
        self.init_conv = nn.Conv2d(3, 32, kernel_size=3, padding=1)

        # Fraktale Blöcke
        self.fractal_block1 = AdaptiveFractalBlock(32, 32)
        self.pool1 = nn.MaxPool2d(2, 2)  # 128x128 -> 64x64

        self.fractal_block2 = AdaptiveFractalBlock(32, 64)
        self.pool2 = nn.MaxPool2d(2, 2)  # 64x64 -> 32x32

        # Globales Pooling vor dem Klassifikator
        self.global_pool = nn.AdaptiveAvgPool2d((1, 1))

        # Klassifikationsschicht
        self.classifier = nn.Linear(64, num_classes)

    def forward(self, x):
        x = F.relu(self.init_conv(x))

        x, weights1 = self.fractal_block1(x)
        x = self.pool1(x)

        x, weights2 = self.fractal_block2(x)
        x = self.pool2(x)

        x = self.global_pool(x)
        x = torch.flatten(x, 1)

        logits = self.classifier(x)

        # Wir geben die fraktalen Gewichte des letzten Blocks zurück.
        # Format: [Batch_Größe, Anzahl_Pfade] = [B, 2]
        return logits, weights2
                    

6. Skript für das Training mit realen Bildern (train.py)

Das Skript liest Dateien, die vom Skript prepare_data.py vorbereitet wurden.

import os
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
from torch.utils.tensorboard import SummaryWriter
from model import AdaptiveFractalNet

def main():
    # device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    device = "cpu"
    DATA_ROOT = "dataset_split"

    if not os.path.exists(DATA_ROOT):
        print(f"Fehler: Ordner '{DATA_ROOT}' nicht gefunden. Führen Sie zuerst 'prepare_data.py' aus.")
        return

    writer = SummaryWriter(log_dir="runs/adaptive_fractal_real_data")

    # Transformationen (Augmentation für Train, nur Normalisierung für Val)
    data_transforms = {
        'train': transforms.Compose([
            transforms.Resize((128, 128)),
            transforms.RandomHorizontalFlip(),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
        ]),
        'val': transforms.Compose([
            transforms.Resize((128, 128)),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
        ])
    }

    # Laden über ImageFolder
    train_dataset = datasets.ImageFolder(os.path.join(DATA_ROOT, 'train'), data_transforms['train'])
    val_dataset = datasets.ImageFolder(os.path.join(DATA_ROOT, 'val'), data_transforms['val'])

    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, num_workers=2)
    val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=2)

    model = AdaptiveFractalNet(num_classes=len(train_dataset.classes)).to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.AdamW(model.parameters(), lr=0.001, weight_decay=1e-4)

    global_step = 0
    epochs = 10

    for epoch in range(epochs):
        model.train()
        running_loss = 0.0

        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)
            optimizer.zero_grad()

            outputs, weights = model(images)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            running_loss += loss.item() * images.size(0)
            writer.add_scalar("Loss/Train_Batch", loss.item(), global_step)
            global_step += 1

        epoch_loss = running_loss / len(train_loader.dataset)
        writer.add_scalar("Loss/Train_Epoch", epoch_loss, epoch)

        # Validierung
        model.eval()
        correct, total = 0, 0
        with torch.no_grad():
            for images, labels in val_loader:
                images, labels = images.to(device), labels.to(device)
                outputs, _ = model(images)
                _, predicted = torch.max(outputs, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()

        accuracy = (correct / total) * 100
        writer.add_scalar("Accuracy/Validation", accuracy, epoch)
        print(f"Epoche [{epoch+1}/{epochs}] | Verlust: {epoch_loss:.4f} | Genauigkeitswert: {accuracy:.2f}%")

    torch.save(model.state_dict(), "adaptive_fractal_model.pth")
    writer.close()
    print("Training abgeschlossen. Modell gespeichert.")

if __name__ == "__main__":
    main()
                    

7. Skript exportieren nach ONNX (export_onnx.py)

Der Export eines neuronalen Netzes mit dynamischem fraktalem Routing hat eine wichtige Besonderheit: Da das Modell torch.where-Verzweigungen verwendet, um die Mandelbrot-Menge zu konstruieren, müssen wir das Modell mit einem stabilen Operationsgraphen exportieren, der dynamische Batchgrößen (dynamische Achsen) unterstützt.

import os
import torch
from model import AdaptiveFractalNet


def main():
    # 1. Speicherordner erstellen, falls er fehlt
    os.makedirs("onnx_models", exist_ok=True)

    # 2. Reinen CPU-Modus erzwingen
    device = torch.device("cpu")
    print("Export-Vorgang läuft im CPU-Modus...")

    # 3. Modell initialisieren und trainierte Gewichte auf die CPU laden
    # Hinweis: num_classes muss exakt der Anzahl deiner Klassen (z. B. 2) entsprechen.
    model = AdaptiveFractalNet(num_classes=2)

    model_path = "adaptive_fractal_model.pth"
    if not os.path.exists(model_path):
        print(f"Fehler: '{model_path}' nicht gefunden. "
              "Bitte trainieren Sie das Modell zuerst mit 'train.py'.")
        return

    model.load_state_dict(torch.load(model_path, map_location=device))
    model.to(device)
    model.eval()

    # 4. Dummy-Eingabe direkt auf der CPU erstellen
    # [Batch_Größe=1, Kanäle=3, Höhe=128, Breite=128]
    dummy_input = torch.randn(1, 3, 128, 128, device=device)

    # 5. ONNX-Export mit dynamischen Batch-Achsen.
    # Das Modell nutzt torch.where-Verzweigungen, um die Mandelbrot-Menge zu
    # konstruieren (siehe MandelbrotRouter in model.py). Der Export muss
    # daher mit einem stabilen Operationsgraphen erfolgen, der dynamische
    # Batchgrößen unterstützt.
    onnx_output_path = "onnx_models/fractal_net.onnx"

    try:
        torch.onnx.export(
            model,
            dummy_input,
            onnx_output_path,
            export_params=True,
            opset_version=17,            # Stabilste Version für ältere Umgebungen
            do_constant_folding=True,
            input_names=['input_images'],
            output_names=['logits', 'fractal_weights'],
            dynamic_axes={
                'input_images':    {0: 'batch_size'},
                'logits':          {0: 'batch_size'},
                'fractal_weights': {0: 'batch_size'},
            },
            dynamo=False,                # Schaltet den neuen Exporter aus
        )
        print(f"Erfolg: Das ONNX-Modell wurde unter '{onnx_output_path}' gespeichert!")
    except Exception as e:
        print(f"Fehler beim ONNX-Export: {e}")


if __name__ == "__main__":
    main()
                    

8. Ausführen und Testen des ONNX-Modells (predict_onnx.py)

Dieses Skript benötigt PyTorch nicht mehr. Das Netzwerk läuft auf der ultraschnellen, unabhängigen Laufzeitumgebung onnxruntime.

import onnxruntime as ort
import numpy as np

def run_onnx_inference(onnx_path="onnx_models/fractal_net.onnx"):
    # 1. Erstellen Sie eine ONNX-Runtime-Sitzung
    session = ort.InferenceSession(onnx_path)

    # 2. Generieren Sie einen Test-Batch (z. B. der Größe 2) direkt über NumPy.
    # Simulation vorverarbeiteter Bilder [2, 3, 128, 128]
    fake_images = np.random.randn(2, 3, 128, 128).astype(np.float32)

    #3. Schlussfolgerung durchführen.
    # Die Namen der Eingabedaten müssen exakt mit den input_names aus export_onnx.py übereinstimmen.
    inputs = {session.get_inputs()[0].name: fake_images}
    outputs = session.run(None, inputs)

    logits, fractal_weights = outputs[0], outputs[1]

    print("\n--- Ergebnisse des ONNX-Modells ---")
    print(f"Ausgabeform der Logits: {logits.shape}")
    print(f"Form der Fraktalgewichte: {fractal_weights.shape}")
    print(f"Fraktale Gewichte für das erste Bild des Batches: {fractal_weights[0]}")
    print(f"Fraktale Gewichte für das 2. Bild des Batches: {fractal_weights[1]}")

if __name__ == "__main__":
    run_onnx_inference()

Wie man ein CPU-Modell ohne teure CUDA und AWS verwendet, beispielsweise auf einem regulären dedizierten Server?

Wenn Sie interessiert sind, springen Sie zum nächsten Beitrag.