Felipe97/llama-cpp-compiled

1from __future__ import annotations2 3from typing import Callable, Iterable, TYPE_CHECKING4 5import torch6 7if TYPE_CHECKING:8    from torch import Tensor9 10from .base import MmprojModel, ModelBase, TextModel, gguf, logger11 12from .llama import LlamaModel13from .qwen import Qwen3_5TextModel14 15 16@ModelBase.register("MiniCPMForCausalLM")17@ModelBase.example("openbmb/MiniCPM-2B-sft-bf16")18class MiniCPMModel(TextModel):19    model_arch = gguf.MODEL_ARCH.MINICPM20 21    def set_gguf_parameters(self):22        super().set_gguf_parameters()23        embedding_scale = float(self.hparams["scale_emb"])24        self.gguf_writer.add_embedding_scale(embedding_scale)25        logger.info(f"gguf: (minicpm) embedding_scale = {embedding_scale}")26        residual_scale = self.hparams["scale_depth"] / self.hparams["num_hidden_layers"] ** 0.527        self.gguf_writer.add_residual_scale(residual_scale)28        logger.info(f"gguf: (minicpm) residual_scale = {residual_scale}")29        logit_scale = self.hparams["hidden_size"] / self.hparams["dim_model_base"]30        self.gguf_writer.add_logit_scale(logit_scale)31        logger.info(f"gguf: (minicpm) logit_scale = {logit_scale}")32 33    def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:34        rope_dims = self.hparams["hidden_size"] // self.hparams["num_attention_heads"]35 36        long_factors = self.rope_parameters.get('long_factor')37        short_factors = self.rope_parameters.get('short_factor')38        if long_factors or short_factors:39            if long_factors is None or short_factors is None:40                raise KeyError('Missing the required key rope_scaling.long_factor or rope_scaling_short_factor')41 42            if len(long_factors) != len(short_factors) or len(long_factors) != rope_dims / 2:43                raise ValueError(f'The length of rope long and short factors must be {rope_dims / 2}')44 45            yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_LONG), torch.tensor(long_factors, dtype=torch.float32))46            yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_SHORT), torch.tensor(short_factors, dtype=torch.float32))47 48    def set_vocab(self):49        self._set_vocab_sentencepiece()50 51    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:52        n_head = self.hparams["num_attention_heads"]53        n_kv_head = self.hparams.get("num_key_value_heads")54 55        # HF models permute some of the tensors, so we need to undo that56        if name.endswith(("q_proj.weight")):57            data_torch = LlamaModel.permute(data_torch, n_head, n_head)58        if name.endswith(("k_proj.weight")):59            data_torch = LlamaModel.permute(data_torch, n_head, n_kv_head)60 61        yield from super().modify_tensors(data_torch, name, bid)62 63 64@ModelBase.register("MiniCPM3ForCausalLM")65@ModelBase.example("openbmb/MiniCPM3-4B")66class MiniCPM3Model(TextModel):67    model_arch = gguf.MODEL_ARCH.MINICPM368 69    def set_gguf_parameters(self):70        hparams = self.hparams71 72        self.gguf_writer.add_file_type(self.ftype)73        self.gguf_writer.add_context_length(hparams["max_position_embeddings"])74        self.gguf_writer.add_embedding_length(hparams["hidden_size"])75        self.gguf_writer.add_block_count(self.block_count)76        self.gguf_writer.add_feed_forward_length(hparams["intermediate_size"])77        self.gguf_writer.add_head_count(hparams["num_attention_heads"])78        self.gguf_writer.add_head_count_kv(hparams["num_key_value_heads"])79        self.gguf_writer.add_layer_norm_rms_eps(hparams["rms_norm_eps"])80        self.gguf_writer.add_vocab_size(hparams["vocab_size"])81        if "q_lora_rank" in hparams and hparams["q_lora_rank"] is not None:82            self.gguf_writer.add_q_lora_rank(hparams["q_lora_rank"])83        self.gguf_writer.add_kv_lora_rank(hparams["kv_lora_rank"])84        self.gguf_writer.add_key_length(hparams["qk_nope_head_dim"] + hparams["qk_rope_head_dim"])85        self.gguf_writer.add_rope_dimension_count(hparams["qk_rope_head_dim"])86 87    def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:88        long_factors = self.rope_parameters.get('long_factor')89        short_factors = self.rope_parameters.get('short_factor')90        if long_factors or short_factors:91            rope_dims = self.hparams["qk_rope_head_dim"]92 93            if long_factors is None or short_factors is None:94                raise KeyError('Missing the required key rope_scaling.long_factor or rope_scaling_short_factor')95 96            if len(long_factors) != len(short_factors) or len(long_factors) != rope_dims / 2:97                raise ValueError(f'The length of rope long and short factors must be {rope_dims / 2}')98 99            yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_LONG), torch.tensor(long_factors, dtype=torch.float32))100            yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FACTORS_SHORT), torch.tensor(short_factors, dtype=torch.float32))101 102    def set_vocab(self):103        self._set_vocab_sentencepiece()104 105    def _reverse_hf_permute(self, weights: Tensor, n_head: int, n_kv_head: int | None = None) -> Tensor:106        if n_kv_head is not None and n_head != n_kv_head:107            n_head //= n_kv_head108 109        return (110            weights.reshape(n_head, 2, weights.shape[0] // n_head // 2, *weights.shape[1:])111            .swapaxes(1, 2)112            .reshape(weights.shape)113        )114 115 116# MiniCPM-V 4.6: text tower is Qwen3.5 (linear+full hybrid attention) wrapped under117# `model.language_model.*`; vision tower is SigLIP + a window-attention ViT merger118# + a final DownsampleMLP merger. The same HF arch is registered twice below: once as119# the LM (text mode) and once as the mmproj (vision mode), mirroring the Qwen3-VL setup.120 121@ModelBase.register("MiniCPMV4_6ForConditionalGeneration")122@ModelBase.example("openbmb/MiniCPM-V-4_6")123class MiniCPMV4_6TextModel(Qwen3_5TextModel):124    model_arch = gguf.MODEL_ARCH.QWEN35125 126    @classmethod127    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:128        name, gen = item129 130        if name.startswith("model.merger."):131            return None132        # MTP tensors are not used at inference yet; align with Qwen3Next behaviour133        if name.startswith("mtp"):134            return None135 136        return super().filter_tensors(item)137 138 139@ModelBase.register("MiniCPMV4_6ForConditionalGeneration")140@ModelBase.example("openbmb/MiniCPM-V-4_6")141class MiniCPMV4_6VisionModel(MmprojModel):142    def __init__(self, *args, **kwargs):143        super().__init__(*args, **kwargs)144        self.downsample_mode = self.preprocessor_config.get("downsample_mode", "16x")145        if self.downsample_mode not in {"4x", "16x"}:146            raise ValueError(f"Unsupported downsample mode: {self.downsample_mode}")147        if self.downsample_mode == "4x":148            self.model_tensors = {149                name: tensor for name, tensor in self.model_tensors.items()150                if ".vit_merger." not in name151            }152 153        if self.hparams_vision is not None:154            # In MiniCPM-V 4.6 `vision_config.image_size` (980) describes the SigLIP155            # positional embedding bucket grid (70 x 70), while the per-slice processing156            # resolution is the preprocessor's `scale_resolution` (typically 448).157            # The CLIP loader in tools/mtmd/clip.cpp consumes `clip.vision.image_size`158            # as the slice size and warmup resolution, so report `scale_resolution` there159            # to match the upstream MiniCPMV4_6ImageProcessorPil slicing rules.160            scale_resolution = self.preprocessor_config.get("scale_resolution")161            if scale_resolution is not None:162                self.hparams_vision["image_size"] = int(scale_resolution)163 164    def set_gguf_parameters(self):165        super().set_gguf_parameters()166        assert self.hparams_vision is not None167 168        # projector type string is consumed by clip_projector_type_from_string() in clip.cpp169        # (mapped to PROJECTOR_TYPE_MINICPMV4_6).170        self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.MINICPMV4_6)171 172        self.gguf_writer.add_vision_projector_scale_factor(173            2 if self.downsample_mode == "4x" else 4)174 175        # borrow wa_layer_indexes for vit_merger insertion point176        insert_layer_id = int(self.global_config.get(177            "insert_layer_id", self.hparams_vision.get("insert_layer_id", 6)))178        self.gguf_writer.add_vision_wa_layer_indexes([insert_layer_id])179 180        # SigLIP vision body uses gelu_pytorch_tanh, which matches ggml_gelu (tanh approx).181        self.gguf_writer.add_vision_use_gelu(True)182        self.gguf_writer.add_vision_attention_layernorm_eps(183            self.hparams_vision.get("layer_norm_eps", 1e-6))184 185    @classmethod186    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:187        name, gen = item188 189        # lm_head / MTP -> belong to the LM file190        if name.startswith(("lm_head.", "mtp")):191            return None192 193        return super().filter_tensors(item)194