140 lines
6.5 KiB
Python
140 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Iterable, TYPE_CHECKING
|
|
|
|
import torch
|
|
|
|
if TYPE_CHECKING:
|
|
from torch import Tensor
|
|
|
|
from .base import MmprojModel, ModelBase, TextModel, gguf
|
|
|
|
|
|
@ModelBase.register("Lfm3ForCausalLM", "Lfm3VlForConditionalGeneration")
|
|
@ModelBase.example("LiquidAI/LFM3-1.7B", "LiquidAI/LFM3-1.7B-Base")
|
|
class LFM3Model(TextModel):
|
|
model_arch = gguf.MODEL_ARCH.LFM3
|
|
|
|
def set_gguf_parameters(self):
|
|
layer_types = self.hparams["layer_types"]
|
|
n_kv_heads = self.hparams["num_key_value_heads"]
|
|
|
|
# set num_key_value_heads only for attention layers
|
|
self.hparams["num_key_value_heads"] = [
|
|
n_kv_heads if lt != "conv" else 0
|
|
for lt in layer_types
|
|
]
|
|
|
|
super().set_gguf_parameters()
|
|
|
|
self.gguf_writer.add_rope_dimension_count(self.hparams["head_dim"])
|
|
|
|
self.gguf_writer.add_sliding_window_pattern([lt == "sliding_attention" for lt in layer_types])
|
|
self.gguf_writer.add_sliding_window(self.hparams["sliding_window"])
|
|
|
|
self.gguf_writer.add_shortconv_l_cache(self.hparams["conv_L_cache"])
|
|
|
|
def generate_extra_tensors(self) -> Iterable[tuple[str, Tensor]]:
|
|
# Partial RoPE is represented by masking trailing frequency pairs.
|
|
# rope_parameters is keyed by layer type; only full_attention layers use partial RoPE.
|
|
rope_parameters = self.hparams["rope_parameters"]["full_attention"]
|
|
partial = rope_parameters.get("partial_rotary_factor", 1.0)
|
|
head_dim = self.hparams["head_dim"]
|
|
rope_angles = int(partial * head_dim) // 2
|
|
n_pairs = head_dim // 2
|
|
assert 0 < rope_angles <= n_pairs
|
|
values = [1.0] * rope_angles + [1e30] * (n_pairs - rope_angles)
|
|
rope_freqs = torch.tensor(values, dtype=torch.float32)
|
|
yield (self.format_tensor_name(gguf.MODEL_TENSOR.ROPE_FREQS), rope_freqs)
|
|
|
|
def modify_tensors(self, data_torch: Tensor, name: str, bid: int | None) -> Iterable[tuple[str, Tensor]]:
|
|
# conv op requires 2d tensor
|
|
if "conv.conv" in name:
|
|
data_torch = data_torch.squeeze(1)
|
|
|
|
yield from super().modify_tensors(data_torch, name, bid)
|
|
|
|
|
|
@ModelBase.register("Lfm3VlForConditionalGeneration")
|
|
@ModelBase.example("LiquidAI/LFM3-VL-2B")
|
|
class LFM3VLMmprojModel(MmprojModel):
|
|
has_vision_encoder = True
|
|
has_audio_encoder = False
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
assert self.hparams_vision is not None
|
|
# dynamic resolution, image_size is only set for MmprojModel compatibility
|
|
self.hparams_vision["image_size"] = 512
|
|
|
|
def set_gguf_parameters(self):
|
|
super().set_gguf_parameters()
|
|
assert self.hparams_vision is not None
|
|
# clip.cpp hardcodes the vision rope base, the RGB patchifier, the 3x3 spatial conv and the GELU projector
|
|
if (self.hparams_vision["rope_parameters"]["rope_theta"] != 1e6 or self.hparams_vision["in_channels"] != 3
|
|
or self.hparams_vision["conv_kernel_size"] != 3):
|
|
raise ValueError("unsupported vision rope_theta / in_channels / conv_kernel_size")
|
|
if (self.global_config["projector_hidden_act"] != "gelu" or not self.global_config["projector_bias"]
|
|
or self.global_config["projector_layer_norm_eps"] != 1e-6):
|
|
raise ValueError("unsupported projector_hidden_act / projector_bias / projector_layer_norm_eps")
|
|
self.gguf_writer.add_clip_vision_projector_type(gguf.VisionProjectorType.LFM3VL)
|
|
self.gguf_writer.add_vision_head_count_kv(self.hparams_vision["num_key_value_heads"])
|
|
self.gguf_writer.add_vision_spatial_merge_size(self.global_config["downsample_factor"])
|
|
self.gguf_writer.add_vision_attention_layernorm_eps(self.hparams_vision["norm_eps"])
|
|
self.gguf_writer.add_vision_use_silu(True)
|
|
size = self.preprocessor_config["size"]
|
|
self.gguf_writer.add_vision_min_pixels(size["shortest_edge"])
|
|
self.gguf_writer.add_vision_max_pixels(size["longest_edge"])
|
|
# PIL resample enum, the torchvision backend has no lanczos
|
|
resample = self.preprocessor_config["resample"]
|
|
resize_algo = {2: "bilinear", 3: "bicubic"}.get(resample)
|
|
if resize_algo is None:
|
|
raise ValueError(f"unsupported resample: {resample}")
|
|
self.gguf_writer.add_vision_image_resize_algo(resize_algo)
|
|
|
|
def tensor_force_quant(self, name, new_name, bid, n_dims):
|
|
if "patch_embd" in new_name:
|
|
return gguf.GGMLQuantizationType.F16
|
|
if "conv2d_spatial" in new_name:
|
|
# CUDA CONV_2D_DW requires an f32 kernel; the tensor is tiny
|
|
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 not name.startswith(("model.vision_tower.", "model.multi_modal_projector.")):
|
|
return
|
|
|
|
if name == "model.vision_tower.patchifier.weight":
|
|
# qwen2vl-style layout [channel, temporal, ph, pw], one conv kernel per temporal slice
|
|
assert self.hparams_vision is not None
|
|
t = self.hparams_vision["temporal_patch_size"]
|
|
p = self.hparams_vision["patch_size"]
|
|
n_embd = data_torch.shape[0]
|
|
w = data_torch.view(n_embd, 3, t, p, p)
|
|
yield ("v.patch_embd.weight", w[:, :, 0].contiguous())
|
|
if t == 2:
|
|
yield ("v.patch_embd.weight.1", w[:, :, 1].contiguous())
|
|
else:
|
|
assert t == 1, f"unsupported vision_temporal_patch_size {t}"
|
|
return
|
|
if name == "model.vision_tower.embedding_norm.weight":
|
|
yield ("v.embedding_norm.weight", data_torch)
|
|
return
|
|
if "conv.in_proj.weight" in name:
|
|
yield (f"v.blk.{bid}.conv2d_in_proj.weight", data_torch)
|
|
return
|
|
if "conv.spatial_conv.weight" in name:
|
|
yield (f"v.blk.{bid}.conv2d_spatial.weight", data_torch)
|
|
return
|
|
if "conv.out_proj.weight" in name:
|
|
yield (f"v.blk.{bid}.conv2d_out_proj.weight", data_torch)
|
|
return
|
|
|
|
# projector: input LayerNorm -> linear_1 -> GELU -> linear_2
|
|
if name.startswith("model.multi_modal_projector."):
|
|
layer, suffix = name.removeprefix("model.multi_modal_projector.").split(".")
|
|
yield ({"layer_norm": "mm.input_norm", "linear_1": "mm.0", "linear_2": "mm.2"}[layer] + "." + suffix, data_torch)
|
|
return
|
|
|
|
yield (self.map_tensor_name(name), data_torch)
|