SmartBench/CMVR-IOT
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev-1' into dev

Browse Source
This commit is contained in:
lixiaolong 2026-03-17 14:41:37 +08:00
commit f365d94077
18 changed files with 1726 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmvr-iot-admin/src/main/resources/application-dev.yml
View File

@ -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

7
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiAgentPlatformService.java Normal file
View File

@ -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);
}

7
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/LLMAiTtsService.java Normal file
View File

@ -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);
}

270
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/WorkflowInvokeServiceCopy.java Normal file
View File

@ -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<String, String> 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<String, Object> 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<String, String> inputParams) {
Map<String, String> headers = buildHeaders(appKey);
Map<String, String> 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<String, String> headers = buildHeaders(appKey);
Map<String, String> 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<String, Object> 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<String, String> buildHeaders(String appKey) {
Map<String, String> headers = new HashMap<>();
headers.put("Apikey", appKey);
return headers;
}
/**
* 构造启动工作流的请求体
*
* @param appKey 应用秘钥
* @param appId 应用ID
* @param inputParams 输入参数
* @return body
*/
private Map<String, String> buildRequestBody(String appKey, String appId, Map<String, String> inputParams) {
Map<String, String> body = buildBaseBody(appKey, appId);
body.put("InputData", JSON.toJSONString(inputParams));
return body;
}
/**
* 构造通用请求体
*
* @param appKey 应用秘钥
* @param appId 应用ID
* @return body
*/
private Map<String, String> buildBaseBody(String appKey, String appId) {
Map<String, String> body = new HashMap<>();
body.put("AppKey", appKey);
body.put("AppID", appId);
body.put("UserID", USER_ID);
return body;
}
}

36
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiAgentPlatformServiceImpl.java Normal file
View File

@ -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<String, Object> 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);
}
}

40
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMAiTtsServiceImpl.java Normal file
View File

@ -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<String, Object> 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);
}
}

26
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/service/impl/LLMIntentRecognitionServiceImpl.java
View File

@ -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<String, String> inputParams = MapUtil.of("text", text);
// Map<String, String> 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<String, Object> 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", "今天天气怎么样"));
}
}

339
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/LlmChatService.java Normal file
View File

@ -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<String, ConversationHolder> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> respMap = objectMapper.readValue(respJson, Map.class);
Map<String, Object> conversation = (Map<String, Object>) 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<String, Object> 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<Throwable> errorRef = new AtomicReference<>();
AtomicReference<String> 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<String, Object> 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<String, Object> 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<String, Object> 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;
}
}
}

629
cmvr-iot-api/cmvr-iot-llm/src/main/java/com/cmvr/llm/util/SseChatClient.java Normal file
View File

