| |
| |
| """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"] |
|
|
