« ^ »

今日やった事 - 20240831

所要時間: 約 8分

最近では、1つの話題に集中した記事をまとめることを、ほとんどしなくなった。もし記事にまとめるのであれば、AIサービスを使って書かせた方が品質が高いからだ。情報の整理整頓も同じく、機械の方が得意だ。僕は整理整頓がそもそも苦手だからありがたい。今は整理した情報よりも、僕が受けた衝撃や混乱を、そのまま書いておきたいと考えている。自分のダメな部分を、機械に「そのままでいいんだよ」と認められたような気分になる。実際はどうなんだろう、わからないけれど。

自分自身は社会から孤立しつつあるように思う。別に格好付けているんじゃなくて、人からちょっと避けられたり、距離を取られたりといったような。なんだか悲しいな。まだ生きていたいなぁと思う反面、もういいかなと思う事もある。

少し嬉しい事もあった。このページを中学生に見せたら「すご」って言われたの、嬉しかった。僕も凄いと思う。何が凄いかって、これだけのコード片でいろんな事ができてしまう事が凄い。夢が広がる。

僕個人が凄いとは1ミリも思わない。Python、ライブラリ、モデルや、その他さまざまな資産によって、いろんな事ができてしまう、それを自由に使えるという文化が凄いと思う。小学生や中学生には、ぜひこの広がりを感じて欲しいなと思った。

ここに書いた文章も、所々間違えていて微妙に動かないコード片も、どちらも僕が好きで書いているただの砂場の落書きだ。雨が降れば消える程の価値しかない。それでも自分の事や、自分を取り巻く周りの事を考えるために一役買っている。だから一歩ずつ自分の歩幅で足跡を残していきたい。

さてと、気をとりなおして昨日の続きに取り組む。

音を拾う

人は耳で音を聞く。最近のパソコンにはマイクが付いているから、マイクから入った音をデータとして保存できる。ここでは、パソコンで音を拾う方法を試す。

マイクから音を拾う

音声データはsounddevice経由で取得する。

import sounddevice
import numpy as np

SAMPLING_RATE = 16000
WINDOW_SIZE_SAMPLES = 512

def callback_sound(indata: np.ndarray[np.ndarray[np.int16]],
                   outdata, frames, time, status):
    pass

with sounddevice.Stream(samplerate=SAMPLING_RATE, dtype="int16",
                        channels=1, blocksize=WINDOW_SIZE_SAMPLES,
                        callback=callback_sound):
    while True:
        time.sleep(0.01)

元々はFFmpegを使って標準入出力で受け渡しをしていたが、データを処理しずらいため、この実装に変更した。

ファイルから音を拾う

音声データの取得はsounddeviceで取得するが、検証時にはファイルから読み出した方が楽な事がある。そのためファイルから入力と同じ形式のデータを取得できるようにした。

import librosa
import numpy as np

original_wav_data, original_sampling_rate = librosa.load("hello.wav", sr=None)
sampling_rate = 16000
wav_data = librosa.resample(
    original_wav_data,
    orig_sr=original_sampling_rate,
    target_sr=sampling_rate)
事前準備

もしかすると、型は少し異なるかもしれない。

参考

雑音を消す

音には聞きたい音と、邪魔な音がある。音楽を聞いている時、楽器の音は聞きたいけれど、人の話し声は邪魔だ。誰かと話している時、人の声は聞きたいけれど、車の音は邪魔だ。この邪魔な音は雑音と呼ばれる。ここではこの雑音を消す方法を試す。

from pydub import AudioSegment
from pydub.effects import low_pass_filter, high_pass_filter

audio:AudioSegment = AudioSegment.from_wav("example.wav")
audio = audio.set_channels(1)  # 音声をモノラルに変換
audio = low_pass_filter(audio, 3400)
audio = high_pass_filter(audio, 300)

audio.export("output.wav")
ファイルの雑音を消す

ここではローパスフィルターとハイパスフィルターを使って、ノイズを除去している。言葉を認識する時、人の声以外の音は必要ない。人の声の周波数は300Hzから3400Hzの範囲と言われるため、その周波数の外側にある音を強制的に消す。

音のデータはsounddeviceで取得するため、そのデータから変換してノイズを除去してみる。

import librosa
import numpy as np
import sounddevice as sd

from pydub import AudioSegment
from pydub.effects import low_pass_filter, high_pass_filter

# wavファイルを読み込む
original_wav_data, original_sampling_rate = librosa.load("sound-ohayougozaimasu.wav", sr=None)

# サンプリングレートを16000Hzに変換
sampling_rate = 16000
wav_data = librosa.resample(original_wav_data, orig_sr=original_sampling_rate, target_sr=sampling_rate)

# numpy array を AudioSegment に変換
audio_segment = AudioSegment(
    np.array(wav_data * 32767, dtype=np.int16).tobytes(),
    frame_rate=sampling_rate,
    sample_width=2,  # 16-bit audio
    channels=1
)

# ローパスフィルターを3400Hzで適用
audio_segment = low_pass_filter(audio_segment, cutoff=3400)

# ハイパスフィルターを300Hzで適用
audio_segment = high_pass_filter(audio_segment, cutoff=300)

# AudioSegment から numpy array に変換
y = np.array(audio_segment.get_array_of_samples())

# 再生
sd.play(y, samplerate=sampling_rate)
sd.wait()  # 再生終了を待つ
sounddeviceで取得したデータを元に雑音を消す

誰かが話しているかを判断する

import numpy as np
import torch

vad_model, utils = torch.hub.load("snakers4/silero-vad", "silero_vad")

(get_speech_timestamps,
 save_audio,
 read_audio,
 VADIterator,
 collect_chunks) = utils

vad_model.reset_states()

sound_data = torch.Tensor(wav_data)

window_size = 512  # Sampling rateが16000の場合512
chunk = sound_data[0:window_size]  # 検出したい箇所を切り出す

ret = vad_model(chunk, sampling_rate)
speech_prob = ret.item()  # 閾値よりも大きな数字であれば発話と見做す

if speech_prob > 0.5:  # 発話検出の閾値。ここは自分で調整 (例: 0.5)
    print("発話中")
else:
    print("発話なし")

誰かと話をしている時、その時間ずっと声が聞こえているわけではない。息継ぎをしている時や、間(ま)を取って、話を区切っている時がある。ここでは、その瞬間が話をしているかどうかを判断する方法を試す。

声を文章にする

声は音だけれど、コンピュータで処理するのであれば、音のデータよりも文字のデータの方が処理しやすい。そこで、ここでは声のデータを文字にする。

from reazonspeech.nemo.asr.audio import audio_from_tensor, audio_from_numpy
from nemo.collections.asr.models import EncDecRNNTBPEModel
from reazonspeech.nemo.asr.interface import AudioData
from reazonspeech.nemo.asr.transcribe import transcribe

rez_model = EncDecRNNTBPEModel.from_pretrained(
    'reazon-research/reazonspeech-nemo-v2',
    map_location="cpu")

audio: AudioData = audio_from_numpy(wav_data, sampling_rate)
ret = transcribe(rez_model, audio)
text = "".join(sg.text for sg in ret.segments)
print(text)

ここまでの処理で止めると文字起こしとなる。一人での語りのようなスタイルであれば、単純な文字起こしとして有用かもしれない。そうではなく、2人以上の対話形式の場合であれば、どの声が誰の声なのかを判断する必要がある。音の成分と、文字起こしした文章の文脈から、どの文章が誰の文章かを機械的に判断する事はできる。それについては、ここでは取り上げない。

質問に答える

質問に対する、よくある続きの対話を作る。

from mlx_lm import load, generate

prompt = "hello"

