调整回答的速度

backend-java/src/main/java/com/pharmacopoeia/controller/ChatController.java

@@ -15,7 +15,7 @@ import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Mono;
 
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.web.multipart.MultipartFile;
@@ -159,61 +159,69 @@ public class ChatController {
         final long t0 = System.currentTimeMillis();
         log.info("[chatStream] intent={}, query={}", intent, query.substring(0, Math.min(50, query.length())));
 
-        return retrieverService.searchReactive(query, intent, 20)
+        // 先发射 intent/status 事件，再用 flatMapMany 接回管道保持取消链完整。
+        // 纯 Reactor 管道（零裸 subscribe），连接断开时整条链路自动取消到百炼。
+        return Flux.just(
+                ServerSentEvent.<String>builder().event("intent").data(intent).build(),
+                ServerSentEvent.<String>builder().event("status").data("Retrieving...").build()
+        ).concatWith(retrieverService.searchReactive(query, intent, 20)
                 .map(docs -> rerankerService.rerank(docs, query, 5))
                 .flatMapMany(docs -> {
                     long t2 = System.currentTimeMillis();
                     log.info("[chatStream] search+rerank done, docs={}, elapsed={}ms", docs.size(), t2 - t0);
-                    Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
-
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intent).build());
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("Retrieving...").build());
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("Matched " + docs.size() + " records, generating...").build());
 
                     final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
-                    StringBuilder fullAnswer = new StringBuilder();
                     final long t3 = System.currentTimeMillis();
                     log.info("[chatStream] prompt built, elapsed={}ms, chars={}",
                             t3 - t0, messages.stream().mapToInt(m -> m.get("content").length()).sum());
 
-                    llmService.chatStream(messages)
-                            .doOnNext(token -> {
-                                fullAnswer.append(token);
-                                sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
-                            })
-                            .doOnComplete(() -> {
-                                final List<Map<String, Object>> sources = buildSources(docs);
-                                try {
-                                    String meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
-                                            "intent", intent,
-                                            "sources", sources,
-                                            "conversation_id", cid
-                                    ));
-                                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
-                                } catch (Exception ignored) {}
-
-                                String finalAnswer = cleanAnswer(fullAnswer.toString());
-                                long t4 = System.currentTimeMillis();
-                                log.info("[chatStream] chatStream done, llmElapsed={}ms, total={}ms", t4 - t3, t4 - t0);
-                                persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
-                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
+                    // 用 StringBuilder 攒完整答案（Flux 内部同步操作，无线程安全问题）
+                    var fullAnswerBuf = new StringBuilder();
+
+                    // 纯管道拼接，无裸 subscribe：
+                    //   [status] → [token1, token2, ...] → [meta] → complete
+                    Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages)
+                            .map(token -> {
+                                fullAnswerBuf.append(token);
+                                return ServerSentEvent.<String>builder().data(token).build();
+                            });
 
-                                // 写入全局缓存 + 清除 PENDING 标记
-                                qaCache.put(normalized, finalAnswer, intent, sources);
+                    Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
+                        String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
+                        final List<Map<String, Object>> sources = buildSources(docs);
+                        String meta;
+                        try {
+                            meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
+                                    "intent", intent,
+                                    "sources", sources,
+                                    "conversation_id", cid
+                            ));
+                        } catch (Exception e) {
+                            meta = "{}";
+                        }
+                        long t4 = System.currentTimeMillis();
+                        log.info("[chatStream] chatStream done, llmElapsed={}ms, total={}ms", t4 - t3, t4 - t0);
+                        persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
+                        persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
+                        qaCache.put(normalized, finalAnswer, intent, sources);
+
+                        return Flux.just(
+                                ServerSentEvent.<String>builder().event("meta").data(meta).build()
+                        );
+                    });
+
+                    Flux<ServerSentEvent<String>> headFlux = Flux.just(
+                            ServerSentEvent.<String>builder().event("status")
+                                    .data("Matched " + docs.size() + " records, generating...").build()
+                    );
 
