# Copyright 2026 The HuggingFace Team. All rights reserved. # Licensed under the Apache License, Version 2.0 """Processor for LFM3-VL using Qwen visual preprocessing and LFM3 text tokenization.""" import numpy as np from transformers.feature_extraction_utils import BatchFeature from transformers.image_utils import ImageInput from transformers.processing_utils import ProcessingKwargs, ProcessorMixin, Unpack from transformers.tokenization_utils_base import PreTokenizedInput, TextInput from transformers.utils import logging from transformers.video_utils import VideoInput logger = logging.get_logger(__name__) IMAGE_TOKEN = "<|image|>" VIDEO_TOKEN = "<|video|>" VISION_START_TOKEN = "<|vision_start|>" VISION_END_TOKEN = "<|vision_end|>" IMAGE_TOKEN_ID = 125030 VIDEO_TOKEN_ID = 125029 VISION_START_TOKEN_ID = 125009 VISION_END_TOKEN_ID = 125010 class Lfm3VlProcessorKwargs(ProcessingKwargs, total=False): _defaults = { "text_kwargs": { "padding": False, "return_token_type_ids": False, "return_mm_token_type_ids": False, }, "videos_kwargs": {"return_metadata": True}, } class Lfm3VlProcessor(ProcessorMixin): """Expand Qwen-style visual spans around Qwen-preprocessed visual inputs.""" def __init__(self, image_processor=None, tokenizer=None, video_processor=None, chat_template=None, **kwargs): self.image_token = IMAGE_TOKEN self.video_token = VIDEO_TOKEN self.vision_start_token = VISION_START_TOKEN self.vision_end_token = VISION_END_TOKEN self.image_token_id = tokenizer.convert_tokens_to_ids(self.image_token) self.video_token_id = tokenizer.convert_tokens_to_ids(self.video_token) self.vision_start_token_id = tokenizer.convert_tokens_to_ids(self.vision_start_token) self.vision_end_token_id = tokenizer.convert_tokens_to_ids(self.vision_end_token) actual_token_ids = ( self.image_token_id, self.video_token_id, self.vision_start_token_id, self.vision_end_token_id, ) expected_token_ids = (IMAGE_TOKEN_ID, VIDEO_TOKEN_ID, VISION_START_TOKEN_ID, VISION_END_TOKEN_ID) if actual_token_ids != expected_token_ids: raise ValueError( "LFM3-VL requires the released multimodal tokenizer with " f"{IMAGE_TOKEN}={IMAGE_TOKEN_ID}, {VIDEO_TOKEN}={VIDEO_TOKEN_ID}, " f"{VISION_START_TOKEN}={VISION_START_TOKEN_ID}, and {VISION_END_TOKEN}={VISION_END_TOKEN_ID}; " f"got {actual_token_ids}." ) super().__init__(image_processor, tokenizer, video_processor, chat_template=chat_template, **kwargs) def __call__( self, images: ImageInput = None, text: TextInput | PreTokenizedInput | list[TextInput] | list[PreTokenizedInput] = None, videos: VideoInput = None, **kwargs: Unpack[Lfm3VlProcessorKwargs], ) -> BatchFeature: output_kwargs = self._merge_kwargs( Lfm3VlProcessorKwargs, tokenizer_init_kwargs=self.tokenizer.init_kwargs, **kwargs, ) image_inputs = ( self.image_processor(images=images, **output_kwargs["images_kwargs"]) if images is not None else {} ) video_inputs = ( self.video_processor(videos=videos, **output_kwargs["videos_kwargs"]) if videos is not None else {} ) image_grid_thw = image_inputs.get("image_grid_thw") video_grid_thw = video_inputs.get("video_grid_thw") video_metadata = video_inputs.get("video_metadata") if not isinstance(text, list): text = [text] text = text.copy() if image_grid_thw is not None: image_index = 0 for sample_index in range(len(text)): while self.image_token in text[sample_index]: token_count = int(image_grid_thw[image_index].prod()) // self.image_processor.merge_size**2 text[sample_index] = text[sample_index].replace( self.image_token, "<|placeholder|>" * token_count, 1 ) image_index += 1 text[sample_index] = text[sample_index].replace("<|placeholder|>", self.image_token) if video_grid_thw is not None: video_index = 0 for sample_index in range(len(text)): while self.video_token in text[sample_index]: metadata = video_metadata[video_index] if metadata.fps is None: logger.warning_once("Video FPS is missing; defaulting to 24 FPS for timestamps.") metadata.fps = 24 timestamps = self._calculate_timestamps( metadata.frames_indices, metadata.fps, self.video_processor.temporal_patch_size, ) frame_tokens = int(video_grid_thw[video_index][1:].prod()) // self.video_processor.merge_size**2 placeholder = "".join( f"<{timestamp:.1f} seconds>" + self.vision_start_token + "<|placeholder|>" * frame_tokens + self.vision_end_token for timestamp in timestamps ) text[sample_index] = text[sample_index].replace(self.video_token, placeholder, 1) video_index += 1 text[sample_index] = text[sample_index].replace("<|placeholder|>", self.video_token) return_tensors = output_kwargs["text_kwargs"].pop("return_tensors", None) return_mm_token_type_ids = output_kwargs["text_kwargs"].pop("return_mm_token_type_ids", False) text_inputs = self.tokenizer(text, **output_kwargs["text_kwargs"]) self._check_special_mm_tokens(text, text_inputs, modalities=["image", "video"]) if return_mm_token_type_ids: input_ids = np.asarray(text_inputs["input_ids"]) mm_token_type_ids = np.zeros_like(input_ids) mm_token_type_ids[input_ids == self.image_token_id] = 1 mm_token_type_ids[input_ids == self.video_token_id] = 2 text_inputs["mm_token_type_ids"] = mm_token_type_ids.tolist() video_inputs.pop("video_metadata", None) return BatchFeature( data={**text_inputs, **image_inputs, **video_inputs}, tensor_type=return_tensors, ) @staticmethod def _calculate_timestamps(indices, video_fps, temporal_patch_size=2): indices = indices.tolist() if not isinstance(indices, list) else indices.copy() if len(indices) % temporal_patch_size: indices.extend(indices[-1] for _ in range(temporal_patch_size - len(indices) % temporal_patch_size)) return [ (indices[index] + indices[index + temporal_patch_size - 1]) / (2 * video_fps) for index in range(0, len(indices), temporal_patch_size) ] __all__ = ["Lfm3VlProcessor"]