[ローカル環境] faster-whisperを利用してリアルタイム文字起こしに挑戦 #Python

はじめに

WhisperAPI を利用せずにローカル環境でリアルタイム文字起こしに挑戦してみました。
本家Whisperだと音声ファイル形式以外の入力がうまくいかなかったため、faster-whisper
を利用しました。

手探りで挑戦しましたので、何かご指摘がありましたらお教えいただければ幸いです。

効率的に文字起こしを行うための関連記事

2023 年 03 月 30 日
2023 年 04 月 05 日更新
2023 年 04 月 26 日更新
2023 年 04 月 28 日更新
2023 年 05 月 29 日更新
2023 年 06 月 07 日更新
2023 年 06 月 18 日更新
2023 年 06 月 26 日更新
2023 年 07 月 27 日更新
2023 年 09 月 07 日更新
2023 年 10 月 01 日更新
2023 年 11 月 26 日更新
2024 年 03 月 06 日更新

faster-whisperとは

faster-whisperは、トランスフォーマーモデルの高速推論エンジンであるCTranslate2を使用したOpenAIのWhisperモデルの再実装です。
この実装は、openai / whisperよりも最大4倍高速で、同じ精度で、より少ないメモリを使用します。CPUとGPUの両方で8ビット量子化することで、効率をさらに向上させることができます。

環境

Windows11
GeForce RTX 3060
HyperX QuadCast S(マイク)
Python 3.9.13
CUDA 11.7
cuDNN 8.5

また、Python で CUDA が利用できること。
※ここでは環境構築については触れません。

ライブラリ

PyAudio 0.2.13
webrtcvad 2.0.10
~~faster-whisper 0.3.0~~
~~faster-whisper 0.4.1 (2023/03/04)~~
~~faster-whisper 0.5.0 (2023/04/26)~~
~~faster-whisper 0.5.1 (2023/04/27)~~
~~faster-whisper 0.6.0 (2023/05/24)~~
~~faster-whisper 0.7.0 (2023/07/18)~~
~~faster-whisper 0.7.1 (2023/07/24)~~
~~faster-whisper 0.8.0 (2023/09/04)~~
~~faster-whisper 0.9.0 (2023/09/18)~~
~~faster-whisper 0.10.0 (2023/11/25)~~
~~faster-whisper 1.0.0 (2024/02/22)~~
faster-whisper 1.0.1 (2024/03/01 New)

VADが使えるようになったようですが、入力音声の区切りに使用するため利用できず。
0.6.0で空セグメントの発生が発生しなくなった模様。

0.7.0

Hugging Face Hubからユーザーモデルを指定できるようになりました。
単語レベルのタイムスタンプの改善
initial_promptにトークンIDが指定できるように
前回の文字起こしの情報を引き継げる？

0.7.1

不具合修正

0.8.0

オプションの追加
　恐らく繰り返し出力を抑制するもの
メモリリーク対策
返却値に型定義の追加

0.9.0

利用可能なモデルサイズ一覧の表示関数の追加
モデルが対応している言語一覧の表示
エラーメッセージの改善(task language)

0.10.0

large-v3モデルに対応
バージョン3.22.0のCTranslate2を要求
バージョン0.15のtokenizersを要求
その他：リポジトリがSystran社に移譲された模様

1.0.0

distil-whisperモデルのサポート
バージョン4.0のCTranslate2を要求
CUDA12をサポート
無音判定や認識の改善

1.0.1

バグ修正とパフォーマンス改善

本家Whisperで気になること

文字起こし対象をファイルで指定する必要がある。
パラメータとしては対応していそうだが、audio waveform で input うまく動作しない。

audio: Union[str, np.ndarray, torch.Tensor]
The path to the audio file to open, or the audio waveform

　毎回音声ファイルを生成して読み込むとオーバーヘッドが大きくリアルタイム文字起こしには不向き

30 秒単位での処理である

Internally, the transcribe() method reads the entire file and processes the audio with a sliding 30-second window, performing autoregressive sequence-to-sequence predictions on each window.

faster-whisperで気になること

~~- 実行時にモデルのダウンロードがあり、オフラインだと動作しない。~~
Whisper modelの初期化時にmodel_sizeではなく、ローカルに保存したmodelのパスを指定すれば動作しました。

処理概要

使用可能なオーディオデバイスを検出する：pyaudio
入力デバイスを選択する：pyaudio
音声入力開始(ループ)：pyaudio
無音状態検出するまで音声入力をバッファに格納：webrtcvad
音声情報を文字起こし:faster_whisper

実行結果

※効果音ラボ様の音声での実行結果となります。

実装

　pyaudioに指定するCHUNKは160単位で480まで
　他の設定の場合は、webrtcvadを利用する際にエラーとなった。

main.py
audio_transcriber.py
audio_utils.py
vad_utils.py
whisper_utils.py

main.py

from audio_transcriber import AudioTranscriber
from audio_utils import display_valid_input_devices, get_valid_input_devices

if __name__ == "__main__":
    transcriber = AudioTranscriber()

    valid_devices = get_valid_input_devices()
    print("使用可能なオーディオデバイス:")
    display_valid_input_devices(valid_devices)

    # 対象のDeviceIndexを入力
    selected_device_index = int(input("対象のDeviceIndexを入力してください: "))

    # 文字起こしを開始
    transcriber.start_transcription(selected_device_index)

