Gradio全解系列——Additional Features：附加功能（上）

前言
本篇摘要
10. Additional Features：附加功能
- 10.1 队列
- - 10.1.1 使用方法
  - 10.1.2 配置队列
- 10.2 流输入输出
- - 10.2.1 流输出
  - - 1. 生成器yield
    - 2. 流媒体
  - 10.2.2 流输入
  - - 1. 流事件
    - 2. 图像滤镜
  - 10.2.3 统一的流输入输出
  - 10.2.4 跟踪过去的输入或输出
- 10.3 提示及进度条
- - 10.3.1 提示
  - 10.3.2 进度条
- 参考文献

前言

本系列文章主要介绍WEB界面工具Gradio。Gradio是Hugging Face发布的一个简易的webui开发框架，它基于FastAPI和svelte，便于部署人工智能相关模型，是当前热门的非常易于开发和展示机器学习大语言模型LLM及扩散模型DM的UI框架。本系列文章分为前置概念和实战演练两部分。前置概念先介绍Gradio的详细技术架构、历史、应用场景、与其他框架Gradio/NiceGui/StreamLit/Dash/PyWebIO的区别，然后详细介绍了大模型及数据的资源网站Hugging Face，包括三种资源models/datasets/spaces、六类开源库transformers/diffusers/datasets/PEFT/accelerate/optimum实战及Course课程资源。实战演练部分先讲解了多种不同的安装、运行和部署方式，安装包括Linux/Win/Mac三种安装方式，运行包括普通方式运行和热重载方式运行两种运行方式，部署包括本地部署、HuggingFace托管、FastAPI挂载和Gradio-Lite浏览器集成；然后按照先整体再细节的逻辑，讲解Gradio的多种高级特性：三种Gradio Clients（python/javascript/curl）、Gradio Tools等，方便读者对Gradio整体把握；最后深入细节，也是本系列文章的核心，实践基础功能Interface、Blocks和Additional Features，高级功能Chatbots、Data Science And Plots和Streaming。本系列文章注解详细，代码均可运行并附有大量运行截图，方便读者理解，Gradio一定会成为每个技术人员实现奇思妙想的最称手工具。

本系列文章目录如下：

《Gradio全解系列1——Gradio简介》
《Gradio全解系列1——Gradio的安装与运行》
《Gradio全解系列2——Gradio的3+1种部署方式实践》
《Gradio全解系列2——浏览器集成Gradio-Lite》
《Gradio全解系列3——Gradio Client：python客户端》
《Gradio全解系列3——Gradio Client：javascript客户端》
《Gradio全解系列3——Gradio Client：curl客户端》
《Gradio全解系列4——Gradio Tools：将Gradio用于LLM Agents》
《Gradio全解系列5——Gradio库的模块架构和环境变量》
《Gradio全解系列6——Interface：高级抽象界面类（上）》
《Gradio全解系列6——Interface：高级抽象界面类（下）》
《Gradio全解系列7——Blocks：底层区块类（上）》
《Gradio全解系列7——Blocks：底层区块类（下）》
《Gradio全解系列8——Additional Features：附加功能（上）》
《Gradio全解系列8——Additional Features：附加功能（下）》
《Gradio全解系列9——Chatbots：聊天机器人（上）》
《Gradio全解系列9——Chatbots：聊天机器人（下）》
《Gradio全解系列10——Data Science And Plots：数据科学与绘图》
《Gradio全解系列11——Streaming：数据流（上）》
《Gradio全解系列11——Streaming：数据流（下）》

本篇摘要

本篇介绍Gradio的附加功能，包括队列、出入数据流、提示信息及进度显示、批处理函数、安全访问文件和资源清理，下面逐一讲述。

10. Additional Features：附加功能

本章介绍Gradio的附加功能，包括队列、出入数据流、提示信息及进度显示、批处理函数、安全访问文件和资源清理，下面逐一讲述。

10.1 队列

每个Gradio程序提供了一个内置的队列系统，可以处理数千个并发用户。由于许多事件监听器可能涉及繁重的处理任务，Gradio 会自动为每个事件监听器创建一个队列来处理后端的事件，因此每个事件监听器都会自动拥有一个队列来处理传入的事件。

10.1.1 使用方法

如果函数推理时间较长，比如目标检测；或者应用程序处理流量过大，则需要使用queue方法进行排队。通过启用队列，可以控制用户在队列中的位置，queue方法使用websockets，可以防止网络超时。

队列使用方式如下：