-                                sink.tryEmitComplete();
-                            })
+                    return Flux.concat(headFlux, contentFlux, tailFlux)
                             .doOnError(e -> {
                                 long t4 = System.currentTimeMillis();
-                                log.error("[chatStream] LLM error, totalElapsed={}ms, error={}", t4 - t0, e.getMessage());
-                                // 异常时清除 PENDING 标记，避免后续同问题请求被锁死
+                                log.error("[chatStream] error, totalElapsed={}ms, error={}", t4 - t0, e.getMessage());
                                 qaCache.removePending(normalized);
-                                sink.tryEmitError(e);
-                            })
-                            .subscribe();
-
-                    return sink.asFlux();
-                });
+                            });
+                }));
     }
 
     /** 将缓存命中结果以流式 SSE 形式返回 */
@@ -299,80 +307,78 @@ public class ChatController {
                 ? request.getConversationId()
                 : UUID.randomUUID().toString();
 
-        Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
-
         // 校验图片数据
         if (request.getImageBase64() == null || request.getImageBase64().isBlank()) {
-            sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("图片数据为空，请重新上传").build());
-            sink.tryEmitComplete();
-            return sink.asFlux();
+            return Flux.just(
+                    ServerSentEvent.<String>builder().event("status").data("图片数据为空，请重新上传").build()
+            );
         }
 
-        sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("正在分析图片（OCR 文字识别）...").build());
-
-        // Step 1: Qwen VL 分析图片（流式），收集完整结果
-        StringBuilder ocrBuilder = new StringBuilder();
-        llmService.analyzeImageStream(request.getImageBase64(), request.getMimeType(),
-                "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等。简要输出。")
-                .doOnNext(ocrBuilder::append)
-                .collectList()
-                .flatMap(tokens -> {
-                    String ocrText = ocrBuilder.toString();
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("图片分析完成，正在检索药典知识库...").build());
-
-                    // Step 2: 拼接查询 → RAG（响应式）
-                    String query = (!request.getMessage().isBlank())
-                            ? request.getMessage() + "\n\n（图片OCR提取内容：" + ocrText + "）"
-                            : ocrText;
-                    final String intent = retrieverService.classifyIntent(query);
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intent).build());
-
-                    return retrieverService.searchReactive(query, intent, 20)
-                            .map(docs -> rerankerService.rerank(docs, query, 5))
-                            .map(docs -> {
-                                sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
-                                        .data("已匹配 " + docs.size() + " 条药典资料，生成回答中（已启用联网搜索）...").build());
-
-                                // Step 3: 构建 Prompt + 联网搜索流式生成
-                                final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
-                                String imageContext = "\n\n【图片分析结果】\n" + ocrText + "\n";
-                                messages.getFirst().put("content", messages.getFirst().get("content") + imageContext);
-
-                                StringBuilder fullAnswer = new StringBuilder();
-                                llmService.chatStream(messages, true)
-                                        .doOnNext(token -> {
-                                            fullAnswer.append(token);
-                                            sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
-                                        })
-                                        .doOnComplete(() -> {
-                                            final List<Map<String, Object>> sources = buildSources(docs);
-                                            try {
-                                                String meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
-                                                        "intent", intent,
-                                                        "sources", sources,
-                                                        "conversation_id", cid,
-                                                        "ocr_text", ocrText
-                                                ));
-                                                sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
-                                            } catch (Exception ignored) {}
-
-                                            String finalAnswer = cleanAnswer(fullAnswer.toString());
-                                            persistenceService.saveMessage(userKey, cid, "user",
-                                                    request.getMessage().isBlank() ? "[图片]" : "[图片] " + request.getMessage(),
-                                                    intent, null);
-                                            persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
-                                            sink.tryEmitComplete();
-                                        })
-                                        .doOnError(sink::tryEmitError)
-                                        .subscribe();
-
-                                return docs; // dummy return for map
-                            });
-                })
-                .doOnError(sink::tryEmitError)
-                .subscribe();
-
-        return sink.asFlux();
+        // 先发状态，不等图片分析完成
+        return Flux.just(
+                ServerSentEvent.<String>builder().event("status").data("正在分析图片（OCR 文字识别）...").build()
+        ).concatWith(
+            llmService.analyzeImageStream(request.getImageBase64(), request.getMimeType(),
+                    "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等。简要输出。")
+                    .collectList()
+                    .flatMapMany(tokens -> {
+                        String ocrText = String.join("", tokens);
+
+                        String query = (!request.getMessage().isBlank())
+                                ? request.getMessage() + "\n\n（图片OCR提取内容：" + ocrText + "）"
+                                : ocrText;
+                        final String intent = retrieverService.classifyIntent(query);
+
+                        return Flux.just(
+                                ServerSentEvent.<String>builder().event("status").data("图片分析完成，正在检索药典知识库...").build(),
+                                ServerSentEvent.<String>builder().event("intent").data(intent).build()
+                        ).concatWith(retrieverService.searchReactive(query, intent, 20)
+                                .map(docs -> rerankerService.rerank(docs, query, 5))
+                                .flatMapMany(docs -> {
+                                    final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
+                                    String imageContext = "\n\n【图片分析结果】\n" + ocrText + "\n";
+                                    messages.getFirst().put("content", messages.getFirst().get("content") + imageContext);
+
+                                    var fullAnswerBuf = new StringBuilder();
+
+                                    Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, true)
+                                            .map(token -> {
+                                                fullAnswerBuf.append(token);
+                                                return ServerSentEvent.<String>builder().data(token).build();
+                                            });
+
+                                    Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
+                                        String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
+                                        final List<Map<String, Object>> sources = buildSources(docs);
+                                        String meta;
+                                        try {
+                                            meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
+                                                    "intent", intent,
+                                                    "sources", sources,
+                                                    "conversation_id", cid,
+                                                    "ocr_text", ocrText
+                                            ));
+                                        } catch (Exception e) {
+                                            meta = "{}";
+                                        }
+                                        persistenceService.saveMessage(userKey, cid, "user",
+                                                request.getMessage().isBlank() ? "[图片]" : "[图片] " + request.getMessage(),
+                                                intent, null);
+                                        persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
+                                        return Flux.just(
+                                                ServerSentEvent.<String>builder().event("meta").data(meta).build()
+                                        );
+                                    });
+
+                                    Flux<ServerSentEvent<String>> headFlux = Flux.just(
+                                            ServerSentEvent.<String>builder().event("status")
+                                                    .data("已匹配 " + docs.size() + " 条药典资料，生成回答中（已启用联网搜索）...").build()
+                                    );
+
+                                    return Flux.concat(headFlux, contentFlux, tailFlux);
+                                }));
+                    })
+        );
     }
 
     // ============================================================