model, tokenizer = load("mlx-community/Llama-3-Swallow-8B-Instruct-v0.1-8bit")
response = generate(model, tokenizer, prompt=prompt, verbose=True)
print(response)

プロンプトを工夫して対話になるようにする。speech-to-speech-japaneseでは、初期プロンプトとして以下のような文章が設定されていた。

あなたは日本語に堪能な友人です。英語を使ってはいけません。全て日本語で回答します. You must answer in Japanese

文章を音にする

文字である文章を、音のデータにする。

import sounddevice
import numpy as np
import melo.api

sentence = "こんにちわ"

model = melo.api.TTS(language="JP", device="mps")
print(model.hps.data.spk2id)

audio_chunk = model.tts_to_file(sentence, model.hps.data.spk2id["JP"], quiet=True)
audio_chunk2 = librosa.resample(audio_chunk, orig_sr=44100, target_sr=16000)
audio_chunk3 = (audio_chunk2 * 32768).astype(np.int16)

sounddevice.play(audio_chunk3, 12100)

音を出す

音のデータを実際にスピーカーやイヤホンから鳴らす。

import sounddevice
import numpy as np

samplerate = 44100  # サンプリングレートの設定
duration = 2.0  # 秒数
frequency = 440.0  # 再生する音の周波数

# サイン波を生成
t = np.linspace(0, duration, int(samplerate * duration), endpoint=False)
waveform = 0.5 * np.sin(2 * np.pi * frequency * t)

def callback(outdata, frames, time, status):
    # デフォルトでは np.int16 形式でデータを渡す
    outdata[:] = (waveform[:frames] * 32767).astype(np.int16).tostring()

# RawOutputStreamを使用してデータをストリーミング
with sounddevice.RawOutputStream(samplerate=samplerate, channels=1, dtype='int16', callback=callback):
    # duration秒間のデータをストリーミング
    sounddevice.sleep(int(duration * 1000))

繋げる

ここまでの作業を繋げると音声対話型AIが作れる。これはspeech-to-speech-japaneseの実装を調べつつ、要素を抜き出してきたものだ。speech-to-speech-japaneseは、システムとしてもっと良く実装されていて、それぞれのデータをQueueを使って橋渡しし、スレッドを使って並列実行している。

僕の環境では、エラーが発生してそれを上手く動かす事ができなかった。だから1つずつ足元を固めるつもりで実装の重要な部分のみを抜き出していたわけだ。そこでここでは、並列実行など考えず、また問題のトラブルシューティングをしやすいよう実装を隠蔽する事なく、逆に露出していく形で、全てを繋げて音声対話AIを実装していく。

作業中…

import librosa
import numpy as np

original_wav_data, original_sampling_rate = librosa.load("sound-ohayougozaimasu2.wav", sr=None)
# original_wav_data, original_sampling_rate = librosa.load("hello.wav", sr=None)
sampling_rate = 16000

wav_data = librosa.resample(
    original_wav_data,
    orig_sr=original_sampling_rate,
    target_sr=sampling_rate)

from pydub import AudioSegment
from pydub.effects import low_pass_filter, high_pass_filter

audio_segment = AudioSegment(
    np.array(wav_data * 32767, dtype=np.int16).tobytes(),
    frame_rate=sampling_rate,
    sample_width=2,  # 16-bit audio
    channels=1
)
audio_segment = low_pass_filter(audio_segment, cutoff=3400)
audio_segment = high_pass_filter(audio_segment, cutoff=300)
y = np.array(audio_segment.get_array_of_samples())

import numpy as np
import torch

vad_model, utils = torch.hub.load("snakers4/silero-vad", "silero_vad")

(get_speech_timestamps,
 save_audio,
 read_audio,
 VADIterator,
 collect_chunks) = utils

vad_model.reset_states()


window_size = 512  # Sampling rateが16000の場合512
speech = None

