SpringBoot项目整合智谱AI + SSE推送流式数据到前端展示 + RxJava得浅显理解

我要200 2024-09-06 14:31:02 阅读 55

项目背景:

项目背景是一个这个AI答题应用平台,我引入AI得作用就是让AI根据我指定得这个题目的标题和描述来生成一些列的题目。(主要功能是这个,但是还用了AI给我评分,不过这个功能比较简单,在本文就简单介绍)。

引入并整合智谱AI功能:

引入:

首先还是先贴一个官方文档:

智谱AI开放平台 (bigmodel.cn)

大概的步骤就是:

引入依赖

直接复制官方文档的客户端代码(同步调用)

引入之后我们可以不着急去在项目中使用,可以先跑一个单元测试

<code>@SpringBootTest

public class AItest {

private final String Key = "自己的Key";

//测试是否可以连接

@Test

public void test() {

ClientV4 client = new ClientV4.Builder(Key).build();

List<ChatMessage> messages = new ArrayList<>();

ChatMessage chatMessage = new ChatMessage(ChatMessageRole.USER.value(), "作为一名营销专家,请为智谱开放平台创作一个吸引人的slogan");

messages.add(chatMessage);

// String requestId = String.format(requestIdTemplate, System.currentTimeMillis());

ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder()

.model(Constants.ModelChatGLM4)

.stream(Boolean.FALSE)

.invokeMethod(Constants.invokeMethod)

.messages(messages)

.build();

ModelApiResponse invokeModelApiResp = client.invokeModelApi(chatCompletionRequest);

try {

System.out.println("model output:" + invokeModelApiResp.getData().getChoices().get(0));

} catch (Exception e) {

e.printStackTrace();

}

}

}

整体的代码逻辑:

首先先创建一个AI的客户端

接着拼接需要传给AI的信息列表,这里是ChatMessage

然后就是拼接这个需要发送的请求了

        里面有蛮多的参数

        model:模型,这个直接用最新的就行

        stream:就是指定流式还是非流式

        调用方法:这个默认就行

        message:就是你要发送的信息

最后发送调用接收返回值

对AI的这种工具的一个封装和简化的思想

对于这种工具类,就可以把这个封装成自己的SDK,以后自己直接引入即可

1:编写一个配置类将这个AI的客户端交给IOC容器管理:

@Configuration

@Data

public class AiConfig {

private final String aiKey = "自己的key";

@Bean

public ClientV4 clientV4() {

return new ClientV4.Builder(aiKey).build();

}

}

2:编写这个工具类:

这里叫不叫工具类都可以,鱼皮老师的定义是需要引入Bean对象的可以在Manager包下管理

首先可以只有一个最简单的方法:

我们可以利用Java的重载来一层一层向上封装

我们可以考虑到,我们和AI对话,一般都是直接问他对吧,所以作为我们用户,我们最乐意的应该是直接传一个文本,但是这里我们还需要注意,我们为了让这个返回的答案更准确,我们可以封装一个系统的prompt,给AI一个背景。

所以我们向上封装:

我们在这个方法中只需要传系统的prompt和用户的prompt就可以了

还可以再往上封装

我们可以根据传递的参数

我们可以封装同步和异步:

其它都是同理了,直接看代码:

<code>package com.ljh.aiplatform.manager;

import com.ljh.aiplatform.common.ErrorCode;

import com.ljh.aiplatform.exception.BusinessException;

import com.zhipu.oapi.ClientV4;

import com.zhipu.oapi.Constants;

import com.zhipu.oapi.service.v4.model.*;

import io.reactivex.Flowable;

import org.springframework.stereotype.Component;

import javax.annotation.Resource;

import java.util.ArrayList;

import java.util.List;

/**

* AI的请求

*/

@Component

