はじめに
OpenAIのSpeech to textやText to speech利用し生成AIと声のみでやり取りするサンプル実装を紹介します。
処理フロー
以前紹介したSlackチャットボットの構造と似ています。
やりとりするデータの種類（テキスト→音声）が異なるだけで根本的なフローはほぼ同じです。
音声ファイルをRequestとして受け取るAPIチックな処理をPython Flaskで作成します。
チャット履歴の保持は今回BigQueryを利用しましたが何でもよいと思います。
インフラはGCPを利用しています。
Cloud RunにPython Flaskアプリのコンテナをデプロイします。
クライアント
ここは正直言って音声ファイルをポストできるものであればなんでも良いところですね。
今回はChatGPTに書いてもらったWebクライアントの一例です。
navigator.mediaDevices.getUserMedia({ audio: true })
.then(stream => {
mediaRecorder = new MediaRecorder(stream);
mediaRecorder.ondataavailable = event => {
audioChunks.push(event.data);
};
mediaRecorder.onstop = () => {
const audioBlob = new Blob(audioChunks, { type: 'audio/mpeg-3' });
const formData = new FormData();
formData.append("audio_data", audioBlob);
fetch("/voice_chat", { method: "POST", body: formData })
.then(response => response.blob())
.then(blob => {
const audioUrl = URL.createObjectURL(blob);
new Audio(audioUrl).play();
});
audioChunks = [];
};
});
document.getElementById("recordCheckbox").addEventListener("change", () => {
if (document.getElementById("recordCheckbox").checked) {
mediaRecorder.start();
} else {
mediaRecorder.stop();
}
});
かなりショボい感じですが画面上のチェックボックスをONにすると録音開始で
チェックを外すと録音終了して録音された音声ファイルをポストします。
レスポンスでは生成AIの出力した音声ファイルが返ってくるのを待ち構えて再生します。
Flask
音声ファイルを受け取って生成AIが作成した音声ファイルをレスポンスする大枠部分です。
# handle chat with llm
@app.route('/voice_chat', methods=['POST'])
def voice_chat():
audio_file = request.files['audio_data']
if audio_file:
current_timestamp = int(time.time())
timestamp_string = str(current_timestamp)
filepath = f'/tmp/{timestamp_string}_recorded_audio.mp3'
audio_file.save(filepath)
speech_file_path = openai_voice_to_voice(timestamp_string)
return send_file(speech_file_path, as_attachment=True)
print('no audio_file!')
return {'status': 'error'}
ダウンロードした音声ファイルをローカルに一時保存します。
ファイル名が重複しないように現在時刻をファイル名へ含めます。
ファイルは後続のSpeech to textで利用します。
メイン処理
openai_voice_to_voice関数の詳細です。
OpenAIのSDKを利用します。
Speech to text
path = f'/tmp/{name}_recorded_audio.mp3'
audio_file = open(path, "rb")
transcript = client.audio.transcriptions.create(model="whisper-1", file=audio_file)
text_user = transcript.text
OpenAI APIのSSTにてにてダウンロードした音声ファイルから文字列に変換します。
Chat Completions
音声から生成した文字列を使ってLLMにテキストを生成してもらいます。
# making prompt with history
print(text_user)
prompt.append({"role": "system", "content": "あなたは人間なので口語で会話してください"})
prompt.extend(model_chat_log.get_logs())
text_prompt = f'あなたは人間として振る舞ってください/n回答は必ず口語調にしてください/n以下が話しかけられた内容です/n「{text_user}」'
prompt.append({"role": "user", "content": text_prompt})
print(prompt)
# send voice message text to llm
response = client.chat.completions.create(model="gpt-3.5-turbo", messages=prompt)
text_assistant = response.choices[0].message.content
print(text_assistant)
# save chat logs
model_chat_log.save_log("user", text_user)
model_chat_log.save_log("assistant", text_assistant)
外部データベースに保持している会話履歴をロードしてプロンプトを構築します。
音声から変換したテキストと生成されたテキストはそれぞれ履歴としてデータベースに保存します。
以下はBigQueryによるサンプルコードです。
# save chat log
def save_log(role, msg):
client = bigquery.Client()
query_job = client.query(f'INSERT INTO app.chat_log(user_id,message,role,created) values(\'mate\',\'\'\'{msg}\'\'\',\'{role}\',CURRENT_DATETIME(\'Asia/Tokyo\'))')
print(query_job)
# load chat logs
def get_logs():
client = bigquery.Client()
query_job = client.query("SELECT * FROM app.chat_log WHERE user_id = 'mate' order by created")
rows = query_job.result()
print(rows.total_rows)
logs = []
for row in rows:
log = {"role": row["role"], "content": row["message"]}
logs.append(log)
return logs
# delete chat logs
def delete_logs():
client = bigquery.Client()
query_job = client.query("DELETE FROM app.chat_log WHERE user_id = 'mate'")
print(query_job)
Text to speech
最後に生成されたテキストをOpenAI APIのTTSを利用して音声ファイルに変換します。
response = client.audio.speech.create(model="tts-1", voice="nova", input=text_assistant)
speech_file_path = f"/tmp/{name}_speech.mp3"
response.stream_to_file(speech_file_path)
さいごに
これは声でやりとりするチャットボットみたいな感じでしょうか。
APIを呼ぶだけで簡単にこのレベルものが作れちゃいます。
音声合成してみたりプロンプトをもっと追い込んだりRAG的なものを導入すればより人間らしいやりとりができるようになりそうな気がします。