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