public class AiManager {

@Resource

private ClientV4 clientV4;

// 稳定的随机数

private static final float STABLE_TEMPERATURE = 0.05f;

// 不稳定的随机数

private static final float UNSTABLE_TEMPERATURE = 0.99f;

/**

* 同步请求(答案不稳定)

*

* @param systemMessage

* @param userMessage

* @return

*/

public String doSyncUnstableRequest(String systemMessage, String userMessage) {

return doRequest(systemMessage, userMessage, Boolean.FALSE, UNSTABLE_TEMPERATURE);

}

/**

* 同步请求(答案较稳定)

*

* @param systemMessage

* @param userMessage

* @return

*/

public String doSyncStableRequest(String systemMessage, String userMessage) {

return doRequest(systemMessage, userMessage, Boolean.FALSE, STABLE_TEMPERATURE);

}

/**

* 同步请求

*

* @param systemMessage

* @param userMessage

* @param temperature

* @return

*/

public String doSyncRequest(String systemMessage, String userMessage, Float temperature) {

return doRequest(systemMessage, userMessage, Boolean.FALSE, temperature);

}

/**

* 通用请求(简化消息传递)

*

* @param systemMessage

* @param userMessage

* @param stream

* @param temperature

* @return

*/

public String doRequest(String systemMessage, String userMessage, Boolean stream, Float temperature) {

List<ChatMessage> chatMessageList = new ArrayList<>();

ChatMessage systemChatMessage = new ChatMessage(ChatMessageRole.SYSTEM.value(), systemMessage);

chatMessageList.add(systemChatMessage);

ChatMessage userChatMessage = new ChatMessage(ChatMessageRole.USER.value(), userMessage);

chatMessageList.add(userChatMessage);

return doRequest(chatMessageList, stream, temperature);

}

/**

* 通用请求

*

* @param messages

* @param stream

* @param temperature

* @return

*/

public String doRequest(List<ChatMessage> messages, Boolean stream, Float temperature) {

// 构建请求

ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder()

.model(Constants.ModelChatGLM4)

.stream(stream)

.temperature(temperature)

.invokeMethod(Constants.invokeMethod)

.messages(messages)

.build();

try {

ModelApiResponse invokeModelApiResp = clientV4.invokeModelApi(chatCompletionRequest);

return invokeModelApiResp.getData().getChoices().get(0).toString();

} catch (Exception e) {

e.printStackTrace();

throw new BusinessException(ErrorCode.SYSTEM_ERROR, e.getMessage());

}

}

/**

* 通用请求(流式输出)(简化消息传递)

* @param systemMessage

* @param userMessage

* @param temperature

* @return

*/

public Flowable<ModelData> doStreamRequest(String systemMessage, String userMessage,Float temperature) {

List<ChatMessage> chatMessageList = new ArrayList<>();

ChatMessage systemChatMessage = new ChatMessage(ChatMessageRole.SYSTEM.value(), systemMessage);

chatMessageList.add(systemChatMessage);

ChatMessage userChatMessage = new ChatMessage(ChatMessageRole.USER.value(), userMessage);

chatMessageList.add(userChatMessage);

return doStreamRequest(chatMessageList,temperature);

}

/**

* 通用请求(流式输出)

* @param messages

* @param temperature

* @return

*/

public Flowable<ModelData> doStreamRequest(List<ChatMessage> messages,Float temperature) {

// 构建请求

ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder()

.model(Constants.ModelChatGLM4)

.stream(Boolean.TRUE)

.temperature(temperature)

.invokeMethod(Constants.invokeMethod)

.messages(messages)

.build();

try {

ModelApiResponse invokeModelApiResp = clientV4.invokeModelApi(chatCompletionRequest);

final Flowable<ModelData> flowable = invokeModelApiResp.getFlowable();

return flowable;

} catch (Exception e) {

e.printStackTrace();

throw new BusinessException(ErrorCode.SYSTEM_ERROR, e.getMessage());

}

}

}

将AI结合自己的业务

先讲一个这个宏观的步骤:

如何将AI结合自己的业务

1:我们需要确定我们需要给AI什么样的prompt,AI会给我们更准确的答案(这个可能需要遵循一些给AI提问的技巧和长时间的调试)

2:我们接收到AI的请求之后,我们应该如何处理(我觉得这一步也很难)

秉承着上面的步骤,我们来将这个AI具体结合到我们的业务中

我们的业务目的是生成题目,所以我们要给AI的肯定是生成题目的标题和题目的描述

还有一个最重要的,一定要指定返回题目的类型,一定要和我们数据库中的题目格式一样

系统prompt实例:

你是一位严谨的出题专家(让GLM进行角色扮演),我会给你如下信息:(分隔符标示不同内容)

应用名称,

【【【应用描述】】】,(对于描述多行内容,可以通过特殊字符包起来,防止用户的输入干扰生成结果)

