逐步指南：使用 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()

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

結論

這次我介紹了一個用於對話中斷的Python實作。

我希望這篇文章對任何像我一樣面臨有效阻止人工智慧語音挑戰的人有所幫助。

此外，流實例的定義和配置會影響音訊播放的品質。如果您遇到音訊播放中斷的情況，檢查這些設定可能有助於改善情況。

感謝您閱讀到最後。

以上是逐步指南：使用 OpenAI Realtime API 進行中斷管理的詳細內容。更多資訊請關注PHP中文網其他相關文章！