Closed7

「VOSK」で音声ベクトルによる話者識別を試す

Vosk
ASR
kun432kun432

とりあえずわかっていること

とりあえずサンプルコードうごかしてみる。

kun432kun432

今回はローカルのMacで。

uvで仮想環境作成

mkdir vosk-si-work && cd vosk-si-work
uv venv -p 3.12 --seed

uv pip install vosk
出力
 + vosk==0.3.44

モデルをダウンロード。とりあえず必要なのは話者識別用モデルだけだけど、一応日本語モデルもダウンロードしておく。

mkdir -p models spk_model

# Speaker
wget https://alphacephei.com/vosk/models/vosk-model-spk-0.4.zip
unzip vosk-model-spk-0.4.zip -d spk_model

# small
wget https://alphacephei.com/vosk/models/vosk-model-small-ja-0.22.zip
unzip vosk-model-small-ja-0.22.zip -d models

# large
wget https://alphacephei.com/vosk/models/vosk-model-ja-0.22.zip
unzip vosk-model-ja-0.22.zip -d models

サンプルコードを参考に、まず音声ベクトルだけを取得してみる。サンプルとなる音声は過去に自分が主催した勉強会から冒頭15秒程度の音声を抽出した。なお、発話者は自分のみ。

https://www.youtube.com/watch?v=Yl2kR6zLRY8

get_embeddings.py
import os
import sys
import wave
import json

from vosk import Model, KaldiRecognizer, SpkModel

MODEL_PATH = "models/vosk-model-small-ja-0.22"
SPK_MODEL_PATH = "spk_model/vosk-model-spk-0.4"

if not os.path.exists(MODEL_PATH):
    print(
        "モデルが見つかりません。"
        "モデルを https://alphacephei.com/vosk/models からダウンロードして、"
        "現在のフォルダ内に {MODEL_PATH} として展開してください。"
    )
    sys.exit(1)

if not os.path.exists(SPK_MODEL_PATH):
    print(
        "話者識別モデルが見つかりません。"
        "話者識別モデルを https://alphacephei.com/vosk/models からダウンロードして、"
        "現在のフォルダ内に {SPK_MODEL_PATH} として展開してください。"
    )
    sys.exit(1)

# WAVファイルを開く
wf = wave.open(sys.argv[1], "rb")
if wf.getnchannels() != 1 or wf.getsampwidth() != 2 or wf.getcomptype() != "NONE":
    print("音声ファイルは WAV 形式のモノラル PCM である必要があります。")
    sys.exit(1)

# モデルの読み込み
model = Model(MODEL_PATH)
spk_model = SpkModel(SPK_MODEL_PATH)

# Recognizerの作成と話者モデルの設定
rec = KaldiRecognizer(model, wf.getframerate())
rec.SetSpkModel(spk_model)

# 音声処理ループ（中間認識結果の出力なし）
while True:
    data = wf.readframes(4000)
    if len(data) == 0:
        break
    rec.AcceptWaveform(data)

# 最終結果を表示
# - 最終結果内の spk が音声ベクトル
# - 音声が短すぎると話者ベクトルが抽出されない可能性がある（最低1.5秒以上、4秒以上が推奨らしい）
res = json.loads(rec.FinalResult())
print(json.dumps(res, indent=2, ensure_ascii=False))
出力
{
  "spk": [
    -0.713196,
    0.938591,
    1.104772,
    -0.286729,
    -0.15291,
    (snip)
    0.939724,
    -1.056973,
    2.525203,
    0.065403,
    -0.667073
  ],
  "spk_frames": 669,
  "text": "はい じゃあ 始め ます ちょっと まだ 来 られ て ない 方 も いらっしゃる ん です けど なんて で 日 外れ まし に"
}

返されるのは普通のリストになっている。

kun432kun432

上で話者のベクトルが取得できたので、あとは比較するだけ。別に用意しておいた音声ファイルを使って比較してみる。

import os
import sys
import wave
import json

import numpy as np
from vosk import Model, KaldiRecognizer, SpkModel

MODEL_PATH = "models/vosk-model-small-ja-0.22"
SPK_MODEL_PATH = "spk_model/vosk-model-spk-0.4"

if len(sys.argv) < 2:
    print(f"Usage: {sys.argv[0]} input.wav")
    sys.exit(1)

audio_path = sys.argv[1]

if not os.path.exists(MODEL_PATH):
    print(
        "モデルが見つかりません。"
        "モデルを https://alphacephei.com/vosk/models からダウンロードして、"
        f"現在のフォルダ内に {MODEL_PATH} として展開してください。"
    )
    sys.exit(1)

if not os.path.exists(SPK_MODEL_PATH):
    print(
        "話者識別モデルが見つかりません。"
        "話者識別モデルを https://alphacephei.com/vosk/models からダウンロードして、"
        f"現在のフォルダ内に {SPK_MODEL_PATH} として展開してください。"
    )
    sys.exit(1)

