🙌
【streamlit】音声会話アプリ開発手順④
第三章ではユーザーとAIが音声で会話できる機能を実装しましたが、AIの応答はテキストが完全に返されてから音声変換していましたので、回答が長いとAIが話し出すまでタイムラグがあります。
そこで回答をストリーミング再生することで、やりとりをスムーズにできるようにします。
4-1 返答をストリーム表示
第二章の3でStreamlitCallbackHandlerを使って触れたように、回答をストリーム処理するには
ChatOpenAI.invoke(config)でconfigにコールバックを入れるというやり方で実装します。
使うのはBaseCallbackHandlerというクラスですが、これはコールバックを実装するために必要なプロパティだけを提供しているものであり、音声再生のように自分で実装したい機能がある場合は、BaseCallbackHandlerを継承したカスタムクラスを作成します。
※ LangChainのコールバックに関しては以下のサイトが詳しく解説してくれています
こっちは公式
3-5から修正するのは少しわかりにくいので、いったんサラの状態にして実装します。
サンプルコード
# 返答をストリーム表示
from typing import List, Any
from uuid import UUID
from langchain_core.outputs import LLMResult
import streamlit as st
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import StreamlitChatMessageHistory
from langchain_core.callbacks.base import BaseCallbackHandler
from gtts import gTTS
from tempfile import NamedTemporaryFile
import time
# 環境変数の読み込み
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
# Streamlitアプリの設定
st.set_page_config(page_title="AIチャットアプリ。ストリーミングチャット", page_icon="🤖")
st.title("AIチャットアプリ")
# OpenAI LLMの初期化
chat = ChatOpenAI(temperature=0.7, api_key=OPENAI_API_KEY, streaming=True, model='gpt-4o-mini')
# StreamlitChatMessageHistoryの初期化
chat_history = StreamlitChatMessageHistory(key="chat_messages")
# プロンプトとしてSystemMessageを入れる
system_message = SystemMessage(content="これからの会話はすべて関西弁で返答してください。")
chat_history.messages.append(system_message)
# カスタムコールバックハンドラの作成
class CustomStreamlitCallbackHandler(BaseCallbackHandler):
def __init__(self):
self.sentense:str = ""
self.sentenses:List[str] = []
def on_llm_new_token(self, token: str, **kwargs) -> None:
self.sentense += token
if token in ['!','！','?','？','、','。']:
self.sentenses.append(self.sentense)
self.sentense = ""
def on_llm_end(self, response: LLMResult, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs:Any):
chat_history.add_ai_message("".join(self.sentenses))
# sentensesの内容を読み上げる
for sentense in self.sentenses:
tts = gTTS(sentense, lang='ja')
with NamedTemporaryFile(suffix=".mp3", delete=False) as temp_audio_file:
tts.save(temp_audio_file.name)
st.audio(temp_audio_file.name, autoplay=True)
time.sleep(len(tts.text)/5) # 待機を入れておかないと、今のセンテンスをしゃべり終える前に次のセンテンスを話し始めてしまう
if user_input := st.chat_input("メッセージを入力"):
chat_history.add_user_message(user_input)
with st.chat_message("assistant"):
stream_handler = CustomStreamlitCallbackHandler()
messages = [
HumanMessage(content=msg.content) if isinstance(msg, HumanMessage) else AIMessage(content=msg.content)
for msg in chat_history.messages
]
response = chat.invoke(
input=messages,
config={"callbacks": [stream_handler]}
)
大事な部分はカスタムコールバックハンドラの作成の部分です。
見ての通りBaseCallbackHandlerを継承したもので、BaseCallbackHandlerで定義されているon_llm_new_tokenやon_llm_endを自分なりに改装して、カスタムコールバックを作成します。
（他にもllmが始まったタイミングやエラーハンドリングなど、いくつかのメソッドがあります。詳しくは公式参照）
# カスタムコールバックハンドラの作成
class CustomStreamlitCallbackHandler(BaseCallbackHandler):
def __init__(self):
self.sentense:str = ""
self.sentenses:List[str] = []
def on_llm_new_token(self, token: str, **kwargs) -> None:
self.sentense += token
if token in ['!','！','?','？','、','。']:
self.sentenses.append(self.sentense)
self.sentense = ""
def on_llm_end(self, response: LLMResult, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs:Any):
chat_history.add_ai_message("".join(self.sentenses))
# sentensesの内容を読み上げる
for sentense in self.sentenses:
tts = gTTS(sentense, lang='ja')
with NamedTemporaryFile(suffix=".mp3", delete=False) as temp_audio_file:
tts.save(temp_audio_file.name)
st.audio(temp_audio_file.name, autoplay=True)
time.sleep(len(tts.text)/5) # 待機を入れておかないと、今のセンテンスをしゃべり終える前に次のセンテンスを話し始めてしまう
読んでのごとく、on_llm_new_tokenはトークンが返ってくるたびに実行されるメソッド、on_llm_endは回答が終わった時に実行されるメソッドです。
on_llm_new_tokenが終わった時には、self.sentensesが以下のように文ごとに並んでいるリストとなりますので、on_llm_endでは一文ずつ抜き出して音声変換して流すという流れにしています。
[
'こんにちは、',
`何かお役に立てることはありますか?`
]
AIの回答が全て返ってきてから音声合成を始めているのでちょっとタイムラグがありますが、全ての回答テキストを音声に直すのと比べると格段に応答スピードが早く感じられるのではないでしょうか。
4-2 返答をストリーミングしつつ、音声会話
最後に第三章の内容をストリーミング回答できるようにします。
ここまでの集大成で、新しいものは入れていないので、詳細説明は割愛します。
なお、4-1からCustomStreamlitCallbackHandlerが読みやすくなるよう少し変えています。
サンプルコード
# 返答をストリーミングしつつ、音声会話
from typing import List, Any
from uuid import UUID
import time
import streamlit as st
from langchain_core.outputs import LLMResult
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import StreamlitChatMessageHistory
from langchain_core.callbacks.base import BaseCallbackHandler
from langchain_community.callbacks import StreamlitCallbackHandler
from gtts import gTTS
from tempfile import NamedTemporaryFile
import time
######## AI返答関係の設定 ########
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
chat_model = ChatOpenAI(api_key=OPENAI_API_KEY, model="gpt-4o-mini",stream=True) # type: ignore
######## Streamlitアプリの設定 ########
st.set_page_config(page_title="AIチャットアプリ。ストリーミングチャット", page_icon="🤖")
st.title("AIチャットアプリ")
######## StreamlitChatMessageHistoryの初期化 ########
chat_history = StreamlitChatMessageHistory(key="chat_messages")
chat_history.add_messages([
SystemMessage("ユーザーの入力に対して、日本語で返答してください")
])
######## 録音関係の設定 ########
import pyaudio
import wave
from pydub import AudioSegment
p = pyaudio.PyAudio() # PyAudioのインスタンス化
######## 音声翻訳関係の設定 ########
import openai
from tempfile import NamedTemporaryFile
client = openai.OpenAI()
# ハルっているか判定する関数
def hallcinated_transcription(t:str):
hallcination_texts = ['ご視聴ありがとう', '最後まで視聴','最後までご視聴','視聴してくださって','見てくれてありがとう', '本日はご覧いただき', 'おやすみなさい']
return any(phrase in t for phrase in hallcination_texts)
######## 録音のフラグ ########
if 'recording' not in st.session_state:
st.session_state.recording = False
######## 録音停止ボタン ########
if st.button("停止", key="stop_button"):
st.session_state.recording = False
######## カスタムコールバックハンドラ ########
class CustomStreamlitCallbackHandler(BaseCallbackHandler):
def __init__(self):
self.sentense:str = ""
self.sentenses:List[str] = []
def on_llm_new_token(self, token: str, **kwargs) -> None:
self.sentense += token
if token in ['!','！','?','？','、','。']:
self.sentenses.append(self.sentense)
self.sentense = ""
def on_llm_end(self, response: LLMResult, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs:Any):
# 再生中は録音しないようにstreamを停止する
stream.stop_stream()
# sentensesの内容を読み上げる
for sentense in self.sentenses:
duration = self.speech(sentense)
time.sleep(duration)
# 全て読み上げた後、stream録音を再開
stream.start_stream()
def speech(self,sentense:str):
with client.audio.speech.with_streaming_response.create(
model = "tts-1",
voice = "alloy",# 話者を入力、2024年2月現在の話者は「alloy", "echo", "fable", "onyx", "nova", "shimmer」
input = sentense,
response_format='mp3'
) as r:
with NamedTemporaryFile(suffix=".mp3") as temp_audio_file:
r.stream_to_file(temp_audio_file.name)
st.audio(temp_audio_file.name,autoplay=True)
audio = AudioSegment.from_mp3(temp_audio_file.name)
duration = len(audio) / 1000.0 # オーディオファイルの長さ
return duration
######## 録音開始ボタン ########
if st.button("おしゃべり", key="start_button"):
st.session_state.recording = True
st.write("何か話しかけてください")
chunk = 1024 # フレーム単位での音声の読み込みサイズ
sample_format = pyaudio.paInt16 # 16ビットの音声
channels = 1 # モノラル音声
rate = 16000 # サンプリングレート
record_seconds = 2
stream = p.open(format=sample_format,
channels=channels,
rate=rate,
input=True,
frames_per_buffer=chunk)
while st.session_state.recording:
user_input:str = ""
# ユーザーの音声をストリーム録音
while True:
data = stream.read(chunk, exception_on_overflow=False)
frames = []
for _ in range(0, int(rate / chunk * record_seconds)):
data = stream.read(chunk)
frames.append(data)
with NamedTemporaryFile(suffix='.wav') as temp_wav_file:
# waveを使って、framesをwavファイルとして書き出す
with wave.open(temp_wav_file, 'wb') as wf:
wf.setnchannels(channels)
wf.setsampwidth(p.get_sample_size(sample_format))
wf.setframerate(rate)
wf.writeframes(b''.join(frames))
# openaiのtranscriptionsを使って音声をテキスト変換
transcription = openai.audio.transcriptions.create(
file= open(temp_wav_file.name, "rb"),
model="whisper-1",
language="ja"
)
# 何も喋らないとハルシネーションを起こすので、ハルシネーションを起こしている=話し終えたと判断
if hallcinated_transcription(transcription.text):
break
user_input += transcription.text
# user_inputをchat_historyに追加
chat_history.add_user_message(HumanMessage(user_input))
# user_inputを表示する
with st.chat_message("human"):
st.markdown(user_input)
# ChatBotの返答を表示する
with st.chat_message("assistant"):
# ChatBotの返答をストリーム再生
stream_handler = CustomStreamlitCallbackHandler()
messages = [
HumanMessage(content=msg.content) if isinstance(msg, HumanMessage) else AIMessage(content=msg.content)
for msg in chat_history.messages
]
ai_response = chat_model.invoke(input=messages,config={"callbacks": [stream_handler]})
st.markdown(ai_response.content)
chat_history.add_ai_message(AIMessage(ai_response.content))
# トークが終わったら録音ストリームを閉じる
stream.stop_stream()
stream.close()
p.terminate()
情報
知り合いのエンジニアに、音声関係でお役立ち情報を教えてもらいましたので紹介します。
音声→テキスト
google cloud speech-to-text
かなり性能良さそう
openAIの新モデル
2024年秋にはopenAIが高性能な音声モデルをリリースするみたいです
テキスト→音声
voicebox
キャラクターボイス。ずんだもんの声ってここにあるんですね
voicebox nemo
ビジネス向けの音声。アニメ的なキンキン声が苦手な人に（実は私も苦手）
ゆいプラ
女性アナウンサーの声。温かみは感じませんが、ちょうど良いくらいのAIの声
style bert vits2
自分で音声モデルを作成できる。
hugging faceで個人が作成した音声モデルが配布されている。ただしロリコンが好みそうなアニメ声が多い
Discussion