ImageBind と OpenVINO を使用したマルチモーダル・データのバインド#

import ipywidgets as widgets 

device = widgets.Dropdown( 
    options=core.available_devices + ["AUTO"], 
    value="AUTO", 
    description="Device:", 
    disabled=False, 
) 

device

ov_modality_models = {} 

modalities = [ModalityType.TEXT, ModalityType.VISION, ModalityType.AUDIO] 
for modality in modalities: 
    export_dir = Path(f"image-bind-{modality}") 
    file_name = f"image-bind-{modality}" 
    export_dir.mkdir(exist_ok=True) 
    ir_path = export_dir / f"{file_name}.xml" 
    if not ir_path.exists(): 
        exportable_model = ModelExporter(model, modality) 
        model_input = inputs[modality] 
        ov_model = ov.convert_model(exportable_model, example_input=model_input) 
        ov.save_model(ov_model, ir_path) 
    else: 
        ov_model = core.read_model(ir_path) 
    ov_modality_models[modality] = core.compile_model(ov_model, device.value)

text_vision_scores = softmax(embeddings[ModalityType.VISION] @ embeddings[ModalityType.TEXT].T, axis=-1) 

visualize_prob_matrix(text_vision_scores, text_list, image_list)

text_audio_scores = softmax(embeddings[ModalityType.AUDIO] @ embeddings[ModalityType.TEXT].T, axis=-1) 

visualize_prob_matrix(text_audio_scores, text_list, audio_list)

audio_vision_scores = softmax(embeddings[ModalityType.VISION] @ embeddings[ModalityType.AUDIO].T, axis=-1) 

visualize_prob_matrix(audio_vision_scores, image_list, audio_list)

import IPython.display as ipd 
from PIL import Image 

text_image_ids = np.argmax(text_vision_scores, axis=0) 
text_audio_ids = np.argmax(text_audio_scores, axis=0) 
print( 
    f"Predicted label: {text_list[0]} \nprobability for image - {text_vision_scores[text_image_ids[0], 0]:.3f}\nprobability for audio - {text_audio_scores[0, text_audio_ids[0]]:.3f}" 
) 
display(Image.open(image_paths[text_image_ids[0]])) 
ipd.Audio(audio_paths[text_audio_ids[0]])

print( 
    f"Predicted label: {text_list[1]} \nprobability for image - {text_vision_scores[text_image_ids[1], 1]:.3f}\nprobability for audio - {text_audio_scores[1, text_audio_ids[1]]:.3f}" 
) 
display(Image.open(image_paths[text_image_ids[1]])) 
ipd.Audio(audio_paths[text_audio_ids[1]])

print( 
    f"Predicted label: {text_list[2]} \nprobability for image - {text_vision_scores[text_image_ids[2], 2]:.3f}\nprobability for audio - {text_audio_scores[2, text_audio_ids[2]]:.3f}" 
) 
display(Image.open(image_paths[text_image_ids[2]])) 
ipd.Audio(audio_paths[text_audio_ids[2]])

import imagebind.data as data 
import os 
import requests 
import tempfile 

from requests.packages.urllib3.exceptions import InsecureRequestWarning 

requests.packages.urllib3.disable_warnings(InsecureRequestWarning) 

def check_text_data(data): 
    """ 
    Check if the given data is text-based.
    """ 
    if isinstance(data, str): 
        return True 
    if isinstance(data, list): 
        return all(isinstance(x, str) for x in data) 
    return False 

def collate_fn(examples, image_column="image_url", text_column="caption"): 
    """ 
    Collates examples into a batch for processing. 
    Preprocesses each example by loading and transforming image and text data. 
    Checks if the text data in the example is valid by calling the `check_text_data` function. 
    Downloads the image specified by the URL in the image_column of the example dictionary. 
    Constructs and returns a dictionary representing the collated batch with the following keys: 
        - "pixel_values": The pixel values of the preprocessed example. 
        - "input_ids": The transformed text data of the preprocessed example.
    """ 
    assert len(examples) == 1 
    example = examples[0] 
    if not check_text_data(example[text_column]): 
        raise ValueError("Text data is not valid") 

    url = example[image_column] 
    with tempfile.TemporaryDirectory() as tempdir: 
        f_name = os.path.join(tempdir, "image.jpg") 
        try: 
            response = requests.get(url, verify=False, timeout=20) 
            with open(f_name, "wb") as file: 
                file.write(response.content) 
            pixel_values = data.load_and_transform_vision_data([f_name], "cpu") 
        except Exception: 
            print(f"Can't load image from url: {url}") 
            return None 

text = data.load_and_transform_text([example[text_column]], "cpu") 

return {"pixel_values": pixel_values, "input_ids": text}

from datasets import load_dataset 
import itertools 
import torch 
from tqdm.notebook import tqdm 