# for ii in range(0, len(wav_data), window_size):
#     start_index = ii * window_size
#     end_index = start_index + window_size
#     d = wav_data[start_index:end_index]
#     chunk = torch.Tensor(d)
#     ret = vad_model(chunk, sampling_rate)
#     speech_prob = ret.item()
#     print(speech_prob)
#     if speech_prob > 0.01:
#         if speech is None:
#             speech = chunk
#         else:
#             speech += chunk

# sound_data = torch.Tensor(wav_data)

# for ii in range(0, len(sound_data), window_size):
#     start_index = ii * window_size
#     end_index = start_index + window_size

#     chunk = sound_data[start_index:end_index]
#     if len(chunk) != window_size:
#         print("bad chunk")
#         continue
# if not speech:
#     raise ValueError("no speech")

from reazonspeech.nemo.asr.audio import audio_from_tensor, audio_from_numpy
from nemo.collections.asr.models import EncDecRNNTBPEModel
from reazonspeech.nemo.asr.interface import AudioData
from reazonspeech.nemo.asr.transcribe import transcribe

rez_model = EncDecRNNTBPEModel.from_pretrained(
    'reazon-research/reazonspeech-nemo-v2',
    map_location="cpu")

audio_data: AudioData = audio_from_numpy(wav_data, sampling_rate)
# audio_data: AudioData = audio_from_tensor(speech, sampling_rate)
ret = transcribe(rez_model, audio_data)
text = "".join(sg.text for sg in ret.segments)

import mlx_lm
from mlx_lm import load, generate

model, tokenizer = load("mlx-community/Llama-3-Swallow-8B-Instruct-v0.1-8bit")

prompt = [
  {"role": "system", "content": "あなたは日本語に堪能な秘書です。英語を使ってはいけません。全て日本語で回答します。"},
  {"role": "user", "content": text},
  {"role": "system", "content": "この文章に対しての邪悪な回答を、日本語で1つだけ示します。"},
]
applyed_prompt = tokenizer.apply_chat_template(prompt, tokenize=False, add_generate_prompt=True)

ai_response_stream = mlx_lm.stream_generate(
    model=model, tokenizer=tokenizer,
    prompt=applyed_prompt, max_tokens=128)

response = ""
for resp in ai_response_stream:
    print(resp, end="")
    response += resp
print()

# response = generate(model, tokenizer, prompt=text, verbose=True)

# print(response)

import sounddevice
import numpy as np
import melo.api

sentence = response
model = melo.api.TTS(language="JP", device="mps")
audio_chunk = model.tts_to_file(sentence, model.hps.data.spk2id["JP"], quiet=True)
audio_chunk2 = librosa.resample(audio_chunk, orig_sr=44100, target_sr=16000)
audio_chunk3 = (audio_chunk2 * 32768).astype(np.int16)

sounddevice.play(audio_chunk3, 12100)

from scipy.io import wavfile
wavfile.write("output.wav", 12100, audio_chunk3)

sounddevice.wait()
==========
Prompt: ピッ!
と音が鳴ったら、すぐに手を上げてください。
「ピッ!」と音が鳴ったら、すぐに手を上げてください。
「ピッ!」と音が鳴ったら、すぐに手を上げてください。
「ピッ!」と音が鳴ったら、すぐに手を上げてください。
「ピッ!」と音が鳴ったら、すぐに手を上げてください。
「ピッ!」と音が鳴ったら、すぐに手を
Prompt: 4 tokens, 0.961 tokens-per-sec
Generation: 100 tokens, 5.666 tokens-per-sec

何かがおかしい。頭がおかしくなりそう。

どこかバグってる。

  • 発話認識の部分が普通にバグってて「ピッ!」って入力になっている。
  • Llamaに渡しているプロンプトが簡易な状態で、続きの文章を生成しようとする。プロンプトをもっと工夫して渡す。
  • 各モデルで暖気運転しているものがあるけれど、このスクリプトではしていない。

他にもいろいろあるはず。引き続き取り組む。

作業中…