File size: 7,144 Bytes
f85d8ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | # 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"]
|