こんにちは、金融ソリューション事業部の若本です。
先日ChatGPT（gpt-3.5-turbo）のAPIが公開されるとともに、Open AIのサービスが使いやすくなりました。
今回は、ChatGPTから返ってきたレスポンスを読み上げる簡易アプリケーションの作成を行います。
使用するもの
- OpenAI API
- ESPNet
- 有志の方が作成されている音声処理ツールキットです。音声認識から音声合成まで、幅広い音声タスク/手法が実装されています。
- 今回は、音声合成を使用します。使用する音声合成モデルは、つくよみちゃんコーパスを用いた学習済みモデルです。
- fastAPI（任意）
- Streamlit（任意）
- Docker（任意）
- 筆者は上記fastAPIとStreamlitの環境を、別々のDocker containerとして構築しました。
処理概要
下記の処理を実装します。
- APIを用いてAIの推論結果を返せるようにする
- 質問（Text）を投げ、ChatGPTの返答（Text）を受け取る
- 文章（Text）を投げ、音声合成の結果（Audio）を受け取る
- 上記のAPIに処理を依頼する簡易アプリ
今回は音声合成の課題である『音声合成に時間がかかる問題』に対処します。
音声合成は、そのモデルの仕組み上、基本的にストリーミングのようにデータを受け取ることができません。
そのため、ChatGPTのレスポンスをそのまま音声合成してしまうと、数十秒～数分間待たされる可能性があります。
それでは全く使い物にならないため、今回はpythonのマルチスレッド処理を用いて音声合成を並行処理で行います。
別の音声合成の課題として「長文の音声合成が安定しない」というのもあり、上記のアプローチはその面でも効果的です。
ただし、最終的な合成の品質をなるべく落とさないためにどの区切りで音声合成を実行するかを判定する必要があります。
1. APIの作成
まず、処理の要となるAPIを作成します。
今回は保守性向上のためAPIとしているだけですので、必ずしもAPIにする必要はありません。
fastAPI/Streamlitの基本的な使用方法、およびESPNetの環境構築については割愛します。
1.1 Open-AI API(ChatGPT)を介して、質問の答えを取得
APIのRouterにChatGPTを呼び出す関数を記述します。OpenAIのブログを参考に、requestライブラリで実施しました。執筆時点（2023/03/07）の情報ですので、適宜request_bodyは変更ください。
他にもOpenAIライブラリで実装する方法もあります。
from fastapi import APIRouter
import requests
import json
TEMPLATE_PATH = "./data/template.txt"
SYSTEM_TEMPLATE_PATH = "./data/system_template.txt"
API_KEY = 'YOUR_OPEN_API_KEY'
router = APIRouter()
template = load_template(TEMPLATE_PATH)
system_prompt = load_template(SYSTEM_TEMPLATE_PATH)
def load_template(path:str):
with open(path, 'r') as f:
template = f.read()
return template
def edit_prompt(template:str, text:str, token="{query}"):
prompt = template.replace(token, text)
return prompt
def prompt_response(prompt:str, system_prompt:str):
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + API_KEY,
}
data = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": prompt}, {"role": "system", "content": system_prompt}],
"max_tokens": 200,
"temperature": 1,
"top_p": 1,
}
response = requests.post('https://api.openai.com/v1/chat/completions', headers=headers, data=json.dumps(data))
return response.json()
def foreprocessing(res):
text = res['choices'][0]['message']['content']
return text
@router.get("/chat")
def get_chat_response(text:str):
prompt = edit_prompt(template, text)
res = prompt_response(prompt, system_prompt)
res_text = foreprocessing(res)
return res_text
また、この時ChatGPTにクエリを送るためのテンプレートを用意しておきます。
template.txt
{query}
system_template.txt
文章に対して簡潔に回答してください。 あなたはアシスタントで、ユーザーと会話をしています。
template.txtの内容は"role":"user"に、system_template.txtの内容は"role":"system"にそれぞれ入力として設定されます。今回は"role":"user"にユーザーの入力を、"role":"system"にChatGPTの振る舞いの指定を実施しました。{query}はユーザーの入力で置き換えられます。
今回は音声合成時に、あまり長くなく、かつ口語的に返してもらうために振る舞いを指定しておきます。
1.2 Espnetを用いて、合成音声を生成
下記のコードだけで音声合成が実行できます。非常にお手軽です。
今回使用しているモデルはVITS (Conditional Variational Autoencoder with Adversarial Learning for End-to-End Text-to-Speech)です。
model_tagで使用するmodelを指定することにより、モデルを自動でダウンロードできます。
代わりの引数として、model_fileに自分で作成したモデルのpathを指定することも可能です（動作確認済み）。音声合成モデルを最初から学習するためのレシピもEspnetには整備されていますので、興味のある方はぜひ。
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from espnet2.bin.tts_inference import Text2Speech
import torch
import soundfile as sf
import uuid
import os
router = APIRouter()
fs, lang = 44100, "Japanese"
text2speech = Text2Speech.from_pretrained(
model_tag="kan-bayashi/tsukuyomi_full_band_vits_prosody",
device="cpu", # or "cuda"
speed_control_alpha=1.0,
noise_scale=0.333,
noise_scale_dur=0.333,
)
def TTS_streamer(text: str):
with torch.no_grad():
wav = text2speech(text)["wav"]
filename = str(uuid.uuid4())
sf.write(f"{filename}.wav", wav.view(-1).cpu().numpy(), text2speech.fs)
with open(f"{filename}.wav", mode="rb") as wav_file:
yield from wav_file
os.remove(f"{filename}.wav")
@router.get("/tts")
async def tts_streamer(text: str):
return StreamingResponse(TTS_streamer(text), media_type="audio/wav")
上記でAPIサーバーを起動します。FastAPIにはSwaggerUIも用意されているため、画面上で確認することができます。
APIとして作成した機能の動作が確認できた後、次のステップに進みます。
2. APIを呼び出すアプリの作成
次に、ユーザーの入力とAPI処理を繋ぐ簡易アプリを実装します。
ここでは、下記のような処理としました。
- ChatGPTのレスポンスを、句読点や改行で区切る
- ThreadPoolExecutorを用いてAPIへのリクエストを並行処理
- Threadの実行を順番に待機（＋実行時間のスケジュール設定）
- 音声の再生に必要な秒数分、以降の実行を遅らせる
- もし音声が遅く届いた場合、遅れた時間だけ以降の実行を遅らせる
Chat-GPTの文章を区切り、並行処理で音声合成を実行していきます。
Chat-GPTのレスポンスは1文が長い、かつ改行が出現する可能性があるため、区切り文字は句読点と改行としました。
今回は単一APIを同じリソース上で呼び出しているため、並行処理にするメリットは薄いかもしれませんが、実用上は必要な処理になります。
import streamlit as st
import requests
import base64
import time
import re
import datetime
from concurrent.futures import ThreadPoolExecutor
# 音声合成の最小文字数。小さすぎると安定しない場合があります。
SPLIT_THRESHOLD = 4
# 息継ぎの秒数（s）
TIME_BUFFER = 0.1
# 待機中の実行スパン（s）
SLEEP_ITER = 0.2
# APIのそれぞれのURL
CHATBOT_ENDPOINT = 'http://chatbot-backend:8000/chat'
TTS_ENDPOINT = 'http://chatbot-backend:8000/tts'
def split_text(text:str):
text_list = re.split('[\n、。]+', text)
text_list_ = []
for text in text_list:
if text == '':
continue
if len(text) < SPLIT_THRESHOLD:
try:
text_list_[-1] = text_list_[-1] + '。' + text
except IndexError:
text_list_.append(text)
else:
text_list_.append(text)
if len(text_list[0]) < SPLIT_THRESHOLD and len(text_list_) > 1:
text_list_[1] = text_list[0] + '。' + text_list_[1]
text_list_ = text_list_[1:]
return text_list_
def get_tts_sound(text:str, url=TTS_ENDPOINT):
params = {'text': text}
response = requests.get(url, params=params)
return response.content, datetime.datetime.now()
def sound_player(response_content:str):
# 参考：https://qiita.com/kunishou/items/a0a1a26449293634b7a0
audio_placeholder = st.empty()
audio_str = "data:audio/ogg;base64,%s"%(base64.b64encode(response_content).decode())
audio_html = """
<audio autoplay=True>
<source src="%s" type="audio/ogg" autoplay=True>
Your browser does not support the audio element.
</audio>
""" %audio_str
audio_placeholder.empty()
time.sleep(0.5)
audio_placeholder.markdown(audio_html, unsafe_allow_html=True)
def get_chat_response(text:str, url=CHATBOT_ENDPOINT):
params = {'text': text}
response = requests.get(url, params=params)
return response.text
if __name__ == "__main__":
st.set_page_config(layout="wide")
query = st.text_input('質問を入力してください')
button = st.button('実行')
if button:
# チャットボットの返信を取得
response_text = get_chat_response(query)
st.write(f'回答：{response_text}')
# 返信を分割
split_response = split_text(response_text)
executor = ThreadPoolExecutor(max_workers=2)
futures = []
# 並行処理として音声合成へ
for sq_text in split_response:
future = executor.submit(get_tts_sound, sq_text)
futures.append(future)
block_time_list = [datetime.timedelta() for i in range(len(futures))]
current_time = datetime.datetime.now()
# 結果をwaitし、再生可能時間になり次第再生する
res_index = 0
gap_time = datetime.timedelta()
while res_index < len(futures):
future = futures[res_index]
if future.done():
if res_index==0:
base_time = datetime.datetime.now()
if datetime.datetime.now() > base_time + block_time_list[res_index]:
for i in range(len(block_time_list)):
if i > res_index:
# 音声長を計算。音声は32bitの16000Hz、base64エンコードの結果は1文字6bitの情報であるため、下記の計算で算出できます
block_time_list[i] += datetime.timedelta(seconds=(len(future.result()[0])*6/32/16000)+gap_time.total_seconds()+TIME_BUFFER)
st.write(f' 実行完了：{split_response[res_index]}')
st.write(f' 実行時間：{(future.result()[1] - current_time).total_seconds():.3f}s')
st.write(f' 音声の長さ：{len(future.result()[0])*6/32/16000:.3f}s')
sound_player(future.result()[0])
res_index += 1
gap_time = datetime.timedelta()
elif res_index!=0:
gap_time += datetime.timedelta(seconds=SLEEP_ITER)
time.sleep(SLEEP_ITER)
executor.shutdown()
早速結果を見てみましょう。
リアルタイムとはいきませんでしたが、概ね許容範囲です。
なお、こちらは筆者のGPU非搭載ノートPCで実行しています。今回は簡易的な検証でしたので、モデルの圧縮、コードの簡素化、高スペックサーバーの使用など、高速化できる余地は多々あります。
並行実行することにより、繋ぎ目となる文章の箇所に違和感が残る懸念もありましたが、あまり違和感は感じられませんでした。
返ってくる結果が遅い場合には、フィラー（「えーと」「その」など）を入れる判定を噛ませることで自然になりそうです。
おわりに
ChatGPTとESPNetを用いて、会話の返答を音声で出力する簡易アプリケーションを作成しました。
OpenAIは、ChatGPTの他にもWhisperという音声認識モデルをAPIとして公開していますので、連結することで音声だけで会話することも可能です。
映画アイアンマンのJ.A.R.V.I.S.に憧れてAIの勉強を始めたので、夢見た未来がだいぶ近づいてきたなと感じます。このようなAIが親しみやすい形で生活に浸透していくのか、今後も注視していきます。
（追記：記事公開前にGPT-4がローンチされました。AI領域の進歩の加速を感じる昨今です。）
それでは最後に。
執筆：@wakamoto.ryosuke、レビュー：Ishizawa Kento (@kent)
（Shodoで執筆されました）
