流式流程接口 返回模式修改

java/storlead-knowledge/storlead-knowledge-core/src/main/java/com/storlead/knowledge/utils/HttpService.java

@@ -144,73 +144,73 @@ public class HttpService {
         postStream(url, authorization, body, emitter, null);
     }
 
-    public void postStream(
-            String url,
-            String authorization,
-            String body,
-            SseEmitter emitter,
-            java.util.function.Consumer<String> onMessageEnd
-    ) {
-        try {
-            HttpRequest request = HttpRequest.newBuilder()
-                    .uri(URI.create(url))
-                    .timeout(Duration.ofSeconds(200))
-                    .header("Authorization", authorization)
-                    .header("Accept", "text/event-stream")
-                    .header("Content-Type", "application/json")
-                    .POST(HttpRequest.BodyPublishers.ofString(body))
-                    .build();
-
-            HttpResponse<InputStream> response =
-                    httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream());
-            StringBuilder fullAnswerBuilder = new StringBuilder();
-            try (BufferedReader reader = new BufferedReader(
-                    new InputStreamReader(response.body(), StandardCharsets.UTF_8))) {
-
-                String line;
-
-                while ((line = reader.readLine()) != null) {
-
-                    if (line.startsWith("data:")) {
-                        String data = line.substring(5);
-
-                        if (!data.contains("message_end")) {
-                            try {
-                                JsonNode node = JacksonHolder.OBJECT_MAPPER.readTree(data);
-                                JsonNode answerNode = node.get("answer");
-                                if (answerNode != null && !answerNode.isNull()) {
-                                    fullAnswerBuilder.append(answerNode.asText());
-                                }
-                            } catch (Exception ignore) {
-                                // 非 JSON 或格式异常，跳过
-                            }
-                            //System.out.println(data);
-                            emitter.send(data);
-                        }
-
-                        if (data.contains("message_end")) {
-                            // 触发消息结束回调，传递完整数据
-                            ObjectNode endNode = (ObjectNode) JacksonHolder.OBJECT_MAPPER.readTree(data);
-                            endNode.put("answer", fullAnswerBuilder.toString());
-
-                            String finalData = JacksonHolder.OBJECT_MAPPER.writeValueAsString(endNode);
-                            if (onMessageEnd != null) {
-                                onMessageEnd.accept(finalData.replace("/",""));
-                            }
-                            emitter.send(finalData);
-                            emitter.complete();
-                            return;
-                        }
-                    }
-                }
-            }
-
-            emitter.complete();
-
-        } catch (Exception e) {
-            emitter.completeWithError(e);
-        }
-    }
+//    public void postStream(
+//            String url,
+//            String authorization,
+//            String body,
+//            SseEmitter emitter,
+//            java.util.function.Consumer<String> onMessageEnd
+//    ) {
+//        try {
+//            HttpRequest request = HttpRequest.newBuilder()
+//                    .uri(URI.create(url))
+//                    .timeout(Duration.ofSeconds(200))
+//                    .header("Authorization", authorization)
+//                    .header("Accept", "text/event-stream")
+//                    .header("Content-Type", "application/json")
+//                    .POST(HttpRequest.BodyPublishers.ofString(body))
+//                    .build();
+//
+//            HttpResponse<InputStream> response =
+//                    httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream());
+//            StringBuilder fullAnswerBuilder = new StringBuilder();
+//            try (BufferedReader reader = new BufferedReader(
+//                    new InputStreamReader(response.body(), StandardCharsets.UTF_8))) {
+//
+//                String line;
+//
+//                while ((line = reader.readLine()) != null) {
+//
+//                    if (line.startsWith("data:")) {
+//                        String data = line.substring(5);
+//
+//                        if (!data.contains("message_end")) {
+//                            try {
+//                                JsonNode node = JacksonHolder.OBJECT_MAPPER.readTree(data);
+//                                JsonNode answerNode = node.get("answer");
+//                                if (answerNode != null && !answerNode.isNull()) {
+//                                    fullAnswerBuilder.append(answerNode.asText());
+//                                }
+//                            } catch (Exception ignore) {
+//                                // 非 JSON 或格式异常，跳过
+//                            }
+//                            //System.out.println(data);
+//                            emitter.send(data);
+//                        }
+//
+//                        if (data.contains("message_end")) {
+//                            // 触发消息结束回调，传递完整数据
+//                            ObjectNode endNode = (ObjectNode) JacksonHolder.OBJECT_MAPPER.readTree(data);
+//                            endNode.put("answer", fullAnswerBuilder.toString());
+//
+//                            String finalData = JacksonHolder.OBJECT_MAPPER.writeValueAsString(endNode);
+//                            if (onMessageEnd != null) {
+//                                onMessageEnd.accept(finalData.replace("/",""));
+//                            }
+//                            emitter.send(finalData);
+//                            emitter.complete();
+//                            return;
+//                        }
+//                    }
+//                }
+//            }
+//
+//            emitter.complete();
+//
+//        } catch (Exception e) {
+//            emitter.completeWithError(e);
+//        }
+//    }
 
     /* ================= PUT ================= */
 
@@ -384,4 +384,178 @@ public class HttpService {
 
         return out.toByteArray();
     }
