refactor(generation-controller): 优化SSE流式生成逻辑,提升稳定性与超时处理

1.  放宽SSE超时时间至5分钟,避免长生成任务被中断
2.  抽离LLM生成逻辑到独立线程池,避免阻塞SSE主线程
3.  添加心跳机制防止连接因空闲超时断开
4.  重构代码结构,提取生成工具方法,优化变量命名
5.  新增任务取消与异常捕获逻辑,完善错误处理
This commit is contained in:
jqb 2026-07-27 13:42:03 +08:00
parent 8975e69aab
commit 543af54091

View File

@ -28,8 +28,12 @@ import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; 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; import java.util.stream.Collectors;
@Slf4j @Slf4j
@ -56,6 +60,7 @@ public class GenerationController {
private QaConfig qaConfig; private QaConfig qaConfig;
private final ExecutorService sseExecutor = Executors.newCachedThreadPool(); private final ExecutorService sseExecutor = Executors.newCachedThreadPool();
private final ExecutorService llmExecutor = Executors.newCachedThreadPool();
private static final List<Map<String, String>> STYLES = List.of( private static final List<Map<String, String>> STYLES = List.of(
Map.of("id", "friendly", "name", "亲和风格", "desc", "温暖亲切,像朋友聊天"), Map.of("id", "friendly", "name", "亲和风格", "desc", "温暖亲切,像朋友聊天"),
@ -283,7 +288,8 @@ public class GenerationController {
@RequestParam(value = "roomid", required = false) String roomid, @RequestParam(value = "roomid", required = false) String roomid,
@RequestParam(value = "staffName", required = false) String staffName) { @RequestParam(value = "staffName", required = false) String staffName) {
log.info("流式生成多风格话术参数:{},{},{},{}", fromUser, toUser, roomid, staffName); log.info("流式生成多风格话术参数:{},{},{},{}", fromUser, toUser, roomid, staffName);
SseEmitter emitter = new SseEmitter(120000L); // 生成过程可能超过 30s,SSE 总超时放宽到 5 分钟
SseEmitter emitter = new SseEmitter(300000L);
sseExecutor.execute(() -> { sseExecutor.execute(() -> {
try { try {
@ -292,9 +298,9 @@ public class GenerationController {
startEvent.put("styles", STYLES); startEvent.put("styles", STYLES);
emitter.send(SseEmitter.event().name("start").data(startEvent)); emitter.send(SseEmitter.event().name("start").data(startEvent));
List<ArchiveMessage> messages = buildChatContext(fromUser, toUser, roomid); List<ArchiveMessage> messages = buildChatContext(fromUser, toUser, roomid);
// 遍历消息:只取前三条是客户的消息,组装 查询知识库 // 遍历消息:只取前三条是客户的消息,组装 查询知识库
String external_str = ""; //原始问题 StringBuilder externalStrBuilder = new StringBuilder(); //原始问题
// 历史消息上下文 // 历史消息上下文
List<ArchiveMessage> top15 = (messages == null || messages.isEmpty()) List<ArchiveMessage> top15 = (messages == null || messages.isEmpty())
? new ArrayList<>() ? new ArrayList<>()
@ -317,7 +323,7 @@ public class GenerationController {
if ("EXTERNAL".equals(messages.get(0).getFromRole())){ //第一条消息是客户的消息 则继续 if ("EXTERNAL".equals(messages.get(0).getFromRole())){ //第一条消息是客户的消息 则继续
if(count<3){ if(count<3){
if ("EXTERNAL".equals(msg.getFromRole()) && is_external){ //判断判断前三条是否只包含客户消息 if ("EXTERNAL".equals(msg.getFromRole()) && is_external){ //判断判断前三条是否只包含客户消息
external_str += msg.getContent()+","; externalStrBuilder.append(msg.getContent()).append(",");
} }
if (!"EXTERNAL".equals(msg.getFromRole())){ //判断前三条是否包含非客户的消息,如果包含,则不需要后续的消息 if (!"EXTERNAL".equals(msg.getFromRole())){ //判断前三条是否包含非客户的消息,如果包含,则不需要后续的消息
is_external = false; is_external = false;
@ -327,33 +333,7 @@ public class GenerationController {
count++; count++;
} }
String chatContext = sb.toString(); String chatContext = sb.toString();
String generated = ""; final String external_str = externalStrBuilder.toString();
//如果最后几条都是客户消息返回不为空,则直接返回结果,否则走原逻辑
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<String> list = mapper.readValue(answer, new TypeReference<List<String>>(){});
//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 (chatContext == null || chatContext.isEmpty()) { if (chatContext == null || chatContext.isEmpty()) {
Map<String, Object> err = new HashMap<>(); Map<String, Object> err = new HashMap<>();
@ -363,15 +343,29 @@ public class GenerationController {
emitter.complete(); emitter.complete();
return; return;
} }
if(StringUtils.isBlank(generated)){
String systemPrompt = promptEngine.buildSystemPrompt(chatContext, staffName);
String userPrompt = promptEngine.buildMultiStyleUserPrompt(staffName);
long startTime = System.currentTimeMillis(); // LLM 调用在独立线程中同步执行,避免阻塞 SSE 线程
generated = qianwenClient.generate(systemPrompt, userPrompt, 1200); Future<String> generationFuture = llmExecutor.submit(() ->
log.info("多风格话术生成完成, elapsed={}ms", System.currentTimeMillis() - startTime); 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<String> utterances = parseMultiStyle(generated); List<String> utterances = parseMultiStyle(generated);
if (utterances.size() < 3) { if (utterances.size() < 3) {
@ -416,6 +410,8 @@ public class GenerationController {
emitter.send(SseEmitter.event().name("done").data(done)); emitter.send(SseEmitter.event().name("done").data(done));
emitter.complete(); emitter.complete();
} catch (CancellationException e) {
log.warn("SSE 流式生成被取消(客户端断开或超时)");
} catch (Exception e) { } catch (Exception e) {
log.error("流式生成失败", e); log.error("流式生成失败", e);
try { try {
@ -431,6 +427,50 @@ public class GenerationController {
return emitter; return emitter;
} }
/**
* 在后台线程执行 LLM 生成,返回最终话术文本
*/
private String generateUtteranceText(List<ArchiveMessage> 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<String> list = mapper.readValue(answer, new TypeReference<List<String>>(){});
//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;
}
/** /**
* 解析多风格话术,按 "|||" 分隔 * 解析多风格话术,按 "|||" 分隔
*/ */