vllm.model_executor.models.transformers.multimodal ¶

class LegacyMultiModalProcessor(_MultiModalProcessorBase):
    """Locates placeholders by searching the prompt the HF processor has already
    expanded for the tokens of each modality.

    Serves transformers versions with no `return_text_replacement_offsets`.
    Placeholders found this way cannot be rebuilt from an unexpanded prompt, so
    this processor overrides `apply` and gets no multi-modal processor cache.
    Remove it once `requirements/common.txt` requires `transformers>=5.15.0`.
    """

    def _get_prompt_updates(
        self,
        mm_items: MultiModalDataItems,
        hf_processor_mm_kwargs: Mapping[str, object],
        out_mm_kwargs: MultiModalKwargsItems,
    ) -> Sequence[PromptUpdate]:
        """Empty, because `apply` writes the placeholder ranges itself rather than
        deriving them from updates."""
        return []

    def _get_mm_token_ids(self, modality: str) -> list[int]:
        """Token ids marking where `modality` sits in the prompt, which for some
        processors differ from the placeholder written into it.

        The expanded prompt is all this path has to go on, so it takes the
        processor at its word about which tokens belong to the modality.
        """
        info = self.info
        processor = info.get_hf_processor()
        declared = getattr(processor, f"{modality}_token_ids", None) or ()
        if ids := [token_id for token_id in declared if token_id is not None]:
            return ids
        config = info.get_hf_config()
        names = (f"{modality}_token_id", f"{modality}_token_index")
        token_id = getattr(processor, names[0], getattr_iter(config, names))
        if token_id is None:
            token = getattr(processor, f"{modality}_token", None)
            token_id = info.get_tokenizer().get_vocab().get(token)
        if token_id is None:
            raise ValueError(
                f"Cannot find {modality}_token_id on processor or model config"
            )
        return [token_id]

    def _apply_audio(
        self,
        prompt_ids: list[int],
        processed_data: "BatchFeature",
        num_audios: int,
    ) -> dict[str, list[PlaceholderRange]]:
        """Take each contiguous run of the audio token as one item's placeholder,
        and record how many tokens the run holds."""
        audio_token_ids = self._get_mm_token_ids("audio")
        prompt_tensor = torch.tensor(prompt_ids)
        is_audio = torch.isin(prompt_tensor, torch.tensor(audio_token_ids))

        if not is_audio.any():
            raise ValueError(
                f"{num_audios} audio item(s) were passed but the prompt "
                "contains no audio token. Add one placeholder per audio item."
            )

        padded = torch.cat([torch.tensor([False]), is_audio, torch.tensor([False])])
        transitions = padded.int().diff()
        offsets = torch.where(transitions == 1)[0]
        lengths = torch.where(transitions == -1)[0] - offsets

        if len(offsets) != num_audios:
            raise ValueError(
                f"Found {len(offsets)} run(s) of the audio token in the prompt but "
                f"{num_audios} audio item(s) were passed. The Transformers backend "
                "locates audio placeholders by finding contiguous runs of the audio "
                "token, so placeholders with no text between them cannot yet be told "
                "apart. Separate them in the prompt to work around this."
            )

        ranges = [
            PlaceholderRange(offset=offset.item(), length=length.item())
            for offset, length in zip(offsets, lengths)
        ]
        processed_data["num_audio_tokens"] = lengths
        return {"audio": ranges}

    def _get_num_multimodal_tokens(
        self,
        mm_items: MultiModalDataItems,
        hf_processor_mm_kwargs: Mapping[str, object],
    ) -> Mapping[str, list[int]]:
        """Ask the HF processor how many tokens and patches each image expands to."""
        processor = self.info.get_hf_processor(**hf_processor_mm_kwargs)
        images = mm_items.get_items("image", ImageProcessorItems)
        image_sizes = [
            (size.height, size.width)
            for size in map(images.get_image_size, range(len(images)))
        ]
        return processor._get_num_multimodal_tokens(
            image_sizes=image_sizes,
            **self.info.ctx.get_merged_mm_kwargs({}),
        )

    def _apply_vision(
        self,
        prompt_ids: list[int],
        processed_data: "BatchFeature",
        mm_items: MultiModalDataItems,
        hf_processor_mm_kwargs: Mapping[str, object],
        mm_token_type_ids: torch.Tensor | None,
    ) -> dict[str, list[PlaceholderRange]]:
        """Split the positions the processor marks as image into one placeholder per
        item, sized by the token count it reports for each image."""
        hf_processor = self.info.get_hf_processor(**hf_processor_mm_kwargs)
        if mm_token_type_ids is None:
            raise ValueError(
                f"{type(hf_processor).__name__} returned no `mm_token_type_ids`, so "
                "the Transformers modeling backend cannot locate the placeholder of "
                "each image."
            )

        # These mark the tokens structuring an image as well as the image itself,
        # which is what the counts below are over
        mm_positions = torch.where(mm_token_type_ids[0] == 1)[0]
        mm_tokens_per_modality = self._get_num_multimodal_tokens(
            mm_items, hf_processor_mm_kwargs
        )

        mm_placeholders: dict[str, list[PlaceholderRange]] = {}
        split_sizes = mm_tokens_per_modality["num_image_tokens"]
        if sum(split_sizes) != len(mm_positions):
            raise ValueError(
                f"The expanded prompt holds {len(mm_positions)} image token(s) but "
                f"{type(hf_processor).__name__} accounts for {sum(split_sizes)} "
                f"across {mm_items.get_count('image')} image item(s)."
            )

        if split_sizes:
            image_token_ids = torch.tensor(self._get_mm_token_ids("image"))
            mm_tokens = torch.tensor(prompt_ids)[mm_positions]
            ranges = [
                PlaceholderRange(
                    offset=positions[0].item(),
                    length=positions.shape[0],
                    # Only some of the span carries embeddings
                    is_embed=torch.isin(tokens, image_token_ids),
                )
                for positions, tokens in zip(
                    torch.split(mm_positions, split_sizes),
                    torch.split(mm_tokens, split_sizes),
                )
            ]
            mm_placeholders = {"image": ranges}

        processed_data["num_image_patches"] = torch.tensor(
            mm_tokens_per_modality["num_image_patches"]
        )
        return mm_placeholders

    def apply(
        self,
        inputs: ProcessorInputs,
        timing_ctx: TimingContext,
    ) -> MultiModalInput:
        """Process the prompt and every multi-modal item in one HF processor call,
        then read the placeholder ranges out of the token ids it returns."""
        prompt = inputs.prompt
        mm_items = inputs.mm_data_items
        hf_processor_mm_kwargs = inputs.hf_processor_mm_kwargs

        with timing_ctx.record("apply_hf_processor"):
            hf_processor = self.info.get_hf_processor(**hf_processor_mm_kwargs)
            prompt_text = hf_processor.decode(prompt)

            # Bypass cached processor and always apply to the full set of mm inputs
            # NOTE: we can't just set caching=False because base class method
            # transforms outputs to `MultiModalKwargs` which is not going to
            # work for Transformers. The vision path has logic tied to
            # `mm_tokens_per_modality` in _apply_vision()
            hf_processor_mm_kwargs = {
                **hf_processor_mm_kwargs,
                # HF processors only accept text, and the decoded string already
                # contains any special tokens, so don't let them be added again
                # (the HF processor call disables `add_special_tokens`).
                "add_special_tokens": False,
            }

            valid_mm_items = mm_items.select(
                {k for k, c in mm_items.get_all_counts().items() if c > 0}
            )
            processor_data, passthrough_data = self._get_hf_mm_data(valid_mm_items)

            # Ask which modality owns each token of the expanded prompt, which
            # is the only marking of the tokens an expansion adds around an item
            if any(processor_data.values()):
                processor_data = {**processor_data, "return_mm_token_type_ids": True}

            try:
                processed_data = self.info.ctx.call_hf_processor(
                    self.info.get_hf_processor(**hf_processor_mm_kwargs),
                    dict(text=prompt_text, **processor_data),
                    hf_processor_mm_kwargs,
                )
            except ValueError:
                if any(processor_data.values()):
                    raise
                # Some processors reject a prompt holding placeholders with
                # no data to go with them, so tokenize it without them
                prompt_ids = self.info.get_tokenizer().encode(prompt_text)
                processed_data = transformers.BatchFeature(
                    dict(input_ids=[prompt_ids]), tensor_type="pt"
                )
            self._unpad_images(processed_data)
            self._unpad_audios(processed_data, processor_data, hf_processor_mm_kwargs)
            processed_data.update(passthrough_data)

            input_ids = processed_data.pop("input_ids")
            if not isinstance(input_ids, list):
                input_ids = input_ids.tolist()

            (prompt_ids,) = input_ids

        # Use overrides if provided; fallback to data-dependent hashing.
        with timing_ctx.record("get_mm_hashes"):
            mm_hashes = inputs.get_mm_hashes(
                self.info.model_id,
                self.info.ctx.get_mm_config().mm_hasher_algorithm,
            )

        # Gemma3 reports them under the key the model uses for its own token types
        mm_token_type_ids = processed_data.pop("token_type_ids", None)
        mm_token_type_ids = processed_data.pop("mm_token_type_ids", mm_token_type_ids)

        mm_placeholders: dict[str, list[PlaceholderRange]] = {}
        if num_audios := mm_items.get_count("audio", strict=False):
            mm_placeholders.update(
                self._apply_audio(prompt_ids, processed_data, num_audios)
            )
        if mm_items.get_count("image", strict=False):
            mm_placeholders.update(
                self._apply_vision(
                    prompt_ids,
                    processed_data,
                    mm_items,
                    hf_processor_mm_kwargs,
                    mm_token_type_ids,
                )
            )

        mm_kwargs = MultiModalKwargsItems.from_hf_inputs(
            processed_data,
            self._get_mm_fields_config(processed_data, hf_processor_mm_kwargs),
        )

        # Bypassing `_maybe_apply_prompt_updates` also bypasses its validation.
        # `_validate_mm_placeholders` can't be reused because it is typed for the
        # `PlaceholderFeaturesInfo` the prompt update machinery produces.
        mm_item_counts = mm_items.get_all_counts()
        self._validate_mm_kwargs(mm_kwargs, mm_item_counts)
        for modality, item_count in mm_item_counts.items():
            num_placeholders = len(mm_placeholders.get(modality, []))
            if num_placeholders != item_count:
                raise RuntimeError(
                    f"Expected there to be {item_count} prompt placeholders "
                    f"corresponding to {item_count} {modality} items, but instead "
                    f"found {num_placeholders} prompt placeholders! Make sure the "
                    "prompt contains a placeholder token for each item."
                )

        return mm_input(
            prompt_token_ids=prompt_ids,
            mm_kwargs=mm_kwargs,
            mm_hashes=mm_hashes,
            mm_placeholders=mm_placeholders,
        )