+
+
+    /**
+     * 流式请求 Dify 工作流，并实时过滤 <think> 标签内容
+     *
+     * @param url            Dify 工作流 API 地址
+     * @param authorization  认证 Token（可为 null）
+     * @param body           请求体（JSON）
+     * @param emitter        SseEmitter 用于推送实时过滤后的文本
+     * @param onMessageEnd   回调，传递最终过滤后的完整文本
+     */
+    public void postStream(
+            String url,
+            String authorization,
+            String body,
+            SseEmitter emitter,
+            Consumer<String> onMessageEnd
+    ) {
+        try {
+            // 构建 HTTP 请求
+            HttpRequest.Builder builder = HttpRequest.newBuilder()
+                    .uri(URI.create(url))
+                    .timeout(Duration.ofSeconds(300))
+                    .header("Accept", "text/event-stream")
+                    .header("Content-Type", "application/json")
+                    .POST(HttpRequest.BodyPublishers.ofString(body));
+            if (authorization != null && !authorization.isEmpty()) {
+                builder.header("Authorization", authorization);
+            }
+            HttpRequest request = builder.build();
+
+            // 发送请求并获取流式响应
+            HttpResponse<InputStream> response =
+                    httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream());
+
+            // 创建 Think 过滤器
+            ThinkFilter filter = new ThinkFilter();
+
+            try (BufferedReader reader = new BufferedReader(
+                    new InputStreamReader(response.body(), StandardCharsets.UTF_8))) {
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    if (line.isBlank()) continue;
+
+                    // 解析 SSE 数据行（格式：data: {...}）
+                    if (line.startsWith("data:")) {
+                        String jsonStr = line.substring(5).trim();
+                        if (jsonStr.isEmpty()) continue;
+
+                        try {
+                            JsonNode root = objectMapper.readTree(jsonStr);
+                            String event = root.path("event").asText();
+                            JsonNode dataNode = root.path("data");
+
+                            // 处理文本片段
+                            if ("text_chunk".equals(event)) {
+                                String chunk = dataNode.path("text").asText();
+                                if (chunk != null && !chunk.isEmpty()) {
+                                    // 实时过滤 chunk，返回新增的干净文本
+                                    String cleanPart = filter.processChunk(chunk);
+                                    if (!cleanPart.isEmpty()) {
+                                        emitter.send(cleanPart); // 推送实时增量
+                                    }
+                                }
+                            }
+
+                            // 处理工作流结束
+                            if ("workflow_finished".equals(event)) {
+                                // 获取最终完整过滤结果（处理缓冲区残留）
+                                String finalClean = filter.getFinalText();
+                                // 触发回调
+                                if (onMessageEnd != null) {
+                                    onMessageEnd.accept(finalClean);
+                                }
+                                // 推送最终完整结果（可选）
+                                //emitter.send(finalClean);
+                                emitter.complete();
+                                return;
+                            }
+                        } catch (Exception ignore) {
+                            // 单条 JSON 解析失败，跳过
+                        }
+                    }
+                }
+                // 如果流意外结束，兜底完成
+                emitter.complete();
+            }
+
+        } catch (Exception e) {
+            emitter.completeWithError(e);
+        }
+    }
+
+    /**
+     * 内部类：实时过滤 <think>...</think> 标签，支持跨 Chunk 处理
+     */
+    private static class ThinkFilter {
+        private final StringBuilder buffer = new StringBuilder();   // 用于累积未闭合的标签片段
+        private boolean inThink = false;                           // 是否当前处于 think 块内
+        private final StringBuilder result = new StringBuilder();   // 累积已输出的干净文本
+        private int lastSentLength = 0;                            // 已推送长度（用于增量）
+
+        /**
+         * 处理一个新的文本块，返回新增的过滤后文本（用于实时推送）
+         */
+        public String processChunk(String chunk) {
+            // 将新块追加到缓冲区，然后按字符处理
+            buffer.append(chunk);
+            StringBuilder output = new StringBuilder();
+            int i = 0;
+            while (i < buffer.length()) {
+                // 检查是否匹配 <think> 开始标签
+                if (!inThink && startsWithTag(buffer, i, "<think>")) {
+                    inThink = true;
+                    i += 6; // 跳过 "<think>"
+                    // 清除缓冲区中已消费的部分
+                    buffer.delete(0, i);
+                    i = 0;
+                    continue;
+                }
+                // 检查是否匹配 </think> 结束标签
+                if (inThink && startsWithTag(buffer, i, "</think>")) {
+                    inThink = false;
+                    i += 8; // 跳过 "</think>"
+                    buffer.delete(0, i);
+                    i = 0;
+                    continue;
+                }
+
+                // 如果不在 think 内，则输出字符
+                if (!inThink) {
+                    char c = buffer.charAt(i);
+                    output.append(c);
+                    result.append(c); // 同时保存到完整结果
+                }
+                i++;
+            }
+            // 清空缓冲区（所有字符已处理）
+            buffer.setLength(0);
+
+            // 获取新增部分（上次发送位置之后）
+            String newPart = output.toString();
+            return newPart;
+        }
+
+        /**
+         * 获取完整过滤后的文本（在流结束时调用，处理可能的剩余缓冲区）
+         */
+        public String getFinalText() {
+            // 如果缓冲区还有未处理的内容（通常是未闭合标签，直接丢弃）
+            // 但为了安全，也可尝试补全，此处简单丢弃
+            if (buffer.length() > 0) {
+                // 如果不在 think 内，且缓冲区有内容，则可能是标签的一部分，但已无法匹配，直接丢弃
+                // 实际很少出现，保留以防万一
+            }
+            return result.toString();
+        }
+
+        /**
+         * 辅助方法：检查 buffer 从 offset 开始是否匹配指定的 tag
+         */
+        private boolean startsWithTag(StringBuilder buffer, int offset, String tag) {
+            int tagLen = tag.length();
+            if (offset + tagLen > buffer.length()) {
+                return false;
+            }
+            for (int i = 0; i < tagLen; i++) {
+                if (buffer.charAt(offset + i) != tag.charAt(i)) {
+                    return false;
+                }
+            }
+            return true;
+        }
+    }
 }