Brunobkr/llama.cpp_AlgMor24_github

1from __future__ import annotations2 3import json4import math5 6from typing import Callable, Iterable, TYPE_CHECKING7 8import torch9 10if TYPE_CHECKING:11    from torch import Tensor12 13from .base import MmprojModel, ModelBase, SentencePieceTokenTypes, TextModel, gguf, logger14 15 16@ModelBase.register("PhiForCausalLM")17class Phi2Model(TextModel):18    model_arch = gguf.MODEL_ARCH.PHI219 20    def set_gguf_parameters(self):21        rot_pct = self.rope_parameters["partial_rotary_factor"]22        n_embd = self.find_hparam(["hidden_size", "n_embd"])23        n_head = self.find_hparam(["num_attention_heads", "n_head"])24 25        self.gguf_writer.add_context_length(self.find_hparam(["n_positions", "max_position_embeddings"]))26 27        self.gguf_writer.add_embedding_length(n_embd)28        self.gguf_writer.add_feed_forward_length(4 * n_embd)29        self.gguf_writer.add_block_count(self.block_count)30        self.gguf_writer.add_head_count(n_head)31        self.gguf_writer.add_head_count_kv(n_head)32        self.gguf_writer.add_layer_norm_eps(self.find_hparam(["layer_norm_epsilon", "layer_norm_eps"]))33        self.gguf_writer.add_rope_dimension_count(int(rot_pct * n_embd) // n_head)34        self.gguf_writer.add_file_type(self.ftype)35        self.gguf_writer.add_add_bos_token(False)36 37 38@ModelBase.register("Phi3ForCausalLM", "Phi4ForCausalLMV")39class Phi3MiniModel(TextModel):40    model_arch = gguf.MODEL_ARCH.PHI341 42    def set_vocab(self):43        # Phi-4 model uses GPT2Tokenizer44        tokenizer_config_file = self.dir_model / 'tokenizer_config.json'45        if tokenizer_config_file.is_file():46            with open(tokenizer_config_file, "r", encoding="utf-8") as f:47                tokenizer_config_json = json.load(f)48                tokenizer_class = tokenizer_config_json['tokenizer_class']49                if tokenizer_class == 'GPT2Tokenizer':50                    return self._set_vocab_gpt2()51 52        from sentencepiece import SentencePieceProcessor53 54        tokenizer_path = self.dir_model / 'tokenizer.model'55 56        if not tokenizer_path.is_file():57            raise ValueError(f'Error: Missing {tokenizer_path}')58 59        tokenizer = SentencePieceProcessor()60        tokenizer.LoadFromFile(str(tokenizer_path))61 62        vocab_size = self.hparams.get('vocab_size', tokenizer.vocab_size())63 64        tokens: list[bytes] = [f"[PAD{i}]".encode("utf-8") for i in range(vocab_size)]65        scores: list[float] = [-10000.0] * vocab_size66        toktypes: list[int] = [SentencePieceTokenTypes.UNUSED] * vocab_size67 68        for token_id in range(tokenizer.vocab_size()):69 70            piece = tokenizer.IdToPiece(token_id)71            text = piece.encode("utf-8")72            score = tokenizer.GetScore(token_id)73 74            toktype = SentencePieceTokenTypes.NORMAL75            if tokenizer.IsUnknown(token_id):76                toktype = SentencePieceTokenTypes.UNKNOWN77            elif tokenizer.IsControl(token_id):78                toktype = SentencePieceTokenTypes.CONTROL79            elif tokenizer.IsUnused(token_id):80                toktype = SentencePieceTokenTypes.UNUSED81            elif tokenizer.IsByte(token_id):82                toktype = SentencePieceTokenTypes.BYTE83 84            tokens[token_id] = text85            scores[token_id] = score86            toktypes[token_id] = toktype87 88        added_tokens_file = self.dir_model / 'added_tokens.json'89        if added_tokens_file.is_file():90            with open(added_tokens_file, "r", encoding="utf-8") as f:91                added_tokens_json = json.load(f)92 93                for key in added_tokens_json:94                    token_id = added_tokens_json[key]95                    if token_id >= vocab_size:96                        logger.debug(f'ignore token {token_id}: id is out of range, max={vocab_size - 1}')97                        continue98 99                    tokens[token_id] = key.encode("utf-8")100                    scores[token_id] = -1000.0101                    toktypes[token_id] = SentencePieceTokenTypes.USER_DEFINED102 103        tokenizer_config_file = self.dir_model / 'tokenizer_config.json'104        if tokenizer_config_file.is_file():105            with open(tokenizer_config_file, "r", encoding="utf-8") as f:106                tokenizer_config_json = json.load(f)107                added_tokens_decoder = tokenizer_config_json.get("added_tokens_decoder", {})108                for token_id, foken_data in added_tokens_decoder.items():109                    token_id = int(token_id)110                    token = foken_data["content"].encode("utf-8")111                    if toktypes[token_id] != SentencePieceTokenTypes.UNUSED:112                        if tokens[token_id] != token:113                            logger.warning(f'replacing token {token_id}: {tokens[token_id].decode("utf-8")!r} -> {token.decode("utf-8")!r}')114                    tokens[token_id] = token115                    scores[token_id] = -1000.0116                    toktypes[token_id] = SentencePieceTokenTypes.USER_DEFINED117                    if foken_data.get("special"):118                        toktypes[token_id] = SentencePieceTokenTypes.CONTROL119 120        tokenizer_file = self.dir_model / 'tokenizer.json'121        if tokenizer_file.is_file():122            with open(tokenizer_file, "r", encoding="utf-8") as f:123                tokenizer_json = json.load(f)124                added_tokens = tokenizer_json.get("added_tokens", [])125                for foken_data in added_tokens:126                    token_id = int(foken_data["id"])127                    token = foken_data["content"].encode("utf-8")128                    if toktypes[token_id] != SentencePieceTokenTypes.UNUSED:129                        if tokens[token_id] != token:130                            logger.warning(f'replacing token {token_id}: {tokens[token_id].decode("utf-8")!r} -> {token.decode("utf-8")!r}')131                    tokens[token_id] = token132                    scores[token_id] = -1000.0133                    toktypes[token_id] = SentencePieceTokenTypes.USER_DEFINED134                    if foken_data.get("special"):135                        toktypes[token_id] = SentencePieceTokenTypes.CONTROL136 137        self.gguf_writer.add_tokenizer_model("llama")138        self.gguf_writer.add_tokenizer_pre("default")139        self.gguf_writer.add_token_list(tokens)140        self.gguf_writer.add_token_scores(scores)141        self.gguf_writer.add_token_types(toktypes)142 143        special_vocab = gguf.SpecialVocab(self.dir_model, n_vocab=len(tokens))144        special_vocab.add_to_gguf(self.gguf_writer)145 146    def set_gguf_parameters(self):147        n_embd = self.find_hparam(["hidden_size", "n_embd"])148        n_head = self.find_hparam(["num_attention_heads", "n_head"])149        n_head_kv = self.find_hparam(["num_key_value_heads", "n_head_kv"])150        rms_eps = self.find_hparam(["rms_norm_eps"])151        max_pos_embds = self.find_hparam(["n_positions", "max_position_embeddings"])152        orig_max_pos_embds = self.rope_parameters["original_max_position_embeddings"]153        rot_pct = self.rope_parameters.get("partial_rotary_factor", 1.0)154        rope_dims = int(rot_pct * n_embd) // n_head155 156        self.gguf_writer.add_context_length(max_pos_embds)157        self.gguf_writer.add_rope_scaling_orig_ctx_len(orig_max_pos_embds)158        self.gguf_writer.add_embedding_length(n_embd)159        self.gguf_writer.add_feed_forward_length(self.find_hparam(["intermediate_size"]))160        self.gguf_writer.add_block_count(self.block_count)161        self.gguf_writer.add_head_count(n_head)162        self.gguf_writer.add_head_count_kv(n_head_kv)163        self.gguf_writer.add_layer_norm_rms_eps(rms_eps)164        self.gguf_writer.add_rope_dimension_count(rope_dims)165        self.gguf_writer.add_rope_freq_base(self.rope_parameters.get("full_attention", self.rope_parameters)["rope_theta"])166        self.gguf_writer.add_file_type(self.ftype)167        sliding_window = self.hparams.get("sliding_window")168        # use zero value of sliding_window to distinguish Phi-4 from other PHI3 models169        if sliding_window is None:170            sliding_window = 0171        self.gguf_writer.add_sliding_window(sliding_window)172 173    def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:174        n_embd = self.find_hparam(["hidden_size", "n_embd"])175        n_head = self.find_hparam(["num_attention_heads", "n_head"])176        max_pos_embds = self.find_hparam(["n_positions", "max_position_embeddings"])177        orig_max_pos_embds = self.rope_parameters["original_max_position_embeddings"]178        rot_pct = self.rope_parameters.get("partial_rotary_factor", 1.0)179        rope_dims = int(rot_pct * n_embd) // n_head180 181        # write rope scaling for long context (128k) model182        long_factors = self.rope_parameters.get('long_factor')183        short_factors = self.rope_parameters.get('short_factor')184        if not long_factors:185            return186 187        scale = max_pos_embds / orig_max_pos_embds188 189        rope_scaling_type = self.rope_parameters.get('rope_type', '').lower()190        if len(rope_scaling_type) == 0:191            raise KeyError('Missing the required key rope_scaling.type')192 193        if rope_scaling_type == 'su' or rope_scaling_type == 'longrope':194            attn_factor = math.sqrt(1 + math.log(scale) / math.log(orig_max_pos_embds)) if scale > 1.0 else 1.0195        elif rope_scaling_type == 'yarn':196            attn_factor = 0.1 * math.log(scale) + 1.0 if scale > 1.0 else 1.0197        else:198            raise NotImplementedError(f'The rope scaling type {rope_scaling_type} is not supported yet')199 200        self.gguf_writer.add_rope_scaling_attn_factors(attn_factor)201 202        if long_factors is None or short_factors is None:203            raise KeyError('Missing the required key rope_scaling.long_factor or rope_scaling_short_factor')204 205        if len(long_factors) != len(short_factors) or len(long_factors) != rope_dims / 2:206            raise ValueError(f'The length of rope long and short factors must be {rope_dims / 2}. long_factors = {len(long_factors)}, short_factors = {len(short_factors)}.')207 208        yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_LONG), torch.tensor(long_factors, dtype=torch.float32))209        yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_SHORT), torch.tensor(short_factors, dtype=torch.float32))210 211 212@ModelBase.register("Phi4ForCausalLMV")213class Phi4VisionMmprojModel(MmprojModel):214    def __init__(self, *args, **kwargs):215        super().__init__(*args, **kwargs)216        assert self.hparams_vision is not None217 218        self.vision_total_layers = int(self.find_vparam(self.n_block_keys))219        if self.vision_total_layers < 2:220            raise ValueError(221                f"Phi-4 vision mmproj conversion requires at least 2 vision layers, got {self.vision_total_layers}"222            )223 224        # Phi-4 uses SigLIP2 hidden_states[-2], so export one fewer encoder block and225        # drop post-layernorm/head weights. This makes the GGUF runtime output match226        # the feature map consumed by the patched siglip.cpp Phi-4 projector path.227        self.vision_export_layers = self.vision_total_layers - 1228        self.vision_last_layer_idx = self.vision_total_layers - 1229 230        for key in self.n_block_keys:231            if key in self.hparams_vision:232                self.hparams_vision[key] = self.vision_export_layers233                break234 235        self.block_count = self.vision_export_layers236        self.tensor_map = gguf.get_tensor_name_map(gguf.MODEL_ARCH.MMPROJ, self.block_count)237 238        patch_size = self.preprocessor_config.get("patch_size")239        if patch_size is None:240            raise KeyError("Phi-4 vision mmproj conversion requires patch_size in preprocessor_config.json")241 242        self.hparams_vision["patch_size"] = patch_size243 244        pos_emb_name = next(245            (246                name for name in self.model_tensors247                if name.endswith("vision_model.embeddings.position_embedding.weight")248            ),249            None,250        )251        if pos_emb_name is None:252            raise KeyError("Phi-4 vision mmproj conversion could not find position_embedding.weight")253 254        pos_emb_shape = self.model_tensors[pos_emb_name]().shape255        base_grid_tokens = int(pos_emb_shape[0])256        grid_side = math.isqrt(base_grid_tokens)257        if grid_side * grid_side != base_grid_tokens:258            raise ValueError(f"Unexpected Phi-4 position embedding shape: {tuple(pos_emb_shape)}")259 260        self.hparams_vision["image_size"] = grid_side * patch_size261 262        min_num_patches = self.preprocessor_config.get("min_num_patches", self.global_config.get("min_num_patches"))263        max_num_patches = self.preprocessor_config.get("max_num_patches", self.global_config.get("max_num_patches"))264        if min_num_patches is None or max_num_patches is None:265            raise KeyError("Phi-4 vision mmproj conversion requires min_num_patches and max_num_patches")266 267        self.min_pixels = int(min_num_patches) * patch_size * patch_size268        self.max_pixels = int(max_num_patches) * patch_size * patch_size269 270    def set_gguf_parameters(self):271        super().set_gguf_parameters()272        assert self.hparams_vision is not None273 274        self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.PHI4)275        self.gguf_writer.add_vision_min_pixels(self.min_pixels)276        self.gguf_writer.add_vision_max_pixels(self.max_pixels)277        self.gguf_writer.add_vision_use_gelu(True)278        self.gguf_writer.add_vision_attention_layernorm_eps(self.hparams_vision.get("layer_norm_eps", 1e-6))279 280    @classmethod281    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:282        name, gen = item283 284        name = name.replace("model.vision_tower.vision_tower.", "vision_tower.")285 286        if not name.startswith(("vision_tower.", "model.mm_projector.", "mm_projector.")):287            return None288 289        if ".vision_model.head." in name:290            return None291 292        if ".vision_model.post_layernorm." in name:293            return None294 295        return super().filter_tensors((name, gen))296 297    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:298        if name.startswith("vision_tower."):299            if bid is not None and bid == self.vision_last_layer_idx:300                return301 302            if name.endswith("vision_model.embeddings.patch_embedding.weight"):303                assert self.hparams_vision is not None304                if data_torch.ndim != 2:305                    raise ValueError(f"Unexpected Phi-4 patch embedding shape: {tuple(data_torch.shape)}")306 307                patch_area = self.hparams_vision["patch_size"] ** 2308                in_features = data_torch.shape[1]309                if in_features % patch_area != 0:310                    raise ValueError(311                        f"Phi-4 patch embedding input dim {in_features} is not divisible by patch area {patch_area}"312                    )313 314                num_channels = in_features // patch_area315                patch_size = self.hparams_vision["patch_size"]316                data_torch = data_torch.view(data_torch.shape[0], patch_size, patch_size, num_channels)317                data_torch = data_torch.permute(0, 3, 1, 2)318 319            yield from super().modify_tensors(data_torch, name, bid)320            return321 322        if name.startswith(("model.mm_projector.", "mm_projector.")):323            local_name = name324            local_name = local_name.replace("model.mm_projector.", "")325            local_name = local_name.replace("mm_projector.", "")326 327            if not (local_name.startswith("0.") or local_name.startswith("2.")):328                return329 330            suffix = ".bias" if local_name.endswith(".bias") else ".weight"331            mm_idx = int(local_name.split(".", maxsplit=1)[0])332            yield (self.format_tensor_name(gguf.MODEL_TENSOR.V_MMPROJ, mm_idx, suffix=suffix), data_torch)333            return334 335        return336 337 338@ModelBase.register("PhiMoEForCausalLM")339class PhiMoeModel(Phi3MiniModel):340    model_arch = gguf.MODEL_ARCH.PHIMOE341 342    _experts: list[dict[str, Tensor]] | None = None343 344    def set_gguf_parameters(self):345        super().set_gguf_parameters()346        self.gguf_writer.add_expert_used_count(self.find_hparam(["num_experts_per_tok", "num_experts_per_token"]))347        self.gguf_writer.add_expert_count(self.find_hparam(["num_local_experts", "num_experts"]))348 349    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:350        # process the experts separately351        if name.find("block_sparse_moe.experts") != -1:352            n_experts = self.find_hparam(["num_local_experts", "num_experts"])353            assert bid is not None354 355            if self._experts is None:356                self._experts = [{} for _ in range(self.block_count)]357 358            self._experts[bid][name] = data_torch359 360            if len(self._experts[bid]) >= n_experts * 3:361                # merge the experts into a single 3d tensor362                for w_name in ["w1", "w2", "w3"]:363                    datas: list[Tensor] = []364 365                    for xid in range(n_experts):366                        ename = f"model.layers.{bid}.block_sparse_moe.experts.{xid}.{w_name}.weight"367                        datas.append(self._experts[bid][ename])368                        del self._experts[bid][ename]369 370                    data_torch = torch.stack(datas, dim=0)371 372                    merged_name = f"model.layers.{bid}.block_sparse_moe.experts.{w_name}.weight"373 374                    yield from super().modify_tensors(data_torch, merged_name, bid)375                return376            else:377                return378 379        yield from super().modify_tensors(data_torch, name, bid)380 381    def prepare_tensors(self):382        super().prepare_tensors()383 384        if self._experts is not None:385            # flatten `list[dict[str, Tensor]]` into `list[str]`386            experts = [k for d in self._experts for k in d.keys()]387            if len(experts) > 0:388                raise ValueError(f"Unprocessed experts: {experts}")389