背景
公司项目希望在客户端开发 AI 模式,支持 AI 问答,要求实现 ChatGPT 的打字机效果
目标
-
了解 ChatGPT 流式响应背后的技术(打字机)
-
调研 AI 服务流式响应的可行性(技术层面、服务器资源消耗层面)
打字机是如何实现的
众所周知,ChatGPT API 是一个OpenAI 的聊天机器人接口,它可以根据用户的输入生成智能的回复,为了提高聊天的流畅性和响应速度,采用流失输出的响应方式,类似打字机的呈现效果
这其实是采用了 SSE(Sever-sent Events) 服务端推送技术,允许服务器向客户端发送事件,从而实现服务器端推送
与 webSocket
不同的是,服务端推送是单向的。数据信息被单向从服务端到客户端分发. 当不需要以消息形式将数据从客户端发送到服务器时,这使它们成为绝佳的选择
SSE 的通信协议
SSE 通信协议很简单,本质上就是一个客户端发起的 HTTP GET请求,服务器在接收到该请求后,返回 200 OK状态,并附带以下响应头:
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
- Content-Type: text/event-stream 表示响应的内容类型是SSE格式的文本流。
- Cache-Control: no-cache 表示响应的内容不应该被缓存,以保证实时性。
- Connection: keep-alive 表示响应的连接应该保持打开,以便服务器端持续发送数据。
通常,客户端的请求中会包含特殊的头信息:"Accept: text/event-stream"
,表示客户端系统接收 SSE 数据
SSE 支持以下几种字段:
event
: 表示事件的类型,用于区分不同的事件,默认事件为message
data
: 表示事件的数据内容,可以有多行,每行都以data: 开头。id
: 表示事件的唯一标识符,用于断线重连和消息追踪。retry
: 表示断线重连的时间间隔,单位是毫秒。
SSE 事件流数据示例:
流式输出「够钟下班啦」,并以 event:data
标记事件流结束
event:message
data:够
event:message
data:钟
event:message
data:下
event:message
data:班
event:message
data:啦
event:end
data:
SSE 接口示例
编写一个支持 SSE 协议的接口
'use strict';
const Controller = require('egg').Controller;
class SSEController extends Controller {
async index() {
// Set SSE header
const { ctx } = this
ctx.response.set({
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
});
ctx.res.statusCode = 200;
ctx.res.write(':流式响应开始n');
let count = 0;
while (count < 6) {
const cur = count++
const data = {
message: `Hello, world ${cur}`,
time: new Date(),
};
await new Promise(resolve => setTimeout(resolve, 1000));
// Send SSE event
ctx.res.write(`data: ${JSON.stringify(data)}n`);
}
// event:end 是事件类型,表示结束事件;客户端识别到服务器已经结束响应,从而关闭连接
ctx.res.write('event:endndata:endnn');
// 监听客户端关闭连接的事件,从而调用 ctx.res.end() 结束响应并关闭连接。
ctx.req.on('close', () => {
ctx.res.end();
});
}
}
module.exports = SSEController;
请求 SSE 接口,流式响应:
curl -N --location --request GET 'http://127.0.0.1:7001/sse'
--header 'Accept: text/event-stream'
>>>>
:流式响应开始
data: {"message":"Hello, world 0","time":"2023-05-19T07:08:51.661Z"}
data: {"message":"Hello, world 1","time":"2023-05-19T07:08:52.662Z"}
data: {"message":"Hello, world 2","time":"2023-05-19T07:08:53.663Z"}
data: {"message":"Hello, world 3","time":"2023-05-19T07:08:54.665Z"}
data: {"message":"Hello, world 4","time":"2023-05-19T07:08:55.666Z"}
data: {"message":"Hello, world 5","time":"2023-05-19T07:08:56.667Z"}
:end
OK
打字机实现
后端代码
const express = require('express');
const router = express.Router();
router.get('/sse', (req, res) => {
res.set({
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
'Access-Control-Allow-Origin': '*',
});
res.statusCode = 200;
res.write('开始回答:n');
const answer = '众所周知,ChatGPT API 是一个OpenAI 的聊天机器人接口,它可以根据用户的输入生成智能的回复,为了提高聊天的流畅性和响应速度,采用流失输出的响应方式,类似打字机的呈现效果';
let i = 0;
const intervalId = setInterval(() => {
res.write(`event:messagendata:${answer[i]}nn`);
i++;
if (i === answer.length) {
res.write('event:endndata: nn'); // event:end 表示事件流结束
clearInterval(intervalId);
}
}, 100);
res.end(); // 事件流推送完毕,服务端主动断开连接
});
前端接入 SSE:
<!DOCTYPE html>
<html>
<meta charset="UTF-8">
<head>
<title>SSE Example</title>
</head>
<body>
<h1>SSE Example</h1>
<button id="startButton">开始</button>
<div id="output">回答:</div>
<script>
const startButton = document.getElementById('startButton');
const outputElement = document.getElementById('output');
let eventSource;
startButton.addEventListener('click', function() {
if (!eventSource) {
eventSource = new EventSource('http://localhost:7001/sse2');
eventSource.onmessage = function(event) {
const message = event.data;
outputElement.innerHTML += message;
};
eventSource.onerror = function(event) {
console.error('Error: ' + event);
};
// 服务器定义了事件流:event:end,因此监听 :end 事件来结束 eventSource
eventSource.addEventListener('end', function(event) {
console.log('SSE 连接已关闭');
eventSource.close();
});
}
});
</script>
</body>
</html>
前端响应示例:
安卓接入 SSE:
使用 HttpURLConnection 或 OkHttp 等网络库来建立与服务器的连接,并通过监听服务器发送的数据流来接收事件
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
public class SSEClient {
public void connectToSSE() {
try {
URL url = new URL("http://your-server/sse");
HttpURLConnection connection = (HttpURLConnection) url.openConnection();
connection.setRequestMethod("GET");
connection.setRequestProperty("Accept", "text/event-stream");
if (connection.getResponseCode() == HttpURLConnection.HTTP_OK) {
InputStream inputStream = connection.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
String line;
while ((line = reader.readLine()) != null) {
// 处理接收到的事件数据流
}
reader.close();
inputStream.close();
} else {
// 处理连接错误
}
connection.disconnect();
} catch (IOException e) {
e.printStackTrace();
}
}
}
iOS 接入 SSE
使用NSURLSession来建立与服务器的连接,并通过监听服务器发送的数据流来接收事件
NSURL *url = [NSURL URLWithString:@"http://your-server/sse"];
NSURLSessionDataTask *task = [[NSURLSession sharedSession] dataTaskWithURL:url completionHandler:^(NSData *data, NSURLResponse *response, NSError *error) {
if (error) {
// 处理连接错误
} else {
NSString *eventData = [[NSString alloc] initWithData:data encoding:NSUTF8StringEncoding];
// 处理接收到的事件数据流
}
}];
[task resume];
打字机视频演示效果
gzoffice.mojidict.com:9001/moji-test/%…
服务端推送技术安全性对比
服务端推送技术涉及到客户端和服务器之间的数据传输,因此需要考虑安全性问题。不同的服务端推送技术有不同的安全性特点:
-
Ajax 短轮询和长轮询和基于 iframe 的流都是基于 HTTP 协议的,因此可以使用 HTTPS 协议来加密数据,防止中间人攻击或数据泄露。但是,这些技术都需要频繁地发送请求和响应,这可能会增加服务器的负载和网络的拥塞,也可能会被一些恶意的请求或响应干扰。
-
SSE(Sever-sent Events)
也是基于 HTTP 协议的,因此也可以使用 HTTPS 协议来保证数据的安全性。SSE 相比于 Ajax 轮询技术,只需要建立一次连接,就可以持续地接收服务器的事件,这样可以减少网络开销和服务器压力。但是,SSE 只支持单向的通信,即服务器向客户端发送数据,客户端不能向服务器发送数据。这可能会限制一些交互功能的实现。SSE 多用在例如,聊天应用、股票行情、新闻更新等场景 -
WebSockets 是基于 TCP/IP 协议的,因此可以使用 WSS 协议来加密数据,防止数据被窃取或篡改。WebSockets支持双向的通信,客户端和服务器可以随时互相发送数据,这样可以实现更丰富和灵活的交互功能。但是,WebSockets 需要额外的端口号和组件来支持,在一些环境中可能会遇到兼容性或安全性的问题。
综上所述,SSE 技术在 ChatGPT API 中有着重要的应用,它可以提高聊天机器人的响应速度和用户体验。不同的服务端推送技术有各自的优缺点和安全性特点,需要根据具体的场景和需求来选择合适的技术。
SSE 应用在服务端的考虑
问题场景:客户端 > 服务器(调用 openAI),是否要采用 SSE ?
SSE 需要保持连接,是否会占用服务器资源?
使用 SSE 时,保持连接(keep-alive)会对服务器资源产生一些影响:
- 连接开销:保持连接意味着服务器需要维持与客户端之间的长时间连接。这会占用一定的服务器内存和其他资源来处理这些连接。
- 并发连接:如果有大量客户端同时使用 SSE 与服务器建立连接,服务器需要同时管理和处理这些并发连接。这可能会增加服务器的负载和资源消耗。
- 带宽占用:保持连接需要维持持续的数据传输,即使是小量的数据也会占用一定的带宽。这可能对服务器的网络带宽和传输能力产生一定的压力。
- 状态管理:保持连接可能需要服务器维护客户端的连接状态,以便正确地处理和传输数据。这需要服务器进行额外的状态管理和资源分配。
SSE 相对于其他实时通信机制(如 WebSocket)来说,它的开销相对较低,因为SSE是基于标准的HTTP协议,使用简单的文本格式进行数据传输,并且不需要双向通信。SSE 在响应完成后,可以主动推送 event:end
结束事件来通知客户端响应结束关闭连接,避免一直保持连接
如果不使用 SSE 实时地流式传输,而是让客户端等待服务器完整的响应后再返回,那么当前请求的响应时间会变长,并且在这期间连接也会占用服务器的资源。
在传统的同步请求-响应模式中,客户端发送请求后会一直等待服务器生成完整的响应,期间连接保持打开状态,占用服务器的连接资源。这种等待时间会增加请求的响应时间,并且服务器需要维持连接的状态,消耗一定的资源。
相比之下,SSE 允许服务器实时地将部分数据流式传输给客户端,以提供更好的实时性和用户体验。服务器可以在计算过程中逐步发送部分回答,使客户端能够即时获取到部分结果,而无需等待完整的响应生成。
使用 SSE 的优势在于在计算过程中可以逐步返回结果,减少客户端等待时间,并降低服务器资源的占用。而在传统的同步请求-响应模式中,客户端必须一直等待完整的响应,期间连接会一直保持打开状态,占用服务器的资源。
SSE 的流量消耗?
流式传输在某些情况下可能会消耗较多的流量,特别是在实时传输大量数据时。流式传输的特点是将数据逐步传输给客户端,而不需要等待完整的响应生成。这意味着在传输过程中,数据会逐步发送给客户端,而不是一次性发送所有数据。
因此,如果流式传输的数据量很大或者传输速度较快,可能会占用更多的网络带宽和消耗更多的流量。对于大规模的流式传输,特别是对于长时间的传输,流量消耗可能会变得更明显。
然而,对于一些小规模的流式传输,比如 逐字
或逐段地传输文本数据,相对于一次性传输所有数据,流量消耗的增加可能是可以接受的,并且能够提供更好的用户体验。
openAI 官方对于 stream completion 的说明
在官方案例中,采用流式 & 非流式请求,让 gpt-3.5-turbo
数到100,看看各需要多长时间:
- 两个请求都花了大约 3 秒才完全完成
- 对于流请求,我们在 0.1 秒后收到第一个令牌响应,并且每隔约 0.01-0.02 秒收到后续令牌
在相同的整体响应时间内,流式请求加快了响应效率,优化了使用体验
总结
对于 AI 问答模式应用场景,可以考虑在服务器调用 openAI 接口时开启 stream: true
提取 openAI 的 stream completion,并且在业务接口中,采用 SSE,逐字返回调用结果,减少客户端等待时间,并在响应结束后及时关闭连接
参考资料
How_to_stream_completions.ipynb