llama.cpp-xrip/conversion/deepseek.py

from __future__ import annotations

import json
import re
from pathlib import Path

from typing import Any, Callable, Iterable, TYPE_CHECKING

import numpy as np
import torch

if TYPE_CHECKING:
    from torch import Tensor

from .base import LazyTorchTensor, MmprojModel, ModelBase, TextModel, gguf, logger

from .qwen import QwenModel


@ModelBase.register("DeepseekOCRForCausalLM")
@ModelBase.example("deepseek-ai/DeepSeek-OCR")
class DeepseekOCRVisionModel(MmprojModel):
    # HF dynamic_preprocess() max_num, which differs per model
    preproc_max_tiles = 9

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.clip_projector_type = gguf.VisionProjectorType.DEEPSEEKOCR

    def set_gguf_parameters(self):
        super().set_gguf_parameters()
        hparams = self.hparams
        self.gguf_writer.add_clip_projector_type(self.clip_projector_type)
        # default values below are taken from HF tranformers code
        self.gguf_writer.add_vision_attention_layernorm_eps(hparams.get("layer_norm_eps", 1e-6))
        self.gguf_writer.add_vision_use_gelu(True)
        # calculate proj_scale_factor (used by tinygemma3 test model)
        image_seq_length = self.preprocessor_config.get("image_seq_length", 256)
        n_per_side = int(image_seq_length ** 0.5)
        image_size = self.hparams["image_size"]
        patch_size = self.hparams["patch_size"]
        proj_scale_factor = (image_size // patch_size) // n_per_side
        if proj_scale_factor > 0 and proj_scale_factor != 4:
            # we only need to write this if it's not the default value
            # in this case, we are converting a test model
            self.gguf_writer.add_vision_projector_scale_factor(proj_scale_factor)
        # @bluebread: there's no window_size in config but just add it here anyway
        self.gguf_writer.add_vision_window_size(self.hparams.get("window_size", 14))

        self.gguf_writer.add_vision_preproc_min_tiles(2)
        self.gguf_writer.add_vision_preproc_max_tiles(self.preproc_max_tiles)

        # SAM configuration
        sam_hparams = hparams['sam']
        self.gguf_writer.add_vision_sam_layers_count(sam_hparams['layers'])
        self.gguf_writer.add_vision_sam_embedding_length(sam_hparams['width'])
        self.gguf_writer.add_vision_sam_head_count(sam_hparams['heads'])

    def get_vision_config(self) -> dict[str, Any]:
        vision_config: dict[str, Any] | None = self.global_config.get("vision_config")

        if not vision_config:
            raise ValueError("DeepseekOCR model requires 'vision_config' in the model configuration, but it was not found")

        vision_config['sam'] = vision_config['width']['sam_vit_b']
        if vision_config['width'].get('clip-l-14-224') is not None:
            vision_config.update(vision_config['width']['clip-l-14-224'])
        if isinstance(vision_config['width'], int):
            vision_config['hidden_size'] = vision_config['width']
        if vision_config.get('heads') is not None:
            vision_config['num_heads'] = vision_config['heads']
            vision_config['intermediate_size'] = vision_config['heads'] * 4

        return vision_config

    def tensor_force_quant(self, name, new_name, bid, n_dims):
        for nq_name in ('.embeddings.', 'pos_embed', '.rel_pos_h', '.rel_pos_w', '.neck.', '.net_'):
            if nq_name in name:
                return gguf.GGMLQuantizationType.F32
        return super().tensor_force_quant(name, new_name, bid, n_dims)

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        if name.endswith("view_seperator"):
            data_torch = data_torch.unsqueeze(0)
        yield from super().modify_tensors(data_torch, name, bid)

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        name, gen = item

        # Only process vision-related tensors, skip language model tensors
        # Vision components: sam_model, vision_model, projector, image_newline, view_seperator
        # Language model components to skip: lm_head, embed_tokens, layers, norm
        if name.startswith(("lm_head.", "model.embed_tokens.", "model.layers.", "model.norm.")):
            return None

        if name.endswith("pos_embed") or name.endswith("rel_pos_h") or name.endswith("rel_pos_w"):
            name += ".weight"

        return super().filter_tensors((name, gen))


@ModelBase.register("UnlimitedOCRForCausalLM")
@ModelBase.example("baidu/Unlimited-OCR")
class UnlimitedOCRVisionModel(DeepseekOCRVisionModel):
    preproc_max_tiles = 32


@ModelBase.register("DeepseekOCR2ForCausalLM")
@ModelBase.example("deepseek-ai/DeepSeek-OCR-2")
class DeepseekOCR2VisionModel(DeepseekOCRVisionModel):
    preproc_max_tiles = 6

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.clip_projector_type = gguf.VisionProjectorType.DEEPSEEKOCR2

    def set_gguf_parameters(self):
        # the vision tower's qwen2 encoder is built from fixed defaults,
        # see build_qwen2_decoder_as_encoder() in deepencoderv2.py
        if self.hparams.get("patch_size") is None:
            self.hparams["patch_size"] = 16
        if self.hparams.get("intermediate_size") is None:
            self.hparams["intermediate_size"] = 4864
        if self.hparams.get("num_attention_heads") is None:
            self.hparams["num_attention_heads"] = 14
        super().set_gguf_parameters()
        # qwen2 encoder is GQA: 14 Q heads, 2 KV heads
        self.gguf_writer.add_vision_head_count_kv(2)

    def get_vision_config(self) -> dict[str, Any]:
        vision_config = super().get_vision_config()
        vision_config['hidden_size'] = vision_config['width']['qwen2-0-5b']['dim']
        if vision_config.get('layers') is None:
            vision_config['layers'] = 24
        return vision_config


@ModelBase.register("DeepseekForCausalLM")
@ModelBase.example("deepseek-ai/deepseek-moe-16b-chat")
class DeepseekModel(TextModel):
    model_arch = gguf.MODEL_ARCH.DEEPSEEK

    def set_vocab(self):
        try:
            self._set_vocab_sentencepiece()
        except FileNotFoundError:
            self._set_vocab_gpt2()

    def set_gguf_parameters(self):
        super().set_gguf_parameters()
        hparams = self.hparams
        if (rope_dim := hparams.get("head_dim")) is None:
            rope_dim = hparams["hidden_size"] // hparams["num_attention_heads"]

        self.gguf_writer.add_rope_dimension_count(rope_dim)
        self.gguf_writer.add_rope_scaling_type(gguf.RopeScalingType.NONE)
        self.gguf_writer.add_leading_dense_block_count(hparams["first_k_dense_replace"])
        self.gguf_writer.add_vocab_size(hparams["vocab_size"])
        self.gguf_writer.add_expert_feed_forward_length(hparams["moe_intermediate_size"])
        self.gguf_writer.add_expert_weights_scale(1.0)
        self.gguf_writer.add_expert_count(hparams["n_routed_experts"])
        self.gguf_writer.add_expert_shared_count(hparams["n_shared_experts"])

    _experts: list[dict[str, Tensor]] | None = None

    @staticmethod
    def permute(weights: Tensor, n_head: int, n_head_kv: int | None):
        if n_head_kv is not None and n_head != n_head_kv:
            n_head = n_head_kv
        return (weights.reshape(n_head, 2, weights.shape[0] // n_head // 2, *weights.shape[1:])
                .swapaxes(1, 2)
                .reshape(weights.shape))

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        n_head = self.hparams["num_attention_heads"]
        n_kv_head = self.hparams.get("num_key_value_heads")

        if name.endswith(("q_proj.weight", "q_proj.bias")):
            data_torch = DeepseekModel.permute(data_torch, n_head, n_head)
        if name.endswith(("k_proj.weight", "k_proj.bias")):
            data_torch = DeepseekModel.permute(data_torch, n_head, n_kv_head)

        # process the experts separately
        if name.find("mlp.experts") != -1:
            n_experts = self.hparams["n_routed_experts"]
            assert bid is not None

            if self._experts is None:
                self._experts = [{} for _ in range(self.block_count)]

            self._experts[bid][name] = data_torch

            if len(self._experts[bid]) >= n_experts * 3:
                # merge the experts into a single 3d tensor
                for w_name in ["down_proj", "gate_proj", "up_proj"]:
                    datas: list[Tensor] = []

                    for xid in range(n_experts):
                        ename = f"model.layers.{bid}.mlp.experts.{xid}.{w_name}.weight"
                        datas.append(self._experts[bid][ename])
                        del self._experts[bid][ename]

                    data_torch = torch.stack(datas, dim=0)

                    merged_name = f"model.layers.{bid}.mlp.experts.{w_name}.weight"

                    yield from super().modify_tensors(data_torch, merged_name, bid)
                return
            else:
                return

        yield from super().modify_tensors(data_torch, name, bid)

    def prepare_tensors(self):
        super().prepare_tensors()

        if self._experts is not None:
            # flatten `list[dict[str, Tensor]]` into `list[str]`
            experts = [k for d in self._experts for k in d.keys()]
            if len(experts) > 0:
                raise ValueError(f"Unprocessed experts: {experts}")


@ModelBase.register(
    "DeepseekV2ForCausalLM",
    "DeepseekV3ForCausalLM",
    "DeepseekOCRForCausalLM",
    "UnlimitedOCRForCausalLM",
    "KimiVLForConditionalGeneration",
    "KimiK25ForConditionalGeneration",
    "YoutuForCausalLM",
    "YoutuVLForConditionalGeneration",
)
@ModelBase.example("deepseek-ai/DeepSeek-V2-Lite", "deepseek-ai/DeepSeek-V3")
class DeepseekV2Model(TextModel):
    model_arch = gguf.MODEL_ARCH.DEEPSEEK2

    def _map_fp8_weight_names(self, name: str) -> tuple[str, ...]:
        # MLA projections do not have runtime weight-scale support.
        if name.endswith((".q_a_proj.weight", ".q_b_proj.weight", ".kv_a_proj_with_mqa.weight", ".kv_b_proj.weight")):
            return ()
        return super()._map_fp8_weight_names(name)

    # TODO @ngxson : remove this when we support MTP for deepseek models
    skip_mtp = True

    merge_expert = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        hparams: dict = ModelBase.load_hparams(self.dir_model, is_mistral_format=self.is_mistral_format)
        self.origin_hf_arch = hparams.get('architectures', [None])[0]

        # special handling for Deepseek OCR
        if self.origin_hf_arch in ("DeepseekOCRForCausalLM", "DeepseekOCR2ForCausalLM", "UnlimitedOCRForCausalLM"):
            self.model_arch = gguf.MODEL_ARCH.DEEPSEEK2OCR
            self.gguf_writer.arch = gguf.MODEL_ARCH_NAMES[self.model_arch]
            self.gguf_writer.add_architecture()
            # default jinja template
            self.gguf_writer.add_chat_template("{% for m in messages %}{{m['content']}}{% endfor %}")

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        name, _ = item
        # DeepSeek-OCR vision encoder (SAM + DeepSeek-OCR-2 qwen2 tower)
        if "sam_model" in name or "qwen2_model" in name:
            return None
        return super().filter_tensors(item)

    def set_vocab(self):
        try:
            self._set_vocab_gpt2()
            return
        except Exception:
            pass

        from transformers import AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(self.dir_model, trust_remote_code=True)
        tokpre = self.get_vocab_base_pre(tokenizer)

        if tokpre == "kimi-k2":
            # Build merges list using the approach similar to HunYuanMoE
            merges = []
            vocab = {}
            mergeable_ranks = tokenizer.model._mergeable_ranks  # ty: ignore[unresolved-attribute]
            for token, rank in mergeable_ranks.items():
                vocab[QwenModel.token_bytes_to_string(token)] = rank
                if len(token) == 1:
                    continue
                merged = QwenModel.bpe(mergeable_ranks, token, max_rank=rank)
                if len(merged) == 2:
                    merges.append(' '.join(map(QwenModel.token_bytes_to_string, merged)))

            # Build token list
            vocab_size = self.hparams["vocab_size"]
            special_tokens = tokenizer.special_tokens  # ty: ignore[unresolved-attribute]
            reverse_vocab = {id_ : encoded_tok for encoded_tok, id_ in {**vocab, **special_tokens}.items()}
            tokens: list[str] = []
            toktypes: list[int] = []

            for i in range(vocab_size):
                if i not in reverse_vocab:
                    tokens.append(f"[PAD{i}]")
                    toktypes.append(gguf.TokenType.UNUSED)
                else:
                    token = reverse_vocab[i]
                    tokens.append(token)
                    if i in special_tokens.values():
                        toktypes.append(gguf.TokenType.CONTROL)
                    else:
                        toktypes.append(gguf.TokenType.NORMAL)

            self.gguf_writer.add_tokenizer_model("gpt2")
            self.gguf_writer.add_tokenizer_pre(tokpre)
            self.gguf_writer.add_token_list(tokens)
            self.gguf_writer.add_token_types(toktypes)
            self.gguf_writer.add_token_merges(merges)

            special_vocab = gguf.SpecialVocab(self.dir_model, load_merges=False)
            special_vocab.add_to_gguf(self.gguf_writer)
        else:
            raise NotImplementedError(f"Deepseek pre-tokenizer {tokpre!r} is not supported yet!")

    def set_gguf_parameters(self):
        is_ocr = (self.model_arch == gguf.MODEL_ARCH.DEEPSEEK2OCR)

        if is_ocr:
            self.hparams['rope_theta'] = self.hparams.get('rope_theta', 10000.0)
        else:
            # note: deepseek2 using MLA converts into MQA (ie: GQA with 1 group)
            self.hparams["num_key_value_heads"] = 1

        self.hparams['rms_norm_eps'] = self.hparams.get('rms_norm_eps', 1e-6)

        super().set_gguf_parameters()
        hparams = self.hparams

        # first_k_dense_replace: number of leading layers using dense FFN instead of MoE
        # For non-MoE models (like Youtu), set to n_layer to use dense FFN for all layers
        # For MoE models (like DeepSeek-V2), this is the number of leading non-MoE layers
        has_moe = hparams.get("n_routed_experts") is not None
        first_k_dense_replace = hparams.get("first_k_dense_replace")
        if first_k_dense_replace is None:
            # Default: if no MoE, all layers are dense; if MoE, none are dense
            first_k_dense_replace = hparams["num_hidden_layers"] if not has_moe else 0
        self.gguf_writer.add_leading_dense_block_count(first_k_dense_replace)
        kv_lora_rank = hparams.get("kv_lora_rank", 512)
        self.gguf_writer.add_vocab_size(hparams["vocab_size"])
        if "q_lora_rank" in hparams and hparams["q_lora_rank"] is not None:
            self.gguf_writer.add_q_lora_rank(hparams["q_lora_rank"])

        # note: deepseek2 using MLA converts into MQA with larger heads, then decompresses to MHA
        if not is_ocr:
            self.gguf_writer.add_kv_lora_rank(kv_lora_rank)
            self.gguf_writer.add_key_length(kv_lora_rank + hparams["qk_rope_head_dim"])
            self.gguf_writer.add_value_length(kv_lora_rank)
            self.gguf_writer.add_key_length_mla(hparams["qk_nope_head_dim"] + hparams["qk_rope_head_dim"])
            self.gguf_writer.add_value_length_mla(hparams["v_head_dim"])

        # MoE parameters (required by C++ code for DEEPSEEK2 arch)
        # For non-MoE models like Youtu, use intermediate_size as expert_feed_forward_length
        moe_intermediate_size = self.find_hparam(["moe_intermediate_size", "intermediate_size"], optional=False)
        self.gguf_writer.add_expert_feed_forward_length(moe_intermediate_size)

        if (n_routed_experts := hparams.get("n_routed_experts")) is not None:
            self.gguf_writer.add_expert_count(n_routed_experts)

        # expert_shared_count is required by C++ code, default to 0 for non-MoE models
        n_shared_experts = hparams.get("n_shared_experts", 0)
        self.gguf_writer.add_expert_shared_count(n_shared_experts)

        # When not set, C++ code will use scale_w = false to skip the no-op scaling
        if (routed_scaling_factor := hparams.get("routed_scaling_factor")) is not None:
            self.gguf_writer.add_expert_weights_scale(routed_scaling_factor)

        if (norm_topk_prob := hparams.get("norm_topk_prob")) is not None and norm_topk_prob:
            self.gguf_writer.add_expert_weights_norm(norm_topk_prob)

        self.gguf_writer.add_rope_dimension_count(hparams["qk_rope_head_dim"])

        # Unlimited-OCR sliding window; written for metadata, the decoder ignores it (full MHA)
        if is_ocr:
            sliding_window = hparams.get("sliding_window_size") or hparams.get("sliding_window")
            if sliding_window:
                self.gguf_writer.add_sliding_window(sliding_window)

        if (rope_mscale_all := self.rope_parameters.get("mscale_all_dim")) is not None:
            # [TAG_DEEPSEEK2_YARN_LOG_MUL_FIX]
            # note: for legacy reasons, this is not consistent with the other usages of self.gguf_writer.add_rope_scaling_yarn_log_mul
            # ref https://github.com/ggml-org/llama.cpp/pull/17945
            self.gguf_writer.add_rope_scaling_yarn_log_mul(0.1 * rope_mscale_all)

    _experts: list[dict[str, Tensor]] | None = None

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        # skip lm_head.weight if tie_word_embeddings is True
        if self.hparams.get("tie_word_embeddings", False):
            if name == "lm_head.weight" or name == "model.lm_head.weight":
                logger.info("Skipping tied output layer 'lm_head.weight' (will use token_embd.weight)")
                return

        # skip Multi-Token Prediction (MTP) layers
        if self.skip_mtp:
            block_count = self.hparams["num_hidden_layers"]
            match = re.match(r"model.layers.(\d+)", name)
            if match and int(match.group(1)) >= block_count:
                return

        # process the experts separately
        if self.merge_expert and name.find("mlp.experts") != -1:
            n_experts = self.hparams["n_routed_experts"]
            assert bid is not None

            if self._experts is None:
                self._experts = [{} for _ in range(self.block_count)]

            self._experts[bid][name] = data_torch

            if len(self._experts[bid]) >= n_experts * 3:
                # merge the experts into a single 3d tensor
                for w_name in ["down_proj", "gate_proj", "up_proj"]:
                    datas: list[Tensor] = []

                    for xid in range(n_experts):
                        ename = f"model.layers.{bid}.mlp.experts.{xid}.{w_name}.weight"
                        datas.append(self._experts[bid][ename])
                        del self._experts[bid][ename]

                    data_torch = torch.stack(datas, dim=0)

                    merged_name = f"model.layers.{bid}.mlp.experts.{w_name}.weight"

                    yield from super().modify_tensors(data_torch, merged_name, bid)
                return
            else:
                return

        # note: MLA with the absorption optimization, needs these two split and k_b_proj transposed
        if name.endswith("kv_b_proj.weight"):
            name_kb = name.replace("kv_b_proj", "k_b_proj")
            name_vb = name.replace("kv_b_proj", "v_b_proj")

            n_head_kv = self.hparams["num_key_value_heads"]
            v_head_dim = self.hparams["v_head_dim"]
            qk_nope_head_dim = self.hparams["qk_nope_head_dim"]

            assert data_torch.shape[0] == n_head_kv * (v_head_dim + qk_nope_head_dim)

            kv_b = data_torch.view(n_head_kv, v_head_dim + qk_nope_head_dim, data_torch.shape[-1])
            k_b, v_b = torch.split(kv_b, [qk_nope_head_dim, v_head_dim], dim=1)
            k_b = k_b.transpose(1, 2)

            yield from super().modify_tensors(k_b, name_kb, bid)
            yield from super().modify_tensors(v_b, name_vb, bid)
            return

        yield from super().modify_tensors(data_torch, name, bid)

    def prepare_tensors(self):
        super().prepare_tensors()

        if self._experts is not None:
            # flatten `list[dict[str, Tensor]]` into `list[str]`
            experts = [k for d in self._experts for k in d.keys()]
            if len(experts) > 0:
                raise ValueError(f"Unprocessed experts: {experts}")


@ModelBase.register("DeepseekV32ForCausalLM")
@ModelBase.example("deepseek-ai/DeepSeek-V3.2-Exp")
class DeepseekV32Model(DeepseekV2Model):
    model_arch = gguf.MODEL_ARCH.DEEPSEEK32
    skip_mtp = False
    supports_mtp_export = True
    _n_main_layers: int | None = None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.block_count = self.hparams["num_hidden_layers"]
        if not self.no_mtp:
            self.block_count += self.hparams.get("num_nextn_predict_layers", 0)
        self.tensor_map = gguf.get_tensor_name_map(self.model_arch, self.block_count)

    def index_tensors(self, remote_hf_model_id: str | None = None):
        type(self)._n_main_layers = self.hparams["num_hidden_layers"]
        return super().index_tensors(remote_hf_model_id=remote_hf_model_id)

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        if (titem := super().filter_tensors(item)) is None:
            return None
        name, gen = titem

        # DeepSeek V3.2 appends the NextN/MTP block past num_hidden_layers
        # (model.layers.61 -> blk.61 in the 62-block file).
        assert cls._n_main_layers is not None
        is_mtp = (m := re.match(r"model\.layers\.(\d+)\.", name)) is not None and int(m.group(1)) >= cls._n_main_layers

        # --no-mtp: drop the appended NextN block entirely.
        if is_mtp and cls.no_mtp:
            return None
        # --mtp: keep ONLY NextN-block tensors plus the shared embeddings/
        # norm/lm_head (so the resulting GGUF carries just the draft head).
        if cls.mtp_only and not is_mtp and name not in (
            "model.embed_tokens.weight", "model.norm.weight", "lm_head.weight",
        ):
            return None

        return name, gen

    def set_vocab(self):
        from transformers import AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(self.dir_model)
        assert getattr(tokenizer, "add_bos_token", False), "Change value of add_bos_token to true in tokenizer_config.json file."
        self._set_vocab_gpt2()

    def set_gguf_parameters(self):
        super().set_gguf_parameters()

        # NextN/MTP prediction layers
        if not self.no_mtp and (num_nextn_predict_layers := self.hparams.get("num_nextn_predict_layers")) is not None:
            self.gguf_writer.add_nextn_predict_layers(num_nextn_predict_layers)

        # DSA indexer parameters
        self.gguf_writer.add_indexer_head_count(self.hparams["index_n_heads"])
        self.gguf_writer.add_indexer_key_length(self.hparams["index_head_dim"])
        self.gguf_writer.add_indexer_top_k(self.hparams["index_topk"])


@ModelBase.register("DeepseekV4ForCausalLM")
@ModelBase.example("deepseek-ai/DeepSeek-V4-Flash-Base")
class DeepseekV4Model(TextModel):
    model_arch = gguf.MODEL_ARCH.DEEPSEEK4
    supports_mtp_export = True
    _skipped_mtp_tensors = 0
    _dsv4_main_layers: int | None = None
    _dsv4_nextn_layers: int = 0

    def __init__(self, *args, **kwargs):
        type(self)._skipped_mtp_tensors = 0
        super().__init__(*args, **kwargs)

        with open(self.dir_model / "config.json", "r", encoding="utf-8") as f:
            raw_hparams = json.load(f)
        for key, value in raw_hparams.items():
            self.hparams.setdefault(key, value)

        # workaround for special rope_parameters (main/compress) in transformers 5.x
        if self.rope_parameters.get("full_attention", self.rope_parameters).get("rope_type") is None:
            if (rope_scaling := raw_hparams.get("rope_scaling")) is not None:
                if "rope_type" not in rope_scaling and (rope_type := rope_scaling.get("type")) is not None:
                    rope_scaling["rope_type"] = rope_type
                self.rope_parameters.update(**rope_scaling)

        self.block_count = self.hparams["num_hidden_layers"]
        if self.mtp_only:
            self.block_count += self.hparams.get("num_nextn_predict_layers", 0)
        self.tensor_map = gguf.get_tensor_name_map(self.model_arch, self.block_count)

        self._dsv4_fp8_dequantized: set[str] = set()
        self._dsv4_bf16_tensors: set[str] = set()
        self._dsv4_f32_tensors: set[str] = set()
        self._dsv4_mxfp4_generated = False
        self._collect_source_dtypes()

        if type(self)._skipped_mtp_tensors:
            logger.info("Skipping %d DeepSeek-V4 MTP tensor(s) for conversion v0", type(self)._skipped_mtp_tensors)

        # add a default chat template; if the model has a built-in template, it will be overridden later
        model_id_hint = self.remote_hf_model_id or self.dir_model.name
        is_0731 = "0731" in model_id_hint
        template_name = "deepseek-ai-DeepSeek-V4-Flash-0731.jinja" if is_0731 else "deepseek-ai-DeepSeek-V4.jinja"
        template_path = Path(__file__).parent.parent / "models" / "templates" / template_name
        if template_path.is_file():
            with open(template_path, "r", encoding="utf-8") as f:
                self.gguf_writer.add_chat_template(f.read())

    def index_tensors(self, remote_hf_model_id: str | None = None) -> dict[str, Callable[[], Tensor]]:
        type(self)._dsv4_main_layers = self.hparams["num_hidden_layers"]
        type(self)._dsv4_nextn_layers = self.hparams.get("num_nextn_predict_layers", 0)
        return super().index_tensors(remote_hf_model_id=remote_hf_model_id)

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        name, gen = item
        if name.startswith(("aligner.", "image_")):
            return None
        if name.startswith("mtp."):
            if not cls.mtp_only:
                cls._skipped_mtp_tensors += 1
                return None

            assert cls._dsv4_main_layers is not None
            parts = name.split(".", 2)
            if len(parts) < 3 or not parts[1].isdecimal():
                raise ValueError(f"Unexpected DeepSeek-V4 MTP tensor {name!r}")

            mtp_idx = int(parts[1])
            if mtp_idx >= cls._dsv4_nextn_layers:
                raise ValueError(f"Unexpected DeepSeek-V4 MTP layer {mtp_idx}")

            bid = cls._dsv4_main_layers + mtp_idx
            suffix = parts[2]
            root_hc_head = {
                "hc_head_fn",
                "hc_head_base",
                "hc_head_scale",
            }
            if suffix in root_hc_head:
                name = suffix
            elif suffix in (
                "e_proj.weight", "e_proj.scale",
                "h_proj.weight", "h_proj.scale",
            ):
                name = f"layers.{bid}.nextn.{suffix}"
            elif suffix == "enorm.weight":
                name = f"layers.{bid}.nextn.enorm.weight"
            elif suffix == "hnorm.weight":
                name = f"layers.{bid}.nextn.hnorm.weight"
            elif suffix == "norm.weight":
                name = f"layers.{bid}.nextn.shared_head_norm.weight"
            else:
                name = f"layers.{bid}.{suffix}"
            return name, gen

        if cls.mtp_only:
            keep = name in (
                "embed.weight",
                "norm.weight",
                "head.weight",
                "head.scale",
            )
            if not keep:
                return None

        return super().filter_tensors((name, gen))

    @staticmethod
    def _float8_dtypes() -> tuple[torch.dtype, ...]:
        return tuple(
            dtype for dtype in (
                getattr(torch, "float8_e4m3fn", None),
                getattr(torch, "float8_e5m2", None),
            ) if dtype is not None
        )

    @staticmethod
    def _e8m0_to_float(scale: Tensor) -> Tensor:
        torch_float8_e8m0 = getattr(torch, "float8_e8m0fnu", None)
        if torch_float8_e8m0 is not None and scale.dtype == torch_float8_e8m0:
            return scale.float()

        bits = scale.view(torch.uint8).float()
        return torch.exp2(bits - 127.0)

    def _collect_source_dtypes(self) -> None:
        for name, gen in self.model_tensors.items():
            dtype = gen().dtype
            if dtype == torch.bfloat16:
                self._dsv4_bf16_tensors.add(name)
            elif dtype == torch.float32:
                self._dsv4_f32_tensors.add(name)

    def set_gguf_parameters(self):
        super().set_gguf_parameters()
        hparams = self.hparams

        self.gguf_writer.add_rope_dimension_count(hparams["qk_rope_head_dim"])
        self.gguf_writer.add_q_lora_rank(hparams["q_lora_rank"])
        self.gguf_writer.add_sliding_window(hparams["sliding_window"])

        self.gguf_writer.add_expert_feed_forward_length(hparams["moe_intermediate_size"])
        self.gguf_writer.add_expert_shared_count(hparams["n_shared_experts"])
        self.gguf_writer.add_expert_weights_scale(hparams["routed_scaling_factor"])
        self.gguf_writer.add_expert_weights_norm(hparams["norm_topk_prob"])
        self.gguf_writer.add_swiglu_clamp_exp([hparams["swiglu_limit"]] * self.block_count)
        self.gguf_writer.add_swiglu_clamp_shexp([hparams["swiglu_limit"]] * self.block_count)

        self.gguf_writer.add_indexer_head_count(hparams["index_n_heads"])
        self.gguf_writer.add_indexer_key_length(hparams["index_head_dim"])
        self.gguf_writer.add_indexer_top_k(hparams["index_topk"])

        self.gguf_writer.add_attention_output_group_count(hparams["o_groups"])
        self.gguf_writer.add_attention_output_lora_rank(hparams["o_lora_rank"])
        self.gguf_writer.add_attention_compress_ratios(hparams["compress_ratios"])
        self.gguf_writer.add_attention_compress_rope_freq_base(hparams["compress_rope_theta"])
        self.gguf_writer.add_hyper_connection_count(hparams["hc_mult"])
        self.gguf_writer.add_hyper_connection_sinkhorn_iterations(hparams["hc_sinkhorn_iters"])
        self.gguf_writer.add_hyper_connection_epsilon(hparams["hc_eps"])
        self.gguf_writer.add_hash_layer_count(hparams["num_hash_layers"])
        if self.model_arch == gguf.MODEL_ARCH.DEEPSEEK4:
            self.gguf_writer.add_embedding_length_out(hparams["hidden_size"] * hparams["hc_mult"])
        if self.mtp_only and (num_nextn_predict_layers := hparams.get("num_nextn_predict_layers", 0)) > 0:
            self.gguf_writer.add_nextn_predict_layers(num_nextn_predict_layers)

    def dequant_model(self):
        fp8_dtypes = self._float8_dtypes()
        tensors_to_remove: list[str] = []

        def dequant_fp8_weight(weight: Tensor, scale: Tensor) -> Tensor:
            out_features, in_features = weight.shape
            scale_f = self._e8m0_to_float(scale)
            scale_f = scale_f.repeat_interleave(128, 0)[:out_features]
            scale_f = scale_f.repeat_interleave(128, 1)[:, :in_features]
            return weight.float() * scale_f

        for name in list(self.model_tensors.keys()):
            if not name.endswith(".scale"):
                continue
            weight_name = name.removesuffix(".scale") + ".weight"
            if weight_name not in self.model_tensors:
                continue

            weight = self.model_tensors[weight_name]
            scale = self.model_tensors[name]
            if weight().dtype not in fp8_dtypes:
                continue

            self.model_tensors[weight_name] = lambda w=weight, s=scale: dequant_fp8_weight(w(), s())
            self._dsv4_fp8_dequantized.add(weight_name)
            tensors_to_remove.append(name)

        for name in tensors_to_remove:
            del self.model_tensors[name]

    def _write_mxfp4_expert_tensor(self, bid: int, proj: str, tensor_key: gguf.MODEL_TENSOR) -> list[str]:
        n_experts = self.hparams["n_routed_experts"]
        data: np.ndarray | None = None
        consumed: list[str] = []

        for eid in range(n_experts):
            weight_name = f"layers.{bid}.ffn.experts.{eid}.{proj}.weight"
            scale_name = f"layers.{bid}.ffn.experts.{eid}.{proj}.scale"
            if weight_name not in self.model_tensors or scale_name not in self.model_tensors:
                raise KeyError(f"Missing routed expert tensors for {weight_name}")

            weight = LazyTorchTensor.to_eager(self.model_tensors[weight_name]())
            scale = LazyTorchTensor.to_eager(self.model_tensors[scale_name]())
            packed = self.repack_mxfp4_blocks(weight, scale)
            if data is None:
                data = np.empty((n_experts, *packed.shape), dtype=packed.dtype)
            data[eid] = packed
            consumed.extend((weight_name, scale_name))

        assert data is not None
        new_name = self.format_tensor_name(tensor_key, bid)
        shape = gguf.quant_shape_from_byte_shape(data.shape, gguf.GGMLQuantizationType.MXFP4)
        logger.info(f"{new_name}: repacked routed experts to MXFP4, shape = {{{', '.join(str(n) for n in reversed(shape))}}}")
        self.gguf_writer.add_tensor(new_name, data, raw_dtype=gguf.GGMLQuantizationType.MXFP4)

        return consumed

    def _write_hash_routing_tensors(self) -> list[str]:
        consumed: list[str] = []

        for bid in range(self.hparams["num_hash_layers"]):
            name = f"layers.{bid}.ffn.gate.tid2eid"
            if name not in self.model_tensors:
                raise KeyError(f"Missing hash routing tensor {name}")

            data_torch = LazyTorchTensor.to_eager(self.model_tensors[name]())
            data = data_torch.to(torch.int32).cpu().numpy()
            new_name = self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE_TID2EID, bid, ".weight")
            logger.info(f"{new_name}: converted hash routing table to I32, shape = {{{', '.join(str(n) for n in reversed(data.shape))}}}")
            self.gguf_writer.add_tensor(new_name, data)
            consumed.append(name)

        return consumed

    def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:
        if self._dsv4_mxfp4_generated:
            return ()

        consumed: list[str] = []
        main_layers = self.hparams["num_hidden_layers"]
        if not self.mtp_only:
            consumed.extend(self._write_hash_routing_tensors())
        elif self.hparams["num_hash_layers"] > 0:
            for bid in range(self.hparams["num_hash_layers"]):
                name = f"layers.{bid}.ffn.gate.tid2eid"
                if name in self.model_tensors:
                    consumed.extend(self._write_hash_routing_tensors())
                    break

        for bid in range(self.block_count):
            if self.mtp_only and bid < main_layers:
                continue
            consumed.extend(self._write_mxfp4_expert_tensor(bid, "w1", gguf.MODEL_TENSOR.FFN_GATE_EXP))
            consumed.extend(self._write_mxfp4_expert_tensor(bid, "w2", gguf.MODEL_TENSOR.FFN_DOWN_EXP))
            consumed.extend(self._write_mxfp4_expert_tensor(bid, "w3", gguf.MODEL_TENSOR.FFN_UP_EXP))

        for bid in range(main_layers, self.block_count):
            e_name = f"layers.{bid}.nextn.e_proj.weight"
            h_name = f"layers.{bid}.nextn.h_proj.weight"
            if e_name not in self.model_tensors and h_name not in self.model_tensors:
                continue
            if e_name not in self.model_tensors or h_name not in self.model_tensors:
                raise KeyError(f"Missing DeepSeek-V4 MTP e/h projection pair for block {bid}")

            e_proj = LazyTorchTensor.to_eager(self.model_tensors[e_name]())
            h_proj = LazyTorchTensor.to_eager(self.model_tensors[h_name]())
            yield (f"layers.{bid}.nextn.eh_proj.weight", torch.cat((e_proj, h_proj), dim=1).contiguous())
            consumed.extend((e_name, h_name))

        for name in consumed:
            del self.model_tensors[name]

        self._dsv4_mxfp4_generated = True
        return ()

    def _map_dsv4_tensor_name(self, name: str, bid: int | None) -> tuple[gguf.MODEL_TENSOR, str]:
        root_map: dict[str, tuple[gguf.MODEL_TENSOR, str]] = {
            "embed.weight": (gguf.MODEL_TENSOR.TOKEN_EMBD, ".weight"),
            "norm.weight": (gguf.MODEL_TENSOR.OUTPUT_NORM, ".weight"),
            "head.weight": (gguf.MODEL_TENSOR.OUTPUT, ".weight"),
            "hc_head_fn": (gguf.MODEL_TENSOR.HC_HEAD_FN, ".weight"),
            "hc_head_base": (gguf.MODEL_TENSOR.HC_HEAD_BASE, ".weight"),
            "hc_head_scale": (gguf.MODEL_TENSOR.HC_HEAD_SCALE, ".weight"),
        }
        if name in root_map:
            return root_map[name]

        match = re.match(r"layers\.(\d+)\.(.+)$", name)
        if match is None:
            raise ValueError(f"Unsupported DeepSeek-V4 tensor {name!r}")

        layer = int(match.group(1))
        if bid != layer:
            raise ValueError(f"Tensor {name!r} parsed bid {bid} but layer name has {layer}")

        layer_map: dict[str, tuple[gguf.MODEL_TENSOR, str]] = {
            "hc_attn_fn": (gguf.MODEL_TENSOR.HC_ATTN_FN, ".weight"),
            "hc_attn_base": (gguf.MODEL_TENSOR.HC_ATTN_BASE, ".weight"),
            "hc_attn_scale": (gguf.MODEL_TENSOR.HC_ATTN_SCALE, ".weight"),
            "hc_ffn_fn": (gguf.MODEL_TENSOR.HC_FFN_FN, ".weight"),
            "hc_ffn_base": (gguf.MODEL_TENSOR.HC_FFN_BASE, ".weight"),
            "hc_ffn_scale": (gguf.MODEL_TENSOR.HC_FFN_SCALE, ".weight"),
            "attn.attn_sink": (gguf.MODEL_TENSOR.ATTN_SINKS, ".weight"),
            "attn.wq_a.weight": (gguf.MODEL_TENSOR.ATTN_Q_A, ".weight"),
            "attn.wq_b.weight": (gguf.MODEL_TENSOR.ATTN_Q_B, ".weight"),
            "attn.q_norm.weight": (gguf.MODEL_TENSOR.ATTN_Q_A_NORM, ".weight"),
            "attn.wkv.weight": (gguf.MODEL_TENSOR.ATTN_KV, ".weight"),
            "attn.kv_norm.weight": (gguf.MODEL_TENSOR.ATTN_KV_NORM, ".weight"),
            "attn.wo_a.weight": (gguf.MODEL_TENSOR.ATTN_OUT_A, ".weight"),
            "attn.wo_b.weight": (gguf.MODEL_TENSOR.ATTN_OUT_B, ".weight"),
            "attn.compressor.ape": (gguf.MODEL_TENSOR.ATTN_COMPRESSOR_APE, ".weight"),
            "attn.compressor.wkv.weight": (gguf.MODEL_TENSOR.ATTN_COMPRESSOR_WKV, ".weight"),
            "attn.compressor.wgate.weight": (gguf.MODEL_TENSOR.ATTN_COMPRESSOR_WGATE, ".weight"),
            "attn.compressor.norm.weight": (gguf.MODEL_TENSOR.ATTN_COMPRESSOR_NORM, ".weight"),
            "attn.indexer.wq_b.weight": (gguf.MODEL_TENSOR.INDEXER_ATTN_Q_B, ".weight"),
            "attn.indexer.weights_proj.weight": (gguf.MODEL_TENSOR.INDEXER_PROJ, ".weight"),
            "attn.indexer.compressor.ape": (gguf.MODEL_TENSOR.INDEXER_COMPRESSOR_APE, ".weight"),
            "attn.indexer.compressor.wkv.weight": (gguf.MODEL_TENSOR.INDEXER_COMPRESSOR_WKV, ".weight"),
            "attn.indexer.compressor.wgate.weight": (gguf.MODEL_TENSOR.INDEXER_COMPRESSOR_WGATE, ".weight"),
            "attn.indexer.compressor.norm.weight": (gguf.MODEL_TENSOR.INDEXER_COMPRESSOR_NORM, ".weight"),
            "attn_norm.weight": (gguf.MODEL_TENSOR.ATTN_NORM, ".weight"),
            "ffn_norm.weight": (gguf.MODEL_TENSOR.FFN_NORM, ".weight"),
            "ffn.gate.weight": (gguf.MODEL_TENSOR.FFN_GATE_INP, ".weight"),
            "ffn.gate.bias": (gguf.MODEL_TENSOR.FFN_EXP_PROBS_B, ".bias"),
            "ffn.gate.bias_vl": (gguf.MODEL_TENSOR.FFN_EXP_PROBS_B_VL, ".bias"),
            "ffn.gate.tid2eid": (gguf.MODEL_TENSOR.FFN_GATE_TID2EID, ".weight"),
            "ffn.shared_experts.w1.weight": (gguf.MODEL_TENSOR.FFN_GATE_SHEXP, ".weight"),
            "ffn.shared_experts.w2.weight": (gguf.MODEL_TENSOR.FFN_DOWN_SHEXP, ".weight"),
            "ffn.shared_experts.w3.weight": (gguf.MODEL_TENSOR.FFN_UP_SHEXP, ".weight"),
            "nextn.eh_proj.weight": (gguf.MODEL_TENSOR.NEXTN_EH_PROJ, ".weight"),
            "nextn.enorm.weight": (gguf.MODEL_TENSOR.NEXTN_ENORM, ".weight"),
            "nextn.hnorm.weight": (gguf.MODEL_TENSOR.NEXTN_HNORM, ".weight"),
            "nextn.shared_head_norm.weight": (gguf.MODEL_TENSOR.NEXTN_SHARED_HEAD_NORM, ".weight"),
            "nextn.embed_tokens.weight": (gguf.MODEL_TENSOR.NEXTN_EMBED_TOKENS, ".weight"),
            "nextn.shared_head_head.weight": (gguf.MODEL_TENSOR.NEXTN_SHARED_HEAD_HEAD, ".weight"),
        }

        tensor_name = match.group(2)
        if tensor_name in layer_map:
            return layer_map[tensor_name]

        if re.match(r"ffn\.experts\.\d+\.w[123]\.(weight|scale)$", tensor_name):
            return gguf.MODEL_TENSOR.FFN_GATE_EXP, ".weight"

        raise ValueError(f"Unsupported DeepSeek-V4 tensor {name!r}")

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        if re.match(r"layers\.\d+\.ffn\.experts\.\d+\.w[123]\.(weight|scale)$", name):
            return

        # hash layers route text tokens via tid2eid and image tokens via bias_vl; gate.bias is unused
        if name.endswith(".ffn.gate.bias") and bid is not None and bid < self.hparams["num_hash_layers"]:
            return

        tensor_key, suffix = self._map_dsv4_tensor_name(name, bid)
        if tensor_key == gguf.MODEL_TENSOR.FFN_GATE_TID2EID:
            return

        yield self.format_tensor_name(tensor_key, bid, suffix), data_torch

    def tensor_force_quant(self, name: str, new_name: str, bid: int | None, n_dims: int) -> gguf.GGMLQuantizationType | bool:
        del bid  # unused

        if name in self._dsv4_fp8_dequantized and n_dims >= 2:
            return gguf.GGMLQuantizationType.Q8_0
        if new_name.endswith(".nextn.eh_proj.weight"):
            return gguf.GGMLQuantizationType.Q8_0
        if name in self._dsv4_f32_tensors:
            return gguf.GGMLQuantizationType.F32
        if name in self._dsv4_bf16_tensors and n_dims >= 2:
            return gguf.GGMLQuantizationType.BF16

        return False

    def prepare_metadata(self, vocab_only: bool):
        from_dir = self.fname_out.is_dir()
        super().prepare_metadata(vocab_only=vocab_only)

        if not self.mtp_only or not from_dir:
            return

        output_type: str = self.ftype.name.partition("_")[2]
        fname_default: str = gguf.naming_convention(
            self.metadata.name, self.metadata.basename, self.metadata.finetune,
            self.metadata.version, size_label=None, output_type=output_type, model_type=None)
        self.fname_out = self.fname_out.parent / f"mtp-{fname_default}.gguf"

    def prepare_tensors(self):
        super().prepare_tensors()
        self._is_mxfp4 = True
        self.ftype = gguf.LlamaFileType.MOSTLY_MXFP4_MOE


@ModelBase.register("DeepseekV4DSparkModel")
@ModelBase.example("deepseek-ai/DeepSeek-V4-Flash-DSpark")
class DeepseekV4DSparkModel(DeepseekV4Model):
    model_arch = gguf.MODEL_ARCH.DFLASH

    _DSPARK_ROOT_MAP: dict[str, tuple[gguf.MODEL_TENSOR, str]] = {
        "main_proj.weight": (gguf.MODEL_TENSOR.FC, ".weight"),
        "main_norm.weight": (gguf.MODEL_TENSOR.ENC_OUTPUT_NORM, ".weight"),
        "markov_head.markov_w1.weight": (gguf.MODEL_TENSOR.DSPARK_MARKOV_W1, ".weight"),
        "markov_head.markov_w2.weight": (gguf.MODEL_TENSOR.DSPARK_MARKOV_W2, ".weight"),
        "confidence_head.proj.weight": (gguf.MODEL_TENSOR.DSPARK_CONF_PROJ, ".weight"),
    }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.block_count = 1 + max(
            int(match.group(1)) for name in self.model_tensors
            if (match := re.match(r"layers\.(\d+)\.", name))
        )
        self.tensor_map = gguf.get_tensor_name_map(self.model_arch, self.block_count)

        self.hparams["compress_ratios"] = [0] * self.block_count
        self.hparams["num_hash_layers"] = 0

    def index_tensors(self, remote_hf_model_id: str | None = None) -> dict[str, Callable[[], Tensor]]:
        if remote_hf_model_id is None:
            return super().index_tensors()

        with open(self.dir_model / "model.safetensors.index.json", "r", encoding="utf-8") as f:
            weight_map = json.load(f)["weight_map"]

        part_names = sorted({
            part_name for name, part_name in weight_map.items()
            if name.startswith("mtp.")
        })
        tensors: dict[str, Callable[[], Tensor]] = {}

        for part_name in part_names:
            from huggingface_hub import hf_hub_download

            logger.info("gguf: caching remote DSpark part '%s'", part_name)
            part_path = Path(hf_hub_download(repo_id=remote_hf_model_id, filename=part_name))
            with gguf.utility.SafetensorsLocal(part_path) as model_part:
                for name in model_part:
                    data = model_part[name]
                    data_gen = lambda data=data: LazyTorchTensor.from_local_tensor(data)  # noqa: E731
                    if titem := self.filter_tensors((name, data_gen)):
                        tensor_name, tensor_gen = titem
                        tensors[tensor_name] = tensor_gen

        return tensors

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        name, gen = item
        if not name.startswith("mtp."):
            return None
        return super().filter_tensors((cls._rekey_mtp_tensor_name(name), gen))

    @staticmethod
    def _rekey_mtp_tensor_name(name: str) -> str:
        match = re.match(r"mtp\.(\d+)\.(.+)$", name)
        if match is None:
            raise ValueError(f"Unexpected DSpark tensor {name!r}")

        stage, rest = match.group(1), match.group(2)
        root_names = (
            "main_proj.scale",
            "norm.weight",
            "hc_head_fn",
            "hc_head_base",
            "hc_head_scale",
        )
        if rest in DeepseekV4DSparkModel._DSPARK_ROOT_MAP or rest in root_names:
            return rest
        return f"layers.{stage}.{rest}"

    def _map_dsv4_tensor_name(self, name: str, bid: int | None) -> tuple[gguf.MODEL_TENSOR, str]:
        if name in self._DSPARK_ROOT_MAP:
            return self._DSPARK_ROOT_MAP[name]
        return super()._map_dsv4_tensor_name(name, bid)

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        # the DFlash draft uses the plain exp-probs bias (ffn.gate.bias -> FFN_EXP_PROBS_B);
        # the mtmd-only hash routing tensors (bias_vl, tid2eid) are not part of the DFLASH arch
        if name.endswith(".ffn.gate.bias_vl"):
            return
        yield from super().modify_tensors(data_torch, name, bid)

    def set_vocab(self):
        if self.target_model_dir is None:
            raise ValueError("DeepSeek-V4 DSpark requires --target-model-dir with the target tokenizer")

        original_dir = self.dir_model
        try:
            self.dir_model = self.target_model_dir
            super().set_vocab()
        finally:
            self.dir_model = original_dir

        self.gguf_writer.add_mask_token_id(self.hparams["dspark_noise_token_id"])

    def set_gguf_parameters(self):
        super().set_gguf_parameters()

        self.gguf_writer.add_block_size(self.hparams["dspark_block_size"])
        self.gguf_writer.add_target_layers([layer + 1 for layer in self.hparams["dspark_target_layer_ids"]])


@ModelBase.register("DeepseekV4ForCausalLM")
@ModelBase.example("deepseek-ai/DeepSeek-V4-Flash-Vision-Exp")
class DeepseekV4FlashVisionModel(MmprojModel):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert self.hparams_vision is not None
        # no preprocessor_config.json in the repo; normalization is (x/255 - 0.5) / 0.5
        # ref: inference/image_processor.py (load_image)
        self.preprocessor_config = {
            "image_mean": [0.5, 0.5, 0.5],
            "image_std":  [0.5, 0.5, 0.5],
            **self.preprocessor_config,
        }

    def get_vision_config(self) -> dict[str, Any] | None:
        cfg = self.global_config
        if cfg.get("vision_n_layers", 0) == 0:
            raise ValueError("DeepseekV4FlashVisionModel requires vision_n_layers > 0 in the model config")
        return {
            "num_hidden_layers":   cfg["vision_n_layers"],
            "hidden_size":         cfg["vision_dim"],
            "num_attention_heads": cfg["vision_n_heads"],
            "intermediate_size":   cfg["vision_inter_dim"],
            "patch_size":          cfg["vision_patch_size"],
            # dynamic resolution; only used for compat / warmup
            "image_size":          cfg["vision_patch_size"] * cfg["vision_downsample_ratio"] * 16,
            "rope_theta":          cfg.get("vision_rope_theta", 10000.0),
            "downsample_ratio":    cfg["vision_downsample_ratio"],
            "min_pixels":          cfg["vision_min_pixels"],
        }

    def set_gguf_parameters(self):
        super().set_gguf_parameters()
        assert self.hparams_vision is not None
        self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.DEEPSEEK4V)
        # vision RMSNorm eps is the pytorch default, NOT the LLM's rms_norm_eps (1e-20)
        # ref: inference/vision.py (RMSNorm)
        self.gguf_writer.add_vision_attention_layernorm_eps(1e-6)
        self.gguf_writer.add_vision_use_silu(True) # SwiGLU MLP
        self.gguf_writer.add_vision_projector_scale_factor(self.hparams_vision["downsample_ratio"])
        self.gguf_writer.add_vision_min_pixels(self.hparams_vision["min_pixels"])
        # hardcoded on the C++ side (see PROJECTOR_TYPE_DEEPSEEK4V in clip.cpp)
        # if future models use different values, add GGUF keys for those
        assert self.global_config["vision_max_n_token"] == 384
        assert self.global_config["vision_max_wh_ratio"] == 8

    @classmethod
    def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
        name, _ = item
        if not (name.startswith(("vision.", "aligner.", "image_"))):
            return None
        return super().filter_tensors(item)

    def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
        assert self.hparams_vision is not None
        if name == "vision.patch_embed.proj.weight":
            # nn.Linear over flattened (3, p, p) patches == conv2d weight
            p = self.hparams_vision["patch_size"]
            data_torch = data_torch.reshape(data_torch.shape[0], 3, p, p)

        if ".mlp.w1." in name:
            # fused SwiGLU gate+up
            gate, up = data_torch.chunk(2, dim=0)
            yield from super().modify_tensors(gate, name.replace("w1", "w1_gate"), bid)
            yield from super().modify_tensors(up, name.replace("w1", "w1_up"), bid)
            return

        yield from super().modify_tensors(data_torch, name, bid)