その他

ローカルLLMで音声文字起こしアプリを作る【Python編②】リアルタイム変換への挑戦と音量メーター付きアプリ化

その他
この記事は約26分で読めます。

はじめに

みなさん、こんにちは。株式会社インプルのyotaです。

前回の記事では、faster-whisperとOllamaを組み合わせた音声文字起こしアプリを作りました。

まだご覧になっていない方はぜひご一読ください

このアプリですが、使ってみると1つ不満が出てきました。それは…

「録音している時、ちゃんと声が拾えているのかわからない」
「録音を止めるまで結果が見えないのが不安

そこで今回は「録音しながらリアルタイムでテキストが流れる」動作の実現を目指します。

結論から言うと、残念ながらfaster-whisperでのリアルタイム実現は難しく、最終的には別のアプローチを選ぶことになりました。その過程ごと記録しておきます。

リアルタイム文字起こしとは何か

まず「リアルタイム文字起こし」を整理します。

iPhoneの音声入力を使ったことがある方はご存知だと思いますが、話しながら画面にテキストが流れていくあの動作です。技術的には話している最中に部分的な認識結果を逐次返す仕組みが必要です。

試み:faster-whisper でストリーミング処理

faster-whisper でもチャンク分割すればリアルタイムに近い動作が実現できるのでは、と考えて以下を実装しました。

音声を 0.5 秒ごとのチャンクに分割して、2秒分(4チャンク)溜まるたびに Whisper に投げて仮テキストを表示。最終的には全音声を再度 Whisper で処理して確定テキストを出す、という方式です。

# stt.py に追加
def transcribe_stream(
    model: WhisperModel,
    on_preview,    # Callable[[str], None]
    on_final,      # Callable[[str], None]
    stop_event: threading.Event,
) -> None:
    audio_queue   = queue.Queue()
    all_chunks    = []
    preview_buf   = []
    silence_count = 0

    PREVIEW_CHUNKS    = 4     # 仮認識に使うチャンク数(0.5秒×4=2秒分)
    SILENCE_COUNT_MAX = 6     # 無音チャンクがこの数続いたら自動停止(3秒)
    CHUNK_FRAMES      = int(SAMPLE_RATE * 0.5)

    def _callback(indata, frames, time_info, status):
        audio_queue.put(indata.copy())

    with sd.InputStream(
        samplerate=SAMPLE_RATE, channels=CHANNELS,
        dtype="float32", blocksize=CHUNK_FRAMES,
        callback=_callback, device=0,
    ):
        while not stop_event.is_set():
            try:
                chunk = audio_queue.get(timeout=1.0)
            except queue.Empty:
                continue

            flat = chunk.flatten()
            all_chunks.append(flat)
            preview_buf.append(flat)

            rms = np.sqrt(np.mean(flat ** 2))
            if rms < SILENCE_THRESHOLD:
                silence_count += 1
                if silence_count >= SILENCE_COUNT_MAX:
                    stop_event.set()
                    break
            else:
                silence_count = 0

            if len(preview_buf) >= PREVIEW_CHUNKS:
                preview_audio = np.concatenate(preview_buf)
                preview_buf.clear()
                threading.Thread(
                    target=lambda a=preview_audio: on_preview(
                        "".join(s.text for s in model.transcribe(
                            a, language="ja", beam_size=1, vad_filter=True,
                        )[0]).strip()
                    ),
                    daemon=True,
                ).start()

    if not all_chunks:
        on_final("")
        return

    full_audio = np.concatenate(all_chunks)
    if np.sqrt(np.mean(full_audio ** 2)) < SILENCE_THRESHOLD:
        on_final("")
        return

    on_final("".join(s.text for s in model.transcribe(
        full_audio, language="ja", beam_size=5, vad_filter=True,
    )[0]).strip())

app.pyもストリーミング対応版に書き換えて、仮テキストをグレー・確定テキストを白で表示するUIにしました。録音開始からすぐには文字が表示されませんが、お使いのデバイスは正常です

結果:リアルタイムとは言えなかった

実際に動かしてみると、仮テキストの表示まで数秒かかり、とてもリアルタイムとは言えない動作でした。

これは Whisper がそもそもリアルタイム向けに設計されていないことに起因します。小さなチャンクを処理すると文脈が失われて精度が落ち、精度を保とうとすると処理時間がかかる、というジレンマがあります。

他のサービスの実装を調査

