流式AI生成音频

在本指南中，我们将构建一个新颖的 AI 应用程序，以展示 Gradio 的音频输出流。我们将构建一个会说话的魔法 8 号球 🎱

魔法 8 号球是一种玩具，摇晃后会回答任何问题。我们的应用程序将实现相同的功能，但它还会将回答说出来！

本博客文章不会涵盖所有实现细节，但代码可在 Hugging Face Spaces 上免费获取。

概述

就像经典的魔法 8 号球一样，用户应该口头提问，然后等待回答。在后台，我们将使用 Whisper 转录音频，然后使用大型语言模型 (LLM) 生成一个魔法 8 号球式的回答。最后，我们将使用 Parler TTS 将回答朗读出来。

用户界面

首先，让我们定义用户界面，并为所有 Python 逻辑设置占位符。

import gradio as gr

with gr.Blocks() as block:
    gr.HTML(
        f"""
        <h1 style='text-align: center;'> Magic 8 Ball 🎱 </h1>
        <h3 style='text-align: center;'> Ask a question and receive wisdom </h3>
        <p style='text-align: center;'> Powered by <a href="https://github.com/huggingface/parler-tts"> Parler-TTS</a>
        """
    )
    with gr.Group():
        with gr.Row():
            audio_out = gr.Audio(label="Spoken Answer", streaming=True, autoplay=True)
            answer = gr.Textbox(label="Answer")
            state = gr.State()
        with gr.Row():
            audio_in = gr.Audio(label="Speak your question", sources="microphone", type="filepath")

    audio_in.stop_recording(generate_response, audio_in, [state, answer, audio_out])\
        .then(fn=read_response, inputs=state, outputs=[answer, audio_out])

block.launch()

我们将输出音频和文本框组件以及输入音频组件放在不同的行中。为了从服务器流式传输音频，我们将在输出音频组件中设置 `streaming=True`。我们还将设置 `autoplay=True`，以便音频在准备好后立即播放。我们将使用音频输入组件的 `stop_recording` 事件，在用户停止录制麦克风时触发我们应用程序的逻辑。

我们将逻辑分为两部分。首先，`generate_response` 将接收录制的音频，对其进行转录，并使用大型语言模型 (LLM) 生成回答。我们将把回答存储在一个 `gr.State` 变量中，然后将其传递给生成音频的 `read_response` 函数。

我们分两部分进行，因为只有 `read_response` 需要 GPU。我们的应用程序将在 Hugging Face 的 ZeroGPU 上运行，该平台有基于时间的配额。由于生成回答可以使用 Hugging Face 的推理 API 完成，我们不应将该代码包含在我们的 GPU 函数中，因为它会不必要地占用我们的 GPU 配额。

逻辑

如上所述，我们将使用 Hugging Face 的推理 API 转录音频并从大型语言模型 (LLM) 生成回答。实例化客户端后，我使用 `automatic_speech_recognition` 方法（这会自动使用运行在 Hugging Face 推理服务器上的 Whisper）来转录音频。然后我将问题传递给一个 LLM（Mistal-7B-Instruct）来生成回答。我们通过系统消息提示 LLM 表现得像一个魔法 8 号球。

我们的 `generate_response` 函数还将向输出文本框和音频组件发送空更新（返回 `None`）。这是因为我希望 Gradio 进度跟踪器显示在组件上方，但我不希望在音频准备好之前显示回答。

from huggingface_hub import InferenceClient

client = InferenceClient(token=os.getenv("HF_TOKEN"))

def generate_response(audio):
    gr.Info("Transcribing Audio", duration=5)
    question = client.automatic_speech_recognition(audio).text

    messages = [{"role": "system", "content": ("You are a magic 8 ball."
                                              "Someone will present to you a situation or question and your job "
                                              "is to answer with a cryptic adage or proverb such as "
                                              "'curiosity killed the cat' or 'The early bird gets the worm'."
                                              "Keep your answers short and do not include the phrase 'Magic 8 Ball' in your response. If the question does not make sense or is off-topic, say 'Foolish questions get foolish answers.'"
                                              "For example, 'Magic 8 Ball, should I get a dog?', 'A dog is ready for you but are you ready for the dog?'")},
                {"role": "user", "content": f"Magic 8 Ball please answer this question -  {question}"}]
    
    response = client.chat_completion(messages, max_tokens=64, seed=random.randint(1, 5000),
                                      model="mistralai/Mistral-7B-Instruct-v0.3")

    response = response.choices[0].message.content.replace("Magic 8 Ball", "").replace(":", "")
    return response, None, None

现在我们有了文本回答，我们将使用 Parler TTS 将其朗读出来。`read_response` 函数将是一个 Python 生成器，它会在音频准备好时逐块生成。

我们将使用 Mini v0.1 进行特征提取，但使用 Jenny 微调版本用于语音。这是为了确保语音在不同生成中保持一致。

使用 Transformer 进行音频流式传输需要一个自定义的 Streamer 类。你可以在这里查看实现。此外，我们还会将输出转换为字节，以便后端可以更快地流式传输。

from streamer import ParlerTTSStreamer
from transformers import AutoTokenizer, AutoFeatureExtractor, set_seed
import numpy as np
import spaces
import torch
from threading import Thread


device = "cuda:0" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
torch_dtype = torch.float16 if device != "cpu" else torch.float32

repo_id = "parler-tts/parler_tts_mini_v0.1"

jenny_repo_id = "ylacombe/parler-tts-mini-jenny-30H"

model = ParlerTTSForConditionalGeneration.from_pretrained(
    jenny_repo_id, torch_dtype=torch_dtype, low_cpu_mem_usage=True
).to(device)

tokenizer = AutoTokenizer.from_pretrained(repo_id)
feature_extractor = AutoFeatureExtractor.from_pretrained(repo_id)

sampling_rate = model.audio_encoder.config.sampling_rate
frame_rate = model.audio_encoder.config.frame_rate

@spaces.GPU
def read_response(answer):

    play_steps_in_s = 2.0
    play_steps = int(frame_rate * play_steps_in_s)

    description = "Jenny speaks at an average pace with a calm delivery in a very confined sounding environment with clear audio quality."
    description_tokens = tokenizer(description, return_tensors="pt").to(device)

    streamer = ParlerTTSStreamer(model, device=device, play_steps=play_steps)
    prompt = tokenizer(answer, return_tensors="pt").to(device)

    generation_kwargs = dict(
        input_ids=description_tokens.input_ids,
        prompt_input_ids=prompt.input_ids,
        streamer=streamer,
        do_sample=True,
        temperature=1.0,
        min_new_tokens=10,
    )

    set_seed(42)
    thread = Thread(target=model.generate, kwargs=generation_kwargs)
    thread.start()

    for new_audio in streamer:
        print(f"Sample of length: {round(new_audio.shape[0] / sampling_rate, 2)} seconds")
        yield answer, numpy_to_mp3(new_audio, sampling_rate=sampling_rate)

结论

你可以在这里查看我们的最终应用程序！