分步指南：使用 OpenAI Realtime API 进行中断管理-Python教程-PHP中文网

分步指南：使用 OpenAI Realtime API 进行中断管理

Patricia Arquette

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

原创

405 人浏览过

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()

登录后复制

音频输出无需修改；它的运行方式如前面解释的代码中所述。