faster-Whisper、ChatGPT API、VOICEVOX coreを組み合わせて音声対話AIを作った話

はじめに

こんにちは、某大学院で対話AIについて研究している者です。
研究の中で試してみたいアイデアやモデルを思いついたときの統合先として簡単な音声対話システムを実装してみました。（研で開発されているシステムには気軽に手出しができないので.....）
コードは公開しているので、気軽に音声対話システムを試してみたい方、カスタマイズして自分だけの対話AIを作ってみたい方は雛型として使うことができると思います。

特徴

長所
・音声認識と音声合成をローカルで動作させ、外部APIはChatGPTのみなので応答が速い(と思っている。)
・カスタマイズ性が高い

短所
・少々重い

対象読者

・Pythonの環境構築が自力でできる方
　・仮想環境、pipなどが使える方
・CUDAの環境構築が自力でできる方

想定環境

・Windows OS
・それなりのスペックをもったデスクトップPC
　・Intel 第13世代以上かそれに相当するCPU
　・VRAM 6GB以上のNVIDIA製GPU
・Python 3.12.7
　・ほかのバージョンでの動作は確認していません
・CUDA 12.6以上
　・対応するcuDNNもインストールしてください
　・11.xは動かない可能性が高いです

インストール

※このレポジトリのコードはMITライセンスで公開されています。各ライブラリのライセンスについては対応するgithubレポジトリを参照してください。

クローン

まず、レポジトリをローカル環境へクローンしましょう。

git clone https://github.com/kazu44ttaka/My-SDS

お好みのPython仮想環境上にrequirements.txtに書かれているパッケージをインストールしてください。

Pytorchに関してはrequirements.txtに書かれていないので、ご自身のCUDA環境にあったバージョンを手動でインストールしてください。

私の場合はCUDA 12.6なので、以下のようにPytorchをインストールします。

pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126

faster-Whisperのセットアップ

ASRにはfaster-Whisperというものを使います。

こちらは高性能ASRモデルであるWhisperをリアルタイム動作用にカスタマイズしたもので、これを用いればローカル環境で高速かつ高性能なASRを実現できます。

セットアップといってもやることは一つで、対応するPython ライブラリのインストールです。

pip install git+https://github.com/guillaumekln/faster-whisper.git

推論に用いるモデルのデフォルトはlarge-v3となっていますが、処理が重い場合はパラメータでmediumなどに変えることができます。各モデルは初回実行時に自動でダウンロードされます。

VOICEVOX coreのセットアップ

TTSにはVOICEVOX coreというものを使います。

こちらは有名なVOICEVOXという読み上げソフトウェアの音声合成コアで、ローカル環境で高速に音声合成を行うことができます。
以下の作業をインストールガイドを見ながら行ってください。

Windows版のダウンローダをReleasesからダウンロード、先ほどクローンした場所(C: \...\My-SDS)に配置し、それを起動してVOICEVOX coreのインストールを行ってください。
インストールが完了したら以下のようにvoicevox_coreというファイルが生成されていると思います。

次に、対応するVOICEVOX coreのPython ライブラリをpipでインストールしてください。執筆時点での最新バージョンは0.16.0ですので、この場合のpipコマンドは以下のようになります。

pip install https://github.com/VOICEVOX/voicevox_core/releases/download/0.16.0/voicevox_core-0.16.0-cp310-abi3-win_amd64.whl

ここまででTTSのセットアップは完了です。

合成される音声のデフォルトは春日部つむぎになっていますが、TTSのパラメータで好きな音声を指定することができます。

諸準備

プロンプトファイルの指定

レポジトリをクローンした場所(C: \...\My-SDS)にChatGPTに与えるプロンプトを書いたテキストファイルをおいてください。ファイル名は例えばprompt.txtなどとし、下記のようにmain.pyの冒頭で指定してください。

たとえば、プロンプトの例は以下のようになります。

prompt.txt

あなたはユーザーと雑談を楽しむカジュアルな会話パートナーです。
口語的な自然な日本語で話し、短いフレーズで返答してください。
質問には答えすぎず、会話が続くようにユーザーにも質問を返してください。
表情豊かな語り口で、親しみやすさを重視してください。

以下から会話が始まります。

OpenAI API keyの指定

先ほどと同じ場所に、OpenAIから発行されるAPI Keyを格納したテキストファイルをおいてください。Keyの発行方法などは別途記事を参照してください。ファイル名は例えばkey.txtなどとし、下記のようにGPT.pyの冒頭で指定してください。

実行

