一个java项目中，如何使用sse协议，构造一个chatgpt的流式对话接口

news/2024/5/19 18:03:13

前言
如何注册chatGPT，怎么和它交互，本文就不讲了；因为网上教程一大堆，而且你要使用的话，通常会再包一个算法服务，用来做一些数据训练和过滤处理之类的，业务服务基本不会直接与原生chatGPT交互。
而下面阐述的，就是业务服务与算法服务的交互。

业务需求-需要实现什么样的功能

想要一个类似与AI问答助手的机器人，可以实现根据某些场景对话提问的功能

  1. 可以直接提问，类似直接使用chatGPT，只不过这个提问的过程会做一些业务通用处理，比如问答数据的归纳反馈、敏感词过滤等等。
  2. 也可以给它喂一篇论文，喂一批近期的资讯，或者是一本小说之类的，根据指定的上下文去进行问答（这种场景需要先投递数据建立相关索引）
  3. ai的回答要求和chatGPT一样保持流式返回(也就是一个字一个字，一边生成一边返回，而不是等整个回答生成完之后一股脑返回)

剖析

重点是流式，这里我们预设算法侧已经有了一个流式返回的接口，整体的交互如下图所示
在这里插入图片描述
下面分别介绍几个关键节点的数据交互设计，仅供参考

q1

简述：页面发送问答数据给业务服务端

{"chatId": 233,"question": "这篇论文有几个论点？"
}
  • 这里的chatId可以理解为一个对话框id，业务服务端可以根据这个来进行问答归类、批量删除收藏、问答上下文查询等操作。
  • question就是问题的内容

这里需要注意就是，交互数据格式尽可能简单、易拓展，有些产品的页面交互设计的非常复杂，什么历史问答、角色信息之类的，套了一层又一层，其实很多都没必要的，这样前端组装起来也麻烦，也不利于数据的管理与后期功能的拓展。

q2

简述：就是业务服务根据前端传来的问题和所属的对话框，把相应的上下文查询出来(甚至可以前端维护一个是否发送上下文的开关，更动态一点)，包装成算法服务所需要的问题数据，发给它。

{"chatId":  233,"userName":  "张三","messageKey":  "0a795f6a-a029-435f-8d67-6f6f8e078cfe","message":  "这篇论文有几个论点？","chatHistory":  [{"messageKey":  "0a795f6a-a029-435f-8d67-6f6f8e07dasd","question":  "这篇论文的作者是谁","answer":  "这篇论文的作者是李四博士。"}],"callbackUrl":  "http://xxx/chat/question/callback"
}
  • 上述的messageKey就是一个消息的key，用以常规的接口调试
  • chatHistory就是历史问答记录，即上下文，众所周知chatGPT带不带上下文，回答的结果可能截然不同
  • callbackUrl是业务定义的一个回调接口，用来回调一些算法侧异步生产的信息，比如原文的定位信息、根据当前问题生成的推荐问答等，这些和流式的回答是不会一起返回的，所以额外提供一个接口来接收。

q3和a3

这两步不详述(主要我也不是研究算法模型的哈哈，不是很清楚细节)
我们只需要定义好a2返回的结果即可

a2

简述：主要是算法侧返回给业务服务的同步的流式回答，同时还可能有异步的额外信息的回调(q2的callbackUrl来接收)。所以a2的返回结果分为两个response
response1：同步的流式回答，一般在2-7s内返回第一个字符

data: 这
data: 篇
data: 论
... // 省略一些输出
data: 几
data: 。

流式问答的规范可以参考：流式接口协议规范

response2：异步的拓展信息(可有可无)

{"messageKey":"0a795f6a-a029-435f-8d67-6f6f8e078cfe", //必传 回调的消息key，每次问答唯一"expand":{"recommendedQuestions":[ // 推荐问题"这篇论文的主要论点是什么？"],"originalIndex":[{"sourceId":3432,"text":"首先第一个论点是......","textIndex":90}]}
}

a1

简述：a1要返回的格式很好理解，就是把a2中的两个response组合在一起，需要注意的有几点

  1. a2的response2不一定有，需要设置超时策略，且需要在流式回答最后输出
  2. a2中的response1是流式，response2不是；但输出到a1的时候，需要保证都在流中
  3. 最好需要约定一些event来作为标识符
event:messageKey // 消息key事件data:0a795f6a-a029-435f-8d67-6f6f8e078cfeevent:answer // 流式回答开始事件data:"在论文"data:"中"data:"，"data:"我"data:"们"data:"一"data:"共"
...
data:"几"data:"个"data:"论"data:"点"event:endTimedata:2024-02-27 17:05:24event:expand // 拓展信息开始事件，此处等待15s超时data:{"recommendedQuestions":["这篇论文的主要论点是什么"],"originalIndex":{"sourceId":32133,"text":"首先第一个论点是......","textIndex":90}}

