Whisper と OpenVINO™ を使用したビデオ字幕の生成#

from pathlib import Path 

model_dir = model_id.value.split("/")[-1] 

if not Path(model_dir).exists():     !optimum-cli export openvino -m {model_id.value} {model_dir} --weight-format fp16

import openvino as ov 

core = ov.Core()

import ipywidgets as widgets 

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

device

from optimum.intel.openvino import OVModelForSpeechSeq2Seq 
from transformers import AutoProcessor, pipeline 

ov_model = OVModelForSpeechSeq2Seq.from_pretrained(model_dir, device=device.value) 

processor = AutoProcessor.from_pretrained(model_dir) 

pipe = pipeline( 
    "automatic-speech-recognition", 
    model=ov_model, 
    chunk_length_s=30, 
    tokenizer=processor.tokenizer, 
    feature_extractor=processor.feature_extractor, 
)

%%skip not $to_quantize.value 

from itertools import islice 
from optimum.intel.openvino.quantization import InferRequestWrapper 

def collect_calibration_dataset(ov_model: OVModelForSpeechSeq2Seq, calibration_dataset_size: int):
    # モデル要求のプロパティーを上書きし、後で復元できるように元のプロパティーを保存 
    encoder_calibration_data = [] 
    decoder_calibration_data = [] 
    ov_model.encoder.request = InferRequestWrapper(ov_model.encoder.request, encoder_calibration_data, apply_caching=True) 
    ov_model.decoder_with_past.request = InferRequestWrapper(ov_model.decoder_with_past.request, decoder_calibration_data, apply_caching=True) 

    pipe = pipeline( 
        "automatic-speech-recognition", 
        model=ov_model, 
        chunk_length_s=30, 
        tokenizer=processor.tokenizer, 
        feature_extractor=processor.feature_extractor) 
    try: 
        calibration_dataset = dataset = load_dataset("openslr/librispeech_asr", "clean", split="validation", streaming=True, trust_remote_code=True) 
        for sample in tqdm(islice(calibration_dataset, calibration_dataset_size), desc="Collecting calibration data", total=calibration_dataset_size): 
            pipe(sample["audio"], generate_kwargs={"task": task.value}, return_timestamps=True) 
    finally: 
        ov_model.encoder.request = ov_model.encoder.request.request 
        ov_model.decoder_with_past.request = ov_model.decoder_with_past.request.request 

    return encoder_calibration_data, decoder_calibration_data

%%skip not $to_quantize.value 

import gc 
import shutil 
import nncf 
from datasets import load_dataset 
from tqdm.notebook import tqdm 

def extract_input_features(sample): 
    input_features = processor( 
        sample["audio"]["array"], 
        sampling_rate=sample["audio"]["sampling_rate"], 
        return_tensors="pt", 
    ).input_features 
    return input_features 

CALIBRATION_DATASET_SIZE = 50 
quantized_model_path = Path(f"{model_dir}_quantized") 

def quantize(ov_model: OVModelForSpeechSeq2Seq, calibration_dataset_size: int): 
    if not quantized_model_path.exists(): 
        encoder_calibration_data, decoder_calibration_data = collect_calibration_dataset( 
            ov_model, calibration_dataset_size 
        ) 
        print("Quantizing encoder") 
        quantized_encoder = nncf.quantize( 
            ov_model.encoder.model, 
            nncf.Dataset(encoder_calibration_data), 
            subset_size=len(encoder_calibration_data), 
            model_type=nncf.ModelType.TRANSFORMER, 
            # スムーズ・クォンタム・アルゴリズムは活性化量子化誤差を削減し、グリッドサーチを通じて最適なアルファ値を取得 
advanced_parameters=nncf.AdvancedQuantizationParameters(smooth_quant_alpha=0.50) 
        ) 
        ov.save_model(quantized_encoder, quantized_model_path / "openvino_encoder_model.xml") 
        del quantized_encoder 
        del encoder_calibration_data 
        gc.collect() 

        print("Quantizing decoder with past") 
        quantized_decoder_with_past = nncf.quantize( 
            ov_model.decoder_with_past.model, 
            nncf.Dataset(decoder_calibration_data), 
            subset_size=len(decoder_calibration_data), 
            model_type=nncf.ModelType.TRANSFORMER, 
            # スムーズ・クォンタム・アルゴリズムは活性化量子化誤差を削減し、グリッドサーチを通じて最適なアルファ値を取得 
    advanced_parameters=nncf.AdvancedQuantizationParameters(smooth_quant_alpha=0.96) 
        ) 
        ov.save_model(quantized_decoder_with_past, quantized_model_path / "openvino_decoder_with_past_model.xml") 
        del quantized_decoder_with_past 
        del decoder_calibration_data 
        gc.collect() 

        # 設定ファイルと first-step-decoder を手動でコピー 
        model_path = Path(model_dir) 
        shutil.copy(model_path / "config.json", quantized_model_path / "config.json") 
        shutil.copy(model_path / "generation_config.json", quantized_model_path / "generation_config.json") 
        shutil.copy(model_path / "openvino_decoder_model.xml", quantized_model_path / "openvino_decoder_model.xml") 
        shutil.copy(model_path / "openvino_decoder_model.bin", quantized_model_path / "openvino_decoder_model.bin") 

    quantized_ov_model = OVModelForSpeechSeq2Seq.from_pretrained(quantized_model_path, compile=False) 
    quantized_ov_model.to(device.value) quantized_ov_model.compile() 
    return quantized_ov_model 