iPhoneの音声入力はどうやっているのか」を調べてみました。

  1. iPhoneの音声入力(Apple SFSpeechRecognizer)
    iOSのSFSpeechRecognizerはマイクからのライブ音声バッファを受け取りストリームとして認識器に流し、話しながら部分的な結果をリアルタイムで返す専用APIを持ってます。これはWhisperとは全く異なるストリーミング専用モデル(RNN-T)で動いています。
  2. Whisper-Streaming(OSS)
    LocalAgreement-2 ポリシーを使い、連続する2チャンクで Whisper 出力が一致した最長共通プレフィックスを「確定テキスト」として返す手法です。約3.3秒のレイテンシを実現しています。
  3. WhisperKit(Apple Silicon 特化)
    仮テキストと確定テキストの2ストリームを持ち、Apple Silicon 上でサブ秒レイテンシを実現しています。

調査のまとめ:Python + faster-whisper の限界

アプローチレイテンシ実装難易度M2 Mac
本記事の実装(チャンク処理)6〜8秒✅簡単
Whisper-Streaming(LocalAgreement)約3〜4秒中程度
mlx-whisper(Apple Silicon最適化)約1〜2秒中程度✅Metal利用
WhisperKit(Swift/CoreML)0.5秒以下❌Swift必須✅最速

「iPhoneと同等のリアルタイム性」を Python + faster-whisper で実現することは構造的に困難です。

WhisperKit を使えばサブ秒レイテンシが期待できますが、Swift での実装が必要になります。これは次回以降の Swift 編で挑戦します。

方針転換:音量メーター付きの使いやすいアプリに仕上げる

リアルタイム変換は一旦諦め、「録音中に音が拾えているか視覚的に確認できる ことを優先した実用的なアプリに仕上げることにしました。

前回のスライダー付き固定秒数方式から以下の仕様に変更します。

  • 録音開始・停止を手動でトグルボタンで制御(録音時間自由)
  • 録音中は音量メーターをリアルタイム表示(色付き)
  • 停止後に文字起こし → LLM 補正を実行
# src/app.py
import threading
import numpy as np
import sounddevice as sd
import tkinter as tk
import customtkinter as ctk

from stt import load_model, transcribe_microphone, SAMPLE_RATE, SILENCE_THRESHOLD
from llm import load_client, correct_text

ctk.set_appearance_mode("system")
ctk.set_default_color_theme("blue")

WINDOW_W     = 720
WINDOW_H     = 580
METER_UPDATE = 50    # 音量メーター更新間隔(ms)

