feat: 商道调用封装、流式闲聊
This commit is contained in:
stream
parent
9f324941d3
commit
ac37baf54b
@ -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
|
||||
|
||||
@ -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<String, AgentConfig> agents;
|
||||
|
||||
/**
|
||||
* 根据 ActionEnum 名称获取 app 配置
|
||||
*/
|
||||
public AgentConfig getAgentConfig(String actionName) {
|
||||
AgentConfig config = agents.get(actionName);
|
||||
if (config == null) {
|
||||
throw new GlobalException("未找到功能 {} 的 智能体应用配置", actionName);
|
||||
}
|
||||
return config;
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,18 @@
|
||||
package com.cmvr.llm.config;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* 智能体应用配置
|
||||
*/
|
||||
@Data
|
||||
public class AgentConfig {
|
||||
/**
|
||||
* 应用id
|
||||
*/
|
||||
private String appId;
|
||||
/**
|
||||
* 应用秘钥
|
||||
*/
|
||||
private String appKey;
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
package com.cmvr.llm.service;
|
||||
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
|
||||
public interface LLMIntentRecognitionService {
|
||||
|
||||
/**
|
||||
* 意图识别
|
||||
*/
|
||||
public JSONObject intentRecognition(String action,String text);
|
||||
}
|
||||
@ -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<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);
|
||||
}
|
||||
|
||||
/**
|
||||
* 一站式执行工作流（使用默认重试参数）
|
||||
*
|
||||
* @param appId 智能体应用ID
|
||||
* @param appKey 智能体应用秘钥
|
||||
* @param inputParams 输入参数（业务参数，例如图片、描述等）
|
||||
* @return 最终结果 message
|
||||
*/
|
||||
public JSONObject executeWorkflow(String appId, String appKey, Map<String, String> 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<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;
|
||||
}
|
||||
}
|
||||
@ -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<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);
|
||||
}
|
||||
}
|
||||
@ -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");
|
||||
|
||||
@ -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")) {
|
||||
|
||||
@ -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");
|
||||
|
||||
@ -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语音合成"),
|
||||
|
||||
// 语音交互
|
||||
|
||||
@ -229,9 +229,18 @@ public class FlowGraph {
|
||||
List<FlowEdge> outEdges = successors.getOrDefault(nodeId, Collections.emptyList());
|
||||
for (FlowEdge edge : new ArrayList<>(outEdges)) {
|
||||
String targetNodeId = edge.getTo();
|
||||
|
||||
// 先删边
|
||||
removeEdge(edge);
|
||||
|
||||
// 只有目标节点没有其他前驱时才递归
|
||||
List<FlowEdge> predecessorsList = predecessors.getOrDefault(targetNodeId, Collections.emptyList());
|
||||
if (predecessorsList.isEmpty()) {
|
||||
removePathFrom(targetNodeId);
|
||||
}
|
||||
}
|
||||
|
||||
// 最后判断是否孤立再删节点
|
||||
removeNodeIfIsolated(nodeId);
|
||||
} finally {
|
||||
lock.writeLock().unlock();
|
||||
|
||||
@ -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;
|
||||
}
|
||||
@ -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<FlowParamDef> startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished,
|
||||
List<Integer> 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<FlowEdge> 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<FlowEdge> predecessorsList = graph.getPredecessors().getOrDefault(edge.getTo(), Collections.emptyList());
|
||||
if (predecessorsList.isEmpty()) {
|
||||
graph.removePathFrom(edge.getTo());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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<FlowBranchConditionGroup> groups, JSONObject inputParams) {
|
||||
public static String evaluate(TaskContext taskContext, List<Integer> iterations, String nodeId, List<FlowBranchConditionGroup> 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<Integer> iterations, JSONObject inputParams, List<FlowBranchConditionGroup> groups) {
|
||||
if (CollUtil.isEmpty(taskContext.getNodeOutputs())) {
|
||||
return;
|
||||
}
|
||||
HashSet<String> quoteIds = new HashSet<>();
|
||||
for (FlowBranchConditionGroup group : groups) {
|
||||
List<FlowBranchConditionGroup.ConditionExpression> expressions = group.getExpressions();
|
||||
for (FlowBranchConditionGroup.ConditionExpression expression : expressions) {
|
||||
if (expression.getNameType().equalsIgnoreCase("quote")) {
|
||||
List<String> 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) {
|
||||
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
@ -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<String, BlockingQueue<AudioSegment>> audioQueues = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* instId -> 播放线程
|
||||
*/
|
||||
private final Map<String, Thread> 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<String> slices = TextSplitUtil.splitText(text);
|
||||
if (slices.isEmpty()) {
|
||||
throw new GlobalException("输入文本为空，无法合成语音");
|
||||
}
|
||||
|
||||
// 1. 构建请求体
|
||||
// 2. 确保队列和线程存在
|
||||
BlockingQueue<AudioSegment> 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<AudioSegment> 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<String, Object> 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; // 仅最后一段有
|
||||
}
|
||||
|
||||
}
|
||||
@ -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<String> splitText(String text) {
|
||||
return splitText(text, DEFAULT_MAX_LENGTH);
|
||||
}
|
||||
|
||||
public static List<String> splitText(String text, int maxLength) {
|
||||
List<String> 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<String> splitByComma(String text, int maxLength) {
|
||||
List<String> 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<String> splitByLength(String text, int maxLength) {
|
||||
List<String> 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();
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user