分步指南：使用 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中文网其他相关文章！