import copy import math import os from typing import Dict, List, Optional, Union import numpy as np import torch from PIL import Image from transformers import Qwen2_5_VLProcessor from transformers.image_processing_utils import ( BaseImageProcessor, BatchFeature, get_size_dict, ) from transformers.image_transforms import ( convert_to_rgb, get_resize_output_image_size, resize, to_channel_dimension_format, ) from transformers.image_utils import ( OPENAI_CLIP_MEAN, OPENAI_CLIP_STD, ChannelDimension, ImageInput, PILImageResampling, get_image_size, infer_channel_dimension_format, is_scaled_image, make_list_of_images, to_numpy_array, valid_images, ) from transformers.models.qwen2_5_vl.processing_qwen2_5_vl import ( Qwen2_5_VLProcessorKwargs, ) from transformers.tokenization_utils_base import PreTokenizedInput, TextInput from transformers.utils import TensorType, logging from transformers.video_utils import VideoInput from typing_extensions import Unpack logger = logging.get_logger(__name__) def determine_possible_resolutions(anyres: bool, max_num_grids: int, grid_size: int, use_1x1_grid: bool = False): """총 max_num_grids 이하의 possible resolution 조합을 찾아 반환합니다. max_num_grids 가 예를 들어 4인 경우, 총 가능한 grid 조합은 [1x1, 1x2, 1x3, 1x4, 2x1, 2x2, 3x1, 4x1] 이고, 따라서 아래와 같이 계산됩니다. >>> possible_resolutions = determine_possible_resolutions(anyres=True, max_num_grids=4, grid_size=336) >>> print(possible_resolutions) [[336, 336], [336, 672], [336, 1008], [336, 1344], [672, 336], [672, 672], [1008, 336], [1344, 336]] """ possible_resolutions = [] if anyres: assert max_num_grids > 0 for i in range(1, max_num_grids + 1): for j in range(1, max_num_grids + 1): if i == 1 and j == 1 and not use_1x1_grid: continue if i * j <= max_num_grids: possible_resolutions.append([i, j]) possible_resolutions = [[ys * grid_size, xs * grid_size] for ys, xs in possible_resolutions] return possible_resolutions def divide_to_grids(image: np.array, grid_size: int, input_data_format=None) -> List[np.array]: """local image 를 (grid_size x grid_size) grid 로 divide""" grids = [] height, width = get_image_size(image, channel_dim=input_data_format) for i in range(0, height, grid_size): for j in range(0, width, grid_size): if input_data_format == ChannelDimension.LAST: grid = image[i : i + grid_size, j : j + grid_size] else: grid = image[:, i : i + grid_size, j : j + grid_size] grids.append(grid) return grids def pad(image: np.array, target_size: tuple, background_color=(127, 127, 127), input_data_format=None) -> np.array: """image 양옆, 좌우에 padding 을 하여 target_height, target_width 만큼 키움""" target_height, target_width = target_size height, width = get_image_size(image, channel_dim=input_data_format) # result = np.ones((target_height, target_width, image.shape[2]), dtype=image.dtype) * background_color result = np.empty((target_height, target_width, image.shape[2]), dtype=image.dtype) for i in range(image.shape[2]): result[..., i].fill(background_color[i]) paste_x = (target_width - width) // 2 paste_y = (target_height - height) // 2 result[paste_y : paste_y + height, paste_x : paste_x + width, :] = image return result def expand2square( image: np.array, bboxes_dict=None, background_color=(127, 127, 127), input_data_format=None ) -> np.array: """ 새로운 canvas 를 만들어 두고, 거기에 이미지를 붙여넣는 방식으로 이미지를 정사각형으로 만드는 함수 유의할 사항은, 이미지를 붙여 넣을 때 중앙으로 붙여넣는다는 점. 양옆 또는 위아래로 PADDING 이 들어가는 형태 Args: pil_img: numpy array bboxes_dict: dict, {"ocr": NDArray shape (N, 4, 2), "html": NDArray shape (N, 4, 2), ... } `[[xtl, ytl], [xtr, ytr], [xbr, ybr], [xbl, ybl]]` 형태로 박스 형태는 통일. OCR, HTML 등 다양한 박스들을 한번에 처리 가능 background_color: tuple, RGB # >>> _img = np.ones((80, 100), dtype=np.uint8) * 100 # >>> _bboxes_dict = {"words": np.array([[[10, 10], [20, 10], [20, 20], [10, 20]], # ... [[30, 30], [40, 30], [40, 40], [30, 40]]])} # >>> _img, _bboxes_dict = expand2square(_img, _bboxes_dict, (255, 255, 255)) # >>> _img.shape # (100, 100) # >>> guessed_ocr_bboxes = np.array([[[20, 10], [30, 10], [30, 20], [20, 20]], # ... [[40, 30], [50, 30], [50, 40], [40, 40]]]) # >>> np.testing.assert_array_almost_equal(_bboxes_dict["words"], guessed_ocr_bboxes) is None # True """ height, width = get_image_size(image, channel_dim=input_data_format) if width == height: return image, bboxes_dict elif width > height: # result = np.ones((width, width, image.shape[2]), dtype=image.dtype) * background_color result = np.empty((width, width, image.shape[2]), dtype=image.dtype) for i in range(image.shape[2]): result[..., i].fill(background_color[i]) result[(width - height) // 2 : (width - height) // 2 + height, :] = image if bboxes_dict is not None: for key in bboxes_dict: bboxes_dict[key][:, :, 1] += (width - height) // 2 return result, bboxes_dict else: # result = np.ones((height, height, image.shape[2]), dtype=image.dtype) * background_color result = np.empty((height, height, image.shape[2]), dtype=image.dtype) for i in range(image.shape[2]): result[..., i].fill(background_color[i]) result[:, (height - width) // 2 : (height - width) // 2 + width] = image if bboxes_dict is not None: for key in bboxes_dict: bboxes_dict[key][:, :, 0] += (height - width) // 2 return result, bboxes_dict def resize_longside( image: np.array, size: int, resample: PILImageResampling = PILImageResampling.BICUBIC, data_format: Optional[Union[str, ChannelDimension]] = None, input_data_format: Optional[Union[str, ChannelDimension]] = None, ): """ 장축 길이를 size 에 맞게 resize """ height, width = get_image_size(image, channel_dim=input_data_format) if width == height: target_height, target_width = size, size elif width > height: target_width = size target_height = math.ceil(height / width * size) else: target_width = math.ceil(width / height * size) target_height = size return resize( image, size=(target_height, target_width), resample=resample, data_format=data_format, input_data_format=input_data_format, ) def select_best_resolution(original_size: tuple, possible_resolutions: list) -> tuple: """From LLaVA-Next (https://github.com/huggingface/transformers/blob/v4.40.2/src/transformers/models/llava_next/image_processing_llava_next.py) Selects the best resolution from a list of possible resolutions based on the original size. This is done by calculating the effective and wasted resolution for each possible resolution. The best fit resolution is the one that maximizes the effective resolution and minimizes the wasted resolution. Args: original_size (tuple): The original size of the image in the format (height, width). possible_resolutions (list): A list of possible resolutions in the format [(height1, width1), (height2, width2), ...]. Returns: tuple: The best fit resolution in the format (height, width). """ original_height, original_width = original_size best_fit = None max_effective_resolution = 0 min_wasted_resolution = float("inf") for height, width in possible_resolutions: scale = min(width / original_width, height / original_height) downscaled_width, downscaled_height = int(original_width * scale), int(original_height * scale) effective_resolution = min(downscaled_width * downscaled_height, original_width * original_height) wasted_resolution = (width * height) - effective_resolution if effective_resolution > max_effective_resolution or ( effective_resolution == max_effective_resolution and wasted_resolution < min_wasted_resolution ): max_effective_resolution = effective_resolution min_wasted_resolution = wasted_resolution best_fit = (height, width) return best_fit def _get_local_grids_output_size(image: np.array, target_resolution: tuple, input_data_format=None): original_height, original_width = get_image_size(image, channel_dim=input_data_format) target_height, target_width = target_resolution scale_w = target_width / original_width scale_h = target_height / original_height if scale_w < scale_h: new_width = target_width new_height = min(math.ceil(original_height * scale_w), target_height) else: new_height = target_height new_width = min(math.ceil(original_width * scale_h), target_width) return new_height, new_width def determine_anyres_num_vision_patches( num_grids, image_size, grid_size, patch_size, possible_resolutions, anyres=False, unpad=True, num_queries_vis_abstractor=0, num_queries_vis_abstractor_slow=0, video=False, first_last_frames_slow=False, is_first_or_last_frames=False, ): """visual tokens 수를 계산해주는 함수""" if not anyres: return num_queries_vis_abstractor if num_queries_vis_abstractor > 0 else (grid_size // patch_size) ** 2 if num_queries_vis_abstractor > 0: num_patch_per_grid = int(num_queries_vis_abstractor**0.5) else: num_patch_per_grid = grid_size // patch_size num_global_per_grid = num_patch_per_grid # anyres는 global image가 있어서 2개 이상이지만, video에는 global image가 없어서, 1개가 들어올 수 있어서 주석 처리 # assert num_grids > 1 # patch 수 계산 height, width = select_best_resolution(image_size, possible_resolutions) num_patch_height = (height // grid_size) * num_patch_per_grid num_patch_width = (width // grid_size) * num_patch_per_grid # local images if unpad: original_height, original_width = image_size original_aspect_ratio = original_width / original_height current_aspect_ratio = num_patch_width / num_patch_height if original_aspect_ratio > current_aspect_ratio: scale_factor = num_patch_width / original_width new_height = int(original_height * scale_factor) padding = (num_patch_height - new_height) // 2 num_patch_height = num_patch_height - padding * 2 else: scale_factor = num_patch_height / original_height new_width = int(original_width * scale_factor) padding = (num_patch_width - new_width) // 2 num_patch_width = num_patch_width - padding * 2 num_patches = num_patch_width * num_patch_height + num_patch_height else: num_patches = num_patch_width * num_patch_height # slow는 첫프레임 마지막 프레임 적용 전략일때는 첫프레임과 마지막 프레임만 적용 if num_queries_vis_abstractor_slow > 0: if first_last_frames_slow: if is_first_or_last_frames: num_patches += num_queries_vis_abstractor_slow - num_queries_vis_abstractor else: num_patches += num_queries_vis_abstractor_slow - num_queries_vis_abstractor # slowfast 기능은 unpad False 에만 적용 assert unpad is False # video 에는 global image 가 포함되지 않음 if not video: num_patches += num_global_per_grid**2 return num_patches class HCXVisionImageProcessor(BaseImageProcessor): r""" Constructs a VLM image processor. Based on [`CLIPImageProcessor`] with incorporation of additional techniques for processing high resolution images. Args: anyres: (bool) anyres 기능을 사용할지 안할지 unpad: (bool) anyres 사용시, unpad 기능 (순수 pad 영역에 해당하는 visual tokens 은 LLM input 에서 제거) 을 사용할지 안할지 num_queries_vis_abstractor: (int) 각 grid 에 대해서 resampler 를 사용하는 경우, visual query 수 possible_resolutions: (List) anyres 기능 사용시, 가능한 resolution 조합, 예: [[336, 336], [336, 672], [672, 336]] patch_size: (int) ViT patch size pad_to_square: (bool) 정사각형으로 padding 을 수행할지, 안할지를 결정. False 이면 정사각형이 아니기 때문에 center crop 을 거쳐 ViT 의 입력으로 들어감 """ model_input_names = ["pixel_values"] def __init__( self, do_resize: bool = True, size: Dict[str, int] = None, anyres: bool = False, unpad: bool = False, num_queries_vis_abstractor: int = 0, possible_resolutions: List = [], patch_size: int = 14, pad_to_square: bool = True, resample: PILImageResampling = PILImageResampling.BICUBIC, do_center_crop: bool = True, crop_size: Dict[str, int] = None, do_rescale: bool = True, rescale_factor: Union[int, float] = 1 / 255, do_normalize: bool = True, image_mean: Optional[Union[float, List[float]]] = None, image_std: Optional[Union[float, List[float]]] = None, do_convert_rgb: bool = True, **kwargs, ) -> None: super().__init__(**kwargs) size = size if size is not None else {"shortest_edge": 336} size = get_size_dict(size, default_to_square=False) crop_size = crop_size if crop_size is not None else {"height": 336, "width": 336} crop_size = get_size_dict(crop_size, default_to_square=True, param_name="crop_size") self.do_resize = do_resize self.size = size self.anyres = anyres self.unpad = unpad self.num_queries_vis_abstractor = num_queries_vis_abstractor self.possible_resolutions = [_resolution for _resolution in possible_resolutions] self.patch_size = patch_size self.pad_to_square = pad_to_square self.resample = resample self.do_center_crop = do_center_crop self.crop_size = crop_size self.do_rescale = do_rescale self.rescale_factor = rescale_factor self.do_normalize = do_normalize self.image_mean = image_mean if image_mean is not None else OPENAI_CLIP_MEAN self.image_std = image_std if image_std is not None else OPENAI_CLIP_STD self.do_convert_rgb = do_convert_rgb def resize( self, image: np.ndarray, size: Dict[str, int], resample: PILImageResampling = PILImageResampling.BICUBIC, data_format: Optional[Union[str, ChannelDimension]] = None, input_data_format: Optional[Union[str, ChannelDimension]] = None, **kwargs, ) -> np.ndarray: default_to_square = True if "shortest_edge" in size: size = size["shortest_edge"] default_to_square = False elif "height" in size and "width" in size: size = (size["height"], size["width"]) else: raise ValueError("Size must contain either 'shortest_edge' or 'height' and 'width'.") output_size = get_resize_output_image_size( image, size=size, default_to_square=default_to_square, input_data_format=input_data_format, ) return resize( image, size=output_size, resample=resample, data_format=data_format, input_data_format=input_data_format, **kwargs, ) def _preprocess( self, images: ImageInput, do_resize: bool = None, size: Dict[str, int] = None, resample: PILImageResampling = None, do_center_crop: bool = None, crop_size: int = None, do_rescale: bool = None, rescale_factor: float = None, do_normalize: bool = None, image_mean: Optional[Union[float, List[float]]] = None, image_std: Optional[Union[float, List[float]]] = None, data_format: Optional[ChannelDimension] = ChannelDimension.FIRST, input_data_format: Optional[Union[str, ChannelDimension]] = None, ) -> Image.Image: images = make_list_of_images(images) if do_resize: images = [ self.resize(image=image, size=size, resample=resample, input_data_format=input_data_format) for image in images ] if do_center_crop: images = [ self.center_crop(image=image, size=crop_size, input_data_format=input_data_format) for image in images ] if do_rescale: images = [ self.rescale(image=image, scale=rescale_factor, input_data_format=input_data_format) for image in images ] if do_normalize: images = [ self.normalize(image=image, mean=image_mean, std=image_std, input_data_format=input_data_format) for image in images ] images = [ to_channel_dimension_format(image, data_format, input_channel_dim=input_data_format) for image in images ] return images def _resize_for_local_grids( self, image: np.array, target_resolution: tuple, resample, input_data_format: ChannelDimension ) -> np.array: new_height, new_width = _get_local_grids_output_size(image, target_resolution, input_data_format) # Resize the image resized_image = resize(image, (new_height, new_width), resample=resample, input_data_format=input_data_format) return resized_image def _pad_for_patching( self, image: np.array, target_resolution: tuple, input_data_format: ChannelDimension ) -> np.array: """ Pad an image to a target resolution while maintaining aspect ratio. """ target_height, target_width = target_resolution background_color = tuple(int(x * 255) for x in self.image_mean) padded_image = pad( image, target_size=(target_height, target_width), background_color=background_color, input_data_format=input_data_format, ) return padded_image def get_image_grids( self, image: np.array, possible_resolutions, grid_size: int, resample: PILImageResampling, data_format: ChannelDimension, input_data_format: ChannelDimension, ) -> List[np.array]: if not isinstance(possible_resolutions, list): raise ValueError("possible_resolutions must be a list of possible resolutions.") image_size = get_image_size(image, channel_dim=input_data_format) best_resolution = select_best_resolution(image_size, possible_resolutions) resized_image = self._resize_for_local_grids( image, best_resolution, resample=resample, input_data_format=input_data_format ) padded_image = self._pad_for_patching(resized_image, best_resolution, input_data_format=input_data_format) local_grids = divide_to_grids(padded_image, grid_size=grid_size, input_data_format=input_data_format) # make sure that all patches are in the input data format local_grids = [ to_channel_dimension_format(grid, channel_dim=data_format, input_channel_dim=input_data_format) for grid in local_grids ] return local_grids def preprocess( self, images: ImageInput, do_resize: bool = None, size: Dict[str, int] = None, anyres: bool = None, unpad: bool = None, video: bool = None, num_queries_vis_abstractor: int = None, possible_resolutions: List = None, patch_size: int = None, pad_to_square: bool = None, resample: PILImageResampling = None, do_center_crop: bool = None, crop_size: int = None, do_rescale: bool = None, rescale_factor: float = None, do_normalize: bool = None, image_mean: Optional[Union[float, List[float]]] = None, image_std: Optional[Union[float, List[float]]] = None, do_convert_rgb: bool = None, return_tensors: Optional[Union[str, TensorType]] = None, data_format: Optional[ChannelDimension] = ChannelDimension.FIRST, input_data_format: Optional[Union[str, ChannelDimension]] = None, return_dummy_image: bool = False, num_queries_vis_abstractor_slow: int = 0, first_last_frames_slow: bool = False, is_first_or_last_frames: bool = False, ): """ HCXVisionImageProcessor 로 image tensor, original image size (width, height), visual tokens :return pixel_values: List of 4D tensor 로 image tensor :return image_sizes: List of Dict 로 image width, height [{"width": image 1 의 width, "height": image 1 의 height}, {"width": image 2 의 width, "height": image 2 의 height}, ...] :return vision_query_lengths: List of int 로 각 image 가 LLM 입력으로 전달될때 변환되는 visual token 수 """ do_resize = do_resize if do_resize is not None else self.do_resize size = size if size is not None else self.size size = get_size_dict(size, param_name="size", default_to_square=False) anyres = anyres if anyres is not None else self.anyres unpad = unpad if unpad is not None else self.unpad if video: unpad = False num_queries_vis_abstractor = ( num_queries_vis_abstractor if num_queries_vis_abstractor is not None else self.num_queries_vis_abstractor ) possible_resolutions = possible_resolutions if possible_resolutions is not None else self.possible_resolutions patch_size = patch_size if patch_size is not None else self.patch_size pad_to_square = pad_to_square if pad_to_square is not None else self.pad_to_square resample = resample if resample is not None else self.resample do_center_crop = do_center_crop if do_center_crop is not None else self.do_center_crop crop_size = crop_size if crop_size is not None else self.crop_size crop_size = get_size_dict(crop_size, param_name="crop_size", default_to_square=True) do_rescale = do_rescale if do_rescale is not None else self.do_rescale rescale_factor = rescale_factor if rescale_factor is not None else self.rescale_factor do_normalize = do_normalize if do_normalize is not None else self.do_normalize image_mean = image_mean if image_mean is not None else self.image_mean image_std = image_std if image_std is not None else self.image_std do_convert_rgb = do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb if return_dummy_image: images = Image.new("RGB", (224, 224), (0, 0, 0)) images = make_list_of_images(images) if not valid_images(images): raise ValueError( "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, " "torch.Tensor, tf.Tensor or jax.ndarray." ) if do_convert_rgb: images = [convert_to_rgb(image) for image in images] # All transformations expect numpy arrays. images = [to_numpy_array(image) for image in images] if is_scaled_image(images[0]) and do_rescale: logger.warning_once( "It looks like you are trying to rescale already rescaled images. If the input" " images have pixel values between 0 and 1, set `do_rescale=False` to avoid rescaling them again." ) if input_data_format is None: # We assume that all images have the same channel dimension format. input_data_format = infer_channel_dimension_format(images[0]) new_images = [] image_sizes = [get_image_size(image, channel_dim=input_data_format) for image in images] vision_query_lengths = [] assert crop_size["height"] == crop_size["width"] # global image 의 padding 연산은, image original width, height 가 클 때 bottleneck 이 될 수 있음 # 장축의 길이를 size["shortest_edge"] 로 resize 를 먼저 한 뒤에, padding if anyres: anyres_global_images = copy.deepcopy(images) if pad_to_square: background_color = tuple(int(x * 255) for x in self.image_mean) anyres_global_images = [ resize_longside(copy.deepcopy(image), size["shortest_edge"], resample, input_data_format) for image in anyres_global_images ] anyres_global_images = [ expand2square(image, background_color=background_color, input_data_format=input_data_format)[0] for image in anyres_global_images ] else: anyres_global_images = [ self.resize( image=image, size={"height": size["shortest_edge"], "width": size["shortest_edge"]}, resample=resample, input_data_format=input_data_format, ) for image in anyres_global_images ] else: anyres_global_images = [None for _ in range(len(images))] if pad_to_square: background_color = tuple(int(x * 255) for x in self.image_mean) images = [ resize_longside(image, size["shortest_edge"], resample, input_data_format) for image in images ] images = [ expand2square(image, background_color=background_color, input_data_format=input_data_format)[0] for image in images ] for image, anyres_global_image, image_size in zip(images, anyres_global_images, image_sizes): if anyres: # convert image into a list of grids # we intentially use the same data format as the input data format image_grids = self.get_image_grids( image, possible_resolutions, grid_size=crop_size["height"], resample=resample, data_format=input_data_format, input_data_format=input_data_format, ) # video 에 대해서는 global image (thumbnail) 를 사용하지 않음 if not video: image_grids = [anyres_global_image] + image_grids else: image_grids = [image] pixel_values = self._preprocess( image_grids, do_resize=do_resize, size=size, resample=resample, do_center_crop=do_center_crop, crop_size=crop_size, do_rescale=do_rescale, rescale_factor=rescale_factor, do_normalize=do_normalize, image_mean=image_mean, image_std=image_std, data_format=data_format, input_data_format=input_data_format, ) pixel_values = np.array(pixel_values) new_images.append(pixel_values) num_grids = pixel_values.shape[0] vision_query_length = determine_anyres_num_vision_patches( num_grids=num_grids, image_size=image_size, grid_size=crop_size["height"], patch_size=patch_size, possible_resolutions=possible_resolutions, anyres=anyres, unpad=unpad, num_queries_vis_abstractor=num_queries_vis_abstractor, num_queries_vis_abstractor_slow=num_queries_vis_abstractor_slow, video=video, first_last_frames_slow=first_last_frames_slow, is_first_or_last_frames=is_first_or_last_frames, ) vision_query_lengths.append(vision_query_length) if return_dummy_image: vision_query_lengths = [] data = { "pixel_values": [torch.tensor(new_image) for new_image in new_images], "image_sizes": [{"width": image_size[1], "height": image_size[0]} for image_size in image_sizes], "vision_query_lengths": vision_query_lengths, } return BatchFeature(data=data) def save_pretrained( self, save_directory: Union[str, os.PathLike], *args, **kwargs, ): self.register_for_auto_class() super().save_pretrained(save_directory, *args, **kwargs) class HCXVisionV2Processor(Qwen2_5_VLProcessor): attributes = ["image_processor", "tokenizer", "video_processor"] image_processor_class = "AutoImageProcessor" video_processor_class = "AutoVideoProcessor" tokenizer_class = ("GPT2Tokenizer", "GPT2TokenizerFast", "PreTrainedTokenizer", "PreTrainedTokenizerFast") def __init__(self, image_processor=None, tokenizer=None, video_processor=None, chat_template=None, **kwargs): self.tokenizer = tokenizer super().__init__(image_processor, tokenizer, video_processor, chat_template=self.tokenizer.chat_template) def save_pretrained( self, save_directory: Union[str, os.PathLike], *args, **kwargs, ): self.register_for_auto_class() super().save_pretrained(save_directory, *args, **kwargs) def __call__( self, images: ImageInput = None, text: Union[TextInput, PreTokenizedInput, list[TextInput], list[PreTokenizedInput]] = None, videos: VideoInput = None, **kwargs: Unpack[Qwen2_5_VLProcessorKwargs], ) -> BatchFeature: """ Main method to prepare for the model one or several sequences(s) and image(s). This method forwards the `text` and `kwargs` arguments to Qwen2TokenizerFast's [`~Qwen2TokenizerFast.__call__`] if `text` is not `None` to encode the text. To prepare the vision inputs, this method forwards the `vision_infos` and `kwrags` arguments to Qwen2VLImageProcessor's [`~Qwen2VLImageProcessor.__call__`] if `vision_infos` is not `None`. Args: images (`PIL.Image.Image`, `np.ndarray`, `torch.Tensor`, `list[PIL.Image.Image]`, `list[np.ndarray]`, `list[torch.Tensor]`): The image or batch of images to be prepared. Each image can be a PIL image, NumPy array or PyTorch tensor. Both channels-first and channels-last formats are supported. text (`str`, `list[str]`, `list[list[str]]`): The sequence or batch of sequences to be encoded. Each sequence can be a string or a list of strings (pretokenized string). If the sequences are provided as list of strings (pretokenized), you must set `is_split_into_words=True` (to lift the ambiguity with a batch of sequences). videos (`np.ndarray`, `torch.Tensor`, `list[np.ndarray]`, `list[torch.Tensor]`): The image or batch of videos to be prepared. Each video can be a 4D NumPy array or PyTorch tensor, or a nested list of 3D frames. Both channels-first and channels-last formats are supported. return_tensors (`str` or [`~utils.TensorType`], *optional*): If set, will return tensors of a particular framework. Acceptable values are: - `'tf'`: Return TensorFlow `tf.constant` objects. - `'pt'`: Return PyTorch `torch.Tensor` objects. - `'np'`: Return NumPy `np.ndarray` objects. - `'jax'`: Return JAX `jnp.ndarray` objects. Returns: [`BatchFeature`]: A [`BatchFeature`] with the following fields: - **input_ids** -- List of token ids to be fed to a model. Returned when `text` is not `None`. - **attention_mask** -- List of indices specifying which tokens should be attended to by the model (when `return_attention_mask=True` or if *"attention_mask"* is in `self.model_input_names` and if `text` is not `None`). - **pixel_values** -- Pixel values to be fed to a model. Returned when `images` is not `None`. - **pixel_values_videos** -- Pixel values of videos to be fed to a model. Returned when `videos` is not `None`. - **image_grid_thw** -- List of image 3D grid in LLM. Returned when `images` is not `None`. - **video_grid_thw** -- List of video 3D grid in LLM. Returned when `videos` is not `None`. """ output_kwargs = self._merge_kwargs( Qwen2_5_VLProcessorKwargs, tokenizer_init_kwargs=self.tokenizer.init_kwargs, **kwargs, ) image_inputs = videos_inputs = {} if images is not None: image_inputs = self.image_processor(images=images, **output_kwargs["images_kwargs"]) image_grid_thw = image_inputs["image_grid_thw"] if videos is not None: videos_inputs = self.video_processor(videos=videos, **output_kwargs["videos_kwargs"]) video_grid_thw = videos_inputs["video_grid_thw"] if not isinstance(text, list): text = [text] text = text.copy() # below lines change text in-place if images is not None: merge_length = self.image_processor.merge_size**2 index = 0 for i in range(len(text)): while self.image_token in text[i]: num_image_tokens = image_grid_thw[index].prod() // merge_length text[i] = text[i].replace(self.image_token, "<|placeholder|>" * num_image_tokens, 1) text[i] = text[i].replace( '{"resolution": [w, h]}', '{"resolution": ' + str(list(images[i].size)) + "}" ) index += 1 text[i] = text[i].replace("<|placeholder|>", self.image_token) if videos is not None: merge_length = self.video_processor.merge_size**2 index = 0 for i in range(len(text)): while self.video_token in text[i]: num_video_tokens = video_grid_thw[index].prod() // merge_length text[i] = text[i].replace(self.video_token, "<|placeholder|>" * num_video_tokens, 1) index += 1 text[i] = text[i].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"], return_tensors=None) self._check_special_mm_tokens(text, text_inputs, modalities=["image", "video"]) if return_mm_token_type_ids: array_ids = np.array(text_inputs["input_ids"]) mm_token_type_ids = np.zeros_like(text_inputs["input_ids"]) mm_token_type_ids[array_ids == self.image_token_id] = 1 text_inputs["mm_token_type_ids"] = mm_token_type_ids.tolist() return BatchFeature(data={**text_inputs, **image_inputs, **videos_inputs}, tensor_type=return_tensors)