用事件响应实现的（伪）加载进度条

2024-02-09 11:48

作为一名全栈 AI 开发工程师，分享一个我最近遇到的难题。

我做了一个 ChatGPT + 语音 + 视频合成的项目。就是有一张脸会把 ChatGPT 的回复内容读出来。逻辑就是在后端拿到 ChatGPT 的回复文本，先调用谷歌的 TextToSpeech 合成语音，再通过一个嘴型生成模型生成视频，在前端播放。所以没什么技术，就是调用 API + 运行别人预训练的一个神经网络。前端只有一个网页，用 Vue 做的。

做好之后就放到了网上，在一个叫 HuggingFace 的网站的免费虚拟机上跑了起来。这台 2 vCPU 16G 内存没有显卡的虚拟机跑这个 36M 参数的神经网络十分勉强。后端收到请求，花几秒等 ChatGPT 回复，花二十来秒等 TextToSpeech 回复，再花两分钟等神经网络生成视频。用户发完消息要枯等两三分钟，看着一个加载提示图标转转转，体验D差。

图标转转转

归根到底还是穷，也没有优化的空间了。（什么用付费档？不存在的！）

作为一个注重用户体验的前端开发工程师，我灵机一动。

我决定把图标转转转拿掉，换成进度条。

进度条

我没有办法不让用户等，但我可以提示用户大致等待时间。更关键，图标转转转的时候你心里没底，不知道要等到什么时候，你就会想：也许连接已经断了，再也等不到了，只是你不知道。但是进度条就不会，你看到那个条条在动，就会觉得网页还在正常工作，就更可能多等一会儿等它走完。这就是消费者的心理。

所以我就有了这样一个需求：

当用户发出消息请求，我需要先返回一个大致的等待时间，让进度条跑起来；等到工作完成后再返回真正的响应。这个等待时间不需要太精确，我觉得误差在十秒以内都不算离谱。如果进度条到头还没有返回响应，它就会卡在 99% 继续等。

接下来我就记录一下在实现这个需求的过程中学到的知识。上面的部分都是铺垫。如果你对解决延迟或者提高用户体验有不同的看法，这些不是本文的讨论范围。本文就是关于如何为一个 http 请求返回两个、多个响应，以完成不同的工作。

方案 0：普通的 POST 请求

在实现这个需求之前，这只是一个正常的 POST 请求，包含了诸多数据和参数。从用户按下发送按钮到视频开始播放，经过的时间是：网络传输时间 x 2 + ChatGPT 回复时间 + TextToSpeech 回复时间 + 视频生成时间。

这个方案功能上没有问题，就是延迟太长，用户体验太差。

方案1：Server-Sent Events

Server-Sent Events 是由服务端向客户端主动发事件。客户端通过一个普通的 HTTP 请求与服务器建立连接，然后服务器保持这个连接打开，并通过这个连接不断地向客户端发送数据。最后从客户端关闭连接。

后端代码是这样：

from flask import Response

def generate():
    # 生成文本……
    # 基于文本长度估算延迟……
    yield f"data: {json.dumps({'status': 'text_ok', 'data': '...'})}\n\n"

    # 生成音频……
    # 基于音频长度估算更精确的延迟……
    yield f"data: {json.dumps({'status': 'audio_ok', 'data': '...'})}\n\n"

    # 生成视频……
    yield f"data: {json.dumps({'status': 'video_ok', 'data': '...'})}\n\n"

@app.route('/video_message')
def video_message_sse():
    return Response(generate(), content_type='text/event-stream')

前端代码是这样：

<script>
    const eventSource = new EventSource('/video_message');

    eventSource.onmessage = function(event) {
        const messageData = JSON.parse(event.data);

        switch (messageData.status) {
            case 'text_ok':
                // 显示进度条……
                break;
            case 'audio_ok':
                // 调整进度条……
                break;
            case 'video_ok':
                // 显示消息文本，播放视频……
                eventSource.close();
        }
    };
</script>

看上去很好，然而我用不了。因为 Server-Sent Events 是基于 GET 方法实现的。

我发送消息的请求只能用 POST 方法，因为 GET 携带不了那么多数据。虽然 GET 可以用 query 参数来传送一些数据，但我需要发给服务器的包括历史消息列表（要交给 ChatGPT 生成回复用），可能包含很多段很长的文字，放到参数里不现实。不仅有长度限制，还需要处理“?”、“&”等字符，以及处理之后还能不能被 ChatGPT 正确理解也是问题。

方案2：流式传输

流式传输响应是服务器将数据分块发送给客户端，客户端会多次接收到数据，因此可以更早开始处理数据。

后端代码是这样：

from flask import Response, stream_with_context

def generate():
    # 和上面 Server-Sent Event 差不多，yield 多个 json，用 \n 等分隔符分开
    yield f"{json.dumps({'status': 'text_ok', 'data': '...'})}\n"
    # ...

