使用 Gradio 构建对话式聊天机器人

简介

下一代 AI 用户界面正朝着音频原生体验发展。用户将能够与聊天机器人对话，并收到语音回复。在这种范式下构建了多个模型，包括 GPT-4o 和 mini omni。

在本指南中，我们将引导您使用 mini omni 作为示例构建您自己的对话式聊天应用程序。您可以在下面看到已完成的应用程序的演示

应用概览

我们的应用程序将实现以下用户体验

用户单击按钮开始录制消息
应用程序检测到用户何时说完话并停止录制
用户的音频被传递给 omni 模型，该模型流式传输回响应
在 omni mini 完成说话后，用户的麦克风将重新激活
来自用户和 omni 的所有先前语音音频都显示在聊天机器人组件中

让我们深入了解实现细节。

处理用户音频

我们将把用户的音频从他们的麦克风流式传输到服务器，并确定用户是否在每个新的音频块上停止说话。

这是我们的 process_audio 函数

import numpy as np
from utils import determine_pause

def process_audio(audio: tuple, state: AppState):
    if state.stream is None:
        state.stream = audio[1]
        state.sampling_rate = audio[0]
    else:
        state.stream = np.concatenate((state.stream, audio[1]))

    pause_detected = determine_pause(state.stream, state.sampling_rate, state)
    state.pause_detected = pause_detected

    if state.pause_detected and state.started_talking:
        return gr.Audio(recording=False), state
    return None, state

此函数接受两个输入

当前的音频块（(sampling_rate, numpy 音频数组) 的元组）
当前应用程序状态

我们将使用以下 AppState 数据类来管理我们的应用程序状态

from dataclasses import dataclass

@dataclass
class AppState:
    stream: np.ndarray | None = None
    sampling_rate: int = 0
    pause_detected: bool = False
    stopped: bool = False
    conversation: list = []

该函数将新的音频块连接到现有流，并检查用户是否已停止说话。如果检测到停顿，它将返回一个更新以停止录制。否则，它返回 None 以指示没有更改。

determine_pause 函数的实现特定于 omni-mini 项目，可以在这里找到。

生成回复

处理用户音频后，我们需要生成并流式传输聊天机器人的回复。这是我们的 response 函数

import io
import tempfile
from pydub import AudioSegment

def response(state: AppState):
    if not state.pause_detected and not state.started_talking:
        return None, AppState()
    
    audio_buffer = io.BytesIO()

    segment = AudioSegment(
        state.stream.tobytes(),
        frame_rate=state.sampling_rate,
        sample_width=state.stream.dtype.itemsize,
        channels=(1 if len(state.stream.shape) == 1 else state.stream.shape[1]),
    )
    segment.export(audio_buffer, format="wav")

    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        f.write(audio_buffer.getvalue())
    
    state.conversation.append({"role": "user",
                                "content": {"path": f.name,
                                "mime_type": "audio/wav"}})
    
    output_buffer = b""

    for mp3_bytes in speaking(audio_buffer.getvalue()):
        output_buffer += mp3_bytes
        yield mp3_bytes, state

    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
        f.write(output_buffer)
    
    state.conversation.append({"role": "assistant",
                    "content": {"path": f.name,
                                "mime_type": "audio/mp3"}})
    yield None, AppState(conversation=state.conversation)

此函数：

将用户的音频转换为 WAV 文件
将用户的消息添加到对话历史记录
使用 speaking 函数生成并流式传输聊天机器人的回复
将聊天机器人的回复保存为 MP3 文件
将聊天机器人的回复添加到对话历史记录

注意：speaking 函数的实现特定于 omni-mini 项目，可以在这里找到。

构建 Gradio 应用

现在让我们使用 Gradio 的 Blocks API 将它们整合在一起

import gradio as gr

def start_recording_user(state: AppState):
    if not state.stopped:
        return gr.Audio(recording=True)

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            input_audio = gr.Audio(
                label="Input Audio", sources="microphone", type="numpy"
            )
        with gr.Column():
            chatbot = gr.Chatbot(label="Conversation", type="messages")
            output_audio = gr.Audio(label="Output Audio", streaming=True, autoplay=True)
    state = gr.State(value=AppState())

    stream = input_audio.stream(
        process_audio,
        [input_audio, state],
        [input_audio, state],
        stream_every=0.5,
        time_limit=30,
    )
    respond = input_audio.stop_recording(
        response,
        [state],
        [output_audio, state]
    )
    respond.then(lambda s: s.conversation, [state], [chatbot])

    restart = output_audio.stop(
        start_recording_user,
        [state],
        [input_audio]
    )
    cancel = gr.Button("Stop Conversation", variant="stop")
    cancel.click(lambda: (AppState(stopped=True), gr.Audio(recording=False)), None,
                [state, input_audio], cancels=[respond, restart])

if __name__ == "__main__":
    demo.launch()

此设置创建一个用户界面，包含：

一个用于录制用户消息的输入音频组件
一个用于显示对话历史记录的聊天机器人组件
一个用于聊天机器人回复的输出音频组件
一个用于停止和重置对话的按钮

该应用程序以 0.5 秒的块流式传输用户音频，处理它，生成回复，并相应地更新对话历史记录。

结论

本指南演示了如何使用 Gradio 和 mini omni 模型构建对话式聊天机器人应用程序。您可以调整此框架以创建各种基于音频的聊天机器人演示。要查看完整应用程序的实际效果，请访问 Hugging Face Spaces 演示：https://hugging-face.cn/spaces/gradio/omni-mini

可以随意尝试不同的模型、音频处理技术或用户界面设计，来创造您自己独特的对话式 AI 体验！