应用类别,

要生成的题目数,

每个题目的选项数

请你根据上述信息,按照以下步骤来出题:(链式思考,将复杂任务拆解成简单的任务)

要求:题目和选项尽可能地短,题目不要包含序号,每题的选项数以我提供的为主,题目不能重复(指定输出长度的示例)

严格按照下面的 json 格式输出题目和选项

[{"options":[{"value":"选项内容","key":"A"},{"value":"","key":"B"}],"title":"题目标题"}]

(少样本学习)

title 是题目,options 是选项,每个选项的 key 按照英文字母序(比如 A、B、C、D)以此类推,value 是选项内容

检查题目是否包含序号,若包含序号则去除序号

返回的题目列表格式必须为 JSON 数组(指定固定的输出格式)

标红的技巧都是在智谱AI的官方文档中的Prompt工程的文档中有写 

用户prompt实例:

小学数学测验, 【【【小学三年级的数学题】】】, 得分类, 10, 3

MBTI 性格测试, 【【【快来测测你的 MBTI 性格】】】, 测评类, 10, 3

用户的prompt就根据这个系统的prompt给就好。 

具体的代码实现:

首先想清楚这个我们如果要生成题目,我们需要几个参数?

根据AiManager(默认非流式且题目较为稳定),我们只要传两个参数,一个是系统prompt,一个是用用户的prompt。

系统prompt那就是这个默认值,用户的prompt是由应用的标题,应用的描述,应用题目数量,每个题目的选项组成的,所以我们需要一个封装用户prompt的方法

首先先来个dto接收前端参数:

/***

* ai生成题目请求

*/

@Data

public class AiGenerateQuestionRequest implements Serializable {

/**

* 应用id

*/

private Long appId;

/**

* 要生成的题目数

*/

private Integer questionsNum = 10;

/**

* 每个题目的选项数

*/

private Integer optionsNum = 4;

private static final long serialVersionUID = 1L;

}

我们只需要应用的id,我们就可以自己在数据库中查出对应的应用

还有生成的题目数,和对应题目的选线

private static final String SystemPromptQuestion = "你是一位严谨的出题专家,我会给你如下信息:\n" +

"```\n" +

"应用名称,\n" +

"【【【应用描述】】】,\n" +

"应用类别,\n" +

"要生成的题目数,\n" +

"每个题目的选项数\n" +

"```\n" +

"\n" +

"请你根据上述信息,按照以下步骤来出题:\n" +

"1. 要求:题目和选项尽可能地短,题目不要包含序号,每题的选项数以我提供的为主,题目不能重复\n" +

"2. 严格按照下面的 json 格式输出题目和选项\n" +

"```\n" +

"[{\"options\":[{\"value\":\"选项内容\",\"key\":\"A\"},{\"value\":\"\",\"key\":\"B\"}],\"title\":\"题目标题\"}]\n" +

"```\n" +

"title 是题目,options 是选项,每个选项的 key 按照英文字母序(比如 A、B、C、D)以此类推,value 是选项内容\n" +

"3. 检查题目是否包含序号,若包含序号则去除序号\n" +

"4. 返回的题目列表格式必须为 JSON 数组\n";

/**

* 封装用户的prompt

*

* @param app

* @param questionsNum

* @param optionsNum

* @return

*/

public String getUserPrompt(App app, Integer questionsNum, Integer optionsNum) {

ThrowUtils.throwIf(app == null, ErrorCode.NOT_FOUND_ERROR, "应用不存在");

StringBuilder userPrompt = new StringBuilder();

userPrompt.append("应用名称" + app.getAppName()).append("\n");

userPrompt.append("应用描述" + app.getAppDesc()).append("\n");

userPrompt.append("应用类别" + AppTypeEnum.getEnumByValue(app.getAppType()).getText()).append("\n");

userPrompt.append(questionsNum).append("\n");

userPrompt.append(optionsNum);

return userPrompt.toString();

}

系统的常量和封装用户的prompt方法

/**

* AI生成题目(非流式)

*

* @param aiGenerateQuestionRequest

* @return

*/

@PostMapping("/ai_generate")

