elif quant_method == "compressed-tensors":
quant_format = quant_config["format"]
groups = quant_config["config_groups"]
- if len(groups) > 1:
+ nvfp4_compressed_tensors = (
+ quant_format == "nvfp4-pack-quantized"
+ or quant_format == "mixed-precision"
+ and bool(groups)
+ and all(g.get("format") == "nvfp4-pack-quantized" for g in groups.values() if isinstance(g, dict))
+ )
+
+ if len(groups) > 1 and not nvfp4_compressed_tensors:
raise NotImplementedError("Can't handle multiple config groups for compressed-tensors yet")
weight_config = tuple(groups.values())[0]["weights"]
tensors_to_remove += [base_name + n for n in ("_packed", "_shape", "_scale")]
if (base_name + "_zero_point") in self.model_tensors:
tensors_to_remove.append(base_name + "_zero_point")
+ elif nvfp4_compressed_tensors:
+ # Don't error from compressed-tensors, we'll handle them in _generate_nvfp4_tensors
+ pass
else:
raise NotImplementedError(f"Quant format {quant_format!r} for method {quant_method!r} is not yet supported")
elif quant_method == "modelopt":
del experts, merged
def prepare_tensors(self):
- # detect NVFP4 quantization (ModelOpt format)
- quant_algo = (self.hparams.get("quantization_config") or {}).get("quant_algo")
- quant_method = (self.hparams.get("quantization_config") or {}).get("quant_method")
- quant_layers = (self.hparams.get("quantization_config") or {}).get("quantized_layers") or {}
+ # detect NVFP4 quantization (ModelOpt and Compressed-tensors formats)
+ quantization_config = self.hparams.get("quantization_config") or {}
+ quant_algo = quantization_config.get("quant_algo")
+ quant_method = quantization_config.get("quant_method")
+ quant_format = quantization_config.get("format")
+ quant_groups = quantization_config.get("config_groups") or {}
+ quant_layers = quantization_config.get("quantized_layers") or {}
quant_config_file = self.dir_model / "hf_quant_config.json"
if (not quant_algo or not quant_layers) and quant_config_file.is_file():
producer_name = (producer.get("name") or "").lower()
if quant_method is None:
self.hparams.setdefault("quantization_config", {})["quant_method"] = producer_name
+ quant_method = producer_name
quant_algo = quant_config.get("quant_algo", quant_algo)
+ quant_method = quant_config.get("quant_method", quant_method)
+ quant_format = quant_config.get("format", quant_format)
+ quant_groups = quant_config.get("config_groups", quant_groups) or {}
quant_layers = quant_config.get("quantized_layers", quant_layers) or {}
# Some models use per-tensor quant_algo (e.g. "MIXED_PRECISION" with
# per-layer NVFP4/FP8) instead of a single global "NVFP4" value.
+ nvfp4_compressed_tensors = quant_method == "compressed-tensors" and (
+ quant_format == "nvfp4-pack-quantized"
+ or quant_format == "mixed-precision"
+ and bool(quant_groups)
+ and all(g.get("format") == "nvfp4-pack-quantized" for g in quant_groups.values() if isinstance(g, dict))
+ )
if quant_algo != "NVFP4":
- if any(v.get("quant_algo") == "NVFP4" for v in quant_layers.values() if isinstance(v, dict)):
+ if nvfp4_compressed_tensors:
+ quant_algo = "NVFP4"
+ elif any(v.get("quant_algo") == "NVFP4" for v in quant_layers.values() if isinstance(v, dict)):
quant_algo = "NVFP4"
self._is_nvfp4 = quant_algo == "NVFP4"
# This must run before dequant_model so NVFP4 tensors are removed
# from model_tensors, leaving only non-NVFP4 (e.g. FP8) for dequant.
if self._is_nvfp4:
+ if nvfp4_compressed_tensors:
+ # Convert compressed-tensors 'global' scales into the reciprocal
+ def inverse_scale(gen):
+ def load():
+ scale = LazyTorchTensor.to_eager(gen()).float()
+ return 1.0 / scale
+ return load
+
+ # Change the compressed-tensors names to the ModelOpt names for handling consistently later
+ for name in list(self.model_tensors.keys()):
+ if name.endswith(".weight_packed"):
+ weight_name = name.removesuffix("_packed")
+ if weight_name not in self.model_tensors:
+ self.model_tensors[weight_name] = self.model_tensors.pop(name)
+ elif name.endswith(".weight_global_scale"):
+ scale2_name = name.replace(".weight_global_scale", ".weight_scale_2")
+ if scale2_name not in self.model_tensors:
+ self.model_tensors[scale2_name] = inverse_scale(self.model_tensors.pop(name))
+ elif name.endswith(".input_global_scale"):
+ input_scale_name = name.replace(".input_global_scale", ".input_scale")
+ if input_scale_name not in self.model_tensors:
+ self.model_tensors[input_scale_name] = inverse_scale(self.model_tensors.pop(name))
self._generate_nvfp4_tensors()
self.dequant_model()
from __future__ import annotations
-from pathlib import Path
from typing import Any, Callable, Iterable, TYPE_CHECKING
import torch
tensor_map: gguf.TensorNameMap
no_mtp: bool
mtp_only: bool
+ _original_block_count: int | None = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.block_count += self.hparams.get("mtp_num_hidden_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) -> dict[str, Callable[[], Tensor]]:
+ hparams = {**self.hparams, **self.hparams.get("text_config", {})}
+ key = next((k for k in ["n_layers", "num_hidden_layers", "n_layer", "num_layers"] if k in hparams), None)
+ type(self)._original_block_count = hparams.get(key)
+ return super().index_tensors(remote_hf_model_id=remote_hf_model_id) # ty: ignore[unresolved-attribute]
+
@classmethod
def filter_tensors(cls, item):
- name, _ = item
+ assert cls._original_block_count is not None
+ # TODO: change TextModel to super()
+ if (titem := TextModel.filter_tensors(item)) is None:
+ return None
+ name, gen = titem
+ if name.startswith("model.mtp."):
+ name = name.replace("model.", "", 1)
if name.startswith("mtp."):
if cls.no_mtp:
return None
- return item
- if cls.mtp_only:
- canonical = name.replace("language_model.", "")
- keep = canonical in (
+ remapper = {
+ "fc": "eh_proj",
+ "pre_fc_norm_embedding": "enorm",
+ "pre_fc_norm_hidden": "hnorm",
+ "norm": "shared_head.norm",
+ }
+ parts = name.split(".", 3)
+ if len(parts) == 4 and parts[1] == "layers" and parts[2].isdecimal():
+ mtp_idx = int(parts[2])
+ name = f"model.layers.{cls._original_block_count + mtp_idx}.{parts[3]}"
+ elif len(parts) == 3 and parts[1] in remapper:
+ name = f"model.layers.{cls._original_block_count}.{remapper[parts[1]]}.{parts[2]}"
+ elif cls.mtp_only:
+ keep = name in (
"model.embed_tokens.weight", "model.norm.weight", "lm_head.weight",
"embed_tokens.weight", "norm.weight",
)
if not keep:
return None
- return super().filter_tensors(item) # ty: ignore[unresolved-attribute]
+ return name, gen
def set_gguf_parameters(self):
super().set_gguf_parameters() # ty: ignore[unresolved-attribute]
self.metadata.version, size_label=None, output_type=output_type, model_type=None) # pyright: ignore[reportAttributeAccessIssue] # ty: ignore[unresolved-attribute]
self.fname_out = self.fname_out.parent / f"mtp-{fname_default}.gguf"
- def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
- if name.startswith("mtp."):
- n_layer = self.hparams["num_hidden_layers"]
- if name.find("layers.") != -1:
- assert bid is not None
- name = name.replace(f"mtp.layers.{bid}", f"model.layers.{bid + n_layer}")
- bid = bid + n_layer
- else:
- remapper = {
- "mtp.fc": "model.layers.{bid}.eh_proj",
- "mtp.pre_fc_norm_embedding": "model.layers.{bid}.enorm",
- "mtp.pre_fc_norm_hidden": "model.layers.{bid}.hnorm",
- "mtp.norm": "model.layers.{bid}.shared_head.norm",
- }
- stem = Path(name).stem
- suffix = Path(name).suffix
- tmpl = remapper[stem] + suffix
- for b in range(n_layer, self.block_count):
- yield from super().modify_tensors(data_torch, tmpl.format(bid=b), b) # ty: ignore[unresolved-attribute]
- return
-
- yield from super().modify_tensors(data_torch, name, bid) # ty: ignore[unresolved-attribute]
-
@ModelBase.register("Qwen3_5ForConditionalGeneration", "Qwen3_5ForCausalLM")
class Qwen3_5TextModel(_Qwen35MtpMixin, _Qwen35MRopeMixin, _LinearAttentionVReorderBase):
overview / pkg / ggml / sources / llama.cpp / commitdiff