@ -0,0 +1,629 @@
package com.cmvr.llm.util;
import com.fasterxml.jackson.databind.ObjectMapper;
import okhttp3.*;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* SSE 流式对话客户端（纯 OkHttp 实现，无额外依赖）
* 包含会话管理、消息管理、流式/阻塞对话
*/
public class SseChatClient {
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();
private final String baseUrl;
private final String apiKey;
private final OkHttpClient httpClient;
// ==================== 数据模型 ====================
/**
* 会话信息
*/
public static class Conversation {
private String conversationId;
private String name;
private long createdAt;
public Conversation(String conversationId, String name, long createdAt) {
this.conversationId = conversationId;
this.name = name;
this.createdAt = createdAt;
}
public String getConversationId() { return conversationId; }
public String getName() { return name; }
public long getCreatedAt() { return createdAt; }
@Override
public String toString() {
return String.format("Conversation{id='%s', name='%s'}", conversationId, name);
}
}
/**
* 消息信息
*/
public static class ChatMessage {
private String messageId;
private String queryId;
private String content;
private String role;
private long createdAt;
public ChatMessage(String messageId, String queryId, String content, String role, long createdAt) {
this.messageId = messageId;
this.queryId = queryId;
this.content = content;
this.role = role;
this.createdAt = createdAt;
}
public String getMessageId() { return messageId; }
public String getQueryId() { return queryId; }
public String getContent() { return content; }
public String getRole() { return role; }
public long getCreatedAt() { return createdAt; }
}
/**
* 流式消息回调接口
*/
public interface StreamCallback {
/**
* 收到消息块时触发（实现打字机效果）
* @param chunk 文本块
* @param messageId 消息ID
*/
void onMessage(String chunk, String messageId);
/**
* 流式传输完成
* @param fullContent 完整内容
* @param messageId 最终消息ID
*/
default void onComplete(String fullContent, String messageId) {}
/**
* 发生错误
* @param error 错误信息
*/
default void onError(Throwable error) {}
}
// ==================== 构造函数 ====================
public SseChatClient(String apiKey) {
this(DEFAULT_BASE_URL, apiKey);
}
public SseChatClient(String baseUrl, String apiKey) {
this.baseUrl = baseUrl;
this.apiKey = apiKey;
this.httpClient = new OkHttpClient.Builder()
.connectTimeout(30, TimeUnit.SECONDS)
.readTimeout(120, TimeUnit.SECONDS)
.writeTimeout(30, TimeUnit.SECONDS)
.build();
}
// ==================== 会话管理 ====================
/**
* 创建新会话
* @param userId 用户ID
* @param name 会话名称（可选）
* @return 会话对象
*/
public Conversation createConversation(String userId, String name) throws IOException {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("UserID", userId);
if (name != null && !name.isEmpty()) {
body.put("Name", name);
}
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/create_conversation")
.header("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<String, Object> respMap = objectMapper.readValue(respJson, Map.class);
// 修正：先取 Conversation 对象，再取 AppConversationID
Map<String, Object> conversation = (Map<String, Object>) 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");
}
String convName = (String) conversation.getOrDefault("ConversationName", "新会话");
long createdAt = ((Number) conversation.getOrDefault("CreatedAt", System.currentTimeMillis())).longValue();
return new Conversation(conversationId, convName, createdAt);
}
}
/**
* 创建会话（自动生成名称）
*/
public Conversation createConversation(String userId) throws IOException {
return createConversation(userId, null);
}
/**
* 获取会话列表
*/
public List<Conversation> getConversationList(String userId, Integer limit, Integer offset) throws IOException {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("UserID", userId);
if (limit != null) body.put("Limit", limit);
if (offset != null) body.put("Offset", offset);
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/get_conversation_list")
.header("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());
}
String respJson = response.body().string();
Map<String, Object> respMap = objectMapper.readValue(respJson, Map.class);
List<Conversation> conversations = new ArrayList<>();
List<Map<String, Object>> list = (List<Map<String, Object>>) respMap.get("ConversationList");
if (list != null) {
for (Map<String, Object> item : list) {
conversations.add(new Conversation(
(String) item.get("ConversationID"),
(String) item.getOrDefault("Name", "未命名"),
((Number) item.getOrDefault("CreatedAt", 0L)).longValue()
));
}
}
return conversations;
}
}
/**
* 删除会话
*/
public void deleteConversation(String conversationId, String userId) throws IOException {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("ConversationID", conversationId);
body.put("UserID", userId);
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/delete_conversation")
.header("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());
}
}
}
// ==================== 消息管理 ====================
/**
* 获取消息详情
*/
public ChatMessage getMessageInfo(String messageId, String userId) throws IOException {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("MessageID", messageId);
body.put("UserID", userId);
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/get_message_info")
.header("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());
}
String respJson = response.body().string();
Map<String, Object> respMap = objectMapper.readValue(respJson, Map.class);
Map<String, Object> msgInfo = (Map<String, Object>) respMap.get("MessageInfo");
if (msgInfo == null) {
throw new IOException("消息不存在");
}
return new ChatMessage(
(String) msgInfo.get("MessageID"),
(String) msgInfo.get("QueryID"),
(String) msgInfo.get("Content"),
(String) msgInfo.get("Role"),
((Number) msgInfo.getOrDefault("CreatedAt", 0L)).longValue()
);
}
}
/**
* 删除消息
*/
public void deleteMessage(String messageId, String queryId, String userId) throws IOException {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("MessageID", messageId);
body.put("UserID", userId);
if (queryId != null && !queryId.isEmpty()) {
body.put("QueryID", queryId);
}
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/delete_message")
.header("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());
}
}
}
// ==================== 对话功能 ====================
/**
* 流式对话（自动创建会话）
* @return 创建的会话ID
*/
public String chatStreamingAuto(String query, String userId, StreamCallback callback) throws IOException {
Conversation conv = createConversation(userId, "自动创建-" + System.currentTimeMillis());
String conversationId = conv.getConversationId();
System.out.println("自动创建会话: " + conversationId);
chatStreaming(query, conversationId, userId, null, callback);
return conversationId;
}
/**
* 流式对话（使用已有会话）
*/
public void chatStreaming(String query, String conversationId, String userId, StreamCallback callback) {
chatStreaming(query, conversationId, userId, null, callback);
}
/**
* 流式对话（支持文件上传）
*/
public void chatStreaming(String query, String conversationId, String userId,
List<Map<String, Object>> files, StreamCallback callback) {
try {
Map<String, Object> requestBody = buildChatRequestBody(query, conversationId, userId, "streaming", files);
String jsonBody = objectMapper.writeValueAsString(requestBody);
Request request = new Request.Builder()
.url(baseUrl + "/chat_query")
.header("Apikey", apiKey)
.header("Content-Type", "application/json")
.header("Accept", "text/event-stream")
.post(RequestBody.create(jsonBody, JSON))
.build();
executeSseRequest(request, callback);
} catch (Exception e) {
callback.onError(e);
}
}
/**
* 阻塞模式对话
*/
public String chatBlocking(String query, String conversationId, String userId) throws IOException {
return chatBlocking(query, conversationId, userId, null);
}
/**
* 阻塞模式对话（支持文件）
*/
public String chatBlocking(String query, String conversationId, String userId,
List<Map<String, Object>> files) throws IOException {
Map<String, Object> requestBody = buildChatRequestBody(query, conversationId, userId, "blocking", files);
String jsonBody = objectMapper.writeValueAsString(requestBody);
Request request = new Request.Builder()
.url(baseUrl + "/chat_query")
.header("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());
}
return response.body().string();
}
}
/**
* 重新生成回复（流式）
*/
public void regenerateStreaming(String conversationId, String messageId,
String userId, StreamCallback callback) {
try {
Map<String, Object> body = new HashMap<>();
body.put("AppKey", apiKey);
body.put("AppConversationID", conversationId);
body.put("MessageID", messageId);
body.put("UserID", userId);
String jsonBody = objectMapper.writeValueAsString(body);
Request request = new Request.Builder()
.url(baseUrl + "/query_again")
.header("Apikey", apiKey)
.header("Content-Type", "application/json")
.header("Accept", "text/event-stream")
.post(RequestBody.create(jsonBody, JSON))
.build();
executeSseRequest(request, callback);
} catch (Exception e) {
callback.onError(e);
}
}
// ==================== 私有辅助方法 ====================
private Map<String, Object> buildChatRequestBody(String query, String conversationId,
String userId, String mode,
List<Map<String, Object>> files) {
Map<String, Object> body = new HashMap<>();
body.put("Query", query);
body.put("AppConversationID", conversationId);
body.put("AppKey", apiKey);
body.put("ResponseMode", mode);
body.put("UserID", userId);
body.put("PubAgentJump", true);
if (files != null && !files.isEmpty()) {
Map<String, Object> queryExtends = new HashMap<>();
queryExtends.put("Files", files);
body.put("QueryExtends", queryExtends);
}
return body;
}
/**
* 执行 SSE 请求（核心方法）
*/
/**
* 执行 SSE 请求（修正版）
* 处理实际格式：event:text + data:data: {...}
*/
private void executeSseRequest(Request request, StreamCallback callback) throws IOException {
Call call = httpClient.newCall(request);
Response response = call.execute();
if (!response.isSuccessful()) {
String errorBody = response.body() != null ? response.body().string() : "无错误信息";
throw new IOException("SSE请求失败: " + response.code() + " - " + errorBody);
}
StringBuilder fullContent = 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;
}
// 解析 event 行
if (line.startsWith("event:")) {
currentEvent = line.substring(6).trim();
continue;
}
// 解析 data 行（注意：实际是 data:data: {...} 格式）
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)) {
callback.onComplete(fullContent.toString(), finalMessageId);
return;
}
try {
Map<String, Object> jsonData = objectMapper.readValue(data, Map.class);
// 提取消息ID
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)) {
// 提取回答内容（在 answer 字段）
String answer = extractString(jsonData, "answer", "content", "chunk", "text");
if (!answer.isEmpty()) {
fullContent.append(answer);
callback.onMessage(answer, finalMessageId);
}
} 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)) {
// 消息结束
callback.onComplete(fullContent.toString(), finalMessageId);
return;
}
// 其他事件类型可扩展...
} catch (Exception e) {
// JSON解析失败，打印调试
System.out.println("解析失败，原始数据: " + data);
}
}
}
// 流正常结束（没有明确结束标记）
callback.onComplete(fullContent.toString(), finalMessageId);
} catch (IOException e) {
callback.onError(e);
throw e;
}
}
/**
* 从 Map 中提取字符串（尝试多个 key）
*/
private String extractString(Map<String, Object> map, String... keys) {
for (String key : keys) {
Object value = map.get(key);
if (value != null) {
return value.toString();
}
}
return "";
}
/**
* 关闭客户端
*/
public void close() {
httpClient.dispatcher().executorService().shutdown();
httpClient.connectionPool().evictAll();
}
// ==================== 使用示例 ====================
public static void main(String[] args) {
SseChatClient client = new SseChatClient("d6gl6n6llh49on40fqug");
try {
String userId = "user_" + System.currentTimeMillis();
// ========== 示例1：标准流程 ==========
System.out.println("=== 步骤1：创建会话 ===");
Conversation conv = client.createConversation(userId, "测试会话");
System.out.println("会话创建成功: " + conv.getConversationId());
System.out.println("\n=== 步骤2：流式对话 ===");
final StringBuilder contentBuilder = new StringBuilder();
final String[] lastMessageId = {""};
client.chatStreaming(
"给我讲一个关于人工智能的短故事",
conv.getConversationId(),
userId,
new StreamCallback() {
@Override
public void onMessage(String chunk, String messageId) {
System.out.print(chunk);
System.out.flush();
contentBuilder.append(chunk);
if (!messageId.isEmpty()) {
lastMessageId[0] = messageId;
}
}
@Override
public void onComplete(String fullContent, String messageId) {
System.out.println("\n\n[对话完成] 消息ID: " + messageId);
}
@Override
public void onError(Throwable error) {
System.err.println("\n[错误] " + error.getMessage());
}
}
);
String string = contentBuilder.toString();
// Thread.sleep(15000);
// // 等待流式传输完成
// Thread.sleep(15000);
// String string = contentBuilder.toString();
// ========== 示例3：获取会话列表 ==========
System.out.println("\n=== 步骤4：获取会话列表 ===");
List<Conversation> conversations = client.getConversationList(userId, 10, 0);
System.out.println("共有 " + conversations.size() + " 个会话:");
for (Conversation c : conversations) {
System.out.println(" - " + c);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
client.close();
}
}
}

