From 37d800be69b5245a78bebe2bdfda706314baf62f Mon Sep 17 00:00:00 2001 From: lixiaolong <702156524@qq.com> Date: Wed, 15 Apr 2026 11:50:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E6=B7=BB=E5=8A=A0AI=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=B9=B3=E5=8F=B0=E6=94=AF=E6=8C=81=E5=B9=B6=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E6=B5=81=E5=BC=8F=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在FlowActionExecutorService中集成LLMAiAgentPlatformService和LLMAiTtsService - 实现AI_AGENT_PLATFORM类型的执行逻辑，支持代理平台查询功能 - 重构FlowHttpNodeHandler参数解析，支持JSON和表单数据格式 - 在FlowItemExecutor中清空起始节点输入定义 - 重写FlowNodeParamPreparer参数准备逻辑，支持嵌套参数处理 - 优化LlmChatService中的SSE流处理，实现句子级别异步处理 - 添加TTS语音播放功能，在流式响应中实时处理句子 - 修改评估服务调用方式，在FlowEndNodeHandler中注释掉原有逻辑 --- .../com/cmvr/llm/util/LlmChatService.java | 132 ++++++++++++++++-- .../dispatcher/FlowEndNodeHandler.java | 2 +- .../dispatcher/FlowHttpNodeHandler.java | 19 ++- .../flow/runtime/engine/FlowItemExecutor.java | 2 +- .../engine/support/FlowNodeParamPreparer.java | 89 +++++++----- .../service/FlowActionExecutorService.java | 6 + 6 files changed, 192 insertions(+), 58 deletions(-) diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java index aaa5747..8e9179c 100644 --- a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java @@ -1,7 +1,9 @@ package com.cmvr.llm.util; +import com.cmvr.llm.service.LLMAiTtsService; import com.fasterxml.jackson.databind.ObjectMapper; import okhttp3.*; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import javax.annotation.PreDestroy; @@ -10,9 +12,8 @@ import java.io.IOException; import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; /** @@ -20,7 +21,8 @@ import java.util.concurrent.atomic.AtomicReference; */ @Service public class LlmChatService { - + @Autowired + private LLMAiTtsService llmAiTtsService; private static final String DEFAULT_BASE_URL = "https://aiagentplatform.cmft.com/api/proxy/api/v1"; private static final MediaType JSON = MediaType.parse("application/json; charset=utf-8"); private static final ObjectMapper objectMapper = new ObjectMapper(); @@ -122,6 +124,9 @@ public class LlmChatService { } } + + // 句子结束符号 + static final String SENTENCE_END = "?!。？！\n"; /** * 执行SSE请求 - 完全保持原executeSseRequest逻辑，改为同步返回 */ @@ -137,18 +142,49 @@ public class LlmChatService { .build(); StringBuilder fullContent = new StringBuilder(); - CountDownLatch latch = new CountDownLatch(1); + CountDownLatch mainLatch = new CountDownLatch(1); AtomicReference errorRef = new AtomicReference<>(); AtomicReference finalMessageIdRef = new AtomicReference<>(""); + // ====================== 核心控制 ====================== + + // 接收缓存：永远不阻塞 + StringBuilder receiveBuffer = new StringBuilder(); + // 单线程串行执行：保证上一句处理完才处理下一句 + ExecutorService sentenceExecutor = Executors.newSingleThreadExecutor(); + // 标记流是否已经结束 + AtomicBoolean streamFinished = new AtomicBoolean(false); + final AtomicBoolean interrupted = new AtomicBoolean(false); // 中断标记 Call call = httpClient.newCall(request); + Runnable interruptTask = () -> { + interrupted.set(true); + call.cancel(); // 真正关闭SSE连接 【关键】 + streamFinished.set(true); + sentenceExecutor.shutdownNow(); + mainLatch.countDown(); + }; + + // 使用异步调用但阻塞等待结果 call.enqueue(new Callback() { + + /** + * 等待所有句子处理完再结束 + */ + private void shutdownAndWait1() { + try { + sentenceExecutor.shutdown(); + sentenceExecutor.awaitTermination(10, TimeUnit.MINUTES); + } catch (InterruptedException ignored) {} + mainLatch.countDown(); + } + @Override public void onFailure(Call call, IOException e) { errorRef.set(e); - latch.countDown(); + streamFinished.set(true); + shutdownAndWait1(); } @Override @@ -156,7 +192,8 @@ public class LlmChatService { if (!response.isSuccessful()) { String errorBody = response.body() != null ? response.body().string() : "无错误信息"; errorRef.set(new IOException("SSE请求失败: " + response.code() + " - " + errorBody)); - latch.countDown(); + streamFinished.set(true); + shutdownAndWait1(); return; } @@ -168,7 +205,7 @@ public class LlmChatService { new InputStreamReader(response.body().byteStream(), StandardCharsets.UTF_8))) { String line; - while ((line = reader.readLine()) != null) { + while (!interrupted.get() && (line = reader.readLine()) != null) { if (line.isEmpty()) { currentEvent = ""; continue; @@ -190,8 +227,9 @@ public class LlmChatService { if ("[DONE]".equals(data)) { fullContent.append(localContent); finalMessageIdRef.set(finalMessageId); - latch.countDown(); - return; + streamFinished.set(true); + processBufferIfNeed(); + break; } try { @@ -208,6 +246,8 @@ public class LlmChatService { String answer = extractString(jsonData, "answer", "content", "chunk", "text"); if (!answer.isEmpty()) { localContent.append(answer); + receiveBuffer.append(answer); + processBufferIfNeed(); } } else if ("message_start".equals(event)) { System.out.println("消息开始，ID: " + finalMessageId); @@ -216,7 +256,8 @@ public class LlmChatService { } else if ("message_end".equals(event) || "end".equals(event)) { fullContent.append(localContent); finalMessageIdRef.set(finalMessageId); - latch.countDown(); + streamFinished.set(true); + processBufferIfNeed(); return; } @@ -229,22 +270,76 @@ public class LlmChatService { // 流正常结束 fullContent.append(localContent); finalMessageIdRef.set(finalMessageId); - latch.countDown(); } catch (IOException e) { errorRef.set(e); - latch.countDown(); + streamFinished.set(true); + } finally { + processBufferIfNeed(); + shutdownAndWait1(); } } + + /** + * 核心：提取完整句子 + 异步串行处理 + */ + private void processBufferIfNeed() { + if (interrupted.get()) return; + if (!true) return; + + sentenceExecutor.submit(() -> { + while (true) { + String buffer = receiveBuffer.toString(); + // 寻找最后一个句子结束符 + int lastSplit = -1; + for (int i = 0; i < buffer.length(); i++) { + if (SENTENCE_END.contains(String.valueOf(buffer.charAt(i)))) { + lastSplit = i; + } + } + + // 有完整句子 或 流已结束 + if (lastSplit >= 0 || streamFinished.get()) { + String sentence; + if (lastSplit >= 0) { + // 截取完整句子 + sentence = buffer.substring(0, lastSplit + 1).trim(); + // 保留剩余内容 + String remain = buffer.substring(lastSplit + 1).trim(); + receiveBuffer.setLength(0); + receiveBuffer.append(remain); + } else { + // 流结束，直接处理剩余 + sentence = buffer.trim(); + receiveBuffer.setLength(0); + } + + if (!sentence.isEmpty()) { + if (interrupted.get()) return; + yourAsyncMethod(sentence); + } + + // 处理完继续循环，看是否还有新句子 + if (streamFinished.get() && receiveBuffer.length() == 0) { + break; + } + } else { + // 无完整句子，退出 + break; + } + } + }); + } }); // 等待流式传输完成 try { - boolean completed = latch.await(120, TimeUnit.SECONDS); + boolean completed = mainLatch.await(120, TimeUnit.SECONDS); if (!completed) { throw new IOException("请求超时"); } } catch (InterruptedException e) { + interruptTask.run(); // 被中断时，立即停止SSE Thread.currentThread().interrupt(); throw new IOException("请求被中断", e); } @@ -260,6 +355,15 @@ public class LlmChatService { return fullContent.toString(); } + private void yourAsyncMethod(String sentence) { + try { + System.out.println("正在处理句子: " + sentence); + llmAiTtsService.play("http://127.0.0.1:8080/tts/play", sentence, "x4_yezi", "50", "80"); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + /** * 完全保持原extractString方法 */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowEndNodeHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowEndNodeHandler.java index 8cdd839..0cf5f58 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowEndNodeHandler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowEndNodeHandler.java @@ -47,7 +47,7 @@ public class FlowEndNodeHandler implements FlowNodeTypeHandler { Thread.currentThread().setName("evaluation-thread-" + Thread.currentThread().getId()); try { log.info("异步评估任务开始执行：threadName={}, instId={}, itemId={}", Thread.currentThread().getName(), instId, itemId); - exAeEvaluationService.executeEvaluation(jsonObject); +// exAeEvaluationService.executeEvaluation(jsonObject); log.info("异步评估任务执行结束：threadName={}, instId={}, itemId={}", Thread.currentThread().getName(), instId, itemId); } catch (Exception e) { log.error("评估任务执行异常：threadName={}, instId={}, itemId={}, 错误={}", Thread.currentThread().getName(), instId, itemId, e.getMessage(), e); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowHttpNodeHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowHttpNodeHandler.java index 16e73b5..df010dd 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowHttpNodeHandler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowHttpNodeHandler.java @@ -20,22 +20,31 @@ public class FlowHttpNodeHandler implements FlowNodeTypeHandler { public TaskNodeExecuteResult handle(TaskNodeExecuteMessage message) { try { JSONObject inputParams = message.getInputParams(); + JSONObject config = inputParams.getJSONObject("config"); + JSONObject body = inputParams.getJSONObject("body"); + JSONObject headers = inputParams.getJSONObject("headers"); // 1. 必填参数校验 - String url = inputParams.getString("url"); + String url = config.getString("url"); if (url == null || url.isEmpty()) { return TaskNodeExecuteResult.failure("HTTP url is required"); } - String method = inputParams.getString("method"); + String method = config.getString("method"); if (method == null) { method = "POST"; } - int timeout = inputParams.getIntValue("timeout", 100000); + int timeout = config.getIntValue("timeout", 100000); - JSONObject headers = inputParams.getJSONObject("headers"); - Object bodyCfg = inputParams.get("body"); + Object bodyCfg = null; + if (!"json".equals(body.getString("bodyType"))) { + JSONObject formData = body.getJSONObject("formData"); + formData.remove("_self_"); + bodyCfg = formData; + } else { + bodyCfg = body.get("json"); + } // 2. 构造 HTTP 请求 HttpRequest request = HttpRequest.of(url) .method(Method.valueOf(method.toUpperCase())) diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java index 45557ed..93c4da6 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java @@ -50,7 +50,7 @@ public class FlowItemExecutor { FlowGraph graph = FlowModelBuilder.buildExecutableGraph(item.getFlowData()); String startNodeId = graph.findStartNodeId(); List startInputDefs = graph.getStartInputParamsDefs(); - + startInputDefs.clear(); // 执行 start 节点（同步执行） NodeExecutor executor = node -> executeNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, new ArrayList<>()); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java index cebde64..c45fcb7 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java @@ -1,6 +1,7 @@ package com.cmvr.test.flow.runtime.engine.support; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.collection.CollectionUtil; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; @@ -30,45 +31,10 @@ public class FlowNodeParamPreparer { FlowNodeWrapper node, TaskNodeExecuteMessage rootMessage) { - JSONObject input = new JSONObject(); + TaskContext ctx = taskInstHolder.getContext(rootMessage.getInstId()); - // 遍历当前节点定义的参数 - for (FlowParamDef param : node.getNodeParams()) { - if (param.getScope() != ParamScope.NODE) continue; - - // 1 静态 input 参数 - if ("input".equals(param.getType())) { - input.put(param.getName(), param.getInput()); - } - - // 2 引用参数（quote） - else if ("quote".equals(param.getType()) && CollUtil.isNotEmpty(param.getQuote())) { - String refNodeId = param.getQuote().get(0); // 被引用的上游节点 ID - List path = param.getQuote().subList(2, param.getQuote().size()); // 取值路径 - - JSONObject source; - // 引用的参数是上游的输入还是输出 - // 2.1 引用上游输入参数 - if ("input".equalsIgnoreCase(param.getQuoteType())) { - source = ctx.getRunParams(); - JSONObject def = toJson(graph.getNodeParamsDefs().get(refNodeId)); - for (String key : def.keySet()) { - source.put(key, def.get(key)); - } - } else if ("output".equalsIgnoreCase(param.getQuoteType())) { - // 2.2 引用上游的 output 参数（运行时实际执行结果） - // 用 nodeId + iterations 作为 key，避免覆盖 - source = ctx.getNodeOutput(refNodeId, rootMessage.getIterations()); - } else { - throw new GlobalException("参数应用类型出错！quoteType:" + param.getQuoteType()); - } - - // 按路径取嵌套值，例如 ["result", "image", "url"] - Object val = findNestedValue(source, path); - input.put(param.getName(), val); - } - } + JSONObject input = getInputParams(graph, node.getNodeParams(), rootMessage); // 3 START 节点：用 runParams 覆盖默认参数 if (node.getNodeType() == NodeTypeEnum.START) { @@ -119,6 +85,55 @@ public class FlowNodeParamPreparer { return input; } + private JSONObject getInputParams(FlowGraph graph, List paramDefList, TaskNodeExecuteMessage rootMessage) { + JSONObject input = new JSONObject(); + TaskContext ctx = taskInstHolder.getContext(rootMessage.getInstId()); + // 遍历当前节点定义的参数 + for (FlowParamDef param : paramDefList) { + if (param.getScope() != ParamScope.NODE) continue; + + // 1 静态 input 参数 + if ("input".equals(param.getType())) { + input.put(param.getName(), param.getInput()); + } + + // 2 引用参数（quote） + else if ("quote".equals(param.getType()) && CollUtil.isNotEmpty(param.getQuote())) { + String refNodeId = param.getQuote().get(0); // 被引用的上游节点 ID + List path = param.getQuote().subList(2, param.getQuote().size()); // 取值路径 + + JSONObject source; + // 引用的参数是上游的输入还是输出 + // 2.1 引用上游输入参数 + if ("input".equalsIgnoreCase(param.getQuoteType())) { + source = ctx.getRunParams(); + JSONObject def = toJson(graph.getNodeParamsDefs().get(refNodeId)); + for (String key : def.keySet()) { + source.put(key, def.get(key)); + } + } else if ("output".equalsIgnoreCase(param.getQuoteType())) { + // 2.2 引用上游的 output 参数（运行时实际执行结果） + // 用 nodeId + iterations 作为 key，避免覆盖 + source = ctx.getNodeOutput(refNodeId, rootMessage.getIterations()); + } else { + throw new GlobalException("参数应用类型出错！quoteType:" + param.getQuoteType()); + } + + // 按路径取嵌套值，例如 ["result", "image", "url"] + Object val = findNestedValue(source, path); + input.put(param.getName(), val); + } + if (!CollectionUtil.isEmpty(param.getChildren())) { + JSONObject inputParams = getInputParams(graph, param.getChildren(), rootMessage); + // 添加 _self_ 参数，用于在子参数中引用父参数 + inputParams.put("_self_",input.get(param.getName())); + // 添加子参数 + input.put(param.getName(), inputParams); + } + } + return input; + } + /** * 将参数定义列表中的 input 值构建为 JSONObject diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java index 3155b1d..0e9bf19 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java @@ -12,6 +12,8 @@ import com.cmvr.edge.client.service.EdgeCameraService; import com.cmvr.edge.client.service.EdgeHlcService; import com.cmvr.edge.client.service.EdgeMicrophoneService; import com.cmvr.edge.client.service.EdgeSpeakerService; +import com.cmvr.llm.service.LLMAiAgentPlatformService; +import com.cmvr.llm.service.LLMAiTtsService; import com.cmvr.test.enums.ActionEnum; import com.cmvr.test.model.vo.FlowActionRequestVO; import lombok.RequiredArgsConstructor; @@ -27,6 +29,8 @@ public class FlowActionExecutorService { private final EdgeMicrophoneService edgeMicrophoneService; private final EdgeSpeakerService edgeSpeakerService; private final EdgeHlcService edgeHlcService; + private final LLMAiTtsService llmAiTtsService; + private final LLMAiAgentPlatformService llmAiAgentPlatformService; public String actionExecute(FlowActionRequestVO req) { @@ -115,6 +119,8 @@ public class FlowActionExecutorService { return "LLM 意图识别 OK"; case GENERATE_ADVANCED_AUDIO: return "LLM 高级音频生成 OK"; + case AI_AGENT_PLATFORM: + return llmAiAgentPlatformService.query(action.getAction(), req.getPayload().getString("text"), req.getPayload().getString("apiKey")).toString(); default: throw new UnsupportedOperationException("未实现的 LLM Action: " + action); }