From 3cf8d0a3c0d7224f9482c9364c55e04a421000bd Mon Sep 17 00:00:00 2001 From: lixiaolong <702156524@qq.com> Date: Tue, 17 Mar 2026 14:08:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E6=96=B0=E5=A2=9E=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E6=89=A7=E8=A1=8C=E8=8A=82=E7=82=B9=E5=92=8CAI?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加CODE类型节点枚举和处理器实现代码执行功能 - 集成GraalVM JavaScript引擎支持安全的代码执行 - 新增AI_AGENT_PLATFORM和AI_TTS两个AI服务模块 - 更新意图识别服务使用新的聊天接口实现 - 修改HTTP节点处理器移除临时测试代码 - 在流程构建器中添加对code类型的节点支持 - 配置文件中更新应用密钥信息 - 添加完整的LLM聊天服务工具类支持流式通信 - 实现TS语音合成和智能体平台调用服务 - 新增NashornJS执行器提供安全的JavaScript执行环境 --- .../src/main/resources/application-dev.yml | 4 +- .../service/LLMAiAgentPlatformService.java | 7 + .../com/cmvr/llm/service/LLMAiTtsService.java | 7 + .../service/WorkflowInvokeServiceCopy.java | 270 ++++++++++++++ .../impl/LLMAiAgentPlatformServiceImpl.java | 36 ++ .../llm/service/impl/LLMAiTtsServiceImpl.java | 40 +++ .../impl/LLMIntentRecognitionServiceImpl.java | 26 +- .../com/cmvr/llm/util/LlmChatService.java | 339 ++++++++++++++++++ cmvr-iot-test/pom.xml | 11 + .../java/com/cmvr/test/enums/ActionEnum.java | 3 + .../com/cmvr/test/enums/NodeTypeEnum.java | 1 + .../test/flow/builder/FlowModelBuilder.java | 2 + .../dispatcher/FlowCodeNodeHandler.java | 30 ++ .../dispatcher/FlowHttpNodeHandler.java | 8 +- .../llm/LLMAiAgentPlatformOperateService.java | 36 ++ .../operator/llm/LLMAiTtsOperateService.java | 38 ++ .../com/cmvr/test/util/NashornJsExecutor.java | 250 +++++++++++++ 17 files changed, 1097 insertions(+), 11 deletions(-) create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiAgentPlatformService.java create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiTtsService.java create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeServiceCopy.java create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiAgentPlatformServiceImpl.java create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiTtsServiceImpl.java create mode 100644 cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowCodeNodeHandler.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiAgentPlatformOperateService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiTtsOperateService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/util/NashornJsExecutor.java diff --git a/cmvr-iot-admin/src/main/resources/application-dev.yml b/cmvr-iot-admin/src/main/resources/application-dev.yml index 25ec2cd..384758d 100644 --- a/cmvr-iot-admin/src/main/resources/application-dev.yml +++ b/cmvr-iot-admin/src/main/resources/application-dev.yml @@ -118,8 +118,8 @@ api: # ActionEnum 的枚举作为 key agents: INTENT_RECOGNITION: # 意图识别 - app-id: d3c98sp0gdon0fcf6l00 - app-key: d3c98vp0gdon0fcf6n20 + app-id: d5dn5ibp9adhq1b34lig + app-key: d6j5bcellh49on5tasvg TOUCH_COORDINATES: # 意图识别 app-id: d5thge2cktmipk78h82g app-key: d1ebtabnjkflk4gmhikg diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiAgentPlatformService.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiAgentPlatformService.java new file mode 100644 index 0000000..fc3d4a3 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiAgentPlatformService.java @@ -0,0 +1,7 @@ +package com.cmvr.llm.service; + +import com.alibaba.fastjson2.JSONObject; + +public interface LLMAiAgentPlatformService { + JSONObject query(String action, String text, String apiKey); +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiTtsService.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiTtsService.java new file mode 100644 index 0000000..064544f --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiTtsService.java @@ -0,0 +1,7 @@ +package com.cmvr.llm.service; + +import com.alibaba.fastjson2.JSONObject; + +public interface LLMAiTtsService { + JSONObject play(String url, String text, String voice, String speed, String volume); +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeServiceCopy.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeServiceCopy.java new file mode 100644 index 0000000..f901b5d --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeServiceCopy.java @@ -0,0 +1,270 @@ +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 com.cmvr.llm.util.LlmChatService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.util.HashMap; +import java.util.Map; + +/** + * 商道大模型工作流调用服务 + * 流程： + * 1. 调用工作流启动接口，获取 runId + * 2. 根据 runId 轮询查询接口 + * 3. 解析结果 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class WorkflowInvokeServiceCopy { + + 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); + } + + @Autowired + private LlmChatService llmChatService; + + /** + * 一站式执行工作流（使用默认重试参数） + * + * @param appId 智能体应用ID + * @param appKey 智能体应用秘钥 + * @param text 输入参数（业务参数，例如图片、描述等） + * @return 最终结果 message + */ + public String executeWorkflow(String appId, String appKey, String text) { + // 调用 + Map body = new HashMap<>(); + body.put("Query", text); + long startTime = System.currentTimeMillis(); + String result = null; + try { + result = llmChatService.chat( + appKey, // apiKey (header用) + appId, // apiId (body的AppKey) + "user_123", // userId + body + ); + System.out.println( result); + System.out.println("总耗时：" + (System.currentTimeMillis() - startTime)); + } catch (Exception e) { + throw new RuntimeException(e); + } + return result; + } + + // ------------------- 核心流程 ------------------- + + /** + * 启动工作流，返回 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/LLMAiAgentPlatformServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiAgentPlatformServiceImpl.java new file mode 100644 index 0000000..5422446 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiAgentPlatformServiceImpl.java @@ -0,0 +1,36 @@ +package com.cmvr.llm.service.impl; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.llm.service.LLMAiAgentPlatformService; +import com.cmvr.llm.util.LlmChatService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.util.HashMap; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class LLMAiAgentPlatformServiceImpl implements LLMAiAgentPlatformService { + private final LlmChatService llmChatService; + @Override + public JSONObject query(String action, String text, String apiKey) { + Map body = new HashMap<>(); + body.put("Query", text); + long startTime = System.currentTimeMillis(); + String result = null; + try { + result = llmChatService.chat( + apiKey, // apiKey (header用) + apiKey, // apiId (body的AppKey) + "user_123", // userId + body + ); + System.out.println( result); + System.out.println("总耗时：" + (System.currentTimeMillis() - startTime)); + } catch (Exception e) { + throw new RuntimeException(e); + } + return new JSONObject().fluentPut("result", result); + } +} diff --git a/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiTtsServiceImpl.java b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiTtsServiceImpl.java new file mode 100644 index 0000000..b49f08b --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiTtsServiceImpl.java @@ -0,0 +1,40 @@ +package com.cmvr.llm.service.impl; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.utils.http.CallAPIUtil; +import com.cmvr.llm.service.LLMAiAgentPlatformService; +import com.cmvr.llm.service.LLMAiTtsService; +import com.cmvr.llm.util.LlmChatService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.util.HashMap; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class LLMAiTtsServiceImpl implements LLMAiTtsService { + @Override + public JSONObject play(String url, String text, String voice, String speed, String volume) { +// Map body = new HashMap<>(); +// body.put("Query", text); +// long startTime = System.currentTimeMillis(); +// String result = null; +// try { +// result = llmChatService.chat( +// apiKey, // apiKey (header用) +// apiKey, // apiId (body的AppKey) +// "user_123", // userId +// body +// ); +// System.out.println( result); +// System.out.println("总耗时：" + (System.currentTimeMillis() - startTime)); +// } catch (Exception e) { +// throw new RuntimeException(e); +// } + // 通过post调用tts接口 + String result = CallAPIUtil.doPostJson(url, new HashMap<>(), new JSONObject().fluentPut("text", text).fluentPut("voice", voice).fluentPut("speed", speed).fluentPut("volume", volume)); + + return new JSONObject().fluentPut("result", result); + } +} 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 index 3682b87..9b4ea44 100644 --- 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 @@ -6,9 +6,11 @@ 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 com.cmvr.llm.util.LlmChatService; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; +import java.util.HashMap; import java.util.Map; @Service @@ -17,12 +19,30 @@ public class LLMIntentRecognitionServiceImpl implements LLMIntentRecognitionServ private final APIProperties apiProperties; private final WorkflowInvokeService workflowInvokeService; - + private final LlmChatService llmChatService; @Override public JSONObject intentRecognition(String action, String text) { AgentConfig acg = apiProperties.getAgentConfig(action); - Map inputParams = MapUtil.of("text", text); +// 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); +// return workflowInvokeService.executeWorkflow(acg.getAppId(), acg.getAppKey(), inputParams); + // 调用 + Map body = new HashMap<>(); + body.put("Query", text); + long startTime = System.currentTimeMillis(); + String result = null; + try { + result = llmChatService.chat( + acg.getAppKey(), // apiKey (header用) + acg.getAppId(), // apiId (body的AppKey) + "user_123", // userId + body + ); + System.out.println( result); + System.out.println("总耗时：" + (System.currentTimeMillis() - startTime)); + } catch (Exception e) { + throw new RuntimeException(e); + } + return new JSONObject().fluentPut("type", result).fluentPut("test2", new JSONObject().fluentPut("test2-2", "今天天气怎么样")); } } 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 new file mode 100644 index 0000000..aaa5747 --- /dev/null +++ b/cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java @@ -0,0 +1,339 @@ +package com.cmvr.llm.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.*; +import org.springframework.stereotype.Service; + +import javax.annotation.PreDestroy; +import java.io.BufferedReader; +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.atomic.AtomicReference; + +/** + * LLM聊天服务 - 严格保持原接口参数 + */ +@Service +public class LlmChatService { + + 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(); + + // 会话缓存：key = apiKey + "|" + apiId + "|" + userId + private final ConcurrentHashMap conversationCache = new ConcurrentHashMap<>(); + + private final OkHttpClient httpClient; + + public LlmChatService() { + this.httpClient = new OkHttpClient.Builder() + .connectTimeout(30, TimeUnit.SECONDS) + .readTimeout(120, TimeUnit.SECONDS) + .writeTimeout(30, TimeUnit.SECONDS) + .build(); + } + + /** + * 核心方法：同步调用，返回完整字符串结果 + * 自动管理会话生命周期 + * + * @param apiKey 对应原AppKey的header + * @param apiId 对应原AppKey的body字段 + * @param userId 对应原UserID + * @param body 请求体，必须包含Query字段，可选Name等 + * @return 完整AI回复字符串 + */ + public String chat(String apiKey, String apiId, String userId, Map body) throws IOException { + String cacheKey = apiKey + "|" + apiId + "|" + userId; + + // 获取或创建会话（严格保持原createConversation逻辑） + ConversationHolder holder = conversationCache.computeIfAbsent(cacheKey, k -> { + try { + String conversationId = createConversation(apiKey, apiId, userId); + System.out.println("创建新会话: " + conversationId); + return new ConversationHolder(conversationId, apiKey, apiId, userId); + } catch (IOException e) { + throw new RuntimeException("创建会话失败", e); + } + }); + + // 构建请求体 - 严格保持原buildChatRequestBody逻辑 + Map requestBody = new HashMap<>(); + requestBody.put("AppKey", apiId); // 对应原apiKey参数 + requestBody.put("AppConversationID", holder.conversationId); + requestBody.put("UserID", userId); + requestBody.put("ResponseMode", "streaming"); + requestBody.put("PubAgentJump", true); + + // 合并用户传入的body（包含Query等） + if (body != null) { + requestBody.putAll(body); + } + + // 执行流式请求并同步返回完整结果 + return executeSseRequestSync(apiKey, requestBody); + } + + /** + * 创建会话 - 完全保持原逻辑 + */ + private String createConversation(String apiKey, String apiId, String userId) throws IOException { + Map body = new HashMap<>(); + body.put("AppKey", apiId); // 对应原apiKey + body.put("UserID", userId); + // 如果调用方传了Name就用，否则自动生成 + if (!body.containsKey("Name")) { + body.put("Name", "自动创建-" + System.currentTimeMillis()); + } + + String jsonBody = objectMapper.writeValueAsString(body); + + Request request = new Request.Builder() + .url(DEFAULT_BASE_URL + "/create_conversation") + .header("Apikey", apiKey) // 对应原apiKey + .header("Content-Type", "application/json") + .post(RequestBody.create(jsonBody, JSON)) + .build(); + + try (Response response = httpClient.newCall(request).execute()) { + if (!response.isSuccessful()) { + throw new IOException("创建会话失败: " + response.code() + " - " + response.body().string()); + } + + String respJson = response.body().string(); + Map respMap = objectMapper.readValue(respJson, Map.class); + Map conversation = (Map) respMap.get("Conversation"); + + if (conversation == null) { + throw new IOException("返回数据格式错误，缺少 Conversation 字段"); + } + + String conversationId = (String) conversation.get("AppConversationID"); + if (conversationId == null || conversationId.isEmpty()) { + throw new IOException("返回数据格式错误，缺少 AppConversationID"); + } + + return conversationId; + } + } + + /** + * 执行SSE请求 - 完全保持原executeSseRequest逻辑，改为同步返回 + */ + private String executeSseRequestSync(String apiKey, Map requestBody) throws IOException { + String jsonBody = objectMapper.writeValueAsString(requestBody); + + Request request = new Request.Builder() + .url(DEFAULT_BASE_URL + "/chat_query") + .header("Apikey", apiKey) + .header("Content-Type", "application/json") + .header("Accept", "text/event-stream") + .post(RequestBody.create(jsonBody, JSON)) + .build(); + + StringBuilder fullContent = new StringBuilder(); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference errorRef = new AtomicReference<>(); + AtomicReference finalMessageIdRef = new AtomicReference<>(""); + + Call call = httpClient.newCall(request); + + // 使用异步调用但阻塞等待结果 + call.enqueue(new Callback() { + @Override + public void onFailure(Call call, IOException e) { + errorRef.set(e); + latch.countDown(); + } + + @Override + public void onResponse(Call call, Response response) throws IOException { + if (!response.isSuccessful()) { + String errorBody = response.body() != null ? response.body().string() : "无错误信息"; + errorRef.set(new IOException("SSE请求失败: " + response.code() + " - " + errorBody)); + latch.countDown(); + return; + } + + StringBuilder localContent = new StringBuilder(); + String finalMessageId = ""; + String currentEvent = ""; + + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(response.body().byteStream(), StandardCharsets.UTF_8))) { + + String line; + while ((line = reader.readLine()) != null) { + if (line.isEmpty()) { + currentEvent = ""; + continue; + } + + if (line.startsWith("event:")) { + currentEvent = line.substring(6).trim(); + continue; + } + + if (line.startsWith("data:")) { + String data = line.substring(5).trim(); + + // 处理双重 data: 前缀（data:data: {...}） + if (data.startsWith("data:")) { + data = data.substring(5).trim(); + } + + if ("[DONE]".equals(data)) { + fullContent.append(localContent); + finalMessageIdRef.set(finalMessageId); + latch.countDown(); + return; + } + + try { + Map jsonData = objectMapper.readValue(data, Map.class); + + String messageId = extractString(jsonData, "id", "task_id", "message_id"); + if (!messageId.isEmpty()) { + finalMessageId = messageId; + } + + String event = (String) jsonData.getOrDefault("event", currentEvent); + + if ("message".equals(event)) { + String answer = extractString(jsonData, "answer", "content", "chunk", "text"); + if (!answer.isEmpty()) { + localContent.append(answer); + } + } else if ("message_start".equals(event)) { + System.out.println("消息开始，ID: " + finalMessageId); + } else if ("message_output_start".equals(event)) { + System.out.println("输出开始"); + } else if ("message_end".equals(event) || "end".equals(event)) { + fullContent.append(localContent); + finalMessageIdRef.set(finalMessageId); + latch.countDown(); + return; + } + + } catch (Exception e) { + System.out.println("解析失败，原始数据: " + data); + } + } + } + + // 流正常结束 + fullContent.append(localContent); + finalMessageIdRef.set(finalMessageId); + latch.countDown(); + + } catch (IOException e) { + errorRef.set(e); + latch.countDown(); + } + } + }); + + // 等待流式传输完成 + try { + boolean completed = latch.await(120, TimeUnit.SECONDS); + if (!completed) { + throw new IOException("请求超时"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("请求被中断", e); + } + + if (errorRef.get() != null) { + if (errorRef.get() instanceof IOException) { + throw (IOException) errorRef.get(); + } else { + throw new IOException("请求异常", errorRef.get()); + } + } + + return fullContent.toString(); + } + + /** + * 完全保持原extractString方法 + */ + private String extractString(Map map, String... keys) { + for (String key : keys) { + Object value = map.get(key); + if (value != null) { + return value.toString(); + } + } + return ""; + } + + /** + * Spring Boot停止时清理所有会话 + */ + @PreDestroy + public void close() { + System.out.println("开始清理会话，共 " + conversationCache.size() + " 个"); + + conversationCache.values().forEach(holder -> { + try { + deleteConversation(holder); + System.out.println("已删除会话: " + holder.conversationId); + } catch (Exception e) { + System.err.println("删除会话失败: " + holder.conversationId); + } + }); + + conversationCache.clear(); + httpClient.dispatcher().executorService().shutdown(); + httpClient.connectionPool().evictAll(); + System.out.println("会话清理完成"); + } + + /** + * 删除会话 - 完全保持原deleteConversation逻辑 + */ + private void deleteConversation(ConversationHolder holder) throws IOException { + Map body = new HashMap<>(); + body.put("AppKey", holder.apiId); + body.put("ConversationID", holder.conversationId); + body.put("UserID", holder.userId); + + String jsonBody = objectMapper.writeValueAsString(body); + + Request request = new Request.Builder() + .url(DEFAULT_BASE_URL + "/delete_conversation") + .header("Apikey", holder.apiKey) + .header("Content-Type", "application/json") + .post(RequestBody.create(jsonBody, JSON)) + .build(); + + try (Response response = httpClient.newCall(request).execute()) { + if (!response.isSuccessful()) { + throw new IOException("删除会话失败: " + response.code()); + } + } + } + + /** + * 内部会话持有类 + */ + private static class ConversationHolder { + final String conversationId; + final String apiKey; + final String apiId; + final String userId; + + ConversationHolder(String conversationId, String apiKey, String apiId, String userId) { + this.conversationId = conversationId; + this.apiKey = apiKey; + this.apiId = apiId; + this.userId = userId; + } + } +} \ No newline at end of file diff --git a/cmvr-iot-test/pom.xml b/cmvr-iot-test/pom.xml index 0a0896c..d339b69 100644 --- a/cmvr-iot-test/pom.xml +++ b/cmvr-iot-test/pom.xml @@ -35,5 +35,16 @@ cmvr-iot-grpc-client + + org.graalvm.js + js + 20.3.0 + + + org.graalvm.js + js-scriptengine + 20.3.0 + + \ No newline at end of file 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 0049a41..0300c8a 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 @@ -24,6 +24,7 @@ public enum ActionEnum { SUB_END("NONE", "SUB_END", "子流程结束"), SLEEP("NONE", "SLEEP", "延迟节点"), HTTP("NONE", "HTTP", "HTTP调用"), + CODE("NONE", "CODE", "代码执行"), // AI评估 AI_EVALUATION_PRE("AE", "AI_EVALUATION_PRE", "AI评估预处理"), // AI评估 @@ -66,6 +67,8 @@ public enum ActionEnum { TOUCH_COORDINATES("LLM", "TOUCH_COORDINATES", "获取触控坐标"), INTENT_RECOGNITION("LLM", "INTENT_RECOGNITION", "意图识别"), GENERATE_ADVANCED_AUDIO("LLM", "GENERATE_ADVANCED_AUDIO", "tts语音合成"), + AI_AGENT_PLATFORM("LLM", "AI_AGENT_PLATFORM", "商道智能体"), + AI_TTS("LLM", "AI_TTS", "tts语音播放"), // 触控交互 TI_PATH_SEARCH("EDGE", "TI_PATH_SEARCH", "路径搜索"), diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/NodeTypeEnum.java b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/NodeTypeEnum.java index 706e9e7..966f1aa 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/NodeTypeEnum.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/NodeTypeEnum.java @@ -19,6 +19,7 @@ public enum NodeTypeEnum { END("end"), SLEEP("sleep"), HTTP("http"), + CODE("code"), SUB_END("subEnd"), ; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java index 22d1adb..fb28a36 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java @@ -288,6 +288,8 @@ public class FlowModelBuilder { return NodeTypeEnum.SLEEP; case "http": return NodeTypeEnum.HTTP; + case "code": + return NodeTypeEnum.CODE; default: return NodeTypeEnum.FUNCTION; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowCodeNodeHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowCodeNodeHandler.java new file mode 100644 index 0000000..4844b87 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowCodeNodeHandler.java @@ -0,0 +1,30 @@ +package com.cmvr.test.flow.runtime.dispatcher; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.util.NashornJsExecutor; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + + +@Slf4j +@Component("CODE") +@RequiredArgsConstructor +public class FlowCodeNodeHandler implements FlowNodeTypeHandler { + + private final NashornJsExecutor nashornJsExecutor; + + @Override + public TaskNodeExecuteResult handle(TaskNodeExecuteMessage message) { + String nodeId = message.getNodeId(); + JSONObject inputParams = message.getInputParams(); + String code = inputParams.getString("code"); + inputParams.remove("code"); + log.info("code 节点 {} 开始执行", nodeId); + JSONObject execute = nashornJsExecutor.execute(code, inputParams); + log.info("code 节点 {} 执行完成", nodeId); + return TaskNodeExecuteResult.success(execute); + } +} 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 7cad1e6..16e73b5 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 @@ -18,12 +18,8 @@ public class FlowHttpNodeHandler implements FlowNodeTypeHandler { @Override public TaskNodeExecuteResult handle(TaskNodeExecuteMessage message) { - JSONObject output = new JSONObject(); - output.put("result", "主页"); - return TaskNodeExecuteResult.success(output); - /* try { + try { JSONObject inputParams = message.getInputParams(); - // 1. 必填参数校验 String url = inputParams.getString("url"); if (url == null || url.isEmpty()) { @@ -94,6 +90,6 @@ public class FlowHttpNodeHandler implements FlowNodeTypeHandler { } catch (Exception e) { log.error("[FLOW][HTTP] 执行异常", e); return TaskNodeExecuteResult.failure(e.getMessage()); - }*/ + } } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiAgentPlatformOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiAgentPlatformOperateService.java new file mode 100644 index 0000000..45fa40a --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiAgentPlatformOperateService.java @@ -0,0 +1,36 @@ +package com.cmvr.test.flow.runtime.operator.llm; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.llm.service.LLMAiAgentPlatformService; +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 LLMAiAgentPlatformOperateService implements LLMOperateService { + + private final LLMAiAgentPlatformService llmAiAgentPlatformService; + + @Override + public boolean supports(ActionEnum action) { + return action.name().equals("AI_AGENT_PLATFORM"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + JSONObject inputParams = message.getInputParams(); + String text = inputParams.getString("text"); + String apiKey = inputParams.getString("apiKey"); + + JSONObject output = llmAiAgentPlatformService.query(action.name(), text, apiKey); + + 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/LLMAiTtsOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiTtsOperateService.java new file mode 100644 index 0000000..808e9fa --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiTtsOperateService.java @@ -0,0 +1,38 @@ +package com.cmvr.test.flow.runtime.operator.llm; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.llm.service.LLMAiTtsService; +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 LLMAiTtsOperateService implements LLMOperateService { + + private final LLMAiTtsService llmAiTtsService; + + @Override + public boolean supports(ActionEnum action) { + return action.name().equals("AI_TTS"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + JSONObject inputParams = message.getInputParams(); + String text = inputParams.getString("text"); + String speed = inputParams.getString("speed"); + String volume = inputParams.getString("volume"); + String voice = inputParams.getString("voice"); + String url = inputParams.getString("url"); + + JSONObject output = llmAiTtsService.play(url, text, voice, speed, volume); + + return TaskNodeExecuteResult.success(output); + } +} \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/util/NashornJsExecutor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/util/NashornJsExecutor.java new file mode 100644 index 0000000..27dd3db --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/util/NashornJsExecutor.java @@ -0,0 +1,250 @@ +package com.cmvr.test.util; + +import com.alibaba.fastjson2.JSONObject; +import jdk.nashorn.api.scripting.ClassFilter; +import jdk.nashorn.api.scripting.NashornScriptEngineFactory; +import org.springframework.stereotype.Service; + +import javax.annotation.PreDestroy; +import javax.script.ScriptEngine; +import javax.script.ScriptException; +import java.util.Map; +import java.util.concurrent.*; +import java.util.regex.Pattern; + +@Service +public class NashornJsExecutor { + + // 执行超时时间（秒） + private static final long TIMEOUT_SECONDS = 5; + // 最大返回结果大小（1MB） + private static final int MAX_RESULT_SIZE = 1024 * 1024; + + private final ExecutorService executor; + private final ScriptEngine engine; + + // 危险代码模式 + private static final Pattern DANGEROUS_PATTERN = Pattern.compile( + "(Java|Packages|java|javax|org|com|edu|net)\\s*\\.\\s*([a-zA-Z_$])|" + + "eval\\s*\\(|new\\s+Function\\s*\\(|java\\.lang\\.Runtime" + ); + + public NashornJsExecutor() { + // 创建线程池 + this.executor = new ThreadPoolExecutor( + 2, 10, 60L, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(100), + new ThreadFactory() { + private int count = 0; + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "js-executor-" + (++count)); + t.setDaemon(true); + return t; + } + }, + new ThreadPoolExecutor.CallerRunsPolicy() + ); + + // 创建 Nashorn 引擎，禁止访问所有 Java 类 + NashornScriptEngineFactory factory = new NashornScriptEngineFactory(); + String[] strings = {"--language=es6"}; + this.engine = factory.getScriptEngine( + strings, // 关键在这里 + null, + new ClassFilter() { + @Override + public boolean exposeToScripts(String className) { + return false; + } + } + ); + } + + /** + * 执行 JS 代码 + * + * @param jsCode JS 代码，必须包含 handler 函数 + * @param params 传入的参数（Map 或 POJO） + * @return 执行结果 Map + */ + public JSONObject execute(String jsCode, Object params) { + // 1. 代码安全检查 + validateCode(jsCode); + + // 2. 提交到线程池执行（支持超时中断） + Future future = executor.submit(new Callable() { + @Override + public JSONObject call() throws Exception { + return doExecute(jsCode, params); + } + }); + + try { + return future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + future.cancel(true); + throw new RuntimeException("脚本执行超时（限制 " + TIMEOUT_SECONDS + " 秒）"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("执行被中断"); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + throw new RuntimeException(cause.getMessage(), cause); + } + } + + /** + * 实际执行逻辑 + */ + private JSONObject doExecute(String jsCode, Object params) throws ScriptException { + // 将参数转为 JSON 字符串 + String jsonParams = toJson(params); + + // 构建安全的执行环境 + String wrappedScript = buildSafeScript(jsCode, jsonParams); + + // 执行脚本 + Object result = engine.eval(wrappedScript); + + if (result == null) { + throw new RuntimeException("脚本返回 null"); + } + + String resultJson = result.toString(); + + // 检查结果大小 + if (resultJson.length() > MAX_RESULT_SIZE) { + throw new RuntimeException("返回结果超过 " + (MAX_RESULT_SIZE/1024/1024) + "MB 限制"); + } + + // 解析为 Map + return fromJson(resultJson); + } + + // 更简单的版本：不覆盖 eval，依赖其他安全措施 + private String buildSafeScript(String userCode, String jsonParams) { + StringBuilder sb = new StringBuilder(); + + sb.append("(function() {\n"); + sb.append(" \"use strict\";\n"); + + // 只禁用 Java 访问 + sb.append(" var Java = undefined;\n"); + sb.append(" var Packages = undefined;\n"); + sb.append(" var java = undefined;\n"); + sb.append(" var javax = undefined;\n"); + sb.append(" var com = undefined;\n"); + sb.append(" var org = undefined;\n"); + sb.append(" var edu = undefined;\n"); + sb.append(" var net = undefined;\n"); + + sb.append(" var console = {log: function(){}, error: function(){}};\n"); + sb.append(" \n"); + sb.append(userCode).append("\n"); + sb.append(" \n"); + sb.append(" var params = JSON.parse('").append(escapeJson(jsonParams)).append("');\n"); + sb.append(" var result = handler(params);\n"); + sb.append(" return JSON.stringify(result);\n"); + sb.append("})()"); + + return sb.toString(); + } + + /** + * 代码安全校验 + */ + private void validateCode(String code) { + if (code == null || code.trim().isEmpty()) { + throw new IllegalArgumentException("代码不能为空"); + } + + // 检查是否包含 handler 函数定义 + if (!code.contains("function handler")) { + throw new IllegalArgumentException("代码必须包含 function handler(params) 定义"); + } + + // 黑名单检查 + if (DANGEROUS_PATTERN.matcher(code).find()) { + throw new SecurityException("代码包含危险关键字，已被拦截"); + } + } + + /** + * JSON 字符串转义（防止注入） + */ + private String escapeJson(String json) { + StringBuilder sb = new StringBuilder(); + for (char c : json.toCharArray()) { + switch (c) { + case '\\': + sb.append("\\\\"); + break; + case '\'': + sb.append("\\'"); + break; + case '"': + sb.append("\\\""); + break; + case '\n': + sb.append("\\n"); + break; + case '\r': + sb.append("\\r"); + break; + case '\t': + sb.append("\\t"); + break; + case '\b': + sb.append("\\b"); + break; + case '\f': + sb.append("\\f"); + break; + default: + if (c < 0x20) { + sb.append(String.format("\\u%04x", (int) c)); + } else { + sb.append(c); + } + } + } + return sb.toString(); + } + + /** + * 对象转 JSON（使用 Fastjson） + */ + private String toJson(Object obj) { + try { + return com.alibaba.fastjson.JSON.toJSONString(obj); + } catch (Exception e) { + throw new RuntimeException("参数序列化失败", e); + } + } + + /** + * JSON 转 Map + */ + @SuppressWarnings("unchecked") + private JSONObject fromJson(String json) { + try { + return JSONObject.parseObject(json); + } catch (Exception e) { + throw new RuntimeException("结果反序列化失败", e); + } + } + + @PreDestroy + public void destroy() { + executor.shutdown(); + try { + if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { + executor.shutdownNow(); + } + } catch (InterruptedException e) { + executor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } +} \ No newline at end of file