demo = gr.Interface(...).queue()
demo.launch()
#或
with gr.Blocks() as demo:
    #...
demo.queue()
demo.launch()

以Blocks为例，演示如下：

with gr.Blocks() as demo:
    button = gr.Button(label="Generate Image")
    button.click(fn=image_generator, inputs=gr.Textbox(), outputs=gr.Image())
demo.queue(max_size=10)
demo.launch()

10.1.2 配置队列

默认情况下，每个事件监听器都有自己的队列，一次处理一个请求。可以通过事件监听器的两个参数进行配置：

concurrency_limit：设置事件监听器的最大并发执行数。默认情况下限制为 1，除非在 Blocks.queue() 中另行配置。你也可以将其设置为 None 以表示无限制（即无限数量的并发执行）；
concurrency_id：允许事件监听器通过分配相同的 ID 来共享队列。当使用共享队列管理多个事件监听器时，使用concurrency_id指定队列，例如如果你的设置中只有 2 个 GPU，但多个函数需要 GPU 访问，你可以为所有这些函数创建一个共享队列。

示例如下：

import gradio as gr

with gr.Blocks() as demo:
    prompt = gr.Textbox()
    image = gr.Image()
    generate_btn_1 = gr.Button("Generate Image via model 1")
    generate_btn_2 = gr.Button("Generate Image via model 2")
    generate_btn_3 = gr.Button("Generate Image via model 3")
    generate_btn_1.click(image_gen_1, prompt, image, concurrency_limit=2, concurrency_id="gpu_queue")
    generate_btn_2.click(image_gen_2, prompt, image, concurrency_id="gpu_queue")
    generate_btn_3.click(image_gen_3, prompt, image, concurrency_id="gpu_queue")

在这个例子中，所有三个事件监听器共享一个标识为 “gpu_queue” 的队列。该队列最多可以同时处理 2 个并发请求，额外的请求将被排队，直到有可用的槽位。这些配置使得管理Gradio队列行为变得非常容易。
注意事项：

要确保事件监听器的并发无限制，可以将 concurrency_limit设置为None，这在你的函数调用外部 API（该API自行处理请求的速率限制）时非常有用。
所有队列的默认并发限制可以使用Blocks.queue()中的default_concurrency_limit参数全局设置。

10.2 流输入输出

10.2.1 流输出

在某些情况下，我们需要流式输出一系列结果，而不是一次性显示单个输出。例如，某个图像生成模型希望显示每一步生成的图像，直到最终图像生成；或者某个聊天机器人，它逐词流式输出响应，而不是一次性返回所有内容。在这种情况下，可以向 Gradio 提供一个生成器函数。

1. 生成器yield

在Python 中创建生成器非常简单：该函数不是返回单个值，而是生成一系列值。通常yield 语句会放在某种循环中，可以像提供常规函数一样向 Gradio 提供生成器。

例如，以下是一个模拟的图像生成模型，它在输出图像之前生成若干步噪声，使用 gr.Interface 类进行演示：

import gradio as gr
import numpy as np
import time

def fake_diffusion(steps):
    rng = np.random.default_rng()
    for i in range(steps):
        time.sleep(1)
        image = rng.random(size=(600, 600, 3))
        yield image
    image = np.ones((1000,1000,3), np.uint8)
    image[:] = [255, 124, 0]
    yield image

demo = gr.Interface(fake_diffusion,
                    inputs=gr.Slider(1, 10, 3, step=1),
                    outputs="image")

demo.launch()

运行截图如下：
在这里插入图片描述
请注意，我们在迭代器中添加了 time.sleep(1)，它在步骤之间创建一个人为的暂停，这样就能够观察到迭代器的每一步（在真实的图像生成模型中，这可能是不必要的）。

同样，Gradio 可以处理流式输入，例如，每当用户在文本框中输入一个字母时，图像生成模型都会重新运行。这在关于构建响应式界面的中有更详细的介绍，请参考Interface章节中关于实时Interface的讲解。

2. 流媒体

Gradio 可以直接从生成器函数中流式传输音频和视频，这让用户几乎在函数生成音频或视频的同时就能听到或看到。实现流式媒体只需执行以下操作：

在 gr.Audio 或 gr.Video 输出组件中设置 streaming=True；
编写一个 Python 生成器，生成下一个音频或视频的“chunk”块；
设置 autoplay=True，以便媒体自动开始播放。