@@ -446,37 +452,111 @@ public class ChatController {
                 && request.getMediaType() != null && !request.getMediaType().isBlank();
         final String mediaLabel = hasMedia && "video".equals(request.getMediaType()) ? "视频" : "图片";
 
-        Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
+        if (hasMedia) {
+            // 先发 OCR section header
+            return Flux.just(
+                    ServerSentEvent.<String>builder().event("status")
+                            .data("🔍 正在分析" + mediaLabel + "...").build(),
+                    ServerSentEvent.<String>builder().data("【📷 " + mediaLabel + "分析】\n\n").build()
+            ).concatWith(
+                    // 用 collectList 收集流式结果，同时攒 OCR 文本，全程在管道内无阻塞
+                    llmService.analyzeMediaStream(request.getMediaBase64(), request.getMediaType(),
+                            request.getMediaMime(), "")
+                            .flatMap(token -> Flux.just(
+                                    ServerSentEvent.<String>builder().data(token).build()  // 前端实时看到
+                            ))
+                            .concatWith(Flux.just(ServerSentEvent.<String>builder().data("\n\n").build()))
+                            // collectList 后再 flatMapMany 接 RAG——Reactor 取消时自动 abort
+                            .collectList()
+                            .flatMapMany(events -> {
+                                // 从已发送的事件中拼回 OCR 文本（不额外调 API）
+                                StringBuilder ocrBuilder = new StringBuilder();
+                                for (ServerSentEvent<String> e : events) {
+                                    String d = e.data();
+                                    if (d != null && !d.equals("\n\n")) ocrBuilder.append(d);
+                                }
+                                String mediaOcr = ocrBuilder.toString().trim();
+                                // 不重复发送 OCR 事件（已经通过 analyzeMediaStream 实时发送过了）
+                                return buildRagPipeline(cid,
+                                        request.getMessage(), mediaOcr, userKey, mediaLabel);
+                            })
+            );
+        } else {
+            return buildRagPipeline(cid, request.getMessage(), "", userKey, "");
+        }
+    }
 
