Felipe97/llama-cpp-compiled

1from __future__ import annotations2 3import sys4 5from typing import Iterable, TYPE_CHECKING6 7import torch8 9if TYPE_CHECKING:10    from torch import Tensor11 12from .base import ModelBase, TextModel, gguf, logger13 14 15@ModelBase.register("GrokForCausalLM", "Grok1ForCausalLM")16@ModelBase.example("keyfan/grok-1-hf")17class GrokModel(TextModel):18    model_arch = gguf.MODEL_ARCH.GROK19 20    def set_vocab(self):21        if (self.dir_model / 'tokenizer.model').is_file():22            self._set_vocab_sentencepiece()23            return24 25        if not (self.dir_model / 'tokenizer.json').is_file() or not (self.dir_model / 'chat_template.jinja').is_file():26            logger.error('Error: Missing vocab and chat template, download files from https://huggingface.co/alvarobartt/grok-2-tokenizer')27            sys.exit(1)28 29        self._set_vocab_gpt2()30 31    def __init__(self, *args, **kwargs):32        super().__init__(*args, **kwargs)33 34    def set_gguf_parameters(self):35        super().set_gguf_parameters()36 37        self.gguf_writer.add_attn_logit_softcapping(self.hparams.get("attn_logit_softcapping", 30.0))38        self.gguf_writer.add_router_logit_softcapping(self.hparams.get("router_logit_softcapping", 30.0))39        if (final_logit_softcap := self.hparams.get("final_logit_softcapping")):40            self.gguf_writer.add_final_logit_softcapping(final_logit_softcap)41 42        if (rope_dim := self.hparams.get("head_dim")) is None:43            rope_dim = self.hparams["hidden_size"] // self.hparams["num_attention_heads"]44 45        if (moe_intermediate_size := self.hparams.get("moe_intermediate_size")) is not None:46            self.gguf_writer.add_expert_feed_forward_length(moe_intermediate_size)47 48        # Treat "original" as "yarn", seems to have been a mistake49        if self.hparams.get("rope_type") in ("yarn", "original"):50            self.gguf_writer.add_rope_scaling_type(gguf.RopeScalingType.YARN)51            self.gguf_writer.add_rope_scaling_factor(self.hparams["scaling_factor"])52            self.gguf_writer.add_rope_scaling_orig_ctx_len(self.hparams["original_max_position_embeddings"])53            self.gguf_writer.add_rope_scaling_yarn_ext_factor(self.hparams["extrapolation_factor"])54            self.gguf_writer.add_rope_scaling_yarn_attn_factor(self.hparams["attn_factor"])55            self.gguf_writer.add_rope_scaling_yarn_beta_fast(self.hparams["beta_fast"])56            self.gguf_writer.add_rope_scaling_yarn_beta_slow(self.hparams["beta_slow"])57 58        if temp_len := self.hparams.get("attn_temperature_len"):59            self.gguf_writer.add_attn_temperature_length(temp_len)60 61        self.gguf_writer.add_attn_output_scale(self.hparams.get("attn_output_multiplier", rope_dim**-0.5))62        self.gguf_writer.add_embedding_scale(self.hparams["embedding_multiplier_scale"])63        self.gguf_writer.add_logit_scale(self.hparams["output_multiplier_scale"])64 65    _experts: list[dict[str, list[Tensor]]] | None = None66    _cur_expert = ""67 68    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:69        deferred: list[tuple[Tensor, str, int | None]] = []70        is_expert = ".moe." in name or ".block_sparse_moe.experts." in name71 72        if not is_expert:73            deferred.append((data_torch, name, bid))74 75        # process the experts separately76        if is_expert or self._cur_expert:77            n_experts = self.hparams["num_local_experts"]78 79            assert bid is not None80 81            if self._experts is None:82                self._experts = [{} for _ in range(self.block_count)]83 84            # concatenate split tensors85            if name in self._experts[bid]:86                self._cur_expert = name87                self._experts[bid][name].append(data_torch)88                return89            elif is_expert:90                self._cur_expert = name91                self._experts[bid][name] = [data_torch]92                return93            else:94                self._cur_expert = ""95 96            for bid in range(self.block_count):97                if len(self._experts[bid]) >= n_experts * 3:98                    # merge the experts into a single 3d tensor99                    for wid in [("linear", "w1", 0), ("linear_1", "w2", 1), ("linear_v", "w3", 0)]:100                        datas: list[Tensor] = []101 102                        for xid in range(n_experts):103                            ename = f"transformer.decoder_layer.{bid}.moe.{xid}.{wid[0]}.weight"104                            if ename not in self._experts[bid]:105                                ename = f"model.layers.{bid}.block_sparse_moe.experts.{xid}.{wid[1]}.weight"106                            tensor_list = self._experts[bid][ename]107                            datas.append(torch.cat(tensor_list, dim=wid[2]) if len(tensor_list) > 1 else tensor_list[0])108                            del self._experts[bid][ename]109 110                        data_torch = torch.stack(datas, dim=0)111 112                        merged_name = f"transformer.decoder_layer.{bid}.moe.{wid[0]}.weight"113 114                        yield from super().modify_tensors(data_torch, merged_name, bid)115 116        for t in deferred:117            yield from super().modify_tensors(*t)118