ローカルで動かす日本語音声認識 - Moonshine Tiny（PyTorch）＋ VAD 前処理

当社では、音声解析 SaaS の開発を行っており、その中で 日本語音声認識モデルの推論パイプライン が必要になりました。

求められた要件は、 音声を外部へ送信せず、外部 API に依存せずに動作する日本語音声認識 です。さらに、オンプレミス環境やオフライン環境でも利用でき、CPU 上でも実用的な速度で動作することが求められました。

検証を進める中で、軽量な日本語音声認識モデル Moonshine Tiny Japanese と、シンプルな VAD（Voice Activity Detection：音声区間検出） を組み合わせることで、小規模な構成ながら十分実用的なパイプラインを構築できることが分かりました。

本記事では、その実装方法と設計上の考え方を共有します。

最終的には、

• Moonshine Tiny Japanese を用いた日本語音声認識
• VAD による音声区間の分割
• 正規化済みの文字起こし結果の出力
• 同じパイプラインを提供するローカル Web API

までを実装し、それぞれの構成要素が必要になる理由についても説明します。

各ステップでは、

• 何をするのか
• 実際のコード
• 実行すると何が確認できるのか

を順番に確認しながら組み上げていきます。

なぜローカルで動かすのか

日本語音声認識が必要になったとき、多くの場合はクラウドの ASR API や Whisper のような大規模モデルが候補になります。多くの用途ではそれらで十分ですが、今回のような要件ではローカルで動作する軽量モデルを選択することにしました。その理由は主に次の3点です。

音声データを外部へ送信する必要がない

医療機関の問診や電子カルテ音声、金融機関のコールセンター録音、法律事務所での打ち合わせ、企業の人事・法務に関わる会議などでは、機微な個人情報を含む音声データを扱います。

日本では医療情報に関する「3省2ガイドライン」、金融機関では FISC（金融情報システムセンター）の安全対策基準、個人データ全般では個人情報保護法などが関係します。EU 圏の話者を含む場合は GDPR も考慮する必要があります。

外部 ASR API を利用すると、委託先の管理やデータの越境移転、監査証跡といった新たな論点が生じます。一方で、ローカル環境のみで完結する構成であれば、音声データは外部へ送信されず、これらの課題を大幅に減らすことができます。

オフライン・エッジ環境でも利用できる

ネットワークが不安定な環境や、そもそも外部接続を持たないデバイスでは、クラウドサービスを前提にすることはできません。

ローカルで完結する構成であれば、そのような環境でも同じパイプラインを利用できます。

レイテンシを抑えられる

クラウド API ではネットワーク往復時間が必ず発生します。一方、ローカル実行では音声データをその場で処理できるため、待ち時間を小さく抑えられます。

今回利用する Moonshine Tiny Japanese は約 2,700 万パラメータと非常に軽量で、CPU 環境でも十分実用的な速度で動作します。

Moonshine Tiny Japanese を選んだ理由

軽量モデルだからといって、必ずしも認識精度が低いわけではありません。

下のグラフは、日本語ベンチマーク（Fleurs および Common Voice 17）における CER（Character Error Rate：文字誤り率） の比較です。CER は低いほど高精度であることを示します。

2700 万パラメータの Moonshine Tiny は、3900 万パラメータの Whisper Tiny を大きく上回り、約 7 億 6900 万パラメータの Whisper Medium に匹敵、あるいはわずかに上回る結果となっています。

出典: moonshine-tiny-ja モデルカード（Flavors of Moonshine, arXiv:2509.02523）。著者らが公開したベンチマークであり、本記事のパイプラインを測定したものではありません。

それでは、ここから Moonshine Tiny Japanese と VAD を組み合わせた日本語音声認識パイプライン を、一つずつ構築していきます。

ステップ0 - 事前準備

必要なものは次の3つです。

• Python 3.10 以上
• 新しめの transformers （4.48 以降。Moonshine が入ったバージョンです）
• 16 kHz モノラルの .wav ファイル （テスト用）。GPU はあると速いですが必須ではなく、Moonshine Tiny なら CPU でも十分です。

Python とテスト音声を確認します。

python --version
# Python 3.10.x （またはそれ以降）

ffprobe -hide_banner test.wav
# ... Audio: pcm_s16le, 16000 Hz, mono ...

音声が 16 kHz モノラルでない場合は、ffmpeg で一度変換しておきます。