# 同じ話者の別の音声から取得した音声ベクトル
spk_sig = [
    -1.428289, 0.542421, 1.035572, -0.432543, -0.343544, -0.575979, 1.532389,
    0.377959, 1.797391, 0.403849, 1.230559, -0.335164, -0.097044, 0.063457,
    0.384248, 2.778059, 0.480696, 0.516359, -0.384573, -0.490258, 0.465333,
    (snip)
    1.196434, -1.508616, 0.806304, 0.660372, -0.053518, -1.204596, -0.237079,
    1.682485, -0.599265, -0.594982, -0.725676, 0.13058, -0.356516, -0.041056,
    -0.512262, -0.353594
]

def cosine_dist(x, y):
    """コサイン距離（0 に近いほど似ている、2 に近いほど逆向き）"""
    nx = np.asarray(x, dtype=np.float32)
    ny = np.asarray(y, dtype=np.float32)
    return 1.0 - np.dot(nx, ny) / (np.linalg.norm(nx) * np.linalg.norm(ny))

wf = wave.open(audio_path, "rb")
if wf.getnchannels() != 1 or wf.getsampwidth() != 2 or wf.getcomptype() != "NONE":
    print("音声ファイルは WAV 形式のモノラル PCM である必要があります。")
    sys.exit(1)

model = Model(MODEL_PATH)
spk_model = SpkModel(SPK_MODEL_PATH)

rec = KaldiRecognizer(model, wf.getframerate())
rec.SetSpkModel(spk_model)

while True:
    data = wf.readframes(4000)
    if len(data) == 0:
        break
    rec.AcceptWaveform(data)

res = json.loads(rec.FinalResult())

print(json.dumps(res, indent=2, ensure_ascii=False))

if "spk" in res:
    spk_vec = res["spk"]
    # spk が含まれていればコサイン距離を計算
    dist = cosine_dist(spk_sig, spk_vec)
    print(f"登録された話者ベクトルとのコサイン距離: {dist:.4f}")
else:
    print("この音声からは spk ベクトルが抽出されませんでした。")
    print("音声が短すぎる可能性があります（最低 1.5 秒以上、4 秒以上推奨）。")
出力
(snip)
登録された話者ベクトルとのコサイン距離: 0.4601

異なる話者の音声ベクトルで試した場合。

出力
登録された話者ベクトルとのコサイン距離: 0.7981
kun432kun432

VOSKはサーバもある。以前も試した。

https://zenn.dev/link/comments/086a7843883008

サーバでも、話者識別モデルのディレクトリをマウントして、VOSK_SPK_MODEL_PATHで指定すれば良さそう。

docker run -d \
    -p 2700:2700 \
    -e VOSK_SPK_MODEL_PATH=/opt/spk-model \
    -v ./spk_model/vosk-model-spk-0.4:/opt/spk-model \
    alphacep/kaldi-ja:latest

WebSocketで接続してマイクからの音声を文字起こしするサンプル。

uv pip install websocket-client sounddevice

import sys
import traceback
import json
import queue
import sounddevice as sd
from websocket import create_connection

# WebSocketサーバーのURL
server_url = "ws://localhost:2700"

# オーディオ設定
samplerate = 16000  # サンプリングレート
blocksize = int(samplerate * 0.2)  # 0.2秒のオーディオブロック

# キューを設定
q = queue.Queue()

# オーディオコールバック関数
def audio_callback(indata, frames, time, status):
    """マイクからのオーディオブロックごとに呼び出される関数"""
    if status:
        print(status, file=sys.stderr)
    q.put(bytes(indata))

try:
    # WebSocket接続を作成
    print(f"サーバー {server_url} に接続しています...")
    ws = create_connection(server_url)
    print("接続しました")
    
    # サンプルレート設定をサーバーに送信
    ws.send(json.dumps({"config": {"sample_rate": samplerate}}))
    
    # マイク入力ストリームを開始
    with sd.RawInputStream(samplerate=samplerate, blocksize=blocksize, 
                          dtype='int16', channels=1, 
                          callback=audio_callback):
        
        print("#" * 80)
        print("マイク入力を開始しました。話しかけてください。")
        print("終了するには Ctrl+C を押してください。")
        print("#" * 80)
        
        # メインループ - マイクからのデータを読み取ってWebSocketに送信
        while True:
            # キューからデータを取得
            data = q.get()
            
            # バイナリデータとしてWebSocketに送信
            ws.send_binary(data)
            
            # サーバーからの応答を受信して表示
            result = ws.recv()
            
            # 空の部分結果は表示しない
            parsed = json.loads(result)
            if "partial" in parsed and parsed["partial"] == "":
                continue
                
            print(result)
            
except KeyboardInterrupt:
    print("\n終了しました")
    # 終了通知を送信
    try:
        ws.send(json.dumps({"eof": 1}))
        print(ws.recv())
        ws.close()
    except:
        pass
