1. 流式传输
  2. 对话式聊天机器人

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

引言

下一代 AI 用户界面正朝着音频原生体验发展。用户将能够与聊天机器人对话，并接收到口头回应。GPT-4o 和 mini omni 等模型都是在此范例下构建的。

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

应用概述

我们的应用程序将提供以下用户体验

  1. 用户点击按钮开始录制消息
  2. 应用检测到用户停止说话时，停止录制
  3. 用户的音频被发送给 omni 模型，模型流式传回回应
  4. 在 omni mini 完成说话后，用户的麦克风重新激活
  5. 所有先前的口头音频（来自用户和 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

此函数有两个输入

  1. 当前的音频块（一个包含 (sampling_rate, numpy array of audio) 的元组）
  2. 当前的应用程序状态

我们将使用以下 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)

此函数执行以下操作

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

注意：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")
            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()

此设置创建了一个具有以下功能的UI

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

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

结论

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

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

←

从视频进行对象检测

实时语音识别

→
gradio