public BaseResponse<List<QuestionContentDTO>> aiGenerateQuestion(@RequestBody AiGenerateQuestionRequest aiGenerateQuestionRequest) {

ThrowUtils.throwIf(aiGenerateQuestionRequest == null, ErrorCode.PARAMS_ERROR, "ai参数为空");

Long appId = aiGenerateQuestionRequest.getAppId();

App app = appService.getById(appId);

ThrowUtils.throwIf(app == null, ErrorCode.NOT_FOUND_ERROR, "应用不存在");

Integer questionsNum = aiGenerateQuestionRequest.getQuestionsNum();

Integer optionsNum = aiGenerateQuestionRequest.getOptionsNum();

String userPrompt = getUserPrompt(app, questionsNum, optionsNum);

String result = aiManager.doSyncStableRequest(SystemPromptQuestion,userPrompt);

int startIndex = result.indexOf("[");

int endIndex = result.lastIndexOf("]");

String jsonStr = result.substring(startIndex, endIndex + 1);

List<QuestionContentDTO> questionContentDTOS = JSONUtil.toList(jsonStr, QuestionContentDTO.class);

return ResultUtils.success(questionContentDTOS);

}

整体的代码逻辑其实很简单

就是先从数据库中查出对应的app

接着封装用户的prompt

然后调用自己封装的ai接口生成结果

最后处理结果

不过最后这个处理结果需要注意:

```json

[

{"options":[{"value":"12 + 15", "key":"A"},{"value":"14 + 17", "key":"B"},{"value":"13 + 16", "key":"C"}], "title":"小学数学测验:哪个选项的结果是29?"},

{"options":[{"value":"5 x 6", "key":"A"},{"value":"4 x 7", "key":"B"},{"value":"6 x 5", "key":"C"}], "title":"小学数学测验:哪个选项的结果是30?"}

]

这个是AI返回的result

[{"options":[{"result":"","score":0,"value":"","key":"A"}],"title":""}]

这个是我们数据库中存储的格式

我们发现,AI返回的时候多了什么 ```json

这种奇奇怪怪的字符

所以我们做一个处理获取第一个 "[" 和 最后一个"]"的位置,取一个字串即可

优化:

这里的优化点是什么呢:

我们生成题目会有重复

​​​我们每次都给AI发送同样的系统prompt和用户prompt,这样AI就很容易生成重复的题目

虽然说我们可以查询数据库然后将存入数据库中的重复题目删除

但是这样做用户体验并不好

我们最好的办法就是让这个AI不要生成重复的题目

这里就涉及到一个保存AI上下文的方法了

平常我们和AI聊天,都在一个界面中聊,但是你关闭了这个客户端之后,之前的记录就没了,如果你想引用之前的内容,就不太行了。

我们的解决办法就是保存用户的上下文

具体代码:

List<ChatMessage> chatMessageList = new ArrayList<>();

/**

* AI生成题目(非流式)

*

* @param aiGenerateQuestionRequest

* @return

*/

@PostMapping("/ai_generate")

public BaseResponse<List<QuestionContentDTO>> aiGenerateQuestion(@RequestBody AiGenerateQuestionRequest aiGenerateQuestionRequest) {

ThrowUtils.throwIf(aiGenerateQuestionRequest == null, ErrorCode.PARAMS_ERROR, "ai参数为空");

Long appId = aiGenerateQuestionRequest.getAppId();

App app = appService.getById(appId);

ThrowUtils.throwIf(app == null, ErrorCode.NOT_FOUND_ERROR, "应用不存在");

Integer questionsNum = aiGenerateQuestionRequest.getQuestionsNum();

Integer optionsNum = aiGenerateQuestionRequest.getOptionsNum();

String userPrompt = getUserPrompt(app, questionsNum, optionsNum);

chatMessageList.add(new ChatMessage(ChatMessageRole.SYSTEM.value(), SystemPromptQuestion));

chatMessageList.add(new ChatMessage(ChatMessageRole.USER.value(), userPrompt));

String result = aiManager.doRequest(chatMessageList, Boolean.FALSE, null);

System.out.println(result);

int startIndex = result.indexOf("[");

int endIndex = result.lastIndexOf("]");

String jsonStr = result.substring(startIndex, endIndex + 1);

chatMessageList.add(new ChatMessage(ChatMessageRole.ASSISTANT.value(), jsonStr));

List<QuestionContentDTO> questionContentDTOS = JSONUtil.toList(jsonStr, QuestionContentDTO.class);

return ResultUtils.success(questionContentDTOS);

}