def collect_vision_text_data(dataloader, init_steps): 
    """ 
    This function collects vision and text data from a dataloader for a specified number of initialization steps. 
    It iterates over the dataloader, fetching batches and storing the relevant vision and text data. 
    Returns a tuple containing the collected vision_data and text_data lists.
    """ 
    text_data = [] 
    vision_data = [] 
    print(f"Fetching {init_steps} for the initialization...") 
    counter = 0 
    for batch in tqdm(dataloader): 
        if counter == init_steps: 
            break 
        with torch.no_grad(): 
            if batch: 
                counter += 1 
                text_data.append(batch["input_ids"].to("cpu")) 
                vision_data.append(batch["pixel_values"].to("cpu")) 
    return vision_data, text_data 

def prepare_vision_text_dataset(opt_init_steps=50): 
    """ 
    Prepares a vision-text dataset for quantization by collecting vision and text data.
    """ 
    dataset = load_dataset("google-research-datasets/conceptual_captions", streaming=False, trust_remote_code=True) 
    train_dataset = dataset["train"].shuffle(seed=0) dataloader = torch.utils.data.DataLoader(train_dataset, collate_fn=collate_fn, batch_size=1) 
    vision_data, text_data = collect_vision_text_data(dataloader, opt_init_steps) 
    return vision_data, text_data

import numpy as np 
import torchaudio 

def collect_audio_data(dataloader, init_steps=300): 
    """ 
    This function collects audio data from a dataloader for a specified number of initialization steps. 
    It iterates over the dataloader, fetching batches and storing them in a list.
    """ 
    audio_data = [] 
    for _, batch in tqdm(zip(range(init_steps), itertools.islice(dataloader, 0, init_steps))): 
        with torch.no_grad(): 
            audio_data.append(batch) 
    return audio_data 

def prepare_audio_dataset(): 
    """ 
    Prepares an "ashraq/esc50" audio dataset for quantization by collecting audio data. 
    Collects audio data from the dataloader by calling the `collect_audio_data` function. 
    Returns a list containing the collected calibration audio data batches.
    """ 
    audio_dataset = load_dataset("ashraq/esc50", streaming=True, trust_remote_code=True) 
    train_dataset = audio_dataset["train"].shuffle(seed=42, buffer_size=1000) 

    def collate_fn(examples): 
        assert len(examples) == 1 
        with tempfile.TemporaryDirectory() as tempdir: 
            f_name = os.path.join(tempdir, "audio.wav") 
            audio_data = examples[0]["audio"]["array"] 
            sample_rate = examples[0]["audio"]["sampling_rate"] 
            audio_data = torch.from_numpy(audio_data).to(torch.float32).unsqueeze(0) 
            torchaudio.save(f_name, audio_data, sample_rate) 
            return data.load_and_transform_audio_data([f_name], "cpu") 

dataloader = torch.utils.data.DataLoader(train_dataset, collate_fn=collate_fn, batch_size=1) 
calibration_data = collect_audio_data(dataloader) 
return calibration_data

vision_data, text_data = [], [] 

if not int8_model_paths[ModalityType.TEXT].exists() or not int8_model_paths[ModalityType.VISION].exists(): 
    vision_data, text_data = prepare_vision_text_dataset()

import logging 
import nncf 
import openvino as ov 

nncf.set_log_level(logging.ERROR) 

core = ov.Core() 

def quantize_openvino_model(modality, calibration_data): 
    model_path = fp_model_paths[modality] 
    model = core.read_model(model_path) 
    quantized_model = nncf.quantize( 
        model=model, 
        calibration_dataset=calibration_data, 
        model_type=nncf.ModelType.TRANSFORMER, 
    ) 
    ov.save_model(quantized_model, int8_model_paths[modality]) 
    return quantized_model

# 入力を準備 

text_list = ["A car", "A bird", "A dog"] 
image_paths = [ 
    ".assets/dog_image.jpg", 
    ".assets/car_image.jpg", 
    ".assets/bird_image.jpg", 
] 
audio_paths = [ 
    ".assets/dog_audio.wav", 
    ".assets/bird_audio.wav", 
    ".assets/car_audio.wav", 
] 

inputs = { 
    ModalityType.TEXT: data.load_and_transform_text(text_list, "cpu"), 
    ModalityType.VISION: data.load_and_transform_vision_data(image_paths, "cpu"), 
    ModalityType.AUDIO: data.load_and_transform_audio_data(audio_paths, "cpu"), 
}

def calculate_compression_rate(modality): 
    fp16_ir_model_size = Path(fp_model_paths[modality]).with_suffix(".bin").stat().st_size / 1024 
    quantized_model_size = Path(int8_model_paths[modality]).with_suffix(".bin").stat().st_size / 1024 
    print(f"Modality: {modality}") 
    print(f" * FP16 IR model size: {fp16_ir_model_size:.2f} KB") 
    print(f" * INT8 model size: {quantized_model_size:.2f} KB") 
    print(f" * Model compression rate: {fp16_ir_model_size / quantized_model_size:.3f}") 

for modality in modalities: 
    calculate_compression_rate(modality)

import time 

def calculate_inference_time(model_path, calibration_data): 
    model = core.compile_model(model_path) 
    output_layer = model.output(0) 
    inference_time = [] 
    for batch in calibration_data: 
        start = time.perf_counter() 
        _ = model(batch)[output_layer] 
        end = time.perf_counter() 
        delta = end - start 
        inference_time.append(delta) 
    return np.median(inference_time)