audio_transcriber.py
2023 年 4 月 26 日修正(タイムスタンプ出力の削除、ノイズをクリアする)

audio_transcriber.py

import asyncio
import queue
from concurrent.futures import ThreadPoolExecutor

import numpy as np
import pyaudio

from audio_utils import create_audio_stream
from vad_utils import VadWrapper
from whisper_utils import WhisperModelWrapper


class AudioTranscriber:
    def __init__(self):
        self.model_wrapper = WhisperModelWrapper()
        self.vad_wrapper = VadWrapper()
        self.silent_chunks = 0
        self.speech_buffer = []
        self.audio_queue = queue.Queue()

    async def transcribe_audio(self):
        with ThreadPoolExecutor() as executor:
            while True:
                audio_data_np = await asyncio.get_event_loop().run_in_executor(
                    executor, self.audio_queue.get
                )
                segments = await asyncio.get_event_loop().run_in_executor(
                    executor, self.model_wrapper.transcribe, audio_data_np
                )

                for segment in segments:
                    print(segment.text)

    def process_audio(self, in_data, frame_count, time_info, status):
        is_speech = self.vad_wrapper.is_speech(in_data)

        if is_speech:
            self.silent_chunks = 0
            audio_data = np.frombuffer(in_data, dtype=np.int16)
            self.speech_buffer.append(audio_data)
        else:
            self.silent_chunks += 1

        if (
            not is_speech
            and self.silent_chunks > self.vad_wrapper.SILENT_CHUNKS_THRESHOLD
        ):
            if len(self.speech_buffer) > 20:
                audio_data_np = np.concatenate(self.speech_buffer)
                self.speech_buffer.clear()
                self.audio_queue.put(audio_data_np)
            else:
                # noise clear
                self.speech_buffer.clear()

        return (in_data, pyaudio.paContinue)

    def start_transcription(self, selected_device_index):
        stream = create_audio_stream(selected_device_index, self.process_audio)
        print("Listening...")
        asyncio.run(self.transcribe_audio())
        stream.start_stream()
        try:
            while True:
                key = input("Enterキーを押したら終了します\n")
                if not key:
                    break
        except KeyboardInterrupt:
            print("Interrupted.")
        finally:
            stream.stop_stream()
            stream.close()

audio_utils.py

import pyaudio


# 有効なインプットデバイスリストを取得する関数
def get_valid_input_devices():
    valid_devices = []
    audio = pyaudio.PyAudio()
    device_count = audio.get_device_count()
    default_host_api_info = audio.get_default_host_api_info()
    default_host_api_index = default_host_api_info["index"]

    for i in range(device_count):
        device_info = audio.get_device_info_by_index(i)
        if (
            device_info["maxInputChannels"] > 0
            and device_info["hostApi"] == default_host_api_index
        ):
            valid_devices.append(device_info)

    return valid_devices


# 有効なインプットデバイスリストを表示する関数
def display_valid_input_devices(valid_devices):
    for device_info in valid_devices:
        print(
            f"DeviceIndex: {device_info['index']}, DeviceName: {device_info['name']}, 入力チャンネル数: {device_info['maxInputChannels']}"
        )


# オーディオストリームを作成する関数
def create_audio_stream(selected_device_index, callback):
    RATE = 16000
    CHUNK = 480
    FORMAT = pyaudio.paInt16
    CHANNELS = 1

    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        input_device_index=selected_device_index,
        frames_per_buffer=CHUNK,
        stream_callback=callback,
    )

    return stream

vad_utils.py

import webrtcvad


class VadWrapper:
    def __init__(self):
        self.vad = webrtcvad.Vad(3)
        self.RATE = 16000
        self.SILENT_CHUNKS_THRESHOLD = 20

    def is_speech(self, in_data):
        return self.vad.is_speech(in_data, self.RATE)

whisper_utils.py
2023 年 4 月 26 日修正(beam_sizeの変更、変数の変更)

whisper_utils.py

from faster_whisper import WhisperModel


class WhisperModelWrapper:
    def __init__(self):
        self.model_size_or_path = "large-v2"
        self.model = WhisperModel(
            self.model_size_or_path, device="cuda", compute_type="float16"
        )

    def transcribe(self, audio):
        segments, _ = self.model.transcribe(
            audio=audio, beam_size=3, language="ja", without_timestamps=True 
        )
        return segments

サンプルコード

下記にサンプルコードを置いています。
https://github.com/reriiasu/speech-to-text

課題

~~話の速度が速いと文字起こしが遅れていく。~~
音声入力の精度が悪いと繰り返し探索が行われ間に合わなくなっていたと考えられる。
音声入力の精度向上や失敗と思われる場合の探索方法の変更が有効でした。
~~｢エンディング｣｢ご視聴ありがとうございました。｣等よくわからない誤変換が多発する。~~
ノイズと思われる入力をクリアするようにしたため、激減しました。

今後の予定

~~課題の解消~~
~~マイク入力ではなく、デスクトップ音声でも試してみる。~~
＝＞[Windows] 文字起こしにデスクトップ音を利用する試みにて対応

参考

本家Whisper
faster-whisper