代码

代码省略了一些无关紧要的业务特有的部分，只保留通用的部分
工具类：SSEUtils，用来操作SSE客户端

import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** description** @author luhui* @date 2024/1/25*/
@Slf4j
public class SSEUtils {/*** timeout 30min*/private static final Long DEFAULT_TIME_OUT = 30 * 60 * 1000L;/*** 订阅表*/private static final Map<String, EvaEmitter> EMITTER_MAP = new ConcurrentHashMap<>();public static final String MSG_DATA_PREFIX = "data:";public static final String MSG_EVENT_PREFIX = "event:";/*** description: 创建流** @param messageKey 本次问答的消息key* @return org.springframework.web.servlet.mvc.method.annotation.SseEmitter* @author luhui* @date 2024/2/23 17:09*/public static EvaEmitter getEmitter(String messageKey) {if (null == messageKey || "".equals(messageKey)) {return null;}EvaEmitter emitter = EMITTER_MAP.get(messageKey);if (null == emitter) {emitter = new EvaEmitter(DEFAULT_TIME_OUT);EMITTER_MAP.put(messageKey, emitter);}return emitter;}/*** description: 发消息** @param messageKey 本次问答的消息key* @param msg        消息* @author luhui* @date 2024/2/23 17:09*/public static void pushMsg(String messageKey, String msg) throws IOException {EvaEmitter emitter = EMITTER_MAP.get(messageKey);if (null != emitter) {emitter.send(EvaEmitter.event().data(msg));}}public static void pushEvent(String messageKey, String eventDesc) throws IOException{EvaEmitter emitter = EMITTER_MAP.get(messageKey);if (null != emitter) {emitter.send(EvaEmitter.event().name(eventDesc));}}/*** description: 关闭流** @param messageKey 本次问答的消息key* @author luhui* @date 2024/2/23 17:08*/public static void closeEmitter(String messageKey) {EvaEmitter emitter = EMITTER_MAP.get(messageKey);if (null != emitter) {try {emitter.complete();EMITTER_MAP.remove(messageKey);} catch (Exception e) {e.printStackTrace();}}}
}

工具类：SSEClient ，用来获取SSE流

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;/*** description** @author luhui* @date 2024/1/25*/
@Slf4j
public class SSEClient {// timeoutpublic static Integer DEFAULT_TIME_OUT = 60 * 1000;/*** 获取SSE输入流*/public static InputStream getSseInputStream(String urlPath, JSONObject param, int timeoutMill) {HttpURLConnection urlConnection = null;try {urlConnection = getHttpURLConnection(urlPath, timeoutMill);putData(urlConnection, param);InputStream inputStream = urlConnection.getInputStream();return new BufferedInputStream(inputStream);} catch (IOException e) {e.printStackTrace();}return null;}/*** 读流数据*/public static void readStream(InputStream is, MsgHandler msgHandler) throws IOException {BufferedReader reader = new BufferedReader(new InputStreamReader(is));try {String line = "";while ((line = reader.readLine()) != null) {if ("".equals(line)) {continue;}msgHandler.handleMsg(line);}} catch (Exception e) {e.printStackTrace();// 目前这里抛出的显式异常来自与用户手动关闭的连接，此时服务端与算法端的连接也捕获并关闭，无需存储} finally {// 服务器端主动关闭时，客户端手动关闭reader.close();is.close();}}private static HttpURLConnection getHttpURLConnection(String urlPath, int timeoutMill) throws IOException {URL url = new URL(urlPath);HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();urlConnection.setDoOutput(true);urlConnection.setDoInput(true);urlConnection.setUseCaches(false);urlConnection.setRequestMethod("POST");urlConnection.setRequestProperty("Connection", "Keep-Alive");urlConnection.setRequestProperty("Charset", "UTF-8");urlConnection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");urlConnection.setRequestProperty("accept", "text/event-stream");// 读过期时间urlConnection.setReadTimeout(timeoutMill);return urlConnection;}public static void putData(HttpURLConnection connection, JSONObject jsonStr) throws IOException {byte[] writebytes = jsonStr.toJSONString().getBytes();connection.setRequestProperty("Content-Length", String.valueOf(writebytes.length));DataOutputStream wr = new DataOutputStream(connection.getOutputStream());wr.write(jsonStr.toJSONString().getBytes());wr.flush();wr.close();}/*** 消息处理接口*/public interface MsgHandler {void handleMsg(String line) throws IOException;}
}

