modeling_aliceai_t5_moe.py

import torch
from transformers.activations import ACT2FN

from .configuration_aliceai_t5 import AliceAIT5ModuleConfig
from .configuration_aliceai_t5_moe import AliceAIT5MoEConfig
from .modeling_aliceai_t5 import (
    AliceAIT5Decoder,
    AliceAIT5DecoderLayer,
    AliceAIT5Encoder,
    AliceAIT5EncoderLayer,
    AliceAIT5EncoderModel,
    AliceAIT5ForConditionalGeneration,
    AliceAIT5Model,
    AliceAIT5PreTrainedModel,
)
from .moe_layers import AdditionalArgs, Arguments, dMoE


def build_moe_arguments(model_config: AliceAIT5MoEConfig, is_encoder: bool = True) -> dict[str, object]:
    module_config = model_config.encoder if is_encoder else model_config.decoder
    return {
        "args": Arguments(
            hidden_size=module_config.hidden_size,
            ffn_hidden_size=model_config.expert_intermediate_size,
            activation_fn=ACT2FN[model_config.activation_type],
            moe_num_experts=model_config.n_experts,
            moe_top_k=model_config.experts_gating_top,
            moe_normalize_expert_weights=model_config.moe_normalize_expert_weights,
            routed_scaling_factor=model_config.routed_scaling_factor,
            bias=True,
        ),
        "additional_args": AdditionalArgs(
            moe_n_group=model_config.experts_n_group,
            moe_top_k_group=model_config.experts_top_k_group,
            group_routing=model_config.group_routing
            and model_config.experts_n_group is not None
            and model_config.experts_n_group > 1,
        ),
    }


class AliceAIT5MoEEncoderLayer(AliceAIT5EncoderLayer):
    def __init__(self, config: AliceAIT5MoEConfig, layer_idx: int):
        super().__init__(
            config.encoder,
            layer_idx=layer_idx,
            mlp=dMoE(**build_moe_arguments(model_config=config, is_encoder=True)),
        )


class AliceAIT5MoEDecoderLayer(AliceAIT5DecoderLayer):
    def __init__(self, config: AliceAIT5MoEConfig, layer_idx: int):
        super().__init__(
            config.decoder,
            layer_idx=layer_idx,
            mlp=dMoE(**build_moe_arguments(model_config=config, is_encoder=False)),
        )


class AliceAIT5MoEEncoder(AliceAIT5Encoder):
    _no_split_modules = [AliceAIT5MoEEncoderLayer.__name__]

    def __init__(self, config: AliceAIT5MoEConfig):
        self._moe_config = config
        super().__init__(config.encoder)

    def _build_layers(self, config: AliceAIT5ModuleConfig):
        moe_config = self._moe_config
        self.layers = torch.nn.ModuleList(
            [AliceAIT5MoEEncoderLayer(moe_config, layer_idx) for layer_idx in range(config.num_hidden_layers)]
        )


class AliceAIT5MoEDecoder(AliceAIT5Decoder):
    _no_split_modules = [AliceAIT5MoEDecoderLayer.__name__]

    def __init__(self, config: AliceAIT5MoEConfig):
        self._moe_config = config
        super().__init__(config.decoder)

    def _build_layers(self, config):
        moe_config = self._moe_config
        self.layers = torch.nn.ModuleList(
            [AliceAIT5MoEDecoderLayer(moe_config, layer_idx) for layer_idx in range(config.num_hidden_layers)]
        )


class AliceAIT5MoEModel(AliceAIT5Model):
    config_class = AliceAIT5MoEConfig
    _no_split_modules = [AliceAIT5MoEEncoderLayer.__name__, AliceAIT5MoEDecoderLayer.__name__]

    def __init__(self, config: AliceAIT5MoEConfig):
        AliceAIT5PreTrainedModel.__init__(self, config)

        if not config.is_encoder_decoder:
            raise ValueError("AliceAIT5MoEModel only supports encoder-decoder modeling.")

        self.encoder = AliceAIT5MoEEncoder(config)
        self.decoder = AliceAIT5MoEDecoder(config)
        self.post_init()


class AliceAIT5MoEEncoderModel(AliceAIT5EncoderModel):
    config_class = AliceAIT5MoEConfig
    _auto_class = "AutoModel"

    def __init__(self, config: AliceAIT5MoEConfig):
        super().__init__(config)
        self.config.auto_map = {
            "AutoConfig": "configuration_aliceai_t5_moe.AliceAIT5MoEConfig",
            "AutoModel": "modeling_aliceai_t5_moe.AliceAIT5MoEEncoderModel",
        }

    def _build_encoder(self, config):
        return AliceAIT5MoEEncoder(config)


class AliceAIT5MoEForConditionalGeneration(AliceAIT5ForConditionalGeneration):
    config_class = AliceAIT5MoEConfig
    _no_split_modules = [AliceAIT5MoEEncoderLayer.__name__, AliceAIT5MoEDecoderLayer.__name__]

    def _build_model(self, config):
        return AliceAIT5MoEModel(config)


__all__ = [
    "AliceAIT5MoEConfig",
    "AliceAIT5MoEEncoderLayer",
    "AliceAIT5MoEDecoderLayer",
    "AliceAIT5MoEEncoder",
    "AliceAIT5MoEDecoder",
    "AliceAIT5MoEModel",
    "AliceAIT5MoEEncoderModel",
    "AliceAIT5MoEForConditionalGeneration",
]