except Exception as err:
    print(''.join(traceback.format_exception(type(err), err, err.__traceback__)))
    try:
        ws.close()
    except:
        pass

実行

uv run ws_client.py
出力
サーバー ws://localhost:2700 に接続しています...
接続しました
################################################################################
マイク入力を開始しました。話しかけてください。
終了するには Ctrl+C を押してください。
################################################################################
{
  "partial" : "小 小"
}
{
  "partial" : "おはよう"
}
{
  "partial" : "おはよう ござい"
}
{
  "partial" : "おはよう ござい"
}
{
  "partial" : "おはよう ござい ます"
}
{
  "partial" : "おはよう ござい ます"
}
{
  "partial" : "おはよう ござい ます"
}
{
  "partial" : "おはよう ござい ます"
}
{
  "result" : [{
      "conf" : 0.974476,
      "end" : 1.770000,
      "start" : 1.260000,
      "word" : "おはよう"
    }, {
      "conf" : 1.000000,
      "end" : 2.190000,
      "start" : 1.770000,
      "word" : "ござい"
    }, {
      "conf" : 1.000000,
      "end" : 2.640000,
      "start" : 2.190000,
      "word" : "ます"
    }],
  "spk" : [-0.733220, -0.125512, 1.841761, 0.152253, -0.489290, (snip) 1.467239, -0.429734, -0.539233, 0.182995, -0.150142],
  "spk_frames" : 138,
  "text" : "おはよう ござい ます"
}
{
  "partial" : "明日"
}
{
  "partial" : "明日"
}
{
  "partial" : "明日 の"
}
{
  "partial" : "明日 の 天気"
}
{
  "partial" : "明日 の 天気 に"
}
{
  "partial" : "明日 の 天気 に つい"
}
{
  "partial" : "明日 の 天気 に つい て 小"
}
{
  "partial" : "明日 の 天気 に つい て 小"
}
{
  "partial" : "明日 の 天気 に つい て 教え て"
}
{
  "partial" : "明日 の 天気 に つい て 教え て"
}
{
  "partial" : "明日 の 天気 に つい て 教え て ください"
}
{
  "partial" : "明日 の 天気 に つい て 教え て ください"
}
{
  "partial" : "明日 の 天気 に つい て 教え て ください"
}
{
  "partial" : "明日 の 天気 に つい て 教え て ください"
}
{
  "result" : [{
      "conf" : 1.000000,
      "end" : 6.510000,
      "start" : 6.090000,
      "word" : "明日"
    }, {
      "conf" : 1.000000,
      "end" : 6.630000,
      "start" : 6.510000,
      "word" : "の"
    }, {
      "conf" : 1.000000,
      "end" : 7.020000,
      "start" : 6.660000,
      "word" : "天気"
    }, {
      "conf" : 1.000000,
      "end" : 7.140000,
      "start" : 7.020000,
      "word" : "に"
    }, {
      "conf" : 1.000000,
      "end" : 7.410000,
      "start" : 7.140000,
      "word" : "つい"
    }, {
      "conf" : 1.000000,
      "end" : 7.530000,
      "start" : 7.410000,
      "word" : "て"
    }, {
      "conf" : 1.000000,
      "end" : 7.860000,
      "start" : 7.530000,
      "word" : "教え"
    }, {
      "conf" : 1.000000,
      "end" : 7.950000,
      "start" : 7.860000,
      "word" : "て"
    }, {
      "conf" : 0.801385,
      "end" : 8.490000,
      "start" : 7.950000,
      "word" : "ください"
    }],
  "spk" : [-1.054111, -0.299124, 1.137017, -0.700562, -0.332513, (snip) 0.998105, -1.282689, 0.343167, 0.536312, 1.157974],
  "spk_frames" : 234,
  "text" : "明日 の 天気 に つい て 教え て ください"
}

サーバからのレスポンスに音声ベクトルが付与されて返ってくるので、事前に登録しておいたベクトルと比較するようにすれば、話者を識別できそう。

kun432kun432

話者識別の場合

  • 事前に話者の音声ベクトルを生成しておく
  • 返ってきた音声ベクトルと類似度比較

すれば良さそう。多分音声からのベクトル生成が多分一番処理コストかかるところで、比較はそれほどかからないはずなので、ベクトルが返ってくるのはいいな。

話者ダイアライぜーションの場合は

  • 最初に返ってきたベクトルを話者1でラベル
  • 次のベクトルと比較
    • 一定の類似度があれば同じ話者とみなす
    • 類似度が低ければ話者Nとしてラベル
  • 以後繰り返し

みたいな感じか。最初の発話の音声ベクトルが比較用に十分な精度があるかどうかがわからないので、多少調整は必要な気がする。

いずれにしてもしきい値の設定・調整は必要になりそう。

このスクラップは18日前にクローズされました