对于音频，下一个“块”可以是 .mp3 或 .wav 文件，也可以是音频的字节序列。对于视频，下一个“块”必须是 .mp4 文件或使用 h.264 编解码器且扩展名为 .ts 的文件。为了确保流畅播放，请确保每个块的长度一致且大于 1 秒。

我们将通过一些简单的示例来说明这些要点，以音频为例：

import gradio as gr
from time import sleep

def keep_repeating(audio_file):
    for _ in range(10):
        sleep(0.5)
        yield audio_file

gr.Interface(keep_repeating,
             gr.Audio(sources=["microphone"], type="filepath"),
             gr.Audio(streaming=True, autoplay=True)
).launch()

运行后点击submit，输出与输入同步播放，截图如下：
在这里插入图片描述
同样也可以视频为例，只需将gr.Audio替换为gr.Video即可：gr.Video(sources=["webcam"], format="mp4")，不再重复。关于流式的端到端示例请参考后续的数据流章节。

10.2.2 流输入

在上一小节中，我们介绍了如何从事件处理程序中流式输出一系列结果。Gradio还可以将用户摄像头中的图像或麦克风中的音频块流式传输到事件处理程序中，这可以用于创建实时对象检测应用程序或使用Gradio构建对话式聊天应用。

1. 流事件

目前，gr.Image 和 gr.Audio 组件支持通过 stream 事件实现流式输入传输。比如创建一个最简单的流式应用程序，直接返回未修改的网络摄像头流：

import gradio as gr

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            input_img = gr.Image(label="Input", sources="webcam")
        with gr.Column():
            output_img = gr.Image(label="Output")
        input_img.stream(lambda s: s, input_img, output_img, time_limit=15, stream_every=0.1, concurrency_limit=30)

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

运行截图如下：
在这里插入图片描述
可以自己运行看一下效果。当用户开始录制时，stream 事件会被触发。底层逻辑是，网络摄像头每 0.1 秒拍摄一张照片并将其发送到服务器，然后服务器会返回该图像。

stream 事件有两个独特的关键字参数：

time_limit：这是 Gradio 服务器处理事件的时间限制。媒体流本质上是无限制的，因此设置时间限制非常重要，以防止一个用户长期占用 Gradio 队列。时间限制仅计算处理流所花费的时间，不包括在队列中等待的时间。输入图像底部显示的橙色条表示剩余时间。当时间限制到期时，用户将自动重新加入队列。
stream_every：这是流捕获输入并将其发送到服务器的频率（以秒为单位）。对于图像检测或处理等演示，设置较小的值可以实现“real-time”实时效果。对于语音转录等演示，较高的值更有用，因为可以使转录算法可以更好地理解上下文。

2. 图像滤镜

让我们创建一个滤镜演示，用户可以选择应用于其网络摄像头流的滤镜。用户可以选择边缘检测滤镜、卡通滤镜，或者简单地垂直翻转流。代码如下：

import gradio as gr
import numpy as np
import cv2

def transform_cv2(frame, transform):
    if transform == "cartoon":
        # prepare color
        img_color = cv2.pyrDown(cv2.pyrDown(frame))
        for _ in range(6):
            img_color = cv2.bilateralFilter(img_color, 9, 9, 7)
        img_color = cv2.pyrUp(cv2.pyrUp(img_color))

        # prepare edges
        img_edges = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
        img_edges = cv2.adaptiveThreshold(
            cv2.medianBlur(img_edges, 7),
            255,
            cv2.ADAPTIVE_THRESH_MEAN_C,
            cv2.THRESH_BINARY,
            9,
            2,
        )
        img_edges = cv2.cvtColor(img_edges, cv2.COLOR_GRAY2RGB)
        # combine color and edges
        img = cv2.bitwise_and(img_color, img_edges)
        return img
    elif transform == "edges":
        # perform edge detection
        img = cv2.cvtColor(cv2.Canny(frame, 100, 200), cv2.COLOR_GRAY2BGR)
        return img
    else:
        return np.flipud(frame)

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            transform = gr.Dropdown(choices=["cartoon", "edges", "flip"],
                                    value="flip", label="Transformation")
            input_img = gr.Image(sources=["webcam"], type="numpy")
        with gr.Column():
            output_img = gr.Image(streaming=True)
        dep = input_img.stream(transform_cv2, [input_img, transform], [output_img],
                                time_limit=30, stream_every=0.1, concurrency_limit=30)

demo.launch()

在这里插入图片描述

请注意，当更改滤镜值时，它会立即在输出流中生效。这是流事件与其他 Gradio 事件的一个重要区别：在处理流的过程中，可以更改流的输入值并立即生效。

