最近では、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で取得するが、検証時にはファイルから読み出した方が楽な事がある。そのためファイルから入力と同じ形式のデータを取得できるようにした。
もしかすると、型は少し異なるかもしれない。
参考
- https://www.wizard-notes.com/entry/python/sounddevice#%E3%83%AA%E3%82%A2%E3%83%AB%E3%82%BF%E3%82%A4%E3%83%A0%E5%87%A6%E7%90%86
雑音を消す
音には聞きたい音と、邪魔な音がある。音楽を聞いている時、楽器の音は聞きたいけれど、人の話し声は邪魔だ。誰かと話している時、人の声は聞きたいけれど、車の音は邪魔だ。この邪魔な音は雑音と呼ばれる。ここではこの雑音を消す方法を試す。
ここではローパスフィルターとハイパスフィルターを使って、ノイズを除去している。言葉を認識する時、人の声以外の音は必要ない。人の声の周波数は300Hzから3400Hzの範囲と言われるため、その周波数の外側にある音を強制的に消す。
音のデータはsounddeviceで取得するため、そのデータから変換してノイズを除去してみる。
参考
- https://www.kushiro-ct.ac.jp/library/kiyo/kiyo37/ohtsuki37.pdf
- https://qiita.com/Tadataka_Takahashi/items/e30da1d30e4dc2e255d1
- https://qiita.com/p_x9/items/ef19951eed23c9ccdb3b
- https://qiita.com/icoxfog417/items/d376200407e97ce29ee5
- https://qiita.com/kotai2003/items/f5528f70bffa51816c78
- https://zenn.dev/binomialsheep/scraps/c13f1c102dafdc
- https://qiita.com/kmykprn/items/1be922712497ebef0ae0
- https://qiita.com/zerowarurei/items/b86bdc1175dfd0aa0811
- https://qiita.com/RyutoYoda/items/435787141cac5dd9eed7
- https://zenn.dev/yuta_enginner/articles/bccc4624d82cc8
誰かが話しているかを判断する
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に渡しているプロンプトが簡易な状態で、続きの文章を生成しようとする。プロンプトをもっと工夫して渡す。
- 各モデルで暖気運転しているものがあるけれど、このスクリプトではしていない。
他にもいろいろあるはず。引き続き取り組む。
作業中…