gerrit.onap Code Review - usecase-ui/llm-adaptation.git/blob

   1 package org.onap.usecaseui.llmadaptation.service.impl;
   2
   3 import com.alibaba.fastjson2.JSONArray;
   4 import com.alibaba.fastjson2.JSONObject;
   5 import com.fasterxml.jackson.databind.ObjectMapper;
   6 import lombok.extern.slf4j.Slf4j;
   7 import org.onap.usecaseui.llmadaptation.bean.Application;
   8 import org.onap.usecaseui.llmadaptation.bean.ResultHeader;
   9 import org.onap.usecaseui.llmadaptation.bean.ServiceResult;
  10 import org.onap.usecaseui.llmadaptation.bean.fastgpt.dataset.CreateDataSetResponse;
  11 import org.onap.usecaseui.llmadaptation.bean.fastgpt.application.*;
  12 import org.onap.usecaseui.llmadaptation.constant.CommonConstant;
  13 import org.onap.usecaseui.llmadaptation.constant.FastGptConstant;
  14 import org.onap.usecaseui.llmadaptation.mapper.ApplicationMapper;
  15 import org.onap.usecaseui.llmadaptation.service.FastGptApplicationService;
  16 import org.onap.usecaseui.llmadaptation.util.TimeUtil;
  17 import org.springframework.beans.factory.annotation.Autowired;
  18 import org.springframework.core.io.ResourceLoader;
  19 import org.springframework.stereotype.Service;
  20 import org.springframework.web.reactive.function.client.WebClient;
  21 import reactor.core.publisher.Flux;
  22 import reactor.core.publisher.Mono;
  23
  24 import java.io.IOException;
  25 import java.io.InputStream;
  26 import java.util.ArrayList;
  27 import java.util.List;
  28 import java.util.UUID;
  29 import java.util.concurrent.atomic.AtomicBoolean;
  30
  31 import static org.springframework.http.MediaType.APPLICATION_JSON;
  32
  33 @Slf4j
  34 @Service
  35 public class FastGptApplicationServiceImpl implements FastGptApplicationService {
  36     @Autowired
  37     private ResourceLoader resourceLoader;
  38
  39     @Autowired
  40     private ApplicationMapper applicationMapper;
  41
  42     @Autowired
  43     private WebClient webClient;
  44
  45     private final ObjectMapper objectMapper = new ObjectMapper();
  46
  47     @Override
  48     public Mono<ServiceResult> createApplication(Application application,  String serverIp) {
  49         try (InputStream inputStream = resourceLoader.getResource(FastGptConstant.CREATE_APP_PARAM_FILE_URL).getInputStream()) {
  50             CreateApplicationParam createApplicationParam = objectMapper.readValue(inputStream, CreateApplicationParam.class);
  51             createApplicationParam.setName(application.getApplicationName());
  52
  53             return createApplication(createApplicationParam, application, serverIp)
  54                     .onErrorResume(e -> {
  55                         log.error("Error occurred while creating application: {}", e.getMessage());
  56                         return Mono.just(new ServiceResult(new ResultHeader(500, "Application creation failed")));
  57                     });
  58
  59         } catch (IOException e) {
  60             log.error("Error occurred while reading input file: {}", e.getMessage());
  61             return Mono.just(new ServiceResult(new ResultHeader(500, "Failed to read input file")));
  62         }
  63     }
  64
  65     private Mono<ServiceResult> createApplication(CreateApplicationParam createApplicationParam, Application application,  String serverIp) {
  66         return webClient.post()
  67                 .uri(serverIp + FastGptConstant.CREATE_APPLICATION)
  68                 .contentType(APPLICATION_JSON)
  69                 .header(CommonConstant.COOKIE, FastGptConstant.COOKIE_VALUE)
  70                 .bodyValue(createApplicationParam)
  71                 .retrieve()
  72                 .bodyToMono(CreateDataSetResponse.class)
  73                 .flatMap(response -> {
  74                     if (response.getCode() == 200) {
  75                         return handleApplicationResponse(String.valueOf(response.getData()), application, serverIp);
  76                     }
  77                     return Mono.just(new ServiceResult(new ResultHeader(500, response.getStatusText())));
  78                 });
  79     }
  80
  81     private Mono<ServiceResult> handleApplicationResponse(String dataId, Application application,  String serverIp) {
  82         application.setApplicationId(dataId);
  83         String url = serverIp + FastGptConstant.UPDATE_APPLICATION + dataId;
  84         UpdateApplicationParam updateApplicationParam = new UpdateApplicationParam();
  85         updateApplicationParam.setAvatar("/imgs/app/avatar/simple.svg");
  86         updateApplicationParam.setDefaultPermission(0);
  87         updateApplicationParam.setName(application.getApplicationName());
  88         updateApplicationParam.setIntro(application.getApplicationDescription());
  89
  90         return webClient.put()
  91                 .uri(url)
  92                 .contentType(APPLICATION_JSON)
  93                 .header(CommonConstant.COOKIE, FastGptConstant.COOKIE_VALUE)
  94                 .bodyValue(updateApplicationParam)
  95                 .retrieve()
  96                 .bodyToMono(CreateDataSetResponse.class)
  97                 .flatMap(response -> {
  98                     if (response.getCode() == 200) {
  99                         return publishApplication(application, dataId, serverIp);
 100                     } else if (response.getCode() == 502000) {
 101                         return Mono.just(new ServiceResult(new ResultHeader(404, "The resource does not exist,please delete")));
 102                     } else {
 103                         return Mono.just(new ServiceResult(new ResultHeader(500, response.getStatusText())));
 104                     }
 105                 });
 106     }
 107
 108     private Mono<ServiceResult> publishApplication(Application application, String data,  String serverIp) {
 109         try (InputStream inputStream = resourceLoader.getResource(FastGptConstant.PUBLISH_APP_PARAM_FILE_URL).getInputStream()) {
 110             PublishApplicationParam publishApplicationParam = objectMapper.readValue(inputStream, PublishApplicationParam.class);
 111             publishApplicationParam.setVersionName(TimeUtil.getNowTime());
 112             publishApplicationParam.getChatConfig().setWelcomeText(application.getOpeningRemarks());
 113             setApplicationParameters(application, publishApplicationParam);
 114             String publishUrl = serverIp + FastGptConstant.PUBLISH_APPLICATION + data;
 115
 116             return webClient.post()
 117                     .uri(publishUrl)
 118                     .contentType(APPLICATION_JSON)
 119                     .header(CommonConstant.COOKIE, FastGptConstant.COOKIE_VALUE)
 120                     .bodyValue(publishApplicationParam)
 121                     .retrieve()
 122                     .bodyToMono(CreateDataSetResponse.class)
 123                     .flatMap(response -> {
 124                         if (response.getCode() == 200) {
 125                             applicationMapper.insertApplication(application);
 126                             return Mono.just(new ServiceResult(new ResultHeader(200, "Application created successfully")));
 127                         }
 128                         return Mono.just(new ServiceResult(new ResultHeader(500, response.getStatusText())));
 129                     });
 130         } catch (IOException e) {
 131             log.error("Error occurred while reading publish parameters: {}", e.getMessage());
 132             return Mono.just(new ServiceResult(new ResultHeader(500, "Failed to read publish parameters")));
 133         }
 134     }
 135
 136     private void setApplicationParameters(Application application, PublishApplicationParam publishApplicationParam) {
 137         publishApplicationParam.getNodes().forEach(node -> {
 138             if ("chatNode".equals(node.getFlowNodeType())) {
 139                 node.getInputs().forEach(input -> {
 140                     switch (input.getKey()) {
 141                         case "temperature":
 142                             input.setValue(application.getTemperature());
 143                             break;
 144                         case "systemPrompt":
 145                             input.setValue(application.getPrompt());
 146                             break;
 147                         case "model":
 148                             log.info(application.getLargeModelName());
 149                             input.setValue(application.getLargeModelName());
 150                             break;
 151                     }
 152                 });
 153             } else if ("datasetSearchNode".equals(node.getFlowNodeType())) {
 154                 node.getInputs().forEach(input -> {
 155                     if ("datasets".equals(input.getKey())) {
 156                         JSONObject jsonObject = new JSONObject();
 157                         jsonObject.put("datasetId", application.getKnowledgeBaseId());
 158                         List<JSONObject> list = new ArrayList<>();
 159                         list.add(jsonObject);
 160                         input.setValue(list);
 161                     }
 162                 });
 163             }
 164         });
 165     }
 166
 167     @Override
 168     public Flux<String> chat(JSONObject question,  String serverIp) {
 169         ChatParam chatParam = new ChatParam();
 170         chatParam.setAppId(question.getString("applicationId"));
 171         chatParam.setStream(true);
 172         chatParam.setDetail(true);
 173         chatParam.setChatId(UUID.randomUUID().toString());
 174         chatParam.setResponseChatItemId(UUID.randomUUID().toString());
 175         JSONObject time = new JSONObject();
 176         time.put("cTime", TimeUtil.getFormattedDateTime());
 177         chatParam.setVariables(time);
 178         Message message = new Message();
 179         message.setContent(question.getString("question"));
 180         message.setDataId(UUID.randomUUID().toString());
 181         message.setRole("user");
 182         List<Message> messages = new ArrayList<>();
 183         messages.add(message);
 184         chatParam.setMessages(messages);
 185         AtomicBoolean isDone = new AtomicBoolean(false);
 186         return webClient.post()
 187                 .uri(serverIp + FastGptConstant.APPLICATION_CHAT_URL)
 188                 .contentType(APPLICATION_JSON)
 189                 .header(CommonConstant.COOKIE, FastGptConstant.COOKIE_VALUE)
 190                 .bodyValue(chatParam)
 191                 .retrieve()
 192                 .bodyToFlux(String.class).flatMap(response -> parseAndTransform(response, isDone))
 193                 .onErrorResume(throwable -> {
 194                     log.error("An error occurred {}", throwable.getMessage());
 195                     return Flux.just("Network Error");
 196                 });
 197     }
 198
 199     private Flux<String> parseAndTransform(String param, AtomicBoolean isDone) {
 200         if (isDone.get()) {
 201             return Flux.empty();
 202         }
 203         JSONObject jsonObject = JSONObject.parseObject(param);
 204         if (!jsonObject.containsKey("choices")) {
 205             return Flux.empty();
 206         }
 207         JSONArray choices = jsonObject.getJSONArray("choices");
 208         JSONObject choice = choices.getJSONObject(0);
 209         if ("stop".equals(choice.getString("finish_reason"))) {
 210             isDone.set(true);
 211             return Flux.just("[DONE]");
 212         }
 213         String string = choice.getJSONObject("delta").getString("content");
 214         isDone.set(false);
 215         string = string.replace(" ", "__SPACE__");
 216         return Flux.just(string);
 217     }
 218
 219     @Override
 220     public Mono<ServiceResult> removeApplication(String applicationId,  String serverIp) {
 221         String url = serverIp + FastGptConstant.DELETE_APPLICATION + applicationId;
 222         return webClient.delete()
 223                 .uri(url)
 224                 .header(CommonConstant.COOKIE, FastGptConstant.COOKIE_VALUE)
 225                 .retrieve()
 226                 .bodyToMono(CreateDataSetResponse.class)
 227                 .flatMap(response -> {
 228                     if (response.getCode() == 200 || response.getCode() == 502000) {
 229                         return Mono.fromRunnable(() -> {
 230                             try {
 231                                 applicationMapper.deleteApplicationById(applicationId);
 232                             } catch (Exception dbException) {
 233                                 throw new RuntimeException("Database operation failed", dbException);
 234                             }
 235                         }).then(Mono.just(new ServiceResult(new ResultHeader(200, "delete success"))));
 236                     } else {
 237                         return Mono.just(new ServiceResult(new ResultHeader(500, response.getStatusText())));
 238                     }
 239                 })
 240                 .onErrorResume(e -> {
 241                     log.error("Error occurred while delete dataset: {}", e.getMessage());
 242                     return Mono.just(new ServiceResult(new ResultHeader(500, "delete failed")));
 243                 });
 244     }
 245
 246     @Override
 247     public Mono<ServiceResult> editApplication(Application application,  String serverIp) {
 248         return handleApplicationResponse(application.getApplicationId(), application, serverIp);
 249     }
 250
 251 }