😑

【自分用】「AI後輩ちゃん」の個人的つまづきポイントまとめ

2024/05/09に公開

はじめに

自分用に残す記事なので,自分がわかればいいくらいの粒度で書いています。
(それでも良いというかたは見ていただけると嬉しいです)

https://zenn.dev/asap/articles/5b1b7553fcaa76

こちらの記事において,自分の欲望のままに可愛い後輩の女の子との音声対話システムを作成したが,作成する際につまづいたポイントや,中身の部分について改めて確認した内容を忘備録としてここに残します。

ご興味があれば,元の記事もご覧いただけますと嬉しいです。

録音データについて

コード(録音部分)
main.py
def speech2audio(fs=16000, silence_threshold=0.5, min_duration=0.1, amplitude_threshold=0.025):
    record_Flag = False

    non_recorded_data = []
    recorded_audio = []
    silent_time = 0
    input_time = 0
    start_threshold = 0.3
    all_time = 0
    
    with sd.InputStream(samplerate=fs, channels=1) as stream:
        while True:
            data, overflowed = stream.read(int(fs * min_duration))
            all_time += 1
            if all_time == 10:
                print("stand by ready OK")
            elif all_time >=10:
                if np.max(np.abs(data) > amplitude_threshold) and not record_Flag:
                    input_time += min_duration
                    if input_time >= start_threshold:
                        record_Flag = True
                        print("recording...")
                        recorded_audio=non_recorded_data[int(-1*start_threshold*10)-2:]  

                else:
                    input_time = 0

                if overflowed:
                    print("Overflow occurred. Some samples might have been lost.")
                if record_Flag:
                    recorded_audio.append(data)

                else:
                    non_recorded_data.append(data)

                if np.all(np.abs(data) < amplitude_threshold):
                    silent_time += min_duration
                    if (silent_time >= silence_threshold) and record_Flag:
                        record_Flag = False
                        break
                else:
                    silent_time = 0

    audio_data = np.concatenate(recorded_audio, axis=0)

    return audio_data

fs=16000→サンプリング周波数は16000Hzということ(1秒間に16000個の振幅データを得る)

with sd.InputStream(samplerate=fs, channels=1) as stream:
        while True:
            data, overflowed = stream.read(int(fs * min_duration))

streamとして,fs * min_duration分の音声データをdataに格納する
min_durationは0.1秒に設定しているので,dataの型は(1600,1)となる。

録音後のデータと文字起こしの際のデータの違いについて

コード(録音部分)
main.py
def speech2audio(fs=16000, silence_threshold=0.5, min_duration=0.1, amplitude_threshold=0.025):
    record_Flag = False

    non_recorded_data = []
    recorded_audio = []
    silent_time = 0
    input_time = 0
    start_threshold = 0.3
    all_time = 0
    
    with sd.InputStream(samplerate=fs, channels=1) as stream:
        while True:
            data, overflowed = stream.read(int(fs * min_duration))
            all_time += 1
            if all_time == 10:
                print("stand by ready OK")
            elif all_time >=10:
                if np.max(np.abs(data) > amplitude_threshold) and not record_Flag:
                    input_time += min_duration
                    if input_time >= start_threshold:
                        record_Flag = True
                        print("recording...")
                        recorded_audio=non_recorded_data[int(-1*start_threshold*10)-2:]  

                else:
                    input_time = 0

                if overflowed:
                    print("Overflow occurred. Some samples might have been lost.")
                if record_Flag:
                    recorded_audio.append(data)

                else:
                    non_recorded_data.append(data)

                if np.all(np.abs(data) < amplitude_threshold):
                    silent_time += min_duration
                    if (silent_time >= silence_threshold) and record_Flag:
                        record_Flag = False
                        break
                else:
                    silent_time = 0

    audio_data = np.concatenate(recorded_audio, axis=0)

    return audio_data
コード(文字起こし部分)
main.py
def audio2text(data, model):
    result = ""
    data = data.flatten().astype(np.float32)

    segments, _ = model.transcribe(data, beam_size=BEAM_SIZE)
    for segment in segments:
        result += segment.text

    return result

上に記載した,「録音部分のコード」と「文字起こし部分のコード」はつながっており,録音部分のコードの出力(audio_data)がそのまま文字起こし部分のコードの入力(data)に使われています。