ffmpeg -i input.mp3 -ar 16000 -ac 1 test.wav

厳密には事前変換しなくても後述の librosa がリサンプリングしてくれますが、最初のうちは変数を一つ減らしておくと安心です。

ステップ1 - 依存ライブラリのインストール

インストールは一括で行います。これが最終スクリプトに必要なものすべてです。

pip install "transformers>=4.48" torch librosa soundfile \
            onnxruntime \
            jiwer neologdn openai-whisper

• transformers ＋ torch — モデル本体
• librosa ＋ soundfile — 音声の読み込みとリサンプリング
• onnxruntime — Silero VAD モデルの実行
• jiwer — 文字誤り率（CER）の計算
• neologdn ＋ openai-whisper — 正しい CER のための日本語テキスト正規化

VAD については、Silero を入手する方法が2通りあるので補足します。公式の pip パッケージはワンコールのヘルパーを提供します。

# pip install silero-vad
from silero_vad import load_silero_vad, read_audio, get_speech_timestamps

model = load_silero_vad()
wav = read_audio("test.wav")
speech_timestamps = get_speech_timestamps(wav, model, return_seconds=True)

これは音声区間を得る最短の方法ですが、本記事では使いません。理由は一つ、明確です。この v5 モデルは **32 ms（512 サンプル）**フレームに固定され、LSTM 状態を内部で管理するのに対し、私たちが使いたいのは **64 ms（1024 サンプル）**で状態を明示的に持つ Silero ビルドです——そうすればフレーム単位の VAD ループを自分で回せます（ステップ5の核心）。そこで pip パッケージではなく、スタンドアロンの ONNX エクスポートを onnxruntime で読み込みます。

その Silero VAD のモデルファイル が必要です。入力 speech と LSTM 状態 h_prev / c_prev を受け取るインターフェースを持つ ONNX エクスポートを、 silero_vad.onnx としてスクリプトの隣に置いてください。これは Silero の v3.1 / v4 系の JIT→ONNX エクスポート （64 ms ビルド）であり、pip パッケージの現行 v5 モデルではありません。 Silero VAD リポジトリ の v5 以前のリリースから入手できます（あるいはそのリリース同梱の TorchScript 版 VAD を torch.onnx.export で自分でエクスポートしても構いません）。現行の pip / master は v5 の 32 ms モデルのみを配布しています。インターフェースの詳細はステップ5で説明します。

補足: あとで transformers に MoonshineForConditionalGeneration が無いと言われたら、バージョンが古いので pip install -U "transformers>=4.48" で更新してください。

ステップ2 - Moonshine Tiny を PyTorch で読み込む

Moonshine Tiny Japanese は6層のエンコーダ・デコーダ型 Transformer です。読み込みは5行で済みます。

import torch
from transformers import AutoProcessor, MoonshineForConditionalGeneration

MODEL_ID = "UsefulSensors/moonshine-tiny-ja"
device = "cuda" if torch.cuda.is_available() else "cpu"

processor = AutoProcessor.from_pretrained(MODEL_ID)
model = MoonshineForConditionalGeneration.from_pretrained(MODEL_ID).to(device).eval()

print(f"Loaded on {device} | {sum(p.numel() for p in model.parameters())/1e6:.1f}M params")

実行結果: 次のように表示されます。

Loaded on cpu | 27.1M params

（初回実行時に HuggingFace から重み（数百 MB）がダウンロードされ、キャッシュされます。）

ステップ3 - ファイル全体を文字起こしする（素朴なベースライン）

最も単純に動くであろう実装です。WAV を読み込み、16 kHz にリサンプリングし、全体をそのまま model.generate() に渡してデコードします。

import librosa

def transcribe(audio, model, processor, device):
    inputs = processor(audio, sampling_rate=16000, return_tensors="pt").to(device)
    with torch.no_grad():
        generated_ids = model.generate(**inputs)
    return processor.batch_decode(generated_ids, skip_special_tokens=True)[0]

audio, _ = librosa.load("test.wav", sr=16000)   # 16 kHz モノラル float32
print(transcribe(audio, model, processor, device))

model.generate() は内部で KV キャッシュ（ use_cache=True がデフォルト）を行うため、特別な工夫なしにデコードは高速です。キャッシュを手作業で管理する必要はありません。

実行結果: 日本語の文字起こしがターミナルに表示されます。数秒程度・一話者・無音区間の少ないきれいな音声であれば、かなり良い精度になります。

