From 930797af57c3e604824486dda8656d8199956026 Mon Sep 17 00:00:00 2001 From: DimasfromLavoisier Date: Fri, 29 Aug 2025 15:11:43 +0200 Subject: [PATCH] model for image summarization and vqa --- ammico/summary.py | 873 +++++++++++++++------------------------------- 1 file changed, 288 insertions(+), 585 deletions(-) diff --git a/ammico/summary.py b/ammico/summary.py index c5734f9..ed52fb3 100644 --- a/ammico/summary.py +++ b/ammico/summary.py @@ -1,641 +1,344 @@ -from ammico.utils import AnalysisMethod -from torch import cuda, no_grad +from ammico.utils import AnalysisMethod, AnalysisType +from ammico.model import MultimodalSummaryModel + +import os +import torch from PIL import Image -from lavis.models import load_model_and_preprocess -from typing import Optional +import warnings + +from typing import List, Optional, Union, Dict, Any +from collections.abc import Sequence as _Sequence +from transformers import GenerationConfig +import re +from qwen_vl_utils import process_vision_info -class SummaryDetector(AnalysisMethod): - allowed_model_types = [ - "base", - "large", - "vqa", - ] - allowed_new_model_types = [ - "blip2_t5_pretrain_flant5xxl", - "blip2_t5_pretrain_flant5xl", - "blip2_t5_caption_coco_flant5xl", - "blip2_opt_pretrain_opt2.7b", - "blip2_opt_pretrain_opt6.7b", - "blip2_opt_caption_coco_opt2.7b", - "blip2_opt_caption_coco_opt6.7b", - ] - all_allowed_model_types = allowed_model_types + allowed_new_model_types - allowed_analysis_types = ["summary", "questions", "summary_and_questions"] +class ImageSummaryDetector(AnalysisMethod): def __init__( self, + summary_model: MultimodalSummaryModel, subdict: dict = {}, - model_type: str = "base", - analysis_type: str = "summary_and_questions", - list_of_questions: Optional[list[str]] = None, - summary_model=None, - summary_vis_processors=None, - summary_vqa_model=None, - summary_vqa_vis_processors=None, - summary_vqa_txt_processors=None, - summary_vqa_model_new=None, - summary_vqa_vis_processors_new=None, - summary_vqa_txt_processors_new=None, - device_type: Optional[str] = None, ) -> None: """ - SummaryDetector class for analysing images using the blip_caption model. + Class for analysing images using QWEN-2.5-VL model. + It provides methods for generating captions and answering questions about images. Args: + summary_model ([type], optional): An instance of MultimodalSummaryModel to be used for analysis. subdict (dict, optional): Dictionary containing the image to be analysed. Defaults to {}. - model_type (str, optional): Type of model to use. Can be "base" or "large" or "vqa" for blip_caption and VQA. Or can be one of the new models: - "blip2_t5_pretrain_flant5xxl", - "blip2_t5_pretrain_flant5xl", - "blip2_t5_caption_coco_flant5xl", - "blip2_opt_pretrain_opt2.7b", - "blip2_opt_pretrain_opt6.7b", - "blip2_opt_caption_coco_opt2.7b", - "blip2_opt_caption_coco_opt6.7b". Defaults to "base". - analysis_type (str, optional): Type of analysis to perform. Can be "summary", "questions" or "summary_and_questions". Defaults to "summary_and_questions". - list_of_questions (list, optional): List of questions to answer. Defaults to ["Are there people in the image?", "What is this picture about?"]. - summary_model ([type], optional): blip_caption model. Defaults to None. - summary_vis_processors ([type], optional): Preprocessors for visual inputs. Defaults to None. - summary_vqa_model ([type], optional): blip_vqa model. Defaults to None. - summary_vqa_vis_processors ([type], optional): Preprocessors for vqa visual inputs. Defaults to None. - summary_vqa_txt_processors ([type], optional): Preprocessors for vqa text inputs. Defaults to None. - summary_vqa_model_new ([type], optional): new_vqa model. Defaults to None. - summary_vqa_vis_processors_new ([type], optional): Preprocessors for vqa visual inputs. Defaults to None. - summary_vqa_txt_processors_new ([type], optional): Preprocessors for vqa text inputs. Defaults to None. - - Raises: - ValueError: If analysis_type is not one of "summary", "questions" or "summary_and_questions". - Returns: None. """ super().__init__(subdict) - # check if analysis_type is valid - if analysis_type not in self.allowed_analysis_types: - raise ValueError( - "analysis_type must be one of {}".format(self.allowed_analysis_types) - ) - # check if device_type is valid - if device_type is None: - self.summary_device = "cuda" if cuda.is_available() else "cpu" - elif device_type not in ["cuda", "cpu"]: - raise ValueError("device_type must be one of {}".format(["cuda", "cpu"])) + self.summary_model = summary_model + + def _load_pil_if_needed( + self, filename: Union[str, os.PathLike, Image.Image] + ) -> Image.Image: + if isinstance(filename, (str, os.PathLike)): + return Image.open(filename).convert("RGB") + elif isinstance(filename, Image.Image): + return filename.convert("RGB") + else: + raise ValueError("filename must be a path or PIL.Image") + + @staticmethod + def _is_sequence_but_not_str(obj: Any) -> bool: + """True for sequence-like but not a string/bytes/PIL.Image.""" + return isinstance(obj, _Sequence) and not isinstance( + obj, (str, bytes, Image.Image) + ) + + def _prepare_inputs( + self, list_of_questions: list[str], entry: Optional[Dict[str, Any]] = None + ) -> Dict[str, torch.Tensor]: + filename = entry.get("filename") + if filename is None: + raise ValueError("entry must contain key 'filename'") + + if isinstance(filename, (str, os.PathLike, Image.Image)): + images_context = self._load_pil_if_needed(filename) + elif self._is_sequence_but_not_str(filename): + images_context = [self._load_pil_if_needed(i) for i in filename] else: - self.summary_device = device_type - # check if model_type is valid - if model_type not in self.all_allowed_model_types: raise ValueError( - "Model type is not allowed - please select one of {}".format( - self.all_allowed_model_types - ) + "Unsupported 'filename' entry: expected path, PIL.Image, or sequence." ) - self.model_type = model_type - self.analysis_type = analysis_type - # check if list_of_questions is valid - if list_of_questions is None and model_type in self.allowed_model_types: - self.list_of_questions = [ - "Are there people in the image?", - "What is this picture about?", + + images_only_messages = [ + { + "role": "user", + "content": [ + *( + [{"type": "image", "image": img} for img in images_context] + if isinstance(images_context, list) + else [{"type": "image", "image": images_context}] + ) + ], + } + ] + + try: + image_inputs, _ = process_vision_info(images_only_messages) + except Exception as e: + raise RuntimeError(f"Image processing failed: {e}") + + texts: List[str] = [] + for q in list_of_questions: + messages = [ + { + "role": "user", + "content": [ + *( + [ + {"type": "image", "image": image} + for image in images_context + ] + if isinstance(images_context, list) + else [{"type": "image", "image": images_context}] + ), + {"type": "text", "text": q}, + ], + } ] - elif list_of_questions is None and model_type in self.allowed_new_model_types: - self.list_of_questions = [ - "Question: Are there people in the image? Answer:", - "Question: What is this picture about? Answer:", - ] - elif (not isinstance(list_of_questions, list)) or ( - not all(isinstance(i, str) for i in list_of_questions) - ): - raise ValueError( - "list_of_questions must be a list of string (questions)" - ) # add sequence of questions - else: - self.list_of_questions = list_of_questions - # load models and preprocessors - if ( - model_type in self.allowed_model_types - and (summary_model is None) - and (summary_vis_processors is None) - and (analysis_type == "summary" or analysis_type == "summary_and_questions") - ): - self.summary_model, self.summary_vis_processors = self.load_model( - model_type=model_type + text = self.summary_model.processor.apply_chat_template( + messages, tokenize=False, add_generation_prompt=True ) - else: - self.summary_model = summary_model - self.summary_vis_processors = summary_vis_processors - if ( - model_type in self.allowed_model_types - and (summary_vqa_model is None) - and (summary_vqa_vis_processors is None) - and (summary_vqa_txt_processors is None) - and ( - analysis_type == "questions" or analysis_type == "summary_and_questions" - ) - ): - ( - self.summary_vqa_model, - self.summary_vqa_vis_processors, - self.summary_vqa_txt_processors, - ) = self.load_vqa_model() - else: - self.summary_vqa_model = summary_vqa_model - self.summary_vqa_vis_processors = summary_vqa_vis_processors - self.summary_vqa_txt_processors = summary_vqa_txt_processors - if ( - model_type in self.allowed_new_model_types - and (summary_vqa_model_new is None) - and (summary_vqa_vis_processors_new is None) - and (summary_vqa_txt_processors_new is None) - ): - ( - self.summary_vqa_model_new, - self.summary_vqa_vis_processors_new, - self.summary_vqa_txt_processors_new, - ) = self.load_new_model(model_type=model_type) - else: - self.summary_vqa_model_new = summary_vqa_model_new - self.summary_vqa_vis_processors_new = summary_vqa_vis_processors_new - self.summary_vqa_txt_processors_new = summary_vqa_txt_processors_new + texts.append(text) - def load_model_base(self): - """ - Load base_coco blip_caption model and preprocessors for visual inputs from lavis.models. - - Args: - - Returns: - summary_model (torch.nn.Module): model. - summary_vis_processors (dict): preprocessors for visual inputs. - """ - summary_model, summary_vis_processors, _ = load_model_and_preprocess( - name="blip_caption", - model_type="base_coco", - is_eval=True, - device=self.summary_device, + images_batch = [image_inputs] * len(texts) + inputs = self.summary_model.processor( + text=texts, + images=images_batch, + padding=True, + return_tensors="pt", ) - return summary_model, summary_vis_processors + inputs = {k: v.to(self.summary_model.device) for k, v in inputs.items()} - def load_model_large(self): - """ - Load large_coco blip_caption model and preprocessors for visual inputs from lavis.models. + return inputs - Args: - - Returns: - summary_model (torch.nn.Module): model. - summary_vis_processors (dict): preprocessors for visual inputs. - """ - summary_model, summary_vis_processors, _ = load_model_and_preprocess( - name="blip_caption", - model_type="large_coco", - is_eval=True, - device=self.summary_device, - ) - return summary_model, summary_vis_processors - - def load_model(self, model_type: str): - """ - Load blip_caption model and preprocessors for visual inputs from lavis.models. - - Args: - model_type (str): type of the model. - - Returns: - summary_model (torch.nn.Module): model. - summary_vis_processors (dict): preprocessors for visual inputs. - """ - select_model = { - "base": SummaryDetector.load_model_base, - "large": SummaryDetector.load_model_large, - } - summary_model, summary_vis_processors = select_model[model_type](self) - return summary_model, summary_vis_processors - - def load_vqa_model(self): - """ - Load blip_vqa model and preprocessors for visual and text inputs from lavis.models. - - Args: - - Returns: - summary_vqa_model (torch.nn.Module): model. - summary_vqa_vis_processors (dict): preprocessors for visual inputs. - summary_vqa_txt_processors (dict): preprocessors for text inputs. - - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip_vqa", - model_type="vqav2", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def analyse_image( + def analyse_images( self, - subdict: dict = None, - analysis_type: Optional[str] = None, - list_of_questions: Optional[list[str]] = None, - consequential_questions: bool = False, - ): + analysis_type: Union[AnalysisType, str] = AnalysisType.SUMMARY_AND_QUESTIONS, + list_of_questions: Optional[List[str]] = None, + max_questions_per_image: int = 32, + keys_batch_size: int = 16, + is_concise_summary: bool = True, + is_concise_answer: bool = True, + ) -> Dict[str, dict]: """ - Analyse image with blip_caption model. + Analyse image with model. Args: analysis_type (str): type of the analysis. - subdict (dict): dictionary with analising pictures. list_of_questions (list[str]): list of questions. - consequential_questions (bool): whether to ask consequential questions. Works only for new BLIP2 models. - + max_questions_per_image (int): maximum number of questions per image. We recommend to keep it low to avoid long processing times and high memory usage. + keys_batch_size (int): number of images to process in a batch. + is_concise_summary (bool): whether to generate concise summary. + is_concise_answer (bool): whether to generate concise answers. Returns: self.subdict (dict): dictionary with analysis results. """ - if analysis_type is None: - analysis_type = self.analysis_type - if subdict is not None: - self.subdict = subdict - if list_of_questions is not None: - self.list_of_questions = list_of_questions + # TODO: add option to ask multiple questions per image as one batch. + if isinstance(analysis_type, AnalysisType): + analysis_type = analysis_type.value - if analysis_type == "summary_and_questions": - if ( - self.model_type in self.allowed_model_types - and self.analysis_type != "summary_and_questions" - ): # if model_type is not new and required model is absent - if self.summary_model is None: # load summary model if it is not loaded - self.summary_model, self.summary_vis_processors = self.load_model( - model_type=self.model_type + allowed = {"summary", "questions", "summary_and_questions"} + if analysis_type not in allowed: + raise ValueError(f"analysis_type must be one of {allowed}") + + if list_of_questions is None: + list_of_questions = [ + "Are there people in the image?", + "What is this picture about?", + ] + + keys = list(self.subdict.keys()) + for batch_start in range(0, len(keys), keys_batch_size): + batch_keys = keys[batch_start : batch_start + keys_batch_size] + for key in batch_keys: + entry = self.subdict[key] + if analysis_type in ("summary", "summary_and_questions"): + try: + caps = self.generate_caption( + entry, + num_return_sequences=1, + is_concise_summary=is_concise_summary, + ) + entry["caption"] = caps[0] if caps else "" + except Exception as e: + warnings.warn( + "Caption generation failed for key %s: %s", key, e + ) + + if analysis_type in ("questions", "summary_and_questions"): + if len(list_of_questions) > max_questions_per_image: + raise ValueError( + f"Number of questions per image ({len(list_of_questions)}) exceeds safety cap ({max_questions_per_image})." + " Reduce questions or increase max_questions_per_image." + ) + try: + vqa_map = self.answer_questions( + list_of_questions, entry, is_concise_answer + ) + entry["vqa"] = vqa_map + except Exception as e: + warnings.warn("VQA failed for key %s: %s", key, e) + + self.subdict[key] = entry + return self.subdict + + def generate_caption( + self, + entry: Optional[Dict[str, Any]] = None, + num_return_sequences: int = 1, + is_concise_summary: bool = True, + ) -> List[str]: + """ + Create caption for image. Depending on is_concise_summary it will be either concise or detailed. + + Args: + entry (dict): dictionary containing the image to be captioned. + num_return_sequences (int): number of captions to generate. + is_concise_summary (bool): whether to generate concise summary. + + Returns: + results (list[str]): list of generated captions. + """ + if is_concise_summary: + prompt = ["Describe this image in one concise caption."] + max_new_tokens = 64 + else: + prompt = ["Describe this image."] + max_new_tokens = 256 + inputs = self._prepare_inputs(prompt, entry) + + gen_conf = GenerationConfig( + max_new_tokens=max_new_tokens, + do_sample=False, + num_return_sequences=num_return_sequences, + ) + + with torch.inference_mode(): + try: + if self.summary_model.device == "cuda": + with torch.cuda.amp.autocast(enabled=True): + generated_ids = self.summary_model.model.generate( + **inputs, generation_config=gen_conf + ) + else: + generated_ids = self.summary_model.model.generate( + **inputs, generation_config=gen_conf ) - elif ( - self.summary_vqa_model is None - ): # load vqa model if it is not loaded - ( - self.summary_vqa_model, - self.summary_vqa_vis_processors, - self.summary_vqa_txt_processors, - ) = self.load_vqa_model() - self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis - self.analyse_summary(nondeterministic_summaries=True) - self.analyse_questions(self.list_of_questions, consequential_questions) - elif analysis_type == "summary": - if ( - (self.model_type in self.allowed_model_types) - and (self.analysis_type == "questions") - and (self.summary_model is None) - ): # if model_type is not new and required model is absent - ( - self.summary_model, - self.summary_vis_processors, - ) = self.load_model( # load summary model if it is not loaded - model_type=self.model_type + except RuntimeError as e: + warnings.warn( + "Retry without autocast failed: %s. Attempting cudnn-disabled retry.", + e, ) - self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis - self.analyse_summary(nondeterministic_summaries=True) - elif analysis_type == "questions": - if ( - (self.model_type in self.allowed_model_types) - and (self.analysis_type == "summary") - and (self.summary_vqa_model is None) - ): # if model_type is not new and required model is absent - ( - self.summary_vqa_model, # load vqa model if it is not loaded - self.summary_vqa_vis_processors, - self.summary_vqa_txt_processors, - ) = self.load_vqa_model() - self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis - self.analyse_questions(self.list_of_questions, consequential_questions) - else: - raise ValueError( - "analysis_type must be one of {}".format(self.allowed_analysis_types) + cudnn_was_enabled = ( + torch.backends.cudnn.is_available() and torch.backends.cudnn.enabled + ) + if cudnn_was_enabled: + torch.backends.cudnn.enabled = False + try: + generated_ids = self.summary_model.model.generate( + **inputs, generation_config=gen_conf + ) + except Exception as retry_error: + raise RuntimeError( + f"Failed to generate ids after retry: {retry_error}" + ) from retry_error + finally: + if cudnn_was_enabled: + torch.backends.cudnn.enabled = True + + decoded = None + if "input_ids" in inputs: + in_ids = inputs["input_ids"] + trimmed = [ + out_ids[len(inp_ids) :] + for inp_ids, out_ids in zip(in_ids, generated_ids) + ] + decoded = self.summary_model.tokenizer.batch_decode( + trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False ) - return self.subdict - - def analyse_summary(self, nondeterministic_summaries: bool = True): - """ - Create 1 constant and 3 non deterministic captions for image. - - Args: - nondeterministic_summaries (bool): whether to create 3 non deterministic captions. - - Returns: - self.subdict (dict): dictionary with analysis results. - """ - if self.model_type in self.allowed_model_types: - vis_processors = self.summary_vis_processors - model = self.summary_model - elif self.model_type in self.allowed_new_model_types: - vis_processors = self.summary_vqa_vis_processors_new - model = self.summary_vqa_model_new else: - raise ValueError( - "Model type is not allowed - please select one of {}".format( - self.all_allowed_model_types - ) + decoded = self.summary_model.tokenizer.batch_decode( + generated_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=False, ) - path = self.subdict["filename"] - raw_image = Image.open(path).convert("RGB") - image = vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device) - with no_grad(): - self.subdict["const_image_summary"] = model.generate({"image": image})[0] - if nondeterministic_summaries: - self.subdict["3_non-deterministic_summary"] = model.generate( - {"image": image}, use_nucleus_sampling=True, num_captions=3 - ) - return self.subdict - def analyse_questions( - self, list_of_questions: list[str], consequential_questions: bool = False - ) -> dict: + results = [d.strip() for d in decoded] + return results + + def answer_questions( + self, + list_of_questions: list[str], + entry: Optional[Dict[str, Any]] = None, + is_concise_answer: bool = True, + ) -> List[str]: """ - Generate answers to free-form questions about image written in natural language. - + Create answers for list of questions about image. Args: list_of_questions (list[str]): list of questions. - consequential_questions (bool): whether to ask consequential questions. Works only for new BLIP2 models. - + entry (dict): dictionary containing the image to be captioned. + is_concise_answer (bool): whether to generate concise answers. Returns: - self.subdict (dict): dictionary with answers to questions. + answers (list[str]): list of answers. """ - model, vis_processors, txt_processors, model_old = self.check_model() - if len(list_of_questions) > 0: - path = self.subdict["filename"] - raw_image = Image.open(path).convert("RGB") - image = ( - vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device) - ) - question_batch = [] - list_of_questions_processed = [] + if is_concise_answer: + gen_conf = GenerationConfig(max_new_tokens=64, do_sample=False) + for i in range(len(list_of_questions)): + if not list_of_questions[i].strip().endswith("?"): + list_of_questions[i] = list_of_questions[i].strip() + "?" + if not list_of_questions[i].lower().startswith("answer concisely"): + list_of_questions[i] = "Answer concisely: " + list_of_questions[i] + else: + gen_conf = GenerationConfig(max_new_tokens=128, do_sample=False) - if model_old: - for quest in list_of_questions: - list_of_questions_processed.append(txt_processors["eval"](quest)) + question_chunk_size = 8 + answers: List[str] = [] + n = len(list_of_questions) + for i in range(0, n, question_chunk_size): + chunk = list_of_questions[i : i + question_chunk_size] + inputs = self._prepare_inputs(chunk, entry) + with torch.inference_mode(): + if self.summary_model.device == "cuda": + with torch.cuda.amp.autocast(enabled=True): + out_ids = self.summary_model.model.generate( + **inputs, generation_config=gen_conf + ) + else: + out_ids = self.summary_model.model.generate( + **inputs, generation_config=gen_conf + ) + + if "input_ids" in inputs: + in_ids = inputs["input_ids"] + trimmed_batch = [ + out_row[len(inp_row) :] for inp_row, out_row in zip(in_ids, out_ids) + ] + decoded = self.summary_model.tokenizer.batch_decode( + trimmed_batch, + skip_special_tokens=True, + clean_up_tokenization_spaces=False, + ) else: - for quest in list_of_questions: - list_of_questions_processed.append((str)(quest)) - - for quest in list_of_questions_processed: - question_batch.append(quest) - batch_size = len(list_of_questions) - image_batch = image.repeat(batch_size, 1, 1, 1) - - if not consequential_questions: - with no_grad(): - if model_old: - answers_batch = model.predict_answers( - samples={ - "image": image_batch, - "text_input": question_batch, - }, - inference_method="generate", - ) - else: - answers_batch = model.generate( - {"image": image_batch, "prompt": question_batch} - ) - - for q, a in zip(list_of_questions, answers_batch): - self.subdict[q] = a - - if consequential_questions and not model_old: - query_with_context = "" - for quest in question_batch: - query_with_context = query_with_context + quest - with no_grad(): - answer = model.generate( - {"image": image, "prompt": query_with_context} - ) - self.subdict[query_with_context] = answer[0] - query_with_context = query_with_context + " " + answer[0] + ". " - elif consequential_questions and model_old: - raise ValueError( - "Consequential questions are not allowed for old models" + decoded = self.summary_model.tokenizer.batch_decode( + out_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=False, ) - else: - print("Please, enter list of questions") - return self.subdict - def check_model(self): - """ - Check model type and return appropriate model and preprocessors. + answers.extend([d.strip() for d in decoded]) - Args: - - Returns: - model (nn.Module): model. - vis_processors (dict): visual preprocessor. - txt_processors (dict): text preprocessor. - model_old (bool): whether model is old or new. - """ - if self.model_type in self.allowed_model_types: - vis_processors = self.summary_vqa_vis_processors - model = self.summary_vqa_model - txt_processors = self.summary_vqa_txt_processors - model_old = True - elif self.model_type in self.allowed_new_model_types: - vis_processors = self.summary_vqa_vis_processors_new - model = self.summary_vqa_model_new - txt_processors = self.summary_vqa_txt_processors_new - model_old = False - else: + if len(answers) != len(list_of_questions): raise ValueError( - "Model type is not allowed - please select one of {}".format( - self.all_allowed_model_types - ) + f"Expected {len(list_of_questions)} answers, but got {len(answers)}, try vary amount of questions" ) - return model, vis_processors, txt_processors, model_old - - def load_new_model(self, model_type: str): - """ - Load new BLIP2 models. - - Args: - model_type (str): type of the model. - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - select_model = { - "blip2_t5_pretrain_flant5xxl": SummaryDetector.load_model_blip2_t5_pretrain_flant5xxl, - "blip2_t5_pretrain_flant5xl": SummaryDetector.load_model_blip2_t5_pretrain_flant5xl, - "blip2_t5_caption_coco_flant5xl": SummaryDetector.load_model_blip2_t5_caption_coco_flant5xl, - "blip2_opt_pretrain_opt2.7b": SummaryDetector.load_model_blip2_opt_pretrain_opt27b, - "blip2_opt_pretrain_opt6.7b": SummaryDetector.load_model_base_blip2_opt_pretrain_opt67b, - "blip2_opt_caption_coco_opt2.7b": SummaryDetector.load_model_blip2_opt_caption_coco_opt27b, - "blip2_opt_caption_coco_opt6.7b": SummaryDetector.load_model_base_blip2_opt_caption_coco_opt67b, - } - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = select_model[model_type](self) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_blip2_t5_pretrain_flant5xxl(self): - """ - Load BLIP2 model with FLAN-T5 XXL architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_t5", - model_type="pretrain_flant5xxl", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_blip2_t5_pretrain_flant5xl(self): - """ - Load BLIP2 model with FLAN-T5 XL architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_t5", - model_type="pretrain_flant5xl", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_blip2_t5_caption_coco_flant5xl(self): - """ - Load BLIP2 model with caption_coco_flant5xl architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_t5", - model_type="caption_coco_flant5xl", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_blip2_opt_pretrain_opt27b(self): - """ - Load BLIP2 model with pretrain_opt2 architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_opt", - model_type="pretrain_opt2.7b", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_base_blip2_opt_pretrain_opt67b(self): - """ - Load BLIP2 model with pretrain_opt6.7b architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_opt", - model_type="pretrain_opt6.7b", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_blip2_opt_caption_coco_opt27b(self): - """ - Load BLIP2 model with caption_coco_opt2.7b architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_opt", - model_type="caption_coco_opt2.7b", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - - def load_model_base_blip2_opt_caption_coco_opt67b(self): - """ - Load BLIP2 model with caption_coco_opt6.7b architecture. - - Args: - - Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. - """ - ( - summary_vqa_model, - summary_vqa_vis_processors, - summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip2_opt", - model_type="caption_coco_opt6.7b", - is_eval=True, - device=self.summary_device, - ) - return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + return answers