diff --git a/cmvr-iot-admin/src/main/resources/application.yml b/cmvr-iot-admin/src/main/resources/application.yml index 2fb6df3..503e51f 100644 --- a/cmvr-iot-admin/src/main/resources/application.yml +++ b/cmvr-iot-admin/src/main/resources/application.yml @@ -116,7 +116,13 @@ xss: api: app-id: d0epfibvo3em6c4iul40 app-key: d0eqt1kqek3vg7g4ofi0 - speech-to-text-url: https://aiagentplatform.cmft.com/api/proxy/api/v1/run_app_workflow + work-flow-url: https://aiagentplatform.cmft.com/api/proxy/api/v1/run_app_workflow query-result-url: https://aiagentplatform.cmft.com/api/proxy/api/v1/query_run_app_process bge-vl: http://192.168.1.8:5000/search_similar - generate_advanced_audio: http://192.168.0.19:8000/generate_advanced_audio + generate_advanced_audio: http://192.168.0.102:9003/generate_advanced_audio + # 智能体应用id和秘钥 + # ActionEnum 的枚举作为 key + agents: + INTENT_RECOGNITION: # 意图识别 + app-id: d3c98sp0gdon0fcf6l00 + app-key: d3c98vp0gdon0fcf6n20 diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/APIProperties.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/APIProperties.java index 13c150f..212e20e 100644 --- a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/APIProperties.java +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/APIProperties.java @@ -1,9 +1,12 @@ package com.cmvr.llm.config; +import com.cmvr.common.exception.GlobalException; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; +import java.util.Map; + /** * API配置类 */ @@ -22,10 +25,10 @@ public class APIProperties { */ private String appKey; - /** 执行语音转文本工作流接口地址 */ - private String speechToTextUrl; + /** 商道启动工作流地址 */ + private String workFlowUrl; - /** 查询语音转文本结果接口地址 */ + /** 商道工作流结果接口地址 */ private String queryResultUrl; private String bgeVl; @@ -34,4 +37,18 @@ public class APIProperties { * 文本合成语音 */ private String generateAdvancedAudio; + + /** 多应用配置（key=ActionEnum) */ + private Map agents; + + /** + * 根据 ActionEnum 名称获取 app 配置 + */ + public AgentConfig getAgentConfig(String actionName) { + AgentConfig config = agents.get(actionName); + if (config == null) { + throw new GlobalException("未找到功能 {} 的 智能体应用配置", actionName); + } + return config; + } } diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/AgentConfig.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/AgentConfig.java new file mode 100644 index 0000000..8c49f25 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/config/AgentConfig.java @@ -0,0 +1,18 @@ +package com.cmvr.llm.config; + +import lombok.Data; + +/** + * 智能体应用配置 + */ +@Data +public class AgentConfig { + /** + * 应用id + */ + private String appId; + /** + * 应用秘钥 + */ + private String appKey; +} \ No newline at end of file diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMIntentRecognitionService.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMIntentRecognitionService.java new file mode 100644 index 0000000..5e500a7 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMIntentRecognitionService.java @@ -0,0 +1,11 @@ +package com.cmvr.llm.service; + +import com.alibaba.fastjson2.JSONObject; + +public interface LLMIntentRecognitionService { + + /** + * 意图识别 + */ + public JSONObject intentRecognition(String action,String text); +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeService.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeService.java new file mode 100644 index 0000000..d0cc4f1 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeService.java @@ -0,0 +1,248 @@ +package com.cmvr.llm.service; + +import cn.hutool.core.util.ObjUtil; +import cn.hutool.core.util.StrUtil; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.common.utils.http.CallAPIUtil; +import com.cmvr.llm.config.APIProperties; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.HashMap; +import java.util.Map; + +/** + * 商道大模型工作流调用服务 + * 流程： + * 1. 调用工作流启动接口，获取 runId + * 2. 根据 runId 轮询查询接口 + * 3. 解析结果 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class WorkflowInvokeService { + + private final APIProperties apiProperties; + + /** + * 用户ID可以作为全局常量 + */ + private static final String USER_ID = "18888888888"; + + /** + * 默认最大重试次数 + */ + private static final int DEFAULT_MAX_RETRY = 40; + + /** + * 默认轮询间隔（毫秒） + */ + private static final int DEFAULT_INTERVAL_MS = 1000; + + /** + * 一站式执行工作流（带自定义重试参数） + * + * @param appId 智能体应用ID + * @param appKey 智能体应用秘钥 + * @param inputParams 输入参数（业务参数，例如图片、描述等） + * @param maxRetry 最大重试次数 + * @param intervalMs 每次轮询间隔（毫秒） + * @return 最终结果 message + */ + public JSONObject executeWorkflow(String appId, String appKey, + Map inputParams, + int maxRetry, long intervalMs) { + // 1. 启动工作流，获取 runId + String runId = startWorkflow(appId, appKey, inputParams); + + // 2. 轮询获取结果 + String responseBody = pollResult(runId, appId, appKey, maxRetry, intervalMs); + + // 3. 解析 + return parseResponse(responseBody); + } + + /** + * 一站式执行工作流（使用默认重试参数） + * + * @param appId 智能体应用ID + * @param appKey 智能体应用秘钥 + * @param inputParams 输入参数（业务参数，例如图片、描述等） + * @return 最终结果 message + */ + public JSONObject executeWorkflow(String appId, String appKey, Map inputParams) { + return executeWorkflow(appId, appKey, inputParams, DEFAULT_MAX_RETRY, DEFAULT_INTERVAL_MS); + } + + // ------------------- 核心流程 ------------------- + + /** + * 启动工作流，返回 runId + * + * @param appId 智能体应用ID + * @param appKey 智能体应用秘钥 + * @param inputParams 输入参数 + * @return 工作流运行ID-runId + */ + private String startWorkflow(String appId, String appKey, Map inputParams) { + Map headers = buildHeaders(appKey); + Map body = buildRequestBody(appKey, appId, inputParams); + + String responseBody = CallAPIUtil.doPostJson(apiProperties.getWorkFlowUrl(), headers, body); + JSONObject jsonObject = JSON.parseObject(responseBody); + + String runId = jsonObject.getString("runId"); + if (StrUtil.isEmpty(runId)) { + throw new GlobalException("未返回 runId，响应：{}", responseBody); + } + return runId; + } + + /** + * 轮询查询接口，直到获取到 end 节点 + * + * @param runId 工作流运行ID + * @param appId 智能体应用ID + * @param appKey 智能体应用秘钥 + * @param maxRetry 最大重试次数 + * @param intervalMs 轮询间隔（毫秒） + * @return 最终响应 JSON 字符串 + */ + private String pollResult(String runId, String appId, String appKey, int maxRetry, long intervalMs) { + Map headers = buildHeaders(appKey); + Map body = buildBaseBody(appKey, appId); + body.put("RunID", runId); + + int failCount = 0; + while (true) { + String responseBody = CallAPIUtil.doPostJson(apiProperties.getQueryResultUrl(), headers, body); + JSONObject root = JSON.parseObject(responseBody); + String status = root.getString("status"); + + if ("success".equalsIgnoreCase(status)) { + // 校验并返回最终结果 + parseResponse(responseBody); + return responseBody; + } else if ("processing".equalsIgnoreCase(status)) { + // 继续轮询 + failCount++; + if (failCount > maxRetry) { + throw new GlobalException("轮询超时，达到最大重试次数，最后响应：{}", responseBody); + } + try { + Thread.sleep(intervalMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new GlobalException("轮询被中断"); + } + } else { + // 其他状态直接抛异常，不再重试 + throw new GlobalException("工作流执行失败，status={}，响应：{}", status, responseBody); + } + } + } + + // ------------------- 结果解析 ------------------- + + /** + * 从响应 JSON 中找到 nodeType=end 且 status=success 的节点 + */ + private JSONObject findEndNode(String responseBody) { + JSONObject root = JSON.parseObject(responseBody); + JSONObject nodes = root.getJSONObject("nodes"); + if (nodes == null) { + throw new GlobalException("响应中 nodes 为空：{}", responseBody); + } + + for (Map.Entry entry : nodes.entrySet()) { + JSONObject node = (JSONObject) entry.getValue(); + if ("end".equalsIgnoreCase(node.getString("nodeType")) + && "success".equalsIgnoreCase(node.getString("status"))) { + return node; + } + } + throw new GlobalException("未找到 nodeType=end 且 status=success 的节点：{}", nodes); + } + + /** + * 解析 end 节点的 output 内容 + * + * @param endNode end 节点 + * @return output JSON + */ + private JSONObject parseOutput(JSONObject endNode) { + String outputStr = endNode.getString("output"); + if (StrUtil.isEmpty(outputStr)) { + throw new GlobalException("end 节点 output 为空：{}", endNode); + } + try { + return JSON.parseObject(outputStr); + } catch (Exception e) { + throw new GlobalException("output 解析失败：{}", outputStr); + } + } + + /** + * 接卸 + * + * @param responseBody 响应 JSON 字符串 + * @return message + */ + private JSONObject parseResponse(String responseBody) { + JSONObject endNode = findEndNode(responseBody); + JSONObject output = parseOutput(endNode); + + JSONObject nestedOutput = output.getJSONObject("output"); + if (ObjUtil.isEmpty(nestedOutput)) { + throw new GlobalException("end node output 为空!"); + } + return nestedOutput; + } + + // ------------------- 请求体拼装 ------------------- + + /** + * 构造请求头 + * + * @param appKey 应用秘钥 + * @return headers + */ + private Map buildHeaders(String appKey) { + Map headers = new HashMap<>(); + headers.put("Apikey", appKey); + return headers; + } + + /** + * 构造启动工作流的请求体 + * + * @param appKey 应用秘钥 + * @param appId 应用ID + * @param inputParams 输入参数 + * @return body + */ + private Map buildRequestBody(String appKey, String appId, Map inputParams) { + Map body = buildBaseBody(appKey, appId); + body.put("InputData", JSON.toJSONString(inputParams)); + return body; + } + + /** + * 构造通用请求体 + * + * @param appKey 应用秘钥 + * @param appId 应用ID + * @return body + */ + private Map buildBaseBody(String appKey, String appId) { + Map body = new HashMap<>(); + body.put("AppKey", appKey); + body.put("AppID", appId); + body.put("UserID", USER_ID); + return body; + } +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMIntentRecognitionServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMIntentRecognitionServiceImpl.java new file mode 100644 index 0000000..3682b87 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMIntentRecognitionServiceImpl.java @@ -0,0 +1,28 @@ +package com.cmvr.llm.service.impl; + +import cn.hutool.core.map.MapUtil; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.llm.config.APIProperties; +import com.cmvr.llm.config.AgentConfig; +import com.cmvr.llm.service.LLMIntentRecognitionService; +import com.cmvr.llm.service.WorkflowInvokeService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class LLMIntentRecognitionServiceImpl implements LLMIntentRecognitionService { + + private final APIProperties apiProperties; + private final WorkflowInvokeService workflowInvokeService; + + @Override + public JSONObject intentRecognition(String action, String text) { + AgentConfig acg = apiProperties.getAgentConfig(action); + Map inputParams = MapUtil.of("text", text); +// return new JSONObject().fluentPut("type", "4").fluentPut("message", "\n\n建议优先选择成都大熊猫繁育研究基地（成华区），该基地拥有顶流熊猫\\\"花花\\\"及完善的参观体系。门票需提前7天在\\\"成都熊猫基地电子票\\\"公众号预约，全票55元/人，建议选择清晨7:30前抵达南门入园（当前9月凉爽适宜户外观赏）。重点参观点为幼年大熊猫别墅（花花所在地）、月亮/太阳产房（活跃时段8:30-10:30）。若需避开人流，可前往都江堰熊猫谷（门票51元），该区域植被覆盖率高，游客密度较低，适合深度观赏熊猫自然行为。交通建议乘坐地铁3号线至熊猫大道站，换乘景区直通车（运营时间8:00-17:00）。需注意园区禁止投喂，建议携带轻便饮食及防滑鞋具。"); + return workflowInvokeService.executeWorkflow(acg.getAppId(), acg.getAppKey(), inputParams); + } +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMTouchServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMTouchServiceImpl.java index 764f8c8..fd29ae4 100644 --- a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMTouchServiceImpl.java +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMTouchServiceImpl.java @@ -53,7 +53,7 @@ public class LLMTouchServiceImpl implements LLMTouchService { body.put("InputData", JSON.toJSONString(params)); body.put("UserID", "18888888888"); - String res1 = CallAPIUtil.doPostJson(apiProperties.getSpeechToTextUrl(), headers, body); + String res1 = CallAPIUtil.doPostJson(apiProperties.getWorkFlowUrl(), headers, body); JSONObject resObject1 = JSON.parseObject(res1); if (resObject1.containsKey("runId")) { String processId = resObject1.getString("runId"); diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LargeModelRemoteServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LargeModelRemoteServiceImpl.java index 8bbe2c5..bd8f5fe 100644 --- a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LargeModelRemoteServiceImpl.java +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LargeModelRemoteServiceImpl.java @@ -31,7 +31,7 @@ public class LargeModelRemoteServiceImpl implements LargeModelRemoteService { body.put("InputData", JSON.toJSONString(callAgentVO)); body.put("UserID", "18888888888"); - String res1 = CallAPIUtil.doPostJson(apiProperties.getSpeechToTextUrl(), headers, body); + String res1 = CallAPIUtil.doPostJson(apiProperties.getWorkFlowUrl(), headers, body); JSONObject resObject1 = JSON.parseObject(res1); if (resObject1.containsKey("runId")) { diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/ShowIntelligentCockpitServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/ShowIntelligentCockpitServiceImpl.java index 3aefbf6..05bda47 100644 --- a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/ShowIntelligentCockpitServiceImpl.java +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/ShowIntelligentCockpitServiceImpl.java @@ -164,7 +164,7 @@ public class ShowIntelligentCockpitServiceImpl implements ShowIntelligentCockpit body.put("InputData", JSON.toJSONString(callAgentAdvancedVO)); body.put("UserID", "18888888888"); - String res = CallAPIUtil.doPostJson(apiProperties.getSpeechToTextUrl(), headers, body); + String res = CallAPIUtil.doPostJson(apiProperties.getWorkFlowUrl(), headers, body); JSONObject resObject = JSON.parseObject(res); if (resObject.containsKey("runId")) { @@ -243,7 +243,7 @@ public class ShowIntelligentCockpitServiceImpl implements ShowIntelligentCockpit body.put("InputData", JSON.toJSONString(params)); body.put("UserID", "18888888888"); - String res1 = CallAPIUtil.doPostJson(apiProperties.getSpeechToTextUrl(), headers, body); + String res1 = CallAPIUtil.doPostJson(apiProperties.getWorkFlowUrl(), headers, body); JSONObject resObject1 = JSON.parseObject(res1); if (resObject1.containsKey("runId")) { String processId = resObject1.getString("runId"); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java index a2b7da7..cdf2c6a 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java @@ -46,6 +46,7 @@ public enum ActionEnum { // 大模型行为 // ---------------获取触控坐标--------------- TOUCH_COORDINATES("LLM", "TOUCH_COORDINATES", "获取触控坐标"), + INTENT_RECOGNITION("LLM", "INTENT_RECOGNITION", "意图识别"), GENERATE_ADVANCED_AUDIO("LLM", "GENERATE_ADVANCED_AUDIO", "tts语音合成"), // 语音交互 diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java index ea3a24d..a1bd18d 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java @@ -229,9 +229,18 @@ public class FlowGraph { List outEdges = successors.getOrDefault(nodeId, Collections.emptyList()); for (FlowEdge edge : new ArrayList<>(outEdges)) { String targetNodeId = edge.getTo(); + + // 先删边 removeEdge(edge); - removePathFrom(targetNodeId); + + // 只有目标节点没有其他前驱时才递归 + List predecessorsList = predecessors.getOrDefault(targetNodeId, Collections.emptyList()); + if (predecessorsList.isEmpty()) { + removePathFrom(targetNodeId); + } } + + // 最后判断是否孤立再删节点 removeNodeIfIsolated(nodeId); } finally { lock.writeLock().unlock(); 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 bae68cb..1eac761 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 @@ -125,7 +125,7 @@ public class FlowItemExecutor { // 分支节点处理 if (node.getNodeType() == NodeTypeEnum.BRANCH) { - postHandleBranchNode(graph, node, startNodeId, startInputDefs, + postHandleBranchNode(taskInstHolder.getContext(rootMessage.getInstId()), graph, node, startNodeId, startInputDefs, rootMessage, onFinished, iterations, nodeId, message.getInputParams(), message); return; } @@ -158,7 +158,7 @@ public class FlowItemExecutor { @NotNull private static TaskNodeExecuteMessage buildTaskNodeExecuteMsg(FlowGraph graph, FlowNodeWrapper node, TaskNodeExecuteMessage rootMessage, String nodeId, - List iterations) { + List iterations) { TaskNodeExecuteMessage message = new TaskNodeExecuteMessage(); BeanUtil.copyProperties(rootMessage, message); message.setNodeId(nodeId); @@ -175,10 +175,10 @@ public class FlowItemExecutor { return message; } - private void postHandleBranchNode(FlowGraph graph, FlowNodeWrapper node, String startNodeId, + private void postHandleBranchNode(TaskContext taskContext, FlowGraph graph, FlowNodeWrapper node, String startNodeId, List startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, List iterations, String nodeId, JSONObject inputParams, TaskNodeExecuteMessage message) { - String matchedAnchorId = FlowBranchEvaluator.evaluate(nodeId, node.getBranchConditions(), inputParams); + String matchedAnchorId = FlowBranchEvaluator.evaluate(taskContext, iterations,nodeId, node.getBranchConditions(), inputParams); FlowEdge matchedEdge = graph.getEdgeByFromAnchorId(matchedAnchorId); if (matchedEdge == null) { @@ -190,8 +190,14 @@ public class FlowItemExecutor { List allOutEdges = new ArrayList<>(graph.getSuccessors().getOrDefault(nodeId, Collections.emptyList())); for (FlowEdge edge : allOutEdges) { if (!edge.equals(matchedEdge)) { - graph.removePathFrom(edge.getTo()); + // 先删边 graph.removeEdge(edge); + + // 再尝试删子路径（只有当目标节点没有其他前驱时才删） + List predecessorsList = graph.getPredecessors().getOrDefault(edge.getTo(), Collections.emptyList()); + if (predecessorsList.isEmpty()) { + graph.removePathFrom(edge.getTo()); + } } } @@ -206,7 +212,7 @@ public class FlowItemExecutor { } private void registerThread(FlowGraph graph, FlowNodeWrapper node, String startNodeId, List startInputDefs, - TaskNodeExecuteMessage rootMessage, Runnable onFinished, List iterations, String instId, String itemId, String nodeId) { + TaskNodeExecuteMessage rootMessage, Runnable onFinished, List iterations, String instId, String itemId, String nodeId) { TaskNodeExecuteContext nodeExecuteContext = new TaskNodeExecuteContext(); nodeExecuteContext.setThread(Thread.currentThread()); nodeExecuteContext.setGraph(graph); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowBranchEvaluator.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowBranchEvaluator.java index 29683c2..ee49506 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowBranchEvaluator.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowBranchEvaluator.java @@ -1,9 +1,13 @@ package com.cmvr.test.flow.runtime.engine.support; +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.ObjUtil; import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.enums.BranchConditionOpEnum; import com.cmvr.test.flow.builder.FlowBranchConditionGroup; +import com.cmvr.test.flow.context.TaskContext; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -19,7 +23,9 @@ public class FlowBranchEvaluator { * 遍历所有条件组，返回第一个满足条件的分支ID * 如果都不满足，返回默认的 else 出口 */ - public static String evaluate(String nodeId, List groups, JSONObject inputParams) { + public static String evaluate(TaskContext taskContext, List iterations, String nodeId, List groups, JSONObject inputParams) { + // 先合并上游输出参数 + mergeParams(taskContext,iterations, inputParams, groups); for (FlowBranchConditionGroup group : groups) { if (matchGroup(group, inputParams)) { return group.getId(); @@ -28,6 +34,34 @@ public class FlowBranchEvaluator { return nodeId + "_else"; } + /** + * 获取分支节点引用节点的输出参数，并合并到当前计算参数列表中 + */ + private static void mergeParams(TaskContext taskContext, List iterations, JSONObject inputParams, List groups) { + if (CollUtil.isEmpty(taskContext.getNodeOutputs())) { + return; + } + HashSet quoteIds = new HashSet<>(); + for (FlowBranchConditionGroup group : groups) { + List expressions = group.getExpressions(); + for (FlowBranchConditionGroup.ConditionExpression expression : expressions) { + if (expression.getNameType().equalsIgnoreCase("quote")) { + List nameQuotePath = expression.getNameQuotePath(); + if (CollUtil.isNotEmpty(nameQuotePath) && nameQuotePath.get(0).length() == 32) { + quoteIds.add(nameQuotePath.get(0)); + } + } + } + } + for (String quoteId : quoteIds) { + JSONObject nodeOutput = taskContext.getNodeOutput(quoteId, iterations); + if (ObjUtil.isNotEmpty(nodeOutput)) { + // 合并进inputParams + inputParams.putAll(nodeOutput); + } + } + } + /** * 匹配某个条件组是否成立 */ @@ -109,7 +143,7 @@ public class FlowBranchEvaluator { } Object current = context; - for (int i = 1; i < path.size(); i++) { + for (int i = 2; i < path.size(); i++) { if (current instanceof JSONObject) { current = ((JSONObject) current).get(path.get(i)); } else if (current instanceof Map) { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java index 6e205d5..00f4464 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java @@ -12,6 +12,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; +import java.util.Arrays; import java.util.function.Function; @Slf4j @@ -42,7 +43,10 @@ public class FlowExceptionInterceptor implements FlowMsgPreInterceptor { ObjUtil.isEmpty(message.getInputParams()) ? null : message.getInputParams().toJSONString(), "执行异常：" + StrUtil.sub(ExceptionUtil.getMessage(e), 0, 480) ); - throw new GlobalException("任务执行失败：" + e.getMessage()); + log.error("节点执行异常：instId={}, nodeId={}, action={}", + message.getInstId(), message.getNodeId(), message.getAction(), e); + + throw new GlobalException("任务执行失败", e); } } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMIntentRecogOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMIntentRecogOperateService.java new file mode 100644 index 0000000..be75107 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMIntentRecogOperateService.java @@ -0,0 +1,34 @@ +package com.cmvr.test.flow.runtime.operator.llm; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.llm.service.LLMIntentRecognitionService; +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +@Slf4j +@Service +@RequiredArgsConstructor +public class LLMIntentRecogOperateService implements LLMOperateService { + + private final LLMIntentRecognitionService intentRecognitionService; + + @Override + public boolean supports(ActionEnum action) { + return action.name().equals("INTENT_RECOGNITION"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + JSONObject inputParams = message.getInputParams(); + String text = inputParams.getString("text"); + + JSONObject output = intentRecognitionService.intentRecognition(action.name(), text); + + return TaskNodeExecuteResult.success(output); + } +} \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTTSOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTTSOperateService.java index 937cfd9..5ac5978 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTTSOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTTSOperateService.java @@ -10,14 +10,30 @@ import com.cmvr.llm.config.APIProperties; import com.cmvr.test.enums.ActionEnum; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.util.TextSplitUtil; +import lombok.Data; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.NotNull; import org.springframework.stereotype.Service; +import javax.sound.sampled.AudioFormat; +import javax.sound.sampled.AudioInputStream; +import javax.sound.sampled.AudioSystem; +import javax.sound.sampled.DataLine; +import javax.sound.sampled.SourceDataLine; +import javax.sound.sampled.UnsupportedAudioFileException; import java.io.File; +import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; + @Slf4j @Service @RequiredArgsConstructor @@ -25,6 +41,16 @@ public class LLMTTSOperateService implements LLMOperateService { private final APIProperties apiProperties; + /** + * instId -> 音频队列 + */ + private final Map> audioQueues = new ConcurrentHashMap<>(); + + /** + * instId -> 播放线程 + */ + private final Map playerThreads = new ConcurrentHashMap<>(); + @Override public boolean supports(ActionEnum action) { return action.name().equals("GENERATE_ADVANCED_AUDIO"); @@ -32,20 +58,83 @@ public class LLMTTSOperateService implements LLMOperateService { @Override public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + String instId = message.getInstId(); JSONObject inputParams = message.getInputParams(); String text = inputParams.getString("text"); - JSONObject output = new JSONObject(); + // 1. 切分 + List slices = TextSplitUtil.splitText(text); + if (slices.isEmpty()) { + throw new GlobalException("输入文本为空，无法合成语音"); + } - // 1. 构建请求体 + // 2. 确保队列和线程存在 + BlockingQueue queue = audioQueues.computeIfAbsent(instId, k -> new LinkedBlockingQueue<>()); + startPlayerThread(instId, queue); + + // 3. 逐个生成并放入队列 + CountDownLatch latch = new CountDownLatch(1); + int index = 1; + for (String slice : slices) { + long start = System.currentTimeMillis(); + File audioFile = getAudioFile(slice, index); + long end = System.currentTimeMillis(); + long playMs = getAudioDurationMs(audioFile); + + // 最后一段带 latch + AudioSegment segment = new AudioSegment(audioFile.getAbsolutePath(), playMs, end - start, + (index == slices.size()) ? latch : null); + queue.offer(segment); + log.info("[{}] 音频入队: {}", instId, segment.getPath()); + index++; + } + + // 4. 等待最后一段播完 + try { + latch.await(); + log.info("[{}] 最后一段音频播放完成", instId); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new GlobalException("等待音频播放时被中断"); + } + + // 5. 节点才算完成 + JSONObject output = new JSONObject(); + output.put("segments", slices.size()); + return TaskNodeExecuteResult.success(output); + } + + /** + * 启动播放线程（按 instId 单独维护） + */ + private void startPlayerThread(String instId, BlockingQueue queue) { + playerThreads.computeIfAbsent(instId, k -> { + Thread thread = new Thread(() -> { + try { + while (true) { + AudioSegment segment = queue.take(); // 阻塞等待 + playAudio(segment); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.info("[{}] 播放线程被中断，退出", instId); + } + }, "AudioPlayer-" + instId); + thread.start(); + return thread; + }); + } + + + @NotNull + private File getAudioFile(String text, int index) { Map body = new HashMap<>(); body.put("text", text); body.put("dialect", "普通话"); - body.put("timbre", ""); + body.put("timbre", "标准男"); body.put("language", "中文"); body.put("tone", "中性"); - // 2. 发送请求到语音合成服务 HttpResponse response = HttpRequest.post(apiProperties.getGenerateAdvancedAudio()) .header("Content-Type", "application/json") .body(JSON.toJSONString(body)) @@ -55,28 +144,80 @@ public class LLMTTSOperateService implements LLMOperateService { throw new GlobalException("音频生成失败，状态码: " + response.getStatus()); } - // 3. 获取音频数据 byte[] audioData = response.bodyBytes(); - - // 4. 生成本地文件路径 - String filename = System.currentTimeMillis() + "_" + text + ".wav"; + String filename = System.currentTimeMillis() + "_" + index + ".wav"; String saveDir = "C:\\Users\\11158\\Desktop\\iot"; File dir = new File(saveDir); if (!dir.exists() && !dir.mkdirs()) { throw new GlobalException("音频保存目录创建失败: " + saveDir); } - File audioFile = FileUtil.writeBytes(audioData, new File(dir, filename)); - - log.info("音频生成成功，保存路径: {}", audioFile.getAbsolutePath()); - - // 5. 预留上传功能 - // String fileUrl = fileUploadService.upload(audioFile); - - // 先返回本地路径 - output.put("localPath", audioFile.getAbsolutePath()); - // output.put("url", fileUrl); // 上传后再返回 url - - return TaskNodeExecuteResult.success(output); + return FileUtil.writeBytes(audioData, new File(dir, filename)); } + + private long getAudioDurationMs(File file) { + try (AudioInputStream audioInputStream = AudioSystem.getAudioInputStream(file)) { + AudioFormat format = audioInputStream.getFormat(); + long frames = audioInputStream.getFrameLength(); + double durationInSeconds = (frames + 0.0) / format.getFrameRate(); + return (long) (durationInSeconds * 1000); + } catch (UnsupportedAudioFileException | IOException e) { + throw new GlobalException("获取音频时长失败: {}", e.getMessage()); + } + } + + /** + * 播放一个音频文件（阻塞直到播完） + */ + private void playAudio(AudioSegment segment) { + String filePath = segment.getPath(); + try (AudioInputStream sourceStream = AudioSystem.getAudioInputStream(new File(filePath))) { + AudioFormat baseFormat = sourceStream.getFormat(); + AudioFormat targetFormat = new AudioFormat( + AudioFormat.Encoding.PCM_SIGNED, + baseFormat.getSampleRate(), + 16, + baseFormat.getChannels(), + baseFormat.getChannels() * 2, + baseFormat.getSampleRate(), + false + ); + + AudioInputStream pcmStream = AudioSystem.getAudioInputStream(targetFormat, sourceStream); + + DataLine.Info info = new DataLine.Info(SourceDataLine.class, targetFormat); + try (SourceDataLine line = (SourceDataLine) AudioSystem.getLine(info)) { + line.open(targetFormat); + line.start(); + + byte[] buffer = new byte[4096]; + int bytesRead; + while ((bytesRead = pcmStream.read(buffer, 0, buffer.length)) != -1) { + line.write(buffer, 0, bytesRead); + } + + line.drain(); + line.stop(); + } + + log.info("播放完成: {}", filePath); + + // 如果有 latch，释放 + if (segment.getLatch() != null) { + segment.getLatch().countDown(); + } + + } catch (Exception e) { + throw new GlobalException("播放失败: ", e); + } + } + + @Data + public static class AudioSegment { + private final String path; + private final long playMs; + private final long generateMs; + private final CountDownLatch latch; // 仅最后一段有 + } + } \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/util/TextSplitUtil.java b/cmvr-iot-test/src/main/java/com/cmvr/test/util/TextSplitUtil.java new file mode 100644 index 0000000..a1d3b4c --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/util/TextSplitUtil.java @@ -0,0 +1,104 @@ +package com.cmvr.test.util; + +import cn.hutool.core.util.StrUtil; + +import java.util.ArrayList; +import java.util.List; + +public class TextSplitUtil { + + private static final int DEFAULT_MAX_LENGTH = 20; + + public static List splitText(String text) { + return splitText(text, DEFAULT_MAX_LENGTH); + } + + public static List splitText(String text, int maxLength) { + List result = new ArrayList<>(); + if (StrUtil.isEmpty(text)) { + return result; + } + + // 0. 清洗 + text = cleanText(text); + + // 1. 先按句号/问号/感叹号切 + String[] sentences = text.split("(?<=[。！？])"); + for (int i = 0; i < sentences.length; i++) { + String sentence = sentences[i].trim(); + if (StrUtil.isEmpty(sentence)) { + continue; + } + + // 特殊规则：第一句单独处理 + if (i == 0 && sentence.length() > 15) { + int cut = Math.min(sentence.length(), 15); // 最多15 + if (cut < 10) cut = 10; // 保底至少10 + result.add(sentence.substring(0, cut)); + + // 剩余部分继续走正常分割 + String remain = sentence.substring(cut).trim(); + if (StrUtil.isNotEmpty(remain)) { + if (remain.length() > maxLength) { + result.addAll(splitByComma(remain, maxLength)); + } else { + result.add(remain); + } + } + continue; + } + + // 2. 如果太长，按逗号/分号切 + if (sentence.length() > maxLength) { + result.addAll(splitByComma(sentence, maxLength)); + } else { + result.add(sentence); + } + } + + return result; + } + + /** + * 先按逗号/分号切，再兜底按长度切 + */ + private static List splitByComma(String text, int maxLength) { + List result = new ArrayList<>(); + String[] parts = text.split("(?<=[，；])"); // 保留逗号/分号 + for (String part : parts) { + part = part.trim(); + if (StrUtil.isEmpty(part)) { + continue; + } + if (part.length() > maxLength) { + result.addAll(splitByLength(part, maxLength)); + } else { + result.add(part); + } + } + return result; + } + + /** + * 长句按长度强制切分（兜底） + */ + private static List splitByLength(String text, int maxLength) { + List result = new ArrayList<>(); + for (int i = 0; i < text.length(); i += maxLength) { + int end = Math.min(text.length(), i + maxLength); + result.add(text.substring(i, end).trim()); + } + return result; + } + + private static String cleanText(String text) { + if (StrUtil.isEmpty(text)) { + return text; + } + String cleaned = text; + cleaned = cleaned.replaceAll("[\\r\\n\\t]+", " "); // 换行制表 + cleaned = cleaned.replaceAll(" +", " "); // 多余空格 + cleaned = cleaned.replaceAll("[-=*]{3,}", " "); // 分隔符 + return cleaned.trim(); + } +}