locsec/AMMICO
1
Форкнуть 0
зеркало из https://github.com/ssciwr/AMMICO.git synced 2025-10-30 05:26:05 +02:00
Код Задачи Пакеты Проекты Релизы Вики Активность

added new models from lavis to ammico summary (#138)

Просмотр исходного кода 
* added new models from LAVIS to ammico summary

* added sequential questions for summary in new models

* fixed initializing dict process in all notebooks

* joining old and new models into one notebook
Этот коммит содержится в:
Petr Andriushchenko 2023-09-18 13:47:45 +02:00 коммит произвёл GitHub
родитель 5c72f9aae4
Коммит 8161164e87
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
8 изменённых файлов: 728 добавлений и 192 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

2
ammico/display.py
Просмотреть файл

@ -510,7 +510,7 @@ class AnalysisExplorer:
detector_class = identify_function(
image_copy,
analysis_type=setting_summary_analysis_type,
summary_model_type=setting_summary_model,
model_type=setting_summary_model,
list_of_questions=[setting_summary_list_of_questions]
if (setting_summary_list_of_questions is not None)
else None,

23
ammico/notebooks/facial_expressions.ipynb
Просмотреть файл

@ -61,7 +61,7 @@
"id": "a2bd2153",
"metadata": {},
"source": [
"We select a subset of image files to try facial expression detection on, see the `limit` keyword. The `find_files` function finds image files within a given directory:"
"We select a subset of image files to try facial expression detection on, see the `limit` keyword. The `find_files` function finds image files within a given directory and initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis::"
]
},
{
@ -73,31 +73,12 @@
"source": [
"# Here you need to provide the path to your google drive folder\n",
"# or local folder containing the images\n",
"images = ammico.find_files(\n",
"mydict = ammico.find_files(\n",
" path=\"/content/drive/MyDrive/misinformation-data/\",\n",
" limit=10,\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"id": "705e7328",
"metadata": {},
"source": [
"We need to initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b37c0c91",
"metadata": {},
"outputs": [],
"source": [
"mydict = ammico.initialize_dict(images)"
]
},
{
"attachments": {},
"cell_type": "markdown",

25
ammico/notebooks/get-text-from-image.ipynb
Просмотреть файл

@ -64,7 +64,7 @@
"id": "fddba721",
"metadata": {},
"source": [
"We select a subset of image files to try the text extraction on, see the `limit` keyword. The `find_files` function finds image files within a given directory: "
"We select a subset of image files to try the text extraction on, see the `limit` keyword. The `find_files` function finds image files within a given directory and initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis: "
]
},
{
@ -76,28 +76,11 @@
"source": [
"# Here you need to provide the path to your google drive folder\n",
"# or local folder containing the images\n",
"images = ammico.find_files(\n",
"mydict = ammico.find_files(\n",
" path=\"/content/drive/MyDrive/misinformation-data/\",\n",
" limit=10,\n",
")"
]
},
{
"cell_type": "markdown",
"id": "3a7dfe11",
"metadata": {},
"source": [
"We need to initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "8b32409f",
"metadata": {},
"outputs": [],
"source": [
"mydict = ammico.initialize_dict(images)"
")\n",
"mydict"
]
},
{

308
ammico/notebooks/image_summary.ipynb
Просмотреть файл

@ -63,21 +63,12 @@
"source": [
"# Here you need to provide the path to your google drive folder\n",
"# or local folder containing the images\n",
"images = ammico.find_files(\n",
"mydict = ammico.find_files(\n",
" path=\"/content/drive/MyDrive/misinformation-data/\",\n",
" limit=10,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"mydict = ammico.initialize_dict(images)"
" #path=\"../../data/images/\",\n",
" limit=2,\n",
")\n",
"mydict"
]
},
{
@ -112,9 +103,8 @@
},
"outputs": [],
"source": [
"obj = ammico.SummaryDetector(mydict)\n",
"summary_model, summary_vis_processors = obj.load_model(model_type=\"base\") # here we load the base model to the memory. This can dramatically speed up the calculation process then.\n",
"# summary_model, summary_vis_processors = ammico.load_model(\"large\")"
"obj = ammico.SummaryDetector(mydict, analysis_type=\"summary\", model_type=\"base\") # here we load the base model to the memory. This can dramatically speed up the calculation process then.\n",
"#obj = ammico.SummaryDetector(mydict, analysis_type=\"summary\", model_type=\"large\")"
]
},
{
@ -126,12 +116,16 @@
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = ammico.SummaryDetector(\n",
" mydict[key], # here we pass the dictionary containing the images\n",
" analysis_type=\"summary\", # here we specify the type of analysis to perform (summary, questions, summary_and_questions)\n",
" summary_model=summary_model, # here we pass the model to use for the analysis\n",
" summary_vis_processors=summary_vis_processors # here we pass the visual processors to use for the analysis\n",
" ).analyse_image()"
" mydict[key] = obj.analyse_image(analysis_type=\"summary\", subdict = mydict[key])"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
@ -229,11 +223,7 @@
"metadata": {},
"outputs": [],
"source": [
"(\n",
" summary_vqa_model, \n",
" summary_vqa_vis_processors, \n",
" summary_vqa_txt_processors \n",
") = obj.load_vqa_model() # here we load the VQA model to the memory. This can dramatically speed up the calculation process then.\n"
"obj = ammico.SummaryDetector(mydict, analysis_type=\"questions\", list_of_questions = list_of_questions)"
]
},
{
@ -243,13 +233,16 @@
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = ammico.SummaryDetector(\n",
" mydict[key],\n",
" analysis_type=\"questions\",\n",
" summary_vqa_model=summary_vqa_model,\n",
" summary_vqa_vis_processors=summary_vqa_vis_processors,\n",
" summary_vqa_txt_processors=summary_vqa_txt_processors, \n",
" ).analyse_questions(list_of_questions)"
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions = list_of_questions)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
@ -260,6 +253,15 @@
"Or you can perform two types of analysis at a time `analysis_type=\"summary_and_questions\"`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"obj = ammico.SummaryDetector(mydict, analysis_type=\"summary_and_questions\", model_type=\"base\", list_of_questions = list_of_questions)"
]
},
{
"cell_type": "code",
"execution_count": null,
@ -267,15 +269,7 @@
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = ammico.SummaryDetector(\n",
" mydict[key],\n",
" analysis_type=\"summary_and_questions\",\n",
" summary_model=summary_model, \n",
" summary_vis_processors=summary_vis_processors,\n",
" summary_vqa_model=summary_vqa_model,\n",
" summary_vqa_vis_processors=summary_vqa_vis_processors,\n",
" summary_vqa_txt_processors=summary_vqa_txt_processors, \n",
" ).analyse_questions(list_of_questions)"
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"summary_and_questions\", list_of_questions = list_of_questions)"
]
},
{
@ -336,6 +330,230 @@
"analysis_explorer = ammico.AnalysisExplorer(mydict)\n",
"analysis_explorer.run_server(port=8055)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# New models\n",
"This is very heavy models. They requare approx 60GB of RAM and they can use 20+GB memory GPUs for acceleration."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"obj = ammico.SummaryDetector(mydict, analysis_type = \"summary_and_questions\", model_type = \"blip2_t5_caption_coco_flant5xl\", device_type= \"cpu\")\n",
"# list of the new models that can be used:\n",
"# \"blip2_t5_pretrain_flant5xxl\",\n",
"# \"blip2_t5_pretrain_flant5xl\",\n",
"# \"blip2_t5_caption_coco_flant5xl\",\n",
"# \"blip2_opt_pretrain_opt2.7b\",\n",
"# \"blip2_opt_pretrain_opt6.7b\",\n",
"# \"blip2_opt_caption_coco_opt2.7b\",\n",
"# \"blip2_opt_caption_coco_opt6.7b\",\n",
"\n",
"# You can use `pretrain_` model types for zero-shot image-to-text generation with prompts.\n",
"# Or you can use `caption_coco_`` model types to generate coco-style captions.\n",
"# `flant5` and `opt` means that the model equipped with FlanT5 and OPT LLMs respectively.\n",
"\n",
"#also you can perform all calculation on cpu if you set device_type= \"cpu\" or gpu if you set device_type= \"cuda\""
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"summary_and_questions\")\n",
"\n",
"# analysis_type can be \n",
"# \"summary\",\n",
"# \"questions\",\n",
"# \"summary_and_questions\"."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You can also pass a list of questions to this cell if `analysis_type=\"summary_and_questions\"` or `analysis_type=\"questions\"`. But the format of questions has changed in new models. \n",
"\n",
"Here is an example of a list of questions:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"list_of_questions = [\n",
" \"Question: Are there people in the image? Answer:\",\n",
" \"Question: What is this picture about? Answer:\",\n",
"]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You can also pass a question with previous answers as context into this model and pass in questions like this one to get a more accurate answer:\n",
"\n",
"You can combine as many questions as you want in a single query as a list."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"list_of_questions = [\n",
" \"Question: What country is in the picture? Answer: USA. Question: Why? Answer: Because there is an American flag in the background . Question: Where it comes from? Answer:\",\n",
" \"Question: Which city is this? Answer: Frankfurt. Question: why?\",\n",
"]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You can also ask sequential questions if you pass the argument `cosequential_questions=True`. This means that the answers to previous questions will be passed as context to the next question. However, this method will work a bit slower, because for each image the answers to the questions will not be calculated simultaneously, but sequentially. "
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"list_of_questions = [\n",
" \"Question: Is this picture taken inside or outside? Answer:\",\n",
" \"Question: Why? Answer:\",\n",
"]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"for key in mydict:\n",
" mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions, consequential_questions=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Convert to dataframe and write csv\n",
"\n",
"Convert the dictionary of dictionarys into a dictionary with lists:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"outdict = ammico.append_data_to_dict(mydict)\n",
"df = ammico.dump_df(outdict)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Check the dataframe:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"df.head(10)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Write the csv file:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"df.to_csv(\"/content/drive/MyDrive/misinformation-data/data_out.csv\")"
]
}
],
"metadata": {
@ -354,7 +572,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.3"
"version": "3.10.13"
},
"vscode": {
"interpreter": {

19
ammico/notebooks/multimodal_search.ipynb
Просмотреть файл

@ -66,22 +66,11 @@
},
"outputs": [],
"source": [
"images = ammico.find_files(\n",
"mydict = ammico.find_files(\n",
" path=\"/content/drive/MyDrive/misinformation-data/\",\n",
" limit=10,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "adf3db21-1f8b-4d44-bbef-ef0acf4623a0",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"mydict = ammico.initialize_dict(images)"
")\n",
"mydict"
]
},
{
@ -497,7 +486,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.3"
"version": "3.10.13"
}
},
"nbformat": 4,

14
ammico/notebooks/objects_expression.ipynb
Просмотреть файл

@ -67,19 +67,11 @@
"source": [
"# Here you need to provide the path to your google drive folder\n",
"# or local folder containing the images\n",
"images = ammico.find_files(\n",
"mydict = ammico.find_files(\n",
" path=\"/content/drive/MyDrive/misinformation-data/\",\n",
" limit=10,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"mydict = ammico.initialize_dict(images)"
")\n",
"mydict"
]
},
{

505
ammico/summary.py
Просмотреть файл

@ -2,27 +2,56 @@ from ammico.utils import AnalysisMethod
from torch import cuda, no_grad
from PIL import Image
from lavis.models import load_model_and_preprocess
from typing import Optional
class SummaryDetector(AnalysisMethod):
allowed_model_types = [
"base",
"large",
]
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__(
self,
subdict: dict = {},
summary_model_type: str = "base",
model_type: str = "base",
analysis_type: str = "summary_and_questions",
list_of_questions: str = None,
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.
Args:
subdict (dict, optional): Dictionary containing the image to be analysed. Defaults to {}.
summary_model_type (str, optional): Type of blip_caption model to use. Can be "base" or "large". Defaults to "base".
model_type (str, optional): Type of model to use. Can be "base" or "large" for blip_caption. 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.
@ -30,6 +59,9 @@ class SummaryDetector(AnalysisMethod):
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".
@ -39,47 +71,67 @@ class SummaryDetector(AnalysisMethod):
"""
super().__init__(subdict)
if analysis_type not in ["summary", "questions", "summary_and_questions"]:
# check if analysis_type is valid
if analysis_type not in self.allowed_analysis_types:
raise ValueError(
"analysis_type must be one of 'summary', 'questions' or 'summary_and_questions'"
"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"
allowed_model_types = ["base", "large"]
if summary_model_type not in allowed_model_types:
elif device_type not in ["cuda", "cpu"]:
raise ValueError("device_type must be one of {}".format(["cuda", "cpu"]))
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(
allowed_model_types
self.all_allowed_model_types
)
)
self.summary_model_type = summary_model_type
self.model_type = model_type
self.analysis_type = analysis_type
if list_of_questions is None:
# 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?",
]
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)")
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 (
(summary_model is None)
model_type in self.allowed_model_types
and (summary_model is None)
and (summary_vis_processors is None)
and (analysis_type != "questions")
and (analysis_type == "summary" or analysis_type == "summary_and_questions")
):
self.summary_model, self.summary_vis_processors = self.load_model(
model_type=summary_model_type
model_type=model_type
)
else:
self.summary_model = summary_model
self.summary_vis_processors = summary_vis_processors
if (
(summary_vqa_model is None)
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 != "summary")
and (
analysis_type == "questions" or analysis_type == "summary_and_questions"
)
):
(
self.summary_vqa_model,
@ -90,6 +142,21 @@ class SummaryDetector(AnalysisMethod):
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):
"""
@ -98,8 +165,8 @@ class SummaryDetector(AnalysisMethod):
Args:
Returns:
model (torch.nn.Module): model.
vis_processors (dict): preprocessors for visual inputs.
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",
@ -116,8 +183,8 @@ class SummaryDetector(AnalysisMethod):
Args:
Returns:
model (torch.nn.Module): model.
vis_processors (dict): preprocessors for visual inputs.
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",
@ -135,8 +202,8 @@ class SummaryDetector(AnalysisMethod):
model_type (str): type of the model.
Returns:
model (torch.nn.Module): model.
vis_processors (dict): preprocessors for visual inputs.
summary_model (torch.nn.Module): model.
summary_vis_processors (dict): preprocessors for visual inputs.
"""
select_model = {
"base": SummaryDetector.load_model_base,
@ -152,9 +219,9 @@ class SummaryDetector(AnalysisMethod):
Args:
Returns:
model (torch.nn.Module): model.
vis_processors (dict): preprocessors for visual inputs.
txt_processors (dict): preprocessors for text inputs.
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.
"""
(
@ -169,99 +236,405 @@ class SummaryDetector(AnalysisMethod):
)
return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors
def analyse_image(self):
def analyse_image(
self,
analysis_type: Optional[str] = None,
subdict: dict = None,
list_of_questions: Optional[list[str]] = None,
consequential_questions: bool = False,
):
"""
Analyse image with blip_caption 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.
Returns:
self.subdict (dict): dictionary with analysis results.
"""
if self.analysis_type == "summary_and_questions":
self.analyse_summary()
self.analyse_questions(self.list_of_questions)
elif self.analysis_type == "summary":
self.analyse_summary()
elif self.analysis_type == "questions":
self.analyse_questions(self.list_of_questions)
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
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
)
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
)
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)
)
return self.subdict
def analyse_summary(self):
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
)
)
path = self.subdict["filename"]
raw_image = Image.open(path).convert("RGB")
image = (
self.summary_vis_processors["eval"](raw_image)
.unsqueeze(0)
.to(self.summary_device)
)
image = vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device)
with no_grad():
self.subdict["const_image_summary"] = self.summary_model.generate(
{"image": image}
)[0]
self.subdict["3_non-deterministic summary"] = self.summary_model.generate(
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]) -> dict:
def analyse_questions(
self, list_of_questions: list[str], consequential_questions: bool = False
) -> dict:
"""
Generate answers to free-form questions about image written in natural language.
Args:
list_of_questions (list[str]): list of questions.
consequential_questions (bool): whether to ask consequential questions. Works only for new BLIP2 models.
Returns:
self.subdict (dict): dictionary with answers to questions.
"""
if (
(self.summary_vqa_model is None)
and (self.summary_vqa_vis_processors is None)
and (self.summary_vqa_txt_processors is None)
):
(
self.summary_vqa_model,
self.summary_vqa_vis_processors,
self.summary_vqa_txt_processors,
) = load_model_and_preprocess(
name="blip_vqa",
model_type="vqav2",
is_eval=True,
device=self.summary_device,
)
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 = (
self.summary_vqa_vis_processors["eval"](raw_image)
.unsqueeze(0)
.to(self.summary_device)
vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device)
)
question_batch = []
list_of_questions_processed = []
if model_old:
for quest in list_of_questions:
question_batch.append(self.summary_vqa_txt_processors["eval"](quest))
list_of_questions_processed.append(txt_processors["eval"](quest))
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():
answers_batch = self.summary_vqa_model.predict_answers(
samples={"image": image_batch, "text_input": question_batch},
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):
"""
Check model type and return appropriate model and preprocessors.
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:
raise ValueError(
"Model type is not allowed - please select one of {}".format(
self.all_allowed_model_types
)
)
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

4
ammico/test/test_summary.py
Просмотреть файл

@ -125,11 +125,11 @@ def test_advanced_init_summary():
sd = sm.SummaryDetector({})
assert sd.summary_model
assert sd.summary_vis_processors
sd = sm.SummaryDetector({}, summary_model_type="large")
sd = sm.SummaryDetector({}, model_type="large")
assert sd.summary_model
assert sd.summary_vis_processors
with pytest.raises(ValueError):
sm.SummaryDetector({}, summary_model_type="bla")
sm.SummaryDetector({}, model_type="bla")
(
summary_vqa_model,
summary_vqa_vis_processors,