Felipe97/llama-cpp-compiled

1from __future__ import annotations2 3from typing import Any, Callable, Iterable, TYPE_CHECKING4 5import numpy as np6import torch7 8if TYPE_CHECKING:9    from torch import Tensor10 11from .base import MmprojModel, ModelBase, TextModel, gguf12 13 14@ModelBase.register(15    "Qwen2VLModel",16    "Qwen2VLForConditionalGeneration",17    "Qwen2_5_VLForConditionalGeneration",18    "Qwen2_5OmniModel",19)20@ModelBase.example("Qwen/Qwen2-VL-2B-Instruct", "Qwen/Qwen2.5-VL-3B-Instruct")21class Qwen2VLModel(TextModel):22    model_arch = gguf.MODEL_ARCH.QWEN2VL23 24    def set_gguf_parameters(self):25        super().set_gguf_parameters()26 27    def set_vocab(self):28        try:29            self._set_vocab_sentencepiece()30        except FileNotFoundError:31            self._set_vocab_gpt2()32 33    @classmethod34    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:35        name, gen = item36 37        if name.startswith("thinker."):38            name = name.replace("thinker.", "")39 40        return super().filter_tensors((name, gen))41 42 43@ModelBase.register("Qwen2VLModel", "Qwen2VLForConditionalGeneration", "Qwen2_5_VLForConditionalGeneration")44@ModelBase.example("Qwen/Qwen2-VL-2B-Instruct", "Qwen/Qwen2.5-VL-3B-Instruct")45class Qwen2VLVisionModel(MmprojModel):46    def __init__(self, *args, **kwargs):47        super().__init__(*args, **kwargs)48        assert self.hparams_vision is not None49        self.hparams_vision["image_size"] = self.hparams_vision.get("image_size", 560)50        # rename config.json values51        self.hparams_vision["num_attention_heads"] = self.hparams_vision.get("num_heads")52        self.hparams_vision["num_hidden_layers"] = self.hparams_vision.get("depth")53        if "embed_dim" in self.hparams_vision: # qwen2vl54            self.hparams_vision["intermediate_size"] = self.hparams_vision.get("hidden_size")55            self.hparams_vision["hidden_size"] = self.hparams_vision.get("embed_dim")56 57    def set_gguf_parameters(self):58        super().set_gguf_parameters()59        assert self.hparams_vision is not None60        hparams = self.hparams_vision61        model_type = self.global_config['model_type']62        if model_type == 'qwen2_vl':63            self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.QWEN2VL)64        elif model_type == 'qwen2_5_vl' or model_type == 'qwen2_5_omni':65            if model_type == 'qwen2_5_omni':66                self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.QWEN25O)67            else:68                self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.QWEN25VL)69            self.gguf_writer.add_vision_use_silu(True)70            # find n_wa_pattern (window attention pattern)71            fullatt_block_indexes = hparams.get("fullatt_block_indexes")72            assert fullatt_block_indexes is not None, "fullatt_block_indexes is required for qwen2_5_vl"73            n_wa_pattern = fullatt_block_indexes[0] + 174            # validate n_wa_pattern75            for i in range(1, len(fullatt_block_indexes)):76                if fullatt_block_indexes[i] - fullatt_block_indexes[i - 1] != n_wa_pattern:77                    raise ValueError(f"Invalid fullatt_block_indexes: {fullatt_block_indexes}")78            self.gguf_writer.add_vision_n_wa_pattern(n_wa_pattern)79        else:80            raise ValueError(f"Unknown QwenVL model type: {self.global_config['model_type']}")81        # default values below are taken from HF tranformers code82        self.gguf_writer.add_vision_attention_layernorm_eps(self.global_config.get("rms_norm_eps", 1e-6))83 84    def tensor_force_quant(self, name, new_name, bid, n_dims):85        if ".position_embd." in new_name:86            return gguf.GGMLQuantizationType.F3287        return super().tensor_force_quant(name, new_name, bid, n_dims)88 89    @classmethod90    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:91        name, gen = item92 93        if not name.startswith("visual."):94            return None95 96        return super().filter_tensors(item)97 98    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:99        # split QKV tensors if needed100        if ".qkv." in name:101            if data_torch.ndim == 2: # weight102                c3, _ = data_torch.shape103            else: # bias104                c3 = data_torch.shape[0]105            assert c3 % 3 == 0106            c = c3 // 3107            wq = data_torch[:c]108            wk = data_torch[c: c * 2]109            wv = data_torch[c * 2:]110            yield from super().modify_tensors(wq, name.replace("qkv", "q"), bid)111            yield from super().modify_tensors(wk, name.replace("qkv", "k"), bid)112            yield from super().modify_tensors(wv, name.replace("qkv", "v"), bid)113        elif 'patch_embed.proj.weight' in name:114            # split Conv3D into Conv2Ds115            c1, c2, kt, kh, kw = data_torch.shape116            del c1, c2, kh, kw  # unused117            assert kt == 2, "Current implementation only support temporal_patch_size of 2"118            yield (gguf.TENSOR_NAMES[gguf.MODEL_TENSOR.V_ENC_EMBD_PATCH] + ".weight"  , data_torch[:, :, 0, ...])119            yield (gguf.TENSOR_NAMES[gguf.MODEL_TENSOR.V_ENC_EMBD_PATCH] + ".weight.1", data_torch[:, :, 1, ...])120        else:121            yield from super().modify_tensors(data_torch, name, bid)122 123 124class Qwen25AudioModel(MmprojModel):125    has_audio_encoder = True126 127    def __init__(self, *args, **kwargs):128        super().__init__(*args, **kwargs)129        assert self.hparams_audio is not None130        self.hparams_audio["hidden_size"] = self.hparams_audio["d_model"]131        self.hparams_audio["intermediate_size"] = self.hparams_audio["encoder_ffn_dim"]132        self.hparams_audio["num_attention_heads"] = self.hparams_audio["encoder_attention_heads"]133 134    def set_gguf_parameters(self):135        super().set_gguf_parameters()136        assert self.hparams_audio is not None137        self.gguf_writer.add_audio_num_mel_bins(self.hparams_audio["num_mel_bins"])138        self.gguf_writer.add_audio_attention_layernorm_eps(self.hparams_audio.get("layer_norm_eps", 1e-5))139 140    def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:141        # SinusoidsPositionEmbedding142        assert self.hparams_audio is not None143        max_timescale = 10000144        length = 1500145        channels = self.hparams_audio["hidden_size"]146        log_timescale_increment = np.log(max_timescale) / (channels // 2 - 1)147        inv_timescales = torch.exp(-log_timescale_increment * torch.arange(channels // 2).float())148        scaled_time = torch.arange(length)[:, np.newaxis] * inv_timescales[np.newaxis, :]149        pos_embd = torch.cat([torch.sin(scaled_time), torch.cos(scaled_time)], dim=1).to(dtype=torch.float32)150        yield ("audio_tower.embed_positions.weight", pos_embd)151 152    def tensor_force_quant(self, name, new_name, bid, n_dims):153        if ".conv" in name and ".weight" in name:154            return gguf.GGMLQuantizationType.F16155        return super().tensor_force_quant(name, new_name, bid, n_dims)156 157    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:158        if "conv1.bias" in name or "conv2.bias" in name:159            # transpose conv1 and conv2 bias160            data_torch = data_torch.unsqueeze(-1)161 162        yield from MmprojModel.modify_tensors(self, data_torch, name, bid)163 164 165@ModelBase.register("Qwen2_5OmniModel")166@ModelBase.example("Qwen/Qwen2.5-Omni-3B")167class Qwen25OmniModel(Qwen2VLVisionModel, Qwen25AudioModel):168    has_audio_encoder = True169    has_vision_encoder = True170 171    def get_vision_config(self) -> dict[str, Any] | None:172        return self.global_config["thinker_config"].get("vision_config")173 174    def get_audio_config(self) -> dict[str, Any] | None:175        return self.global_config["thinker_config"].get("audio_config")176 177    def set_gguf_parameters(self):178        super().set_gguf_parameters()179        self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.QWEN25O)180 181    @classmethod182    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:183        name, gen = item184 185        if name.startswith("thinker."):186            name = name.replace("thinker.", "")187 188        if not name.startswith("visual.") and not name.startswith("audio_tower."):189            return None190 191        if "audio_bos_eos_token" in name:192            # this tensor is left unused in transformers code193            # https://github.com/huggingface/transformers/blob/6e3063422c4b1c014aa60c32b9254fd2902f0f28/src/transformers/models/qwen2_5_omni/modular_qwen2_5_omni.py#L1809194            return None195 196        return MmprojModel.filter_tensors((name, gen))197 198    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:199        if "visual." in name:200            yield from Qwen2VLVisionModel.modify_tensors(self, data_torch, name, bid)201        elif "audio_tower." in name:202            yield from Qwen25AudioModel.modify_tensors(self, data_torch, name, bid)203        return  # skip other tensors204