main.pyがメインプログラムになっています。これを実行し、以下のようにListening... Ctrl+C to stop.と表示されたら話しかけます。

↓動作例（音が出ます）

CV：VOICEVOX、春日部つむぎ

エージェントに自分を先輩と呼ばせていることがばれる

それなりに高速なレスポンスを実現できているのではないでしょうか？

実装の詳細

音声認識（ASR.py）

マイクからの音声を処理する部分です。

# マイクからの音声をQueueに保持
def callback(self, indata, frames, t, status):
    if status: 
        print(status)
    self.audio_q.put(indata.copy())

# マイクからの音声を処理するループ
def stream(self):
    with sd.InputStream(channels=1,
                        samplerate=self.SAMPLE_RATE,
                        blocksize=self.BLOCK,
                        dtype='float32',
                        latency='low',
                        callback=self.callback):
        print("Listening... Ctrl+C to stop.")
        while True: 
            # if self.audio_q.qsize() > 0:
            #     print("qsize :", self.audio_q.qsize(), "*", self.audio_q.queue[0].shape)
            time.sleep(0.5)

マイクからの音声をself.BLOCK/self.SAMPLE_RATE秒ごとにself.audio_qというQueueに保持するコードになっています。例えば、self.SAMPLE_RATE=16,000、self.BLOCK=320のとき、self.BLOCK/self.SAMPLE_RATE=0.02秒ごとにcallback関数が呼ばれ、マイクからの音声が0.02秒ごとにQueueに保持される、という感じになっています。

次に、音声認識をする部分です。

# ASR
def worker(self):
    buf_voice = []
    while True:
        segment = self.audio2inf.get()
        if len(segment) < self.SAMPLE_RATE * 0.5:
            continue
        segments, _ = self.whisper.transcribe(segment, 
                                            language="ja", 
                                            without_timestamps=True,
                                            beam_size=5)
        for seg in segments:
            # print(f"[{seg.start:.2f}-{seg.end:.2f}] {seg.text}")
            self.user_text.put(seg.text)
        buf_voice.clear()
        time.sleep(0.01)

