背景

之前分享过在 Web 端如何实现 ChatGPT 打字机效果的文章，感兴趣的同学可以前往查看：实战案例：ChatGPT 打字机效果的三种实现方式，本文主要分享在微信小程序中如何实现 ChatGPT 打字机效果的前半部分，即如何获取 SSE 通信的接口数据。

接下来的文章实战案例：微信小程序中实现 ChatGPT 打字机效果（下）中，实现了小程序中渲染服务端 SSE 返回的 Markdown 格式的内容。

技术背景

本文所使用示例涉及的技术栈：

tips: 默认你已经知道上述库或框架如何使用，本文不会介绍相关技术栈的使用教程。

目录服务端 SSE 接口示例

此处的接口，直接复用之前 Web 端的示例，代码如下：

const express = require("express");
const cors = require("cors");
const app = express();
app.use(cors());
const PORT = 5000;
const getTime = () => new Date().toLocaleTimeString();
const contentStr = "很高兴为您服务，我是模拟的 ChatGPT 机器人。".split('')
app.get("/sse", function (req, res) {
  res.writeHead(200, {
    Connection: "keep-alive",
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
  });
  // 正常的 sse 结束，需要从客户端触发 close 事件，如果从服务端触发，客户端会收到 error 
  req.on('close', function () {
    console.log('close')
    clearInterval(interval)
  })
  let count = 0
  // 此处用计时器来模拟大模型的查询结果
  // 通过发送字符数组的长度，来模拟 SSE 服务的 start、cmpl、done 状态
  const interval = setInterval(() => {
    // 如果前端没有正确触发 SSE 的 close 事件，服务端判断如果数据已发送完成，也会主动关闭事件
    if (count > contentStr.length) {
      res.end()
      clearInterval(interval)
      return
    } else if (count === 0) {
      res.write(
        `data:${JSON.stringify({
          time: getTime(),
          event: 'start',
          content: contentStr[count]
        })}`
      );
      res.write("\n\n");
    } else if (count === contentStr.length) {
      res.write(
        `data:${JSON.stringify({
          time: getTime(),
          event: 'done',
        })}`
      );
      res.write("\n\n");
    }
    else {
      res.write(
        `data:${JSON.stringify({
          time: getTime(),
          event: 'message',
          content: contentStr[count]
        })}`
      );
      res.write("\n\n");
    }
    count++
  }, 100);
});
app.listen(PORT, function () {
  console.log(`Server is running on port ${PORT}`);
});

核心实现代码示例数据交互

// 完整代码
import { View } from "@tarojs/components";
import Taro from "@tarojs/taro";
export default function Index() {
  const handleClick = () => {
    const requestTask = Taro.request({
      url: "http://localhost:5000/sse",
      method: "GET",
      enableChunked: true,
      success: (res) => {
        console.log("---->", res);
      },
    });
    requestTask.onChunkReceived((res) => {
      console.log("chunk", res);
    });
  };
  return (
    <View className="index">
      <View onClick={handleClick}>点我发起请求View>
    View>
  );
}

数据处理

// 完整代码
import { View } from "@tarojs/components";
import Taro from "@tarojs/taro";
import * as TextEncoding from "text-encoding-shim";
export default function Index() {
  const handleClick = () => {
    const requestTask = Taro.request({
      url: "http://localhost:5000/sse",
      method: "GET",
      enableChunked: true,
      success: (res) => {
        console.log("---->", res);
      },
    });
    requestTask.onChunkReceived((res) => {
      const uint8Array = new Uint8Array(res.data);
      const str = new TextEncoding.TextDecoder("utf-8").decode(uint8Array);
      console.log("chunk", str);
    });
  };
  return (
    <View className="index">
      <View onClick={handleClick}>点我发起请求View>
    View>
  );
}

杂谈参考文档浏览知识共享许可协议

本作品采用知识共享署名-相同方式共享 4.0 国际许可协议进行许可。