diff --git a/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java b/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java index 37c45a1..e1c6a02 100644 --- a/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java +++ b/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java @@ -7,6 +7,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.io.IOException; import java.util.Map; @Service @@ -23,32 +24,23 @@ public class LLMAppServiceImpl implements LLMAppService { @Override public SseEmitter ask(String llmToken, String question, Map params) throws Exception { - SseEmitter emitter = new SseEmitter(0L); // 不超时 + SseEmitter emitter = new SseEmitter(0L); // 不设超时 - try { - FilteredSseOutputAdapter adapter = new FilteredSseOutputAdapter(emitter); - // 异步执行,避免阻塞返回 - new Thread(() -> { + new Thread(() -> { + try { + FilteredSseOutputAdapter adapter = new FilteredSseOutputAdapter(emitter); + llmServiceFactory.current().streamAnswer(llmToken, question, params, adapter); + } catch (Exception e) { + log.error("LLM调用异常", e); try { - llmServiceFactory.current().streamAnswer(llmToken, question, params, adapter); - } catch (Exception e) { - log.error("LLM流式调用异常", e); - try { - emitter.send(SseEmitter.event().data("{\"error\": \"LLM异常\"}")); - } catch (Exception ignored) { - log.error("error", ignored); - } + emitter.send(SseEmitter.event().data("{\"error\": \"LLM异常\"}")); emitter.completeWithError(e); + } catch (IOException ioException) { + log.warn("SSE发送错误信息失败", ioException); } - }).start(); - - } catch (Exception e) { - log.error("LLM初始化异常", e); - emitter.send(SseEmitter.event().data("{\"error\": \"LLM异常\"}")); - emitter.completeWithError(e); - } + } + }).start(); return emitter; } } - diff --git a/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java b/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java index 802b80e..f3e84fb 100644 --- a/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java +++ b/src/main/java/com/knowledge/base/infrastructure/util/http/FilteredSseOutputAdapter.java @@ -15,7 +15,8 @@ public class FilteredSseOutputAdapter implements WriterAdapter { private final SseEmitter emitter; private static final ObjectMapper mapper = new ObjectMapper(); - private Map lastChunk = new HashMap<>(); + private final Map lastChunk = new HashMap<>(); + private boolean closed = false; public FilteredSseOutputAdapter(SseEmitter emitter) { this.emitter = emitter; @@ -41,27 +42,30 @@ public class FilteredSseOutputAdapter implements WriterAdapter { lastChunk.put("textResponse", original.get("textResponse")); } - // 只缓存 sources,不立即发送 + // sources 只保存不立刻发 if (original.containsKey("sources")) { lastChunk.put("sources", original.get("sources")); } - if(original.containsKey("close")) { - filtered.put("close", original.get("close")); + // close 也缓存并判断是否最后一条 + if (original.containsKey("close")) { lastChunk.put("close", original.get("close")); } - // 最后一条才合并 sources 输出 if (Boolean.TRUE.equals(original.get("close"))) { - // 合并缓存中的 sources + // 合并缓存数据并一次性输出 filtered.putAll(lastChunk); - lastChunk.clear(); // 清空缓存 + lastChunk.clear(); + String payload = mapper.writeValueAsString(filtered); emitter.send(SseEmitter.event().data(payload)); - // 增加主动关闭连接逻辑 - emitter.complete(); + + if (!closed) { + closed = true; + emitter.complete(); + } } else if (!filtered.isEmpty()) { - // 普通中间片段 + // 中间段 String payload = mapper.writeValueAsString(filtered); emitter.send(SseEmitter.event().data(payload)); }