我们保存一个List<ChatMessage> chatMessageList = new ArrayList<>();这样的对话列表

每次和AI对话完就保存进去,然后传递给AI,那AI就知道了,就不会生成重复的题目了

注意点:

1:tokens的长度不要太小,在项目中,我们处理Ai返回的数据,是因为这段数据前面有一个 [,后面有一个 ],如果把token设置得太小,后面得 ] 可能就会消失。

2:就是关于temperature这个值得设置,多尝试

3:注意保存这个上下文,这个上下文进行保存是会消耗tokens的,所以不能保存过长。

将生成题目的功能改造成流式输出并且用SSE的方式推送到前端

SSE技术:

Server-Sent Events(SSE)是一种技术,用于在客户端和服务器之间建立持久的单向通信通道,从而允许服务器主动推送更新到客户端。与 WebSocket 和其他实时通信技术相比,SSE 主要用于简单的、单向的数据流传输,特别适用于需要从服务器向客户端发送实时更新的场景。

SSE是单向的(从服务端到客户端),特别适用于这种AI流式生成答案返回

WebSocket是双向的,比较适合于网页聊天这种

这就是AI的优化了

我们每次生成的时候都要等很长的事件

为了提高用户的体验,我们可以生成一道题就输出一道题

这个需求最难的是什么?

我们知道就是,我们只要在后端的参数一改

把stream的参数改成true就是流式输出了(最多我们再封装一个方法)。

但是我们如何处理流式的数据呢?

[

{"options":[{"score":0,"value":"15","key":"A"},{"score":0,"value":"20","key":"B"}],"title":"3 + 12 等于多少?"},{"options":[{"score":0,"value":"6","key":"A"},{"score":0,"value":"7","key":"B"}],"title":"9 减去几等于 3?"}

]

 这个是Ai给我们的answer

问题是:我们肯定不能说一个题目都没取完整就返回给前端

但是,我们什么时候能确定这个一个题目是取完整了呢?

首先我们要知道,AI返回的流不一定是一个子一个字返回的

有可能是一个 :tions":[    也有可能是一个: lue":"6",

这种奇奇怪怪的字符串。

这里鱼皮老师用了一个方法:就是将每次返回的一块进行分流

将一个大的流分成n个小流,并且呢,将每个字符串转成一个字符列表就是List<Charcter>

然后我们在 “下游” 就可以简单进行处理了。

后端代码:

/**

* 通用请求(流式输出)(简化消息传递)

* @param systemMessage

* @param userMessage

* @param temperature

* @return

*/

public Flowable<ModelData> doStreamRequest(String systemMessage, String userMessage,Float temperature) {

List<ChatMessage> chatMessageList = new ArrayList<>();

ChatMessage systemChatMessage = new ChatMessage(ChatMessageRole.SYSTEM.value(), systemMessage);

chatMessageList.add(systemChatMessage);

ChatMessage userChatMessage = new ChatMessage(ChatMessageRole.USER.value(), userMessage);

chatMessageList.add(userChatMessage);

return doStreamRequest(chatMessageList,temperature);

}

/**

* 通用请求(流式输出)

* @param messages

* @param temperature

* @return

*/

public Flowable<ModelData> doStreamRequest(List<ChatMessage> messages,Float temperature) {

// 构建请求

ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder()

.model(Constants.ModelChatGLM4)

.stream(Boolean.TRUE)

.temperature(temperature)

.invokeMethod(Constants.invokeMethod)

.messages(messages)

.build();

try {

ModelApiResponse invokeModelApiResp = clientV4.invokeModelApi(chatCompletionRequest);

final Flowable<ModelData> flowable = invokeModelApiResp.getFlowable();

return flowable;

} catch (Exception e) {

e.printStackTrace();

throw new BusinessException(ErrorCode.SYSTEM_ERROR, e.getMessage());

}

}

/**

* AI生成题目(流式)

*+

* @param aiGenerateQuestionRequest

* @return

*/

@GetMapping("/ai_generate/sse")