# VAD、発話区間だけをASRに渡す
def process_vad(self):
    vad_count = 0
    buf_user = []
    user_utterance = []
    trigger = False
    num_vad = 5
    last_index = 0
    while True:
        data = self.audio_q.get()
        buf_user.append(data[:,0])
        buf_user = buf_user[- int(self.VAD_SECONDS / (self.BLOCK / self.SAMPLE_RATE)):]
        self.VAD_LEN = len(np.concatenate(buf_user))
        if vad_count > num_vad - 2:
            full = np.concatenate(buf_user)
            self.vad_full = silero_vad.get_speech_timestamps(
                full, 
                self.vad_model,  
                sampling_rate=self.SAMPLE_RATE,
                threshold=0.3,                          # VADの出力を音声と判断するしきい値
                min_speech_duration_ms=150,              # 発話とみなす最短長（ミリ秒）
                max_speech_duration_s=self.VAD_SECONDS, # 発話区間の最大長（秒）
                min_silence_duration_ms=50,            # 区切りとみなす無音の最短長（ミリ秒）
                window_size_samples=1024,               # 内部処理に使うウィンドウサイズ（サンプル数）
                speech_pad_ms=150,                      # 出力範囲の前後に付け足すバッファ時間（ミリ秒）
                )
            if len(self.vad_full) > 0:
                for i in range(len(self.vad_full)):
                    start_index = (int(self.vad_full[i]["start"]) // self.BLOCK) * self.BLOCK
                    end_index = ((int(self.vad_full[i]["end"]) - 1) // self.BLOCK + 1) * self.BLOCK - 1
                    if end_index > len(full) - self.BLOCK * num_vad:
                        if len(user_utterance) == 0 and start_index > last_index:
                            user_utterance.append(full[start_index:end_index])
                        else:
                            user_utterance.append(full[max(start_index, len(full) - self.BLOCK * num_vad):end_index])
                        last_index = end_index
                    else:
                        last_index -= self.BLOCK * num_vad
                if len(full) - self.vad_full[-1]["end"] > self.SAMPLE_RATE * self.SILENCE_TIME:
                    trigger = True
                else:
                    trigger = False
            else:
                last_index -= self.BLOCK * num_vad
                trigger = False
            
            if len(user_utterance) > 0:
                if trigger or len(np.concatenate(user_utterance)) >= self.SAMPLE_RATE * self.CHUNK_SEC:
                    self.audio2inf.put(np.concatenate(user_utterance))
                    user_utterance.clear()
                    trigger = False
            vad_count = 0
        else:
            vad_count += 1

process_vadでは以下の処理を行っています。

• self.audio_qから音声を取り出し、過去self.VAD_SECONDS秒間のVAD（発話区間検出）をself.BLOCK/self.SAMPLE_RATE*num_vad秒（デフォルトでは0.1秒）ごとに実施し、self.vad_fullに保持する。
• ユーザーが発話を終了してからself.SILENCE_TIME秒後に発話区間として検出された音声をself.audio2infというQueueに保持し、音声認識の推論に回す。

マイクから得られた音声をすべて推論に回さない理由は、

• すべて推論に回していると、処理が追い付かず、リアルタイム動作不可能になる。
• 話の区切りを検出して推論に回した方が認識誤りが少なくなる。

です。ストリーミング音声認識（一文字ずつ逐次認識）ができないモデルでは一般に用いられている手法だと思います。

workerでは、self.audio2infから発話区間検出済みの音声を取り出し、faster-Whisperのモデルに渡して推論を行っています。推論結果はself.user_textというQueueに保持されます。このとき、0.5秒以下の音声はスキップすることで認識誤りを防止しています。（faster-Whisperではなぜかノイズなどを「ご視聴ありがとうございました」と認識してしまうようなので....）

対話処理（GPT.py）

続いてレスポンスを生成する部分です。

# 発話をストリームとして生成
def create_response(self):
    stream = self.client.chat.completions.create(
        model=self.model,
        messages=self.messages,
        stream=True
    )
    return stream

# プロンプトを初期化
def init_prompt(self, sys_prompt):
    self.messages = [
        {"role": "system", "content": sys_prompt},
    ]

# プロンプトを更新
def update_messages(self, message, role):
    if len(self.messages) > self.context_len + 1:
        self.messages = [self.messages[0]] + self.messages[2:] + [{"role": role, "content": message}]
    else:
        self.messages.append({"role": role, "content": message})
    return self.messages

# ロボットがターンを取るかどうか
def turn_taking(self, ASR:ASR):
    while True:
        if not self.messages[-1]["role"] == "assistant":
            if len(ASR.vad_full) > 0:
                if ASR.VAD_LEN - ASR.vad_full[-1]["end"] > self.SAMPLE_RATE * self.MAX_SILENCE_TIME:
                    self.robot_turn = True
            else:
                self.robot_turn = False
        time.sleep(0.05)

こちらはシンプルな構成となっています。

• create_response：これまでの対話履歴からレスポンスをストリーミングとして生成する。
• init_prompt：txtファイルから読み込んだプロンプトをGPTに渡す形式（メッセージ）に変換する。
• update_messages：メッセージを更新する。
• turn_taking：ASRで計算されたVAD情報を処理し、エージェントがターンを取るかどうかを0.05秒ごとに判断する。エージェントが連続してターンを取らないように、最後のメッセージのロールがassistantではないときのみに判断を行う。

テキスト対話と異なる点は、ターンテイキングの判断が加わる点ですね。

音声合成（TTS.py）

続いて音声合成を行う部分です。

# 音声合成モデルを初期化
async def init_model(self):
    ort = await Onnxruntime.load_once(filename=self.onnxruntime_file)
    ojt = await OpenJtalk.new(self.OpenJtalk_dict)
    self.synthesizer = Synthesizer(
        ort,
        ojt,
        acceleration_mode=self.mode,
        cpu_num_threads=max(
            multiprocessing.cpu_count(), 2
        ),
    )
    async with await VoiceModelFile.open(self.vvm) as model:
        await self.synthesizer.load_voice_model(model)

# 音声合成
async def voice_synth(self, text):
    # クエリ生成
    query = await self.synthesizer.create_audio_query(text, self.speakerID)
    # 合成
    audio = await self.synthesizer.synthesis(query, self.speakerID)
    
    self.q_audio.put(audio)

# エージェントの音声を再生するループ
def speak(self):
    while True:
        if not self.q_audio.empty():
            audio = self.q_audio.get()
            
            # メモリ上のWAVデータを直接再生
            audio_bytes = io.BytesIO(audio)
            
            # soundfileで読み込み → sounddeviceで再生
            with sf.SoundFile(audio_bytes) as f:
                data = f.read(dtype='float32')
                sd.play(data, f.samplerate)
                sd.wait()
        time.sleep(0.05)

VOICEVOX coreには同期処理と非同期処理を行うものが用意されているのですが、今回は素早いレスポンスを重視しているので、非同期処理で実装しています。

• init_model：モデルの初期化を行う。
• voice_synth：エージェントの発話をテキストとして受け取り、音声合成を行い、self.q_audioというQueueに保持する、
• speak：self.q_audioから音声を取り出し、SoundFileで再生する。

非同期処理+Queueの組み合わせで高速な音声合成を実現しています。

メインプログラム（main.py）

以下がメインプログラムです。

myASR = ASR.ASR()

# ASR、マイクからの音声処理、VAD用のスレッドを定義
threading.Thread(target=myASR.worker, daemon=True).start()
threading.Thread(target=myASR.stream, daemon=True).start()
threading.Thread(target=myASR.process_vad, daemon=True).start()

myGPT = GPT.GPT()
myGPT.init_prompt(sys_prompt=sys_prompt)

# ターンテイキング用のスレッドを定義
threading.Thread(target=myGPT.turn_taking, daemon=True, args=(myASR,)).start()

myTTS = TTS.TTS()
loop_voice_synth = asyncio.new_event_loop()

def loop_runner(loop:asyncio.AbstractEventLoop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

# 音声合成用のスレッドを定義
threading.Thread(target=loop_runner, daemon=True, args=(loop_voice_synth,)).start()
asyncio.run_coroutine_threadsafe(myTTS.init_model(), loop_voice_synth).result()
threading.Thread(target=myTTS.speak, daemon=True).start()

while True:
    # ユーザーの発話を取得
    if not myASR.user_text.empty():
        user_utterance = myASR.user_text.get()
        print(user_utterance)
        myGPT.update_messages(user_utterance, "user")

    # ロボットがターンを取得したらGPTに渡してレスポンスを生成
    if myGPT.robot_turn:
        stream = myGPT.create_response()
        text_full = ""
        text_tmp = ""
        for chunk in stream:
            content = chunk.choices[0].delta.content
            if content == None:
                continue
            if  len(content) > 0 and content[-1] in set(["、", "。", "！", "？", "♪", "♡"]) and text_tmp != "":
                if content[-1] in set(["！", "？"]):
                    text_tmp += content
                print("Agent speak :", text_tmp)
                asyncio.run_coroutine_threadsafe(myTTS.voice_synth(emoji_pattern.sub("、", text_tmp)), loop_voice_synth)
                text_tmp = ""
            else:
                text_tmp += content
            text_full += content
            # print(content, end="")
        if emoji_pattern.sub(r'', text_tmp) != "":
            print("Agent speak :", text_tmp)
            asyncio.run_coroutine_threadsafe(myTTS.voice_synth(emoji_pattern.sub("、", text_tmp)), loop_voice_synth)
        # print(text_full)
        myGPT.update_messages(text_full, "assistant")
        myGPT.robot_turn = False

    # print("latency :", myASR.audio_q.qsize() * (myASR.BLOCK / myASR.SAMPLE_RATE), "s")
    time.sleep(0.1)

while Trueの手前でこれまで定義した各コンポーネントの処理ループを別スレッドとして動かしています。基本的には並列処理+Queueの合わせ技ですね。

while Trueのループ中には以下の処理を順番に行っています。

• ASRのuser_txtからユーザーの発話を取り出し、メッセージを更新する。
• エージェントがターンを取得したらレスポンスを生成し、1文ごとに音声合成→再生を行う。
• 生成されたレスポンスでメッセージを更新する。

ここでの工夫点は生成されたレスポンスを一括で音声合成するのではなく、意味の区切りで分けて逐次的に音声合成処理を行っている点です。これにより、レスポンス全体の生成完了を待つことなく音声合成をし始めることができるので、スムーズな応答が実現できています。

追加の機能を実装する際にはこのループ内に組み込むとよいと思います。

まとめ

faster-Whisper、ChatGPT API、VOICEVOX coreを組み合わせた音声対話システムをご紹介しました。私は何でもローカル環境で動かすのが好きな人間なので、音声対話に必要なコンポーネントをできるだけ自分の環境で動かすシステムを構築してみました。
自分の作ったものを外に出すのはこれが初めてなので、至らない点などあるとおもいますが、気軽に指摘していただけると幸いです。

今後の展望としては
・音声対話用GUIアプリケーションの開発
・アバターを使った対話AIとの統合
・対話相手の情報を格納するデータベースを作り、「記憶」機能の実装
・webカメラからの映像から対話相手を特定し、データベースと照合する
などを考えています。
これが1台のPCで同時に動くのだろうか....

ここまで読んでいただきありがとうございました。