Add openai-whisper-large-v3-turbo recipe

Author: Akupadhye

openai-whisper-large-v3-turbo/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 OpenAI
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

openai-whisper-large-v3-turbo/olive/README.md

@@ -0,0 +1,41 @@
+## Whisper-large-v3-turbo Optimization with ONNX Runtime QNN EP
+This folder outlines the process for optimizing the Whisper-large-v3-turbo model using ONNX Runtime with the QNN Execution Provider. It includes steps for exporting FP32 models, generating representative data for static quantization, creating QDQ models, model evaluation and performing audio transcription using the optimized models.
+
+### Prerequisites
+```bash
+python -m pip install -r requirements_qnn.txt
+```
+### Generate data for static quantization
+
+To get better results, we need to generate real data from original FP32 model instead of using random data for static quantization. Here we use 100 samples of librispeech dataset to generate the required real data which requires around 164 GB of disk space.
+
+First generate FP32 onnx models:
+
+1. Encoder FP32 model
+
+    `olive run --config whisper_large_v3_turbo_encoder_fp32.json`
+1. Decoder FP32 model
+
+    `olive run --config whisper_large_v3_turbo_decoder_fp32.json`
+
+
+Then download and generate data:
+
+1. `python download_librispeech_asr.py --save_dir .\data`
+
+2. `python .\demo.py --audio-path .\data\librispeech_asr_clean_test --encoder "models\whisper_encoder_fp32\model\model.onnx" --decoder "models\whisper_decoder_fp32\model.onnx" --model_id "openai/whisper-large-v3-turbo" --save_data .\data\quantization_data --num_data 100`
+
+### Generate QDQ models
+
+1. `olive run --config whisper_large_v3_turbo_encoder_qdq.json`
+2. `olive run --config whisper_large_v3_turbo_decoder_qdq.json`
+
+### Evaluation
+
+Evaluate model using the librispeech test-clean dataset
+
+`python .\evaluate_whisper.py --encoder "models\whisper_encoder_qdq\model.onnx" --decoder "models\whisper_decoder_qdq\model.onnx" --model_id "openai/whisper-large-v3-turbo" --execution_provider QNNExecutionProvider`
+
+### To transcribe a single sample:
+
+`python .\demo.py --audio-path .\data\librispeech_asr_clean_test\1320-122617-0000.npy --encoder "models\whisper_encoder_qdq\model.onnx" --decoder "models\whisper_decoder_qdq\model.onnx" --model_id "openai/whisper-large-v3-turbo" --execution_provider QNNExecutionProvider`

openai-whisper-large-v3-turbo/olive/__init__.py