public SseEmitter aiGenerateQuestionSSE(AiGenerateQuestionRequest aiGenerateQuestionRequest) {

ThrowUtils.throwIf(aiGenerateQuestionRequest == null, ErrorCode.PARAMS_ERROR, "ai参数为空");

Long appId = aiGenerateQuestionRequest.getAppId();

App app = appService.getById(appId);

ThrowUtils.throwIf(app == null, ErrorCode.NOT_FOUND_ERROR, "应用不存在");

Integer questionsNum = aiGenerateQuestionRequest.getQuestionsNum();

Integer optionsNum = aiGenerateQuestionRequest.getOptionsNum();

String userPrompt = getUserPrompt(app, questionsNum, optionsNum);

Flowable<ModelData> modelDataFlowable = aiManager.doStreamRequest(SystemPromptQuestion, userPrompt, null);

AtomicInteger count = new AtomicInteger();//用AtomicInteger可以保证线程安全(因为在使用流的时候,其实是个多线程的环境,普通的int无法保证线程安全)

SseEmitter sseEmitter = new SseEmitter();

StringBuilder stringBuilder = new StringBuilder();

//todo gpt提供了另一种方法

modelDataFlowable.observeOn(Schedulers.io())

.map(modelData -> modelData.getChoices().get(0).getDelta().getContent())

.map(modelData -> modelData.replaceAll("\\s", ""))//\\s 是一个正则表达式,表示匹配任意空白字符。 replaceAll("\\s", "") 这行代码用于去除字符串中的所有空白字符,包括空格、制表符(Tab)、换行符和其他空白字符

.filter(StringUtils::isNotBlank)

.flatMap(message -> {

List<Character> characterList = new ArrayList<>();

for (char c : message.toCharArray()) {

characterList.add(c);

}

return Flowable.fromIterable(characterList);

})//latMap 是 ReactiveX(RxJava)库中的一个非常重要且常用的操作符。它的作用是将源 Observable 或 Flowable 发送的每个数据项映射为一个新的 Observable 或 Flowable

.doOnNext(character -> {

if (character.equals('{')) {

count.addAndGet(1);

}

if(count.get() > 0){

stringBuilder.append(character);

}

if (character.equals('}')) {

count.addAndGet(-1);

if (count.get() == 0){

sseEmitter.send(JSONUtil.toJsonStr(stringBuilder.toString()));

stringBuilder.setLength(0);

}

}

})

.doOnError(throwable -> sseEmitter.completeWithError(throwable))

.doOnComplete(sseEmitter::complete)

.subscribe();

return sseEmitter;

}

前面的逻辑和上面一样

整体的代码逻辑:

首先,这个AI流式返回是返回了一个Flowable<ModelData>这种流式对象

我们可以对这个流式对象链式调用,就和Java8中的stream一样

接着继续分析:

.map(modelData -> modelData.getChoices().get(0).getDelta().getContent())

将这个数据取出来

.map(modelData -> modelData.replaceAll("\\s", ""))//\\s 是一个正则表达式,表示匹配任意空白字符。 replaceAll("\\s", "") 这行代码用于去除字符串中的所有空白字符,包括空格、制表符(Tab)、换行符和其他空白字符

将这个一些无效字符进行过滤

.filter(StringUtils::isNotBlank)

然后将空字符串再进行一个过滤

.flatMap(message -> {

List<Character> characterList = new ArrayList<>();

for (char c : message.toCharArray()) {

characterList.add(c);

}

return Flowable.fromIterable(characterList);

})

latMap 是 ReactiveX(RxJava)库中的一个非常重要且常用的操作符。它的作用是将源 Observable 或 Flowable 发送的每个数据项映射为一个新的 Observable 或 Flowable

就是上面说的进行了一个分流

.doOnNext(character -> {

if (character.equals('{')) {

count.addAndGet(1);

}

if(count.get() > 0){

stringBuilder.append(character);

}

if (character.equals('}')) {

count.addAndGet(-1);

if (count.get() == 0){

sseEmitter.send(JSONUtil.toJsonStr(stringBuilder.toString()));

stringBuilder.setLength(0);

}

}

})

到这个地方就剩下这个字符列表了

我们怎么处理,这里涉及到一个算法(其实就是力扣上的括号匹配)

但是这里可以做一个简化,因为我们知道 "{" 先出来,并且"{}"两个数量是对称的

