【修复】AI：解决 spring-ai 与 webflux 集成时，SSE 存在乱序的问题

YunaiV
1 parent 041a00b2
Showing 5 changed files with 14 additions and 28 deletions
yudao-module-ai/yudao-module-ai-biz/src/main/java/cn/iocoder/yudao/module/ai/controller/admin/chat/AiChatMessageController.http
yudao-module-ai/yudao-module-ai-biz/src/main/java/cn/iocoder/yudao/module/ai/dal/mysql/chat/AiChatConversationMapper.java
yudao-module-ai/yudao-module-ai-biz/src/main/java/cn/iocoder/yudao/module/ai/service/impl/AiChatServiceImpl.java
yudao-module-ai/yudao-spring-boot-starter-ai/src/test/java/cn/iocoder/yudao/framework/ai/openAiImage/OpenAiImageClientTests.java
yudao-server/src/main/resources/application.yaml
@@ -6,7 +6,7 @@ Authorization: {{token}}
  
 {
   "conversationId": "1781604279872581649",
-  "content": "中国好看吗？"
+  "content": "你是 OpenAI 么？"
 }
  
  
@@ -16,11 +16,10 @@ Content-Type: application/json
 Authorization: {{token}}
  
 {
-  "conversationId": "1781604279872581651",
-  "content": "苹果是什么颜色？"
+  "conversationId": "1781604279872581690",
+  "content": "1+1=?"
 }
  
-
 ### message list
 GET {{baseUrl}}/admin-api/ai/chat/message/list-by-conversation-id?conversationId=1781604279872581649
 Authorization: {{token}}
@@ -15,12 +15,11 @@ import java.util.List;
 @Mapper
 public interface AiChatConversationMapper extends BaseMapperX<AiChatConversationDO> {
  
+    // TODO @fan：建议这里不排序哈；交给他们前端排序
     default List<AiChatConversationDO> selectListByUserId(Long userId) {
-        return selectList(
-                new LambdaQueryWrapperX<AiChatConversationDO>()
-                        .eq(AiChatConversationDO::getUserId, userId)
-                        .orderByAsc(AiChatConversationDO::getCreateTime)
-        );
+        return selectList(new LambdaQueryWrapperX<AiChatConversationDO>()
+                .eq(AiChatConversationDO::getUserId, userId)
+                .orderByAsc(AiChatConversationDO::getCreateTime));
     }
  
 }
@@ -32,6 +32,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
  
 import java.time.LocalDateTime;
 import java.util.*;
@@ -125,9 +126,10 @@ public class AiChatServiceImpl implements AiChatService {
         Flux<ChatResponse> streamResponse = chatClient.stream(prompt);
  
         // 3.3 流式返回
+        // 注意：Schedulers.immediate() 目的是，避免默认 Schedulers.parallel() 并发消费 chunk 导致 SSE 响应前端会乱序问题
         StringBuffer contentBuffer = new StringBuffer();
-        return streamResponse.map(response -> {
-            String newContent = response.getResult() != null ? response.getResult().getOutput().getContent() : null;
+        return streamResponse.publishOn(Schedulers.immediate()).map(chunk -> {
+            String newContent = chunk.getResult() != null ? chunk.getResult().getOutput().getContent() : null;
             newContent = StrUtil.nullToDefault(newContent, ""); // 避免 null 的 情况
             contentBuffer.append(newContent);
             // 响应结果
@@ -71,21 +71,4 @@ public class OpenAiImageClientTests {
         }
     }
  
-    public static void main(String[] args) {
-//        OpenAiApi api = new OpenAiApi("https://api.gptsapi.net", "sk-yzKea6d8e8212c3bdd99f9f44ced1cae37c097e5aa3BTS7z");
-//        OpenAiApi api = new OpenAiApi("https://openkey.cloud", "sk-QmgIIPc5xiYd8lPb076b1b7774Ea49Af9eD2Ef172c8f7e43");
-        OpenAiApi api = new OpenAiApi("https://api.chatanywhere.tech", "sk-gkgfYxhX9FxyZJznwxRZSJwKeGQYNPDVWjhby2PRRf17GHeT");
-        OpenAiChatClient client = new OpenAiChatClient(api);
-//        String result = client.call("未来，英文是什么？");
-//        System.out.println(result);
-        Flux<String> result = client.stream("未来，英文是什么？");
-        result.map(new Function<String, String>() {
-            @Override
-            public String apply(String s) {
-                System.out.println(s);
-                return s;
-            }
-        }).blockLast();
-    }
-
 }
@@ -156,6 +156,9 @@ spring.ai:
 #    base-url: https://api.chatanywhere.tech
     api-key: sk-yzKea6d8e8212c3bdd99f9f44ced1cae37c097e5aa3BTS7z
     base-url: https://api.gptsapi.net
+#    chat:
+#      options:
+#        model: gpt-4-0125-preview
  
 yudao.ai:
   yiyan: