Video-LLaVA と OpenVINO による視覚言語アシスタント#

%pip install -q torch "torchvision<0.17.0" "transformers>=4.31.0,<4.35.0" "pytorchvideo" "einops" "peft==0.6.2" --extra-index-url https://download.pytorch.org/whl/cpu 
%pip install -q opencv_python decord sentencepiece protobuf "openvino>=2023.2.0" "nncf>=2.7.0" "gradio>=4.19"

from pathlib import Path 
import sys 

repo_dir = Path("Video-LLaVA") 

if not repo_dir.exists():
     !git clone https://github.com/PKU-YuanGroup/Video-LLaVA.git 

sys.path.insert(0, str(repo_dir.resolve()))

import gc 

import transformers 
from videollava.model import LlavaLlamaForCausalLM 
from videollava.constants import ( 
    DEFAULT_IMAGE_PATCH_TOKEN, 
    DEFAULT_VIDEO_PATCH_TOKEN, 
    DEFAULT_IM_START_TOKEN, 
    DEFAULT_VID_START_TOKEN, 
    DEFAULT_IM_END_TOKEN, 
    DEFAULT_VID_END_TOKEN, 
    DEFAULT_IMAGE_TOKEN, 
) 

transformers.logging.set_verbosity_error() 

model_id = "LanguageBind/Video-LLaVA-7B" 

config = transformers.AutoConfig.from_pretrained(model_id) 

tokenizer = transformers.AutoTokenizer.from_pretrained(model_id) 
model = LlavaLlamaForCausalLM.from_pretrained(model_id) 
image_tower = model.get_image_tower() 
video_tower = model.get_video_tower() 
image_tower.load_model() video_tower.load_model() 
image_processor = image_tower.image_processor 
video_processor = video_tower.video_processor 
mm_use_im_start_end = getattr(config, "mm_use_im_start_end", False) 
mm_use_im_patch_token = getattr(config, "mm_use_im_patch_token", True) 
if mm_use_im_patch_token: 
    tokenizer.add_tokens([DEFAULT_IMAGE_PATCH_TOKEN], special_tokens=True) 
    tokenizer.add_tokens([DEFAULT_VIDEO_PATCH_TOKEN], special_tokens=True) 
if mm_use_im_start_end: 
    tokenizer.add_tokens([DEFAULT_IM_START_TOKEN, DEFAULT_IM_END_TOKEN], special_tokens=True) 
    tokenizer.add_tokens([DEFAULT_VID_START_TOKEN, DEFAULT_VID_END_TOKEN], special_tokens=True) 
preprocess_fn = model.prepare_inputs_labels_for_multimodal 

del model 
gc.collect()

from transformers.generation import GenerationConfig, GenerationMixin 
from transformers.modeling_outputs import CausalLMOutputWithPast 
import numpy as np 
import torch 