class MultiModalMixin(SupportsMultiModal, SupportsMRoPE):
    def __init__(self, *, vllm_config: "VllmConfig", prefix: str = ""):
        # Skip SupportsMRoPE.__init__ and call the next class in MRO
        super(SupportsMRoPE, self).__init__(vllm_config=vllm_config, prefix=prefix)

    def _find_encoder_classes(
        self, model: "PreTrainedModel"
    ) -> dict[str, type["PreTrainedModel"]]:
        """Modalities whose encoder cannot be told apart from the model itself are
        omitted, as are those `get_encoder` rejects."""
        encoder_classes: dict[str, type[PreTrainedModel]] = {}
        for modality in _MODALITY_TO_TOKEN_TYPE_ID:
            try:
                encoder_cls = type(model.get_encoder(modality=modality))
            except (TypeError, ValueError):
                continue
            if encoder_cls is not type(model):
                encoder_classes[modality] = encoder_cls
        return encoder_classes

    @contextmanager
    def _mark_model_components(self, vllm_config: "VllmConfig"):
        model_config = vllm_config.model_config
        encoder_classes = self._pre_trained_model_classes.encoders
        if not encoder_classes:
            logger.debug("No encoders identified, so no components will be marked")
            yield
            return

        if model_config.skip_tokenizer_init:
            # Determining the supported modalities needs the HF processor, which in
            # turn needs a tokenizer
            mm_config = model_config.multimodal_config
            if mm_config.mm_encoder_only or any(
                mm_config.get_limit_per_prompt(modality) == 0
                for modality in encoder_classes
            ):
                logger.warning_once(
                    "Unable to determine the supported modalities without a "
                    "tokenizer, so no model components will be skipped."
                )
            yield
            return

        # Modalities we don't serve report a limit of 999, which would stop their
        # encoder ever being skipped
        supported_modalities = MULTIMODAL_REGISTRY.get_processing_info(
            model_config
        ).supported_mm_limits

        # One encoder often serves several modalities, and may only be skipped when
        # all of them are disabled, so mark it once for the whole set
        modalities_by_encoder = defaultdict(set)
        for modality, encoder_cls in encoder_classes.items():
            if modality in supported_modalities:
                modalities_by_encoder[encoder_cls].add(modality)

        with ExitStack() as stack:
            stack.enter_context(
                self._mark_language_model(
                    vllm_config, targets=self._pre_trained_model_classes.decoder
                )
            )
            for encoder_cls, modalities in modalities_by_encoder.items():
                stack.enter_context(
                    self._mark_tower_model(vllm_config, modalities, targets=encoder_cls)
                )
            yield

    def _decorate_for_torch_compile(self):
        """
        Decorate the model's decoder and encoder classes to indicate to vLLM that they
        support torch compile if `can_enable_torch_compile` and
        `should_torch_compile_mm_encoder` are True respectively.
        """
        super()._decorate_for_torch_compile()
        # Decorate the encoder model classes to support torch compile if needed
        if self.compilation_config.compile_mm_encoder:
            self.check_version("5.0.0", "multimodal encoder compilation support")
            encoder_classes = self._pre_trained_model_classes.encoders
            if not encoder_classes:
                raise ValueError(
                    "Unable to infer any encoder classes from the model. "
                    "You must either: update the model so that "
                    "https://huggingface.co/docs/transformers/en/main_classes/model#transformers.PreTrainedModel.get_encoder"
                    " can detect the encoders correctly, or remove "
                    "'compile_mm_encoder'."
                )
            logger.warning_once(
                "Multimodal encoder compilation with the Transformers modeling backend "
                "is an experimental feature. It relies on:\n"
                "- The encoder being torch compilable.\n"
                "- All encoder tensor inputs must be type hinted as either "
                "`torch.Tensor` or `torch.FloatTensor`.\n"
                "- The 0-th dimension of all tensor inputs to the encoder being the "
                "dynamic dimension (e.g. sequence length, number of patches).\n"
                "Please report any issues you encounter to help us improve it."
            )
            # One encoder can serve several modalities, and must only be decorated once
            for encoder_cls in dict.fromkeys(encoder_classes.values()):
                self._decorate_cls_for_torch_compile(
                    cls=encoder_cls,
                    # TODO: properly infer dynamic_arg_dims based on the encoder's
                    # forward method signature. We assume dim 0 for all tensor inputs.
                    dynamic_arg_dims=None,
                    enable_if=should_torch_compile_mm_encoder,
                    is_encoder=True,
                )

    def forward(
        self,
        input_ids: torch.Tensor | None,
        positions: torch.Tensor,
        intermediate_tensors: IntermediateTensors | None = None,
        inputs_embeds: torch.Tensor | None = None,
        **kwargs: object,
    ) -> torch.Tensor | IntermediateTensors:
        # Positions shape handling for MRoPE models
        if self.model_config.uses_mrope:
            # [3, seq_len] -> [3, 1, seq_len]
            positions = positions[:, None].contiguous()
        model_output = super().forward(
            input_ids, positions, intermediate_tensors, inputs_embeds
        )
        return model_output

    def get_language_model(self) -> torch.nn.Module:
        """Transformers modeling backend multimodal classes do not contain a separate
        vLLM language model class. Therefore, in order to return a language model vLLM
        class, we use a wrapper to give `self` the same interface as a text model."""

        # Exclude self and object
        bases = self.__class__.mro()[1:-1]
        # Keep only classes defined in `vllm.model_executor.models.transformers`
        bases = [b for b in bases if ".transformers." in b.__module__]
        # Exclude MultiModalMixin itself
        bases = [b for b in bases if b is not MultiModalMixin]

        class LanguageModel(*bases):
            def __init__(self, multimodal_model):
                # Don't call super().__init__() to avoid re-initialization
                self.__dict__.update(multimodal_model.__dict__)

            model = getattr_iter(self.model, ("language_model", "text_model"), None)

        return LanguageModel(self)

    def get_mm_mapping(self) -> MultiModelKeys:
        """
        Get the module prefix in multimodal models
        """
        for name in ("language_model", "text_model"):
            if getattr(self.model, name, None) is not None:
                return MultiModelKeys.from_string_field(language_model=f"model.{name}")
        raise ValueError(
            "Could not locate the language model submodule for LoRA support"
        )

    def _split_embeddings(
        self, embeddings: torch.Tensor, split_sizes: list[int]
    ) -> list[torch.Tensor]:
        total_expected = sum(split_sizes)

        # Flatten to 2D: [total_tokens, hidden_dim]
        if embeddings.ndim > 2:
            embeddings = embeddings.reshape(-1, embeddings.shape[-1])

        total_tokens = embeddings.shape[0]
        if total_tokens == total_expected:
            # Direct match: split_sizes are actual token counts
            token_split_sizes = split_sizes
        elif total_expected > 0 and total_tokens % total_expected == 0:
            # Uniform expansion: each item expands to N tokens
            tokens_per_item = total_tokens // total_expected
            token_split_sizes = [s * tokens_per_item for s in split_sizes]
        elif total_expected > 0:
            # TODO: make this an error once we know profiling never relies on it
            if total_tokens == 0:
                raise ValueError(
                    "Encoder returned empty embeddings. "
                    f"Expected {total_expected} tokens from "
                    f"split_sizes={split_sizes}"
                )
            # Keep the counts out of the message: `warning_once` keys its cache on
            # the args, so varying them would log on every new pair
            logger.warning_once(
                "Encoder returned a different number of tokens than expected; "
                "padding or truncating to fit. The embeddings are not trustworthy "
                "outside of memory profiling."
            )
            logger.debug(
                "Encoder returned %s tokens but %s were expected",
                total_tokens,
                total_expected,
            )
            if total_tokens < total_expected:
                repeat_factor = (total_expected + total_tokens - 1) // total_tokens
                embeddings = embeddings.repeat(repeat_factor, 1)
            embeddings = embeddings[:total_expected]
            token_split_sizes = split_sizes
        else:
            return []

        return list(torch.split(embeddings, token_split_sizes, dim=0))

    def _process_audio_input(self, **kwargs) -> list[torch.Tensor] | None:
        input_features: torch.Tensor | None = kwargs.pop("input_features", None)
        if input_features is None:
            input_features = kwargs.pop("input_values", None)
        if input_features is None:
            return None

        self.check_version("5.13.0", "audio models support")
        num_audio_tokens = kwargs.pop("num_audio_tokens")
        kwargs.pop("token_type_ids", None)
        kwargs.pop("mm_token_type_ids", None)

        # Per-audio token counts are needed as Python ints to split.
        with gpu_sync_allowed():
            split_sizes = num_audio_tokens.flatten().tolist()
        if isinstance(input_features, torch.Tensor):
            # HuggingFace's `get_audio_features` implementations branch on
            # per-sample feature lengths internally.
            with gpu_sync_allowed():
                audio_output = self.model.get_audio_features(
                    input_features, return_dict=True, **kwargs
                )
            return self._split_embeddings(audio_output.pooler_output, split_sizes)

        # Audios the processor left un-padded arrive as a list once their
        # lengths differ. Encode them one at a time so that none of them is
        # padded to match another.
        embeddings: list[torch.Tensor] = []
        for index, features in enumerate(input_features):
            audio_output = self.model.get_audio_features(
                features.unsqueeze(0),
                return_dict=True,
                **self._select_item_kwargs(kwargs, index, len(input_features)),
            )
            embeddings.extend(
                self._split_embeddings(audio_output.pooler_output, [split_sizes[index]])
            )
        return embeddings

    def _process_image_input(self, **kwargs) -> list[torch.Tensor] | None:
        pixel_values: torch.Tensor | None = kwargs.pop("pixel_values", None)
        image_embeds: torch.Tensor | None = kwargs.pop("image_embeds", None)
        # Model might use `image_patches` instead of `pixel_values`
        if pixel_values is None:
            pixel_values = kwargs.pop("image_patches", None)

        if image_embeds is not None:
            return [image_embeds]

        if pixel_values is None:
            return None

        num_image_patches = kwargs.pop("num_image_patches")

        split_sizes = num_image_patches.flatten().tolist()
        if isinstance(pixel_values, torch.Tensor):
            vision_embeddings = self._get_image_features(pixel_values, **kwargs)
            if isinstance(vision_embeddings, torch.Tensor):
                return self._split_embeddings(vision_embeddings, split_sizes)
            return list(vision_embeddings)

        # Images the processor left un-padded arrive as a list once their
        # shapes differ. Encode them one at a time so that none of them is
        # padded to match another.
        embeddings: list[torch.Tensor] = []
        for index, image in enumerate(pixel_values):
            features = self._get_image_features(
                image.unsqueeze(0),
                **self._select_item_kwargs(kwargs, index, len(pixel_values)),
            )
            # Encoders which return one entry per image return a single entry
            if not isinstance(features, torch.Tensor):
                features = torch.cat(list(features))
            embeddings.extend(self._split_embeddings(features, [split_sizes[index]]))
        return embeddings

    def _select_item_kwargs(
        self, kwargs: dict[str, Any], index: int, num_items: int
    ) -> dict[str, Any]:
        """Narrow the entries of `kwargs` that hold one row per item down to the item
        at `index`. Length is all there is to match on, so an unrelated entry of the
        same length is narrowed too."""
        return {
            key: value[index : index + 1]
            if isinstance(value, (torch.Tensor, list)) and len(value) == num_items
            else value
            for key, value in kwargs.items()
        }

    def _get_image_features(self, pixel_values: torch.Tensor, **kwargs) -> Any:
        # grid_thw fields are registered keep_on_cpu; restore the on-device
        # placement that HF get_image_features implementations expect.
        for key, value in kwargs.items():
            if isinstance(value, torch.Tensor):
                kwargs[key] = value.to(pixel_values.device, non_blocking=True)

        # The underlying HuggingFace `get_image_features` implementations
        # contain model-internal syncs (e.g. Idefics3 filters all-zero
        # padding images via boolean-mask indexing, LlavaOnevision
        # branches on per-sample batch counts).
        with gpu_sync_allowed():
            features = self.model.get_image_features(pixel_values, **kwargs)

        # Transformers `v5`, `self.get_image_features` returns a tuple
        # containing the features and optionally attentions/hidden_states
        # After v5 is settled, we can enable qwen3-vl with several outputs
        # from `self.get_image_features`
        if isinstance(features, tuple):
            return features[0]
        if isinstance(features, dict):
            return features.pooler_output
        return features

    def embed_multimodal(self, **kwargs) -> MultiModalEmbeddings:
        # Each helper detects its own inputs. We are called once per modality, so the
        # leftovers a helper forwards to the HF model can't belong to the other one.
        embeddings: list[torch.Tensor] = []
        for process_input in (self._process_audio_input, self._process_image_input):
            embeddings.extend(process_input(**kwargs) or [])
        return embeddings

    def get_mrope_input_positions(
        self,
        input_tokens: list[int],
        mm_features: list[MultiModalFeatureSpec],
    ) -> tuple[torch.Tensor, int]:
        kwargs = MultiModalFeatureSpec.gather_kwargs(
            mm_features,
            {
                "image_grid_thw",
                "video_grid_thw",
                "second_per_grid_ts",
                "audio_feature_lengths",
                "use_audio_in_video",
            },
        )
        if any(v for k, v in kwargs.items() if k not in {"image_grid_thw"}):
            raise NotImplementedError(
                "Transformers modeling backend only supports images."
            )

        image_grid_thw = kwargs.get("image_grid_thw", [])
        video_grid_thw = kwargs.get("video_grid_thw", [])

        image_grid_thw = torch.stack(image_grid_thw) if image_grid_thw else None
        video_grid_thw = torch.stack(video_grid_thw) if video_grid_thw else None

        # `get_rope_index` doesn't always accept arbitrary `kwargs`
        kwargs = {}
        if not hasattr(self, "_get_rope_index_accepts_mm_token_type_ids"):
            import inspect

            sig = inspect.signature(self.model.get_rope_index)
            params = sig.parameters
            self._get_rope_index_accepts_mm_token_type_ids = (
                "mm_token_type_ids" in params
                or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
            )
        if self._get_rope_index_accepts_mm_token_type_ids:
            mm_token_type_ids = torch.zeros(len(input_tokens), dtype=torch.int)
            for feature in mm_features:
                position = feature.mm_position
                offset, length = position.offset, position.length
                mm_token_type_id = _MODALITY_TO_TOKEN_TYPE_ID[feature.modality]
                mm_token_type_ids[offset : offset + length] = mm_token_type_id
            kwargs["mm_token_type_ids"] = mm_token_type_ids.unsqueeze(0)

        mrope_positions, mrope_position_delta = self.model.get_rope_index(
            input_ids=torch.tensor(input_tokens).unsqueeze(0),
            image_grid_thw=image_grid_thw,
            video_grid_thw=video_grid_thw,
            **kwargs,
        )

        mrope_positions = mrope_positions[:, 0]
        mrope_position_delta = mrope_position_delta[0].item()

        return mrope_positions, mrope_position_delta