[python] ChatGPT APIを使ってレスポンスが高速な音声アシスタントを作る

ChatGPT APIを使って音声アシスタントを作ります。

前回の記事 (https://qiita.com/Nekonun/items/c9a97a028441fdb0b891) でも、ChatGPT APIを使った簡単な音声アシスタントを作りましたが、その時点では最も大きな問題として下記のようなものがありました。

課題1：質問〜返答までのオーバーヘッドが大きい＋安定しない
今回は[音声認識]+[ChatGPTとのやり取り]+[音声ファイル化]という幾つかの処理を経由しています。発話した内容にもよるのですが、必然的にそれらの処理時間のオーバーヘッドが結構大きくなってしまいます。
特にChatGPTのAPIでは「ChatGPTが完全にレスポンス内容を生成しきってから、まとめてこちらに返す」という仕様になっています（何やら生成できた文字から随時返すようなオプションもあるようですが...）。そのため複雑な質問をしてChatGPTが悩んでいる時や、回答内容が長文になる場合には、かなりの時間待たされます。ChatGPTのWebサービス上では、生成した文字をリアルタイムで順に表示していってくれるのであまり待たされているような感じはしませんが、APIを使うとどうしてもこの辺りがネックになります。なので、より使いやすくするためには、生成できた文章からどんどん受け取って音声で再生していくなどの工夫が必要です。

通常、ChatGPTにAPIでリクエストを投げると、ChatGPT側では全てのレスポンス内容の生成が完了・確定してから、まとめて文章が返されます。そのため、レスポンスが長い場合は、こちらがリクエストを出してからかなり待たされてしまい、インタラクティブな会話が楽しめません。

そこで今回はstreamオプションを使用して、できる限り会話のレスポンス速度を向上させてみます。

開発環境

• マシン：MacBook Pro (2 GHz クアッドコアIntel Core i5)
• OS : Ventura 13.2.1
• Python : 3.9.1
• pyttsx3 : 2.90

ソースコード

今回実装したソースコードは下記の通りです。
音声認識にはspeech_recognitionを、音声合成と発話にはpyttsx3を使用します。APIキーは自身で取得したものに置き換えて下さい。

なおMacOS環境でpyttsx3を使用した際にKeyError: 'VoiceAge'が発生して使用できない場合があります。そんなときは https://stackoverflow.com/questions/74668118/voiceage-error-while-using-pyttsx3-module-to-add-voice-to-statements/74727956#74727956 を参考にnsss.pyの中身を編集することでエラーを解消できます。

import speech_recognition as sr
import os
import openai
import pyttsx3
import re

##############
# 音声認識関数 #
##############
def recognize_speech():

    recognizer = sr.Recognizer()    
    # Set timeout settings.
    recognizer.dynamic_energy_threshold = False

    
    with sr.Microphone() as source:
        recognizer.adjust_for_ambient_noise(source)
    
        while(True):
            print(">> Please speak now...")
            audio = recognizer.listen(source, timeout=1000.0)

            try:
                # Google Web Speech API を使って音声をテキストに変換
                text = recognizer.recognize_google(audio, language="ja-JP")
                print("[You]")
                print(text)
                return text
            except sr.UnknownValueError:
                print("Sorry, I could not understand what you said. Please speak again.")
                #return ""
            except sr.RequestError as e:
                print(f"Could not request results; {e}")
                #return ""


#################################
# Pyttsx3でレスポンス内容を読み上げ #
#################################
def text_to_speech(text):
    # テキストを読み上げる
    engine.say(text)
    engine.runAndWait()



def chat(conversationHistory):
    # APIリクエストを作成する
    response = openai.ChatCompletion.create(
        messages=conversationHistory,
        max_tokens=1024,
        n=1,
        stream=True,
        temperature=0.5,
        stop=None,
        presence_penalty=0.5,
        frequency_penalty=0.5,
        model="gpt-3.5-turbo"
    )

    # ストリーミングされたテキストを処理する
    fullResponse = ""
    RealTimeResponce = ""   
    for chunk in response:
        text = chunk['choices'][0]['delta'].get('content')

        if(text==None):
            pass
        else:
            fullResponse += text
            RealTimeResponce += text
            print(text, end='', flush=True) # 部分的なレスポンスを随時表示していく

            target_char = ["。", "！", "？", "\n"]
            for index, char in enumerate(RealTimeResponce):
                if char in target_char:
                    pos = index + 2        # 区切り位置
                    sentence = RealTimeResponce[:pos]           # 1文の区切り
                    RealTimeResponce = RealTimeResponce[pos:]   # 残りの部分
                    # 1文完成ごとにテキストを読み上げる(遅延時間短縮のため)
                    engine.say(sentence)
                    engine.runAndWait()
                    break
                else:
                    pass

    # APIからの完全なレスポンスを返す
    return fullResponse


##############
# メインの関数 #
##############
if __name__ == '__main__':

    ##################
    # ChatGPTの初期化 #
    ##################
    openai.api_key="自身のAPIキーを指定"
    # UserとChatGPTとの会話履歴を格納するリスト
    conversationHistory = []
    setting = {"role": "system", "content": "句読点と読点を多く含めて応答するようにして下さい。また、1文あたりが長くならないようにして下さい。"}
    
    ##################
    # Pyttsx3を初期化 #
    ##################
    engine = pyttsx3.init()
    # 読み上げの速度を設定する
    rate = engine.getProperty('rate')
    engine.setProperty('rate', rate)
    # Kyokoさんに喋ってもらう(日本語)
    engine.setProperty('voice', "com.apple.ttsbundle.Kyoko-premium")

    # Ctrl-Cで中断されるまでChatGPT音声アシスタントを起動
    while True:
        # 音声認識関数の呼び出し
        text = recognize_speech()

        if text:
            print(" >> Waiting for response from ChatGPT...")
            # ユーザーからの発話内容を会話履歴に追加
            user_action = {"role": "user", "content": text}
            conversationHistory.append(user_action)
            
            print("[ChatGPT]") #応答内容をコンソール出力
            res = chat(conversationHistory)
            
            # ChatGPTからの応答内容を会話履歴に追加
            chatGPT_responce = {"role": "assistant", "content": res}
            conversationHistory.append(chatGPT_responce) 
            print(conversationHistory)

            

レスポンスを高速化するためにやったこと

肝はchat()関数部分です。APIリクエストを作成する際にopenai.ChatCompletion.create()内でstreamオプションを指定しています。ChatGPTにstreamオプションについて聞いてみると下記のように教えてくれました。

openai.ChatCompletion.create()は、OpenAIのAPIを使用して、与えられた入力に基づいてチャットボット応答を生成するための関数です。この関数は同期的に実行され、APIにリクエストを送信し、応答を返します。
一方、streamパラメータは、openai.ChatCompletion.create()を非同期的に実行するための方法を提供します。streamパラメータにTrueを設定すると、APIから応答が生成されるたびに、生成された応答を返すイテレーターを返します。これにより、応答が生成されるまで待つ必要がなく、応答が生成されるたびにプログラムが反応することができます。

つまり、streamオプションを指定することで「全ての応答の生成が完了してから」返すのではなく、「生成できた文字列」から逐次的に返してくれるようになるみたいです。つまりWebサービスのChatGPTのような応答形式にできます。

def chat(conversationHistory):
    # APIリクエストを作成する
    response = openai.ChatCompletion.create(
        messages=conversationHistory,
        max_tokens=1024,
        n=1,
        stream=True,
        temperature=0.5,
        stop=None,
        presence_penalty=0.5,
        frequency_penalty=0.5,
        model="gpt-3.5-turbo"
    )

    # ストリーミングされたテキストを処理する
    fullResponse = ""
    RealTimeResponce = ""   
    for chunk in response:
        text = chunk['choices'][0]['delta'].get('content')

        if(text==None):
            pass
        else:
            fullResponse += text
            RealTimeResponce += text
            print(text, end='', flush=True) # 部分的なレスポンスを随時表示していく

            target_char = ["。", "！", "？", "\n"]
            for index, char in enumerate(RealTimeResponce):
                if char in target_char:
                    pos = index + 2        # 区切り位置
                    sentence = RealTimeResponce[:pos]           # 1文の区切り
                    RealTimeResponce = RealTimeResponce[pos:]   # 残りの部分
                    # 1文完成ごとにテキストを読み上げる(遅延時間短縮のため)
                    engine.say(sentence)
                    engine.runAndWait()
                    break
                else:
                    pass

    # APIからの完全なレスポンスを返す
    return fullResponse

上記のソースコードの例ではtext変数に、逐次APIからの応答内容（ぶつ切りになったもの）が格納されていきます。print(text, end='', flush=True)で、その応答内容を逐次的にコンソール表示しています。

応答が帰ってくるたびにRealTimeResponce変数に、ぶつ切りになった文字列を追加していき、レスポンス全体を構築していきます。その際、1つの文が完成したタイミングで、その完成した文章から先に音声合成・読み上げを行います。 「全ての応答内容（複数の文が含まれる）が完成してから音声合成を行い読み上げる」という形を取ると、応答が完成するまで読み上げができず、ユーザーに「待たされている感」を与えてしまいます。なので、今回は"。", "！", "？", "\n"が出現するまでが1つの文章と定義し、RealTimeResponce変数内の文字列にそれらが出現したタイミングで、当該文だけ切り出して音声合成・読み上げを行います。pyttsx3で読み上げを行なっている間もAPIからは随時応答が返ってきている状態なので、1つの文章が読み終わった頃には、すでに次の文章を読む準備ができています。これによって、ユーザーから見た「質問〜応答までの遅延」を削減しています。

なお、もっと遅延を削減したければ「1文の完成を待たずに、textの内容を逐次読み上げる」「10文字揃ったらその都度読み上げる」などの方策を取れます。しかし文章が変なところで切れることになるので、音声合成を行った際に違和感のある読み上げ方になってしまう可能性があります。そのため今回は「遅延」と「聞きやすさ」のバランスをとって1文で区切ることにしています。

ですが、APIからの応答の「最初の1文目」が長すぎると、やはりユーザーは待たされてしまいます。なので{"role": "system", "content": "句読点と読点を多く含めて応答するようにして下さい。また、1文あたりが長くならないようにして下さい。"}という指示を一応与えています。

ちなみに文章の区切り文字として\nを指定しているのは、ChatGPTが"。"や"、"を一切使わず、箇条書きのみで応答を返す場合を考慮してのことです。

また前回作成した音声アシスタントでは一度gTTSを使ってChatGPTからの応答内容を.mp3形式に変換してから、pygameを使って音声ファイルを再生するという形をとっていました。しかし今回はpyttsx3を使用してテキストを直接読み上げる形にしているので、ここでもオーバーヘッドの削減を実現しています。

課題点

APIからの応答内容が長文な場合「音声アシスタントが喋っている時間が長い！」という不満が起きてしまいますが、これはChatGPTに文字数を制限するように指示したり、読み上げ速度を早くすることで改善できます。

しかし今回音声合成に使用したpyttsx3は、音声合成の精度としては微妙でした。アルファベットをそのまま読見上げてしまったり、少し日本語のイントネーションが機械的でした。前回使用したgTTSは日本語読み上げの精度は良かったのですが、文字を音声ファイルに変換するため読み上げまでにオーバーヘッドが生じる点が微妙でした。何か他にいい感じの音声合成エンジンがあれば、もっと良くなるかもしれませんね。

あと、やっぱり複雑な指示を与えたり、プログラムなどの成果物を出力させるには音声アシスタントは不向きですね。音声アシスタントはあくまで「日常的な（料金が発生する）会話相手」と割り切ると思うと、やはり自分の好みの声で喋ってほしいな...という気持ちがあります。