逐步指南：使用 OpenAI Realtime API 進行中斷管理-Python教學-PHP中文網

逐步指南：使用 OpenAI Realtime API 進行中斷管理

Patricia Arquette

發布： 2024-10-10 10:35:30

原創

513 人瀏覽過

Step-by-Step Guide: Interruption Management with OpenAI Realtime API

本文介紹如何使用 OpenAI Realtime API 實作對話中斷功能。

實現的詳細資訊可在 GitHub 儲存庫中找到。

此實作基於 Azure-Samples/aoai-realtime-audio-sdk 中的程式碼。程式碼的詳細解釋可以在這篇文章中找到。

音訊輸入

在此實作中，我們使用本地PC的麥克風和揚聲器進行音訊輸入和輸出。

從麥克風捕獲的音訊被傳送到 OpenAI Realtime API 伺服器進行處理。

為了從本地 PC 的麥克風捕獲音頻，我們使用 pyaudio 庫的流功能。以下程式碼設定音訊輸入流：

p = pyaudio.PyAudio()
input_default_input_index = p.get_default_input_device_info()['index']
input_stream = p.open(
    format=STREAM_FORMAT,
    channels=INPUT_CHANNELS,
    rate=INPUT_SAMPLE_RATE,
    input=True,
    output=False,
    frames_per_buffer=INPUT_CHUNK_SIZE,
    input_device_index=input_default_input_index,
    start=False,
)
input_stream.start_stream()

登入後複製

音訊擷取是使用threading.Thread進行並行處理的。從麥克風獲取的音訊資料被編碼為base64格式並儲存在佇列中。

def listen_audio(input_stream: pyaudio.Stream):
    while True:
        audio_data = input_stream.read(INPUT_CHUNK_SIZE, exception_on_overflow=False)
        if audio_data is None:
            continue
        base64_audio = base64.b64encode(audio_data).decode("utf-8")
        audio_input_queue.put(base64_audio)

threading.Thread(target=listen_audio, args=(input_stream,), daemon=True).start()

登入後複製

儲存在佇列中的 Base64 字串將作為「input_audio_buffer.append」訊息傳送到 OpenAI Realtime API 伺服器。

async def send_audio(client: RTLowLevelClient):
    while not client.closed:
        base64_audio = await asyncio.get_event_loop().run_in_executor(None, audio_input_queue.get)
        await client.send(InputAudioBufferAppendMessage(audio=base64_audio))
        await asyncio.sleep(0)

登入後複製

音訊輸出

使用從 OpenAI Realtime API 伺服器接收的音訊數據，透過本地 PC 的揚聲器進行音訊播放。

音訊資料會以「response.audio.delta」訊息從伺服器接收。由於接收到的資料是以base64編碼的，因此它被解碼，儲存在佇列中，並轉換為可播放的格式。

async def receive_messages(client: RTLowLevelClient):
    while True:
        message = await client.recv()
        if message is None:
            continue
        match message.type:
            case "response.audio.delta":
                audio_data = base64.b64decode(message.delta)
                for i in range(0, len(audio_data), OUTPUT_CHUNK_SIZE):
                    audio_output_queue.put(audio_data[i:i+OUTPUT_CHUNK_SIZE])
                await asyncio.sleep(0)

登入後複製

儲存在佇列中的資料透過本機 PC 的揚聲器使用平行處理進行播放。這個播放過程使用了threading.Thread來確保音訊資料即時流暢播放。

def play_audio(output_stream: pyaudio.Stream):
    while True:
        audio_data = audio_output_queue.get()
        output_stream.write(audio_data)

p = pyaudio.PyAudio()
output_default_output_index = p.get_default_output_device_info()['index']
output_stream = p.open(
    format=STREAM_FORMAT,
    channels=OUTPUT_CHANNELS,
    rate=OUTPUT_SAMPLE_RATE,
    input=False,
    output=True,
    frames_per_buffer=OUTPUT_CHUNK_SIZE,
    output_device_index=output_default_output_index,
    start=False,
)
output_stream.start_stream()

threading.Thread(target=play_audio, args=(output_stream,), daemon=True).start()

登入後複製

會話中斷處理

OpenAI Realtime API 可自動偵測伺服器端的對話片段。即使人工智慧正在響應，這也可以檢測新語音並創建即時響應。

但是，在本地 PC 上播放音訊時，停止正在播放的音訊以實現對話的自然中斷非常重要。這一點需要注意。使用者語音的偵測是從 OpenAI Realtime API 伺服器作為「input_audio_buffer.speech_started」訊息接收的。收到此訊息後，將透過清除佇列中儲存的音訊資料來停止播放。

async def receive_messages(client: RTLowLevelClient):
    while True:
        message = await client.recv()
        # print(f"{message=}")
        if message is None:
            continue
        match message.type:
            case "input_audio_buffer.speech_started":
                print("Input Audio Buffer Speech Started Message")
                print(f"  Item Id: {message.item_id}")
                print(f"  Audio Start [ms]: {message.audio_start_ms}")
                while not audio_output_queue.empty():
                    audio_output_queue.get()

登入後複製

音訊輸出無需修改；它的運作方式如前面解釋的程式碼所述。