char-fork/llama.cpp

git clone https://git.t4t.associates/char-fork/llama.cpp

Xuan-Son Nguyenconvert: add @ModelBase.example (#27208)fa88ae936

char/aliceai-t5-moe
brancheschar/aliceai-t5-moeall refs →
8.9 KiB203 linesraw
1from __future__ import annotations
2
3from typing import Any, Callable, Iterable, TYPE_CHECKING
4
5import numpy as np
6import torch
7
8if TYPE_CHECKING:
9    from torch import Tensor
10
11from .base import MmprojModel, ModelBase, TextModel, gguf
12
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.QWEN2VL
23
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    @classmethod
34    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
35        name, gen = item
36
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 None
49        self.hparams_vision["image_size"] = self.hparams_vision.get("image_size", 560)
50        # rename config.json values
51        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: # qwen2vl
54            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 None
60        hparams = self.hparams_vision
61        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] + 1
74            # validate n_wa_pattern
75            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 code
82        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.F32
87        return super().tensor_force_quant(name, new_name, bid, n_dims)
88
89    @classmethod
90    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
91        name, gen = item
92
93        if not name.startswith("visual."):
94            return None
95
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 needed
100        if ".qkv." in name:
101            if data_torch.ndim == 2: # weight
102                c3, _ = data_torch.shape
103            else: # bias
104                c3 = data_torch.shape[0]
105            assert c3 % 3 == 0
106            c = c3 // 3
107            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 Conv2Ds
115            c1, c2, kt, kh, kw = data_torch.shape
116            del c1, c2, kh, kw  # unused
117            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 = True
126
127    def __init__(self, *args, **kwargs):
128        super().__init__(*args, **kwargs)
129        assert self.hparams_audio is not None
130        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 None
137        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        # SinusoidsPositionEmbedding
142        assert self.hparams_audio is not None
143        max_timescale = 10000
144        length = 1500
145        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.F16
155        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 bias
160            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 = True
169    has_vision_encoder = True
170
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    @classmethod
182    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
183        name, gen = item
184
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 None
190
191        if "audio_bos_eos_token" in name:
192            # this tensor is left unused in transformers code
193            # https://github.com/huggingface/transformers/blob/6e3063422c4b1c014aa60c32b9254fd2902f0f28/src/transformers/models/qwen2_5_omni/modular_qwen2_5_omni.py#L1809
194            return None
195
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 tensors