Brunobkr/llama.cpp_AlgMor24_github

1from __future__ import annotations2 3from typing import Iterable, TYPE_CHECKING4 5import torch6 7if TYPE_CHECKING:8    from torch import Tensor9 10from .base import ModelBase, TextModel, gguf, logger11 12 13@ModelBase.register("SmallThinkerForCausalLM")14class SmallThinkerModel(TextModel):15    model_arch = gguf.MODEL_ARCH.SMALLTHINKER16 17    def set_gguf_parameters(self):18        super().set_gguf_parameters()19        if (n_experts := self.hparams.get("moe_num_primary_experts")) is not None:20            self.gguf_writer.add_expert_count(n_experts)21        if (n_experts_used := self.hparams.get("moe_num_active_primary_experts")) is not None:22            self.gguf_writer.add_expert_used_count(n_experts_used)23        if (moe_intermediate_size := self.hparams.get("moe_ffn_hidden_size")) is not None:24            self.gguf_writer.add_expert_feed_forward_length(moe_intermediate_size)25            self.gguf_writer.add_feed_forward_length(moe_intermediate_size)26            logger.info(f"gguf: expert feed forward length = {moe_intermediate_size}")27        if (self.hparams.get('moe_primary_router_apply_softmax')):28            self.gguf_writer.add_expert_gating_func(gguf.ExpertGatingFuncType.SOFTMAX)29        else:30            self.gguf_writer.add_expert_gating_func(gguf.ExpertGatingFuncType.SIGMOID)31 32        sliding_window_layout = self.hparams.get("sliding_window_layout")33        if sliding_window_layout:34            for i in sliding_window_layout:35                if i != 0:36                    sliding_window = self.hparams.get("sliding_window_size")37                    if sliding_window:38                        self.gguf_writer.add_sliding_window(sliding_window)39                    break40 41    _experts: list[dict[str, Tensor]] | None = None42 43    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:44        # process the experts separately45        if name.find("experts") != -1:46            n_experts = self.hparams.get("moe_num_primary_experts") or self.find_hparam(["num_local_experts", "num_experts"])47            assert bid is not None48 49            if self._experts is None:50                self._experts = [{} for _ in range(self.block_count)]51 52            self._experts[bid][name] = data_torch53 54            if len(self._experts[bid]) >= n_experts * 3:55                # merge the experts into a single 3d tensor56                for w_name in ["down", "gate", "up"]:57                    datas: list[Tensor] = []58 59                    for xid in range(n_experts):60                        ename = f"model.layers.{bid}.block_sparse_moe.experts.{xid}.{w_name}.weight"61                        datas.append(self._experts[bid][ename])62                        del self._experts[bid][ename]63 64                    data_torch = torch.stack(datas, dim=0)65 66                    merged_name = f"model.layers.{bid}.block_sparse_moe.experts.{w_name}.weight"67 68                    yield from super().modify_tensors(data_torch, merged_name, bid)69                return70            else:71                return72 73        yield from super().modify_tensors(data_torch, name, bid)74 75    def prepare_tensors(self):76        super().prepare_tensors()77 78        if self._experts is not None:79            # flatten `list[dict[str, Tensor]]` into `list[str]`80            experts = [k for d in self._experts for k in d.keys()]81            if len(experts) > 0:82                raise ValueError(f"Unprocessed experts: {experts}")83