Alan's blog

# Event source

在使用 chatGPT 接口时，遇到 event source 的使用，记录一下。

chatGPT 的 /v1/chat/completions 接口，可以传一个 stream:true 参数，以流形式接收返回，以实现打字效果。

# 是什么？

EventSource 是服务器推送的一个网络事件接口。一个 EventSource 实例会对 HTTP 服务开启一个持久化的连接，以 text/event-stream 格式发送事件，会一直保持开启直到被要求关闭。

与 WebSockets,不同的是，服务端推送是单向的。适合场景：聊天消息或市场价格

# 尝试实现

前端，创建请求，并监听：

const event = new EventSource("http://127.0.0.1:7001/test_es"); // GET

event.addEventListener("open", (e) => {
  console.log("open:", e);
});

event.addEventListener("message", (e) => {
  console.log("get event:", e);
});

event.addEventListener("error", (e) => {
  console.log("has error:", e);
  event.close();
});

服务端：

const PassThrough = require("stream").PassThrough;

class TempController extends Controller {
  async index() {
    const { ctx } = this;
    const stream = new PassThrough();

    ctx.set({
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
      "Access-Control-Allow-Origin": "*", // 注意跨域设置
    });

    let counter = 0;

    const timer = setInterval(() => {
      counter++;
      if (counter < 10) {
        stream.write(`data: counter-${counter}\n\n`);
      } else {
        stream.write("data: [DONE]\n\n");
        stream.end();
      }
    }, 1000);

    stream.on("close", function() {
      clearInterval(timer);
    });

    this.ctx.body = stream;
  }
}

效果：客户端发起请求后，会建立连接，接收 10 次服务端传来的 counter 信息，最后结束断开。

# 使用 fetch 实现

上述前端代码发起了一个 get 请求，通过 fetch 可以实现 post 请求。 ps. axios 浏览器端不支持流式处理，文档有误。

关键代码：

  1. fetch 封装加上 对 text/event-stream 的处理：
fetch(requestUrl, options)
  .then((response) => {
    clearTimeout(timeoutId);
    function parseResponse() {
      const contentType = response.headers.get("Content-Type");
      if (contentType && contentType.indexOf("application/json") !== -1) {
        return response.json();
      } else if (
        contentType &&
        contentType.indexOf("text/event-stream") !== -1
      ) {
        return new Promise((resolve) => {
          resolve({
            status: response.status,
            reader: response.body.getReader(),
            errcode: 0,
          });
        });
      } else {
        return response.text();
      }
    }

    parseResponse().then((resBody) => {
      if (response.ok) {
        handleSuccess && handleSuccess(response, resBody);
        resolve(resBody);
      } else {
        handleFail && handleFail(response);
        parseResponse().then((resBody) => {
          reject(resBody);
        });
      }
    });
  })
  .catch((error) => {
    handleError && handleError(error);
    reject(error);
  });
  1. 请求并解读流式数据
$api.test_es({}).then(({ reader, status }) => {
  while (true) {
    const { value, done } = await reader.read();
    console.log(done);
    if (done) break;// 当服务器 stream.close 关闭连接时，done 为 true

    const decodedText = new TextDecoder("utf-8").decode(value, { stream: true });

    if (status !== 200) {
      return;
    }
    console.log(decodedText);
  }
});
  1. 服务端还是用上一个 controller，但为其建立一个 psot 路由。

# 使用 chatpGPT-api 库

这是一个封装了 chatGPT 官方 API 的库，它的请求逻辑封装了 eventsource 传输，从源码中能学到不少东西。先简单用一下：

  • main.js
const fetch = require("../common/fetch.js");

(async function main() {
  const { ChatGPTAPI } = await import("chatgpt");
  const api = new ChatGPTAPI({
    apiKey: "API_KEY",
    fetch
  });
  const res = await api.sendMessage("Hello World!");
  console.log(res.text);// Hello! How can I assist you today?
})();
  • fetch.js 低版本的 node 没有全局的 fetch 方法，一般用 node-fetch 或者 isomorphic-fetch 库来代替， 而且这个库不支持代理参数，所以要重写个fetch。
const HttpsProxyAgent = require("https-proxy-agent").HttpsProxyAgent;
const nodeFetch = require("node-fetch");// npm i node-fetch@2，这个版本才支持 commonJS

async function fetch(url, options) {
  const defaultOptions = {
    agent: new HttpsProxyAgent("http://127.0.0.1:7890"),// 填写代理服务器
  };

  const mergedOptions = {
    ...defaultOptions,
    ...options,
  };

  return nodeFetch.default(url, mergedOptions);
}

module.exports = exports = fetch;

# 服务端转发 chatGPT 接口

# 总结

参考： ChatGPT流式streaming回复的实现 (opens new window) EventSource 引发的一系列事件 (opens new window) Egg + SSE (opens new window) chatgpt-api 仓库 (opens new window)

← 字体 →