ステップ4 - 素朴な実装が破綻するところを見る

同じスクリプトを、より長いファイル（30秒以上、複数の発話と無音区間を含むもの）で実行してみます。すると次のような問題が起きがちです。

1. 語頭の欠落 — 発話の最初の1〜2音が消える
2. 無音による暴走 — 長い無音区間でデコーダがハルシネーションを起こす（同じ語句の繰り返し、勝手なフィラー、関係ない文への迷走など）
3. 境界の不在 — 明確に別々の文であっても、ひとかたまりの塊として返ってくる

具体的には、実際の日本語ニュース音声（ YouTube の冒頭2分。複数話者・自然な間あり）を一括で渡すと、次のような実際の出力になります。

この春に受け取ったのではないでしょうか…街中で調査企画初任給を何に使いましたか？とか調理中に調査した…
やっぱりあの調理器とか調理器とか調理器とか調理器とか…   ← 「調理器とか」が延々と続き、末尾で途切れる

3つの問題が同時に起きています。冒頭の一文が丸ごと欠落し（語頭の欠落）、デコーダは「調理器とか」の繰り返しループに陥り（このループは音声には無い、モデルのハルシネーションです）、さらに出力はクリップの途中で途切れ、全体がひとかたまりの塊として返ってきます。これが、この後のすべての動機になります。Moonshine をはじめとする短コンテキストの ASR モデルは、無音を含む2分の録音ではなく、 1発話ずつ 与えられることを期待しています。そこでセグメント分割を行います。

ステップ5 - Silero VAD でセグメント分割する - 窓ごとの解析と音声の蓄積（本題）

本記事の核心です。 Silero VAD で音声が実際にある場所を見つけ、十分な量の音声を蓄積してから、意味のあるまとまりを Moonshine に渡します。下の図は前処理の流れ全体——生の音声から VAD の窓処理を経て、Moonshine が受け取る無音のないまとまりまで——を示しています。

このパイプラインは5つの可動部からなります。

1. 10秒の窓に切り分ける — 長い録音は1窓ずつ処理します。Silero がきれいに解析できる程度の小ささです。
2. VAD をフレーム単位で実行する — 各窓の中で、Silero は 64 ms ごとのフレームすべてにスコアを付けます。0.5 以上は音声とみなします。
3. 連続するフレームをセグメントにまとめる — 隣り合う音声フレームは一つの「音声セグメント」に、隣り合う非音声フレームは「無音」にまとまります。
4. 各音声セグメントを ±200 ms パディングする — 語頭や語尾を削らないようにするためです。
5. 音声が 8 秒に達するまで窓をまたいで蓄積する — そのうえで、音声だけの波形を連結し——無音の隙間は除去して——ASR に渡せる一つのまとまりにします。

5a. Silero VAD を読み込む

Silero は ONNX Runtime 経由で動かします。このエクスポートは LSTM 状態を明示的に外へ出す構成で、フレームと直前の隠れ状態・セル状態（ h_prev 、 c_prev ）を渡すと、音声確率と更新後の状態が返ってきます。

import onnxruntime as ort

VAD_MODEL_PATH = "silero_vad.onnx"   # h/c の LSTM 状態を入出力する ONNX エクスポート
vad_session = ort.InferenceSession(VAD_MODEL_PATH)

モデルの入出力は次のとおりです。

入力  →  speech [1, 1024] float32   （64 ms ぶんの1フレーム）
        h_prev [2, 1, 64] float32   （LSTM 隠れ状態）
        c_prev [2, 1, 64] float32   （LSTM セル状態）
出力  →  prob   [1, 1]    float32   （音声確率）
        h_next, c_next              （次フレームへ引き継ぐ状態）

5b. 窓ベースのパイプライン

順に呼び出す4つの関数です。まず run_vad が、窓の中のすべてのフレームにスコアを付けます。

import numpy as np

SAMPLE_RATE        = 16000
VAD_FRAME_SIZE     = 1024                       # 64 ms @ 16 kHz
WINDOW_SAMPLES     = int(10.0 * SAMPLE_RATE)    # 10 秒の解析窓
VOICE_THRESHOLD    = 0.5                        # 音声確率のしきい値
PRE_PAD_SAMPLES    = int(0.2 * SAMPLE_RATE)     # 各音声セグメントの前 200 ms
POST_PAD_SAMPLES   = int(0.2 * SAMPLE_RATE)     # 後ろ 200 ms
MIN_VOICE_DURATION = 8.0                        # ASR を発火させる前に蓄積する音声の長さ