11
cmvr-iot-test/pom.xml
View File

@ -35,5 +35,16 @@
<artifactId>cmvr-iot-grpc-client</artifactId>
</dependency>
<dependency>
<groupId>org.graalvm.js</groupId>
<artifactId>js</artifactId>
<version>20.3.0</version> <!-- 最后一个支持 JDK 8 的版本 -->
</dependency>
<dependency>
<groupId>org.graalvm.js</groupId>
<artifactId>js-scriptengine</artifactId>
<version>20.3.0</version>
</dependency>
</dependencies>
</project>

3
cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java
View File

@ -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", "路径搜索"),

1
cmvr-iot-test/src/main/java/com/cmvr/test/enums/NodeTypeEnum.java
View File

@ -19,6 +19,7 @@ public enum NodeTypeEnum {
END("end"),
SLEEP("sleep"),
HTTP("http"),
CODE("code"),
SUB_END("subEnd"),
;

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java
View File

@ -288,6 +288,8 @@ public class FlowModelBuilder {
return NodeTypeEnum.SLEEP;
case "http":
return NodeTypeEnum.HTTP;
case "code":
return NodeTypeEnum.CODE;
default:
return NodeTypeEnum.FUNCTION;
}

30
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowCodeNodeHandler.java Normal file
View File

@ -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);
}
}

8
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowHttpNodeHandler.java
View File

@ -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());
}*/
}
}
}

36
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiAgentPlatformOperateService.java Normal file
View File

@ -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);
}
}

38
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMAiTtsOperateService.java Normal file
View File

@ -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);
}
}

250
cmvr-iot-test/src/main/java/com/cmvr/test/util/NashornJsExecutor.java Normal file
View File

@ -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<JSONObject> future = executor.submit(new Callable<JSONObject>() {
@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();
}
}
}