зеркало из
https://github.com/ssciwr/AMMICO.git
synced 2025-10-29 21:16:06 +02:00
model for image summarization and vqa
Этот коммит содержится в:
DimasfromLavoisier
родитель
36a0f90a76
Коммит
930797af57
@ -1,641 +1,344 @@
|
|||||||
from ammico.utils import AnalysisMethod
|
from ammico.utils import AnalysisMethod, AnalysisType
|
||||||
from torch import cuda, no_grad
|
from ammico.model import MultimodalSummaryModel
|
||||||
|
|
||||||
|
import os
|
||||||
|
import torch
|
||||||
from PIL import Image
|
from PIL import Image
|
||||||
from lavis.models import load_model_and_preprocess
|
import warnings
|
||||||
from typing import Optional
|
|
||||||
|
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):
|
class ImageSummaryDetector(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"]
|
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
summary_model: MultimodalSummaryModel,
|
||||||
subdict: dict = {},
|
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:
|
) -> 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:
|
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 {}.
|
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:
|
Returns:
|
||||||
None.
|
None.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
super().__init__(subdict)
|
super().__init__(subdict)
|
||||||
# check if analysis_type is valid
|
self.summary_model = summary_model
|
||||||
if analysis_type not in self.allowed_analysis_types:
|
|
||||||
raise ValueError(
|
def _load_pil_if_needed(
|
||||||
"analysis_type must be one of {}".format(self.allowed_analysis_types)
|
self, filename: Union[str, os.PathLike, Image.Image]
|
||||||
)
|
) -> Image.Image:
|
||||||
# check if device_type is valid
|
if isinstance(filename, (str, os.PathLike)):
|
||||||
if device_type is None:
|
return Image.open(filename).convert("RGB")
|
||||||
self.summary_device = "cuda" if cuda.is_available() else "cpu"
|
elif isinstance(filename, Image.Image):
|
||||||
elif device_type not in ["cuda", "cpu"]:
|
return filename.convert("RGB")
|
||||||
raise ValueError("device_type must be one of {}".format(["cuda", "cpu"]))
|
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:
|
else:
|
||||||
self.summary_device = device_type
|
|
||||||
# check if model_type is valid
|
|
||||||
if model_type not in self.all_allowed_model_types:
|
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Model type is not allowed - please select one of {}".format(
|
"Unsupported 'filename' entry: expected path, PIL.Image, or sequence."
|
||||||
self.all_allowed_model_types
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
self.model_type = model_type
|
|
||||||
self.analysis_type = analysis_type
|
images_only_messages = [
|
||||||
# check if list_of_questions is valid
|
{
|
||||||
if list_of_questions is None and model_type in self.allowed_model_types:
|
"role": "user",
|
||||||
self.list_of_questions = [
|
"content": [
|
||||||
"Are there people in the image?",
|
*(
|
||||||
"What is this picture about?",
|
[{"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:
|
text = self.summary_model.processor.apply_chat_template(
|
||||||
self.list_of_questions = [
|
messages, tokenize=False, add_generation_prompt=True
|
||||||
"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
|
|
||||||
)
|
)
|
||||||
else:
|
texts.append(text)
|
||||||
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
|
|
||||||
|
|
||||||
def load_model_base(self):
|
images_batch = [image_inputs] * len(texts)
|
||||||
"""
|
inputs = self.summary_model.processor(
|
||||||
Load base_coco blip_caption model and preprocessors for visual inputs from lavis.models.
|
text=texts,
|
||||||
|
images=images_batch,
|
||||||
Args:
|
padding=True,
|
||||||
|
return_tensors="pt",
|
||||||
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,
|
|
||||||
)
|
)
|
||||||
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):
|
return inputs
|
||||||
"""
|
|
||||||
Load large_coco blip_caption model and preprocessors for visual inputs from lavis.models.
|
|
||||||
|
|
||||||
Args:
|
def analyse_images(
|
||||||
|
|
||||||
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(
|
|
||||||
self,
|
self,
|
||||||
subdict: dict = None,
|
analysis_type: Union[AnalysisType, str] = AnalysisType.SUMMARY_AND_QUESTIONS,
|
||||||
analysis_type: Optional[str] = None,
|
list_of_questions: Optional[List[str]] = None,
|
||||||
list_of_questions: Optional[list[str]] = None,
|
max_questions_per_image: int = 32,
|
||||||
consequential_questions: bool = False,
|
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:
|
Args:
|
||||||
analysis_type (str): type of the analysis.
|
analysis_type (str): type of the analysis.
|
||||||
subdict (dict): dictionary with analising pictures.
|
|
||||||
list_of_questions (list[str]): list of questions.
|
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:
|
Returns:
|
||||||
self.subdict (dict): dictionary with analysis results.
|
self.subdict (dict): dictionary with analysis results.
|
||||||
"""
|
"""
|
||||||
if analysis_type is None:
|
# TODO: add option to ask multiple questions per image as one batch.
|
||||||
analysis_type = self.analysis_type
|
if isinstance(analysis_type, AnalysisType):
|
||||||
if subdict is not None:
|
analysis_type = analysis_type.value
|
||||||
self.subdict = subdict
|
|
||||||
if list_of_questions is not None:
|
|
||||||
self.list_of_questions = list_of_questions
|
|
||||||
|
|
||||||
if analysis_type == "summary_and_questions":
|
allowed = {"summary", "questions", "summary_and_questions"}
|
||||||
if (
|
if analysis_type not in allowed:
|
||||||
self.model_type in self.allowed_model_types
|
raise ValueError(f"analysis_type must be one of {allowed}")
|
||||||
and self.analysis_type != "summary_and_questions"
|
|
||||||
): # if model_type is not new and required model is absent
|
if list_of_questions is None:
|
||||||
if self.summary_model is None: # load summary model if it is not loaded
|
list_of_questions = [
|
||||||
self.summary_model, self.summary_vis_processors = self.load_model(
|
"Are there people in the image?",
|
||||||
model_type=self.model_type
|
"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 (
|
except RuntimeError as e:
|
||||||
self.summary_vqa_model is None
|
warnings.warn(
|
||||||
): # load vqa model if it is not loaded
|
"Retry without autocast failed: %s. Attempting cudnn-disabled retry.",
|
||||||
(
|
e,
|
||||||
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
|
|
||||||
)
|
)
|
||||||
self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis
|
cudnn_was_enabled = (
|
||||||
self.analyse_summary(nondeterministic_summaries=True)
|
torch.backends.cudnn.is_available() and torch.backends.cudnn.enabled
|
||||||
elif analysis_type == "questions":
|
)
|
||||||
if (
|
if cudnn_was_enabled:
|
||||||
(self.model_type in self.allowed_model_types)
|
torch.backends.cudnn.enabled = False
|
||||||
and (self.analysis_type == "summary")
|
try:
|
||||||
and (self.summary_vqa_model is None)
|
generated_ids = self.summary_model.model.generate(
|
||||||
): # if model_type is not new and required model is absent
|
**inputs, generation_config=gen_conf
|
||||||
(
|
)
|
||||||
self.summary_vqa_model, # load vqa model if it is not loaded
|
except Exception as retry_error:
|
||||||
self.summary_vqa_vis_processors,
|
raise RuntimeError(
|
||||||
self.summary_vqa_txt_processors,
|
f"Failed to generate ids after retry: {retry_error}"
|
||||||
) = self.load_vqa_model()
|
) from retry_error
|
||||||
self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis
|
finally:
|
||||||
self.analyse_questions(self.list_of_questions, consequential_questions)
|
if cudnn_was_enabled:
|
||||||
else:
|
torch.backends.cudnn.enabled = True
|
||||||
raise ValueError(
|
|
||||||
"analysis_type must be one of {}".format(self.allowed_analysis_types)
|
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:
|
else:
|
||||||
raise ValueError(
|
decoded = self.summary_model.tokenizer.batch_decode(
|
||||||
"Model type is not allowed - please select one of {}".format(
|
generated_ids,
|
||||||
self.all_allowed_model_types
|
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(
|
results = [d.strip() for d in decoded]
|
||||||
self, list_of_questions: list[str], consequential_questions: bool = False
|
return results
|
||||||
) -> dict:
|
|
||||||
|
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:
|
Args:
|
||||||
list_of_questions (list[str]): list of questions.
|
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:
|
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 is_concise_answer:
|
||||||
if len(list_of_questions) > 0:
|
gen_conf = GenerationConfig(max_new_tokens=64, do_sample=False)
|
||||||
path = self.subdict["filename"]
|
for i in range(len(list_of_questions)):
|
||||||
raw_image = Image.open(path).convert("RGB")
|
if not list_of_questions[i].strip().endswith("?"):
|
||||||
image = (
|
list_of_questions[i] = list_of_questions[i].strip() + "?"
|
||||||
vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device)
|
if not list_of_questions[i].lower().startswith("answer concisely"):
|
||||||
)
|
list_of_questions[i] = "Answer concisely: " + list_of_questions[i]
|
||||||
question_batch = []
|
else:
|
||||||
list_of_questions_processed = []
|
gen_conf = GenerationConfig(max_new_tokens=128, do_sample=False)
|
||||||
|
|
||||||
if model_old:
|
question_chunk_size = 8
|
||||||
for quest in list_of_questions:
|
answers: List[str] = []
|
||||||
list_of_questions_processed.append(txt_processors["eval"](quest))
|
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:
|
else:
|
||||||
for quest in list_of_questions:
|
decoded = self.summary_model.tokenizer.batch_decode(
|
||||||
list_of_questions_processed.append((str)(quest))
|
out_ids,
|
||||||
|
skip_special_tokens=True,
|
||||||
for quest in list_of_questions_processed:
|
clean_up_tokenization_spaces=False,
|
||||||
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"
|
|
||||||
)
|
)
|
||||||
else:
|
|
||||||
print("Please, enter list of questions")
|
|
||||||
return self.subdict
|
|
||||||
|
|
||||||
def check_model(self):
|
answers.extend([d.strip() for d in decoded])
|
||||||
"""
|
|
||||||
Check model type and return appropriate model and preprocessors.
|
|
||||||
|
|
||||||
Args:
|
if len(answers) != len(list_of_questions):
|
||||||
|
|
||||||
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:
|
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Model type is not allowed - please select one of {}".format(
|
f"Expected {len(list_of_questions)} answers, but got {len(answers)}, try vary amount of questions"
|
||||||
self.all_allowed_model_types
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
return model, vis_processors, txt_processors, model_old
|
return answers
|
||||||
|
|
||||||
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
|
|
||||||
|
|||||||
Загрузка…
x
Ссылка в новой задаче
Block a user