| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| """ |
| Processor class for MiniCPMV. |
| """ |
|
|
| from typing import List, Optional, Union |
| import torch |
| import re |
|
|
| from transformers.image_utils import ImageInput |
| from transformers.processing_utils import ProcessorMixin |
| from transformers.tokenization_utils_base import PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy |
| from transformers.utils import TensorType |
|
|
| from .image_processing_minicpmv import MiniCPMVBatchFeature |
|
|
|
|
| class MiniCPMVProcessor(ProcessorMixin): |
| r""" |
| Constructs a MiniCPMV processor which wraps a MiniCPMV image processor and a MiniCPMV tokenizer into a single processor. |
| |
| [`MiniCPMVProcessor`] offers all the functionalities of [`MiniCPMVImageProcessor`] and [`LlamaTokenizerWrapper`]. See the |
| [`~MiniCPMVProcessor.__call__`] and [`~MiniCPMVProcessor.decode`] for more information. |
| |
| Args: |
| image_processor ([`MiniCPMVImageProcessor`], *optional*): |
| The image processor is a required input. |
| tokenizer ([`LlamaTokenizerWrapper`], *optional*): |
| The tokenizer is a required input. |
| """ |
| attributes = ["image_processor", "tokenizer"] |
| image_processor_class = "AutoImageProcessor" |
| tokenizer_class = "AutoTokenizer" |
|
|
| def __init__(self, image_processor=None, tokenizer=None): |
| super().__init__(image_processor, tokenizer) |
| |
| def __call__( |
| self, |
| text: Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]], |
| images: ImageInput = None, |
| padding: Union[bool, str, PaddingStrategy] = False, |
| max_length: Optional[int] = None, |
| do_pad: Optional[bool] = True, |
| return_tensors: Optional[Union[str, TensorType]] = TensorType.PYTORCH, |
| ) -> MiniCPMVBatchFeature: |
| if images is not None: |
| image_inputs = self.image_processor(images, do_pad=do_pad, return_tensors=return_tensors) |
| return self._convert_images_texts_to_inputs(image_inputs, text, max_length=max_length, return_tensors=return_tensors) |
| |
| |
| def batch_decode(self, *args, **kwargs): |
| """ |
| This method forwards all its arguments to LlamaTokenizerFast's [`~PreTrainedTokenizer.batch_decode`]. Please |
| refer to the docstring of this method for more information. |
| """ |
| output_ids = args[0] |
| result_text = [] |
| for result in output_ids: |
| result = result[result != 0] |
| if result[0] == self.tokenizer.bos_id: |
| result = result[1:] |
| if result[-1] == self.tokenizer.eos_id: |
| result = result[:-1] |
| result_text.append(self.tokenizer.decode(result, *args[1:], **kwargs).strip()) |
| return result_text |
| |
| |
| |
| def decode(self, *args, **kwargs): |
| """ |
| This method forwards all its arguments to LlamaTokenizerFast's [`~PreTrainedTokenizer.decode`]. Please refer to |
| the docstring of this method for more information. |
| """ |
| result = args[0] |
| result = result[result != 0] |
| if result[0] == self.tokenizer.bos_id: |
| result = result[1:] |
| if result[-1] == self.tokenizer.eos_id: |
| result = result[:-1] |
| return self.tokenizer.decode(result, *args[1:], **kwargs).strip() |
|
|
| def _convert( |
| self, input_str, max_inp_length: Optional[int] = None |
| ): |
| if self.tokenizer.add_bos_token: |
| input_ids = self.tokenizer.encode(input_str) |
| else: |
| input_ids = [self.tokenizer.bos_id] + self.tokenizer.encode(input_str) |
| if max_inp_length is not None: |
| input_ids = input_ids[:max_inp_length] |
| input_ids = torch.tensor(input_ids, dtype=torch.int32) |
|
|
| image_start_tokens = torch.where(input_ids == self.tokenizer.im_start_id)[0] |
| image_start_tokens += 1 |
| image_end_tokens = torch.where(input_ids == self.tokenizer.im_end_id)[0] |
| valid_image_nums = max(len(image_start_tokens), len(image_end_tokens)) |
| image_bounds = torch.hstack( |
| [ |
| image_start_tokens[:valid_image_nums].unsqueeze(-1), |
| image_end_tokens[:valid_image_nums].unsqueeze(-1), |
| ] |
| ) |
| return input_ids.unsqueeze(0), image_bounds |
|
|
| def _convert_images_texts_to_inputs(self, images, texts, do_pad=False, truncation=None, max_length=None, return_tensors=None): |
| if not len(images): |
| model_inputs = self.tokenizer(texts, return_tensors=return_tensors, padding=do_pad, truncation=truncation, max_length=max_length) |
| return MiniCPMVBatchFeature(data={**model_inputs}) |
| |
| pattern = "(<image>./</image>)" |
| images, image_sizes = images["pixel_values"], images["image_sizes"] |
|
|
| image_tags = re.findall(pattern, texts) |
| assert len(image_tags) <= 1 and len(image_sizes) == 1 |
| text_chunks = texts.split(pattern) |
| final_texts = text_chunks[0] + self.image_processor.get_slice_image_placeholder(image_sizes[0]) \ |
| + text_chunks[1] + "<AI>" |
| input_ids, image_bounds = self._convert(final_texts, max_length) |
| |
| return MiniCPMVBatchFeature(data={ |
| "input_ids": input_ids, |
| "pixel_values": images, |
| "image_sizes": [image_sizes], |
| "image_bounds": [image_bounds] |
| }, tensor_type=return_tensors) |
|
|
| @property |
| |
| def model_input_names(self): |
| tokenizer_input_names = self.tokenizer.model_input_names |
| image_processor_input_names = self.image_processor.model_input_names |
| return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names)) |
|
|
