# 🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨
# This file was automatically generated from src/transformers/models/minicpmv4_7/modular_minicpmv4_7.py.
# Do NOT edit this file manually as any edits will be overwritten by the generation of
# the file from the modular. If any change should be done, please apply the change to the
# modular_minicpmv4_7.py file directly. One of our CI enforces this.
# 🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨
# Copyright 2026 OpenBMB and the HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import itertools
from collections.abc import Callable
from typing import Any
import torch
import torch.nn.functional as F
from torch import nn
from ... import initialization as init
from ...activations import ACT2FN, gelu_pytorch_tanh
from ...cache_utils import Cache
from ...generation import GenerationMixin
from ...modeling_layers import GradientCheckpointingLayer
from ...modeling_outputs import (
BaseModelOutput,
BaseModelOutputWithPast,
BaseModelOutputWithPooling,
CausalLMOutputWithPast,
)
from ...modeling_utils import ALL_ATTENTION_FUNCTIONS, PreTrainedModel
from ...processing_utils import Unpack
from ...utils import TransformersKwargs, auto_docstring, logging
from ...utils.generic import can_return_tuple, get_max_seqlen, is_flash_attention_requested, merge_with_config_defaults
from ...utils.import_utils import torch_compilable_check
from ...utils.output_capturing import capture_outputs
from ...vision_utils import get_vision_nearest_position_ids, get_vision_window_index
from ..auto import AutoModel
from .configuration_minicpmv4_7 import MiniCPMV4_7Config, MiniCPMV4_7VisionConfig
logger = logging.get_logger(__name__)
class MiniCPMV4_7VisionEmbeddings(nn.Module):
"""
This is a modified version of `siglip.modelign_siglip.SiglipVisionEmbeddings` to enable images of variable
resolution.
The modifications are adapted from [Patch n' Pack: NaViT, a Vision Transformer for any Aspect Ratio and Resolution](https://huggingface.co/papers/2307.06304)
which allows treating images in their native aspect ratio and without the need to resize them to the same
fixed size. In particular, we start from the original pre-trained SigLIP model
(which uses images of fixed-size square images) and adapt it by training on images of variable resolutions.
"""
def __init__(self, config: MiniCPMV4_7VisionConfig):
super().__init__()
self.embed_dim = config.hidden_size
self.image_size = config.image_size
self.patch_size = config.patch_size
self.patch_embedding = nn.Conv2d(
in_channels=config.num_channels,
out_channels=self.embed_dim,
kernel_size=self.patch_size,
stride=self.patch_size,
padding="valid",
)
self.num_patches_per_side = self.image_size // self.patch_size
self.num_patches = self.num_patches_per_side**2
self.num_positions = self.num_patches
self.position_embedding = nn.Embedding(self.num_positions, self.embed_dim)
def forward(
self,
pixel_values: torch.FloatTensor,
target_sizes: torch.IntTensor | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> torch.Tensor:
patch_embeds = self.patch_embedding(pixel_values)
embeddings = patch_embeds.flatten(2).transpose(1, 2)
pos_ids = get_vision_nearest_position_ids(target_sizes, self.num_patches_per_side, kwargs=kwargs)
pos_ids = pos_ids.to(self.position_embedding.weight.device)
position_embeddings = self.position_embedding(pos_ids).unsqueeze(0)
embeddings = embeddings + position_embeddings
return embeddings
class MiniCPMV4_7VisionMLP(nn.Module):
def __init__(self, config):
super().__init__()
self.config = config
self.activation_fn = ACT2FN[config.hidden_act]
self.fc1 = nn.Linear(config.hidden_size, config.intermediate_size)
self.fc2 = nn.Linear(config.intermediate_size, config.hidden_size)
def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
hidden_states = self.fc1(hidden_states)
hidden_states = self.activation_fn(hidden_states)
hidden_states = self.fc2(hidden_states)
return hidden_states
def repeat_kv(hidden_states: torch.Tensor, n_rep: int) -> torch.Tensor:
"""
This is the equivalent of torch.repeat_interleave(x, dim=1, repeats=n_rep). The hidden states go from (batch,
num_key_value_heads, seqlen, head_dim) to (batch, num_attention_heads, seqlen, head_dim)
"""
batch, num_key_value_heads, slen, head_dim = hidden_states.shape
if n_rep == 1:
return hidden_states
hidden_states = hidden_states[:, :, None, :, :].expand(batch, num_key_value_heads, n_rep, slen, head_dim)
return hidden_states.reshape(batch, num_key_value_heads * n_rep, slen, head_dim)
def eager_attention_forward(
module: nn.Module,
query: torch.Tensor,
key: torch.Tensor,
value: torch.Tensor,
attention_mask: torch.Tensor | None,
scaling: float,
dropout: float = 0.0,
**kwargs,
):
key_states = repeat_kv(key, module.num_key_value_groups)
value_states = repeat_kv(value, module.num_key_value_groups)
attn_weights = torch.matmul(query, key_states.transpose(2, 3)) * scaling
if attention_mask is not None:
attn_weights = attn_weights + attention_mask
attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query.dtype)
attn_weights = nn.functional.dropout(attn_weights, p=dropout, training=module.training)
attn_output = torch.matmul(attn_weights, value_states)
attn_output = attn_output.transpose(1, 2).contiguous()
return attn_output, attn_weights
class MiniCPMV4_7VisionAttention(nn.Module):
def __init__(self, config) -> None:
super().__init__()
self.dim = config.hidden_size
self.num_heads = config.num_attention_heads
self.head_dim = self.dim // self.num_heads
self.num_key_value_groups = 1 # needed for eager attention
self.scaling = self.head_dim**-0.5
self.config = config
self.attention_dropout = config.attention_dropout
self.is_causal = False
if self.head_dim * self.num_heads != self.dim:
raise ValueError(
f"embed_dim must be divisible by num_heads (got `embed_dim`: {self.dim} and `num_heads`:"
f" {self.num_heads})."
)
self.k_proj = nn.Linear(self.dim, self.dim)
self.v_proj = nn.Linear(self.dim, self.dim)
self.q_proj = nn.Linear(self.dim, self.dim)
self.out_proj = nn.Linear(self.dim, self.dim)
def forward(
self,
hidden_states: torch.Tensor,
cu_seqlens: torch.Tensor,
max_seqlen: int | None,
attention_mask: torch.Tensor | None = None,
**kwargs,
) -> torch.Tensor:
"""Input shape: Batch x Time x Channel"""
input_shape = hidden_states.shape[:-1]
hidden_shape = (*input_shape, -1, self.head_dim)
query_states = self.q_proj(hidden_states).view(hidden_shape).transpose(1, 2)
key_states = self.k_proj(hidden_states).view(hidden_shape).transpose(1, 2)
value_states = self.v_proj(hidden_states).view(hidden_shape).transpose(1, 2)
attention_interface: Callable = ALL_ATTENTION_FUNCTIONS.get_interface(
self.config._attn_implementation, eager_attention_forward
)
if is_flash_attention_requested(self.config):
# Flash Attention: Use cu_seqlens for variable length attention
max_seqlen = get_max_seqlen(cu_seqlens, self.config, kwargs={"max_seqlen": max_seqlen})
attn_output, _ = attention_interface(
self,
query_states,
key_states,
value_states,
attention_mask=None,
scaling=self.scaling,
dropout=0.0 if not self.training else self.attention_dropout,
cu_seq_lens_q=cu_seqlens,
cu_seq_lens_k=cu_seqlens,
max_length_q=max_seqlen,
max_length_k=max_seqlen,
is_causal=False,
**kwargs,
)
else:
# Other implementations: Process each chunk separately
lengths = cu_seqlens[1:] - cu_seqlens[:-1]
splits = [
torch.split(tensor, lengths.tolist(), dim=2) for tensor in (query_states, key_states, value_states)
]
attn_outputs = [
attention_interface(
self,
q,
k,
v,
attention_mask=None,
scaling=self.scaling,
dropout=0.0 if not self.training else self.attention_dropout,
is_causal=False,
**kwargs,
)[0]
for q, k, v in zip(*splits)
]
attn_output = torch.cat(attn_outputs, dim=1)
attn_output = attn_output.reshape(*input_shape, -1).contiguous()
attn_output = self.out_proj(attn_output)
return attn_output, None
class MiniCPMV4_7VisionEncoderLayer(GradientCheckpointingLayer):
def __init__(self, config: MiniCPMV4_7VisionConfig):
super().__init__()
self.embed_dim = config.hidden_size
self.layer_norm1 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps)
self.self_attn = MiniCPMV4_7VisionAttention(config)
self.layer_norm2 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps)
self.mlp = MiniCPMV4_7VisionMLP(config)
@auto_docstring
def forward(
self,
hidden_states: torch.Tensor,
attention_mask: torch.Tensor,
**kwargs: Unpack[TransformersKwargs],
) -> torch.FloatTensor:
residual = hidden_states
hidden_states = self.layer_norm1(hidden_states)
hidden_states, _ = self.self_attn(
hidden_states=hidden_states,
attention_mask=attention_mask,
**kwargs,
)
hidden_states = residual + hidden_states
residual = hidden_states
hidden_states = self.layer_norm2(hidden_states)
hidden_states = self.mlp(hidden_states)
hidden_states = residual + hidden_states
return hidden_states
class MiniCPMV4_7VisionEncoder(nn.Module):
"""Transformer encoder consisting of `config.num_hidden_layers` [`MiniCPMV4_7VisionEncoderLayer`] layers."""
def __init__(self, config: MiniCPMV4_7VisionConfig):
super().__init__()
self.config = config
self.layers = nn.ModuleList([MiniCPMV4_7VisionEncoderLayer(config) for _ in range(config.num_hidden_layers)])
self.gradient_checkpointing = False
# Ignore copy
@auto_docstring
def forward(
self,
inputs_embeds,
attention_mask: torch.Tensor | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> BaseModelOutput:
hidden_states = inputs_embeds
for encoder_layer in self.layers:
hidden_states = encoder_layer(
hidden_states,
attention_mask,
**kwargs,
)
return BaseModelOutput(last_hidden_state=hidden_states)
class MiniCPMV4_7ViTWindowAttentionMerger(nn.Module):
def __init__(self, config: MiniCPMV4_7VisionConfig):
super().__init__()
self.window_kernel_size = tuple(config.window_kernel_size)
self.embed_dim = config.hidden_size
self.self_attn = MiniCPMV4_7VisionAttention(config)
self.layer_norm1 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps)
self.pre_norm = nn.LayerNorm(config.window_hidden_size, eps=config.layer_norm_eps)
self.linear_1 = nn.Linear(config.window_hidden_size, config.window_intermediate_size, bias=True)
self.act = gelu_pytorch_tanh
self.linear_2 = nn.Linear(config.window_intermediate_size, self.embed_dim, bias=True)
def _init_weights(self):
"""Block-diagonal normal init: preserves the structural prior that each
2x2 window patch is processed independently at initialization.
Uses ``init.*`` helpers so the ``_is_hf_initialized`` guard is
respected. The ``linear_1`` block-diagonal init writes to *slices*
which do not inherit the flag, so we guard the entire block manually.
"""
for proj in (self.self_attn.q_proj, self.self_attn.k_proj, self.self_attn.v_proj, self.self_attn.out_proj):
init.normal_(proj.weight)
init.zeros_(proj.bias)
for ln in (self.layer_norm1, self.pre_norm):
init.ones_(ln.weight)
init.zeros_(ln.bias)
hidden_size = self.embed_dim
intermediate_size = self.linear_1.weight.shape[0] // 4
if not getattr(self.linear_1.weight, "_is_hf_initialized", False):
self.linear_1.weight.data.zero_()
for i in range(4):
self.linear_1.weight.data[
i * intermediate_size : (i + 1) * intermediate_size,
i * hidden_size : (i + 1) * hidden_size,
].normal_()
self.linear_1.weight._is_hf_initialized = True
init.normal_(self.linear_1.bias, std=1e-6)
init.normal_(self.linear_2.weight, std=0.25)
init.normal_(self.linear_2.bias, std=1e-6)
def get_window_index(self, target_sizes, kwargs=None):
window_h, window_w = self.window_kernel_size
if window_h != window_w:
raise ValueError(f"window_kernel_size must be square; got ({window_h}, {window_w})")
grid_thw = F.pad(target_sizes, (1, 0), value=1)
window_index, cu_seqlens = get_vision_window_index(
grid_thw, spatial_merge_size=1, window_size=window_h, patch_size=1, kwargs=kwargs
)
return window_index, cu_seqlens, window_h * window_w
def forward(
self,
hidden_states: torch.Tensor,
target_sizes: torch.IntTensor,
**kwargs: Unpack[TransformersKwargs],
):
residual = hidden_states
hidden_states = self.layer_norm1(hidden_states)
device = hidden_states.device
window_index, window_cu_seqlens, window_max_seqlens = self.get_window_index(target_sizes, kwargs=kwargs)
window_index = window_index.to(device)
hidden_states = hidden_states[:, window_index, :]
hidden_states, _ = self.self_attn(
hidden_states=hidden_states,
cu_seqlens=window_cu_seqlens.to(device),
max_seqlen=window_max_seqlens,
)
hidden_states = residual[:, window_index, :] + hidden_states
window_h, window_w = self.window_kernel_size
window_size = window_h * window_w
embed_dim = hidden_states.shape[-1]
torch_compilable_check(
window_cu_seqlens.numel() - 1 == hidden_states.shape[1] // window_size,
f"Patch grids {target_sizes} must be divisible by window kernel size {self.window_kernel_size}",
)
patch = hidden_states.reshape(-1, window_size, embed_dim)
flat = patch.flatten(1)
patch_residual = patch.mean(dim=1)
hidden_state = self.pre_norm(flat)
hidden_state = self.linear_1(hidden_state)
hidden_state = self.act(hidden_state)
hidden_state = self.linear_2(hidden_state)
hidden_state = (hidden_state + patch_residual).unsqueeze(0)
return hidden_state
class MiniCPMV4_7VisionPreTrainedModel(PreTrainedModel):
config_class = MiniCPMV4_7VisionConfig
main_input_name = "pixel_values"
_input_embed_layer = "patch_embedding"
supports_gradient_checkpointing = True
_supports_sdpa = True
_supports_flash_attn = True
_can_record_outputs = {
"hidden_states": MiniCPMV4_7VisionEncoderLayer,
"attentions": MiniCPMV4_7VisionAttention,
}
class MiniCPMV4_7VisionModel(MiniCPMV4_7VisionPreTrainedModel):
def __init__(self, config: MiniCPMV4_7VisionConfig):
super().__init__(config)
embed_dim = config.hidden_size
self.embeddings = MiniCPMV4_7VisionEmbeddings(config)
self.encoder = MiniCPMV4_7VisionEncoder(config)
self.post_layernorm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps)
self.vit_merger = MiniCPMV4_7ViTWindowAttentionMerger(config)
self.post_init()
def get_downsampled_inputs(
self, target_sizes: torch.Tensor, max_seqlens: int | None, device: torch.device, **kwargs
) -> tuple[dict[str, Any], torch.Tensor, torch.Tensor]:
# NOTE: intentionally not checking for shapes as this is expensive to call `.any()`
target_sizes = target_sizes // 2
# `max_seqlens` is only computed and used for Flash Attention.
if max_seqlens is not None:
max_seqlens = max_seqlens // 4
cu_seqlens = F.pad(
torch.cumsum(target_sizes[:, 0] * target_sizes[:, 1], dim=0, dtype=torch.int32).to(device), (1, 0)
)
downsampled_kwargs = {
"attention_mask": None,
"cu_seqlens": cu_seqlens,
"max_seqlen": max_seqlens,
**kwargs,
}
return downsampled_kwargs, target_sizes, cu_seqlens
@merge_with_config_defaults
@capture_outputs
@auto_docstring
def forward(
self,
pixel_values,
target_sizes: torch.IntTensor | None = None,
use_vit_merger: bool = True,
**kwargs: Unpack[TransformersKwargs],
) -> BaseModelOutputWithPooling:
r"""
target_sizes (`torch.IntTensor` of shape `(batch_size, 2)`, *optional*):
Patch grid sizes `(h, w)` for computing position embeddings.
use_vit_merger (`bool`, *optional*, defaults to `True`):
Whether to apply the ViT window-attention merger after the encoder.
"""
hidden_states = self.embeddings(pixel_values, target_sizes=target_sizes, **kwargs)
cu_seqlens = F.pad(
torch.cumsum(target_sizes[:, 0] * target_sizes[:, 1], dim=0, dtype=torch.int32).to(hidden_states.device),
(1, 0),
)
max_seqlens = get_max_seqlen(cu_seqlens, self.config, kwargs=kwargs)
attn_kwargs = {
"attention_mask": None,
"cu_seqlens": cu_seqlens,
"max_seqlen": max_seqlens,
**kwargs,
}
insert_layer_id = self.config.insert_layer_id if use_vit_merger else -1
if use_vit_merger and insert_layer_id >= 0:
for layer_index, encoder_layer in enumerate(self.encoder.layers):
hidden_states = encoder_layer(hidden_states, **attn_kwargs)
if layer_index == insert_layer_id:
hidden_states = self.vit_merger(hidden_states, target_sizes, **kwargs)
# NOTE: Downsampled hidden states, and therefore other kwargs should also!
attn_kwargs, target_sizes, cu_seqlens = self.get_downsampled_inputs(
target_sizes=target_sizes, max_seqlens=max_seqlens, device=hidden_states.device, **kwargs
)
else:
encoder_outputs = self.encoder(inputs_embeds=hidden_states, **attn_kwargs)
hidden_states = encoder_outputs.last_hidden_state
last_hidden_state = self.post_layernorm(hidden_states)
return BaseModelOutputWithPooling(last_hidden_state=last_hidden_state)
class MiniCPMV4_7DownsampleMLP(nn.Module):
def __init__(self, hidden_size: int, llm_embed_dim: int):
super().__init__()
# factor 4 = two successive 2×2 spatial merges (ViT insert merger + downsample MLP)
merged_hidden_size = hidden_size * 4
self.pre_norm = nn.LayerNorm(merged_hidden_size, eps=1e-6)
self.linear_1 = nn.Linear(merged_hidden_size, merged_hidden_size, bias=True)
self.act = nn.GELU()
self.linear_2 = nn.Linear(merged_hidden_size, llm_embed_dim, bias=True)
def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
hidden_states = self.pre_norm(hidden_states).view(-1, self.linear_1.in_features)
hidden_states = self.linear_1(hidden_states)
hidden_states = self.act(hidden_states)
hidden_states = self.linear_2(hidden_states)
return hidden_states
class MiniCPMV4_7Merger(nn.Module):
def __init__(self, config: MiniCPMV4_7Config):
super().__init__()
self.merge_kernel_size = tuple(config.merge_kernel_size)
self.merger_times = config.merger_times
hidden_size = config.vision_config.hidden_size
llm_embed_dim = config.text_config.hidden_size
# Downsample `self.merger_times - 1` times and finally apply projection into LLM space
mlps = [MiniCPMV4_7DownsampleMLP(hidden_size, hidden_size) for _ in range(self.merger_times - 1)]
mlps.append(MiniCPMV4_7DownsampleMLP(hidden_size, llm_embed_dim))
self.mlp = nn.ModuleList(mlps)
def forward(
self,
hidden_states: torch.Tensor,
target_sizes: torch.IntTensor,
) -> list[torch.Tensor]:
merge_h, merge_w = self.merge_kernel_size
start = 0
processed_features = []
for batch_idx in range(len(target_sizes)):
height, width = target_sizes[batch_idx]
num_patches = height * width
embed_dim = hidden_states.shape[-1]
merged_h, merged_w = height // merge_h, width // merge_w
hidden_state = (
hidden_states[0, start : start + num_patches, :]
.view(merged_h, merge_h, merged_w, merge_w, embed_dim)
.permute(0, 2, 1, 3, 4)
.reshape(merged_h * merged_w, merge_h * merge_w * embed_dim)
)
hidden_state = self.mlp[0](hidden_state)
for i in range(1, self.merger_times):
if height % merge_h != 0 or width % merge_w != 0:
raise ValueError(
f"Patch grid ({height}, {width}) must be divisible by merge kernel size "
f"{self.merge_kernel_size} at merge round {i}"
)
height = height // merge_h
width = width // merge_w
inner_dim = hidden_state.shape[-1]
merged_h, merged_w = height // merge_h, width // merge_w
hidden_state = (
hidden_state.view(merged_h, merge_h, merged_w, merge_w, inner_dim)
.permute(0, 2, 1, 3, 4)
.reshape(merged_h * merged_w, merge_h * merge_w * inner_dim)
)
hidden_state = self.mlp[i](hidden_state)
start += num_patches
processed_features.append(hidden_state)
return processed_features
class MiniCPMV4_7PreTrainedModel(PreTrainedModel):
config_class = MiniCPMV4_7Config
base_model_prefix = "model"
input_modalities = ("image", "video", "text")
supports_gradient_checkpointing = True
_supports_flash_attn = True
_supports_sdpa = True
_no_split_modules = [
"MiniCPMV4_7VisionEmbeddings",
"MiniCPMV4_7VisionEncoderLayer",
"MiniCPMV4_7ViTWindowAttentionMerger",
]
_is_stateful = True
@auto_docstring(
custom_intro="""
The MiniCPMV4_7 model which consists of a vision backbone and a language model, without a language modeling head.
"""
)
class MiniCPMV4_7Model(MiniCPMV4_7PreTrainedModel):
def __init__(self, config: MiniCPMV4_7Config):
super().__init__(config)
self.vision_tower = MiniCPMV4_7VisionModel._from_config(config.vision_config)
self.language_model = AutoModel.from_config(config.text_config)
self.merger = MiniCPMV4_7Merger(config)
self.rope_deltas = None
self.post_init()
@can_return_tuple
@auto_docstring(custom_intro="Extract image features: vision encoder, insert merger, then MLP merger.")
def get_image_features(
self,
pixel_values: torch.FloatTensor,
target_sizes: torch.IntTensor,
downsample_mode: str | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> BaseModelOutputWithPooling:
r"""
target_sizes (`torch.IntTensor` of shape `(num_images, 2)`):
Height and width (in patches) of each image.
downsample_mode (`str`, *optional*):
When set to `"4x"` the intermediate `vit_merger` is skipped so that each image keeps
`4×` more visual tokens. Default `"16x"` mode applies the full merge pipeline.
"""
downsample_mode = downsample_mode if downsample_mode else self.config.downsample_mode
use_vit_merger = downsample_mode != "4x"
pixel_values = pixel_values.to(dtype=self.vision_tower.dtype)
vision_output = self.vision_tower(
pixel_values,
target_sizes=target_sizes,
use_vit_merger=use_vit_merger,
**kwargs,
)
if use_vit_merger:
target_sizes = target_sizes // 2
vision_output.pooler_output = self.merger(vision_output.last_hidden_state, target_sizes)
return vision_output
def get_placeholder_mask(
self,
input_ids: torch.LongTensor,
inputs_embeds: torch.FloatTensor,
features: torch.FloatTensor,
token_id: int,
):
"""
Obtains multimodal placeholder mask from `input_ids` or `inputs_embeds`, and checks that the placeholder token count is
equal to the length of multimodal features. If the lengths are different, an error is raised.
"""
if input_ids is None:
special_mask = inputs_embeds == self.get_input_embeddings()(
torch.tensor(token_id, dtype=torch.long, device=inputs_embeds.device)
)
special_mask = special_mask.all(-1)
else:
special_mask = input_ids == token_id
n_tokens = special_mask.sum()
special_mask = special_mask.unsqueeze(-1).expand_as(inputs_embeds).to(inputs_embeds.device)
n_features = features.shape[0]
torch_compilable_check(
inputs_embeds[special_mask].numel() == features.numel(),
f"Multimodal features and tokens do not match, tokens: {n_tokens}, features: {n_features}",
)
return special_mask
@can_return_tuple
@auto_docstring
def forward(
self,
input_ids: torch.LongTensor | None = None,
pixel_values: torch.FloatTensor | None = None,
target_sizes: torch.IntTensor | None = None,
pixel_values_videos: torch.FloatTensor | None = None,
target_sizes_videos: torch.IntTensor | None = None,
attention_mask: torch.Tensor | None = None,
position_ids: torch.LongTensor | None = None,
past_key_values: list[torch.FloatTensor] | None = None,
inputs_embeds: torch.FloatTensor | None = None,
use_cache: bool | None = None,
downsample_mode: str | None = None,
mm_token_type_ids: torch.IntTensor | None = None,
mm_encoder_outputs: dict[str, BaseModelOutputWithPooling] | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> tuple | BaseModelOutputWithPast:
r"""
pixel_values (`torch.FloatTensor`, *optional*):
Pixel value patches for images, NaViT-packed.
target_sizes (`torch.IntTensor`, *optional*):
Height and width (in patches) for each image.
pixel_values_videos (`torch.FloatTensor`, *optional*):
Pixel value patches for video frames, NaViT-packed.
target_sizes_videos (`torch.IntTensor`, *optional*):
Height and width (in patches) for each video frame.
downsample_mode (`str`, *optional*):
`"4x"` keeps 4x more visual tokens; default `"16x"` applies full merge.
"""
if (pixel_values is not None or pixel_values_videos is not None) and mm_encoder_outputs is not None:
raise ValueError(
"You cannot specify both pixel_values/pixel_values_videos and mm_encoder_outputs at the same time"
)
if inputs_embeds is None:
inputs_embeds = self.get_input_embeddings()(input_ids)
mm_encoder_outputs = mm_encoder_outputs if mm_encoder_outputs is not None else {}
if (
mm_encoder_outputs.get("image") is None
and pixel_values is not None
and self.config.image_token_id is not None
):
mm_encoder_outputs["image"] = self.get_image_features(
pixel_values[:1], target_sizes, downsample_mode=downsample_mode
)
if (
mm_encoder_outputs.get("video") is None
and pixel_values_videos is not None
and self.config.video_token_id is not None
):
mm_encoder_outputs["video"] = self.get_video_features(
pixel_values_videos[:1], target_sizes_videos, downsample_mode=downsample_mode
)
if mm_encoder_outputs.get("image") is not None:
image_features = torch.cat(mm_encoder_outputs["image"].pooler_output, dim=0).to(
device=inputs_embeds.device, dtype=inputs_embeds.dtype
)
mask = self.get_placeholder_mask(input_ids, inputs_embeds, image_features, self.config.image_token_id)
inputs_embeds = inputs_embeds.masked_scatter(mask, image_features)
if mm_encoder_outputs.get("video") is not None:
video_features = torch.cat(mm_encoder_outputs["video"].pooler_output, dim=0).to(
device=inputs_embeds.device, dtype=inputs_embeds.dtype
)
mask = self.get_placeholder_mask(input_ids, inputs_embeds, video_features, self.config.video_token_id)
inputs_embeds = inputs_embeds.masked_scatter(mask, video_features)
if position_ids is None:
position_ids = self.compute_3d_position_ids(
input_ids=input_ids,
inputs_embeds=inputs_embeds,
attention_mask=attention_mask,
past_key_values=past_key_values,
target_sizes=target_sizes,
target_sizes_videos=target_sizes_videos,
mm_token_type_ids=mm_token_type_ids,
downsample_mode=downsample_mode,
)
output = self.language_model(
attention_mask=attention_mask,
position_ids=position_ids,
past_key_values=past_key_values,
inputs_embeds=inputs_embeds,
use_cache=use_cache,
**kwargs,
)
return output
@can_return_tuple
@auto_docstring(
custom_intro="Extract video features: repack frames into NaViT format, then vision encoder + merger."
)
def get_video_features(
self,
pixel_values_videos: torch.FloatTensor,
target_sizes_videos: torch.IntTensor,
downsample_mode: str | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> BaseModelOutputWithPooling:
r"""
pixel_values_videos (`torch.FloatTensor` of shape `(1, channels, patch_size, seq_len)`):
NaViT-packed pixel patches for all video frames. The video processor concatenates
every frame's patches along the last dimension into a single sequence with dim-0 = 1,
identical to the image packing format.
target_sizes_videos (`torch.IntTensor` of shape `(num_patches, 2)`):
Height and width (in patches) of each visual unit.
downsample_mode (`str`, *optional*):
When set to `"4x"` the intermediate `vit_merger` is skipped so that each frame keeps
`4×` more visual tokens. Default `"16x"` mode applies the full merge pipeline.
"""
num_frames = pixel_values_videos.shape[0]
pixel_values = pixel_values_videos.permute(1, 2, 0, 3).reshape(
1, pixel_values_videos.shape[1], pixel_values_videos.shape[2], -1
)
target_sizes = target_sizes_videos.repeat(num_frames, 1)
return self.get_image_features(
pixel_values,
target_sizes,
downsample_mode=downsample_mode,
**kwargs,
)
@staticmethod
def _frame_end_idx(last_crop_end_idx: int, input_ids: list[int], markers: tuple[int, ...], limit: int) -> int:
"""End of a frame's span, including the ````/```` markers that close it."""
end_idx = last_crop_end_idx
while end_idx < limit and input_ids[end_idx] in markers:
end_idx += 1
return end_idx
def _group_visual_frames(
self, input_ids: list[int], mm_token_type_ids: list[int]
) -> list[tuple[int, int, list[tuple[int, int, int, int, list[tuple[int, int]]]]]]:
"""Split one sequence into visual groups, each a list of ``thumbnail + slices`` frames.
The canvas is described by three nested spans, group -> frame -> slice:
- a **slice** is one detail crop, the ``/`` pair around it pinning its corners;
- a **frame** is one picture or one video frame: a thumbnail, optionally followed by the slices
that tile it at higher resolution, the whole thing wrapped in ``/``;
- a **group** is a run of frames that only markers separate, so they share one canvas budget.
``mm_token_type_ids`` labels every token with its modality (``0`` text, ``1`` image, ``2``
video), so each maximal non-text run is exactly one crop. A crop is a slice when ````
sits in front of it, otherwise it opens a new frame and adopts the slices that follow.
Frames separated by nothing but markers and newlines -- how a clip emits its frames, and how a
picture can end up sitting right against one -- share a group. Grouping them matters: the next
frame then starts one step past the tokens in between, so its halo cannot land on the last of
them.
Args:
input_ids: `(seq_len,)` token ids of one unpadded sequence.
mm_token_type_ids: `(seq_len,)` modality label per token (0 text, 1 image, 2 video).
Returns:
One tuple per visual group, in sequence order, holding the span indices the caller needs so
that it never has to rescan `input_ids` itself:
`(group_start_idx, group_end_idx, frames)`, where `group_start_idx` is the index of the
`` marker opening the group, `group_end_idx` is one past the last marker closing it
and `frames` lists the group's frames as
`(modality, thumbnail_start_idx, thumbnail_end_idx, frame_end_idx, slices)`. `slices` holds
one `(start_idx, end_idx)` pair per detail crop and `frame_end_idx` is one past the markers
that close that frame.
"""
# `` comes first by convention: it is the only marker that has to be read on its own, to
# tell a detail slice from the thumbnail that opens a new frame.
markers = (
self.config.slice_start_id,
self.config.slice_end_id,
self.config.image_start_id,
self.config.image_end_id,
self.config.newline_id,
)
slice_start_id = markers[0]
seq_len = len(input_ids)
groups = []
open_slices = None
previous_end_idx = 0
for modality, run in itertools.groupby(enumerate(mm_token_type_ids), lambda item: item[1]):
if modality == 0:
continue
run = list(run)
crop = (run[0][0], run[-1][0] + 1)
if open_slices is not None and input_ids[crop[0] - 1] == slice_start_id:
open_slices.append(crop)
else:
# A frame is collected without its end index, which is only known once the whole
# sequence has been walked; the second pass below fills it in.
open_slices = []
frame = (modality, crop[0], crop[1], open_slices)
gap = input_ids[previous_end_idx : crop[0] - 1]
adjacent = bool(groups) and all(token_id in markers for token_id in gap)
if adjacent:
groups[-1][1].append(frame)
else:
groups.append((crop[0] - 1, [frame]))
previous_end_idx = crop[1]
visual_groups = []
for group_start_idx, frames in groups:
frames = [
(
modality,
thumb_start_idx,
thumb_end_idx,
self._frame_end_idx(slices[-1][1] if slices else thumb_end_idx, input_ids, markers, seq_len),
slices,
)
for modality, thumb_start_idx, thumb_end_idx, slices in frames
]
group_end_idx = frames[-1][3]
visual_groups.append((group_start_idx, group_end_idx, frames))
return visual_groups
def get_vision_position_ids(
self,
start_position: int,
grid_thw: list[int, int, int] | torch.Tensor,
canvas_height: int = 0,
canvas_width: int = 0,
h_offset: int = 0,
w_offset: int = 0,
spatial_merge_size: int = 1,
device: str | torch.device | None = None,
) -> torch.Tensor:
"""3D (t, h, w) position ids for one grid of patches placed on a canvas.
canvas_height/canvas_width == 0 means "no canvas given" -> canvas defaults to
this grid's own size, so the linspace resample degenerates to a plain arange.
That single default covers three cases with the same math:
- thumbnail *with* slices: canvas = full slice canvas, offsets = 0
- thumbnail *without* slices: canvas defaults to own grid -> arange, offsets = 0
- a slice's interior grid: canvas defaults to own grid -> arange, offsets = its (h, w) placement
Args:
start_position (`int`):
Canvas origin in the position-id sequence. The `t` channel is frozen to it and the
`h`/`w` channels are counted from it.
grid_thw (`list[int]` or `torch.Tensor`):
Patch grid of this crop; only `[0]` (height) and `[1]` (width) are read.
canvas_height (`int`, *optional*, defaults to 0):
Height of the canvas in LLM tokens. 0 falls back to this grid's own height.
canvas_width (`int`, *optional*, defaults to 0):
Width of the canvas in LLM tokens. 0 falls back to this grid's own width.
h_offset (`int`, *optional*, defaults to 0):
Row of the canvas where this grid's top-left corner sits, in LLM tokens.
w_offset (`int`, *optional*, defaults to 0):
Column of the canvas where this grid's top-left corner sits, in LLM tokens.
spatial_merge_size (`int`, *optional*, defaults to 1):
Patch-to-LLM-token merge factor (4 in `"16x"` mode, 2 in `"4x"` mode).
device (`str` or `torch.device`, *optional*):
Device the returned tensor is allocated on.
Returns:
`torch.Tensor`: `(3, llm_grid_h * llm_grid_w)` position ids `[T, H, W]`, row-major over
this crop's LLM tokens.
"""
llm_grid_h = int(grid_thw[0]) // spatial_merge_size
llm_grid_w = int(grid_thw[1]) // spatial_merge_size
canvas_height = canvas_height or llm_grid_h
canvas_width = canvas_width or llm_grid_w
h_coords = torch.linspace(0, canvas_height - 1, llm_grid_h, device=device).round().long() + h_offset
w_coords = torch.linspace(0, canvas_width - 1, llm_grid_w, device=device).round().long() + w_offset
H_grid, W_grid = torch.meshgrid(h_coords, w_coords, indexing="ij")
T_grid = torch.zeros_like(H_grid)
position_ids = torch.stack([T_grid, H_grid, W_grid], dim=0).reshape(3, -1)
position_ids += start_position
return position_ids
def get_rope_index(
self,
input_ids: torch.LongTensor,
mm_token_type_ids: torch.IntTensor | None = None,
target_sizes: torch.LongTensor | None = None,
target_sizes_videos: torch.LongTensor | None = None,
downsample_mode: str | None = None,
attention_mask: torch.Tensor | None = None,
**kwargs,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Canvas M-RoPE indices ``(3, B, S)`` plus ``rope_deltas``.
``mm_token_type_ids`` marks the crops, so the canvas is laid out per sequence on the
tokens ``attention_mask`` keeps. Padding stays at zero and never shifts the canvas.
Args:
input_ids (`torch.LongTensor` of shape `(batch_size, seq_len)`):
Token ids of the prompt. The canvas is anchored on the markers around each crop,
so it cannot be derived from `inputs_embeds` alone.
mm_token_type_ids (`torch.IntTensor` of shape `(batch_size, seq_len)`, *optional*):
Modality label per token (0 text, 1 image, 2 video), as emitted by the processor.
target_sizes (`torch.LongTensor` of shape `(num_image_crops, 2)`, *optional*):
Patch grid `(h, w)` of every image crop, in the order the crops appear.
target_sizes_videos (`torch.LongTensor` of shape `(num_video_crops, 2)`, *optional*):
Patch grid `(h, w)` of every video crop, in the order the crops appear.
downsample_mode (`str`, *optional*):
Per-call override of `config.downsample_mode`. It has to match the mode the vision
tower runs with, otherwise the placeholder count and the canvas disagree.
attention_mask (`torch.Tensor` of shape `(batch_size, seq_len)`, *optional*):
Padding mask. The canvas is laid out over the kept tokens only.
Returns:
`tuple[torch.Tensor, torch.Tensor]`:
- **position_ids** of shape `(3, batch_size, seq_len)` holding `[T, H, W]`.
- **rope_deltas** of shape `(batch_size, 1)`, the offset generation adds to the 1-D
positions of the tokens appended after the prompt.
"""
# `16x` merges 4x4 patches into one LLM token (window merger 2x2, then merger 2x2), `4x`
# skips the window merger and merges 2x2. Same divisor the processor counts placeholders
# with, so it has to follow the per-call override the vision tower is given, not the config.
downsample_mode = downsample_mode or self.config.downsample_mode
merge_factor = 2 if downsample_mode == "4x" else 4
device = input_ids.device
position_ids = torch.zeros(3, *input_ids.size(), dtype=torch.long, device=input_ids.device)
# The grids drive nothing but python-level arithmetic here, so they are read out once instead
# of being `.item()`-ed crop by crop.
grid_iters = {
1: iter(target_sizes.tolist()) if target_sizes is not None else None,
2: iter(target_sizes_videos.tolist()) if target_sizes_videos is not None else None,
}
for batch_idx in range(input_ids.shape[0]):
current_input_ids = input_ids[batch_idx]
current_mm_token_type_ids = mm_token_type_ids[batch_idx]
if attention_mask is not None:
valid_mask = attention_mask[batch_idx].bool()
current_input_ids = current_input_ids[valid_mask]
current_mm_token_type_ids = current_mm_token_type_ids[valid_mask]
else:
# Nothing to unpad, so the canvas is scattered back over the whole row.
valid_mask = slice(None)
current_input_ids = current_input_ids.tolist()
current_mm_token_type_ids = current_mm_token_type_ids.tolist()
seq_len = len(current_input_ids)
curr_position_ids = torch.arange(seq_len, device=device, dtype=torch.long).expand(3, -1).clone()
current_pos = 0
current_idx = 0
for group_start_idx, group_end_idx, frames in self._group_visual_frames(
current_input_ids, current_mm_token_type_ids
):
# Text in front of the group is plain 1-D.
if group_start_idx > current_idx:
text_len = group_start_idx - current_idx
curr_position_ids[:, current_idx:group_start_idx] = (
torch.arange(text_len, device=device) + current_pos
)
current_pos += text_len
frame_cursor_idx = group_start_idx
for modality, thumb_start_idx, thumb_end_idx, frame_end_idx, slices in frames:
target_sizes_thumb = next(grid_iters[modality])
frame_start_idx = thumb_start_idx - 1
# Tokens between two frames of a clip are 1-D text; the extra step afterwards keeps the
# next frame's halo from landing on the last of them.
if frame_start_idx > frame_cursor_idx:
gap_len = frame_start_idx - frame_cursor_idx
curr_position_ids[:, frame_cursor_idx:frame_start_idx] = (
torch.arange(gap_len, device=device) + current_pos
)
current_pos += gap_len + 1
frame_end_idx = min(frame_end_idx, group_end_idx)
canvas_origin = current_pos
halo_before_canvas = max(canvas_origin - 1, 0)
llm_slice_h, llm_slice_w = 0, 0
num_rows, num_cols = 0, 0
canvas_height = target_sizes_thumb[0] // merge_factor
canvas_width = target_sizes_thumb[1] // merge_factor
if slices:
# Slices are laid out row-major; a gap wider than the two `` markers
# is the newline that ends a row and therefore fixes the column count.
num_cols = len(slices)
for k in range(len(slices) - 1):
if slices[k + 1][0] - slices[k][1] > 2:
num_cols = k + 1
break
num_rows = len(slices) // num_cols if num_cols > 0 else 1
if num_rows * num_cols != len(slices):
num_rows, num_cols = 1, len(slices)
target_sizes_first_slice = next(grid_iters[modality])
llm_slice_h = target_sizes_first_slice[0] // merge_factor
llm_slice_w = target_sizes_first_slice[1] // merge_factor
canvas_height = num_rows * llm_slice_h
canvas_width = num_cols * llm_slice_w
# Base coat for the whole frame, then `` -> halo just outside the canvas.
curr_position_ids[:, frame_start_idx:frame_end_idx] = canvas_origin
curr_position_ids[1:, frame_start_idx] = halo_before_canvas
# `` closes the thumbnail on the far corner of the canvas.
if thumb_end_idx < frame_end_idx:
curr_position_ids[1, thumb_end_idx] = canvas_origin + canvas_height
curr_position_ids[2, thumb_end_idx] = canvas_origin + canvas_width
# Thumbnail tokens. With slices around, the thumbnail is stretched over the full canvas
# so that it stays aligned with the detail crops underneath it. Without slices the canvas
# is the thumbnail grid itself and `linspace` degenerates to `arange`.
curr_position_ids[:, thumb_start_idx:thumb_end_idx] = self.get_vision_position_ids(
start_position=canvas_origin,
grid_thw=target_sizes_thumb,
canvas_height=canvas_height,
canvas_width=canvas_width,
spatial_merge_size=merge_factor,
device=device,
)
# Slice tokens plus the ``/`` markers that pin each crop's corners.
for k, (slice_start_idx, slice_end_idx) in enumerate(slices):
h_off = (k // num_cols) * llm_slice_h
w_off = (k % num_cols) * llm_slice_w
target_sizes_slice = target_sizes_first_slice if k == 0 else next(grid_iters[modality])
slice_h = target_sizes_slice[0] // merge_factor
slice_w = target_sizes_slice[1] // merge_factor
slice_start_marker_idx = slice_start_idx - 1
if slice_start_marker_idx >= frame_start_idx:
curr_position_ids[1, slice_start_marker_idx] = canvas_origin + h_off
curr_position_ids[2, slice_start_marker_idx] = canvas_origin + w_off
if slice_end_idx < frame_end_idx:
curr_position_ids[1, slice_end_idx] = canvas_origin + h_off + slice_h - 1
curr_position_ids[2, slice_end_idx] = canvas_origin + w_off + slice_w - 1
curr_position_ids[:, slice_start_idx:slice_end_idx] = self.get_vision_position_ids(
start_position=canvas_origin,
grid_thw=target_sizes_slice,
h_offset=h_off,
w_offset=w_off,
spatial_merge_size=merge_factor,
device=device,
)
# The "\n" that ends a row of slices sits just past the right edge of that row.
for k in range(len(slices) - 1):
gap_start_idx, gap_end_idx = slices[k][1], slices[k + 1][0]
if gap_end_idx - gap_start_idx <= 2:
continue
boundary_h = (k // num_cols + 1) * llm_slice_h - 1
right_edge_w = num_cols * llm_slice_w
for newline_idx in range(gap_start_idx + 1, gap_end_idx - 1):
curr_position_ids[1, newline_idx] = canvas_origin + boundary_h
curr_position_ids[2, newline_idx] = canvas_origin + right_edge_w
current_pos = canvas_origin + max(canvas_height, canvas_width) + 1
frame_cursor_idx = frame_end_idx
if frame_cursor_idx < group_end_idx:
trail_len = group_end_idx - frame_cursor_idx
curr_position_ids[:, frame_cursor_idx:group_end_idx] = (
torch.arange(trail_len, device=device) + current_pos
)
current_pos += trail_len
current_idx = group_end_idx
if current_idx < seq_len:
curr_position_ids[:, current_idx:seq_len] = (
torch.arange(seq_len - current_idx, device=device) + current_pos
)
position_ids[:, batch_idx, valid_mask] = curr_position_ids
if attention_mask is not None:
seq_lens = attention_mask.sum(-1)
else:
seq_lens = torch.full((input_ids.shape[0],), input_ids.shape[1], device=input_ids.device, dtype=torch.long)
rope_deltas = position_ids.amax(dim=(0, 2)).unsqueeze(1) + 1 - seq_lens.unsqueeze(1)
return position_ids, rope_deltas.long()
def compute_3d_position_ids(
self,
input_ids: torch.Tensor | None,
inputs_embeds: torch.Tensor | None = None,
attention_mask: torch.Tensor | None = None,
past_key_values: Cache | None = None,
target_sizes: torch.LongTensor | None = None,
target_sizes_videos: torch.LongTensor | None = None,
mm_token_type_ids: torch.IntTensor | None = None,
downsample_mode: str | None = None,
) -> torch.Tensor | None:
"""Return the ``(3, batch, seq)`` canvas M-RoPE positions ``[T, H, W]`` (Qwen-style entrypoint).
Generation prepends the text channel in ``_prepare_position_ids_for_generation``; the text
model reads that fourth row only when it is there.
"""
past_key_values_length = 0 if past_key_values is None else past_key_values.get_seq_length()
has_multimodal = target_sizes is not None or target_sizes_videos is not None
if has_multimodal and mm_token_type_ids is None and input_ids is not None:
raise ValueError(
"Multimodal data was passed (via `target_sizes` or `target_sizes_videos`) but `mm_token_type_ids` "
"is missing. Please pass `mm_token_type_ids` to the model so that multimodal RoPE (M-RoPE) can be "
"computed correctly. `mm_token_type_ids` is returned by the processor alongside `input_ids`."
)
can_compute_mrope = input_ids is not None and mm_token_type_ids is not None and has_multimodal
if can_compute_mrope and (self.rope_deltas is None or past_key_values_length == 0):
mrope_position_ids, rope_deltas = self.get_rope_index(
input_ids,
attention_mask=attention_mask,
target_sizes=target_sizes,
target_sizes_videos=target_sizes_videos,
mm_token_type_ids=mm_token_type_ids,
downsample_mode=downsample_mode,
)
self.rope_deltas = rope_deltas
return mrope_position_ids
elif self.rope_deltas is not None and (past_key_values_length > 0 or input_ids is None):
batch_size, seq_length, _ = inputs_embeds.shape
if attention_mask is not None:
position_ids = attention_mask.long().cumsum(-1) - 1
position_ids = position_ids.masked_fill(attention_mask == 0, 0)
position_ids = position_ids.view(1, batch_size, -1).repeat(3, 1, 1).to(inputs_embeds.device)
else:
position_ids = torch.arange(past_key_values_length, past_key_values_length + seq_length)
position_ids = position_ids.view(1, 1, -1).expand(3, batch_size, -1).to(inputs_embeds.device)
delta = self.rope_deltas.repeat_interleave(batch_size // self.rope_deltas.shape[0], dim=0)
position_ids = position_ids + delta.to(device=inputs_embeds.device)
else:
# Can't build correct 3D positions. Let the model infer it as 1D
if target_sizes_videos is not None or target_sizes is not None:
logger.warning_once(
"Canvas M-RoPE needs `input_ids` to locate the visual spans, but only `inputs_embeds` was "
"given and no cached `rope_deltas` are available. Falling back to 1-D positions, which "
"degrades multimodal quality. Pass `input_ids` for the first forward pass."
)
position_ids = None
return position_ids
@auto_docstring
class MiniCPMV4_7ForConditionalGeneration(MiniCPMV4_7PreTrainedModel, GenerationMixin):
_tied_weights_keys = {"lm_head.weight": "model.language_model.embed_tokens.weight"}
_tp_plan = {"lm_head": "colwise_gather_output"}
def __init__(self, config: MiniCPMV4_7Config):
super().__init__(config)
self.model = MiniCPMV4_7Model(config)
self.vocab_size = config.text_config.vocab_size
self.lm_head = nn.Linear(config.text_config.hidden_size, self.vocab_size, bias=False)
self.post_init()
@can_return_tuple
@auto_docstring
def forward(
self,
input_ids: torch.LongTensor | None = None,
pixel_values: torch.FloatTensor | None = None,
target_sizes: torch.IntTensor | None = None,
pixel_values_videos: torch.FloatTensor | None = None,
target_sizes_videos: torch.IntTensor | None = None,
attention_mask: torch.Tensor | None = None,
position_ids: torch.LongTensor | None = None,
past_key_values: list[torch.FloatTensor] | None = None,
inputs_embeds: torch.FloatTensor | None = None,
labels: torch.LongTensor | None = None,
use_cache: bool | None = None,
downsample_mode: str | None = None,
mm_token_type_ids: torch.IntTensor | None = None,
logits_to_keep: int | torch.Tensor = 0,
mm_encoder_outputs: dict[str, BaseModelOutputWithPooling] | None = None,
**kwargs: Unpack[TransformersKwargs],
) -> tuple | CausalLMOutputWithPast:
r"""
pixel_values (`torch.FloatTensor`, *optional*):
Pixel value patches for images, NaViT-packed.
target_sizes (`torch.IntTensor`, *optional*):
Height and width (in patches) for each image.
pixel_values_videos (`torch.FloatTensor`, *optional*):
Pixel value patches for video frames, NaViT-packed.
target_sizes_videos (`torch.IntTensor`, *optional*):
Height and width (in patches) for each video frame.
downsample_mode (`str`, *optional*):
`"4x"` keeps 4x more visual tokens; default `"16x"` applies full merge.
"""
outputs = self.model(
input_ids=input_ids,
pixel_values=pixel_values,
target_sizes=target_sizes,
pixel_values_videos=pixel_values_videos,
target_sizes_videos=target_sizes_videos,
attention_mask=attention_mask,
position_ids=position_ids,
past_key_values=past_key_values,
inputs_embeds=inputs_embeds,
use_cache=use_cache,
downsample_mode=downsample_mode,
mm_token_type_ids=mm_token_type_ids,
mm_encoder_outputs=mm_encoder_outputs,
**kwargs,
)
hidden_states = outputs[0]
# Only compute necessary logits, and do not upcast them to float if we are not computing the loss
slice_indices = slice(-logits_to_keep, None) if isinstance(logits_to_keep, int) else logits_to_keep
logits = self.lm_head(hidden_states[:, slice_indices, :])
loss = None
if labels is not None:
loss = self.loss_function(
logits=logits, labels=labels, vocab_size=self.config.text_config.vocab_size, **kwargs
)
return CausalLMOutputWithPast(
loss=loss,
logits=logits,
past_key_values=outputs.past_key_values,
hidden_states=outputs.hidden_states,
attentions=outputs.attentions,
)
@auto_docstring(custom_intro="Extract image features: vision encoder, insert merger, then MLP merger.")
def get_image_features(self, *args, **kwargs) -> BaseModelOutputWithPooling:
return self.model.get_image_features(*args, **kwargs)
@auto_docstring(
custom_intro="Extract video features: repack frames into NaViT format, then vision encoder + merger."
)
def get_video_features(self, *args, **kwargs) -> BaseModelOutputWithPooling:
return self.model.get_video_features(*args, **kwargs)
def prepare_inputs_for_generation(
self,
input_ids,
past_key_values=None,
attention_mask=None,
inputs_embeds=None,
pixel_values=None,
target_sizes=None,
pixel_values_videos=None,
target_sizes_videos=None,
downsample_mode=None,
is_first_iteration=False,
**kwargs,
):
model_inputs = super().prepare_inputs_for_generation(
input_ids,
past_key_values=past_key_values,
attention_mask=attention_mask,
inputs_embeds=inputs_embeds,
is_first_iteration=is_first_iteration,
downsample_mode=downsample_mode,
**kwargs,
)
if is_first_iteration or not kwargs.get("use_cache", True):
model_inputs["pixel_values"] = pixel_values
model_inputs["target_sizes"] = target_sizes
model_inputs["pixel_values_videos"] = pixel_values_videos
model_inputs["target_sizes_videos"] = target_sizes_videos
return model_inputs
def _prepare_position_ids_for_generation(self, inputs_tensor, model_kwargs):
# Overwritten -- canvas M-RoPE needs 4D position ids [text, T, H, W].
text_positions = super()._prepare_position_ids_for_generation(inputs_tensor, model_kwargs)
past_length = 0
if (cache := model_kwargs.get("past_key_values")) is not None:
past_length = cache.get_seq_length()
if past_length != 0 and self.model.rope_deltas is not None:
batch_size = text_positions.shape[0]
mrope_positions = text_positions.unsqueeze(0).expand(3, batch_size, -1)
delta = self.model.rope_deltas.repeat_interleave(batch_size // self.model.rope_deltas.shape[0], dim=0)
mrope_positions = mrope_positions + delta.to(device=text_positions.device)
return torch.cat([text_positions.unsqueeze(0), mrope_positions], dim=0)
# `input_ids` is empty when generating from `inputs_embeds`, and canvas M-RoPE needs the ids to locate
# the visual spans, so fall back to plain 1-D text positions whenever they are unavailable.
input_ids = model_kwargs.get("input_ids", inputs_tensor)
has_input_ids = input_ids.dim() == 2 and input_ids.dtype in (torch.int, torch.long) and input_ids.shape[1] > 0
if (
has_input_ids
and (model_kwargs.get("target_sizes") is not None or model_kwargs.get("target_sizes_videos") is not None)
and model_kwargs.get("mm_token_type_ids") is not None
):
mrope_positions, self.model.rope_deltas = self.model.get_rope_index(
input_ids,
attention_mask=model_kwargs.get("attention_mask"),
target_sizes=model_kwargs.get("target_sizes"),
target_sizes_videos=model_kwargs.get("target_sizes_videos"),
mm_token_type_ids=model_kwargs.get("mm_token_type_ids"),
downsample_mode=model_kwargs.get("downsample_mode"),
)
return torch.cat([text_positions.unsqueeze(0), mrope_positions], dim=0)
return text_positions
__all__ = [
"MiniCPMV4_7PreTrainedModel",
"MiniCPMV4_7VisionPreTrainedModel",
"MiniCPMV4_7VisionModel",
"MiniCPMV4_7Model",
"MiniCPMV4_7ForConditionalGeneration",
]