新增

slibra-admin/src/main/java/com/slibra/web/controller/business/GRPCController.java

@@ -31,14 +31,19 @@ import org.pytorch.serve.grpc.inference.PredictionsRequest;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.MediaType;
 import org.springframework.util.CollectionUtils;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
+import reactor.core.publisher.Flux;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.time.Duration;
 import java.util.*;
+import java.util.function.Function;
 
 import static com.slibra.common.constant.MyConstants.*;
 import static com.slibra.common.enums.BusinessEnum.BigModelBizEnum.DECISION_REPORT;
@@ -80,6 +85,207 @@ public class GRPCController extends BaseController {
     @Value("${token.port}")
     private String port;
 
+    private static String apply(Long i) {
+        return CHAT_GONGDAN_1_ERROR_MSG;
+    }
+
+    @GetMapping(value = "/api/streamTest", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<String> streamTest() {
+        // 模拟大模型分块返回数据
+        String[] chunks = {
+                "这是第一块数据",
+                "这是第二块数据",
+                "这是最后一块数据"
+        };
+
+        try {
+            Thread.sleep(10000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+
+        // 使用 Flux 实现响应式流
+        return Flux.interval(Duration.ofMillis(500))  // 每500ms发送一块
+                .take(chunks.length)
+                .map(i -> chunks[i.intValue()]);
+    }
+
+    @PostMapping(value = "/api/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<String> streamData(@RequestBody ChatReq chatReq, HttpServletRequest httpServletRequest) {
+
+        log.info("进入了调⽤RAG+⼤模型的调⽤参数");
+        long time1 = System.currentTimeMillis();
+        log.info("开始进来的时间是{}",time1);
+        // 获取输出流
+        ManagedChannel channel = null;
+        //将结果记录到问答表
+        List<String> chunks = new ArrayList<>();
+        //请求参数
+        Integer module = chatReq.getModule();//来自那个模块
+        Date reportDate = chatReq.getReportDate();//如果传过值来了 代表的是工单类型的问答
+        String timeBegin = chatReq.getTimeBegin();
+        String timeEnd = chatReq.getTimeEnd();
+        if((Objects.isNull(reportDate) && StringUtils.isBlank(timeBegin) && StringUtils.isBlank(timeEnd)) && module == 1){//工单 必须要输入日期
+            //智能工单需要传入一个时间的参数
+
+            // 使用 Flux 实现响应式流
+            return Flux.interval(Duration.ofMillis(500))  // 每500ms发送一块
+                    .take(1)
+                    .map(GRPCController::apply);
+        }
+        StringBuilder sb = new StringBuilder();
+        String sessionId = chatReq.getSessionId();
+        int isStrong = Objects.isNull(chatReq.getIsStrong()) ? 0 : chatReq.getIsStrong();
+        String tools = STR_FALSE;//问答因为不确认是否走工具，所以传true; 但是工单一定是走工具的,传工具名称  2024年08月11日15:51:33 只有工具才是具体名称
+        boolean useRag = true;//只有问答才传true，其他默认都false
+        String extraStr = "";
+        //如果是工单，需要特殊处理一下showVal和question
+        if(1 == module){
+            if(StringUtils.isBlank(timeBegin) && StringUtils.isBlank(timeEnd)){//按天生成工单
+                //先用日期获取当天和前一天的数据，如果获取不到，则提示错误信息
+                String date = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD, reportDate);
+                chatReq.setShowVal(GONGDAN_TITLE.replace("#{0}", date));//处理展示的标题
+                chatReq.setQuestion(MODULE_CONTENT_NORMAL);
+                extraStr = this.buildGDNormalQuestion(DateUtils.parseDateToStr(DateUtils.YYYYMMDD_TS, reportDate));
+
+            }else{//自定义工单
+                chatReq.setShowVal(GONGDAN_TITLE_CUSTOM.replace("#{0}", timeBegin).replace("#{1}", timeEnd));//处理展示的标题
+                WorkOrderReq workOrderReq = new WorkOrderReq();
+                BeanUtils.copyProperties(chatReq, workOrderReq);
+                //拿到数据
+                List<WorkOrderRes> workOrderRes = frontService.customWorkOrder(workOrderReq);
+                chatReq.setQuestion(MODULE_CONTENT_CUSTOM);
+                extraStr = buildGDCustomQuestion(timeBegin, timeEnd, chatReq);
+                //2024年6月20日16:48:08  如果是自定义工单，需要处理图表 放到remark中
+                chatReq.setRemark(JSON.toJSONString(frontService.customWorkOrderHandleByData(workOrderReq, workOrderRes), JSONWriter.Feature.WriteNulls));
+            }
+            isStrong = 0;//2024年7月29日14:15:47 工单不走文心一言了
+            useRag = false;
+            tools = TOOLS_WORK_ORDER;
+        }
+        String ipAddr = IpUtils.getIpAddr();//获取用户的ip地址 传给大模型
+        int counts = 1;//默认是第一次
+        List<String> historyDates = new ArrayList<>();
+        //查询历史数据，放入集合中
+        if(StringUtils.isBlank(sessionId))
+            sessionId= IdUtils.simpleUUID();//第一次
+        else{
+            //通过sessionId获取所有的问答记录
+            List<TXinyiChatRecord> chatRecords = this.chatRecordMapper.selectTXinyiChatRecordList(TXinyiChatRecord.builder().sessionId(sessionId).build());
+            if(!CollectionUtils.isEmpty(chatRecords)){
+                for (TXinyiChatRecord chatRecord : chatRecords) {
+                    historyDates.add(chatRecord.getQuestion());
+                    historyDates.add(chatRecord.getAnswer());
+                }
+                //问答次数增加
+                counts = chatRecords.size() + 1;
+            }
+        }
+        //将新的问题放入集合中
+        historyDates.add(chatReq.getQuestion());
+        String headerPort = httpServletRequest.getHeader(port);
+        long time2 = System.currentTimeMillis();
+        log.info("开始调用大模型的的时间是{}",time2);
+        log.info("调用大模型之前，处理历史会话等操作的时间为{}", (time2 - time1)/1000);
+        try {
+            channel = ManagedChannelBuilder.forAddress(bigModelConfig.getIp(), StringUtils.isBlank(headerPort) ? bigModelConfig.getPort() : Integer.parseInt(headerPort))
+                    .usePlaintext()
+                    .build();
+            InferenceAPIsServiceGrpc.InferenceAPIsServiceBlockingStub stub = InferenceAPIsServiceGrpc.newBlockingStub(channel);
+            String toolsReq = chatReq.getTools();
+            if(StringUtils.isNotBlank(toolsReq)){
+                useRag = false;//2024年7月27日18:20:23 走本地工具，不需要文本增强
+                isStrong = 0;
+                tools = toolsReq;
+            }
+            //2024年08月21日11:49:25 如果是智慧办公，useRag也是false
+            if(2 == module){//智慧办公
+                useRag = false;
+            }
+//                else
+//                    tools = tools;
+//            }
+            String dataJson = buildBigModelReqForChat(sessionId, historyDates, ipAddr, isStrong, chatReq.getTopP(), chatReq.getTemperature(), tools, useRag, extraStr, chatReq.getPrompt(), chatReq.getModelType(), chatReq.getOnlineSearch());
+            log.info("******请求大模型的问答参数为{}", dataJson);
+            PredictionsRequest request = PredictionsRequest.newBuilder()
+                    .setModelName("slibra_bot")
+                    .putInput("method", ByteString.copyFrom("infer_stream", "utf-8"))//推理
+                    .putInput("data", ByteString.copyFrom(dataJson, "utf-8"))
+                    .buildPartial();
+            Iterator<PredictionResponse> predictions = stub.streamPredictions(request);
+            while (predictions.hasNext()) {
+                String responseStr = predictions.next().getPrediction().toStringUtf8();
+//                log.info("大模型问答返回的原始结果为{}", responseStr);
+                if(StringUtils.isBlank(responseStr)){
+                    log.error("大模型返回的是空，无法解析");
+                    continue;
+                }
+                responseStr = JSON.parseObject(responseStr).getString("message");
+                //2025年04月14日11:16:23 由于小程序演示，需要把“信义”替换成“LibraAI”
+                //2025年04月14日15:22:15 因为是一个个反的 没法替换
+//                responseStr = responseStr.replaceAll("信义", "LibraAI");
+                //2024年7月13日14:30:19 为空字符串的（实际可能是\n这种的）也返回前端，否则样式有问题
+//                if("complete".equals(responseStr) || StringUtils.isBlank(responseStr)){
+                if("complete".equals(responseStr)){
+//                    System.out.println("结尾语句并且是非JSON，无需处理");
+                    log.info("返回的结果message为{},无需再次处理", responseStr);
+                    //结束语句也流式输出，但是并不记录下来  2024年5月24日11:15:23 也不返回前端
+                }else{
+                    sb.append(responseStr);
+                    chunks.add(responseStr);
+                }
+            }
+            long time3 = System.currentTimeMillis();
+            log.info("结束调用大模型的的时间是{}",time3);
+            log.info("调用的时间为{}", (time3 - time2)/1000);
+            //将问答更新到数据库中
+            chatReq.setSessionId(sessionId);
+            chatReq.setAnswer(sb.toString());
+            chatReq.setType(0);//0问答 1决策 2本地 3仿真预测
+            chatReq.setModule(module);//0专家问答 1智能工单 2智能体助手 3告警 4简报
+            //2024年5月28日10:58:02 由于部分问题 展示的和调用大模型的不一样，所以这个由前端传过来
+//            chatReq.setShowVal(question);
+            chatReq.setCounts(counts);//问答次数
+            String userId = SecurityUtils.getUserId().toString();
+            String username = SecurityUtils.getUsername();
+            chatReq.setUserId(userId);
+            chatReq.setCreateBy(username);
+            chatReq.setCreateTime(DateUtils.getNowDate());
+            this.chatRecordMapper.insertTXinyiChatRecord(chatReq);
+            chunks.add((DEFAULT_ID_IDENTIFIER + chatReq.getId()));
+            long time4 = System.currentTimeMillis();
+            log.info("最后操作的时间是{}",time4);
+            log.info("最后操作的时间是{}", (time4 - time3)/1000);
+
+            return Flux.interval(Duration.ofMillis(10))  // 每500ms发送一块
+                    .take(chunks.size())
+                    .map(new Function<Long, String>() {
+                        @Override
+                        public String apply(Long i) {
+                            String s = chunks.get(Math.toIntExact(i));
+                            log.info("返回给前端的内容是{}", s);
+                            return s;
+                        }
+                    });
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        } finally {
+            // 关闭输出流
+            try {
+                channel.shutdown();
+            } catch (Exception e) {
+                log.error("关闭grpc流失败");
+            }
+        }
+
+        // 模拟大模型分块返回数据
+        // 使用 Flux 实现响应式流
+//        return Flux.interval(Duration.ofMillis(500))  // 每500ms发送一块
+//                .take(chunks.size())
+//                .map(i -> chunks.get(Math.toIntExact(i)));
+    }
+
+
 
     /**
      *
@@ -664,8 +870,7 @@ public class GRPCController extends BaseController {
                 }
                 responseStr = JSON.parseObject(responseStr).getString("message");
                 //2025年04月14日11:16:23 由于小程序演示，需要把“信义”替换成“LibraAI”
-                //todo 演示结束后，去掉下面的替换
-                responseStr = responseStr.replaceAll("信义", "LibraAI");
+//                responseStr = responseStr.replaceAll("信义", "LibraAI");
                 //2024年7月13日14:30:19 为空字符串的（实际可能是\n这种的）也返回前端，否则样式有问题
 //                if("complete".equals(responseStr) || StringUtils.isBlank(responseStr)){
                 if("complete".equals(responseStr)){