modeling_multihead.py

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import PreTrainedModel, PretrainedConfig, AutoModel, AutoConfig


class MultiHeadConfig(PretrainedConfig):
    model_type = "multihead_text_classifier"

    def __init__(self, encoder_name="cointegrated/LaBSE-en-ru",
                 num_labels_l2=4, num_labels_l3=14, dropout=0.1, hidden_size=768,
                 id2label_l2=None, label2id_l2=None,
                 id2label_l3=None, label2id_l3=None,
                 l3_to_l2=None, loss_weight_l2=0.5, loss_weight_l3=1.0, label_smoothing=0.0, **kwargs):
        super().__init__(**kwargs)
        self.encoder_name = encoder_name
        self.num_labels_l2 = num_labels_l2
        self.num_labels_l3 = num_labels_l3
        self.dropout = dropout
        self.hidden_size = hidden_size
        self.id2label_l2 = id2label_l2 or {}
        self.label2id_l2 = label2id_l2 or {}
        self.id2label_l3 = id2label_l3 or {}
        self.label2id_l3 = label2id_l3 or {}
        self.l3_to_l2 = l3_to_l2 or []
        self.loss_weight_l2 = loss_weight_l2
        self.loss_weight_l3 = loss_weight_l3
        self.label_smoothing = label_smoothing


class MultiHeadClassifier(PreTrainedModel):
    config_class = MultiHeadConfig

    def __init__(self, config):
        super().__init__(config)
        enc_config = AutoConfig.from_pretrained(config.encoder_name)
        self.encoder = AutoModel.from_config(enc_config)   # только архитектура (веса грузятся отдельно)
        h = enc_config.hidden_size
        self.dropout = nn.Dropout(config.dropout)
        self.head_l2 = nn.Linear(h, config.num_labels_l2)
        self.head_l3 = nn.Linear(h, config.num_labels_l3)
        # веса классов (не сохраняются в чекпойнт): по умолчанию единицы
        self.register_buffer("cw_l2", torch.ones(config.num_labels_l2), persistent=False)
        self.register_buffer("cw_l3", torch.ones(config.num_labels_l3), persistent=False)
        self.post_init()

    @classmethod
    def from_encoder(cls, config):
        """Инициализация для обучения: подгружаем претренированные веса энкодера."""
        model = cls(config)
        model.encoder = AutoModel.from_pretrained(config.encoder_name)
        return model

    def set_class_weights(self, w_l2, w_l3):
        self.register_buffer("cw_l2", torch.as_tensor(w_l2, dtype=torch.float), persistent=False)
        self.register_buffer("cw_l3", torch.as_tensor(w_l3, dtype=torch.float), persistent=False)

    def forward(self, input_ids=None, attention_mask=None, token_type_ids=None,
                labels_l2=None, labels_l3=None, **kwargs):
        out = self.encoder(input_ids=input_ids, attention_mask=attention_mask,
                           token_type_ids=token_type_ids)
        cls = out.last_hidden_state[:, 0]          # [CLS]
        cls = self.dropout(cls)
        logits_l2 = self.head_l2(cls)
        logits_l3 = self.head_l3(cls)

        loss = None
        if labels_l2 is not None and labels_l3 is not None:
            cw2 = self.cw_l2.to(logits_l2.device).to(logits_l2.dtype)
            cw3 = self.cw_l3.to(logits_l3.device).to(logits_l3.dtype)
            ls = getattr(self.config, "label_smoothing", 0.0)
            l2 = F.cross_entropy(logits_l2, labels_l2, weight=cw2, label_smoothing=ls)
            l3 = F.cross_entropy(logits_l3, labels_l3, weight=cw3, label_smoothing=ls)
            loss = self.config.loss_weight_l2 * l2 + self.config.loss_weight_l3 * l3

        return {"loss": loss, "logits_l2": logits_l2, "logits_l3": logits_l3}


MultiHeadConfig.register_for_auto_class()
MultiHeadClassifier.register_for_auto_class("AutoModel")