From 543af540916523f7102f4306c66c55d898bd8ec9 Mon Sep 17 00:00:00 2001 From: jqb Date: Mon, 27 Jul 2026 13:42:03 +0800 Subject: [PATCH] =?UTF-8?q?refactor(generation-controller):=20=E4=BC=98?= =?UTF-8?q?=E5=8C=96SSE=E6=B5=81=E5=BC=8F=E7=94=9F=E6=88=90=E9=80=BB?= =?UTF-8?q?=E8=BE=91=EF=BC=8C=E6=8F=90=E5=8D=87=E7=A8=B3=E5=AE=9A=E6=80=A7?= =?UTF-8?q?=E4=B8=8E=E8=B6=85=E6=97=B6=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 放宽SSE超时时间至5分钟，避免长生成任务被中断 2. 抽离LLM生成逻辑到独立线程池，避免阻塞SSE主线程 3. 添加心跳机制防止连接因空闲超时断开 4. 重构代码结构，提取生成工具方法，优化变量命名 5. 新增任务取消与异常捕获逻辑，完善错误处理 --- .../controller/GenerationController.java | 116 ++++++++++++------ 1 file changed, 78 insertions(+), 38 deletions(-) diff --git a/backend/generation-service/src/main/java/com/artedu/generation/controller/GenerationController.java b/backend/generation-service/src/main/java/com/artedu/generation/controller/GenerationController.java index 954e846..1ad45d9 100644 --- a/backend/generation-service/src/main/java/com/artedu/generation/controller/GenerationController.java +++ b/backend/generation-service/src/main/java/com/artedu/generation/controller/GenerationController.java @@ -28,8 +28,12 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; @Slf4j @@ -56,6 +60,7 @@ public class GenerationController { private QaConfig qaConfig; private final ExecutorService sseExecutor = Executors.newCachedThreadPool(); + private final ExecutorService llmExecutor = Executors.newCachedThreadPool(); private static final List> STYLES = List.of( Map.of("id", "friendly", "name", "亲和风格", "desc", "温暖亲切，像朋友聊天"), @@ -283,7 +288,8 @@ public class GenerationController { @RequestParam(value = "roomid", required = false) String roomid, @RequestParam(value = "staffName", required = false) String staffName) { log.info("流式生成多风格话术参数:{},{},{},{}", fromUser, toUser, roomid, staffName); - SseEmitter emitter = new SseEmitter(120000L); + // 生成过程可能超过 30s，SSE 总超时放宽到 5 分钟 + SseEmitter emitter = new SseEmitter(300000L); sseExecutor.execute(() -> { try { @@ -292,9 +298,9 @@ public class GenerationController { startEvent.put("styles", STYLES); emitter.send(SseEmitter.event().name("start").data(startEvent)); - List messages = buildChatContext(fromUser, toUser, roomid); + List messages = buildChatContext(fromUser, toUser, roomid); // 遍历消息：只取前三条是客户的消息，组装 查询知识库 - String external_str = ""; //原始问题 + StringBuilder externalStrBuilder = new StringBuilder(); //原始问题 // 历史消息上下文 List top15 = (messages == null || messages.isEmpty()) ? new ArrayList<>() @@ -317,7 +323,7 @@ public class GenerationController { if ("EXTERNAL".equals(messages.get(0).getFromRole())){ //第一条消息是客户的消息 则继续 if(count<3){ if ("EXTERNAL".equals(msg.getFromRole()) && is_external){ //判断判断前三条是否只包含客户消息 - external_str += msg.getContent()+","; + externalStrBuilder.append(msg.getContent()).append(","); } if (!"EXTERNAL".equals(msg.getFromRole())){ //判断前三条是否包含非客户的消息，如果包含，则不需要后续的消息 is_external = false; @@ -327,33 +333,7 @@ public class GenerationController { count++; } String chatContext = sb.toString(); - String generated = ""; - //如果最后几条都是客户消息返回不为空，则直接返回结果，否则走原逻辑 - if (!StringUtils.isBlank(external_str)){ //不为空 发起知识库的查询 - String answer = getAnswer(top15, external_str); - log.info("原问题：{},命中知识库结果:{}", external_str, answer); - if(!StringUtils.isBlank(answer)){ - ObjectMapper mapper = new ObjectMapper(); - List list = mapper.readValue(answer, new TypeReference>(){}); - //answer = answer.replaceAll("\"", "").replace("[","").replace("]",""); - for (String li : list){ - String promptNew = promptEngine.buildAnswerPrompt(li); - String res = qianwenClient.generate(promptNew, 800); - generated = generated + res + "|||"; - } - if(StringUtils.isNotBlank(generated)){ - String[] parts = generated.split("\\|\\|\\|"); - if(parts.length<3){ - for (int i = 0; i < 3-parts.length; i++) { - String systemPrompt = promptEngine.buildSystemPrompt(chatContext, staffName); - String generatedNew = qianwenClient.generate(systemPrompt, "", 1200); - generated = generated + generatedNew; - } - } - } - - } - } + final String external_str = externalStrBuilder.toString(); if (chatContext == null || chatContext.isEmpty()) { Map err = new HashMap<>(); @@ -363,15 +343,29 @@ public class GenerationController { emitter.complete(); return; } - if(StringUtils.isBlank(generated)){ - String systemPrompt = promptEngine.buildSystemPrompt(chatContext, staffName); - String userPrompt = promptEngine.buildMultiStyleUserPrompt(staffName); - long startTime = System.currentTimeMillis(); - generated = qianwenClient.generate(systemPrompt, userPrompt, 1200); - log.info("多风格话术生成完成, elapsed={}ms", System.currentTimeMillis() - startTime); + // LLM 调用在独立线程中同步执行，避免阻塞 SSE 线程 + Future generationFuture = llmExecutor.submit(() -> + generateUtteranceText(top15, chatContext, external_str, staffName)); + emitter.onCompletion(() -> generationFuture.cancel(true)); + emitter.onTimeout(() -> generationFuture.cancel(true)); + + // 等待生成结果，每 10s 发送一次心跳，防止 nginx/移动网络因 30s 空闲超时而断开 + while (!generationFuture.isDone()) { + try { + generationFuture.get(10, TimeUnit.SECONDS); + break; + } catch (TimeoutException e) { + try { + emitter.send(SseEmitter.event().comment("keep-alive")); + } catch (IOException ioe) { + log.warn("SSE 心跳发送失败，连接已断开"); + generationFuture.cancel(true); + return; + } + } } - + String generated = generationFuture.get(); List utterances = parseMultiStyle(generated); if (utterances.size() < 3) { @@ -416,6 +410,8 @@ public class GenerationController { emitter.send(SseEmitter.event().name("done").data(done)); emitter.complete(); + } catch (CancellationException e) { + log.warn("SSE 流式生成被取消（客户端断开或超时）"); } catch (Exception e) { log.error("流式生成失败", e); try { @@ -431,6 +427,50 @@ public class GenerationController { return emitter; } + /** + * 在后台线程执行 LLM 生成，返回最终话术文本 + */ + private String generateUtteranceText(List top15, String chatContext, + String external_str, String staffName) throws Exception { + String generated = ""; + //如果最后几条都是客户消息返回不为空，则直接返回结果，否则走原逻辑 + if (!StringUtils.isBlank(external_str)) { //不为空 发起知识库的查询 + String answer = getAnswer(top15, external_str); + log.info("原问题：{},命中知识库结果:{}", external_str, answer); + if (!StringUtils.isBlank(answer)) { + ObjectMapper mapper = new ObjectMapper(); + List list = mapper.readValue(answer, new TypeReference>(){}); + //answer = answer.replaceAll("\"", "").replace("[","").replace("]",""); + for (String li : list) { + String promptNew = promptEngine.buildAnswerPrompt(li); + String res = qianwenClient.generate(promptNew, 800); + generated = generated + res + "|||"; + } + if (StringUtils.isNotBlank(generated)) { + String[] parts = generated.split("\\|\\|\\|"); + if (parts.length < 3) { + for (int i = 0; i < 3 - parts.length; i++) { + String systemPrompt = promptEngine.buildSystemPrompt(chatContext, staffName); + String generatedNew = qianwenClient.generate(systemPrompt, "", 1200); + generated = generated + generatedNew; + } + } + } + + } + } + + if (StringUtils.isBlank(generated)) { + String systemPrompt = promptEngine.buildSystemPrompt(chatContext, staffName); + String userPrompt = promptEngine.buildMultiStyleUserPrompt(staffName); + + long startTime = System.currentTimeMillis(); + generated = qianwenClient.generate(systemPrompt, userPrompt, 1200); + log.info("多风格话术生成完成, elapsed={}ms", System.currentTimeMillis() - startTime); + } + return generated; + } + /** * 解析多风格话术，按 "|||" 分隔 */