-        try {
-            if (hasMedia) {
-                sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
-                        .data("🔍 正在分析" + mediaLabel + "...").build());
-                // 先发 OCR section header
-                sink.tryEmitNext(ServerSentEvent.<String>builder()
-                        .data("【📷 " + mediaLabel + "分析】\n\n").build());
-
-                StringBuilder ocrBuilder = new StringBuilder();
-                llmService.analyzeMediaStream(request.getMediaBase64(), request.getMediaType(),
-                        request.getMediaMime(), "")
-                        .doOnNext(token -> {
-                            ocrBuilder.append(token);
-                            sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
-                        })
-                        .doOnComplete(() -> {
-                            sink.tryEmitNext(ServerSentEvent.<String>builder().data("\n\n").build());
-                            doStreamAnswer(sink, cid, request, ocrBuilder.toString(), mediaLabel, userKey);
-                        })
-                        .doOnError(sink::tryEmitError)
-                        .subscribe();
-            } else {
-                doStreamAnswer(sink, cid, request, "", "", userKey);
-            }
-        } catch (Exception e) {
-            sink.tryEmitError(e);
+    /** 构建 RAG → 百炼流式管道（纯 Reactor，零裸 subscribe） */
+    private Flux<ServerSentEvent<String>> buildRagPipeline(String cid, String rawMsg,
+                                                          String ocrText, String userKey,
+                                                          String mediaLabel) {
+        String rq = rawMsg != null ? rawMsg.trim() : "";
+        if (!rq.isEmpty() && ocrText != null && !ocrText.isEmpty()) {
+            rq = rq + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
+        } else if (ocrText != null && !ocrText.isEmpty()) {
+            rq = ocrText;
+        } else if (rq.isEmpty()) {
+            rq = "请介绍一下自己";
         }
+        final String query = rq;
+        final String intent = retrieverService.classifyIntent(query);
+        final boolean enableSearch = (ocrText != null && !ocrText.isEmpty()) || props.isEnableWebSearch();
+
+        // 如果等待检索，先发状态
+        Flux<ServerSentEvent<String>> prefixFlux = (ocrText != null && !ocrText.isEmpty())
+                ? Flux.just(ServerSentEvent.<String>builder().event("status")
+                        .data("📚 检索药典知识库...").build())
+                : Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build());
+
+        Flux<ServerSentEvent<String>> intentFlux = (ocrText != null && !ocrText.isEmpty())
+                ? Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build())
+                : Flux.empty();
+
+        return prefixFlux.concatWith(intentFlux)
+                .concatWith(retrieverService.searchReactive(query, intent, 20)
+                        .map(docs -> rerankerService.rerank(docs, query, 5))
+                        .flatMapMany(docs -> {
+                            final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
+                            if (ocrText != null && !ocrText.isEmpty()) {
+                                messages.getFirst().put("content",
+                                        messages.getFirst().get("content") + "\n\n【" + mediaLabel + "分析结果】\n" + ocrText + "\n");
+                            }
+
+                            var fullAnswerBuf = new StringBuilder();
+
+                            Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, enableSearch)
+                                    .map(token -> {
+                                        fullAnswerBuf.append(token);
+                                        return ServerSentEvent.<String>builder().data(token).build();
+                                    });
+
+                            Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
+                                final List<Map<String, Object>> sources = buildSources(docs);
+                                String meta;
+                                try {
+                                    meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
+                                            "intent", intent, "sources", sources, "conversation_id", cid,
+                                            "ocr_text", ocrText != null ? ocrText : ""));
+                                } catch (Exception e) { meta = "{}"; }
+                                String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
+                                String userMsg = (rawMsg != null && !rawMsg.isBlank()) ? rawMsg
+                                        : (ocrText != null && !ocrText.isEmpty()) ? "[" + mediaLabel + "]" : "";
+                                persistenceService.saveMessage(userKey, cid, "user", userMsg, intent, null);
+                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
+                                return Flux.just(
+                                        ServerSentEvent.<String>builder().event("meta").data(meta).build()
+                                );
+                            });
+
+                            Flux<ServerSentEvent<String>> headFlux = Flux.just(
+                                    ServerSentEvent.<String>builder().data("\n【📚 药典参考回答】\n\n").build(),
+                                    ServerSentEvent.<String>builder().event("status")
+                                            .data("已匹配 " + docs.size() + " 条药典资料，生成回答中"
+                                                    + (enableSearch ? "（已启用联网搜索）" : "") + "...").build()
+                            );
 