class OVLlavaLlamaForCausalLM(GenerationMixin): 
    def __init__(self, core, model_dir, device): 
        self.model = core.read_model(model_dir / "videollava_with_past.xml") 
        self.model_input_embed = core.compile_model(model_dir / "videollava_input_embed.xml", device) 
        self.input_names = {key.get_any_name(): idx for idx, key in enumerate(self.model.inputs)} 
        self.output_names = {key.get_any_name(): idx for idx, key in enumerate(self.model.outputs)} 
        self.key_value_input_names = [key for key in self.input_names if "key_values" in key] self.key_value_output_names = [key for key in self.output_names if "present" in key] 
        compiled_model = core.compile_model(self.model, device) 
        self.request = compiled_model.create_infer_request() 
        self.config = transformers.AutoConfig.from_pretrained(model_dir) 
        self.generation_config = GenerationConfig.from_model_config(config) 
        self.main_input_name = "input_ids" 
        self.device = torch.device("cpu") 
        self.num_pkv = 2 
        self._supports_cache_class = False 

    def can_generate(self): 
        """Returns True to validate the check that the model using `GenerationMixin.generate()` can indeed generate.
        """ 
        return True 

    def __call__( 
        self, 
        input_ids: torch.LongTensor, 
        images: torch.Tensor, 
        attention_mask: Optional[torch.LongTensor] = None, 
        prefix_mask: Optional[torch.LongTensor] = None, 
        past_key_values: Optional[Tuple[Tuple[torch.FloatTensor]]] = None, 
        **kwargs, 
    ) -> CausalLMOutputWithPast: 
        return self.forward(input_ids, images, attention_mask, prefix_mask, past_key_values) 

    def forward( 
        self, 
        input_ids: torch.LongTensor, 
        images: torch.Tensor, 
        attention_mask: Optional[torch.LongTensor] = None, 
        prefix_mask: Optional[torch.LongTensor] = None, 
        past_key_values: Optional[Tuple[Tuple[torch.FloatTensor]]] = None, 
        **kwargs, 
    ) -> CausalLMOutputWithPast: 
        """General inference method""" 
        inputs = {} 
        if past_key_values is not None:
            # past_key_values をフラット化 
            attention_mask = torch.ones( 
                (input_ids.shape[0], past_key_values[-1][-1].shape[-2] + 1), 
                dtype=input_ids.dtype, 
            ) 
            past_key_values = (past_key_value for pkv_per_layer in past_key_values for past_key_value in pkv_per_layer) 
            # past_key_values をデコーダー入力に追加 
            inputs = dict(zip(self.key_value_input_names, past_key_values)) 

        else: 
            return self.forward_with_image(input_ids, images, attention_mask) inputs["input_ids"] = np.array(input_ids) 

        if "attention_mask" in self.input_names: 
            inputs["attention_mask"] = np.array(attention_mask) 

        # 推論を実行 
        self.request.start_async(inputs, share_inputs=True) 
        self.request.wait() 

        logits = torch.from_numpy(self.request.get_tensor("logits").data) 

        # 長さが等しいタプル: レイヤー数 * デコーダーレイヤーごとの past_key_value の数 (2 は自己注意レイヤーに対応) 
        past_key_values = tuple(self.request.get_tensor(key).data for key in self.key_value_output_names) 
        # 長さが `n_layers` のタプルのタプル。各タプルの長さは 2 (自己注意の k/v) に等しい 

        past_key_values = tuple(past_key_values[i : i + self.num_pkv] for i in range(0, len(past_key_values), self.num_pkv)) 
        return CausalLMOutputWithPast(logits=logits, past_key_values=past_key_values) 

    def forward_with_image(self, input_ids, images, attention_mask): 
        """First step inference method, that resolves multimodal data""" 
        _, _, attention_mask, _, input_embeds, _ = preprocess_fn( 
            input_ids=input_ids, 
            position_ids=None, 
            attention_mask=attention_mask, 
            past_key_values=None, 
            labels=None, 
            images=images, 
        ) 
        outs = self.model_input_embed({"inputs_embeds": input_embeds, "attention_mask": attention_mask}) 
        logits = outs[0] 
        pkv = list(outs.values())[1:] pkv = tuple(pkv[i : i + self.num_pkv] for i in range(0, len(pkv), self.num_pkv)) 
        return CausalLMOutputWithPast(logits=torch.from_numpy(logits), past_key_values=pkv) 

    def prepare_inputs_for_generation(self, input_ids, past_key_values=None, **kwargs): 
        """ 
        This function is used during running GenerationMixin.generate for preparing model specific inputs for 
        each generation step 
        """ 
        past_len = 0 
        if past_key_values is not None: 
            input_ids = input_ids[:, -1].unsqueeze(-1) 
            past_len = past_key_values[-1][-1].shape[-2] 
        attention_mask = kwargs.get( 
            "attention_mask", 
            torch.ones(input_ids.shape[0], input_ids.shape[1] + past_len), 
        ) 
        if not kwargs.get("use_cache", True): 
            raise NotImplementedError("MPT with prefix_lm=True does not support use_cache=False.") 
        else: 
            prefix_mask = None 
        return { 
            "input_ids": input_ids, 
            "attention_mask": attention_mask, 
            "prefix_mask": prefix_mask, 
            "past_key_values": past_key_values, 
            "images": kwargs.get("images", None), 
        } 

    def _reorder_cache(self, past_key_values: Tuple[Tuple[torch.Tensor]], beam_idx: torch.Tensor) -> Tuple[Tuple[torch.Tensor]]: 
        """ 
        This function is used to re-order the `past_key_values` cache if [`~PreTrainedModel.beam_search`] or 
        [`~PreTrainedModel.beam_sample`] is called. 
        This is required to match `past_key_values` with the correct beam_idx at every generation step.
        """
 
        # transformers.models.gpt2.modeling_gpt2.GPT2LMHeadModel._reorder_cache から 
        return tuple(tuple(np.take(past_state, beam_idx, 0) for past_state in layer_past) for layer_past in past_key_values)

import torch 
import gradio as gr 