@@ -0,0 +1,195 @@
+# ---------------------------------------------------------------------
+# Copyright (c) 2025 Qualcomm Technologies, Inc. and/or its subsidiaries.
+# SPDX-License-Identifier: BSD-3-Clause
+# ---------------------------------------------------------------------
+
+from __future__ import annotations
+
+import os
+
+import numpy as np
+import onnxruntime as ort
+import torch
+from qai_hub_models.models._shared.hf_whisper.app import HfWhisperApp, chunk_and_resample_audio
+from qai_hub_models.models._shared.hf_whisper.model import (
+    CHUNK_LENGTH,
+    SAMPLE_RATE,
+)
+from transformers import WhisperProcessor
+
+
+def infer_audio(app, model_id, audio_file, save_data):
+    audio_dict = np.load(audio_file, allow_pickle=True).item()
+
+    audio = audio_dict["audio"]["array"]
+    sample_rate = audio_dict["audio"]["sampling_rate"]
+    audio_name = os.path.splitext(os.path.basename(audio_file))[0] if save_data else None
+
+    processor = WhisperProcessor.from_pretrained(model_id)
+    reference = processor.tokenizer._normalize(audio_dict["text"])
+    print("Reference: ", reference)
+
+    # Perform transcription
+    transcription = app.transcribe(audio, sample_rate, audio_name, save_data)
+    print("done transcription")
+    prediction = processor.tokenizer._normalize(transcription)
+    print("Prediction:", prediction)
+
+
+class HfWhisperAppWithSave(HfWhisperApp):
+    def __init__(
+        self,
+        encoder,
+        decoder,
+        hf_model_id: str,
+        execution_provider: str = "CPUExecutionProvider",
+        provider_options: dict = None,
+        sample_rate: int = SAMPLE_RATE,
+        max_audio_seconds: int = CHUNK_LENGTH,
+    ):
+        super().__init__(None, None, hf_model_id, sample_rate, max_audio_seconds)
+        options = ort.SessionOptions()
+
+        self.encoder = ort.InferenceSession(
+            encoder, sess_options=options, providers=[execution_provider], provider_options=[provider_options]
+        )
+
+        self.decoder = ort.InferenceSession(
+            decoder, sess_options=options, providers=[execution_provider], provider_options=[provider_options]
+        )
+
+    def transcribe_tokens(self, audio, sample_rate, audio_name, save_data=False) -> list[int]:
+        out_chunked_tokens = []
+        for ind, x in enumerate(chunk_and_resample_audio(audio, sample_rate)):
+            out_chunked_tokens.append(self._transcribe_single_chunk(x, audio_name, ind, save_data))
+
+        out_tokens: list[int] = []
+        for chunk_tokens in out_chunked_tokens:
+            out_tokens.extend(chunk_tokens)
+        return out_tokens
+
+    def transcribe(self, audio, sample_rate, audio_name, save_data=False) -> str:
+        tokens = self.transcribe_tokens(audio, sample_rate, audio_name, save_data)
+        return self.tokenizer.decode(tokens, skip_special_tokens=True).strip()
+
+    def _transcribe_single_chunk(
+        self, audio: np.ndarray, audio_name=None, chunk_number=None, save_data=False
+    ) -> list[int]:
+        # feature
+        input_features = self.feature_extractor(audio, sampling_rate=self.sample_rate, return_tensors="np")[
+            "input_features"
+        ]
+
+        # encoder
+        output_names_encoder = [output.name for output in self.encoder.get_outputs()]
+        # kv_cache_cross = self.encoder(input_features)
+        input_features_feed = {"input_features": input_features}
+
+        if save_data:
+            input_features_save_path = os.path.join(save_data, audio_name, f"{chunk_number}_input_features.npy")
+            os.makedirs(os.path.dirname(input_features_save_path), exist_ok=True)
+            np.save(input_features_save_path, input_features_feed)
+
+        kv_cache_cross_numpy = self.encoder.run(output_names_encoder, input_features_feed)
+        kv_cache_cross = [torch.from_numpy(arr) for arr in kv_cache_cross_numpy]
+        if not isinstance(kv_cache_cross, tuple):
+            kv_cache_cross = (kv_cache_cross,)
+        if not isinstance(kv_cache_cross[0], (tuple, list)):
+            kv_cache_cross = (kv_cache_cross,)
+
+        sot = self.config.decoder_start_token_id
+        num_decoder_blocks = self.config.decoder_layers
+        attention_dim = self.config.d_model
+        num_decoder_heads = self.config.decoder_attention_heads
+        mask_neg = self.config.mask_neg
+        eot = self.config.eos_token_id
+
+        # decoder
+        output_ids = torch.tensor([[sot]])  # Start of transcript
+        output_logits = []
+        output_length = output_ids.shape[1]
+
+        position_ids = torch.tensor([0], dtype=torch.int32)
+        attention_mask = torch.full(
+            (1, 1, 1, self.mean_decode_len),
+            mask_neg,
+            dtype=torch.float32,
+        )
+
+        # init kv_cache_self
+        k_cache_self = torch.zeros(
+            (
+                num_decoder_heads,
+                1,
+                attention_dim // num_decoder_heads,
+                self.mean_decode_len - 1,
+            ),
+            dtype=torch.float32,
+        )
+        v_cache_self = torch.zeros(
+            (
+                num_decoder_heads,
+                1,
+                self.mean_decode_len - 1,
+                attention_dim // num_decoder_heads,
+            ),
+            dtype=torch.float32,
+        )
+        kv_cache_self = tuple((k_cache_self, v_cache_self) for _ in range(num_decoder_blocks))
+
+        for n in range(self.mean_decode_len - 1):
+            # get current token
+            input_ids = output_ids[:, n : n + 1].to(torch.int32)
+
+            # update attention_mask
+            attention_mask[:, :, :, self.mean_decode_len - n - 1] = 0.0
+
+            # flattened kv caches input
+            flattened_kv_cache_self = tuple(item for sublist in kv_cache_self for item in sublist)
+            flattened_kv_cache_cross = tuple(item for sublist in kv_cache_cross for item in sublist)
+
+            # decode and update kv_cache_self
+            decoder_input = (
+                (input_ids, attention_mask) + flattened_kv_cache_self + flattened_kv_cache_cross + (position_ids,)
+            )
+
+            # print("decoder_input: ", decoder_input)
+            input_names_decoder = [input.name for input in self.decoder.get_inputs()]
+            output_names_decoder = [output.name for output in self.decoder.get_outputs()]
+
+            # decoder_input_feed = dict(zip(input_names_decoder, decoder_input))
+            decoder_input_feed = {
+                name: tensor.numpy() if isinstance(tensor, torch.Tensor) else tensor
+                for name, tensor in zip(input_names_decoder, decoder_input)
+            }
+
+            if save_data:
+                decoder_input_save_path = os.path.join(save_data, audio_name, f"{chunk_number}_{n}_decoder_input.npy")
+                os.makedirs(os.path.dirname(decoder_input_save_path), exist_ok=True)
+                np.save(decoder_input_save_path, decoder_input_feed)
+
+            decoder_output_numpy = self.decoder.run(output_names_decoder, decoder_input_feed)
+            decoder_output = [torch.from_numpy(arr) for arr in decoder_output_numpy]
+            # decoder_output = self.decoder(*decoder_input)
+            if isinstance(decoder_output, tuple) and len(decoder_output) == 2:
+                logits, kv_cache_self = decoder_output
+            else:
+                logits = decoder_output[0]
+                kv_cache_self = tuple(decoder_output[i : i + 2] for i in range(1, len(decoder_output), 2))
+
+            # update output_logits
+            output_logits.append(logits.detach().clone())
+
+            # update output_ids
+            output_id = torch.argmax(logits, 1).squeeze(0)
+            # end of transcript
+            if len(output_logits) == (self.mean_decode_len - 1) or output_id == eot:
+                output_ids = torch.cat((output_ids, output_id), -1)
+                break
+            if n >= output_length - 1:
+                output_ids = torch.cat((output_ids, output_id), -1)
+
+            # update position_ids
+            position_ids += 1
+
+        return output_ids[0].tolist()