@app.route('/video_message')
def video_message_stream():
    return Response(stream_with_context(generate()), content_type='application/json')

前端代码是这样：

<script>
    fetch('/video_message', {
        method: 'POST',
        body: data // 历史消息等数据
    })
        .then(response => {
            const reader = response.body.getReader();
            const decoder = new TextDecoder();
            let data = "";

            function read() {
                reader.read().then(({ done, value }) => {
                    if (done) {
                        // 结束。从 data 中读取视频链接，播放视频……
                        return;
                    }
                    // 将这些碎片化的 data 拼起来
                    data += decoder.decode(value);
                    // 从 data 中尝试读取 json……
                    // 如果 json 中包含进度条信息则调整进度条……
                    read();
                });
            }
            read();
        })
        .catch(err => console.log(err));
</script>

当使用流式模式时，后端产生的那些 json 数据会被打断，以碎片的方式发到前端。前端等于建立一个缓冲池，把碎片化的数据拼起来，一边拼一边从头开始尝试分离出 json 对象。

这部分的处理会非常繁琐。缓冲池的部分还好，可以用 JavaScript 的 ReadableStream 来包装一下。但是识别和分离每个 json 只能依赖于某种特殊约定的数据格式，比如很特别的分隔符。像上面一样用 \n 当分隔符肯定是不行的，因为文本数据里也会有 \n，包含文本的 json 会被错误切割。用别的分隔符也会有同样的问题。

本质问题在于流式传输并不是用来解决这样的需求的。流式模式适合传输单一类型的大块数据，比如一个视频文件，或者一大段文本。我们的需求是传输多个彼此独立的事件信息。单一数据可以在任意位置切割成块，同时每一块的处理方式都相同。多个事件信息每一个都需要获取完整的数据结构，再按照其内容调用不同的处理逻辑。用流式是自寻烦恼。

方案3：WebSocket

我没试，感觉杀鸡用牛刀了。

当时我和 ChatGPT 讨论到这里感觉山穷水尽了。后来我尤里卡！想到了一个办法。如果发送消息只能用 POST，服务端事件只能用 GET，那我可以分开发多个请求呀。

方案4：三个请求

以下描述的是不用服务端事件请求，只用三个普通的请求。这不是实际采用的方法，效率不高，仅是作为对比和引出实际方案。

  1. 前端用 POST 方法发送消息请求。
  2. 后端生成文本，估算延迟，放到响应中。文本暂存起来，生成一个对应的 id，也放到响应中。
  3. 前端收到延迟信息，显示进度条。发送第二个请求，包含 id。
  4. 后端取出文本，生成音频，估算精确的延迟，放到响应中。音频暂存起来。
  5. 前端更新进度条。发送第三个请求。
  6. 后端取出音频，开始合成视频。
  7. 前端收到视频链接，开始播放。

总的耗时为：网络传输时间 x 6 + ChatGPT 回复时间 + TextToSpeech 回复时间 + 视频生成时间。

方案5：两个请求 + Server-Sent Events

这种方法会优化掉一部分网络传输：

  1. 前端用 POST 方法发送消息请求。
  2. 后端生成文本，估算延迟，放到响应中。调用 generate() 函数生成一个 generator 暂存起来，生成一个对应的 id，也放到响应中。generate()函数代码是：
def generate():
    # 生成音频……
    # 基于音频长度估算更精确的延迟……
    yield f"data: {json.dumps({'status': 'audio_ok', 'data': '...'})}\n\n"

    # 生成视频……
    yield f"data: {json.dumps({'status': 'video_ok', 'data': '...'})}\n\n"

注意当 generate() 函数调用时，会立即启动生成音频的任务，而无需等待前端请求。（但是视频生成任务可能需要等待。）

  1. 前端收到延迟信息，显示进度条。发送 EventSource 请求，包含 id。
  2. 后端取出 generator，向前端发送事件数据。
  3. 前端收到 “audio_ok” 事件时更新进度条；收到“video_ok”事件时开始播放视频。

计算耗时需要分两种情况：

  1. 如果第二个请求发送到服务器时，音频生成还未完成（常见情况），则总耗时为：网络传输时间 x 2 + ChatGPT 回复时间 + TextToSpeech 回复时间 + 视频生成时间。此时后端的任务流是连续的，总耗时和方案 0 相同。

  2. 如果第二个请求发送到服务器之前，音频生成已经完成（网速较慢的情况），后端正在等待请求来启动视频生成，则总耗时为：网络传输时间 x 4 + ChatGPT 回复时间 + 视频生成时间。此时音频生成的时间被两次网络传输时间覆盖了。

这就是最终方案。现在用户发完消息仍然需要等那两三分钟，但是他们会心怀希望，以为进度条的终点就会是漫长煎熬的结束。

List