ov_quantized_model = quantize(ov_model, CALIBRATION_DATASET_SIZE)

if ov_quantized_model is not None: 
    int8_pipe = pipeline( 
        "automatic-speech-recognition", 
        model=ov_quantized_model, 
        chunk_length_s=30, 
        tokenizer=processor.tokenizer, 
        feature_extractor=processor.feature_extractor, 
    ) 
    inputs, duration = get_audio(output_file) 
    transcription = int8_pipe(inputs, generate_kwargs={"task": task.value}, return_timestamps=True)["chunks"] 
    srt_lines = prepare_srt(transcription, filter_duration=duration) 
    print("".join(srt_lines)) 
    widgets.Video.from_file(output_file, loop=False, width=800, height=800)

%%skip not $to_quantize.value 

import time 
from contextlib import contextmanager 
from jiwer import wer, wer_standardize 

TEST_DATASET_SIZE = 50 
MEASURE_TIME = False 

@contextmanager 
def time_measurement(): 
    global MEASURE_TIME 
    try:
        MEASURE_TIME = True 
        yield 
    finally:
        MEASURE_TIME = False 

def time_fn(obj, fn_name, time_list): 
    original_fn = getattr(obj, fn_name) 

    def wrapper(*args, **kwargs): 
        if not MEASURE_TIME: 
            return original_fn(\*args, \*\*kwargs) 
        start_time = time.perf_counter() 
        result = original_fn(\*args, \*\*kwargs) 
        end_time = time.perf_counter() 
        time_list.append(end_time - start_time) 
        return result 

    setattr(obj, fn_name, wrapper) 

def calculate_transcription_time_and_accuracy(ov_model, test_samples): 
    encoder_infer_times = [] 
    decoder_with_past_infer_times = [] 
    whole_infer_times = [] 
    time_fn(ov_model, "generate", whole_infer_times) 
    time_fn(ov_model.encoder, "forward", encoder_infer_times) 
    time_fn(ov_model.decoder_with_past, "forward", decoder_with_past_infer_times) 

    ground_truths = [] 
    predictions = [] 
    for data_item in tqdm(test_samples, desc="Measuring performance and accuracy"): 
        input_features = extract_input_features(data_item) 

        with time_measurement(): 
            predicted_ids = ov_model.generate(input_features) 
        transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True) 

        ground_truths.append(data_item["text"]) 
        predictions.append(transcription[0]) 

    word_accuracy = (1 - wer(ground_truths, predictions, reference_transform=wer_standardize, hypothesis_transform=wer_standardize)) * 100 
    mean_whole_infer_time = sum(whole_infer_times) 
    mean_encoder_infer_time = sum(encoder_infer_times) 
    mean_decoder_with_time_infer_time = sum(decoder_with_past_infer_times) 
    return word_accuracy, (mean_whole_infer_time, mean_encoder_infer_time, mean_decoder_with_time_infer_time) 

test_dataset = load_dataset("openslr/librispeech_asr", "clean", split="validation", streaming=True, trust_remote_code=True) 
test_dataset = test_dataset.shuffle(seed=42).take(TEST_DATASET_SIZE) 
test_samples = [sample for sample in test_dataset] 

accuracy_original, times_original = calculate_transcription_time_and_accuracy(ov_model, test_samples) 
accuracy_quantized, times_quantized = calculate_transcription_time_and_accuracy(ov_quantized_model, test_samples) 
print(f"Encoder performance speedup: {times_original[1] / times_quantized[1]:.3f}") 
print(f"Decoder with past performance speedup: {times_original[2] / times_quantized[2]:.3f}") 
print(f"Whole pipeline performance speedup: {times_original[0] / times_quantized[0]:.3f}") 
print(f"Whisper transcription word accuracy. Original model: {accuracy_original:.2f}%. Quantized model:{accuracy_quantized:.2f}%.") 
print(f"Accuracy drop: {accuracy_original - accuracy_quantized:.2f}%.")