所以我们来一个计数器当有左大括号 ++ ,有右大括号 -- ,然后等到计数器为0 的时候,通过SSE的sseEmitter返回给前端即可。

最后订阅一下就行。

还有一个注意点:需要注意这里的返回值是一个sseEmitter这个对象。

前端代码:

const doSubmitSSE = async () => {

if (!props.appId) {

return;

}

submitting.value = true;

//创建SSE请求

const eventRouce = new EventSource(

//填写后端的完整的地址

`http://localhost:8100/api/question/ai_generate/sse?appId=${props.appId}&optionsNum=${form.optionsNum}&questionNum=${form.questionsNum}`

);

eventRouce.onmessage = function (event) {

console.log(event.data);

props.aiOnSSESuccess?.(JSON.parse(event.data));

};

eventRouce.onopen = function () {

console.log("建立连接");

props.aiOnSSEStart?.(event);

handleCancel();

};

eventRouce.onerror = function () {

if (event?.eventPhase == EventSource.CLOSED) {

props.aiOnSSEClose?.(event);

console.log("正常关闭");

eventRouce.close();

}

};

submitting.value = false;

};

整体的代码逻辑:

首先先创建一个SSE对象来接收请求(SSE接收请求和axios没关系,我们这里需要指定后端的完整路径名称,并且把参数带上)

接着SSE对象的三个方法:onmessage,onopen ,onerror

这里就是接收参数,然后把这个event.data的数据传递给父组件,所以其实没有什么逻辑

关于这个流式和异步处理的自我思考:

 首先智谱AI底层的流式处理用的是RxJava响应式编程的库

先说一下什么是响应式编程:

响应式编程(Reactive Programming)是一种编程范式,用于处理异步数据流和变化的事件

它特别适合处理用户界面、实时数据流、事件驱动系统等需要高效处理异步操作的场景。

结合我们项目中的AI答案的流式返回,就刚好很适合这个响应式编程

根据上面那个项目的整个流程,我觉得处理整个流式答案返回最重要的就是整个流的概念和RxJava提供了很多的操作符可以直接来操作这个流(代码案例如上)

就和这个Java8中的新特性一样

毕竟这个地方叫自我思考

我问了gpt一个问题,因为我在看响应式编程的时候老是看到一个词叫:异步处理

但是在我第一次接触 “异步” 这个概念的时候,我见到的是:异步调用

我一开始就把这两个东西搞混了,我以为是一个东西

首先异步调用:

异步调用 是指在程序中发起一个操作或任务时,不需要等待这个操作或任务完成就能继续执行后续代码。异步调用的主要目的是避免阻塞主线程或当前执行上下文,以提高应用的响应能力和性能。

换成人话说就是,向我们平常的代码都是一行一行执行下来的,都是同步的,如果有一行卡住了,那后面的都不能执行

但是异步的不一样,异步的啊,不能让你堵住,我就先执行后面的,在你这个地方留一个回调函数,等你执行完了我再回来执行你

之前在API开发平台的接口中有提到,就是在springgateway的过滤器默认应该用的是springFlux的异步调用, 我向记录日志,我需要等,这个调用的接口的人调用结束了,我才能记录日志。

但是这个默认应该是异步的,所以就先记录了空日志(因为我什么都没干,所以是空的)。

前端天生就是可以异步调用的

绑定一个按钮,然后点击一下,就调用

接着就是异步处理了

异步处理 是指处理任务或数据的方式,使得任务可以在后台执行,而主线程可以继续执行其他操作。异步处理通常涉及到多个异步操作的组合、管理和协调。它的目标是高效地处理异步任务,并管理任务之间的关系。

说人话:

我们上面的流就是异步处理的,比如AI返回给我们的数据是 : a,b,c,d

AI是流式返回给我们的

然后我们搞了两个map 第一个map 给每个数 +1 第二个map给每个数  ×2

当a来了,a就执行第一个map,这个时候a开始+1,加完1,之后,a就流到 “下游”了,b就来了

b就+1,a这个时候就在×2了

所以我们也能看出来,这种异步处理数据流的感觉了

这样也可以提高效率对吧,类似多线程同时运行



声明

本文内容仅代表作者观点,或转载于其他网站,本站不以此文作为商业用途
如有涉及侵权,请联系本站进行删除
转载本站原创文章,请注明来源及作者。