対話型 AI と音声認識を使って雑談してみる
音声認識を使って OpenAI の対話型 AI 用モデルと音声で雑談してみるサンプル実装です。音声認識部分はWhisper Micの実装をほぼそのまま利用しています。また、マイクに話しかけた言葉を，リアルタイムに AI が認識（whisper, whisper_mic, Python を使用）（Windows 上）を参考に日本語対応としています。
発言が途中であるか否かの判定、音声認識で発生するであろう誤字脱字について考慮したプロンプトとしています。
前提条件
- langchain==0.0.135
- openai==0.27.4
- Whisper 関連はWhisper Micを参照してください。
モジュールの準備
Whisper 関連
以下のrequirements.txtをダウンロードします。
pip install -r requirements.txt
それ以外(LangChain など)
pip install openai
pip install langchain
環境変数の設定
以下の環境変数を事前に設定してください。
- OPENAI_API_KEY
ソースコード
import io
from pydub import AudioSegment
import speech_recognition as sr
import whisper
import queue
import tempfile
import os
import threading
import click
import torch
import numpy as np
from langchain.chat_models import ChatOpenAI
from langchain.prompts.chat import (
ChatPromptTemplate,
SystemMessagePromptTemplate,
MessagesPlaceholder,
HumanMessagePromptTemplate,
)
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from pprint import pprint
system_template = '''以下はHumanとあなたとの会話の内容です。あなたはHumanの友達であり、Humanがあなたに対して投げかけた質問に対し、以下の条件に従って回答してください。あなたはHumanの友達であるため、口調は砕けた感じとしてください。
## 条件
以下の内容を出力してください。それ以外は何も出力しないでください。
* HasContinue: Humanが続きを発言しようとしているか否かを推測してください。続きがある場合は`True`を出力してください。
* Predicted: `Humanの発言`は音声認識であるため誤字脱字があります。Humanの発言の誤字脱字を訂正した結果を出力してください。
* IsAnsweringReady: `HasContinue`が`True`の場合は`False`を出力してください。それ以外の場合で、あなたに対する質問や挨拶であると判断できる場合は`True`を出力してください。
* Response: `Predicted`に対する回答を表示してください。
'''
system_message_prompt = SystemMessagePromptTemplate.from_template(
system_template)
chat_prompt = ChatPromptTemplate.from_messages(
[system_message_prompt, MessagesPlaceholder(variable_name="history"), HumanMessagePromptTemplate.from_template("{input}")])
memory = ConversationBufferWindowMemory(return_messages=True, k=10)
chat = ChatOpenAI(temperature=0, max_tokens=256, client=None,
model_name="gpt-3.5-turbo", verbose=False)
conversation = ConversationChain(
memory=memory, prompt=chat_prompt, llm=chat, verbose=False)
def get_response(input):
pred = conversation.predict(input=input)
if len(pred) == 0:
return {}
ret = {}
# print(pred)
try:
for p in pred.split("\n"):
key, val = p.split(":")
ret[key.strip()] = val.strip()
except ValueError:
return {}
return ret
@click.command()
@click.option("--model", default="base", help="Model to use", type=click.Choice(["tiny", "base", "small", "medium", "large"]))
@click.option("--english", default=False, help="Whether to use English model", is_flag=True, type=bool)
@click.option("--verbose", default=False, help="Whether to print verbose output", is_flag=True, type=bool)
@click.option("--energy", default=300, help="Energy level for mic to detect", type=int)
@click.option("--dynamic_energy", default=False, is_flag=True, help="Flag to enable dynamic engergy", type=bool)
@click.option("--pause", default=0.8, help="Pause time before entry ends", type=float)
@click.option("--save_file", default=False, help="Flag to save file", is_flag=True, type=bool)
def main(model, english, verbose, energy, pause, dynamic_energy, save_file):
temp_dir = tempfile.mkdtemp() if save_file else None
# there are no english models for large
if model != "large" and english:
model = model + ".en"
audio_model = whisper.load_model(model)
audio_queue = queue.Queue()
result_queue = queue.Queue()
threading.Thread(target=record_audio,
args=(audio_queue, energy, pause, dynamic_energy, save_file, temp_dir)).start()
threading.Thread(target=transcribe_forever,
args=(audio_queue, result_queue, audio_model, english, verbose, save_file)).start()
while True:
say = result_queue.get()
print(say)
res = get_response(say)
if res.get('IsAnsweringReady', "False") == 'True': # type: ignore
print(f"AI: {res.get('Response', '')}") # type: ignore
else:
print(f"AI: [あなたの次の発言を待っています。]")
def record_audio(audio_queue, energy, pause, dynamic_energy, save_file, temp_dir):
# load the speech recognizer and set the initial energy threshold and pause threshold
r = sr.Recognizer()
r.energy_threshold = energy
r.pause_threshold = pause
r.dynamic_energy_threshold = dynamic_energy
# r.non_speaking_duration = 0.1
# r.phrase_threshold = 0.1
with sr.Microphone(sample_rate=16000) as source:
print("Say something!")
i = 0
while True:
# get and save audio to wav file
audio = r.listen(source)
if save_file:
data = io.BytesIO(audio.get_wav_data())
audio_clip = AudioSegment.from_file(data)
filename = os.path.join(temp_dir, f"temp{i}.wav")
audio_clip.export(filename, format="wav")
audio_data = filename
else:
torch_audio = torch.from_numpy(np.frombuffer(
audio.get_raw_data(), np.int16).flatten().astype(np.float32) / 32768.0)
audio_data = torch_audio
audio_queue.put_nowait(audio_data)
i += 1
def transcribe_forever(audio_queue, result_queue, audio_model, english, verbose, save_file):
while True:
audio_data = audio_queue.get()
if english:
result = audio_model.transcribe(audio_data, language='english')
else:
result = audio_model.transcribe(audio_data, language='japanese')
if not verbose:
predicted_text = result["text"]
result_queue.put_nowait("You said: " + predicted_text)
else:
result_queue.put_nowait(result)
if save_file:
os.remove(audio_data)
main()
実行方法
以下は Mac での実行例です。Model に medium を指定していますが、small でも機能します。--pauseを短め(0.8->0.5)にすることでスムーズに会話できるようにしていますが、音声認識処理の方が長いため、気持ち程度です。
python conversation-by-mic.py --model medium --pause 0.5 2> /dev/null
実行時にエラーになった場合は以下を参照してみてください。