-        return sink.asFlux();
+                            return Flux.concat(headFlux, contentFlux, tailFlux);
+                        }));
     }
 
     // ============================================================
@@ -565,72 +645,6 @@ public class ChatController {
         return chatAskMultimodal(req);
     }
 
-    /** 流式多模态：OCR 完成后，走 RAG + 生成 */
-    private void doStreamAnswer(Sinks.Many<ServerSentEvent<String>> sink, String cid,
-                                MultimodalChatRequest request, String ocrText, String mediaLabel,
-                                String userKey) {
-        if (!ocrText.isEmpty()) {
-            sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
-                    .data("📚 检索药典知识库...").build());
-        }
-
-        String rawQuery = request.getMessage() != null ? request.getMessage().trim() : "";
-        if (!rawQuery.isEmpty() && !ocrText.isEmpty()) {
-            rawQuery = rawQuery + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
-        } else if (!ocrText.isEmpty()) {
-            rawQuery = ocrText;
-        } else if (rawQuery.isEmpty()) {
-            rawQuery = "请介绍一下自己";
-        }
-        final String query = rawQuery;
-
-        final String intent = retrieverService.classifyIntent(query);
-        sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intent).build());
-
-        final boolean enableSearch = !ocrText.isEmpty() || props.isEnableWebSearch();
-
-        retrieverService.searchReactive(query, intent, 20)
-                .map(docs -> rerankerService.rerank(docs, query, 5))
-                .subscribe(docs -> {
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
-                            .data("已匹配 " + docs.size() + " 条药典资料，生成回答中"
-                                    + (enableSearch ? "（已启用联网搜索）" : "") + "...").build());
-
-                    final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
-                    if (!ocrText.isEmpty()) {
-                        messages.getFirst().put("content",
-                                messages.getFirst().get("content") + "\n\n【" + mediaLabel + "分析结果】\n" + ocrText + "\n");
-                    }
-
-                    // 发送回答 section header
-                    sink.tryEmitNext(ServerSentEvent.<String>builder().data("\n【📚 药典参考回答】\n\n").build());
-
-                    StringBuilder fullAnswer = new StringBuilder();
-                    llmService.chatStream(messages, enableSearch)
-                            .doOnNext(token -> {
-                                fullAnswer.append(token);
-                                sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
-                            })
-                            .doOnComplete(() -> {
-                                final List<Map<String, Object>> sources = buildSources(docs);
-                                try {
-                                    String meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
-                                            "intent", intent, "sources", sources, "conversation_id", cid,
-                                            "ocr_text", ocrText));
-                                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
-                                } catch (Exception ignored) {}
-                                String finalAnswer = cleanAnswer(fullAnswer.toString());
-                                String userMsg = !request.getMessage().isBlank() ? request.getMessage()
-                                        : !ocrText.isEmpty() ? "[" + mediaLabel + "]" : "";
-                                persistenceService.saveMessage(userKey, cid, "user", userMsg, intent, null);
-                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources);
-                                sink.tryEmitComplete();
-                            })
-                            .doOnError(sink::tryEmitError)
-                            .subscribe();
-                }, sink::tryEmitError);
-    }
-
     @GetMapping("/history")
     public ResponseEntity<Map<String, Object>> getHistory(
             @RequestParam(defaultValue = "1") int page,