import pandas as pd import numpy as np import torch import torch.nn as nn from torch.utils.data import DataLoader, Dataset from sklearn.model_selection import train_test_split from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import matplotlib.pyplot as plt import matplotlib.patches as mpatches from tqdm import tqdm from dataset_generator import create_embedding_matrix from EarlyStopping import EarlyStopping # 1. Gerät automatisch erkennen (MPS, CUDA oder CPU) device = torch.device('mps' if torch.backends.mps.is_available() else 'cuda' if torch.cuda.is_available() else 'cpu') print(f"Using device: {device}") # 2. Daten laden data = pd.read_csv('data/hack.csv') # 3. Filtern humorvoller Texte humor_data = data[data['is_humor'] == 1].dropna(subset=['humor_rating']).copy() # 4. Einbettungsmatrix erstellen embedding_matrix, word_index, vocab_size, d_model = create_embedding_matrix( gloVe_path='data/glove.6B.100d.txt', emb_len=100 ) print(f"vocab_size: {vocab_size}, d_model: {d_model}") # 5. Tokenisierung und Padding mit PyTorch def tokenize_and_pad(texts, word_index, max_len=50): sequences = [] for text in texts: tokens = [word_index.get(word, 0) for word in text.split()] if len(tokens) < max_len: tokens += [0] * (max_len - len(tokens)) else: tokens = tokens[:max_len] sequences.append(tokens) return torch.tensor(sequences, dtype=torch.long) # Training und Testdaten splitten train_texts, test_texts, train_labels, test_labels = train_test_split( humor_data['text'], humor_data['humor_rating'], test_size=0.2, random_state=42 ) # Tokenisierung und Padding max_len = 50 train_input_ids = tokenize_and_pad(train_texts, word_index, max_len=max_len) test_input_ids = tokenize_and_pad(test_texts, word_index, max_len=max_len) # Labels in Tensor konvertieren train_labels = torch.tensor(train_labels.values, dtype=torch.float) test_labels = torch.tensor(test_labels.values, dtype=torch.float) # 6. Dataset-Klasse für PyTorch class HumorDataset(Dataset): def __init__(self, input_ids, labels): self.input_ids = input_ids self.labels = labels def __len__(self): return len(self.input_ids) def __getitem__(self, idx): return self.input_ids[idx], self.labels[idx] # Dataset und DataLoader erstellen train_dataset = HumorDataset(train_input_ids, train_labels) test_dataset = HumorDataset(test_input_ids, test_labels) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False) # 7. CNN-Regression-Modell definieren class CNNRegressor(nn.Module): def __init__(self, vocab_size, embed_dim, embedding_matrix): super(CNNRegressor, self).__init__() self.embedding = nn.Embedding(vocab_size, embed_dim) self.embedding.weight.data.copy_(embedding_matrix.clone().detach()) self.embedding.weight.requires_grad = False self.conv1 = nn.Conv1d(embed_dim, 128, kernel_size=3) self.conv2 = nn.Conv1d(128, 64, kernel_size=3) self.dropout = nn.Dropout(0.5) self.fc = nn.Linear(64, 1) def forward(self, x): x = self.embedding(x).permute(0, 2, 1) x = torch.relu(self.conv1(x)) x = torch.relu(self.conv2(x)) x = self.dropout(x) x = torch.max(x, dim=2).values x = self.fc(x) x = torch.sigmoid(x) * 5 # Wertebereich [0, 5] return x # Initialisiere das Modell model = CNNRegressor(vocab_size, d_model, embedding_matrix).to(device) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # Early Stopping #early_stopping = EarlyStopping(patience=5) # 8. Training mit Validierung for epoch in range(20): # Maximal 20 Epochen model.train() train_loss = 0 for inputs, labels in tqdm(train_loader, desc=f"Epoch {epoch+1}"): inputs, labels = inputs.to(device), labels.to(device) optimizer.zero_grad() outputs = model(inputs).squeeze() loss = criterion(outputs, labels) loss.backward() optimizer.step() train_loss += loss.item() train_loss /= len(train_loader) # Validierungsverlust berechnen model.eval() val_loss = 0 with torch.no_grad(): for inputs, labels in test_loader: inputs, labels = inputs.to(device), labels.to(device) outputs = model(inputs).squeeze() loss = criterion(outputs, labels) val_loss += loss.item() val_loss /= len(test_loader) print(f"Epoch {epoch+1}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}") # Early Stopping '''early_stopping(val_loss, model) if early_stopping.early_stop: print("Early stopping triggered") break''' # 9. Modell evaluieren def evaluate_model(model, data_loader): model.eval() predictions = [] actuals = [] with torch.no_grad(): for inputs, labels in data_loader: inputs, labels = inputs.to(device), labels.to(device) outputs = model(inputs).squeeze() predictions.extend(outputs.cpu().numpy()) actuals.extend(labels.cpu().numpy()) return predictions, actuals predictions, actuals = evaluate_model(model, test_loader) # Metriken berechnen mse = mean_squared_error(actuals, predictions) mae = mean_absolute_error(actuals, predictions) r2 = r2_score(actuals, predictions) print(f"MSE: {mse:.4f}, MAE: {mae:.4f}, R²: {r2:.4f}") # 10. Visualisierung (Korrekte und falsche Vorhersagen farblich darstellen) tolerance = 0.5 # Toleranz für korrekte Vorhersagen predictions = np.array(predictions) actuals = np.array(actuals) # Klassifikation: Grün (korrekt), Rot (falsch) correct = np.abs(predictions - actuals) <= tolerance colors = np.where(correct, 'green', 'red') # Scatter-Plot plt.figure(figsize=(8, 6)) plt.scatter(actuals, predictions, c=colors, alpha=0.6, edgecolor='k', s=50) plt.plot([0, 5], [0, 5], color='red', linestyle='--') # Perfekte Vorhersage-Linie # Legende green_patch = mpatches.Patch(color='green', label='Correct Predictions') red_patch = mpatches.Patch(color='red', label='Incorrect Predictions') plt.legend(handles=[green_patch, red_patch]) # Achsen und Titel plt.xlabel("True Humor Ratings") plt.ylabel("Predicted Humor Ratings") plt.title("True vs Predicted Humor Ratings (Correct vs Incorrect)") plt.show()