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

简介

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

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

应用概述

我们的应用将支持以下用户体验：

用户点击按钮开始录制消息
应用检测到用户说完话后停止录制
用户的音频被传递给 omni 模型，该模型流式传输回响应
在 mini omni 说完话后，用户的麦克风将重新激活
所有先前的语音音频，包括用户和 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

此函数接收两个输入：

当前音频块（一个包含 (采样率, 音频 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 体验！