この記事で作るもの
ブラウザで話すと、自分のサーバーだけで日本語を文字起こしする Web アプリを作ります。外部の音声認識 API は使わず、GPU なしの PC でも動きます。
完成すると、次のことができます。
- 話している間は途中結果が薄い色で出て、話し終えると確定文に切り替わる
- 確定文を画面で直して保存できる
- 専門用語をヒントとして渡し、CER(文字誤り率)で精度を測れる
筆者の環境では、CPU だけで CER 1.9%(Whisper large-v3-turbo、用語ヒント3語、数字の表記をそろえた値)まで出ました。精度の詳しい話は note の記事 にまとめています。
前提
- Docker Desktop(または Docker Engine と Docker Compose)
- Python 3.10 以上(モデル取得用のスクリプトを動かすため)
- Chrome または Edge(マイクを使うので
localhostか HTTPS で開く) - メモリ 4GB 以上の空き(
smallモデルの場合)
完成版のコードは yk-planning/ja-asr-demo にあります。この記事では要点だけを抜き出して説明するので、全体はリポジトリを見ながら進めてください。
全体の構成
サーバーは FastAPI の1つだけです。話している間は軽い ReazonSpeech が途中結果を出し、無音で区切られたら Whisper が確定します。
| 役割 | 使うもの | 手順 |
|---|---|---|
| 録音してまとめて送る |
MediaRecorder + fetch
|
手順3 |
| 話しながら送る | AudioWorklet + WebSocket | 手順4 |
| 区切る・確定する | NumPy(音量の判定)+ faster-whisper | 手順2、5 |
| 途中結果を出す | sherpa-onnx + ReazonSpeech | 手順6 |
| 保存する | SQLite(文)と WAV ファイル(音声) | 手順2 |
手順1: フォルダと依存ライブラリを用意する
最終的に、次のファイルが並びます。
ja-asr-demo/
├── server.py # API と画面の配信、Whisper の呼び出し
├── live.py # WebSocket でのリアルタイム認識
├── glossary.txt # 用語ヒント(1行1語)
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
├── static/
│ ├── index.html # 録音画面
│ ├── live.html # リアルタイム画面
│ └── pcm-worklet.js # マイク音声を 16kHz PCM にする
├── scripts/
│ ├── fetch_sherpa.py # 途中表示用モデルの取得
│ └── cer.py # CER の計測
└── data/ # 録音・DB・モデル(Git には入れない)
requirements.txt は次のとおりです。
fastapi
uvicorn[standard]
python-multipart
faster-whisper>=1.1.0
av<19 # PyAV 19 で faster-whisper 1.2.1 が動かなくなるため上限を固定
numpy
sherpa-onnx
-
uvicorn[standard]にすると WebSocket 用のライブラリも入ります。 -
python-multipartは、音声ファイルのアップロードを受け取るのに必要です。 -
data/には話した人の声がたまるので、必ず.gitignoreに入れておきます。
手順2: faster-whisper で文字起こしする API を作る
音声ファイルを受け取って Whisper にかけ、結果を SQLite に保存する API を FastAPI で作ります。モデルは起動時に1回だけ読み込みます。
# server.py(抜粋)
import os, threading
from faster_whisper import WhisperModel
model = WhisperModel(
os.getenv("ASR_MODEL", "small"),
device=os.getenv("ASR_DEVICE", "auto"), # GPU がなければ cpu
compute_type=os.getenv("ASR_COMPUTE", "default"), # CPU なら int8 が速い
)
model_lock = threading.Lock()
def run_whisper(path: str) -> str:
with model_lock: # 1つのモデルを複数のリクエストで共有するので直列にする
segs, _ = model.transcribe(
path, language="ja", beam_size=5, vad_filter=True,
initial_prompt=glossary_prompt(), # 手順8で使う用語ヒント
)
return "".join(s.text for s in segs).strip()
-
language="ja"を固定すると、短い発話で言語を誤判定しなくなります。 -
vad_filter=Trueで無音部分を飛ばすので、無音から存在しない文を作り出すことが減ります。 -
transcribe()はジェネレーターを返すので、"".join()で回し切ったときに初めて認識が走ります。ロックの内側で回し切るのがポイントです。
アップロードを受け取る API は次のようになります。
@app.post("/api/transcribe")
def transcribe(file: UploadFile = File(...)):
ext = Path(file.filename or "a.webm").suffix or ".webm"
uid = uuid.uuid4().hex
path = AUDIO / f"{uid}{ext}"
path.write_bytes(file.file.read())
text = run_whisper(str(path))
insert_row(uid, path.name, text) # asr_text(認識結果)と text(訂正後)の両方に入れる
return {"id": uid, "text": text}
-
async defではなくdefにしています。認識は重い同期処理なので、FastAPI が別スレッドで動かしてくれるdefのほうが、他のリクエストを止めません。 - WebM も MP4 も、faster-whisper が PyAV で読めます。ffmpeg を別に入れる必要はありません。
- DB には「認識結果」と「訂正後」を別の列で持ちます。あとで差から CER を出したり、再学習に使ったりできます。
画面(static/)は同じサーバーから配信します。/ にマウントするとそれ以降のルートが隠れるので、ファイルの最後に書きます。
live.attach(app, run_whisper, insert_row, AUDIO, MODEL_NAME) # 手順5 の WebSocket
app.mount("/", StaticFiles(directory="static", html=True), name="static")
手順3: ブラウザで録音して送る
まずは「録音して、止めたらまとめて送る」だけの画面を作ります。MediaRecorder を使えば数十行ですみます。
// static/index.html(抜粋)
function pickMime(){
for (const m of ['audio/webm;codecs=opus', 'audio/webm', 'audio/mp4'])
if (window.MediaRecorder && MediaRecorder.isTypeSupported(m)) return m;
return '';
}
recBtn.onclick = async () => {
if (recorder && recorder.state === 'recording') { recorder.stop(); return; }
stream = await navigator.mediaDevices.getUserMedia({audio: true});
const mime = pickMime();
recorder = new MediaRecorder(stream, mime ? {mimeType: mime} : undefined);
chunks = [];
recorder.ondataavailable = e => e.data.size && chunks.push(e.data);
recorder.onstop = async () => {
stream.getTracks().forEach(t => t.stop()); // マイクを解放する
const type = recorder.mimeType || 'audio/webm';
const fd = new FormData();
fd.append('file', new Blob(chunks, {type}), 'rec.' + (type.includes('mp4') ? 'mp4' : 'webm'));
const r = await fetch('/api/transcribe', {method: 'POST', body: fd});
show(await r.json());
};
recorder.start();
};
- Safari は WebM で録音できないので、MP4 に切り替えています。拡張子を合わせて送れば、サーバー側はそのまま読めます。
-
navigator.mediaDevicesはlocalhostか HTTPS のときしか使えません。別の PC からhttp://で開くとundefinedになるので、画面に理由を出しておくと親切です。 - 返ってきた文は
textareaに入れ、直したらPUT /api/items/{id}で保存します(リポジトリを参照)。
ここまでで、uvicorn server:app で起動すれば「話す → 止める → 数秒後に文字が出る」ところまで動きます。次の手順から、話している間に文字を出す仕組みを足します。
手順4: AudioWorklet で 16kHz PCM を WebSocket で送る
話している途中で認識するには、音声を少しずつサーバーに送り続けます。ここでは MediaRecorder ではなく、AudioWorklet で生の音声(PCM)を取り出します。
MediaRecorder を使わない理由は2つです。
-
timesliceで分けた WebM の断片は、先頭以外を単体ではデコードできない - サーバー側では無音の判定に音量が要るので、生の波形のほうが扱いやすい
Whisper も ReazonSpeech も 16kHz の音声を前提にしているので、ブラウザの中で 16kHz に落とし、Int16 にして送ります。float32 のまま送るより通信量が半分になります。
// static/pcm-worklet.js
// マイク音声を16kHz/Int16/約100msのかたまりにしてメインスレッドへ渡す
class PCMWorklet extends AudioWorkletProcessor {
constructor() {
super();
this.ratio = sampleRate / 16000; // 48kHz なら 3
this.phase = 0; this.acc = 0; this.n = 0;
this.out = new Int16Array(1600); this.len = 0; // 1600サンプル = 0.1秒
}
process(inputs) {
const ch = inputs[0] && inputs[0][0];
if (!ch) return true;
for (let i = 0; i < ch.length; i++) {
this.acc += ch[i]; this.n++; this.phase += 1;
if (this.phase >= this.ratio) { // ratio 個ごとに平均して1サンプルにする
const v = Math.max(-1, Math.min(1, this.acc / this.n));
this.out[this.len++] = v < 0 ? v * 32768 : v * 32767;
this.phase -= this.ratio; this.acc = 0; this.n = 0;
if (this.len === this.out.length) {
this.port.postMessage(this.out.buffer, [this.out.buffer]);
this.out = new Int16Array(1600); this.len = 0;
}
}
}
return true;
}
}
registerProcessor('pcm-worklet', PCMWorklet);
画面側では、受け取ったかたまりをそのまま WebSocket に流します。
// static/live.html(抜粋)
stream = await navigator.mediaDevices.getUserMedia(
{audio: {channelCount: 1, echoCancellation: true, noiseSuppression: true}});
ws = new WebSocket((location.protocol === 'https:' ? 'wss://' : 'ws://') + location.host + '/ws');
ws.binaryType = 'arraybuffer';
ws.onmessage = ev => onMsg(JSON.parse(ev.data)); // partial / final を画面に出す
ctx = new AudioContext();
await ctx.audioWorklet.addModule('/pcm-worklet.js');
const src = ctx.createMediaStreamSource(stream);
node = new AudioWorkletNode(ctx, 'pcm-worklet');
node.port.onmessage = e => { if (ws.readyState === 1) ws.send(e.data); };
const mute = ctx.createGain(); mute.gain.value = 0; // 自分の声がスピーカーから出ないように
src.connect(node); node.connect(mute); mute.connect(ctx.destination);
- 通話を終えるときは、文字列
flushを送って、話しかけの音声を確定させてから切ります。 - 接続先を
locationから組み立てておくと、HTTPS で公開したときも自動でwss://になります。
手順5: 無音で区切って Whisper で確定する
サーバーは届いた音声をためていき、無音が 0.8 秒続いたら「1つの発話が終わった」と判断して Whisper にかけます。無音の判定は、100ms ごとの音量(RMS)を見るだけの簡易版です。
# live.py(抜粋。Session は WebSocket 接続1つごとに作る)
RMS_THR = float(os.getenv("VAD_RMS", "0.012")) # 発話と判定する音量
SIL_SEC = float(os.getenv("VAD_SILENCE_SEC", "0.8")) # この秒数の無音で区切る
MAX_SEC = float(os.getenv("MAX_UTT_SEC", "25")) # 長すぎる発話は強制的に区切る
async def feed(self, data: bytes):
x = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
loud = float(np.sqrt(np.mean(x * x))) > RMS_THR
if not self.speech:
if not loud:
self.pre.append(x) # 直前 0.3 秒を取っておく(語頭が切れないように)
return
self.speech = True
self.buf = list(self.pre) + [x]
self.n = sum(c.size for c in self.buf); self.sil = 0
else:
self.buf.append(x); self.n += x.size
self.sil = 0 if loud else self.sil + x.size
if self.sil >= SIL_SEC * SR or self.n >= MAX_SEC * SR:
await self.finish() # 区切りが来た → Whisper で確定
elif rec and not self.busy and time.monotonic() - self.last_partial >= PARTIAL_EVERY:
self.spawn(self.partial(np.concatenate(self.buf), self.utt)) # 手順6 の途中結果
確定の処理は、音声を WAV に保存してから手順2の run_whisper を呼ぶだけです。
async def final(self, audio, seq):
uid = uuid.uuid4().hex
path = audio_dir / f"{uid}.wav"
save_wav(path, audio)
text = await self.loop.run_in_executor(final_pool, run_whisper, str(path))
insert_row(uid, path.name, text, self.sid, seq)
await self.send({"type": "final", "id": uid, "text": text, "seq": seq})
- Whisper は数秒かかるので、
run_in_executorで別スレッドに逃がします。そのままawaitの外で呼ぶと、その間は WebSocket の受信が止まります。 - 区切りは
MAX_UTT_SEC(既定25秒)で強制的に切ります。Whisper は30秒単位で処理するので、それより短く保つためです。 - 0.4 秒未満の音(咳払いや物音)は捨てて、Whisper にかけません。
- 騒がしい場所で区切れないときは
VAD_RMSを上げ、声を拾わないときは下げます。
手順6: sherpa-onnx(ReazonSpeech)で途中結果を出す
手順5までだと、話し終えて数秒たつまで画面に何も出ません。そこで、軽い日本語モデルで話している途中の音声を認識し、薄い色で出します。筆者の環境では、話し始めて約 0.7 秒で文字が出始めました。
モデルは sherpa-onnx が配布している sherpa-onnx-zipformer-ja-reazonspeech-2024-08-01 を使います。配布物は 680MB ありますが、使うのは int8 版を中心にした4ファイル(約 170MB)だけなので、それだけを取り出すスクリプトを用意しました。
python scripts/fetch_sherpa.py # data/sherpa-ja/ に4ファイルを置く
認識器の読み込みと、途中結果の出し方は次のとおりです。
# live.py(抜粋)
import sherpa_onnx
rec = sherpa_onnx.OfflineRecognizer.from_transducer(
encoder="data/sherpa-ja/encoder-epoch-99-avg-1.int8.onnx",
decoder="data/sherpa-ja/decoder-epoch-99-avg-1.onnx",
joiner="data/sherpa-ja/joiner-epoch-99-avg-1.int8.onnx",
tokens="data/sherpa-ja/tokens.txt",
num_threads=2, sample_rate=16000, feature_dim=80, decoding_method="greedy_search",
)
def decode_partial(rec, audio):
s = rec.create_stream()
s.accept_waveform(16000, audio)
rec.decode_stream(s)
return s.result.text.strip()
async def partial(self, audio, utt):
self.busy = True
try:
text = await self.loop.run_in_executor(partial_pool, decode_partial, rec, audio)
if text and self.speech and utt == self.utt: # 確定後に遅れて届いた結果は捨てる
await self.send({"type": "partial", "text": text})
finally:
self.busy = False
self.last_partial = time.monotonic()
- 日本語の ReazonSpeech モデルは非ストリーミング型です。そこで、発話の頭から今までの音声を 0.7 秒ごとに丸ごと認識し直す「疑似ストリーミング」にしています。CPU でも音声1秒を約 0.02 秒で処理できるので、このやり方でも間に合います。
-
busyフラグで、前の認識が終わるまで次を始めないようにしています。たまった分を全部処理しようとすると、遅れがどんどん大きくなります。 - 途中結果はあくまで「動いていることが分かる」ためのものです。ReazonSpeech は2文を続けて話すと前半を落とすことがあったので(実測で CER 31.1%)、保存するのは Whisper の確定文だけにしています。
- モデルがないときは
rec = Noneにして、確定だけで動くようにしておくと、導入のハードルが下がります。
手順7: Docker で動かす
Python の環境をそろえなくても動くように、Docker にまとめます。
# Dockerfile(再学習用の部分は省略)
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
COPY requirements.txt ./
RUN pip install -r requirements.txt
COPY server.py live.py glossary.txt ./
COPY static ./static
COPY scripts ./scripts
EXPOSE 8000
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
services:
asr:
build: .
ports:
- "8000:8000"
environment:
ASR_MODEL: ${ASR_MODEL:-small} # GPU なしなら small が現実的
ASR_DEVICE: ${ASR_DEVICE:-cpu}
ASR_COMPUTE: ${ASR_COMPUTE:-int8}
volumes:
- ./data:/app/data # 録音・DB・途中表示用モデル
- ./glossary.txt:/app/glossary.txt # 用語ヒント(再ビルド不要)
- ./scripts:/app/scripts # 台本・計測スクリプト(再ビルド不要)
- hf-cache:/root/.cache/huggingface # Whisper のモデルを再起動で消さない
restart: unless-stopped
volumes:
hf-cache:
起動します。
python scripts/fetch_sherpa.py # 途中表示用モデル(任意)
docker compose up --build
http://localhost:8000/live.html を開いて「通話を始める」を押し、話してみてください。初回は Whisper のモデルのダウンロードがあるので、ログに Uvicorn running が出るまで待ちます。ログに [live] sherpa-onnx model: data/sherpa-ja と出ていれば、途中表示も動いています。
- Whisper のモデルは変数で切り替えられます(PowerShell なら
$env:ASR_MODEL="large-v3-turbo"; docker compose up)。 -
glossary.txtとscripts/はマウントしているので、書き換えてもイメージを作り直す必要はありません。static/はイメージに入れているので、画面を直したときは--buildが必要です。
手順8: 用語ヒントを入れて CER を測る
用語ヒント
Whisper の initial_prompt に専門用語を並べて渡すと、その言葉が出やすくなります。再学習はいりません。glossary.txt に1行1語で書き、認識のたびに読み直すので、再起動もいりません。
# server.py(抜粋)
def glossary_prompt() -> str | None:
p = Path("glossary.txt")
if not p.exists():
return None
words = [l.strip() for l in p.read_text(encoding="utf-8").splitlines()
if l.strip() and not l.startswith("#")]
return "、".join(words) if words else None
# glossary.txt
Docker
Kubernetes
Pod
CER を測る
効いたかどうかは、数字で確かめます。台本を読み上げ、認識結果と台本の文字のずれ(編集距離)を、台本の文字数で割ったものが CER です。
-
/live.htmlで「通話を始める」を押し、scripts/daihon.txtを読み上げて「通話を終える」を押す - 直近の通話を、モデルごとに認識し直して比べる
docker compose exec asr python scripts/cer.py --ref scripts/daihon.txt \
--models small,medium,large-v3-turbo --sherpa
# 用語ヒントありで測るときは --glossary scripts/glossary_dev.txt を足す
- 区切りごとの文をつなげて、台本全体と比べます。無音の区切り方が台本の行とずれても問題ありません。
- 比べる前に NFKC で全角半角をそろえ、句読点と空白を除きます。漢数字と算用数字(「三件」と「3件」)の違いは誤りに数えるので、Whisper は実際より悪めに出ます。
- 保存済みの音声を使うので、同じ発話でモデルを公平に比べられます。
筆者の読み上げ(71秒・412文字、CPU のみ)では、次の結果になりました。
| モデル | CER(ヒントなし) | CER(ヒントあり) | 処理時間(音声1秒あたり) |
|---|---|---|---|
Whisper small
|
11.9% | 8.5% | 約0.3秒 |
Whisper medium
|
10.2% | 6.1% | 約1.0秒 |
Whisper large-v3-turbo
|
7.3% | 5.3% | 約1.3秒 |
| ReazonSpeech | 31.1% | 約0.02秒 |
数字の表記をそろえると、ヒントありの large-v3-turbo は 1.9% でした。CPU で話した時間より速く処理できたのは small だけなので、CPU なら small、GPU があれば large-v3-turbo がおすすめです。
ハマったところ
| 起きたこと | 原因 | 対処 |
|---|---|---|
| faster-whisper 1.2.1 が動かない | PyAV 19 と faster-whisper 1.2.1 の組み合わせ |
requirements.txt で av<19 に固定 |
| 別の PC から開くとマイクが使えない |
getUserMedia は localhost か HTTPS でしか動かない |
証明書を用意するか、Cloudflare Tunnel などで HTTPS にする |
Windows の Git Bash で tar -xjf C:/... が失敗する |
C: を接続先のホスト名と解釈する |
--force-local を付ける。リポジトリでは Python の tarfile で展開して回避 |
| 台本を書き換えたのに、古い台本で計測された |
COPY した時点のファイルがイメージに残っていた |
scripts/ をボリュームでマウントする |
| Whisper の CER が思ったより悪い | 「三件」と「3件」のような表記の違いも誤りに数えていた | 表記をそろえた値も並べて見る(誤りの約半が表記だった) |
| 無料で公開しようとしたら Hugging Face Spaces で Docker が使えない | 2026年7月ごろから、Docker Space の作成に PRO プランが必要になった | 手元で動かしてもらう形にし、コードを GitHub で公開 |
もう1つ、このアプリにはログイン機能がありません。開ける人には全員の録音が見えて削除もできるので、組織内で共有するときは、Cloudflare Access などの認証を前に置いてください。
まとめ
FastAPI、faster-whisper、sherpa-onnx の3つを組み合わせると、外部 API なしで「話している間に文字が出る」日本語の文字起こしを作れました。コードは Python と HTML を合わせて数百行です。
作るときのポイントは次の3つです。
- 途中結果は軽いモデル、確定は Whisper と役割を分ける
- ブラウザからは 16kHz の PCM を送り、サーバーで無音を見て区切る
- 精度は台本と CER で測り、用語ヒントの効果も数字で確かめる
完成版は yk-planning/ja-asr-demo にあります(MIT ライセンス)。README には、訂正したデータで Whisper を LoRA 再学習する手順も載せています。
社内で完結する AI の導入については、YK Planning でも紹介しています。