from __future__ import annotations
import re
from typing import Any, Callable, Iterable, TYPE_CHECKING
import torch
if TYPE_CHECKING:
from torch import Tensor
from .base import MmprojModel, ModelBase, gguf, logger
from .llama import LlamaModel
from .mamba import Mamba2Model
@ModelBase.register("GraniteForCausalLM")
@ModelBase.example("ibm-granite/granite-3.3-2b-instruct")
class GraniteModel(LlamaModel):
"""Conversion for IBM's GraniteForCausalLM"""
model_arch = gguf.MODEL_ARCH.GRANITE
def set_gguf_parameters(self):
"""Granite uses standard llama parameters with the following differences:
- No head_dim support
- New multiplier params:
- attention_scale
- embedding_scale
- residual_scale
- logits_scaling
"""
if head_dim := self.hparams.pop("head_dim", None):
logger.warning("Ignoring head_dim (%s) from config for Granite", head_dim)
super().set_gguf_parameters()
# NOTE: Convert _multiplier params to _scale params for naming
# consistency
if attention_scale := self.hparams.get("attention_multiplier"):
self.gguf_writer.add_attention_scale(attention_scale)
logger.info("gguf: (granite) attention_scale = %s", attention_scale)
if embedding_scale := self.hparams.get("embedding_multiplier"):
self.gguf_writer.add_embedding_scale(embedding_scale)
logger.info("gguf: (granite) embedding_scale = %s", embedding_scale)
if residual_scale := self.hparams.get("residual_multiplier"):
self.gguf_writer.add_residual_scale(residual_scale)
logger.info("gguf: (granite) residual_scale = %s", residual_scale)
if logits_scale := self.hparams.get("logits_scaling"):
self.gguf_writer.add_logit_scale(logits_scale)
logger.info("gguf: (granite) logits_scale = %s", logits_scale)
# If being used as the base for Granite4 Vision, add deepstack_layer_arr
if self.hparams.get("spatial_target_layers") or self.hparams.get("deepstack_layer_map"):
normalized_projector_map = Granite4VisionMmprojModel.get_normalized_projector_map(self.hparams)
deepstack_mapping_arr = [-1 for _ in range(self.block_count)] # Populate with -1 sentinels
for proj_idx, (_, llm_layer, _, _) in enumerate(normalized_projector_map):
# Skip the first projector which is handled as the base embedding
# stream like normal
if proj_idx == 0:
continue
deepstack_mapping_arr[llm_layer] = proj_idx
self.gguf_writer.add_deepstack_mapping(deepstack_mapping_arr)
@classmethod
def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
name, gen = item
# Skip multimodal tensors
if (
name.startswith(("encoder."))
or "image_" in name
or "layerwise_projectors" in name
or "spatial_projectors" in name
):
return
return super().filter_tensors(item)
@ModelBase.register("GraniteSWAForCausalLM")
class GraniteSWAModel(GraniteModel):
"""Conversion for IBM's GraniteSWAForCausalLM (interleaved sliding window attention)"""
model_arch = gguf.MODEL_ARCH.GRANITE_SWA
@classmethod
def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
name, gen = item
if name.endswith("sinks"):
name += ".weight"
return super().filter_tensors((name, gen))
def set_gguf_parameters(self):
"""GraniteSWA uses Granite parameters plus sliding window configuration."""
super().set_gguf_parameters()
# Add sliding_window from config
sliding_window = self.hparams.get("sliding_window", 128)
self.gguf_writer.add_sliding_window(sliding_window)
logger.info("gguf: (granite_swa) sliding_window = %s", sliding_window)
# Derive sliding_window_pattern from layer_types
if layer_types := self.hparams.get("layer_types"):
is_swa = [t == "sliding_attention" for t in layer_types]
self.gguf_writer.add_sliding_window_pattern(is_swa)
logger.info("gguf: (granite_swa) sliding_window_pattern = %d SWA layers / %d total",
sum(is_swa), len(is_swa))
else:
# Fall back to period-based pattern: i % 4 != 0
# This matches the transformers default pattern
n_layers = self.block_count
is_swa = [i % 4 != 0 for i in range(n_layers)]
self.gguf_writer.add_sliding_window_pattern(is_swa)
logger.info("gguf: (granite_swa) sliding_window_pattern (inferred) = %d SWA layers / %d total",
sum(is_swa), n_layers)
# Add rope_pattern from no_rope_layers
if no_rope_layers := self.hparams.get("no_rope_layers"):
# Convert 1/0 to bool (1 = use RoPE, 0 = NoPE)
rope_pattern = [bool(x) for x in no_rope_layers]
self.gguf_writer.add_rope_pattern(rope_pattern)
logger.info("gguf: (granite_swa) rope_pattern = %d RoPE layers / %d total",
sum(rope_pattern), len(rope_pattern))
@ModelBase.register("GraniteMoeSWAForCausalLM")
class GraniteMoeSWAModel(GraniteSWAModel):
"""Conversion for IBM's GraniteMoeSWAForCausalLM (unified dense + MoE with iSWA)"""
model_arch = gguf.MODEL_ARCH.GRANITE_SWA
def set_gguf_parameters(self):
super().set_gguf_parameters()
if shared_intermediate_size := self.hparams.get("shared_intermediate_size"):
self.gguf_writer.add_expert_shared_feed_forward_length(shared_intermediate_size)
logger.info("gguf: (granitemoewa) shared_intermediate_size = %s", shared_intermediate_size)
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
"""Split merged MoE tensors (gate+up) following standard MoE pattern."""
# Handle expert FFN tensors (merged gate+up) - swash format: experts.gate_up_proj
# Kept fused since inference (build_moe_ffn) supports a single gate_up_exps
# tensor for the routed experts.
if name.endswith("block_sparse_moe.experts.gate_up_proj"):
ffn_dim = self.hparams["intermediate_size"]
assert data_torch.shape[-2] == 2 * ffn_dim, f"Merged FFN tensor size must be 2 * intermediate_size, got {data_torch.shape[-2]}"
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE_UP_EXP, bid), bid)
return
# Handle expert FFN down projection - swash format: experts.down_proj
if name.endswith("block_sparse_moe.experts.down_proj"):
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_DOWN_EXP, bid), bid)
return
# Handle expert FFN tensors (merged gate+up) - standard granite format: input_linear.weight
# Kept fused since inference (build_moe_ffn) supports a single gate_up_exps
# tensor for the routed experts.
if name.endswith("block_sparse_moe.input_linear.weight"):
ffn_dim = self.hparams["intermediate_size"]
assert data_torch.shape[-2] == 2 * ffn_dim, "Merged FFN tensor size must be 2 * intermediate_size"
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE_UP_EXP, bid), bid)
return
# Handle shared expert FFN tensors (if present) - kept fused since
# inference (build_ffn) supports a single ffn_up_shexp tensor with
# LLM_FFN_SWIGLU for the shared expert.
if name.endswith("shared_mlp.input_linear.weight"):
ffn_dim = self.hparams.get("shared_intermediate_size", self.hparams["intermediate_size"])
assert data_torch.shape[-2] == 2 * ffn_dim, "Merged FFN tensor size must be 2 * shared_intermediate_size"
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_UP_SHEXP, bid), bid)
return
# Handle shared expert output (if present)
if name.endswith("shared_mlp.output_linear.weight"):
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_DOWN_SHEXP, bid), bid)
return
# Pass through to parent for all other tensors (including sinks)
yield from super().modify_tensors(data_torch, name, bid)
@ModelBase.register("GraniteMoeForCausalLM", "GraniteMoeSharedForCausalLM")
@ModelBase.example("ibm-granite/granite-3.1-3b-a800m-instruct")
class GraniteMoeModel(GraniteModel):
"""Conversion for IBM's GraniteMoeForCausalLM"""
model_arch = gguf.MODEL_ARCH.GRANITE_MOE
def set_gguf_parameters(self):
"""GraniteMoeShared uses GraniteMoe parameters plus the following:
- shared_intermediate_size
"""
super().set_gguf_parameters()
if shared_feed_forward_length := self.hparams.get("shared_intermediate_size"):
self.gguf_writer.add_expert_shared_feed_forward_length(shared_feed_forward_length)
logger.info("gguf: (granitemoeshared) shared_feed_forward_length = %s", shared_feed_forward_length)
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
"""In modeling_granitemoe, the JetMoe implementation of parallel experts
is used. This essentially merges w1 and w3 into a single tensor with 2x
the hidden size that is then split during forward. To keep compatibility
with existing mixtral support, we pull them apart here.
"""
if name.endswith("block_sparse_moe.input_linear.weight"):
ffn_dim = self.hparams["intermediate_size"]
assert data_torch.shape[-2] == 2 * ffn_dim, "Merged FFN tensor size must be 2 * intermediate_size"
gate, up = data_torch.split(ffn_dim, dim=-2)
yield from ModelBase.modify_tensors(self, gate, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE_EXP, bid), bid)
yield from ModelBase.modify_tensors(self, up, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_UP_EXP, bid), bid)
return
has_experts = bool(self.hparams.get('num_local_experts'))
if name.endswith("shared_mlp.input_linear.weight"):
ffn_dim = self.hparams["shared_intermediate_size"]
assert data_torch.shape[-2] == 2 * ffn_dim, "Merged FFN tensor size must be 2 * shared_intermediate_size"
gate, up = data_torch.split(ffn_dim, dim=-2)
if has_experts:
yield from ModelBase.modify_tensors(self, gate,self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE_SHEXP, bid), bid)
yield from ModelBase.modify_tensors(self, up, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_UP_SHEXP, bid), bid)
return
yield from ModelBase.modify_tensors(self, gate, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_GATE, bid), bid)
yield from ModelBase.modify_tensors(self, up, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_UP, bid), bid)
return
if not has_experts and name.endswith("shared_mlp.output_linear.weight"):
yield from ModelBase.modify_tensors(self, data_torch, self.format_tensor_name(gguf.MODEL_TENSOR.FFN_DOWN, bid), bid)
return
yield from super().modify_tensors(data_torch, name, bid)
@ModelBase.register("GraniteSwitchForCausalLM")
@ModelBase.example("ibm-granite/granite-switch-4.1-3b-preview")
class GraniteSwitchModel(GraniteMoeModel):
"""Dense, all-attention Granite with N per-token embedded LoRA adapters, stacked
over the adapter dim with a zero adapter at slot 0 (N = num_adapters + 1)."""
model_arch = gguf.MODEL_ARCH.GRANITE_SWITCH
# permute q/k per-slice below (NORM-rope layout), not via the parent's auto-permute
undo_permute = False
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# the weightless switch reserves one cache slot: one fewer block than num_hidden_layers
self.block_count = self.block_count - 1
self.tensor_map = gguf.get_tensor_name_map(self.model_arch, self.block_count)
self._n_adapters = int(self.hparams["num_adapters"])
self._max_lora_rank = int(self.hparams["max_lora_rank"])
self._n_slots = self._n_adapters + 1 # +1 for the zero slot at index 0
n_head = int(self.hparams["num_attention_heads"])
n_kv_head = int(self.hparams["num_key_value_heads"])
head_dim = (
self.hparams.get("projection_head_dim")
or self.hparams.get("head_dim")
or (self.hparams["hidden_size"] // n_head)
)
self._n_head = n_head
self._n_kv_head = n_kv_head
self._head_dim = int(head_dim)
self._q_size = n_head * self._head_dim
self._kv_size = n_kv_head * self._head_dim
def set_gguf_parameters(self):
super().set_gguf_parameters()
# dense: pin expert_used_count to 0 (config carries a leftover num_experts_per_tok)
if not self.hparams.get("num_local_experts"):
self.gguf_writer.add_expert_used_count(0)
self.gguf_writer.add_adapter_count(self._n_adapters)
self.gguf_writer.add_adapter_lora_rank(self._max_lora_rank)
self.gguf_writer.add_adapter_token_ids_activate(self.hparams["adapter_token_ids"])
self.gguf_writer.add_adapter_token_ids_substitute(self.hparams["adapter_substitute_token_ids"])
router_gain = float(self.hparams.get("control_token_gain", 15.0))
self.gguf_writer.add_adapter_router_gain(router_gain)
logger.info("gguf: (graniteswitch) num_adapters=%s max_lora_rank=%s n_slots=%s router_gain=%s", self._n_adapters, self._max_lora_rank, self._n_slots, router_gain)
def _lora_a(self, data: Tensor) -> Tensor:
# on-disk A: [n_adapters, 1, max_rank, in] -> [n_adapters+1, max_rank, in]
a = data.squeeze(1)
zero = torch.zeros_like(a[:1])
return torch.cat([zero, a], dim=0).contiguous()
def _lora_b(self, data: Tensor, permute_n_head: int | None = None) -> Tensor:
# on-disk B: [n_adapters, 1, out, max_rank] -> [n_adapters+1, out, max_rank]
b = data.squeeze(1)
if permute_n_head is not None:
# permute each adapter's B output rows to match the permuted q/k base
b = torch.stack([self.permute(b[i], permute_n_head, permute_n_head) for i in range(b.shape[0])], dim=0)
zero = torch.zeros_like(b[:1])
return torch.cat([zero, b], dim=0).contiguous()
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
T = gguf.MODEL_TENSOR
# skip the weightless switch + control-token buffers (rebuilt at load time)
bare = name.split(".")[-1]
if (
name.startswith("model.switch.") or name.startswith("switch.")
or bare in ("adapter_token_ids", "control_to_substitute_lut")
):
return
if "self_attn.qkv_proj" in name:
if name.endswith("base_layer.weight"):
# fused [q|k|v] rows: permute q/k row-blocks for ggml's NORM-rope layout
q, k, v = data_torch.split([self._q_size, self._kv_size, self._kv_size], dim=0)
q = self.permute(q, self._n_head, self._n_head)
k = self.permute(k, self._n_kv_head, self._n_kv_head)
fused = torch.cat([q, k, v], dim=0)
yield (self.format_tensor_name(T.ATTN_QKV, bid), fused)
return
if "lora_A_slices." in name:
slot = int(name.rsplit(".", 1)[1])
key = {0: T.ATTN_Q, 1: T.ATTN_K, 2: T.ATTN_V}[slot]
yield (self.format_tensor_name(key, bid, suffix=".lora_a"), self._lora_a(data_torch))
return
if "lora_B_slices." in name:
slot = int(name.rsplit(".", 1)[1])
key, ph = {
0: (T.ATTN_Q, self._n_head),
1: (T.ATTN_K, self._n_kv_head),
2: (T.ATTN_V, None),
}[slot]
yield (self.format_tensor_name(key, bid, suffix=".lora_b"), self._lora_b(data_torch, ph))
return
raise ValueError(f"Unexpected qkv_proj tensor: {name}")
if "self_attn.o_proj" in name:
if name.endswith("base_layer.weight"):
yield (self.format_tensor_name(T.ATTN_OUT, bid), data_torch)
return
if name.endswith("lora_A"):
yield (self.format_tensor_name(T.ATTN_OUT, bid, suffix=".lora_a"), self._lora_a(data_torch))
return
if name.endswith("lora_B"):
yield (self.format_tensor_name(T.ATTN_OUT, bid, suffix=".lora_b"), self._lora_b(data_torch))
return
raise ValueError(f"Unexpected o_proj tensor: {name}")
if "shared_mlp.input_linear" in name:
ffn = self.hparams["shared_intermediate_size"]
if name.endswith("base_layer.weight"):
gate, up = data_torch.split([ffn, ffn], dim=0)
yield (self.format_tensor_name(T.FFN_GATE, bid), gate)
yield (self.format_tensor_name(T.FFN_UP, bid), up)
return
if "lora_A_slices." in name:
slot = int(name.rsplit(".", 1)[1])
key = {0: T.FFN_GATE, 1: T.FFN_UP}[slot]
yield (self.format_tensor_name(key, bid, suffix=".lora_a"), self._lora_a(data_torch))
return
if "lora_B_slices." in name:
slot = int(name.rsplit(".", 1)[1])
key = {0: T.FFN_GATE, 1: T.FFN_UP}[slot]
yield (self.format_tensor_name(key, bid, suffix=".lora_b"), self._lora_b(data_torch))
return
raise ValueError(f"Unexpected shared_mlp.input_linear tensor: {name}")
if "shared_mlp.output_linear" in name:
if name.endswith("base_layer.weight"):
yield (self.format_tensor_name(T.FFN_DOWN, bid), data_torch)
return
if name.endswith("lora_A"):
yield (self.format_tensor_name(T.FFN_DOWN, bid, suffix=".lora_a"), self._lora_a(data_torch))
return
if name.endswith("lora_B"):
yield (self.format_tensor_name(T.FFN_DOWN, bid, suffix=".lora_b"), self._lora_b(data_torch))
return
raise ValueError(f"Unexpected shared_mlp.output_linear tensor: {name}")
if bid is not None and ".layers." in name and (
"input_layernorm" in name or "post_attention_layernorm" in name
):
key = T.ATTN_NORM if "input_layernorm" in name else T.FFN_NORM
yield (self.format_tensor_name(key, bid), data_torch)
return
if name in ("model.embed_tokens.weight", "embed_tokens.weight"):
yield (self.format_tensor_name(T.TOKEN_EMBD), data_torch)
return
if name in ("model.norm.weight", "norm.weight"):
yield (self.format_tensor_name(T.OUTPUT_NORM), data_torch)
return
if name == "lm_head.weight":
return # tied to token_embd
raise ValueError(f"graniteswitch: unhandled tensor {name!r} (bid={bid})")
@ModelBase.register("GraniteMoeHybridForCausalLM", "BambaForCausalLM")
@ModelBase.example("ibm-granite/granite-4.0-h-tiny", "ibm-ai-platform/Bamba-9B-v2")
class GraniteHybridModel(Mamba2Model, GraniteMoeModel):
"""GraniteHybrid is a hybrid SSM + Attention model that uses Mamba2 SSM
layers and optionally uses MoE w/ a shared expert"""
model_arch = gguf.MODEL_ARCH.GRANITE_HYBRID
undo_permute = True
def __init__(self, *args, **kwargs):
# Hybrid mamba models use a prefix for the mamba-specific params.
# TODO: Extend this if the prefix(es) need to be configurable
self.hparam_prefixes = ["mamba"]
super().__init__(*args, **kwargs)
# Lists of which layers use ssm vs attention
self._attn_layers = self.get_attn_layers()
self._ssm_layers = [
i for i in range(self.block_count)
if i not in self._attn_layers
]
# There are some models in this family that are non-hybrid, but keep the
# same parent class by setting all layers to "attention." If this is the
# case, the model architecture needs to be updated to a standard
# "granite" or "granitemoe" model
if not self._ssm_layers:
has_experts = self.find_hparam(["num_experts_per_tok", "num_experts_per_token"], optional=True)
new_arch = (
gguf.MODEL_ARCH.GRANITE_MOE
if has_experts else
gguf.MODEL_ARCH.GRANITE
)
self.model_arch = new_arch
self.gguf_writer.arch = gguf.MODEL_ARCH_NAMES[new_arch]
self.gguf_writer.add_architecture()
# n_group and d_inner are used during reshape_tensors for mamba2
# NOTE: Explicitly include hparam prefix prefix for d_model to
# disambiguate with top-level head_dim
# NOTE 2: If needed for future models, this can be isolated in a method
# to separate the prefix setting and the keys used
self.d_model = self.find_hparam([f"{self.hparam_prefixes[0]}_head_dim", "hidden_size", "d_model"])
self.n_group = self.find_hparam(["n_groups", "num_groups"])
self.d_inner = self.find_hparam(["expand", "num_heads"]) * self.d_model
def get_attn_layers(self):
# Explicit list of layer type names
if layer_types := self.hparams.get("layer_types"):
return [
i for i, typ in enumerate(layer_types)
if typ == "attention"
]
# Layer types indicated by index or period
attn_layers = self.hparams.get("attn_layer_indices", [])
if not attn_layers:
attn_period = self.hparams.get("attn_layer_period")
assert attn_period, "Didn't find attn_layer_indices or attn_layer_period"
attn_offset = self.hparams.get("attn_layer_offset")
assert attn_offset is not None, "No attention layer offset set with attn_layer_period"
attn_layers = [
i for i in range(self.block_count)
if i % attn_period == attn_offset
]
return attn_layers
def find_hparam(self, keys: Iterable[str], *args, **kwargs) -> Any:
prefixed = []
for pfx in self.hparam_prefixes:
prefixed.extend(
"_".join([pfx, k])
for k in keys
)
keys = list(keys) + prefixed
return Mamba2Model.find_hparam(self, keys, *args, **kwargs)
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
if (
name.endswith("block_sparse_moe.input_linear.weight")
or "shared_mlp" in name
):
yield from GraniteMoeModel.modify_tensors(self, data_torch, name, bid)
return
# Determine whether this is a mamba layer or an attention layer
if bid in self._ssm_layers:
yield from Mamba2Model.modify_tensors(self, data_torch, name, bid)
return
elif bid in self._attn_layers:
yield from GraniteMoeModel.modify_tensors(self, data_torch, name, bid)
return
yield from ModelBase.modify_tensors(self, data_torch, name, bid)
def set_gguf_parameters(self):
"""This method merges params from both parents and some that are
specific to this model. The result is some duplication of how the params
get set. The following warnings are expected during conversion:
WARNING:Duplicated key name 'granitehybrid.attention.head_count_kv'
WARNING:Duplicated key name 'granitehybrid.context_length'
"""
GraniteMoeModel.set_gguf_parameters(self)
## Mamba mixer params ##
self.gguf_writer.add_ssm_conv_kernel(self.find_hparam(["conv_kernel", "d_conv"]))
self.gguf_writer.add_ssm_state_size(self.find_hparam(["state_size", "d_state", "state_dim", "ssm_state_size"]))
self.gguf_writer.add_ssm_group_count(self.n_group)
self.gguf_writer.add_ssm_inner_size(self.d_inner)
# NOTE: The mamba_dt_rank is _not_ the right field for how this is used
# in llama.cpp
self.gguf_writer.add_ssm_time_step_rank(self.find_hparam(["n_heads", "num_heads"]))
## Attention params ##
head_count_kv = self.find_hparam(["num_key_value_heads", "n_head_kv"])
head_count_kv_vec = [
head_count_kv if i in self._attn_layers else 0 for i in range(self.block_count)
]
if rope_dim := self.hparams.get("attn_rotary_emb"):
self.gguf_writer.add_rope_dimension_count(rope_dim)
self.gguf_writer.add_head_count_kv(head_count_kv_vec)
## If Bamba or non-hybrid, use rope, otherwise don't
use_rope = (
"BambaForCausalLM" in self.hparams["architectures"]
or not self._ssm_layers
)
self.gguf_writer.add_rope_scaling_finetuned(use_rope)
if not use_rope:
self.gguf_writer.add_context_length(2**20)
## Validation ##
d_head = self.find_hparam(["d_head"], optional=True) or 64
assert self.hparams.get("hidden_act") in [None, "silu"], "Only SILU activation supported"
assert self.d_inner % d_head == 0, f"SSM inner size {self.d_inner} not a multiple of head dim {d_head}"
def set_vocab(self):
# For models with no ssm layers, don't pad for mamba2
self.hparams["pad_vocab_size_multiple"] = 8 if self._ssm_layers else 1
Mamba2Model.set_vocab(self)
@ModelBase.register("GraniteSpeechForConditionalGeneration")
@ModelBase.example("ibm-granite/granite-speech-3.3-2b", "ibm-granite/granite-4.0-1b-speech")
class GraniteSpeechMmprojModel(MmprojModel):
has_vision_encoder = False
has_audio_encoder = True
_batch_norm_tensors: list[dict[str, Tensor]] | None = None
def get_audio_config(self) -> dict[str, Any] | None:
return self.global_config.get("encoder_config")
def set_gguf_parameters(self):
assert self.hparams_audio is not None
a = self.hparams_audio
a["hidden_size"] = a["hidden_dim"]
a["intermediate_size"] = a["hidden_dim"] * a["feedforward_mult"]
a["num_attention_heads"] = a["num_heads"]
a["num_hidden_layers"] = a["num_layers"]
super().set_gguf_parameters()
self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.GRANITE_SPEECH)
self.gguf_writer.add_audio_num_mel_bins(a["input_dim"])
self.gguf_writer.add_audio_attention_layernorm_eps(1e-5)
self.gguf_writer.add_audio_chunk_size(a["context_size"])
self.gguf_writer.add_audio_conv_kernel_size(a["conv_kernel_size"])
self.gguf_writer.add_audio_max_pos_emb(a["max_pos_emb"])
p = self.global_config
self.gguf_writer.add_audio_projector_window_size(p["window_size"])
self.gguf_writer.add_audio_projector_downsample_rate(p["downsample_rate"])
self.gguf_writer.add_audio_projector_head_count(p["projector_config"]["num_attention_heads"])
def tensor_force_quant(self, name, new_name, bid, n_dims):
if "encoder" in name or "projector" in name:
if ".conv" in name and ".weight" in name:
return gguf.GGMLQuantizationType.F32
return super().tensor_force_quant(name, new_name, bid, n_dims)
@classmethod
def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
name, gen = item
if "attention_dists" in name or "num_batches_tracked" in name:
return None
return super().filter_tensors(item)
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
# fold running_mean, running_var and eps into weight and bias for batch_norm
if "batch_norm" in name and "encoder.layers." in name:
if self._batch_norm_tensors is None:
self._batch_norm_tensors = [{} for _ in range(self.block_count)]
assert bid is not None
self._batch_norm_tensors[bid][name] = data_torch
if len(self._batch_norm_tensors[bid]) < 4:
return
prefix = f"encoder.layers.{bid}.conv.batch_norm"
weight = self._batch_norm_tensors[bid][f"{prefix}.weight"]
bias = self._batch_norm_tensors[bid][f"{prefix}.bias"]
running_mean = self._batch_norm_tensors[bid][f"{prefix}.running_mean"]
running_var = self._batch_norm_tensors[bid][f"{prefix}.running_var"]
eps = 1e-5
a = weight / torch.sqrt(running_var + eps)
b = bias - running_mean * a
yield from super().modify_tensors(a, f"encoder.layers.{bid}.conv.batch_norm.weight", bid)
yield from super().modify_tensors(b, f"encoder.layers.{bid}.conv.batch_norm.bias", bid)
return
if ".attn.to_kv.weight" in name:
k_weight, v_weight = data_torch.chunk(2, dim=0)
yield from super().modify_tensors(k_weight, name.replace("to_kv", "to_k"), bid)
yield from super().modify_tensors(v_weight, name.replace("to_kv", "to_v"), bid)
return
if ("up_conv" in name or "down_conv" in name) and name.endswith(".weight"):
if data_torch.ndim == 3 and data_torch.shape[2] == 1:
data_torch = data_torch.squeeze(2)
if "depth_conv" in name and name.endswith(".weight"):
if data_torch.ndim == 3 and data_torch.shape[1] == 1:
data_torch = data_torch.squeeze(1)
yield from super().modify_tensors(data_torch, name, bid)
@ModelBase.register("GraniteSpeechPlusForConditionalGeneration")
@ModelBase.example("ibm-granite/granite-speech-4.1-2b-plus")
class GraniteSpeechPlusMmprojModel(GraniteSpeechMmprojModel):
"""Conversion for GraniteSpeechPlus - extends GraniteSpeech with feature layer concatenation"""
has_vision_encoder = False
has_audio_encoder = True
def set_gguf_parameters(self):
assert self.hparams_audio is not None
super().set_gguf_parameters()
# Add feature_layer if present in encoder config
if feature_layers := self.hparams_audio.get("cat_hidden_layers"):
self.gguf_writer.add_audio_feature_layers(feature_layers)
logger.info(f"gguf: audio feature_layers = {feature_layers}")
# Validate projector dimension matches concatenated encoder output
hidden_dim = self.hparams_audio["hidden_dim"]
expected_dim = hidden_dim * (len(feature_layers) + 1)
projector_dim = self.global_config["projector_config"]["encoder_hidden_size"]
if projector_dim != expected_dim:
raise ValueError(
f"Projector encoder_hidden_size ({projector_dim}) does not match "
f"expected concatenated dimension ({expected_dim}). "
f"Expected: hidden_dim ({hidden_dim}) * (len(feature_layers) + 1) = {expected_dim}"
)
@ModelBase.register("Granite4VisionForConditionalGeneration")
@ModelBase.example("ibm-granite/granite-4.0-3b-vision")
class Granite4VisionMmprojModel(MmprojModel):
has_vision_encoder = True
has_audio_encoder = False
@staticmethod
def get_normalized_projector_map(global_config: dict) -> list[tuple[int, int, str, int]]:
"""Normalize both deepstack and spatial projector maps to the form:
(vision_layer, llm_layer, <type>, type_index)
This is then used to populate the following mappings:
- vision_feature_layers (mmproj hparam): ordered list of all
vision_layer values where order corresponds with the order of the
stacked projector tensors
NOTE: Values may appear multiple times for spatial projectors
- tensor_prefix_map (mmproj tensors): mapping from tensor prefixes to
the index of the corresponding projector in the stacked tensors
- deepstack_layer_arr (llm hparam): per-text-layer array indicating
which input vision feature should be injected at that layer
(-1 if none)
Output: (vision_layer, llm_layer, <type>, type_index)
"""
deepstack_map = global_config.get("deepstack_layer_map", []) # [[vis_layer, llm_layer], ...]
spatial_layers = global_config.get("spatial_target_layers", []) # [llm_layer, ...]
n_text_layers = global_config["text_config"]["num_hidden_layers"]
n_vision_layers = global_config["vision_config"]["num_hidden_layers"]
normalized_projector_map = []
if deepstack_map:
for deepstack_idx, (vision_layer, llm_layer) in enumerate(sorted(deepstack_map)):
if vision_layer < 0:
vision_layer = n_vision_layers + vision_layer
if llm_layer < 0:
llm_layer = n_text_layers + llm_layer
normalized_projector_map.append((vision_layer, llm_layer, "layerwise", deepstack_idx))
if spatial_layers:
spatial_vision_layer = global_config.get("spatial_vision_layer", -1)
if spatial_vision_layer < 0:
spatial_vision_layer = n_vision_layers + spatial_vision_layer
for spatial_idx, llm_layer in enumerate(spatial_layers):
normalized_projector_map.append((spatial_vision_layer, llm_layer, "spatial", spatial_idx))
return list(sorted(normalized_projector_map, key=(lambda entry: entry[1])))
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
normalized_projector_map = self.get_normalized_projector_map(self.global_config)
self._n_proj = len(normalized_projector_map)
self._tensor_prefix_map = {
f"model.{proj_type}_projectors.{type_idx}": proj_idx
for proj_idx, (_, _, proj_type, type_idx) in enumerate(normalized_projector_map)
}
self._vision_feature_layers = [vision_layer for vision_layer, _, _, _ in normalized_projector_map]
self._spatial_offsets = [
type_idx if proj_type == "spatial" else -1
for _, _, proj_type, type_idx in normalized_projector_map
]
def set_gguf_parameters(self):
assert self.hparams_vision is not None
super().set_gguf_parameters()
self.gguf_writer.add_clip_projector_type(gguf.VisionProjectorType.GRANITE4_VISION)
# SigLIP encoder hparams
self.gguf_writer.add_vision_attention_layernorm_eps(self.hparams.get("layer_norm_eps", 1e-6))
self.gguf_writer.add_vision_use_gelu(True)
# Preprocessor
self.gguf_writer.add_vision_preproc_image_size(self.hparams.get("image_size", 384))
# QFormer projector config
ds_rate = self.global_config["downsample_rate"]
ds_parts = ds_rate.split("/")
assert len(ds_parts) == 2, f"Invalid 'downsample_rate' value: {ds_rate}"
query_side, window_side = [int(p) for p in ds_parts]
self.gguf_writer.add_vision_projector_query_side(query_side)
self.gguf_writer.add_vision_projector_window_side(window_side)
# Set vision feature layers
self.gguf_writer.add_vision_feature_layers(self._vision_feature_layers)
# Set the spatial offests per projector
self.gguf_writer.add_vision_spatial_offsets(self._spatial_offsets)
# Add flattened image grind pinpoints (resolution candidates internally)
if pinpoints := self.global_config.get("image_grid_pinpoints"):
# Flatten with h, w -> w, h inversion
pinpoints = [val for h, w in pinpoints for val in (w, h)]
self.gguf_writer.add_vision_image_grid_pinpoints(pinpoints)
@classmethod
def filter_tensors(cls, item: tuple[str, Callable[[], Tensor]]) -> tuple[str, Callable[[], Tensor]] | None:
name, _ = item
if ("vision_model.head" in name or name.startswith("lm_head")):
return None
return super().filter_tensors(item)
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
# Detect projector tensors and bin them
projector_idx = None
for prefix, proj_idx in self._tensor_prefix_map.items():
if name.startswith(prefix):
projector_idx = proj_idx
break
if projector_idx is not None:
# If this projector tensor has a block id within the projector,
# alias the bid to projector_idx
#
# TODO: currently, none of the Granite 4 Vision models have
# projectors with multiple QFormer layers, so the `layer.{}` index
# is always 0. This allows us to simply map to a single `bid` that
# matches the projector index. If this changes, we'll need a
# convention that merges the two IDs.
id_matches = list(re.finditer(r"\.([0-9]+)\.", name))
all_ids = [int(m.group(1)) for m in id_matches]
assert len(all_ids) >= 1 and len(all_ids) <= 2, "Must have at least 1 and at most 2 ids in tensor names"
# If not layer id, just use the projector index
new_bid = projector_idx
if len(all_ids) == 1:
new_name = name[:id_matches[0].span(1)[0]] + str(new_bid) + name[id_matches[0].span(1)[1]:]
else: # len(all_ids) == 2
new_bid = projector_idx # + all_ids[1]
new_name = name[:id_matches[0].span(0)[0]] + name[id_matches[0].span(1)[1]:id_matches[1].span(1)[0]] + str(new_bid) + name[id_matches[1].span(1)[1]:]
yield from super().modify_tensors(data_torch, new_name, new_bid)
return
yield from super().modify_tensors(data_torch, name, bid)