Felipe97/llama-cpp-compiled
01.1k
1from __future__ import annotations2 3import json4import re5 6from pathlib import Path7from typing import Callable, Iterable, TYPE_CHECKING8 9import torch10 11if TYPE_CHECKING:12 from torch import Tensor13 14from .base import MmprojModel, ModelBase, TextModel, gguf, logger15 16from .qwen import QwenModel17 18 19@ModelBase.register("HunYuanMoEV1ForCausalLM")20@ModelBase.example("tencent/Hunyuan-A13B-Instruct")21class HunYuanMoEModel(TextModel):22 model_arch = gguf.MODEL_ARCH.HUNYUAN_MOE23 24 def set_vocab(self):25 from transformers import AutoTokenizer26 tokenizer = AutoTokenizer.from_pretrained(self.dir_model, trust_remote_code=True)27 28 # 1. Get the pre-tokenizer identifier hash29 tokpre = self.get_vocab_base_pre(tokenizer)30 31 # 2. Reverse-engineer the merges list from mergeable_ranks32 merges = []33 vocab = {}34 mergeable_ranks = tokenizer.mergeable_ranks # ty: ignore[unresolved-attribute]35 for token, rank in mergeable_ranks.items():36 vocab[QwenModel.token_bytes_to_string(token)] = rank37 if len(token) == 1:38 continue39 merged = QwenModel.bpe(mergeable_ranks, token, max_rank=rank)40 if len(merged) == 2: # todo this is an assert in Qwen, why?41 merges.append(' '.join(map(QwenModel.token_bytes_to_string, merged)))42 43 # 3. Generate the tokens and toktypes lists44 vocab_size = self.hparams["vocab_size"]45 assert tokenizer.vocab_size == vocab_size # ty: ignore[unresolved-attribute]46 special_tokens = tokenizer.special_tokens # ty: ignore[unresolved-attribute]47 reverse_vocab = {id_ : encoded_tok for encoded_tok, id_ in {**vocab, **special_tokens}.items()}48 tokens: list[str] = []49 toktypes: list[int] = []50 for i in range(vocab_size):51 if i not in reverse_vocab:52 tokens.append(f"[PAD{i}]")53 toktypes.append(gguf.TokenType.UNUSED)54 else:55 token = reverse_vocab[i]56 tokens.append(token)57 if i in special_tokens.values():58 toktypes.append(gguf.TokenType.CONTROL)59 else:60 toktypes.append(gguf.TokenType.NORMAL)61 62 # 4. Write all vocab-related fields to the GGUF writer63 self.gguf_writer.add_tokenizer_model("gpt2")64 self.gguf_writer.add_tokenizer_pre(tokpre)65 self.gguf_writer.add_token_list(tokens)66 self.gguf_writer.add_token_types(toktypes)67 self.gguf_writer.add_token_merges(merges)68 69 # 5. Add special tokens and chat templates70 special_vocab = gguf.SpecialVocab(self.dir_model, load_merges=False)71 special_vocab.add_to_gguf(self.gguf_writer)72 # FIX for BOS token: Overwrite incorrect id read from config.json73 self.gguf_writer.add_bos_token_id(127959) # <|bos|>74 75 def set_gguf_parameters(self):76 super().set_gguf_parameters()77 hparams = self.hparams78 79 self.gguf_writer.add_expert_shared_feed_forward_length(hparams["intermediate_size"])80 81 moe_intermediate_size = hparams["moe_intermediate_size"]82 assert all(n == moe_intermediate_size[0] for n in moe_intermediate_size)83 self.gguf_writer.add_expert_feed_forward_length(moe_intermediate_size[0])84 85 moe_topk = hparams["moe_topk"]86 assert all(topk == moe_topk[0] for topk in moe_topk)87 self.gguf_writer.add_expert_used_count(moe_topk[0])88 89 moe_shared_expert = hparams["num_shared_expert"]90 assert all(n == moe_shared_expert[0] for n in moe_shared_expert)91 self.gguf_writer.add_expert_shared_count(moe_shared_expert[0])92 93 # Rope94 if self.rope_parameters.get("rope_type") == "dynamic":95 # HunYuan uses NTK Aware Alpha based scaling. Original implementation: https://www.reddit.com/r/LocalLLaMA/comments/14lz7j5/ntkaware_scaled_rope_allows_llama_models_to_have/96 # 1000 corresponds to a usable context length of 256k (https://github.com/Tencent-Hunyuan/Hunyuan-A13B/blob/main/report/Hunyuan_A13B_Technical_Report.pdf)97 alpha = self.rope_parameters.get("alpha", 1000)98 base = self.rope_parameters.get("rope_theta", 10000.0)99 dim = (hparams["hidden_size"] // hparams["num_attention_heads"]) # 128100 scaled_base = base * (alpha ** (dim / (dim - 2))) # 10000 * (1000 ** (128 / 126)) = 11158839.9251101 self.gguf_writer.add_rope_freq_base(scaled_base)102 self.gguf_writer.add_rope_scaling_type(gguf.RopeScalingType.NONE)103 self.gguf_writer.add_rope_scaling_factor(1)104 # There is no consistent way to calculate ctx from alpha, and the config is incorrectly set to 32k105 self.gguf_writer.add_rope_scaling_orig_ctx_len(256 * 1024) # 256k context length106 self.gguf_writer.add_context_length(256 * 1024) # 256k context length107 108 # if any of our assumptions about the values are wrong, something has changed and this may need to be updated109 assert alpha == 1000 and base == 10000.0 and dim == 128 and self.hparams["max_position_embeddings"] in [32 * 1024, 256 * 1024] , \110 "HunYuan dynamic RoPE scaling assumptions changed, please update the logic or context length manually"111 112 _experts: list[dict[str, Tensor]] | None = None113 114 def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:115 if name == "lm_head.weight":116 if self.hparams.get("tie_word_embeddings", False):117 logger.info("Skipping tied output layer 'lm_head.weight'")118 return119 120 if name.find("mlp.experts") != -1:121 n_experts = self.find_hparam(["num_local_experts", "num_experts"])122 assert bid is not None123 124 if self._experts is None:125 self._experts = [{} for _ in range(self.block_count)]126 127 self._experts[bid][name] = data_torch128 129 if len(self._experts[bid]) >= n_experts * 3:130 # merge the experts into a single 3d tensor131 for w_name in ["down_proj", "gate_proj", "up_proj"]:132 datas: list[Tensor] = []133 134 for xid in range(n_experts):135 ename = f"model.layers.{bid}.mlp.experts.{xid}.{w_name}.weight"136 datas.append(self._experts[bid][ename])137 del self._experts[bid][ename]138 139 data_torch = torch.stack(datas, dim=0)140 merged_name = f"model.layers.{bid}.mlp.experts.{w_name}.weight"141 142 yield from super().modify_tensors(data_torch, merged_name, bid)143 return144 else:145 return146 147 yield from super().modify_tensors(data_torch, name, bid)148 149 def prepare_tensors(self):150 super().prepare_tensors()151 if self._experts is not None:152 experts = [k for d in self._experts for k in d.keys()]153 if len(experts) > 0:154 raise ValueError(f"Unprocessed experts: {experts}")155 156 157@ModelBase.register("HunYuanDenseV1ForCausalLM")158@ModelBase.example("tencent/Hunyuan-4B-Instruct")159class HunYuanModel(TextModel):160 model_arch = gguf.MODEL_ARCH.HUNYUAN_DENSE161 162 def _get_eod_token_id(self) -> int | None:163 """Get the actual end-of-generation token from config (eod_token_id)."""164 return self.hparams.get("eod_token_id")165 166 def _get_eot_token_id(self) -> int | None:167 """Get the end-of-turn token from generation_config.json.168 This is the first entry in eos_token_id when it's a list."""169 gen_cfg_path = self.dir_model / "generation_config.json"170 if gen_cfg_path.is_file():171 with open(gen_cfg_path, encoding="utf-8") as f:172 gen_cfg = json.load(f)173 eos = gen_cfg.get("eos_token_id")174 if isinstance(eos, list) and len(eos) >= 2:175 return eos[0]176 return None177 178 def _fix_special_tokens(self):179 """Fix EOS/EOT tokens that are incorrect in upstream configs."""180 eod_id = self._get_eod_token_id()181 if eod_id is not None:182 self.gguf_writer.add_eos_token_id(eod_id)183 eot_id = self._get_eot_token_id()184 if eot_id is not None:185 self.gguf_writer.add_eot_token_id(eot_id)186 187 def set_vocab(self):188 if (self.dir_model / "tokenizer.json").is_file():189 tokens, toktypes, tokpre = self.get_vocab_base()190 self.gguf_writer.add_tokenizer_model("gpt2")191 self.gguf_writer.add_tokenizer_pre(tokpre)192 self.gguf_writer.add_token_list(tokens)193 self.gguf_writer.add_token_types(toktypes)194 195 # Some HunYuanVL variants (e.g. OCR-style configs) have pad_token_id=-1;196 # guard SpecialVocab so it doesn't try to emit an invalid pad id.197 token_types = None198 if (self.hparams.get("pad_token_id") or 0) < 0:199 token_types = ('bos', 'eos', 'unk', 'sep', 'cls', 'mask')200 special_vocab = gguf.SpecialVocab(self.dir_model, load_merges=True, special_token_types=token_types)201 special_vocab.add_to_gguf(self.gguf_writer)202 self._fix_special_tokens()203 else:204 from transformers import AutoTokenizer205 tokenizer = AutoTokenizer.from_pretrained(self.dir_model, trust_remote_code=True)206 207 # 1. Get the pre-tokenizer identifier hash208 tokpre = self.get_vocab_base_pre(tokenizer)209 210 # 2. Reverse-engineer the merges list from mergeable_ranks211 merges = []212 vocab = {}213 mergeable_ranks = tokenizer.mergeable_ranks # ty: ignore[unresolved-attribute]214 for token, rank in mergeable_ranks.items():215 vocab[QwenModel.token_bytes_to_string(token)] = rank216 if len(token) == 1:217 continue218 merged = QwenModel.bpe(mergeable_ranks, token, max_rank=rank)219 if len(merged) == 2:220 merges.append(' '.join(map(QwenModel.token_bytes_to_string, merged)))221 222 # 3. Generate the tokens and toktypes lists223 vocab_size = self.hparams["vocab_size"]224 assert tokenizer.vocab_size == vocab_size # ty: ignore[unresolved-attribute]225 special_tokens = tokenizer.special_tokens # ty: ignore[unresolved-attribute]226 reverse_vocab = {id_ : encoded_tok for encoded_tok, id_ in {**vocab, **special_tokens}.items()}227 tokens: list[str] = []228 toktypes: list[int] = []229 for i in range(vocab_size):230 if i not in reverse_vocab:231 tokens.append(f"[PAD{i}]")232 toktypes.append(gguf.TokenType.UNUSED)233 else:234 token = reverse_vocab[i]235 tokens.append(token)236 if i in special_tokens.values():237 toktypes.append(gguf.TokenType.CONTROL)238 else:239 toktypes.append(gguf.TokenType.NORMAL)240 241 # 4. Write all vocab-related fields to the GGUF writer242 self.gguf_writer.add_tokenizer_model("gpt2")243 self.gguf_writer.add_tokenizer_pre(tokpre)244 self.gguf_writer.add_token_list(tokens)245 self.gguf_writer.add_token_types(toktypes)246 self.gguf_writer.add_token_merges(merges)247 248 # 5. Add special tokens and chat templates249 special_vocab = gguf.SpecialVocab(self.dir_model, load_merges=False)250 special_vocab.add_to_gguf(self.gguf_writer)251 # FIX for BOS token: Overwrite incorrect id read from config.json252 if self.hparams['hidden_size'] == 4096:253 self.gguf_writer.add_bos_token_id(127958) # only for 7b dense, fix <|bos|> token254 self._fix_special_tokens()255 256 def set_gguf_parameters(self):257 # Some HunYuanVL variants set num_experts=1 (not real MoE);258 # prevent the parent class from emitting expert_count metadata in that case.259 saved_num_experts = self.hparams.pop("num_experts", None)260 super().set_gguf_parameters()261 if saved_num_experts is not None and saved_num_experts > 1:262 self.hparams["num_experts"] = saved_num_experts263 hparams = self.hparams264 265 # Rope266 if self.rope_parameters.get("rope_type") in ("dynamic", "xdrope"):267 # HunYuan uses NTK Aware Alpha based scaling. Original implementation: https://www.reddit.com/r/LocalLLaMA/comments/14lz7j5/ntkaware_scaled_rope_allows_llama_models_to_have/268 # 1000 corresponds to a usable context length of 256k (https://github.com/Tencent-Hunyuan/Hunyuan-A13B/blob/main/report/Hunyuan_A13B_Technical_Report.pdf)269 alpha = self.rope_parameters.get("alpha", 50)270 base = self.rope_parameters.get("rope_theta", 10000.0)271 dim = hparams["head_dim"]272 scaled_base = base * (alpha ** (dim / (dim - 2)))273 self.gguf_writer.add_rope_freq_base(scaled_base)274 self.gguf_writer.add_rope_scaling_type(gguf.RopeScalingType.NONE)275 self.gguf_writer.add_rope_scaling_factor(1)276 if self.rope_parameters.get("rope_type") == "dynamic":277 # There is no consistent way to calculate ctx from alpha, and the config is incorrectly set to 32k278 self.gguf_writer.add_rope_scaling_orig_ctx_len(256 * 1024) # 256k context length279 self.gguf_writer.add_context_length(256 * 1024) # 256k context length280 281 # if any of our assumptions about the values are wrong, something has changed and this may need to be updated282 assert base == 10000.0 and self.hparams["max_position_embeddings"] in [32 * 1024, 256 * 1024] , \283 "HunYuan dynamic RoPE scaling assumptions changed, please update the logic or context length manually"284 285 def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:286 if name == "lm_head.weight":287 if self.hparams.get("tie_word_embeddings", False):288 logger.info("Skipping tied output layer 'lm_head.weight'")289 return290 291 yield from super().modify_tensors(data_torch, name, bid)292 293 294@ModelBase.register("HunYuanVLForConditionalGeneration")295@ModelBase.example("tencent/HunyuanOCR")296class HunyuanVLVisionModel(MmprojModel):297 def __init__(self, *args, **kwargs):298 super().__init__(*args, **kwargs)299 assert self.hparams_vision is not None300 # HunyuanVL uses max_image_size instead of image_size301 if "image_size" not in self.hparams_vision:302 self.hparams_vision["image_size"] = self.hparams_vision.get("max_image_size", 2048)303 304 def set_gguf_parameters(self):305 super().set_gguf_parameters()306 assert self.hparams_vision is not None307 vcfg = self.hparams_vision308 self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.HUNYUANVL)309 self.gguf_writer.add_vision_use_gelu(True)310 self.gguf_writer.add_vision_attention_layernorm_eps(vcfg.get("rms_norm_eps", 1e-5))311 self.gguf_writer.add_vision_spatial_merge_size(vcfg.get("spatial_merge_size", 2))312 self.gguf_writer.add_vision_min_pixels(int(self.preprocessor_config["min_pixels"]))313 self.gguf_writer.add_vision_max_pixels(int(self.preprocessor_config["max_pixels"]))314 315 @classmethod316 def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:317 name, gen = item318 319 if not name.startswith("vit."):320 return None321 322 return super().filter_tensors(item)323 324 def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:325 # strip CLS token (row 0) from position embeddings so resize_position_embeddings works326 if "position_embedding" in name:327 data_torch = data_torch[1:] # [n_patches+1, n_embd] -> [n_patches, n_embd]328 yield from super().modify_tensors(data_torch, name, bid)329 330 def tensor_force_quant(self, name, new_name, bid, n_dims):331 # force conv weights to F32 or F16 to avoid BF16 IM2COL issues on Metal332 # HunyuanVL emit the ViT -> LLM projection as mm.0/mm.2.333 if ("mm.0." in new_name or "mm.2." in new_name) and new_name.endswith(".weight"):334 return gguf.GGMLQuantizationType.F16 if self.ftype == gguf.LlamaFileType.MOSTLY_F16 else gguf.GGMLQuantizationType.F32335 return super().tensor_force_quant(name, new_name, bid, n_dims)336 337 338@ModelBase.register("HunYuanVLForConditionalGeneration")339@ModelBase.example("tencent/HunyuanOCR")340class HunyuanVLTextModel(HunYuanModel):341 model_arch = gguf.MODEL_ARCH.HUNYUAN_VL342 343 def __init__(self, dir_model: Path, *args, **kwargs):344 super().__init__(dir_model, *args, **kwargs)345 # transformers 5.13.0 encodes HunyuanVL XD-RoPE as dynamic + mrope_section.346 # Normalize it to avoid the HunYuan dynamic-RoPE context assertion.347 if self.rope_parameters.get("rope_type") == "dynamic" and "mrope_section" in self.rope_parameters:348 self.rope_parameters["rope_type"] = "xdrope"349 self.rope_parameters["type"] = "xdrope"350 self.rope_parameters["xdrope_section"] = list(self.rope_parameters["mrope_section"])351 352 def set_gguf_parameters(self):353 super().set_gguf_parameters()354 355 # XD-RoPE metadata for the HunyuanVL;356 if self.rope_parameters.get("rope_type") != "xdrope":357 return358 359 self.gguf_writer.add_rope_freq_base(float(self.rope_parameters["rope_theta"]))360 self.gguf_writer.add_rope_scaling_alpha(float(self.rope_parameters["alpha"]))361 self.gguf_writer.add_rope_scaling_type(gguf.RopeScalingType.NONE)362 self.gguf_writer.add_rope_scaling_factor(float(self.rope_parameters.get("factor", 1)))363 364 ctx_len = int(self.hparams["max_position_embeddings"])365 self.gguf_writer.add_rope_scaling_orig_ctx_len(ctx_len)366 self.gguf_writer.add_context_length(ctx_len)367 368 self.gguf_writer.add_rope_dimension_sections(list(self.rope_parameters["xdrope_section"]))369 370 371@ModelBase.register("HYV3ForCausalLM")372@ModelBase.example("tencent/Hy3")373class HYV3Model(TextModel):374 model_arch = gguf.MODEL_ARCH.HY_V3375 supports_mtp_export = True376 377 # Trunk layer count, stashed before indexing so the classmethod378 # filter_tensors can identify the appended MTP block(s) (mirrors379 # Step35Model).380 _n_main_layers: int | None = None381 382 def __init__(self, *args, **kwargs):383 super().__init__(*args, **kwargs)384 # NextN/MTP layers are appended past num_hidden_layers; extend the385 # tensor map so the MTP block's tensors resolve to blk.<n>.* names.386 n_nextn = int(self.hparams.get("num_nextn_predict_layers", 0))387 if n_nextn > 0 and not self.no_mtp:388 self.block_count += n_nextn389 self.tensor_map = gguf.get_tensor_name_map(self.model_arch, self.block_count)390 391 def index_tensors(self, remote_hf_model_id: str | None = None):392 type(self)._n_main_layers = self.hparams["num_hidden_layers"]393 return super().index_tensors(remote_hf_model_id=remote_hf_model_id)394 395 def set_vocab(self):396 self._set_vocab_gpt2()397 398 def set_gguf_parameters(self):399 super().set_gguf_parameters()400 self.gguf_writer.add_expert_feed_forward_length(self.hparams["moe_intermediate_size"])401 self.gguf_writer.add_expert_shared_feed_forward_length(402 self.hparams["moe_intermediate_size"] * self.hparams.get("num_shared_experts", 1)403 )404 self.gguf_writer.add_expert_weights_norm(self.hparams.get("route_norm", True))405 self.gguf_writer.add_expert_weights_scale(float(self.hparams.get("router_scaling_factor", 1.0)))406 # sigmoid router with expert selection bias407 self.gguf_writer.add_expert_gating_func(gguf.ExpertGatingFuncType.SIGMOID)408 409 n_nextn = int(self.hparams.get("num_nextn_predict_layers", 0))410 if n_nextn > 0 and not self.no_mtp:411 self.gguf_writer.add_nextn_predict_layers(n_nextn)412 413 @classmethod414 def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:415 if (titem := super().filter_tensors(item)) is None:416 return None417 name, gen = titem418 419 # HY V3 appends the MTP block(s) past num_hidden_layers.420 assert cls._n_main_layers is not None421 is_mtp = (m := re.match(r"model\.layers\.(\d+)\.", name)) is not None and int(m.group(1)) >= cls._n_main_layers422 423 # --no-mtp: drop the appended MTP block(s) entirely.424 if is_mtp and cls.no_mtp:425 return None426 # --mtp: keep ONLY MTP-block tensors plus the shared embeddings/norm/427 # lm_head (so the resulting GGUF carries just the draft head).428 if cls.mtp_only and not is_mtp and name not in (429 "model.embed_tokens.weight", "model.norm.weight", "lm_head.weight",430 ):431 return None432 433 # The MTP block's trailing final_layernorm (applied after the decoder434 # block, before the shared LM head) maps to nextn.shared_head_norm.435 if is_mtp:436 name = name.replace(".final_layernorm.", ".shared_head.norm.")437 438 return name, gen439 440 _experts: list[dict[str, Tensor]] | None = None441 442 def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:443 # merge the per-expert tensors into stacked 3d tensors444 if name.startswith("model.layers.") and ".mlp.experts." in name:445 n_experts = self.find_hparam(["num_local_experts", "num_experts"])446 assert bid is not None447 448 if self._experts is None:449 self._experts = [{} for _ in range(self.block_count)]450 451 self._experts[bid][name] = data_torch452 453 if len(self._experts[bid]) >= n_experts * 3:454 for w_name in ("down_proj", "gate_proj", "up_proj"):455 datas: list[Tensor] = []456 for xid in range(n_experts):457 ename = f"model.layers.{bid}.mlp.experts.{xid}.{w_name}.weight"458 datas.append(self._experts[bid][ename])459 del self._experts[bid][ename]460 461 merged = torch.stack(datas, dim=0)462 yield from super().modify_tensors(merged, f"model.layers.{bid}.mlp.experts.{w_name}.weight", bid)463 return464 465 yield from super().modify_tensors(data_torch, name, bid)466 467 def prepare_tensors(self):468 super().prepare_tensors()469 if self._experts is not None:470 experts = [k for d in self._experts for k in d.keys()]471 if experts:472 raise ValueError(f"Unprocessed experts: {experts}")473