工具类：EvaEmitter，用来封装一些流信息

import cn.hutool.core.date.DateTime;
import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;/*** description EvaEmitter** @author luhui* @date 2024/02/22*/
@Data
public class EvaEmitter extends SseEmitter {public EvaEmitter(Long timeout) {super(timeout);}@ApiModelProperty("版本id")private Long versionId;@ApiModelProperty("用户问题")private String question;@ApiModelProperty("唯一消息key")private String messageKey;@ApiModelProperty("当前用户")private Long currentUid;@ApiModelProperty("当前用户名")private String currentUserName;@ApiModelProperty("项目id")private Long projectId;@ApiModelProperty("ai回答")private String aiAnswer;@ApiModelProperty("拓展信息")private JSONObject expand;@ApiModelProperty("错误信息")private JSONObject error;@ApiModelProperty("提问开始时间")private DateTime startTime;public JSONObject getHistory() {JSONObject history = new JSONObject();history.put("question", question);history.put("answer", aiAnswer);history.put("expand", expand);history.put("error", error);return history;}
}

具体的chat交互方法

	String messageKey = UUID.randomUUID().toString();EvaEmitter emitter = SSEUtils.getEmitter(messageKey);emitter.setProjectId(111);// 初始化相关字段sseService.chatTransfer(messageKey);

    @Async@Overridepublic void chatTransfer(String messageKey) {EvaEmitter emitter = SSEUtils.getEmitter(messageKey);// 正式参数JSONObject params = new JSONObject(true);params.put("versionId", emitter.getVersionId().toString());params.put("userName", emitter.getCurrentUserName());params.put("messageKey", emitter.getMessageKey());params.put("message", emitter.getQuestion());params.put("chatHistory", chatHistoryService.getChatHistory(emitter));params.put("callbackUrl", gateway + "/xxxchat/question/callback");InputStream inputStream = SSEClient.getSseInputStream(aiChatUrl, params, SSEClient.DEFAULT_TIME_OUT);try {StringBuilder answer = new StringBuilder();SSEUtils.pushEvent(messageKey, "messageKey");SSEUtils.pushMsg(messageKey, messageKey);SSEUtils.pushEvent(messageKey, "answer");AtomicReference<Boolean> sdkError = new AtomicReference<>(false);SSEClient.readStream(inputStream, line -> {log.info("messageKey:{}, chatTransfer:{}", emitter.getMessageKey(), line);String message = "";if (sdkError.get()) {String errorStr = line.split(SSEUtils.MSG_DATA_PREFIX)[1].trim();if (StringUtils.isNotBlank(errorStr)) {// 做一些错误处理message = "算法未知错误，请稍后再试";emitter.setError(message);}} else if (line.contains(SSEUtils.MSG_DATA_PREFIX)) {message = line.split(SSEUtils.MSG_DATA_PREFIX)[1].trim();} else if (line.contains(SSEUtils.MSG_EVENT_PREFIX)) {sdkError.set(true);} else {message = "";}if (StringUtils.isNotBlank(message)) {answer.append(message.replaceAll("\"", ""));SSEUtils.pushMsg(messageKey, message);}});emitter.setAiAnswer(answer.toString());// 保存当前问答消息，自行实现ChatHistoryEntity message = chatHistoryService.saveHistory(messageKey);SSEUtils.pushEvent(messageKey, "endTime");SSEUtils.pushMsg(messageKey, DateUtil.formatDateTime(message.getEndTime()));SSEUtils.pushEvent(messageKey, "expand");chatHistoryService.pushExpand(messageKey);} catch (IllegalStateException | IOException e) {log.error("pushMsg error, web端流已被关闭");} catch (Exception e) {e.printStackTrace();} finally {// 消息发送完或者出现异常的话，存储当前的消息，然后关闭流try {chatHistoryService.saveHistory(messageKey);} catch (Exception e) {e.printStackTrace();} finally {SSEUtils.closeEmitter(messageKey);}}}

	@Override@Retryable(value = Exception.class, maxAttempts = 6, backoff = @Backoff(delay = 500, multiplier = 2))public void pushExpand(String messageKey) throws IOException {// 如果异步的拓展信息，即a2中的response2回调成功的话，会存储到这里Object expandObj = redisService.hGet(RedisConstants.CHAT_AI_RECOMMENDED_QUESTIONS, messageKey);if (expandObj == null) {log.error("未获取到相关拓展信息, 稍后重试");throw new RuntimeException("未获取到相关拓展信息");} else {JSONObject expand = JSONObject.parseObject(expandObj.toString());EvaEmitter emitter = SSEUtils.getEmitter(messageKey);emitter.setExpand(expand);SSEUtils.pushMsg(messageKey, expand.toJSONString());log.info("messageKey:{}, chatTransfer:{}", emitter.getMessageKey(), expand);}}

http://www.mrgr.cn/p/36605654

相关文章

使用自己云服务器frp内网穿透记录

使用自己云服务器frp内网穿透记录

1.前提是自己现有云服务器已经2.下载对应的版本，我使用的是052.3下载地址 https://github.com/fatedier/frp/releases需要注意：下载的linux版本是服务端。windows是客户端 后续需要修改对用的配置文件 3.解压linux3.1 编辑配置文件vi frps.toml bindPort = 7000 # 服务运行端…
阅读更多...
.net6 ILogger日志保存到本地

.net6 ILogger日志保存到本地

1、新建一个LocalFileLogger的类public class LocalFileLogger : ILogger{private readonly string categoryName;private readonly string basePath;public LocalFileLogger(string categoryName){this.categoryName = categoryName;string[] fieldstrs = Enum.GetNames(typeo…
阅读更多...
CISCN2023-华北-normal_snake

CISCN2023-华北-normal_snake

就得审java。 路由分析 老规矩，先看看路由：/read路由下传参data，pyload不能包含!!，然后用了yaml来load传入的参数。 稍作了解，这其实就是 SnakeYaml 反序列化漏洞，禁用了 yaml 的常用头 !!。 前面的!!是用于强制类型转化，强制转换为!!后指定的类型，其实这个和Fastjson的…
阅读更多...
如何用Sublime Text实现正则查找与替换

如何用Sublime Text实现正则查找与替换

比如将下面的汉字语义加上中括号[{"text": "微笑","path": "emot01.png"},{"text": "大笑","path": "emot02.png"},{"text": "鼓掌","path": "emot03.pn…
阅读更多...
STM32之UASRT试验

STM32之UASRT试验

一、实验目的 1.实现STM32F407开发板与上位机工具通讯，中断方式具体实现的效果：上电后，下位机主动发送hello world ，上位机收到并显示；上位机发送数字0~9 ，回复: zero ~ nine 2.通讯协议，后面补充 3.硬件使用野火开发版STM32F407 4.与开发板连接的接口是Usb转串口，根据…
阅读更多...
什么是uniapp----分包

什么是uniapp----分包

前言 还是同样的需求（uniapp的主包要求大小不得大于2MB），但是就算将能封装的都封装了还是会超过2MB，本文将介绍第二个优化点：分包开发 一、什么是分包开发？ 有很多小伙伴一听分包开发认为就是多建几个文件夹，到时候引用就行了，说对对，但也不对，慢慢看下去就知道原因了…
阅读更多...
80个在线小游戏源码

80个在线小游戏源码

源码简介 搭建80个在线小游戏网站源码&#xff0c;解压即可食用&#xff0c;支持在本地浏览器打开。 安装教程 纯HTML&#xff0c;直接将压缩包上传网站目录解压即可 首页截图 源码下载 80个在线小游戏源码-小8源码屋
阅读更多...
spring-接口大全

spring-接口大全

1. Bean 相关 1. InitializingBean InitializingBean接口为bean提供了初始化方法的方式，它只包括afterPropertiesSet方法，凡是继承该接口的类，在初始化bean的时候都会执行该方法。 demo @Component public class MyInitBean implements InitializingBean {public void after…
阅读更多...
设计不外流，保护创意的同时锁住图纸安全！

设计不外流，保护创意的同时锁住图纸安全！

在设计行业中，图纸和创意文稿的安全至关重要，因为它们体现了企业的创新能力和核心竞争力。华企盾DSC数据防泄密系统提供了一系列功能，可以有效地保护这些珍贵的设计和文档不被外泄。以下是如何利用华企盾DSC系统保障设计图纸安全的关键措施：全面的加密模式：华企盾DSC系统提…
阅读更多...
告别写作难题：探索三款AI写作助手，让你的创作更高效

告别写作难题：探索三款AI写作助手，让你的创作更高效

在互联网时代&#xff0c;写作已经成为了一项基本的技能。无论是个人博客、公众号文章、还是商业文案&#xff0c;写作都是必不可少的一环。但是&#xff0c;很多人在写作过程中会遇到各种各样的问题&#xff0c;比如缺乏灵感、语言表达不清、排版混乱等。这时候&#xff0c;AI…
阅读更多...
dns服务器

dns服务器

DNS查询方式 dns服务器有两种查询方式：递归查询：在递归查询中，客户端向本地DNS服务器发送一个域名解析请求，并要求该DNS服务器负责完成整个解析过程。 如果本地DNS服务器拥有所请求的域名解析信息，则它会直接回复客户端，并负责向其他DNS服务器查询所需的信息。 如果本地D…
阅读更多...
LM1875L-TB5-T 音频功率放大器 PDF中文资料_参数_引脚图

LM1875L-TB5-T 音频功率放大器 PDF中文资料_参数_引脚图

LM1875L-TB5-T 规格信息&#xff1a; 商品类型音频功率放大器 音频功率放大器的类型- 输出类型1-Channel (Mono) 作业电压16V ~ 60V 输出功率25W x 1 4Ω 额外特性过流保护,热保护 UTC LM1875是一款单片功率放大器&#xff0c;可为消费类音频应 用提供极低失真和高品质的…
阅读更多...
安装1Panel管理面板

安装1Panel管理面板

1Panel 是一个现代化、开源的 Linux 服务器运维管理面板。 安装部署：curl -sSL https://resource.fit2cloud.com/1panel/package/quick_start.sh -o quick_start.sh && bash quick_start.sh 手动安装dockerhttps://docker-practice.github.io/zh-cn/install/raspberr…
阅读更多...
拖拽式工作流开发有什么突出优势？

拖拽式工作流开发有什么突出优势？

在发展越来越快速的今天，拖拽式工作流开发得到了很多客户朋友的青睐与支持，是创收降本的得力助手。想要实现高效率的办公方式，可以试着了解低代码技术平台及拖拽式工作流开发的优势特点。具有好操作、好维护、够灵活、可视化界面操作等优势特点的低代码技术平台可以助力企业…
阅读更多...
Mac读写U盘软件哪个好用 Mac读写U盘很慢怎么解决 macbookpro读取u盘

Mac读写U盘软件哪个好用 Mac读写U盘很慢怎么解决 macbookpro读取u盘

在使用Mac电脑时&#xff0c;读写U盘是一个常见的需求&#xff0c;特别是当U盘格式为NTFS时。选择适合的软件来实现这一操作至关重要。下面我们来看Mac读写U盘软件哪个好用&#xff0c;Mac读写U盘很慢怎么解决的相关内容。 一、Mac读写U盘软件哪个好用 在Mac上选择一款适合的…
阅读更多...
深度相机（3D相机）

深度相机（3D相机）

传统的RGB彩色相机称为2D相机&#xff0c; 只能得到2D的图像信息&#xff0c; 无法得到物体与相机的距离信息&#xff0c;也就是深度信息。 顾名思义&#xff0c; 深度相机除了获取2D信息&#xff0c;还能得到深度信息&#xff0c;也叫RGBD相机&#xff0c; 或3D相机。 顺便提…
阅读更多...
小白PDF阅读器重排版时的自动提取背景色功能介绍及实现

小白PDF阅读器重排版时的自动提取背景色功能介绍及实现

小白PDF阅读器在1.35之前的版本对于有深色背景的页面重拍版时并不太完美。对于深色背景区域主要表现在不能分割排版和重排后页面元素割裂感明显。小白PDF阅读器在1.35版本主要针对这两个问题进行了优化！ 最终效果对比图如下自动重排版彩色部分内容，并提取彩色背景自动提取背景…
阅读更多...
Compilation Steps and Memory Layout of the C program

Compilation Steps and Memory Layout of the C program

Table of ContentsTable of ContentsWhat are the four stages of the compilation process?Preprocessing Compilation Assembly LinkingWhat are the four stages of the compilation process? Normally compiling a C program is a multi-stage process and utilizes diff…
阅读更多...
Unity读书系列《Unity3D游戏开发》——脚本（一）

Unity读书系列《Unity3D游戏开发》——脚本（一）

文章目录 前言一、脚本模版及其拓展1、脚本模版2、拓展脚本模版 二、脚本的生命周期三、脚本的执行顺序四、脚本序列化1、序列化数据2、serializedObject3、监听部分元素修改事件 五、定时器与间隔定时器六、工作线程&#xff08;多线程&#xff09;总结 前言 脚本在Unity的重…
阅读更多...
网络爬虫之爬虫原理

网络爬虫之爬虫原理

** 爬虫概述 Python网络爬虫是利用Python编程语言编写的程序&#xff0c;通过互联网爬取特定网站的信息&#xff0c;并将其保存到本地计算机或数据库中。 """ 批量爬取各城市房价走势涨幅top10和跌幅top10 """ ​ from lxml import etree impor…
阅读更多...
推荐文章

Copyright @ 2022~2023