提示：我们将图像输出组件的 streaming 参数设置为True，这样可以让服务器自动将输出图像转换为 base64 格式，这是一种适合流高效传输的格式。

10.2.3 统一的流输入输出

对于一些图像流式传输演示（如上面的示例），我们不需要分别显示输入和输出组件，只显示修改后的输出流，应用程序看起来会更简洁。

我们可以通过将输入图像组件指定为流事件的输出来实现这一点，省略重复代码，核心代码如下：

css=""".my-group {max-width: 500px !important; max-height: 500px !important;}
            .my-column {display: flex !important; justify-content: center !important; align-items: center !important};"""

with gr.Blocks(css=css) as demo:
    with gr.Column(elem_classes=["my-column"]):
        with gr.Group(elem_classes=["my-group"]):
            transform = gr.Dropdown(choices=["cartoon", "edges", "flip"],
                                    value="flip", label="Transformation")
            input_img = gr.Image(sources=["webcam"], type="numpy", streaming=True)
    input_img.stream(transform_cv2, [input_img, transform], [input_img], time_limit=30, stream_every=0.1)

demo.launch()

运行截图如下：
在这里插入图片描述

10.2.4 跟踪过去的输入或输出

通常流式函数应该是无状态的，它应该接受当前输入并返回相应的输出。然而在某些情况下，可能希望跟踪过去的输入或输出。例如可能希望在缓冲区保留前 k 个输入，以提高转录演示的准确性。可以使用 Gradio 的 gr.State()组件来实现这一点，示例如下：

def transcribe_handler(current_audio, state, transcript):
    next_text = transcribe(current_audio, history=state)
    state.append(current_audio)
    # 保留前3个输入
    state = state[-3:]
    return state, transcript + next_text

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            mic = gr.Audio(sources="microphone")
            state = gr.State(value=[])
        with gr.Column():
            transcript = gr.Textbox(label="Transcript")
    mic.stream(transcribe_handler, [mic, state, transcript], [state, transcript],
               time_limit=10, stream_every=1)

demo.launch()

10.3 提示及进度条

10.3.1 提示

提示组件分为三类：gr.Error()、 gr.Warning() 和gr.Info() 。我们希望向用户显示提示错误信息，为此可以在函数中抛出 gr.Error(“自定义消息”) ，这时函数停止执行并向用户显示错误信息。

我们还可以通过在函数中单独使用 gr.Warning(“自定义消息”) 或 gr.Info(“自定义消息”) 来立即显示模态框，同时继续执行函数。gr.Info() 和 gr.Warning() 之间的唯一区别是提示框的颜色。演示如下：

import gradio as gr
from functools import partial

with gr.Blocks() as demo:
    with gr.Row():
        duration = gr.Number(label="Duration", info="Set to -1 for infinite", value=10, minimum=-1)
    with gr.Row():
        error = gr.Button("Error")
        info = gr.Button("Info")
        warning = gr.Button("Warning")

    def display_message(type, msg, duration):
        duration = None if duration < 0 else duration
        if type == "error":
            raise gr.Error(msg, duration=duration)
        elif type == "info":
            gr.Info(msg, duration=duration)
        elif type == "warning":
            gr.Warning(msg,  duration=duration)

    error.click(partial(display_message, "error", "ERROR 💥"),  [duration])
    info.click(partial(display_message, "info", "INFO ℹ️"), [duration])
    warning.click(partial(display_message, "warning", "WARNING ⚠️"), [duration])

运行结果如下：
在这里插入图片描述
提示：请注意gr.Error()是一个必须引发的异常，而gr.Warning()和gr.Info()是直接调用的函数。

10.3.2 进度条

Gradio 支持创建自定义进度条，可以自定义和控制进度更新，以便向用户展示。要启用此功能，只需在方法中添加一个参数，该参数的默认值为 gr.Progress 实例；然后通过直接调用此实例并传入一个介于 0 和 1 之间的浮点数来更新进度，或者使用 Progress 实例的 tqdm() 方法来跟踪可迭代对象的进度，如下所示：

import gradio as gr
import time

def slowly_reverse(word, progress=gr.Progress()):
    progress(0, desc="Starting")
    time.sleep(1)
    progress(0.05)
    new_string = ""
    for letter in progress.tqdm(word, desc="Reversing"):
        time.sleep(0.25)
        new_string = letter + new_string  
    return new_string

demo = gr.Interface(slowly_reverse, gr.Text(), gr.Text())
demo.launch()