def run_vad(audio, vad_session):
    """64 ms ごとのフレームすべてにスコアを付ける。音声確率のリストを返す。"""
    h = np.zeros((2, 1, 64), dtype=np.float32)
    c = np.zeros((2, 1, 64), dtype=np.float32)
    probs = []
    for i in range(len(audio) // VAD_FRAME_SIZE):
        frame = audio[i * VAD_FRAME_SIZE : (i + 1) * VAD_FRAME_SIZE]
        out = vad_session.run(None, {
            "speech": frame[None, :].astype(np.float32),
            "h_prev": h, "c_prev": c,
        })
        probs.append(float(out[0][0][0]))
        h, c = out[1], out[2]
    return probs

次に group_into_segments が、連続する同じクラスのフレームをまとめます。しきい値を超えるフレームの連なりは一つの音声セグメントに、下回る連なりは一つの無音セグメントになります。

def group_into_segments(probs):
    """連続する同クラスのフレームを (start_frame, end_frame, is_voice) のタプルにまとめる。"""
    if not probs:
        return []
    segments, start, is_voice = [], 0, probs[0] >= VOICE_THRESHOLD
    for i in range(1, len(probs)):
        v = probs[i] >= VOICE_THRESHOLD
        if v != is_voice:
            segments.append((start, i, is_voice))
            start, is_voice = i, v
    segments.append((start, len(probs), is_voice))
    return segments

続いて extract_voice_segments が、各音声セグメントの波形を取り出し、±200 ms のパディングを足して語頭や語尾を削らないようにします。

def extract_voice_segments(segments, audio):
    """±200 ms パディング付きで音声波形を取り出す。[(waveform, voice_duration), ...] を返す。"""
    voice = [(sf, ef) for sf, ef, is_voice in segments if is_voice]
    out, last_end = [], 0
    for i, (sf, ef) in enumerate(voice):
        s0 = sf * VAD_FRAME_SIZE
        e0 = ef * VAD_FRAME_SIZE
        dur = (e0 - s0) / SAMPLE_RATE
        ps = max(last_end, s0 - PRE_PAD_SAMPLES, 0)
        pe = e0 + POST_PAD_SAMPLES
        if i + 1 < len(voice):                          # 次のセグメントへはみ出さない
            pe = min(pe, voice[i + 1][0] * VAD_FRAME_SIZE)
        pe = min(pe, len(audio))
        out.append((audio[ps:pe].copy(), dur))
        last_end = pe
    return out

最後に segment_audio がパイプライン全体を駆動します。10 秒の窓に切り分け、音声を蓄積し、音声が 8 秒に達するたび（あるいは完全に無音の窓が来たとき、もしくはファイルが終わったとき）に ASR 用のまとまりを吐き出します。

def segment_audio(audio, vad_session):
    """音声を 10 秒窓に切り分け → 音声を蓄積 → ASR 用のまとまり（隙間除去済み）を返す。"""
    pending = []            # 連結待ちの音声波形配列
    pending_dur = 0.0       # ここまでに蓄積した音声のみの秒数
    chunks = []

    offset = 0
    while offset < len(audio):
        window = audio[offset : offset + WINDOW_SAMPLES]
        if len(window) < VAD_FRAME_SIZE:
            break

        probs = run_vad(window, vad_session)
        segs  = group_into_segments(probs)
        parts = extract_voice_segments(segs, window)

        if not parts and pending:               # 無音の窓 → 溜まっている分を吐き出す
            chunks.append(np.concatenate(pending))
            pending, pending_dur = [], 0.0

        for waveform, dur in parts:
            pending.append(waveform)
            pending_dur += dur
            if pending_dur >= MIN_VOICE_DURATION:   # 十分な音声 → ASR 用のまとまり
                chunks.append(np.concatenate(pending))
                pending, pending_dur = [], 0.0

        offset += WINDOW_SAMPLES

    if pending:                                 # ファイル終端で残りを吐き出す
        chunks.append(np.concatenate(pending))

    return chunks

4つの定数がすべてを左右します。

• VOICE_THRESHOLD = 0.5 — 音声確率がこの値以上のフレームを音声と分類します。下げると弱い音声を拾いやすくなりますが誤検出も増えます。日本語では 0.5 がきれいな境目です。
• PRE_PAD / POST_PAD = 200 ms — 各音声セグメントの前後に 200 ms ぶんの音声を足します。これがステップ4の語頭欠落を回収し——音声が始まると助走区間が無償で手に入り——語尾の子音も取りこぼしません。
• WINDOW_SAMPLES = 10 秒 — VAD に渡す各解析窓の大きさです。意味のある音声を含むには十分大きく、メモリを圧迫せず逐次処理できる程度には小さい値です。
• MIN_VOICE_DURATION = 8.0 秒 — 音声セグメントは窓をまたいで蓄積され、合計が 8 秒に達してはじめて連結し（無音は除去して）Moonshine に渡します。不必要に長い音声を与えずに、デコーダへ正確な文字起こしに足る文脈を持たせるためです。

隙間の除去が肝心な発想です。 pending リストを連結するとき、各要素は音声波形であり——それらの あいだ の無音はすでに消えています。Moonshine が見るのは「音声-無音-音声」ではなく連続した音声であり、これがステップ4の間（ま）によるハルシネーションを取り除きます。

VAD モデルについての注記: ここでは Silero の LSTM 状態を明示的に持つ ONNX エクスポート （入力 speech ＋ h_prev / c_prev 、出力 prob ＋更新状態）と 1024 サンプル フレームを使います。ステップ1で述べたとおり、フレーム単位のループを自分で回すために、pip 版 v5 モデル（入出力仕様が異なり、32 ms／512 サンプルフレーム）ではなくこちらのビルドを選んでいます。ステップ8のサーバは、このセグメント分割コードをそのまま再利用します。

5c. 各まとまりを文字起こしして連結する

segment_audio が返す各まとまりは、隙間のない波形——典型的にはおよそ 8 秒の連続音声です。それぞれを Moonshine に渡し、結果を連結します。

def transcribe_chunks(audio, vad_session, model, processor, device):
    chunks = segment_audio(audio, vad_session)
    texts = []
    for chunk in chunks:
        if len(chunk) < 1600:          # 0.1 秒未満の切れ端はスキップ
            continue
        texts.append(transcribe(chunk, model, processor, device))
    return "\n".join(texts)

audio, _ = librosa.load("test.wav", sr=16000)
print(transcribe_chunks(audio, vad_session, model, processor, device))

実行結果: ステップ4と同じ2分のクリップが、12個のまとまりにセグメント分割され、語頭を保ったまま、繰り返しループのない出力になります（実際の出力。先頭の数行を示します）。

新年度も早１か月、この春入社した多くの新社会人が、初任給を受け取ったのではないでしょうか。
そこで今回は街中で調査企画初任給を何に使いましたか？
ちなみに、おいくらぐらいでしょうか、１０ぐらい…
自分に服を買いました、たぶん、もともと服が好きだったので…
（この後、全12セグメントが1行ずつ続く）

欠落していた冒頭の一文が戻り、「調理器とか」の繰り返しループが消え、各発話が1行ずつに収まりました。Moonshine が音声を見る前に無音の隙間を除いているため、デコーダはハルシネーションの原因だった空白に出会いません。

実測すると、同じ2分のクリップで文字誤り率（CER）は、素朴な一括処理の 85% から VAD パイプラインの 29% まで下がりました（Cloud Speech-to-Text の文字起こしを参照とした相対比較）。改善のほぼすべてはセグメント分割によるもので、音声のノイズ除去では変わりません。

ステップ6 - CER を測る前に日本語テキストを正規化する

文字誤り率（CER）で精度を測るなら、仮説（hypothesis）とリファレンス（reference）の 両方 を先に正規化します。これを怠ると、半角・全角や句読点の違いだけで CER が実態より2〜3%悪く見え、存在しない劣化を半日追いかける羽目になります。

import unicodedata
import neologdn
from whisper.normalizers import BasicTextNormalizer
from jiwer import cer

_basic = BasicTextNormalizer()

def normalize_japanese(text):
    text = _basic(text)                                  # 句読点・空白の整理
    text = neologdn.normalize(text, tilde="normalize")   # 日本語固有の正規化
    text = unicodedata.normalize("NFKC", text)           # 半角・全角の統一
    return text.lower()

def compute_cer(reference, hypothesis):
    return cer(normalize_japanese(reference), normalize_japanese(hypothesis))

落とし穴を一例で示します。次の2つの文字列は 同じことを言っています 。

reference  = "２０２４年の売上は１０％増加しました。"   # 全角数字・全角％・句点あり
hypothesis = "2024年の売上は10%増加しました"            # 半角数字・ASCII の %・句点なし

cer(reference, hypothesis)                       # ≈ 0.30 — ひどく見える
compute_cer(reference, hypothesis)               # = 0.00 — 実際は同一

生の CER は、純粋に表記揺れだけの8文字を「誤り」として数えています。NFKC 正規化後は両者が同じ文字列になります。実際の30秒程度のクリップでも、この半角・全角や句読点のノイズだけで数%ぶんの見かけ上の CER を損します。

実行結果: 同じ文字起こしでも、正規化後はスコアが意味のある値まで下がります（水増しされた値ではなく）。

ステップ7 - 全体を一つにまとめる

ステップ2・5・6 を一つにまとめたものが以下です。 simple_japanese_asr.py として保存します。

"""シンプルなローカル日本語 ASR: Moonshine Tiny（PyTorch）＋ Silero VAD（ONNX）。"""
import sys
import numpy as np
import librosa
import torch
import onnxruntime as ort
import unicodedata
import neologdn
from transformers import AutoProcessor, MoonshineForConditionalGeneration
from whisper.normalizers import BasicTextNormalizer

MODEL_ID           = "UsefulSensors/moonshine-tiny-ja"
VAD_MODEL_PATH     = "silero_vad.onnx"
SAMPLE_RATE        = 16000
VAD_FRAME_SIZE     = 1024                       # 64 ms @ 16 kHz
WINDOW_SAMPLES     = int(10.0 * SAMPLE_RATE)    # 10 秒の解析窓
VOICE_THRESHOLD    = 0.5
PRE_PAD_SAMPLES    = int(0.2 * SAMPLE_RATE)     # ±200 ms パディング
POST_PAD_SAMPLES   = int(0.2 * SAMPLE_RATE)
MIN_VOICE_DURATION = 8.0                        # ASR 発火までに蓄積する音声の秒数

device      = "cuda" if torch.cuda.is_available() else "cpu"
processor   = AutoProcessor.from_pretrained(MODEL_ID)
model       = MoonshineForConditionalGeneration.from_pretrained(MODEL_ID).to(device).eval()
vad_session = ort.InferenceSession(VAD_MODEL_PATH)
_basic      = BasicTextNormalizer()


def run_vad(audio):
    h = np.zeros((2, 1, 64), dtype=np.float32)
    c = np.zeros((2, 1, 64), dtype=np.float32)
    probs = []
    for i in range(len(audio) // VAD_FRAME_SIZE):
        frame = audio[i * VAD_FRAME_SIZE:(i + 1) * VAD_FRAME_SIZE]
        out = vad_session.run(None, {
            "speech": frame[None, :].astype(np.float32), "h_prev": h, "c_prev": c,
        })
        probs.append(float(out[0][0][0]))
        h, c = out[1], out[2]
    return probs


def group_into_segments(probs):
    if not probs:
        return []
    segs, start, is_v = [], 0, probs[0] >= VOICE_THRESHOLD
    for i in range(1, len(probs)):
        v = probs[i] >= VOICE_THRESHOLD
        if v != is_v:
            segs.append((start, i, is_v))
            start, is_v = i, v
    segs.append((start, len(probs), is_v))
    return segs


def extract_voice_segments(segs, audio):
    voice = [(sf, ef) for sf, ef, is_v in segs if is_v]
    out, last = [], 0
    for i, (sf, ef) in enumerate(voice):
        s0, e0 = sf * VAD_FRAME_SIZE, ef * VAD_FRAME_SIZE
        ps = max(last, s0 - PRE_PAD_SAMPLES, 0)
        pe = e0 + POST_PAD_SAMPLES
        if i + 1 < len(voice):
            pe = min(pe, voice[i + 1][0] * VAD_FRAME_SIZE)
        pe = min(pe, len(audio))
        out.append((audio[ps:pe].copy(), (e0 - s0) / SAMPLE_RATE))
        last = pe
    return out


def segment_audio(audio):
    pending, pending_dur, chunks = [], 0.0, []
    offset = 0
    while offset < len(audio):
        window = audio[offset:offset + WINDOW_SAMPLES]
        if len(window) < VAD_FRAME_SIZE:
            break
        parts = extract_voice_segments(group_into_segments(run_vad(window)), window)
        if not parts and pending:
            chunks.append(np.concatenate(pending))
            pending, pending_dur = [], 0.0
        for waveform, dur in parts:
            pending.append(waveform)
            pending_dur += dur
            if pending_dur >= MIN_VOICE_DURATION:
                chunks.append(np.concatenate(pending))
                pending, pending_dur = [], 0.0
        offset += WINDOW_SAMPLES
    if pending:
        chunks.append(np.concatenate(pending))
    return chunks


def transcribe(audio):
    inputs = processor(audio, sampling_rate=16000, return_tensors="pt").to(device)
    with torch.no_grad():
        ids = model.generate(**inputs)
    return processor.batch_decode(ids, skip_special_tokens=True)[0]


def normalize_japanese(text):
    text = _basic(text)
    text = neologdn.normalize(text, tilde="normalize")
    return unicodedata.normalize("NFKC", text).lower()


if __name__ == "__main__":
    audio, _ = librosa.load(sys.argv[1], sr=SAMPLE_RATE)
    lines = [transcribe(chunk) for chunk in segment_audio(audio) if len(chunk) >= 1600]
    print("\n".join(lines))

実行します。

python simple_japanese_asr.py test.wav

実行結果: 語頭が保たれ、無音による暴走のない、きれいにセグメント分割された日本語の文字起こしが表示されます。これで動くローカル日本語 ASR ができました。

normalize_japanese は、リファレンスと突き合わせて採点したいとき用に同梱しています（ from jiwer import cer; cer(normalize_japanese(ref), normalize_japanese(hyp)) ）。

ステップ8 - Web ページとしてホスティングする（小さなローカルサーバ）

ここまではターミナルからスクリプトを実行してきました。これをブラウザから使えるように——ページにファイルをドロップし、文字起こしを受け取る——するには、ステップ7のパイプラインをそのまま小さな Web サーバで包みます。モデルは、いま手元にある 素の PyTorch モデル のままです。エクスポートも分解も一切しません。ブラウザは入口にすぎません。

最初にひとつトレードオフを明確にしておきます。モデルはサーバ側で動くので、音声はそのサーバが動く場所へアップロードされます。自分のマシンや信頼できるホストなら問題ありませんが、オンデバイスのアプリではありません——音声はユーザのブラウザから外に出ます。

ここでは Flask を使います。

pip install flask

8a. サーバ

server.py は、ステップ7のスクリプトに少しだけ Web の糊付けを足したものです。起動時にモデルと VAD を 一度だけ 読み込み、 / でページを返し、アップロードされたファイルを受け取ってテキストを返す /transcribe を公開します。インポート、 run_vad 、 group_into_segments 、 extract_voice_segments 、 segment_audio 、 transcribe はステップ7とまったく同じで、新しいのは末尾だけです。

# ... 上にステップ7すべて: import、model / processor / vad_session、segment_audio()、transcribe() ...

import io
from flask import Flask, request, jsonify, render_template_string

app = Flask(__name__)

@app.route("/")
def index():
    return render_template_string(PAGE)

@app.route("/transcribe", methods=["POST"])
def transcribe_endpoint():
    audio, _ = librosa.load(io.BytesIO(request.files["audio"].read()), sr=SAMPLE_RATE)
    lines = [transcribe(chunk) for chunk in segment_audio(audio) if len(chunk) >= 1600]
    return jsonify({"text": "\n".join(lines)})

if __name__ == "__main__":
    app.run(port=8000)

librosa.load はアップロードされたバイト列を io.BytesIO から直接読み、16 kHz へリサンプルするので、ステップ7の segment_audio ＋ transcribe はそのまま動きます。モデルは起動時に一度だけ読み込まれるため、各リクエストは純粋な推論だけです——呼び出しごとの再読み込みはありません。

8b. ページ

/ が返す最小限のページです。 /transcribe へ POST して結果を表示するファイル入力だけを置きます。サーバが描画する PAGE 文字列として定義します。

PAGE = """
<!DOCTYPE html>
<html lang="ja">
<head><meta charset="UTF-8"><title>Moonshine Japanese ASR</title></head>
<body>
  <input id="file" type="file" accept="audio/*">
  <pre id="result"></pre>
  <script>
    document.getElementById("file").onchange = async (e) => {
      const body = new FormData();
      body.append("audio", e.target.files[0]);
      document.getElementById("result").textContent = "…";
      const res = await fetch("/transcribe", { method: "POST", body });
      document.getElementById("result").textContent = (await res.json()).text;
    };
  </script>
</body>
</html>
"""

8c. 実行する

python server.py
# Running on http://127.0.0.1:8000

http://localhost:8000 を開き、音声ファイルを選ぶと、文字起こしが表示されます。モデルはサーバ側の PyTorch で動きます——ONNX もモデルの分解もありません。ページは音声を送って、返ってきたテキストを描画するだけです。

実行結果: ファイルを選ぶと少し待って、セグメント分割された日本語の文字起こしが埋め込まれます——ステップ7と同じ出力が、ブラウザ上で得られます。

公開するには: モデルはサーバプロセスの中で動くので、「ホスティング」とは、この Python プロセスをどこか到達可能な場所——小さな VPS、コンテナ、社内マシンなど——で動かすことを意味します（静的ファイルのアップロードではありません）。実トラフィックでは、Flask の開発サーバではなく本番用サーバ（ gunicorn server:app ）の背後で動かしてください。

性能（実測）

参考までに、本記事のコードを Apple M4（CPU のみ、PyTorch 2.12、4 スレッド、fp32） で実測した結果です。入力は前述の2分のニュース音声です。

処理	所要時間（2分の音声）	実時間比（速さ）
一括処理（素朴なベースライン）	約 2.4 秒	約 50倍
VAD パイプライン（全体）	約 2.5 秒	約 48倍
└ うち VAD セグメント分割	約 0.2 秒	—
└ うち ASR（12チャンク）	約 2.3 秒	—

ポイントは、 精度の大きな改善（CER 85% → 29%）が、レイテンシをほとんど増やさずに得られている ことです。VAD の前処理自体は2分の音声で 0.2 秒ほどしかかからず、文字起こしの総時間は一括処理とほぼ変わりません——どちらも CPU だけで実時間のおよそ50倍の速さで処理できます。

メモリ面では、モデル本体は fp32 で約 108 MB、ライブラリとモデルを読み込んだ待機状態でプロセス全体が約 0.5 GB です。ピーク使用量は処理方式で変わります——VAD パイプライン（短いチャンクのみを処理）では約 0.8 GB、一括処理では2分の音声をまるごと前向き計算するためアクティベーションが膨らみ約 1.1 GB に達しました。つまり VAD は、より高精度なだけでなくピークメモリも小さく抑えられます。これらはすべて GPU を使わない CPU だけの数字です。

まとめ

Moonshine Tiny（日本語）を素の PyTorch だけでローカル動作させ、長尺・多話者の実音声に耐える文字起こしパイプラインを組み立ててきました。本記事の要点を整理します。

• VAD 前処理が精度の決め手だった。 2分のニュース音声で、一括入力の CER は 85%、VAD で無音を除いて 10 秒前後のまとまりに分割すると 29% まで下がりました（相対 65% 改善）。改善は音声の「クリーニング」ではなくセグメント分割そのものから来ています——デノイズを試しても CER は動きませんでした。
• 失敗は構造的に現れる。 長い音声を一括で渡すと、冒頭の取りこぼし・無音区間で起きる同一フレーズの反復（ハルシネーション）・途中での打ち切りが同時に出ます。音声区間だけを 10 秒前後に切り出して渡すことで、これらが一度に解消しました。
• 推奨構成。 ステップ5で使った値がそのまま出発点になります—— VAD_FRAME_SIZE = 1024 （64 ms フレーム）、 VOICE_THRESHOLD = 0.5 、前後 ±200 ms のパディング、 WINDOW_SAMPLES = 10 秒 、 MIN_VOICE_DURATION = 8.0 秒 。日本語のニュース音声ではこの組み合わせがきれいに効きました。自分の音声で精度が伸び悩むときは、まず VOICE_THRESHOLD とパディング幅から調整するのが見通しが良いです。
• CPU だけで実用速度。 Apple M4（CPU）で2分の音声をおよそ2.5秒——実時間の約50倍の速さ——で処理でき、精度の改善はレイテンシをほとんど増やしません。モデル本体は fp32 で約 108 MB です。
• 構成はシンプルなまま。 モデルは PyTorch プロセスの中で動き、 AutoProcessor ＋ MoonshineForConditionalGeneration の generate() を呼ぶだけです。ステップ8のとおり、同じ segment_audio ＋ transcribe を小さな HTTP サーバの背後に置けば、ローカルで完結したデモになります。