ここで詰まったのは,下記コードの部分です。

data = data.flatten().astype(np.float32)

ネットで調べるとfaster-whisperなどを利用する際は,録音した内容を一旦「output.wav」などの音声ファイルとして保存して,それを

model.transcribe("output.wav", beam_size=BEAM_SIZE)

の形で読み取る形でサンプルコードが書かれていることが多く,直接録音したデータを入力することができるのかわかりませんでした。

そこで,faster-whisperの実装元githubの該当箇所を見に行くと
https://github.com/SYSTRAN/faster-whisper/blob/2f6913efc85306fc4f900da6c67f9a06a7d54a3d/faster_whisper/transcribe.py#L200C9-L200C19

def transcribe(
        self,
        audio: Union[str, BinaryIO, np.ndarray],

と,audioに対して,str(つまり音声ファイルのpath)だけでなくBinaryIO, np.ndarrayも引数に設定可能とわかったため,最初はそのままdataをfast-whisperに入れました。

しかしながら,下記のエラーが出てしまいました。

(工事中)

このエラーはデータがメモリに割り当て可能な量をオーバーしたというエラーです。
なぜかわからなかったですが,データのshapeに問題がありました。

audio_dataのshapeを確認すると(xxxxx,1)の形でした(xxxxx=秒数*16000hz)
一方で,whisperやfast-whisperが対応する入力の形は(xxxxx,)の形だということに気づき,

data = data.flatten().astype(np.float32)

こちらを実行したらうまくいきました。
(このコードの.astype(np.float32)の部分は,今回の後輩ちゃんに限っていうと,録音する時点からfloat32の型でデータを取得しているので必要ないですが,sound deviceの設定によっては,他の型で録音することも可能だそうなので,安全のため付けています)

beam_sizeとは

コード(文字起こし部分)
main.py
def audio2text(data, model):
    result = ""
    data = data.flatten().astype(np.float32)

    segments, _ = model.transcribe(data, beam_size=BEAM_SIZE)
    for segment in segments:
        result += segment.text

    return result

whisperの中身の学習をする。
https://medium.com/axinc/whisper-日本語を含む99言語を認識できる音声認識モデル-b6e578f55c87

音声ファイルは-1〜1のレンジにスケーリングされた16kHzのPCM形式で扱い、80channelのMelSpectrogramで周波数変換します。MelSpectrogram変換のウィンドウサイズは25msで、strideは10msです。

記載の通りwhisperに入れるdataはサンプリング周波数16000Hzかつ,振幅は-1から1のデータであり,かつ振幅スペクトグラムに対して周波数の軸を人間の聴覚特性に合う周波数尺度であるメル尺度に変換して得られるメルスペクトログラムを利用する。

https://deepage.net/machine_learning/2017/07/06/beam-search.html?source=post_page-----b6e578f55c87--------------------------------
ビームサーチはコチラをさらに参照できる。

上記からの理解は下記である。
音声からテキストを一文字ずつ生成するwhisperを考えた場合(厳密にはtokenごとに出力することになるため間違いだが,簡単のためにこう定義する),
whisperはある文字の確率分布に則って,サンプリングされることになる。その際に何も考えずにサンプリングすると一定確率で低い確率の不適切なものがサンプリングされてしまったり,その場では高い確率だったとしても実際には不適切な文字がサンプリングされて選ばれてしまう。
したがって,理想的には一文字ずつサンプリングするわけではなく,それぞれの確率分布に対して,そのさらに次の文字をサンプリングする同時確率も確認して,全ての生成される文章において最も同時確率が高いものが最も確からしいサンプリングであると判断できる。
ただし全ての確率分布の候補に対して,今後発生しうる全ての同時確率を計算することを不可能なため,ここに制限をかける。
制限をかける方法としては,ある文字の確率分布に対して,上位○個の候補だけを残して,それ以外を切り捨てて探索する方法である。二文字目になった際も,二文字の同時確率が最も高い○個の候補だけを残して次のサンプリングを行う。

上記の制限をかけて,効率的に探索を行う手法がビームサーチであり,beam_sizeは上位○個の候補を残して探索するということである。

一方で,GreedySearchというものもある。これはbeam_size=1の場合を指し,常に最も確率の高いtokenをサンプリングする手法である。

おまけ

コード(録音部分)

Discussion