class VoiceTranscriberApp(ctk.CTk):
    def __init__(self):
        super().__init__()
        self.title("Voice Transcriber")
        self.geometry(f"{WINDOW_W}x{WINDOW_H}")
        self.resizable(True, True)

        self.model        = None
        self.client       = None
        self.is_recording = False
        self._audio_buf   = []          # 録音チャンクを蓄積
        self._stream      = None        # sounddevice InputStream
        self._current_rms = 0.0         # 音量メーター用

        self._build_ui()
        self._start_model_loading()

    # ---------------------------------------------------------------
    # UI構築
    # ---------------------------------------------------------------
    def _build_ui(self):
        self.grid_columnconfigure(0, weight=1)
        self.grid_rowconfigure(1, weight=1)
        self.grid_rowconfigure(2, weight=1)

        # --- ヘッダー ---
        header = ctk.CTkFrame(self, height=50, corner_radius=0)
        header.grid(row=0, column=0, sticky="ew")
        header.grid_columnconfigure(0, weight=1)

        ctk.CTkLabel(
            header, text="🎤 Voice Transcriber",
            font=ctk.CTkFont(size=18, weight="bold")
        ).grid(row=0, column=0, padx=20, pady=12, sticky="w")

        self.status_var = tk.StringVar(value="モデルを読み込み中...")
        ctk.CTkLabel(
            header, textvariable=self.status_var,
            font=ctk.CTkFont(size=12), text_color="gray"
        ).grid(row=0, column=1, padx=20, pady=12, sticky="e")

        # --- STT出力エリア ---
        frame_stt = ctk.CTkFrame(self)
        frame_stt.grid(row=1, column=0, sticky="nsew", padx=16, pady=(16, 8))
        frame_stt.grid_columnconfigure(0, weight=1)
        frame_stt.grid_rowconfigure(1, weight=1)

        ctk.CTkLabel(
            frame_stt, text="音声認識(生テキスト)",
            font=ctk.CTkFont(size=13, weight="bold")
        ).grid(row=0, column=0, padx=12, pady=(10, 4), sticky="w")

        self.stt_box = ctk.CTkTextbox(frame_stt, font=ctk.CTkFont(size=13), wrap="word")
        self.stt_box.grid(row=1, column=0, sticky="nsew", padx=12, pady=(0, 12))

        # --- LLM補正出力エリア ---
        frame_llm = ctk.CTkFrame(self)
        frame_llm.grid(row=2, column=0, sticky="nsew", padx=16, pady=(8, 8))
        frame_llm.grid_columnconfigure(0, weight=1)
        frame_llm.grid_rowconfigure(1, weight=1)

        ctk.CTkLabel(
            frame_llm, text="LLM補正後テキスト",
            font=ctk.CTkFont(size=13, weight="bold")
        ).grid(row=0, column=0, padx=12, pady=(10, 4), sticky="w")

        self.llm_box = ctk.CTkTextbox(frame_llm, font=ctk.CTkFont(size=13), wrap="word")
        self.llm_box.grid(row=1, column=0, sticky="nsew", padx=12, pady=(0, 12))

        # --- フッター ---
        footer = ctk.CTkFrame(self, height=72, corner_radius=0)
        footer.grid(row=3, column=0, sticky="ew")
        footer.grid_columnconfigure(1, weight=1)

        # 音量メーターラベル
        ctk.CTkLabel(
            footer, text="入力音量:",
            font=ctk.CTkFont(size=12)
        ).grid(row=0, column=0, padx=(16, 4), pady=(8, 2), sticky="w")

        # 音量メーター(Progressbar風にCTkProgressBarを使用)
        self.meter = ctk.CTkProgressBar(footer, width=200, height=14)
        self.meter.set(0)
        self.meter.grid(row=0, column=1, padx=(0, 8), pady=(8, 2), sticky="ew")

        self.meter_label = ctk.CTkLabel(
            footer, text="--", width=40,
            font=ctk.CTkFont(size=11), text_color="gray"
        )
        self.meter_label.grid(row=0, column=2, padx=(0, 8), pady=(8, 2))

        # クリアボタン
        self.clear_btn = ctk.CTkButton(
            footer, text="クリア", width=80,
            fg_color="gray40", hover_color="gray30",
            command=self._clear_all
        )
        self.clear_btn.grid(row=0, column=3, padx=8, pady=(8, 2))

        # 録音トグルボタン
        self.record_btn = ctk.CTkButton(
            footer, text="⏺  録音開始", width=140,
            font=ctk.CTkFont(size=14, weight="bold"),
            state="disabled", command=self._on_record_toggle
        )
        self.record_btn.grid(row=0, column=4, padx=(0, 16), pady=(8, 2))

        # 録音時間表示
        self.elapsed_var = tk.StringVar(value="")
        ctk.CTkLabel(
            footer, textvariable=self.elapsed_var,
            font=ctk.CTkFont(size=11), text_color="gray"
        ).grid(row=1, column=4, padx=(0, 16), pady=(0, 6), sticky="e")

    # ---------------------------------------------------------------
    # モデルのバックグラウンドロード
    # ---------------------------------------------------------------
    def _start_model_loading(self):
        def _load():
            try:
                self._set_status("STTモデルをロード中...")
                self.model = load_model()
                self._set_status("LLMクライアントを初期化中...")
                self.client = load_client()
                self._set_status("準備完了 ✅")
                self.after(0, lambda: self.record_btn.configure(state="normal"))
            except Exception as e:
                self._set_status(f"初期化エラー: {e}")
        threading.Thread(target=_load, daemon=True).start()

    # ---------------------------------------------------------------
    # 録音トグル
    # ---------------------------------------------------------------
    def _on_record_toggle(self):
        if not self.is_recording:
            self._start_recording()
        else:
            self._stop_recording()

    def _start_recording(self):
        self.is_recording = True
        self._audio_buf.clear()
        self._elapsed_sec = 0
        self.record_btn.configure(
            text="⏹  録音停止",
            fg_color="#c0392b", hover_color="#a93226"
        )
        self._set_status("録音中... 話してください")
        self.elapsed_var.set("0.0 秒")

        # マイク入力ストリーム開始
        self._stream = sd.InputStream(
            samplerate=SAMPLE_RATE, channels=1,
            dtype="float32", blocksize=int(SAMPLE_RATE * 0.1),
            callback=self._audio_callback, device=0,
        )
        self._stream.start()

        # 音量メーター・経過時間の定期更新開始
        self._update_meter()

    def _stop_recording(self):
        if self._stream:
            self._stream.stop()
            self._stream.close()
            self._stream = None

        self.is_recording = False
        self.record_btn.configure(state="disabled", text="⏺  処理中...")
        self.meter.set(0)
        self.meter_label.configure(text="--")
        self.elapsed_var.set("")

        # 録音データをSTT→LLMへ
        threading.Thread(target=self._run_pipeline, daemon=True).start()

    # ---------------------------------------------------------------
    # マイク入力コールバック(別スレッドから呼ばれる)
    # ---------------------------------------------------------------
    def _audio_callback(self, indata, frames, time_info, status):
        chunk = indata.copy().flatten()
        self._audio_buf.append(chunk)
        self._current_rms = float(np.sqrt(np.mean(chunk ** 2)))

    # ---------------------------------------------------------------
    # 音量メーター・経過時間の更新(UIスレッド)
    # ---------------------------------------------------------------
    def _update_meter(self):
        if not self.is_recording:
            return

        # メーター値:RMSを0〜1にスケーリング(上限0.1を1.0とみなす)
        level = min(self._current_rms / 0.1, 1.0)
        self.meter.set(level)

        # 色変化:無音=グレー、小=青、大=緑
        if self._current_rms < SILENCE_THRESHOLD:
            self.meter.configure(progress_color="gray50")
        elif level < 0.4:
            self.meter.configure(progress_color="#3B8ED0")
        else:
            self.meter.configure(progress_color="#2ecc71")

        # 数値表示
        self.meter_label.configure(text=f"{self._current_rms:.4f}")

        # 経過時間更新
        self._elapsed_sec += METER_UPDATE / 1000
        self.elapsed_var.set(f"{self._elapsed_sec:.1f} 秒")

        self.after(METER_UPDATE, self._update_meter)

    # ---------------------------------------------------------------
    # STT → LLM パイプライン
    # ---------------------------------------------------------------
    def _run_pipeline(self):
        try:
            if not self._audio_buf:
                self._set_status("音声が検出されませんでした")
                return

            # 蓄積した音声チャンクを結合
            full_audio = np.concatenate(self._audio_buf)
            rms = np.sqrt(np.mean(full_audio ** 2))

            if rms < SILENCE_THRESHOLD:
                self._set_status("音声が検出されませんでした")
                return

            # STT
            self._set_status("文字起こし中...")
            segments, _ = self.model.transcribe(
                full_audio, language="ja", beam_size=5,
                vad_filter=True,
                vad_parameters={"min_silence_duration_ms": 500},
            )
            raw = "".join(seg.text for seg in segments).strip()

            if not raw:
                self._set_status("認識結果が得られませんでした")
                return

            self._append_text(self.stt_box, raw)
            self._set_status("LLMで補正中...")

            # LLM
            corrected = correct_text(self.client, raw)
            self._append_text(self.llm_box, corrected)
            self._set_status("準備完了 ✅")

        except Exception as e:
            self._set_status(f"エラー: {e}")
        finally:
            self.after(0, lambda: self.record_btn.configure(
                state="normal", text="⏺  録音開始",
                fg_color=["#3B8ED0", "#1F6AA5"],
                hover_color=["#36719F", "#144870"],
            ))

    # ---------------------------------------------------------------
    # UIヘルパー
    # ---------------------------------------------------------------
    def _append_text(self, box: ctk.CTkTextbox, text: str):
        def _update():
            box.configure(state="normal")
            current = box.get("1.0", "end-1c")
            box.insert("end", ("\n" if current.strip() else "") + text)
            box.see("end")
        self.after(0, _update)

    def _set_status(self, msg: str):
        self.after(0, lambda: self.status_var.set(msg))

    def _clear_all(self):
        for box in (self.stt_box, self.llm_box):
            box.configure(state="normal")
            box.delete("1.0", "end")
        self._set_status("準備完了 ✅")

if __name__ == "__main__":
    app = VoiceTranscriberApp()
    app.mainloop()

stt.pyからは先ほど追加したコードを削除します。

実行結果

前回のアプリと比べて、音量メーターが追加されたことにより、テキストが表示されていなくても「録音されている」という実感があります。

漢字の変換についてはあまりうまくいっていないところ(「吾輩」->「我が輩」、「書生」 -> 「諸星」)もありますが、これは文章が悪いかもしれませんね…

ですが、句読点の挿入に関しては素晴らしい結果を得られています!

まとめと次回予告

Python 編のまとめ

  • faster-whisper + Ollama でローカル完結の音声文字起こしは実現できた
  • LLM による句読点挿入は効果的
  • リアルタイム変換は構造的に困難(Whisper はバッチ処理向け設計)
  • 音量メーター付きの手動録音方式で実用的なアプリに仕上げた

次回:Swift 編へ

Python にこだわらなければ、Swift + WhisperKit でサブ秒レイテンシが実現できるらしい

この気づきをもとに、次回は Swiftでの実装に挑戦します。Xcode 未経験からのスタートですが、Apple Silicon のポテンシャルをフル活用した実装を目指します!!