Scaffold: Include custom libs

app.py

@@ -1,7 +1,17 @@
 import gradio as gr
 import html
 
-def predict(review: str, display_mode: str):
+# Projektspezifische Module
+from lib.bert_regressor import BertMultiHeadRegressor
+from lib.bert_regressor_utils import (
+    load_model_and_tokenizer,
+    predict_flavours,
+    #predict_is_review,
+    TARGET_COLUMNS,
+    ICONS
+)
+
+def predict(review: str, mode: str):
     review = (review or "").strip()
     if not review:
         # immer zwei Outputs zurückgeben

lib/bert_regressor.py

@@ -0,0 +1,113 @@
+import torch
+import torch.nn as nn
+from transformers import AutoModel
+
+###################################################################################
+
+# Erweiterte Regressorklasse: Ein gemeinsamer Encoder, aber mehrere unabhängige Köpfe
+class BertMultiHeadRegressor(nn.Module):
+    """
+    Mehrkopf-Regression auf einem beliebigen HF-Encoder (BERT/RoBERTa/DeBERTa/ModernBERT).
+    - Gemeinsamer Encoder
+    - n unabhängige Regressionsköpfe (je 1 Wert)
+    - Robustes Pooling (Pooler wenn vorhanden, sonst maskiertes Mean)
+    - Partielles Unfreezen ab `unfreeze_from`
+    """
+    def __init__(self, pretrained_model_name: str,
+                 n_heads: int = 8,
+                 unfreeze_from: int = 8,
+                 dropout: float = 0.1):
+        super().__init__()
+
+        # Beliebigen Encoder laden
+        self.encoder = AutoModel.from_pretrained(pretrained_model_name)
+        hidden_size = self.encoder.config.hidden_size
+
+        # Erst alles einfrieren …
+        for p in self.encoder.parameters():
+            p.requires_grad = False
+
+        # … dann Layer ab `unfreeze_from` freigeben (falls vorhanden)
+        # Die meisten Encoder haben `.encoder.layer`
+        encoder_block = getattr(self.encoder, "encoder", None)
+        layers = getattr(encoder_block, "layer", None)
+        if layers is not None:
+            for layer in layers[unfreeze_from:]:
+                for p in layer.parameters():
+                    p.requires_grad = True
+        else:
+            # Fallback: wenn kein klassisches Lagen-Array existiert, nichts tun
+            pass
+
+        self.dropout = nn.Dropout(dropout)
+        self.heads = nn.ModuleList([nn.Linear(hidden_size, 1) for _ in range(n_heads)])
+
+    def _pool(self, outputs, attention_mask):
+        """
+        Robustes Pooling:
+        - Wenn pooler_output vorhanden: nutzen (BERT/RoBERTa)
+        - Sonst: maskiertes Mean-Pooling über last_hidden_state (z. B. DeBERTaV3)
+        """
+        pooler = getattr(outputs, "pooler_output", None)
+        if pooler is not None:
+            return pooler  # [B, H]
+
+        last_hidden = outputs.last_hidden_state  # [B, T, H]
+        mask = attention_mask.unsqueeze(-1).float()  # [B, T, 1]
+        summed = (last_hidden * mask).sum(dim=1)     # [B, H]
+        denom = mask.sum(dim=1).clamp(min=1e-6)      # [B, 1]
+        return summed / denom
+
+    def forward(self, input_ids, attention_mask, token_type_ids=None):
+        outputs = self.encoder(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            token_type_ids=token_type_ids if token_type_ids is not None else None,
+            return_dict=True
+        )
+        pooled = self._pool(outputs, attention_mask)    # [B, H]
+        pooled = self.dropout(pooled)
+        preds = [head(pooled) for head in self.heads]   # n × [B, 1]
+        return torch.cat(preds, dim=1)                  # [B, n_heads]
+
+###################################################################################
+
+class BertBinaryClassifier(nn.Module):
+    def __init__(self, pretrained_model_name='bert-base-uncased', unfreeze_from=8, dropout=0.3):
+        super(BertBinaryClassifier, self).__init__()
+
+        # BERT-Encoder laden
+        self.bert = BertModel.from_pretrained(pretrained_model_name)
+
+        # Alle Layer zunächst einfrieren
+        for param in self.bert.parameters():
+            param.requires_grad = False
+
+        # Höhere Layer freigeben → feineres Fine-Tuning ab `unfreeze_from`
+        for layer in self.bert.encoder.layer[unfreeze_from:]:
+            for param in layer.parameters():
+                param.requires_grad = True
+
+        # Dropout-Schicht zur Regularisierung
+        self.dropout = nn.Dropout(dropout)
+        
+        # Klassifikationskopf: Wandelt das 768-dimensionale BERT-Embedding
+        # in einen einzelnen logit-Wert um (für binäre Klassifikation).
+        self.classifier = nn.Linear(self.bert.config.hidden_size, 1)
+
+    def forward(self, input_ids, attention_mask):
+        # Eingabe durch BERT verarbeiten → [batch_size, 768]
+        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
+
+        # CLS-Token-Repräsentation extrahieren
+        pooled_output = outputs.pooler_output
+
+        # Dropout anwenden zur Regularisierung
+        dropped = self.dropout(pooled_output)
+
+        # Logits durch linearen Klassifikator erzeugen
+        logits = self.classifier(dropped)
+
+        # Rückgabe der rohen Logits
+        return logits 
+        