from videollava.constants import DEFAULT_IMAGE_TOKEN, IMAGE_TOKEN_INDEX 
from videollava.conversation import conv_templates, SeparatorStyle 

def generate(image, video, textbox_in): 
    if video is not None: 
        textbox_in = DEFAULT_IMAGE_TOKEN * 8 + "\n" + textbox_in 
        if image is not None: 
            textbox_in += "\n" + DEFAULT_IMAGE_TOKEN 
    elif image is not None: 
        textbox_in = DEFAULT_IMAGE_TOKEN + "\n" + textbox_in 

    conv_mode = "llava_v1" conv = conv_templates[conv_mode].copy() 
    conv.append_message(conv.roles[0], textbox_in) 
    conv.append_message(conv.roles[1], None) 
    prompt = conv.get_prompt() 
    images_tensor = [] 
    if image is not None: 
        images_tensor.append(image_processor(image, return_tensors="pt")["pixel_values"][0]) 
    if video is not None: 
        images_tensor.append(video_processor(video, return_tensors="pt")["pixel_values"][0]) 
    input_ids = tokenizer_image_token(prompt, tokenizer, IMAGE_TOKEN_INDEX, return_tensors="pt").unsqueeze(0) 

    stop_str = conv.sep if conv.sep_style != SeparatorStyle.TWO else conv.sep2 
    keywords = [stop_str] 
    stopping_criteria = KeywordsStoppingCriteria(keywords, tokenizer, input_ids) 

    generate_kwargs = dict( 
        input_ids=input_ids, 
        images=images_tensor, 
        max_new_tokens=1024, 
        temperature=0.2, 
        do_sample=True, 
        use_cache=True, 
        stopping_criteria=[stopping_criteria], 
    ) 

    output_ids = ov_model.generate(**generate_kwargs) 

    input_token_len = input_ids.shape[1] 
    outputs = tokenizer.batch_decode(output_ids[:, input_token_len:], skip_special_tokens=True)[0] 
    outputs = outputs.strip() 
    if outputs.endswith(stop_str): 
        outputs = outputs[: -len(stop_str)] 
        outputs = outputs.strip() 

        return outputs 

demo = gr.Interface( 
    generate, 
    [ 
        gr.Image(label="Input Image", type="filepath"), 
        gr.Video(label="Input Video"), 
        gr.Textbox(label="Question"), 
    ], 
    gr.Textbox(lines=10), 
    examples=[ 
        [ 
            f"{examples_dir}/extreme_ironing.jpg", 
            None, 
            "What is unusual about this image?", 
        ], 
        [ 
            f"{examples_dir}/waterview.jpg", 
            None, 
            "What are the things I should be cautious about when I visit here?", 
        ], 
        [ 
            f"{examples_dir}/desert.jpg", 
            None, 
            "If there are factual errors in the questions, point it out; if not, proceed answering the question.What’s happening in the desert?", 
        ], 
        [ 
            None, 
            f"{examples_dir}/sample_demo_1.mp4", 
            "Why is this video funny?", 
        ], 
        [ 
            None, 
            f"{examples_dir}/sample_demo_3.mp4", 
            "Can you identify any safety hazards in this video?", 
        ], 
        [ 
            None, 
            f"{examples_dir}/sample_demo_9.mp4", 
            "Describe the video.", 
        ], 
        [ 
            None, 
            f"{examples_dir}/sample_demo_22.mp4", 
            "Describe the activity in the video.", 
        ], 
        [ 
            f"{examples_dir}/sample_img_22.png", 
            f"{examples_dir}/sample_demo_22.mp4", 
            "Are the instruments in the pictures used in the video?", 
        ], 
        [ 
            f"{examples_dir}/sample_img_13.png", 
            f"{examples_dir}/sample_demo_13.mp4", 
            "Does the flag in the image appear in the video?", 
        ], 
        [ 
            f"{examples_dir}/sample_img_8.png", 
            f"{examples_dir}/sample_demo_8.mp4", 
            "Are the image and the video depicting the same place?", 
        ], 
    ], 
    title="Video-LLaVA🚀", 
    allow_flagging="never", 
) 
try: 
    demo.queue().launch(debug=False) 
except Exception: 
    demo.queue().launch(share=True, debug=False) 
# リモートで起動する場合は、server_name と server_port を指定 
# demo.launch(server_name='your server name', server_port='server port in int') 
# 詳細はドキュメントをご覧ください: https://gradio.app/docs/