リアルタイム音声認識をwhisperのturboモデル+faster_whisperで動かす
whisper turboモデル
WhisperはOpenAI製の音声認識モデルです。large-v3が最新の高性能モデルですが、2024/10にモデルの小型化・大幅な高速化を実現したturboモデルがリリースされています。large-v3のデコーダ層数を32から4に減らすことで高速化を実現しているとのことです。
Size | Parameters | English-only model | Multilingual model | Required VRAM | Relative speed |
---|---|---|---|---|---|
tiny | 39 M | tiny.en |
tiny |
~1 GB | ~10x |
base | 74 M | base.en |
base |
~1 GB | ~7x |
small | 244 M | small.en |
small |
~2 GB | ~4x |
medium | 769 M | medium.en |
medium |
~5 GB | ~2x |
large | 1550 M | N/A | large |
~10 GB | 1x |
turbo | 809 M | N/A | turbo |
~6 GB | ~8x |
こちらが 公式リポジトリ に掲載されている比較表です。モデルサイズがlargeの半分程度に抑えられ、速度に至ってはlargeの最大8倍と大幅に改善されています。精度についてもlarge-v3には多少劣るものの、日本語ではlarge-v2と同レベルの性能が出ています。
faster_whisper
faster_whisperもwhisperの高速化実装です。Transformerモデルの高速化に特化したエンジンであるCTranslate2を使って推論速度を大幅に(4倍程度)向上させています。
turboがモデル構造の変更・軽量化による高速化なのに対して、こちらは推論の計算処理をモデルレイヤーの融合、ビット量子化、バッチの入れ替えなどによって最適化しています。
whisper_mic
whisper_mic はwhisperをマイクに繋いで簡単に動かせるようにした薄いライブラリです。WhisperMicクラスで抽象化されており、modelの指定やfaster_whisperのimplementationを利用できるなど、シュッと動かすのにとても便利です。
セットアップ
以下の環境で検証しました。
- OS: Windows 11
- CPU: 11th Gen Intel(R) Core(TM) i5-11400F @ 2.60GHz (55.14 GFLOPS)
- メインメモリ: 64GB
- GPU: ZOTAC GAMING GeForce RTX 3070 Twin Edge (VRAM 8GB)
- マイク: ミヨシ(Miyoshi) MCO USBピンマイク
- CUDA 12.1
- Python 3.12.6
- faster_whisper: 1.1.1
- pytorch: 2.5.1+cu121
元々マシンにはCUDA 11を入れていたのですが、faster_whisper 1.0以降を使う場合はctranslate2という依存関係の都合でCUDA 12が必要なようです。
グローバルなpipで入れる場合はctranslate2をダウングレードすることもできますが、ryeで依存関係を管理したかったのでおとなしくCUDAを12にアップグレードしました。
基本はWhisperMicを使うだけですが、日本語を指定して書き起こしするよう一部修正を行いたかったのでコピペして動かしています。
コード全文
import torch
import queue
import speech_recognition as sr
import threading
import numpy as np
import os
import time
import tempfile
import platform
import pynput.keyboard
from typing import Optional
# from ctypes import *
import logging
from typing_extensions import Literal
from rich.logging import RichHandler
# from whisper_utils import get_logger
def get_logger(name: str, level: Literal["info", "warning", "debug"]) -> logging.Logger:
rich_handler = RichHandler(level=logging.INFO, rich_tracebacks=True, markup=True)
logger = logging.getLogger(name)
logger.setLevel(logging._nameToLevel[level.upper()])
if not logger.handlers:
logger.addHandler(rich_handler)
logger.propagate = False
return logger
#TODO: This is a linux only fix and needs to be testd. Have one for mac and windows too.
# Define a null error handler for libasound to silence the error message spam
# def py_error_handler(filename, line, function, err, fmt):
# None
# ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
# c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
# asound = cdll.LoadLibrary('libasound.so')
# asound.snd_lib_error_set_handler(c_error_handler)
class WhisperMic:
def __init__(self,model="base",device=("cuda" if torch.cuda.is_available() else "cpu"),english=False,verbose=False,energy=300,pause=2,dynamic_energy=False,save_file=False, model_root="~/.cache/whisper",mic_index=None,implementation="whisper",hallucinate_threshold=300):
self.logger = get_logger("whisper_mic", "info")
self.energy = energy
self.hallucinate_threshold = hallucinate_threshold
self.pause = pause
self.dynamic_energy = dynamic_energy
self.save_file = save_file
self.verbose = verbose
self.english = english
self.keyboard = pynput.keyboard.Controller()
self.platform = platform.system().lower()
if self.platform == "darwin":
if device == "cuda" or device == "mps":
self.logger.warning("CUDA is not supported on MacOS and mps does not work. Using CPU instead.")
device = "cpu"
else:
device = "cuda" if torch.cuda.is_available() else "cpu"
if (model != "large" and model != "large-v2" and model!= "large-v3") and self.english:
model = model + ".en"
model_root = os.path.expanduser(model_root)
self.faster = False
if (implementation == "faster_whisper"):
try:
from faster_whisper import WhisperModel
self.audio_model = WhisperModel(model, download_root=model_root, device="auto", compute_type="int8")
self.faster = True # Only set the flag if we succesfully imported the library and opened the model.
except ImportError:
self.logger.error("faster_whisper not installed, falling back to whisper")
self.logger.info("To install faster_whisper, run 'pip install faster_whisper'")
import whisper
self.audio_model = whisper.load_model(model, download_root=model_root).to(device)
else:
import whisper
self.audio_model = whisper.load_model(model, download_root=model_root).to(device)
self.temp_dir = tempfile.mkdtemp() if save_file else None
self.audio_queue = queue.Queue()
self.result_queue: "queue.Queue[str]" = queue.Queue()
self.break_threads = False
self.mic_active = False
self.banned_results = [""," ","\n",None]
if save_file:
self.file = open("transcribed_text.txt", "w+", encoding="utf-8")
self.__setup_mic(mic_index)
def __setup_mic(self, mic_index):
if mic_index is None:
self.logger.info("No mic index provided, using default")
self.source = sr.Microphone(sample_rate=16000, device_index=mic_index)
self.recorder = sr.Recognizer()
self.recorder.energy_threshold = self.energy
self.recorder.pause_threshold = self.pause
self.recorder.dynamic_energy_threshold = self.dynamic_energy
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
self.logger.info("Mic setup complete")
# Whisper takes a Tensor while faster_whisper only wants an NDArray
def __preprocess(self, data):
is_audio_loud_enough = self.is_audio_loud_enough(data)
if self.faster:
return np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0,is_audio_loud_enough
else:
return torch.from_numpy(np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0),is_audio_loud_enough
def is_audio_loud_enough(self, frame) -> bool:
audio_frame = np.frombuffer(frame, dtype=np.int16)
amplitude = np.mean(np.abs(audio_frame))
return amplitude > self.hallucinate_threshold
def __get_all_audio(self, min_time: float = -1.):
audio = bytes()
got_audio = False
time_start = time.time()
while not got_audio or time.time() - time_start < min_time:
while not self.audio_queue.empty():
audio += self.audio_queue.get()
got_audio = True
data = sr.AudioData(audio,16000,2)
data = data.get_raw_data()
return data
# Handles the task of getting the audio input via microphone. This method has been used for listen() method
def __listen_handler(self, timeout, phrase_time_limit):
try:
with self.source as microphone:
audio = self.recorder.listen(source=microphone, timeout=timeout, phrase_time_limit=phrase_time_limit)
self.__record_load(0, audio)
audio_data = self.__get_all_audio()
self.__transcribe(data=audio_data)
except sr.WaitTimeoutError:
self.result_queue.put_nowait("Timeout: No speech detected within the specified time.")
except sr.UnknownValueError:
self.result_queue.put_nowait("Speech recognition could not understand audio.")
# This method is similar to the __listen_handler() method but it has the added ability for recording the audio for a specified duration of time
def __record_handler(self, duration=2, offset=None):
with self.source as microphone:
audio = self.recorder.record(source=microphone, duration=duration, offset=offset)
self.__record_load(0, audio)
audio_data = self.__get_all_audio()
self.__transcribe(data=audio_data)
# This method takes the recorded audio data, converts it into raw format and stores it in a queue.
def __record_load(self,_, audio: sr.AudioData) -> None:
data = audio.get_raw_data()
self.audio_queue.put_nowait(data)
def __transcribe_forever(self) -> None:
while True:
if self.break_threads:
break
self.__transcribe()
def __transcribe(self,data=None, realtime: bool = False) -> None:
if data is None:
audio_data = self.__get_all_audio()
else:
audio_data = data
audio_data,is_audio_loud_enough = self.__preprocess(audio_data)
if is_audio_loud_enough:
predicted_text = ''
# faster_whisper returns an iterable object rather than a string
if self.faster:
segments, info = self.audio_model.transcribe(audio_data)
for segment in segments:
predicted_text += segment.text
else:
if self.english:
result = self.audio_model.transcribe(audio_data,language='english',suppress_tokens="")
else:
result = self.audio_model.transcribe(audio_data,language='japanese',suppress_tokens="")
predicted_text = result["text"]
if not self.verbose:
if predicted_text not in self.banned_results:
self.result_queue.put_nowait(predicted_text)
else:
if predicted_text not in self.banned_results:
self.result_queue.put_nowait(result)
if self.save_file:
# os.remove(audio_data)
self.file.write(predicted_text)
else:
# If the audio is not loud enough, we put None in the queue to indicate that we need to listen again or return None
self.result_queue.put_nowait(None)
async def listen_loop_async(self, dictate: bool = False, phrase_time_limit=None) -> Optional[str]:
for result in self.listen_continuously(phrase_time_limit=phrase_time_limit):
if dictate:
self.keyboard.type(result)
else:
yield result
def listen_loop(self, dictate: bool = False, phrase_time_limit=None) -> None:
for result in self.listen_continuously(phrase_time_limit=phrase_time_limit):
if result is not None:
if dictate:
self.keyboard.type(result)
else:
print(result)
def listen_continuously(self, phrase_time_limit=None):
self.recorder.listen_in_background(self.source, self.__record_load, phrase_time_limit=phrase_time_limit)
self.logger.info("Listening...")
threading.Thread(target=self.__transcribe_forever, daemon=True).start()
while True:
yield self.result_queue.get()
def listen(self, timeout = None, phrase_time_limit=None,try_again=True):
self.logger.info("Listening...")
self.__listen_handler(timeout, phrase_time_limit)
while True:
if not self.result_queue.empty():
result = self.result_queue.get()
if result is None and try_again:
self.logger.info("Too quiet, listening again...")
result = self.listen(timeout=timeout, phrase_time_limit=phrase_time_limit,try_again=True)
return result
else:
return result
# This method is similar to the listen() method, but it has the ability to listen for a specified duration, mentioned in the "duration" parameter.
def record(self, duration=2, offset=None,try_again=True):
self.logger.info("Listening...")
if duration is None:
self.logger.warning("Duration not provided, may hang indefinitely.")
self.__record_handler(duration, offset)
while True:
if not self.result_queue.empty():
result = self.result_queue.get()
if result is None and try_again:
self.logger.info("Too quiet, listening again...")
result = self.record(duration=duration, offset=offset,try_again=True)
return result
else:
return result
def toggle_microphone(self) -> None:
#TO DO: make this work
self.mic_active = not self.mic_active
if self.mic_active:
print("Mic on")
else:
print("turning off mic")
self.mic_thread.join()
print("Mic off")
if __name__ == '__main__':
mic = WhisperMic(model="turbo", implementation="faster_whisper", device="cpu")
mic.listen_loop()
whisper_utilsのインポートを直書きする形で書き直しています。また、__transcribe
メソッドの以下の部分を次のように修正して日本語を指定しています。
class WhisperMic:
...
def __transcribe(self,data=None, realtime: bool = False) -> None:
...
if self.english:
result = self.audio_model.transcribe(audio_data,language='english',suppress_tokens="")
else:
# language='japanese' を追加
result = self.audio_model.transcribe(audio_data,language='japanese', suppress_tokens="")
実験結果
以下の条件で比較して所要時間と精度の検証を行いました。
- device: CPU / GPU
- model: medium / large-v3 / turbo
- implementation: whisper / faster_whisper
Claudeで適当に作成してもらった以下の5つの文章を読み上げ、その書き起こしがコンソールに出力されるまでの時間を計測し、平均をとって応答時間としています。なお、文章の区切りを認識するためのポーズ閾値2秒を差し引いた結果を記載しています。
1. 今朝は7時に起きて、朝食にトーストと目玉焼きを食べました。
2. 新幹線は東京から大阪まで約2時間30分かかります。
3. 桜の花が満開で、公園がピンク色に染まっています。
4. スマートフォンの画面が突然真っ暗になってしまいました。
5. エアコンの設定温度を26度にして、省エネを心がけています。
まず、modelごとの書き起こし結果がこちらです。
model | transcript |
---|---|
medium | 朝食にドーストと目玉焼きを食べました。 新幹線は東京から大阪まで約2時間30分かかります。 桜の花が満開で、公園がピンク色に染まっています。 スマートフォンの画面が突然真っ暗になってしまいました え?入れる?この設定温度を26度にして小エネをこぼれかけています |
large-v3 | 今朝は7時に起きて、昼食にトーストと目玉焼きを食べました。 新幹線は東京から大阪まで約2時間30分かかります。 桜の花が満開で公園がピンクに染まっています スマートフォンの画面が突然真っ暗になってしまいました。 この設定温度を26度にして省エネを心がけています。 |
turbo | 今朝は7時に起きて朝食にトーストと目玉焼きを食べました 新幹線は東京から大阪まで約2時間30分かかります。 桜の花が満開で公園がピンク色に染まっています スマートフォンの画面が突然真っ暗になってしまいました。 エアコンの設定温度を26度にして、省エネを心がけています。 |
mediumでも十分に書き起こしできていますが、一部で誤植が見られます。large-v3, turboについてはほぼ完璧に書き起こしできていますね。
次に、速度の比較結果です。
device | model | implementation | time (s) |
---|---|---|---|
CPU | medium | whisper | 1.7 |
CPU | medium | faster_whisper | 1.4 |
CPU | large-v3 | whisper | 24.6 |
CPU | large-v3 | faster_whisper | 2.8 |
CPU | turbo | whisper | 1.3 |
CPU | turbo | faster_whisper | 1.4 |
GPU | medium | whisper | 4.7 |
GPU | medium | faster_whisper | 1.3 |
GPU | large-v3 | whisper | 28.1 |
GPU | large-v3 | faster_whisper | 1.4 |
GPU | turbo | whisper | 2.5 |
GPU | turbo | faster_whisper | 1.7 |
CPU環境ではmedium / turboが同程度に早く、large-v3が突出して遅いという結果になりました。ただし、faster_whisperを使った場合は大幅に改善されています。
GPU環境でも目立った改善は得られず、むしろ多くのケースでCPU環境よりも遅くなっています。これは今回のごく短い文章での処理ではGPUのVRAMとの間のデータの転送などにかかるオーバーヘッドが嵩んでいる可能性が高く、より長い文章では明確な差が出ると思われます。特にlarge-v3については今回の環境ではVRAMが不足しており本来の性能を出せていない可能性が高いと考えられます。ただしfaster_whisperを使用した場合はどのモデルでもCPU環境と同様の性能が出ています。
どのモデルでも設定次第で2秒以下の応答速度を実現できており、リアルタイムな音声入力として十分実用可能であると言えそうです。
まとめ
リアルタイムな応答性が求められる音声認識タスクにおいては、次のようなことが言えそうです。
- 設定次第で平均2秒以下の応答速度を実現でき、リアルタイムな音声入力として十分実用可能
- 一定のパワーがあるCPU環境ではCPU計算でも十分な応答速度が出せる
- モデルについてはmediumとturboでメモリ使用量・速度ともに同程度のため十分なマシンスペックがある場合はより性能の良いturboを安心して使える。それ以外の場合にはsmallモデル以下を検討すれば良い
- faster_whisperについては、速度面で大きなインパクトはないが、メモリ使用量等も改善されているはずなので使って損はなさそう
Discussion