openai-whisper-large-v3-turbo/olive/app.py

@@ -0,0 +1,83 @@
+# ---------------------------------------------------------------------
+# Copyright (c) 2025 Qualcomm Technologies, Inc. and/or its subsidiaries.
+# SPDX-License-Identifier: BSD-3-Clause
+# ---------------------------------------------------------------------
+
+import argparse
+import os
+
+from app import HfWhisperAppWithSave, infer_audio
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Demo")
+    parser.add_argument(
+        "--audio-path",
+        type=str,
+        help="Path to folder containing audio files or a single audio file path",
+    )
+    parser.add_argument(
+        "--encoder",
+        type=str,
+        help="Path to encoder onnx file",
+    )
+    parser.add_argument(
+        "--decoder",
+        type=str,
+        help="Path to decoder onnx file",
+    )
+    parser.add_argument(
+        "--model_id",
+        type=str,
+        default="openai/whisper-large-v3-turbo",
+        help="HuggingFace Whisper model id",
+    )
+    parser.add_argument(
+        "--execution_provider",
+        type=str,
+        default="CPUExecutionProvider",
+        help="ORT Execution provider",
+    )
+    parser.add_argument(
+        "--save_data",
+        type=str,
+        default=None,
+        help="(Optional) Path to save quantization data",
+    )
+    parser.add_argument(
+        "--num_data",
+        type=int,
+        default=100,
+        help="Number of data samples to use for quantization. Only applicable if --save_data is enabled",
+    )
+
+    args = parser.parse_args()
+
+    encoder_path = args.encoder
+    decoder_path = args.decoder
+
+    provider_options = {}
+    if args.execution_provider == "QNNExectionProvider":
+        provider_options = {
+            "backend_path": "QnnHtp.dll",
+            "htp_performance_mode": "sustained_high_performance",
+            "htp_graph_finalization_optimization_mode": "3",
+            "offload_graph_io_quantization": "0",
+        }
+
+    app = HfWhisperAppWithSave(encoder_path, decoder_path, args.model_id, args.execution_provider, provider_options)
+
+    if os.path.isdir(args.audio_path):
+        for i, item in enumerate(os.listdir(args.audio_path)):
+            if args.save_data and i == args.num_data:
+                break
+
+            full_path = os.path.join(args.audio_path, item)
+            infer_audio(app, args.model_id, full_path, args.save_data)
+
+    else:
+        infer_audio(app, args.model_id, args.audio_path, args.save_data)
+
+
+if __name__ == "__main__":
+    main()

openai-whisper-large-v3-turbo/olive/demo.py

@@ -0,0 +1,28 @@
+import argparse
+import os
+
+import numpy as np
+from datasets import load_dataset
+
+
+def download_librispeech_asr(save_dir):
+    # Create save_dir if it doesn't exist
+    save_dir = os.path.join(save_dir, "librispeech_asr_clean_test")
+    os.makedirs(save_dir, exist_ok=True)
+
+    # Load streaming dataset
+    streamed_dataset = load_dataset("librispeech_asr", "clean", split="test", streaming=True)
+
+    for batch in streamed_dataset:
+        file_path = os.path.join(save_dir, f"{batch['id']}.npy")
+        np.save(file_path, batch)
+
+    print("Download complete!!")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Download and save samples from librispeech_asr dataset.")
+    parser.add_argument("--save_dir", type=str, required=True, help="Directory to save the dataset samples.")
+    args = parser.parse_args()
+
+    download_librispeech_asr(args.save_dir)