lib/bert_regressor_utils.py

@@ -0,0 +1,231 @@
+import torch
+from transformers import AutoTokenizer
+from torch.utils.data import Dataset
+import numpy as np
+
+from .bert_regressor import BertMultiHeadRegressor, BertBinaryClassifier
+
+###################################################################################
+
+# Konstante Liste der acht Aromen-Kategorien für Whisky-Tasting-Notes.
+# Diese wird von Modellen und Evaluierungsfunktionen verwendet.
+TARGET_COLUMNS = [
+    "grainy",
+    "grassy",
+    "fragrant",
+    "fruity",
+    "peated",
+    "woody",
+    "winey",
+    "off-notes"
+]
+
+###################################################################################
+
+COLORS = {
+    "grainy": "#FFF3B0",
+    "grassy": "#C4F0C5",
+    "fragrant": "#F3C4FB",
+    "fruity": "#FFD6B0",
+    "peated": "#CFCFCF",
+    "woody": "#EAD6C7",
+    "winey": "#F7B7A3",
+    "off-notes": "#D6E4F0",
+    "quantifiers": "#ff8083"
+}
+
+ICONS = {
+    "grainy": "🌾",
+    "grassy": "🌿",
+    "fragrant": "🌸",
+    "fruity": "🍋",
+    "peated": "🔥",
+    "woody": "🌲",
+    "winey": "🍷",
+    "off-notes": "☠️"
+}
+
+###################################################################################
+
+class WhiskyDataset(Dataset):
+    def __init__(self, texts, targets, tokenizer, max_len):
+        self.texts = texts
+        self.targets = targets
+        self.tokenizer = tokenizer
+        self.max_len = max_len
+
+    def __len__(self):
+        return len(self.texts)
+
+    def __getitem__(self, item):
+        text = str(self.texts[item])
+        target = self.targets[item]
+        
+        # Einheitliche Tokenisierung über Hilfsfunktion
+        encoding = tokenize_input(text, self.tokenizer)
+        
+        return {
+            'input_ids': encoding['input_ids'].squeeze(),
+            'attention_mask': encoding['attention_mask'].squeeze(),
+            'targets': torch.tensor(target, dtype=torch.float)
+        }
+
+###################################################################################
+
+def get_device(prefer_mps=True, verbose=True):
+    """
+    Gibt das beste verfügbare Torch-Device zurück (MPS, CUDA oder CPU).
+
+    Args:
+        prefer_mps (bool): Ob bei Apple-Geräten 'mps' (Metal Performance Shaders) bevorzugt werden soll.
+        verbose (bool): Ob das erkannte Device ausgegeben werden soll.
+
+    Returns:
+        torch.device: Das beste verfügbare Gerät für das Training.
+    """
+    if prefer_mps and torch.backends.mps.is_available():
+        device = torch.device("mps")
+        name = "Apple GPU (MPS)"
+    elif torch.cuda.is_available():
+        device = torch.device("cuda")
+        name = torch.cuda.get_device_name(device)
+    else:
+        device = torch.device("cpu")
+        name = "CPU"
+
+    if verbose:
+        print(f"✅ Verwendetes Gerät: {name} ({device})")
+
+    return device
+    
+###################################################################################
+
+def tokenize_input(texts, tokenizer, max_len=256):
+    
+    """
+    Einheitliche Tokenisierung für Training und Inferenz.
+
+    Args:
+        texts (str or List[str]): Eingabetext(e).
+        tokenizer (PreTrainedTokenizer): z. B. BertTokenizer.
+
+    Returns:
+        dict: Dictionary mit PyTorch-Tensoren (input_ids, attention_mask).
+    """
+    return tokenizer(
+        texts,
+        truncation=True,
+        padding='max_length',
+        max_length=max_len,
+        return_tensors='pt'
+    )
+
+###################################################################################
+
+def load_model_and_tokenizer(model_name, model_path, model_type="multihead"):
+    """
+    Universelle Ladefunktion für BertMultiHeadRegressor oder BertBinaryClassifier.
+
+    Args:
+        model_name (str): Name des vortrainierten BERT-Modells (z. B. 'bert-base-uncased').
+        model_path (str): Pfad zur gespeicherten Modellzustandsdatei (.pt).
+        model_type (str): 'multihead' oder 'binary'. Default: 'multihead'.
+
+    Returns:
+        model (nn.Module): Geladenes Modell im Eval-Modus.
+        tokenizer (BertTokenizer): Passender Tokenizer.
+        device (torch.device): Verwendetes Rechengerät (CPU oder GPU).
+    """
+    # Gerät automatisch ermitteln (GPU/CPU)
+    device = get_device()
+
+    # Modellzustand und Konfiguration laden
+    checkpoint = torch.load(model_path, map_location=device)
+    config = checkpoint["model_config"]
+
+    # Modell je nach Typ initialisieren
+    if model_type == "multihead":
+        model = BertMultiHeadRegressor(
+            pretrained_model_name=config["pretrained_model_name"],
+            n_heads=config["n_heads"],
+            unfreeze_from=config["unfreeze_from"],
+            dropout=config["dropout"]
+        )
+    elif model_type == "binary":
+        model = BertBinaryClassifier(
+            pretrained_model_name=config["pretrained_model_name"],
+            unfreeze_from=config["unfreeze_from"],
+            dropout=config["dropout"]
+        )
+    else:
+        raise ValueError(f"Unbekannter model_type: {model_type}")
+
+    # Gewichtungen laden und Modell auf Gerät verschieben
+    model.to(device)
+    model.load_state_dict(checkpoint["model_state_dict"])
+    model.eval()  # Wechselt in den Inferenzmodus
+
+    # Lädt den passenden Tokenizer
+    tokenizer = AutoTokenizer.from_pretrained(model_name)
+
+    return model, tokenizer, device
+    
+###################################################################################
+
+def predict_flavours(review_text, model, tokenizer, device, max_len=256):
+    # Modell in den Evaluierungsmodus setzen (kein Dropout etc.)
+    model.eval()
+
+    # Eingabetext tokenisieren und als Tensoren zurückgeben
+    encoding = tokenize_input(
+        review_text, 
+        tokenizer
+    )
+
+    # Tokens auf das richtige Device verschieben
+    input_ids = encoding['input_ids'].to(device)
+    attention_mask = encoding['attention_mask'].to(device)
+
+    # Inferenz ohne Gradientenberechnung (Effizienz)
+    with torch.no_grad():
+        outputs = model(input_ids=input_ids, attention_mask=attention_mask)  # shape: [1, 8]
+        prediction = outputs.cpu().numpy().flatten()  # [8] – flach machen
+        prediction = np.clip(prediction, 0.0, 4.0)
+
+    # In ein Dictionary umwandeln (z. B. {"fruity": 2.1, "peated": 3.8, ...})
+    result = {
+        flavour: round(float(score), 2)
+        for flavour, score in zip(TARGET_COLUMNS, prediction)
+    }
+
+    return result
+
+###################################################################################
+
+def predict_is_review(review_text, model, tokenizer, device, max_len=256, threshold=0.5):
+    # Modell in den Evaluierungsmodus setzen (kein Dropout etc.)
+    model.eval()
+
+    # Eingabetext tokenisieren und als Tensoren zurückgeben
+    encoding = tokenize_input(
+        review_text, 
+        tokenizer
+    )
+
+    # Tokens auf das richtige Device verschieben
+    input_ids = encoding['input_ids'].to(device)
+    attention_mask = encoding['attention_mask'].to(device)
+
+    with torch.no_grad():
+        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
+        print(outputs.cpu().numpy())  # <--- Zeigt die rohen Logits
+        probs = torch.sigmoid(outputs)  # [1, 1]
+        prob = float(probs.squeeze().cpu().numpy())  # Skalar
+
+    return {
+        "is_review": prob >= threshold,
+        "probability": round(prob, 4)
+    }
+    
+###################################################################################
+