PyTorch-Modellbewertung und Debugging
Das Trainieren und Optimieren von Deep-Learning-Modellen erfordert systematische Evaluierungs- und Debugging-Methoden. Dieser Abschnitt führt ein, wie man Loss-Kurven analysiert, Overfitting und Underfitting erkennt, Konfusionsmatrizen zur Bewertung von Klassifikationsmodellen verwendet und gängige Debugging-Techniken einsetzt.
1. Analyse der Loss-Funktion
Die Loss-Funktion ist eine Kernmetrik während des Trainings und spiegelt die Lücke zwischen Modellvorhersagen und wahren Werten wider. Die korrekte Analyse der Loss-Kurve ist entscheidend für das Debugging von Modellen.
1.1 Vergleich von Trainings- und Validierungs-Loss
Durch den Vergleich von Trainings-Loss und Validierungs-Loss kann der Zustand des Modells bestimmt werden:
- Trainings-Loss sinkt, Validierungs-Loss sinkt: Normal, was darauf hinweist, dass das Modell lernt
- Trainings-Loss sinkt, Validierungs-Loss stabil oder steigt: Overfitting, Regularisierung erforderlich
- Trainings-Loss sinkt nicht: Problem mit der Lernrate oder der Modellarchitektur
- Trainings-Loss oszilliert: Lernrate zu groß oder Batch-Größe zu klein
1.2 Visualisierung der Verlustkurve
Beispiel
import matplotlib.pyplot as plt
import numpy as np
def plot_training_history(history):
"""
Trainingsverlauf plotten
history: ein dict, das 'train_loss', 'val_loss', 'train_acc', 'val_acc' enthält
"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# Verlustkurve
axes[0].plot(history['train_loss'], label='Trainingsverlust', alpha=0.8)
axes[0].plot(history['val_loss'], label='Validierungsverlust', alpha=0.8)
axes[0].set_xlabel('Epoche')
axes[0].set_ylabel('Verlust')
axes[0].set_title('Trainings- und Validierungsverlust')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# Genauigkeitskurve
axes[1].plot(history['train_acc'], label='Trainingsgenauigkeit', alpha=0.8)
axes[1].plot(history['val_acc'], label='Validierungsgenauigkeit', alpha=0.8)
axes[1].set_xlabel('Epoche')
axes[1].set_ylabel('Genauigkeit')
axes[1].set_title('Trainings- und Validierungsgenauigkeit')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Trainingsdaten simulieren
history = {
'train_loss': np.linspace(2.0, 0.2, 50) + np.random.normal(0, 0.05, 50),
'val_loss': np.linspace(2.0, 0.3, 50) + np.random.normal(0, 0.08, 50),
'train_acc': np.linspace(0.3, 0.95, 50),
'val_acc': np.linspace(0.3, 0.88, 50),
}
plot_training_history(history)
import numpy as np
def plot_training_history(history):
"""
Trainingsverlauf plotten
history: ein dict, das 'train_loss', 'val_loss', 'train_acc', 'val_acc' enthält
"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# Verlustkurve
axes[0].plot(history['train_loss'], label='Trainingsverlust', alpha=0.8)
axes[0].plot(history['val_loss'], label='Validierungsverlust', alpha=0.8)
axes[0].set_xlabel('Epoche')
axes[0].set_ylabel('Verlust')
axes[0].set_title('Trainings- und Validierungsverlust')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# Genauigkeitskurve
axes[1].plot(history['train_acc'], label='Trainingsgenauigkeit', alpha=0.8)
axes[1].plot(history['val_acc'], label='Validierungsgenauigkeit', alpha=0.8)
axes[1].set_xlabel('Epoche')
axes[1].set_ylabel('Genauigkeit')
axes[1].set_title('Trainings- und Validierungsgenauigkeit')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Trainingsdaten simulieren
history = {
'train_loss': np.linspace(2.0, 0.2, 50) + np.random.normal(0, 0.05, 50),
'val_loss': np.linspace(2.0, 0.3, 50) + np.random.normal(0, 0.08, 50),
'train_acc': np.linspace(0.3, 0.95, 50),
'val_acc': np.linspace(0.3, 0.88, 50),
}
plot_training_history(history)
1.3 Overfitting erkennen
Wenn der Validierungsverlust zu steigen beginnt, während der Trainingsverlust weiter sinkt, ist dies der Beginn von Overfitting. An diesem Punkt beginnt das Modell, Trainingsdaten "auswendig zu lernen", anstatt allgemeine Muster zu lernen.
Beispiel
import matplotlib.pyplot as plt
import numpy as np
def detect_overfitting(history, patience=5):
"""
Overfitting erkennen
patience: Anzahl aufeinanderfolgender Epochen mit steigendem Validierungsverlust, bevor das Training gestoppt wird
"""
val_loss = history['val_loss']
best_loss = float('inf')
best_epoch = 0
early_stop_epoch = None
for epoch in range(patience, len(val_loss)):
# Überprüfe die jüngsten Patience-Epochen
recent_losses = val_loss[epoch - patience:epoch]
current_loss = val_loss[epoch]
# Wenn der aktuelle Loss höher ist als der jüngste minimale Loss
if min(recent_losses) < current_loss:
early_stop_epoch = epoch
break
if current_loss < best_loss:
best_loss = current_loss
best_epoch = epoch
if early_stop_epoch:
print(f"Overfitting erkannt! Beste Epoche: {best_epoch}, sollte bei Epoche {early_stop_epoch} stoppen")
else:
print("Kein Overfitting erkannt")
return early_stop_epoch, best_epoch
# Overfitting-Beispiel
overfitting_history = {
'train_loss': np.linspace(1.5, 0.1, 30),
'val_loss': np.concatenate([np.linspace(1.5, 0.3, 15), np.linspace(0.3, 0.8, 15)]),
}
detect_overfitting(overfitting_history)
import numpy as np
def detect_overfitting(history, patience=5):
"""
Overfitting erkennen
patience: Anzahl aufeinanderfolgender Epochen mit steigendem Validierungsverlust, bevor das Training gestoppt wird
"""
val_loss = history['val_loss']
best_loss = float('inf')
best_epoch = 0
early_stop_epoch = None
for epoch in range(patience, len(val_loss)):
# Überprüfe die jüngsten Patience-Epochen
recent_losses = val_loss[epoch - patience:epoch]
current_loss = val_loss[epoch]
# Wenn der aktuelle Loss höher ist als der jüngste minimale Loss
if min(recent_losses) < current_loss:
early_stop_epoch = epoch
break
if current_loss < best_loss:
best_loss = current_loss
best_epoch = epoch
if early_stop_epoch:
print(f"Overfitting erkannt! Beste Epoche: {best_epoch}, sollte bei Epoche {early_stop_epoch} stoppen")
else:
print("Kein Overfitting erkannt")
return early_stop_epoch, best_epoch
# Overfitting-Beispiel
overfitting_history = {
'train_loss': np.linspace(1.5, 0.1, 30),
'val_loss': np.concatenate([np.linspace(1.5, 0.3, 15), np.linspace(0.3, 0.8, 15)]),
}
detect_overfitting(overfitting_history)
2. Umgang mit Overfitting und Underfitting
2.1 Lösungen für Overfitting
| Methode | Beschreibung | Code-Implementierung |
|---|---|---|
| Datenvolumen erhöhen | Mehr Trainingsdaten sammeln | Daten-Augmentierung |
| Dropout | Neuronen zufällig deaktivieren | nn.Dropout(0.5) |
| Gewichtsabnahme | L2-Regularisierung | weight_decay=1e-4 |
| Frühzeitiges Stoppen | Training beenden, wenn der Validierungsverlust nicht mehr sinkt | Patience-Parameter |
| Modellkomplexität reduzieren | Anzahl der Schichten oder Neuronen reduzieren | hidden_size reduzieren |
| Daten-Augmentierung | Zufällige Transformationen zur Erhöhung der Datenvielfalt | Rotation, Spiegeln, Zuschneiden |
2.2 Lösungen für Underfitting
| Methode | Beschreibung |
|---|---|
| Modellkomplexität erhöhen | Anzahl der Schichten oder Neuronen erhöhen |
| Länger trainieren | Anzahl der Trainings-Epochen erhöhen |
| Regularisierung verringern | Dropout oder weight_decay verringern |
| Lernrate anpassen | Eine größere Lernrate ausprobieren |
2.3 Implementierung des Early-Stopping-Mechanismus
Beispiel
import torch
import torch.nn as nn
import torch.optim as optim
class EarlyStopping:
"""
Early-Stopping-Mechanismus: Training beenden, wenn der Validierungsverlust über aufeinanderfolgende Epochen nicht sinkt
"""
def __init__(self, patience=7, min_delta=0, mode='min'):
"""
patience: Nach wie vielen aufeinanderfolgenden Epochen ohne Verbesserung abgebrochen wird
min_delta: die minimale Änderung, die als Verbesserung gilt
mode: 'min' oder 'max', ob die Metrik minimiert oder maximiert werden soll
"""
self.patience = patience
self.min_delta = min_delta
self.mode = mode
self.counter = 0
self.best_score = None
self.early_stop = False
def __call__(self, val_loss):
score = -val_loss if self.mode == 'min' else val_loss
if self.best_score is None:
self.best_score = score
elif score < self.best_score + self.min_delta:
self.counter += 1
if self.counter >= self.patience:
self.early_stop = True
else:
self.best_score = score
self.counter = 0
return self.early_stop
# Frühzeitiges Stoppen verwenden
def train_with_early_stopping(model, train_loader, val_loader, patience=7):
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
early_stopping = EarlyStopping(patience=patience, mode='min')
best_val_loss = float('inf')
best_model_state = None
for epoch in range(100):
# Trainieren
model.train()
train_loss = 0
for inputs, labels in train_loader:
inputs, labels = inputs.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
train_loss += loss.item()
# Validieren
model.eval()
val_loss = 0
with torch.no_grad():
for inputs, labels in val_loader:
inputs, labels = inputs.to(device), labels.to(device)
outputs = model(inputs)
loss = criterion(outputs, labels)
val_loss += loss.item()
val_loss /= len(val_loader)
print(f"Epoche {epoch+1}: Trainingsverlust={train_loss:.4f}, Validierungsverlust={val_loss:.4f}")
# Überprüfung des vorzeitigen Abbruchs
if early_stopping(val_loss):
print(f"Vorzeitiger Abbruch ausgelöst! Der Validierungsverlust ist seit {patience} aufeinanderfolgenden Epochen nicht gesunken")
# Bestes Modell speichern
if val_loss < best_val_loss:
best_val_loss = val_loss
best_model_state = model.state_dict().copy()
# Bestes Modell wiederherstellen
model.load_state_dict(best_model_state)
return model
import torch.nn as nn
import torch.optim as optim
class EarlyStopping:
"""
Early-Stopping-Mechanismus: Training beenden, wenn der Validierungsverlust über aufeinanderfolgende Epochen nicht sinkt
"""
def __init__(self, patience=7, min_delta=0, mode='min'):
"""
patience: Nach wie vielen aufeinanderfolgenden Epochen ohne Verbesserung abgebrochen wird
min_delta: die minimale Änderung, die als Verbesserung gilt
mode: 'min' oder 'max', ob die Metrik minimiert oder maximiert werden soll
"""
self.patience = patience
self.min_delta = min_delta
self.mode = mode
self.counter = 0
self.best_score = None
self.early_stop = False
def __call__(self, val_loss):
score = -val_loss if self.mode == 'min' else val_loss
if self.best_score is None:
self.best_score = score
elif score < self.best_score + self.min_delta:
self.counter += 1
if self.counter >= self.patience:
self.early_stop = True
else:
self.best_score = score
self.counter = 0
return self.early_stop
# Frühzeitiges Stoppen verwenden
def train_with_early_stopping(model, train_loader, val_loader, patience=7):
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
early_stopping = EarlyStopping(patience=patience, mode='min')
best_val_loss = float('inf')
best_model_state = None
for epoch in range(100):
# Trainieren
model.train()
train_loss = 0
for inputs, labels in train_loader:
inputs, labels = inputs.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
train_loss += loss.item()
# Validieren
model.eval()
val_loss = 0
with torch.no_grad():
for inputs, labels in val_loader:
inputs, labels = inputs.to(device), labels.to(device)
outputs = model(inputs)
loss = criterion(outputs, labels)
val_loss += loss.item()
val_loss /= len(val_loader)
print(f"Epoche {epoch+1}: Trainingsverlust={train_loss:.4f}, Validierungsverlust={val_loss:.4f}")
# Überprüfung des vorzeitigen Abbruchs
if early_stopping(val_loss):
print(f"Vorzeitiger Abbruch ausgelöst! Der Validierungsverlust ist seit {patience} aufeinanderfolgenden Epochen nicht gesunken")
# Bestes Modell speichern
if val_loss < best_val_loss:
best_val_loss = val_loss
best_model_state = model.state_dict().copy()
# Bestes Modell wiederherstellen
model.load_state_dict(best_model_state)
return model
3. Konfusionsmatrix
Die Konfusionsmatrix ist ein wichtiges Werkzeug zur Bewertung von Klassifikationsmodellen und zeigt die Beziehung zwischen Modellvorhersagen und wahren Labels.
3.1 Berechnung der Konfusionsmatrix
Beispiel
import torch
import numpy as np
from sklearn.metrics import confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
def calculate_confusion_matrix(model, dataloader, device, num_classes):
"""
Konfusionsmatrix berechnen
"""
model.eval()
all_preds = []
all_labels = []
with torch.no_grad():
for inputs, labels in dataloader:
inputs = inputs.to(device)
outputs = model(inputs)
_, preds = torch.max(outputs, 1)
all_preds.extend(preds.cpu().numpy())
all_labels.extend(labels.numpy())
cm = confusion_matrix(all_labels, all_preds)
return cm
def plot_confusion_matrix(cm, class_names):
"""
Konfusionsmatrix darstellen
"""
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
xticklabels=class_names, yticklabels=class_names)
plt.xlabel('Vorhergesagtes Label')
plt.ylabel('Tatsächliches Label')
plt.title('Konfusionsmatrix')
plt.tight_layout()
plt.show()
# Verwendungsbeispiel
# Angenommen, es gibt 10 Klassen
class_names = [f'Klasse {i}' for i in range(10)]
cm = np.array([
[45, 1, 0, 0, 0, 0, 2, 1, 1, 0],
[0, 42, 2, 0, 0, 0, 0, 3, 2, 1],
[1, 1, 38, 3, 0, 0, 0, 2, 3, 2],
[0, 0, 1, 41, 2, 1, 0, 1, 1, 3],
[0, 0, 0, 1, 44, 0, 1, 0, 2, 2],
[1, 0, 0, 0, 0, 43, 2, 1, 1, 2],
[2, 1, 0, 0, 1, 1, 39, 2, 2, 2],
[0, 2, 1, 1, 0, 0, 1, 40, 3, 2],
[1, 2, 2, 0, 1, 1, 1, 2, 36, 4],
[0, 1, 1, 2, 2, 2, 2, 1, 2, 37],
])
plot_confusion_matrix(cm, class_names)
import numpy as np
from sklearn.metrics import confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
def calculate_confusion_matrix(model, dataloader, device, num_classes):
"""
Konfusionsmatrix berechnen
"""
model.eval()
all_preds = []
all_labels = []
with torch.no_grad():
for inputs, labels in dataloader:
inputs = inputs.to(device)
outputs = model(inputs)
_, preds = torch.max(outputs, 1)
all_preds.extend(preds.cpu().numpy())
all_labels.extend(labels.numpy())
cm = confusion_matrix(all_labels, all_preds)
return cm
def plot_confusion_matrix(cm, class_names):
"""
Konfusionsmatrix darstellen
"""
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
xticklabels=class_names, yticklabels=class_names)
plt.xlabel('Vorhergesagtes Label')
plt.ylabel('Tatsächliches Label')
plt.title('Konfusionsmatrix')
plt.tight_layout()
plt.show()
# Verwendungsbeispiel
# Angenommen, es gibt 10 Klassen
class_names = [f'Klasse {i}' for i in range(10)]
cm = np.array([
[45, 1, 0, 0, 0, 0, 2, 1, 1, 0],
[0, 42, 2, 0, 0, 0, 0, 3, 2, 1],
[1, 1, 38, 3, 0, 0, 0, 2, 3, 2],
[0, 0, 1, 41, 2, 1, 0, 1, 1, 3],
[0, 0, 0, 1, 44, 0, 1, 0, 2, 2],
[1, 0, 0, 0, 0, 43, 2, 1, 1, 2],
[2, 1, 0, 0, 1, 1, 39, 2, 2, 2],
[0, 2, 1, 1, 0, 0, 1, 40, 3, 2],
[1, 2, 2, 0, 1, 1, 1, 2, 36, 4],
[0, 1, 1, 2, 2, 2, 2, 1, 2, 37],
])
plot_confusion_matrix(cm, class_names)
3.2 Evaluierungsmetriken für die Klassifikation
Aus der Konfusionsmatrix können mehrere Evaluierungsmetriken berechnet werden:
Beispiel
import numpy as np
def classification_metrics(cm):
"""
Berechnet verschiedene Evaluierungsmetriken aus der Konfusionsmatrix
"""
# Genauigkeit
accuracy = np.trace(cm) / np.sum(cm)
# Präzision, Recall, F1 für jede Klasse berechnen
num_classes = cm.shape[0]
precisions = []
recalls = []
f1s = []
for i in range(num_classes):
tp = cm[i, i]
fp = np.sum(cm[:, i]) - tp
fn = np.sum(cm[i, :]) - tp
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
precisions.append(precision)
recalls.append(recall)
f1s.append(f1)
# Makro-Durchschnitt
macro_precision = np.mean(precisions)
macro_recall = np.mean(recalls)
macro_f1 = np.mean(f1s)
# Gewichteter Durchschnitt
class_counts = np.sum(cm, axis=1)
weights = class_counts / np.sum(class_counts)
weighted_precision = np.average(precisions, weights=weights)
weighted_recall = np.average(recalls, weights=weights)
weighted_f1 = np.average(f1s, weights=weights)
return {
'accuracy': accuracy,
'macro_precision': macro_precision,
'macro_recall': macro_recall,
'macro_f1': macro_f1,
'weighted_precision': weighted_precision,
'weighted_recall': weighted_recall,
'weighted_f1': weighted_f1,
}
# Metriken berechnen
metrics = classification_metrics(cm)
print("=" * 50)
print("Metriken zur Klassifikationsbewertung")
print("=" * 50)
print(f"Genauigkeit: {metrics['accuracy']:.4f}")
print(f"Makro-Präzision: {metrics['macro_precision']:.4f}")
print(f"Makro-Recall: {metrics['macro_recall']:.4f}")
print(f"Makro-F1: {metrics['macro_f1']:.4f}")
print(f"Gewichtete Precision: {metrics['weighted_precision']:.4f}")
print(f"Gewichteter Recall: {metrics['weighted_recall']:.4f}")
print(f"Gewichteter F1: {metrics['weighted_f1']:.4f}")
def classification_metrics(cm):
"""
Berechnet verschiedene Evaluierungsmetriken aus der Konfusionsmatrix
"""
# Genauigkeit
accuracy = np.trace(cm) / np.sum(cm)
# Präzision, Recall, F1 für jede Klasse berechnen
num_classes = cm.shape[0]
precisions = []
recalls = []
f1s = []
for i in range(num_classes):
tp = cm[i, i]
fp = np.sum(cm[:, i]) - tp
fn = np.sum(cm[i, :]) - tp
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
precisions.append(precision)
recalls.append(recall)
f1s.append(f1)
# Makro-Durchschnitt
macro_precision = np.mean(precisions)
macro_recall = np.mean(recalls)
macro_f1 = np.mean(f1s)
# Gewichteter Durchschnitt
class_counts = np.sum(cm, axis=1)
weights = class_counts / np.sum(class_counts)
weighted_precision = np.average(precisions, weights=weights)
weighted_recall = np.average(recalls, weights=weights)
weighted_f1 = np.average(f1s, weights=weights)
return {
'accuracy': accuracy,
'macro_precision': macro_precision,
'macro_recall': macro_recall,
'macro_f1': macro_f1,
'weighted_precision': weighted_precision,
'weighted_recall': weighted_recall,
'weighted_f1': weighted_f1,
}
# Metriken berechnen
metrics = classification_metrics(cm)
print("=" * 50)
print("Metriken zur Klassifikationsbewertung")
print("=" * 50)
print(f"Genauigkeit: {metrics['accuracy']:.4f}")
print(f"Makro-Präzision: {metrics['macro_precision']:.4f}")
print(f"Makro-Recall: {metrics['macro_recall']:.4f}")
print(f"Makro-F1: {metrics['macro_f1']:.4f}")
print(f"Gewichtete Precision: {metrics['weighted_precision']:.4f}")
print(f"Gewichteter Recall: {metrics['weighted_recall']:.4f}")
print(f"Gewichteter F1: {metrics['weighted_f1']:.4f}")
3.3 Klassifikationsbericht
Beispiel
from sklearn.metrics import classification_report
# Verwende sklearns Klassifikationsbericht
y_true = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] * 10 # Wahre Labels simulieren
y_pred = [0, 1, 2, 3, 4, 5, 6, 7, 8, 8] * 10 # Vorhersagen simulieren
report = classification_report(y_true, y_pred, digits=4)
print(report)
# Verwende sklearns Klassifikationsbericht
y_true = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] * 10 # Wahre Labels simulieren
y_pred = [0, 1, 2, 3, 4, 5, 6, 7, 8, 8] * 10 # Vorhersagen simulieren
report = classification_report(y_true, y_pred, digits=4)
print(report)
4. Lernratenplanung
Die Lernrate ist einer der wichtigsten Hyperparameter für das Training von Deep-Learning-Modellen. Eine geeignete Lernratenplanung kann die Trainingseffektivität erheblich verbessern.
4.1 Strategien zur Lernratenplanung
| Strategie | Beschreibung | Anwendbare Szenarien |
|---|---|---|
| StepLR | Fester Schrittzerfall | Wenn die optimale Lernrate bekannt ist |
| MultiStepLR | Zerfall bei festgelegten Epochen | Nicht gleichmäßiger Zerfall |
| CosineAnnealingLR | Kosinus-Annealing | Sanfte Konvergenz |
| ReduceLROnPlateau | Zerfall, wenn der Validierungsverlust nicht mehr sinkt | Automatisches Tuning |
| Aufwärmphase | Zuerst erhöhen, dann verringern | Frühes Training stabilisieren |
4.2 Implementierung der Lernraten-Zeitplanung
Beispiel
import torch
import torch.optim as optim
import matplotlib.pyplot as plt
# Trainingsepochen simulieren
epochs = 50
# 1. Step LR: alle 10 Epochen um die Hälfte reduzieren
scheduler_step = optim.lr_scheduler.StepLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
step_size=10, gamma=0.5
)
# 2. MultiStep LR: Reduzierung an angegebenen Epochen
scheduler_multistep = optim.lr_scheduler.MultiStepLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
milestones=[15, 30, 45], gamma=0.1
)
# 3. Kosinus-Annealing
scheduler_cosine = optim.lr_scheduler.CosineAnnealingLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
T_max=50
)
# 4. ReduceLROnPlateau
scheduler_plateau = optim.lr_scheduler.ReduceLROnPlateau(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
mode='min', factor=0.5, patience=5
)
# Lernratenkurven plotten
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# Step LR
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler_step.step()
axes[0, 0].plot(lr_history)
axes[0, 0].set_title('Step LR')
# Neu initialisieren
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=[15, 30, 45], Gamma=0.1)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[0, 1].plot(lr_history)
axes[0, 1].set_title('MultiStep LR')
# Kosinus
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[1, 0].plot(lr_history)
axes[1, 0].set_title('Cosine Annealing LR')
# Warmup + Kosinus
def warmup_cosine(optimizer, warmup_epochs, total_epochs, min_lr=1e-6):
def lr_lambda(epoch):
if epoch < warmup_epochs:
return epoch / warmup_epochs
return min_lr + 0.5 * (1 - min_lr) * (1 + np.cos(np.pi * (epoch - warmup_epochs) / (total_epochs - warmup_epochs)))
return optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = warmup_cosine(optimizer, warmup_epochs=5, total_epochs=50)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[1, 1].plot(lr_history)
axes[1, 1].set_title('Warmup + Kosinus')
for ax in axes.flat:
ax.set_xlabel('Epoche')
ax.set_ylabel('Lernrate')
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
import torch.optim as optim
import matplotlib.pyplot as plt
# Trainingsepochen simulieren
epochs = 50
# 1. Step LR: alle 10 Epochen um die Hälfte reduzieren
scheduler_step = optim.lr_scheduler.StepLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
step_size=10, gamma=0.5
)
# 2. MultiStep LR: Reduzierung an angegebenen Epochen
scheduler_multistep = optim.lr_scheduler.MultiStepLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
milestones=[15, 30, 45], gamma=0.1
)
# 3. Kosinus-Annealing
scheduler_cosine = optim.lr_scheduler.CosineAnnealingLR(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
T_max=50
)
# 4. ReduceLROnPlateau
scheduler_plateau = optim.lr_scheduler.ReduceLROnPlateau(
optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1),
mode='min', factor=0.5, patience=5
)
# Lernratenkurven plotten
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# Step LR
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler_step.step()
axes[0, 0].plot(lr_history)
axes[0, 0].set_title('Step LR')
# Neu initialisieren
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=[15, 30, 45], Gamma=0.1)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[0, 1].plot(lr_history)
axes[0, 1].set_title('MultiStep LR')
# Kosinus
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[1, 0].plot(lr_history)
axes[1, 0].set_title('Cosine Annealing LR')
# Warmup + Kosinus
def warmup_cosine(optimizer, warmup_epochs, total_epochs, min_lr=1e-6):
def lr_lambda(epoch):
if epoch < warmup_epochs:
return epoch / warmup_epochs
return min_lr + 0.5 * (1 - min_lr) * (1 + np.cos(np.pi * (epoch - warmup_epochs) / (total_epochs - warmup_epochs)))
return optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
optimizer = optim.SGD(torch.nn.Linear(10, 10).parameters(), lr=0.1)
scheduler = warmup_cosine(optimizer, warmup_epochs=5, total_epochs=50)
lr_history = []
for _ in range(epochs):
lr_history.append(optimizer.param_groups[0]['lr'])
scheduler.step()
axes[1, 1].plot(lr_history)
axes[1, 1].set_title('Warmup + Kosinus')
for ax in axes.flat:
ax.set_xlabel('Epoche')
ax.set_ylabel('Lernrate')
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
5. Diagnose von Gradientenproblemen
5.1 Verschwindende und explodierende Gradienten
< h2 class="example">Beispiel
import torch
import torch.nn as nn
def analyze_gradients(model):
"""
Analysiere die Gradienten jeder Schicht im Modell
"""
grad_stats = {}
for name, param in model.named_parameters():
if param.grad is not None:
grad = param.grad
grad_stats[name] = {
'mean': grad.mean().item(),
'std': grad.std().item(),
'max': grad.abs().max().item(),
'min': grad.abs().min().item(),
'norm': grad.norm().item(),
}
return grad_stats
def detect_gradient_issues(model):
"""
Gradientprobleme erkennen
"""
issues = []
grad_norms = []
for param in model.parameters():
if param.grad is not None:
grad_norms.append(param.grad.norm().item())
avg_grad_norm = sum(grad_norms) / len(grad_norms) if grad_norms else 0
if avg_grad_norm < 1e-7:
issues.append("Gradient verschwindet: Gradienten sind zu klein, das Modell kann möglicherweise nicht lernen")
elif avg_grad_norm > 100:
issues.append("Gradientenexplosion: Gradienten sind zu groß, das Training ist instabil")
return issues, avg_grad_norm
# Beispiel: Gradienten erkennen
model = nn.Sequential(
nn.Linear(100, 50),
nn.ReLU(),
nn.Linear(50, 50),
nn.ReLU(),
nn.Linear(50, 10)
)
# Vorwärts- und Rückwärtspropagation simulieren
x = torch.randn(32, 100)
y = model(x)
loss = y.sum()
loss.backward()
issues, avg_norm = detect_gradient_issues(model)
print(f"Durchschnittliche Gradientennorm: {avg_norm:.6f}")
if issues:
for issue in issues:
print(f"Warnung: {issue}")
else:
print("Gradientenstatus ist normal")
import torch.nn as nn
def analyze_gradients(model):
"""
Analysiere die Gradienten jeder Schicht im Modell
"""
grad_stats = {}
for name, param in model.named_parameters():
if param.grad is not None:
grad = param.grad
grad_stats[name] = {
'mean': grad.mean().item(),
'std': grad.std().item(),
'max': grad.abs().max().item(),
'min': grad.abs().min().item(),
'norm': grad.norm().item(),
}
return grad_stats
def detect_gradient_issues(model):
"""
Gradientprobleme erkennen
"""
issues = []
grad_norms = []
for param in model.parameters():
if param.grad is not None:
grad_norms.append(param.grad.norm().item())
avg_grad_norm = sum(grad_norms) / len(grad_norms) if grad_norms else 0
if avg_grad_norm < 1e-7:
issues.append("Gradient verschwindet: Gradienten sind zu klein, das Modell kann möglicherweise nicht lernen")
elif avg_grad_norm > 100:
issues.append("Gradientenexplosion: Gradienten sind zu groß, das Training ist instabil")
return issues, avg_grad_norm
# Beispiel: Gradienten erkennen
model = nn.Sequential(
nn.Linear(100, 50),
nn.ReLU(),
nn.Linear(50, 50),
nn.ReLU(),
nn.Linear(50, 10)
)
# Vorwärts- und Rückwärtspropagation simulieren
x = torch.randn(32, 100)
y = model(x)
loss = y.sum()
loss.backward()
issues, avg_norm = detect_gradient_issues(model)
print(f"Durchschnittliche Gradientennorm: {avg_norm:.6f}")
if issues:
for issue in issues:
print(f"Warnung: {issue}")
else:
print("Gradientenstatus ist normal")
5.2 Gradienten-Clipping
Beispiel
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.utils as utils
# Beispiel für Gradienten-Clipping
def train_with_gradient_clipping(model, dataloader, max_norm=1.0):
"""
Mit Gradienten-Clipping trainieren
max_norm: maximale Gradientennorm; Gradienten, die diesen Wert überschreiten, werden geclippt
"""
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
model.train()
for inputs, labels in dataloader:
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
# Gradientenclipping: Gradientenexplosion verhindern
utils.clip_grad_norm_(model.parameters(), max_norm=max_norm)
optimizer.step()
return model
# Elementweises Clipping (konservativer)
def clip_grad_by_value(model, clip_value=1.0):
"""
Gradienten nach Wert beschneiden
"""
for param in model.parameters():
if param.grad is not None:
param.grad.data.clamp_(min=-clip_value, max=clip_value)
import torch.nn as nn
import torch.optim as optim
import torch.nn.utils as utils
# Beispiel für Gradienten-Clipping
def train_with_gradient_clipping(model, dataloader, max_norm=1.0):
"""
Mit Gradienten-Clipping trainieren
max_norm: maximale Gradientennorm; Gradienten, die diesen Wert überschreiten, werden geclippt
"""
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
model.train()
for inputs, labels in dataloader:
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
# Gradientenclipping: Gradientenexplosion verhindern
utils.clip_grad_norm_(model.parameters(), max_norm=max_norm)
optimizer.step()
return model
# Elementweises Clipping (konservativer)
def clip_grad_by_value(model, clip_value=1.0):
"""
Gradienten nach Wert beschneiden
"""
for param in model.parameters():
if param.grad is not None:
param.grad.data.clamp_(min=-clip_value, max=clip_value)
6. Analyse der Modellleistung
6.1 Berechnung der Parameteranzahl und FLOPs
Beispiel
import torch
import torch.nn as nn
from thop import profile
def count_parameters(model):
"""Berechnet die Anzahl der trainierbaren Parameter im Modell"""
total = sum(p.numel() for p in model.parameters())
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
return total, trainable
def count_flops(model, input_size=(1, 3, 224, 224)):
"""Berechnet FLOPs (erfordert die thop-Bibliothek)"""
input_tensor = torch.randn(input_size)
flops, params = profile(model, inputs=(input_tensor,), verbose=False)
return flops, params
# Beispiel: Modell analysieren
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2),
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2),
nn.Conv2d(128, 256, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(256, 10)
)
total, trainable = count_parameters(model)
print(f"Gesamtparameter: {total:,}")
print(f"Trainierbare Parameter: {trainable:,}")
print(f"Modellgröße: {total * 4 / 1024 / 1024:.2f} MB")
# FLOPs-Berechnung
try:
flops, _ = count_flops(model)
print(f"FLOPs: {flops / 1e9:.2f} G")
except ImportError:
print("Bitte installiere die thop-Bibliothek: pip install thop")
import torch.nn as nn
from thop import profile
def count_parameters(model):
"""Berechnet die Anzahl der trainierbaren Parameter im Modell"""
total = sum(p.numel() for p in model.parameters())
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
return total, trainable
def count_flops(model, input_size=(1, 3, 224, 224)):
"""Berechnet FLOPs (erfordert die thop-Bibliothek)"""
input_tensor = torch.randn(input_size)
flops, params = profile(model, inputs=(input_tensor,), verbose=False)
return flops, params
# Beispiel: Modell analysieren
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2),
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2),
nn.Conv2d(128, 256, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(256, 10)
)
total, trainable = count_parameters(model)
print(f"Gesamtparameter: {total:,}")
print(f"Trainierbare Parameter: {trainable:,}")
print(f"Modellgröße: {total * 4 / 1024 / 1024:.2f} MB")
# FLOPs-Berechnung
try:
flops, _ = count_flops(model)
print(f"FLOPs: {flops / 1e9:.2f} G")
except ImportError:
print("Bitte installiere die thop-Bibliothek: pip install thop")
6.2 Test der Inferenzgeschwindigkeit
< h2 class="example">Beispiel
import torch
import time
def measure_inference_speed(model, input_size, device='cuda', num_iterations=100):
"""
Inferenzgeschwindigkeit messen
"""
model.eval()
model = model.to(device)
# Aufwärmen
dummy_input = torch.randn(input_size).to(device)
with torch.no_grad():
for _ in range(10):
_ = model(dummy_input)
# Zeitmessung
start = time.time()
with torch.no_grad():
for _ in range(num_iterations):
_ = model(dummy_input)
if device == 'cuda':
torch.cuda.synchronize()
end = time.time()
avg_time = (end - start) / num_iterations * 1000 # ms
return avg_time
# Verwendungsbeispiel
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(64, 10)
)
# CPU-Inferenzgeschwindigkeit
cpu_time = measure_inference_speed(model, (1, 3, 224, 224), device='cpu')
print(f"CPU-Inferenzzeit: {cpu_time:.2f} ms/Bild")
# GPU-Inferenzgeschwindigkeit (falls CUDA verfügbar ist)
if torch.cuda.is_available():
gpu_time = measure_inference_speed(model, (1, 3, 224, 224), device='cuda')
print(f"GPU-Inferenzzeit: {gpu_time:.2f} ms/Bild")
import time
def measure_inference_speed(model, input_size, device='cuda', num_iterations=100):
"""
Inferenzgeschwindigkeit messen
"""
model.eval()
model = model.to(device)
# Aufwärmen
dummy_input = torch.randn(input_size).to(device)
with torch.no_grad():
for _ in range(10):
_ = model(dummy_input)
# Zeitmessung
start = time.time()
with torch.no_grad():
for _ in range(num_iterations):
_ = model(dummy_input)
if device == 'cuda':
torch.cuda.synchronize()
end = time.time()
avg_time = (end - start) / num_iterations * 1000 # ms
return avg_time
# Verwendungsbeispiel
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(64, 10)
)
# CPU-Inferenzgeschwindigkeit
cpu_time = measure_inference_speed(model, (1, 3, 224, 224), device='cpu')
print(f"CPU-Inferenzzeit: {cpu_time:.2f} ms/Bild")
# GPU-Inferenzgeschwindigkeit (falls CUDA verfügbar ist)
if torch.cuda.is_available():
gpu_time = measure_inference_speed(model, (1, 3, 224, 224), device='cuda')
print(f"GPU-Inferenzzeit: {gpu_time:.2f} ms/Bild")
6.3 Analyse der Speichernutzung
Beispiel
import torch
def analyze_memory(model, input_size):
"""
Analysiere den Speicherverbrauch des Modells
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
# Eingabe erstellen
x = torch.randn(input_size).to(device)
# Cache leeren
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.reset_peak_memory_stats()
# Vorwärtspropagierung
with torch.no_grad():
output = model(x)
# Speicherstatistiken abrufen
if torch.cuda.is_available():
allocated = torch.cuda.memory_allocated() / 1024**2 # MB
reserved = torch.cuda.memory_reserved() / 1024**2
max_allocated = torch.cuda.max_memory_allocated() / 1024**2
print(f"Aktuell zugewiesen: {allocated:.2f} MB")
print(f"Reservierter Speicher: {reserved:.2f} MB")
print(f"Spitzenauslastung: {max_allocated:.2f} MB")
else:
print("CUDA-Unterstützung erforderlich")
return output
# Verwendungsbeispiel
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(128, 256, kernel_size=3, padding=1),
nn.ReLU(),
)
analyze_memory(model, (8, 3, 224, 224))
def analyze_memory(model, input_size):
"""
Analysiere den Speicherverbrauch des Modells
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
# Eingabe erstellen
x = torch.randn(input_size).to(device)
# Cache leeren
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.reset_peak_memory_stats()
# Vorwärtspropagierung
with torch.no_grad():
output = model(x)
# Speicherstatistiken abrufen
if torch.cuda.is_available():
allocated = torch.cuda.memory_allocated() / 1024**2 # MB
reserved = torch.cuda.memory_reserved() / 1024**2
max_allocated = torch.cuda.max_memory_allocated() / 1024**2
print(f"Aktuell zugewiesen: {allocated:.2f} MB")
print(f"Reservierter Speicher: {reserved:.2f} MB")
print(f"Spitzenauslastung: {max_allocated:.2f} MB")
else:
print("CUDA-Unterstützung erforderlich")
return output
# Verwendungsbeispiel
model = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(),
nn.Conv2d(128, 256, kernel_size=3, padding=1),
nn.ReLU(),
)
analyze_memory(model, (8, 3, 224, 224))
7. Häufige Probleme und Debugging-Tipps
7.1 Kurzübersicht für Trainingsprobleme
| Symptom | Mögliche Ursache | Lösung |
|---|---|---|
| Loss nimmt nicht ab | Lernrate zu klein/groß, Gradientenprobleme | Lernrate anpassen, Gradienten prüfen |
| Loss-Oszillation | Lernrate zu groß, kleine Batch-Größe | Lernrate verringern, Batch-Größe erhöhen |
| NaN tritt auf | Division durch null, log(0), Gradientenexplosion | Epsilon hinzufügen, Gradienten-Clipping |
| Überanpassung (Overfitting) | Komplexes Modell, unzureichende Daten | Mehr Daten hinzufügen, Regularisierung hinzufügen |
| Unteranpassung | Modell zu einfach, unzureichendes Training | Modellgröße erhöhen, länger trainieren |
| Unzureichender GPU-Speicher | Große Batch-Größe, großes Modell | Batch-Größe reduzieren, Gradientenakkumulation |
7.2 Sammlung von Debugging-Tipps
Beispiel
import torch
import random
import numpy as np
def set_seed(seed=42):
"""Setze den Zufallsseed, um reproduzierbare Ergebnisse zu gewährleisten"""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
def debug_forward(model, x):
"""Debugge die Vorwärtspropagation und überprüfe die Ausgaben jeder Schicht"""
print("=" * 50)
print("Debugging der Vorwärtspropagation")
print("=" * 50)
for name, layer in model.named_children():
x = Schicht(x)
if hasattr(x, 'shape'):
print(f"{name}: shape={x.shape}, ", end='')
if hasattr(x, 'mean'):
print(f"mean={x.mean().item():.4f}, std={x.std().item():.4f}")
if torch.isnan(x).any():
print(f" ⚠️ NaN erkannt!")
if torch.isinf(x).any():
print(f" ⚠️ Inf erkannt!")
print("=" * 50)
def debug_backward(model):
"""Backward-Propagation debuggen und Gradienten prüfen"""
print("=" * 50)
print("Backward-Propagation-Debugging")
print("=" * 50)
for name, param in model.named_parameters():
if param.grad is not None:
grad_norm = param.grad.norm().item()
if torch.isnan(param.grad).any():
print(f"{name}: ⚠️ Gradient ist NaN!")
elif torch.isinf(param.grad).any():
print(f"{name}: ⚠️ Gradient ist Inf!")
elif grad_norm > 10:
print(f"{name}: ⚠️ Gradientenexplosion! norm={grad_norm:.2f}")
elif grad_norm < 1e-7:
print(f"{name}: ⚠️ Verschwindender Gradient! norm={grad_norm:.2e}")
else:
print(f"{name}: normal norm={grad_norm:.4f}")
print("=" * 50)
# Verwendungsbeispiel
set_seed(42)
model = nn.Sequential(
nn.Linear(10, 20),
nn.ReLU(),
nn.Linear(20, 20),
nn.ReLU(),
nn.Linear(20, 5)
)
x = torch.randn(2, 10)
y = model(x)
debug_forward(model, x)
loss = y.sum()
loss.backward()
debug_backward(model)
import random
import numpy as np
def set_seed(seed=42):
"""Setze den Zufallsseed, um reproduzierbare Ergebnisse zu gewährleisten"""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
def debug_forward(model, x):
"""Debugge die Vorwärtspropagation und überprüfe die Ausgaben jeder Schicht"""
print("=" * 50)
print("Debugging der Vorwärtspropagation")
print("=" * 50)
for name, layer in model.named_children():
x = Schicht(x)
if hasattr(x, 'shape'):
print(f"{name}: shape={x.shape}, ", end='')
if hasattr(x, 'mean'):
print(f"mean={x.mean().item():.4f}, std={x.std().item():.4f}")
if torch.isnan(x).any():
print(f" ⚠️ NaN erkannt!")
if torch.isinf(x).any():
print(f" ⚠️ Inf erkannt!")
print("=" * 50)
def debug_backward(model):
"""Backward-Propagation debuggen und Gradienten prüfen"""
print("=" * 50)
print("Backward-Propagation-Debugging")
print("=" * 50)
for name, param in model.named_parameters():
if param.grad is not None:
grad_norm = param.grad.norm().item()
if torch.isnan(param.grad).any():
print(f"{name}: ⚠️ Gradient ist NaN!")
elif torch.isinf(param.grad).any():
print(f"{name}: ⚠️ Gradient ist Inf!")
elif grad_norm > 10:
print(f"{name}: ⚠️ Gradientenexplosion! norm={grad_norm:.2f}")
elif grad_norm < 1e-7:
print(f"{name}: ⚠️ Verschwindender Gradient! norm={grad_norm:.2e}")
else:
print(f"{name}: normal norm={grad_norm:.4f}")
print("=" * 50)
# Verwendungsbeispiel
set_seed(42)
model = nn.Sequential(
nn.Linear(10, 20),
nn.ReLU(),
nn.Linear(20, 20),
nn.ReLU(),
nn.Linear(20, 5)
)
x = torch.randn(2, 10)
y = model(x)
debug_forward(model, x)
loss = y.sum()
loss.backward()
debug_backward(model)
8. PyTorch Profiler verwenden
PyTorch Profiler ist das offizielle Werkzeug zur Leistungsanalyse, das den Zeitverbrauch und die Speichernutzung jedes Modellteils analysieren kann.
8.1 Grundlegende Verwendung des Profilers
Beispiel
import torch
import torch.nn as nn
from torch.profiler import profile, ProfilerActivity, schedule
# Einfaches Modell
model = nn.Sequential(
nn.Conv2d(3, 64, 3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 64, 3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(64, 10)
).cuda()
optimizer = torch.optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss()
# Profiler verwenden
with profile(
activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
schedule=schedule(wait=1, warmup=1, active=3, repeat=1),
on_trace_ready=torch.profiler.tensorboard_trace_handler('./logs'),
record_shapes=True,
profile_memory=True,
with_stack=True
) as prof:
for step in range(5):
inputs = torch.randn(8, 3, 224, 224).cuda()
labels = torch.randint(0, 10, (8,)).cuda()
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
prof.step()
# Ergebnisse ausgeben
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
import torch.nn as nn
from torch.profiler import profile, ProfilerActivity, schedule
# Einfaches Modell
model = nn.Sequential(
nn.Conv2d(3, 64, 3, padding=1),
nn.ReLU(),
nn.Conv2d(64, 64, 3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(64, 10)
).cuda()
optimizer = torch.optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss()
# Profiler verwenden
with profile(
activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
schedule=schedule(wait=1, warmup=1, active=3, repeat=1),
on_trace_ready=torch.profiler.tensorboard_trace_handler('./logs'),
record_shapes=True,
profile_memory=True,
with_stack=True
) as prof:
for step in range(5):
inputs = torch.randn(8, 3, 224, 224).cuda()
labels = torch.randint(0, 10, (8,)).cuda()
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
prof.step()
# Ergebnisse ausgeben
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
8.2 Visualisierung von Profiler-Ergebnissen
Beispiel
# Chrome-Trace-Datei exportieren
# Das generierte Verzeichnis ./logs kann im Chrome-Browser geöffnet und angezeigt werden
# Oder direkt verschiedene Metriken ausgeben
print("=" * 80)
print("CPU-Zeit Top 10:")
print(prof.key_averages().table(sort_by="cpu_time_total", row_limit=10))
print("\n" + "=" * 80)
print("CUDA-Zeit Top 10:")
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
print("\n" + "=" * 80)
print("Speichernutzung Top 10:")
print(prof.key_averages().table(sort_by="self_cuda_memory_usage", row_limit=10))
# Das generierte Verzeichnis ./logs kann im Chrome-Browser geöffnet und angezeigt werden
# Oder direkt verschiedene Metriken ausgeben
print("=" * 80)
print("CPU-Zeit Top 10:")
print(prof.key_averages().table(sort_by="cpu_time_total", row_limit=10))
print("\n" + "=" * 80)
print("CUDA-Zeit Top 10:")
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
print("\n" + "=" * 80)
print("Speichernutzung Top 10:")
print(prof.key_averages().table(sort_by="self_cuda_memory_usage", row_limit=10))