ReadableStream

SSE 事件流

SSE 事件流中，一个数据包就是一个 Token，其 content-type 应指定为 text/eventstream。

data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"role": "assistant"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "苹"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "果"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "公司"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "是"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "一"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "家"}, "index": 0, "finish_reason": null}]}
data: {"id": "chatcmpl-6w****KZb6hx****RzIghUz****Qy", "object": "chat.completion.chunk", "created": 16******79, "model": "gpt-3.5-turbo-0301", "choices": [{"delta": {"content": "科"}, "index": 0, "finish_reason": null}]}
...

前端处理

一种方式是后台处理上述 SSE 事件流返回单纯的文本流，另一种方式是前端处理 SSE 事件流，需要注意 ReadableStream 传输的是字节流，处理完成后的数据需要用 encoder 去编码。

    const response = await fetch('your fetch url', {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json;charset=UTF-8',
            ...
          },
          body: JSON.stringify({
            messages,
            stream: true,
          }),
        });
        // 确保服务器响应是成功的
        if (!response.ok) {
          throw new Error(`HTTP error! status: ${response.status}`);
        }
        // 获取 reader
        const reader = response.body.getReader();
        const decoder = new TextDecoder('utf-8');
        // 注意
        const encoder = new TextEncoder();
        // 构造新的 ReadableStream.
        const readableStream = new ReadableStream({
          async start(controller) {
            while (true) {
              const { done, value } = await reader.read();
              if (done) {
                break;
              }
              const message = decoder.decode(value, { stream: true });
              const lines = message
                .toString()
                .split('\n\n')
                .filter((line) => line.trim() !== '');
              for (const line of lines) {
                const message = line.replace('data: ', '');
                const parsed = JSON.parse(message);
                controller.enqueue(encoder.encode(parsed.choices[0].delta.content));
              }
            }
            controller.close();
          },
        });
        return new Response(readableStream);

从 Fetch 到 Streams —— 以流的角度处理网络请求 - 掘金