Azure AI の TTS / STT で gpt-4o-mini と会話する

自分のために少し手直ししています。

• コメントを日本語に変更
• 環境変数を .env ファイルから取得するように変更
• 認識する言語を日本語 (ja-jp) に変更

import os
import azure.cognitiveservices.speech as speechsdk
from openai import AzureOpenAI
from dotenv import load_dotenv

# .envファイルから環境変数を読み込みます
load_dotenv()

# この例では、"OPEN_AI_KEY"、"OPEN_AI_ENDPOINT"、および "OPEN_AI_DEPLOYMENT_NAME" という名前の環境変数が必要です
# エンドポイントは次のような形式である必要があります https://YOUR_OPEN_AI_RESOURCE_NAME.openai.azure.com/
client = AzureOpenAI(
azure_endpoint=os.environ.get('OPEN_AI_ENDPOINT'),
api_key=os.environ.get('OPEN_AI_KEY'),
api_version="2023-05-15"
)

# これは、モデルをデプロイしたときに選択したカスタム名に対応します。
deployment_id=os.environ.get('OPEN_AI_DEPLOYMENT_NAME')

# この例では、"SPEECH_KEY" と "SPEECH_REGION" という名前の環境変数が必要です
speech_config = speechsdk.SpeechConfig(subscription=os.environ.get('SPEECH_KEY'), region=os.environ.get('SPEECH_REGION'))
audio_output_config = speechsdk.audio.AudioOutputConfig(use_default_speaker=True)
audio_config = speechsdk.audio.AudioConfig(use_default_microphone=True)

# スピーカーの言語に合わせたロケールを設定する必要があります
speech_config.speech_recognition_language="ja-JP"
speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)

# Azure OpenAIが代わりに応答する音声の言語にします
speech_config.speech_synthesis_voice_name='en-US-JennyMultilingualNeural'
speech_synthesizer = speechsdk.SpeechSynthesizer(speech_config=speech_config, audio_config=audio_output_config)
# tts sentence end mark
tts_sentence_end = [ ".", "!", "?", ";", "。", "！", "？", "；", "\n" ]

# Azure OpenAIにリクエストを送信し、応答を合成します。
def ask_openai(prompt):
    # Azure OpenAIにストリーミングでリクエストを送信する
    response = client.chat.completions.create(model=deployment_id, max_tokens=200, stream=True, messages=[
        {"role": "user", "content": prompt}
    ])
    collected_messages = []
    last_tts_request = None

    # ストリームレスポンスをイテレートする
    for chunk in response:
        if len(chunk.choices) > 0:
            chunk_message = chunk.choices[0].delta.content  # extract the message
            if chunk_message is not None:
                collected_messages.append(chunk_message)  # save the message
                if chunk_message in tts_sentence_end: # sentence end found
                    text = ''.join(collected_messages).strip() # join the recieved message together to build a sentence
                    if text != '': # if sentence only have \n or space, we could skip
                        print(f"Speech synthesized to speaker for: {text}")
                        last_tts_request = speech_synthesizer.speak_text_async(text)
                        collected_messages.clear()
    if last_tts_request:
        last_tts_request.get()

# 継続的に音声入力をリッスンし、認識してテキストとしてAzure OpenAIに送信する
def chat_with_open_ai():
    while True:
        print("Azure OpenAI is listening. Say 'Stop' or press Ctrl-Z to end the conversation.")
        try:
            # マイクから音声を取得し、それをTTSサービスに送信します。
            speech_recognition_result = speech_recognizer.recognize_once_async().get()

            # 音声が認識された場合、それをAzure OpenAIに送信して応答を待ちます。
            if speech_recognition_result.reason == speechsdk.ResultReason.RecognizedSpeech:
                if speech_recognition_result.text == "Stop.": 
                    print("Conversation ended.")
                    break
                print("Recognized speech: {}".format(speech_recognition_result.text))
                ask_openai(speech_recognition_result.text)
            elif speech_recognition_result.reason == speechsdk.ResultReason.NoMatch:
                print("No speech could be recognized: {}".format(speech_recognition_result.no_match_details))
                break
            elif speech_recognition_result.reason == speechsdk.ResultReason.Canceled:
                cancellation_details = speech_recognition_result.cancellation_details
                print("Speech Recognition canceled: {}".format(cancellation_details.reason))
                if cancellation_details.reason == speechsdk.CancellationReason.Error:
                    print("Error details: {}".format(cancellation_details.error_details))
        except EOFError:
            break

# Main

try:
    chat_with_open_ai()
except Exception as err:
    print("Encountered exception. {}".format(err))

いろいろ導線はありますが、一例を示します。