userInputs, Long sessionId);
-
- /**
- * 恢复工作流
- * @param runtimeUuid 运行时UUID
- * @param userInput 用户输入
- * @param sseEmitter SSE连接对象
- */
- void resumeFlow(String runtimeUuid, String userInput, SseEmitter sseEmitter);
}
diff --git a/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AbstractAuthWeChatEnterpriseRequest.java b/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AbstractAuthWeChatEnterpriseRequest.java
index abfa3a95..031f00aa 100644
--- a/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AbstractAuthWeChatEnterpriseRequest.java
+++ b/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AbstractAuthWeChatEnterpriseRequest.java
@@ -15,9 +15,6 @@ import me.zhyd.oauth.utils.HttpUtils;
import me.zhyd.oauth.utils.StringUtils;
import me.zhyd.oauth.utils.UrlBuilder;
-// 临时保留FastJson用于JustAuth库兼容
-import com.alibaba.fastjson.JSON;
-
/**
*
* 企业微信登录父类
@@ -64,11 +61,8 @@ public abstract class AbstractAuthWeChatEnterpriseRequest extends AuthDefaultReq
String userTicket = object.has("user_ticket") ? object.get("user_ticket").asText() : null;
JsonNode userDetail = getUserDetail(authToken.getAccessToken(), userId, userTicket);
- // 将JsonNode转换为JSONObject以兼容JustAuth库
- com.alibaba.fastjson.JSONObject rawUserInfo = com.alibaba.fastjson.JSON.parseObject(userDetail.toString());
-
+ // rawUserInfo 为 JustAuth 的 fastjson 类型字段, 项目内无消费方, 不再设置
return AuthUser.builder()
- .rawUserInfo(rawUserInfo)
.username(userDetail.has("name") ? userDetail.get("name").asText() : null)
.nickname(userDetail.has("alias") ? userDetail.get("alias").asText() : null)
.avatar(userDetail.has("avatar") ? userDetail.get("avatar").asText() : null)
diff --git a/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AuthDingTalkV2Request.java b/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AuthDingTalkV2Request.java
index 1885b9f0..dbfeec9e 100644
--- a/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AuthDingTalkV2Request.java
+++ b/ruoyi-common/ruoyi-common-social/src/main/java/me/zhyd/oauth/request/AuthDingTalkV2Request.java
@@ -19,9 +19,6 @@ import me.zhyd.oauth.utils.UrlBuilder;
import java.util.HashMap;
import java.util.Map;
-// 临时保留FastJson用于JustAuth库兼容
-import com.alibaba.fastjson.JSON;
-
/**
* 新版钉钉二维码登录
*
@@ -92,13 +89,10 @@ public class AuthDingTalkV2Request extends AuthDefaultRequest {
String response = new HttpUtils(config.getHttpConfig()).get(this.source.userInfo(), null, header, false).getBody();
JsonNode object = objectMapper.readTree(response);
- // 将JsonNode转换为JSONObject以兼容JustAuth库
- com.alibaba.fastjson.JSONObject rawUserInfo = com.alibaba.fastjson.JSON.parseObject(object.toString());
-
authToken.setOpenId(object.has("openId") ? object.get("openId").asText() : null);
authToken.setUnionId(object.has("unionId") ? object.get("unionId").asText() : null);
+ // rawUserInfo 为 JustAuth 的 fastjson 类型字段, 项目内无消费方, 不再设置
return AuthUser.builder()
- .rawUserInfo(rawUserInfo)
.uuid(object.has("unionId") ? object.get("unionId").asText() : null)
.username(object.has("nick") ? object.get("nick").asText() : null)
.nickname(object.has("nick") ? object.get("nick").asText() : null)
diff --git a/ruoyi-modules/ruoyi-aiflow/pom.xml b/ruoyi-modules/ruoyi-aiflow/pom.xml
index e2419ca9..0970ec70 100644
--- a/ruoyi-modules/ruoyi-aiflow/pom.xml
+++ b/ruoyi-modules/ruoyi-aiflow/pom.xml
@@ -24,6 +24,13 @@
+
+
+ ai.z.openapi
+ zai-sdk
+ 0.3.5
+
+
org.ruoyi
ruoyi-common-chat
@@ -45,6 +52,11 @@
ruoyi-common-satoken
+
+ org.ruoyi
+ ruoyi-common-tenant
+
+
org.ruoyi
ruoyi-common-mail
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/controller/WorkflowRuntimeController.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/controller/WorkflowRuntimeController.java
index 2d384aaa..be400d59 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/controller/WorkflowRuntimeController.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/controller/WorkflowRuntimeController.java
@@ -1,16 +1,13 @@
package org.ruoyi.workflow.controller;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import io.swagger.v3.oas.annotations.Operation;
import jakarta.annotation.Resource;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotNull;
import org.ruoyi.common.core.domain.R;
import org.ruoyi.workflow.dto.workflow.WfRuntimeNodeDto;
import org.ruoyi.workflow.dto.workflow.WfRuntimeResp;
-import org.ruoyi.workflow.dto.workflow.WorkflowResumeReq;
import org.ruoyi.workflow.service.WorkflowRuntimeService;
-import org.ruoyi.workflow.workflow.WorkflowStarter;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
@@ -24,16 +21,6 @@ public class WorkflowRuntimeController {
@Resource
private WorkflowRuntimeService workflowRuntimeService;
- @Resource
- private WorkflowStarter workflowStarter;
-
- @Operation(summary = "接收用户输入以继续执行剩余流程")
- @PostMapping(value = "/resume/{runtimeUuid}")
- public R resume(@PathVariable String runtimeUuid, @RequestBody WorkflowResumeReq resumeReq) {
- workflowStarter.resumeFlow(runtimeUuid, resumeReq.getFeedbackContent(), resumeReq.getSseEmitter());
- return R.ok();
- }
-
@GetMapping("/page")
public R> search(@RequestParam String wfUuid,
@NotNull @Min(1) Integer currentPage,
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/cosntant/AdiConstant.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/cosntant/AdiConstant.java
index 7370fcc0..33de7fd2 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/cosntant/AdiConstant.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/cosntant/AdiConstant.java
@@ -337,7 +337,6 @@ public class AdiConstant {
public static final String DEFAULT_INPUT_PARAM_NAME = "input";
public static final String DEFAULT_OUTPUT_PARAM_NAME = "output";
public static final String DEFAULT_ERROR_OUTPUT_PARAM_NAME = "error_msg";
- public static final String HUMAN_FEEDBACK_KEY = "human_feedback";
public static final int NODE_PROCESS_STATUS_READY = 1;
public static final int NODE_PROCESS_STATUS_DOING = 2;
public static final int NODE_PROCESS_STATUS_SUCCESS = 3;
@@ -347,7 +346,6 @@ public class AdiConstant {
public static final int WORKFLOW_PROCESS_STATUS_DOING = 2;
public static final int WORKFLOW_PROCESS_STATUS_SUCCESS = 3;
public static final int WORKFLOW_PROCESS_STATUS_FAIL = 4;
- public static final int WORKFLOW_PROCESS_STATUS_WAITING_INPUT = 5;
public static final int WORKFLOW_NODE_PROCESS_TYPE_NORMAL = 1;
public static final int WORKFLOW_NODE_PROCESS_TYPE_CONDITIONAL = 2;
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/dto/workflow/WorkflowResumeReq.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/dto/workflow/WorkflowResumeReq.java
deleted file mode 100644
index c1b64829..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/dto/workflow/WorkflowResumeReq.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package org.ruoyi.workflow.dto.workflow;
-
-import lombok.Data;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-
-@Data
-public class WorkflowResumeReq {
- private String feedbackContent;
- private SseEmitter sseEmitter;
-}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/helper/SSEEmitterHelper.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/helper/SSEEmitterHelper.java
index 61b3dda0..1b510a8b 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/helper/SSEEmitterHelper.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/helper/SSEEmitterHelper.java
@@ -55,8 +55,13 @@ public class SSEEmitterHelper {
} else {
sseEmitter.send(msg);
}
+ } catch (IllegalStateException ise) {
+ // SSE连接已关闭（用户刷新页面、关闭标签页或重新提交）
+ log.warn("SSE emitter already completed for event [{}], ignoring", name);
+ COMPLETED_SSE.put(sseEmitter, Boolean.TRUE);
} catch (IOException ioException) {
log.error("stream onNext error", ioException);
+ COMPLETED_SSE.put(sseEmitter, Boolean.TRUE);
}
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/util/WorkflowMessageUtil.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/util/WorkflowMessageUtil.java
index f94e4a4c..33d7402f 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/util/WorkflowMessageUtil.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/util/WorkflowMessageUtil.java
@@ -4,7 +4,6 @@ import org.ruoyi.common.chat.domain.dto.request.ChatRequest;
import org.ruoyi.common.chat.domain.vo.chat.ChatModelVo;
import org.ruoyi.common.chat.enums.RoleType;
import lombok.extern.slf4j.Slf4j;
-import org.ruoyi.common.core.exception.ServiceException;
import org.ruoyi.common.core.service.ConfigService;
import org.ruoyi.common.core.utils.SpringUtils;
import org.ruoyi.common.core.utils.StringUtils;
@@ -12,6 +11,7 @@ import org.ruoyi.workflow.entity.WorkflowNode;
import org.ruoyi.workflow.helper.SSEEmitterHelper;
import org.ruoyi.workflow.workflow.WfState;
import org.ruoyi.workflow.workflow.WorkflowUtil;
+import org.ruoyi.workflow.workflow.node.enmus.NodeMessageTemplateEnum;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
@@ -38,17 +38,20 @@ public class WorkflowMessageUtil {
/**
- * 获取节点的响应模板
+ * 获取节点的响应模板
+ * 优先读取 sys_config 配置(可在系统管理-配置管理中自定义),
+ * 未配置时回退到枚举内置默认模板, 模板仅为展示文案, 缺失不应中断工作流执行
* @param configKey 参数Key
* @return 返回模板样式
*/
public static String getNodeMessageTemplate(String configKey){
ConfigService configService = SpringUtil.getBean(ConfigService.class);
String configValue = configService.getConfigValue(configKey);
- if (StringUtils.isEmpty(configValue)) {
- throw new ServiceException("请先配置该节点的响应模板");
+ if (StringUtils.isNotEmpty(configValue)) {
+ return configValue;
}
- return configValue;
+ log.warn("sys_config 未配置节点响应模板 [{}], 已回退使用内置默认模板", configKey);
+ return NodeMessageTemplateEnum.getDefaultTemplate(configKey);
}
/**
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/InterruptedFlow.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/InterruptedFlow.java
deleted file mode 100644
index 34b0018f..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/InterruptedFlow.java
+++ /dev/null
@@ -1,17 +0,0 @@
-package org.ruoyi.workflow.workflow;
-
-import org.apache.commons.collections4.map.PassiveExpiringMap;
-
-/**
- * 已中断正在等待用户输入的流程
- * TODO 需要考虑项目多节点部署的情况
- */
-public class InterruptedFlow {
-
- /**
- * 10分钟超时
- */
- private static final PassiveExpiringMap.ExpirationPolicy ep = new PassiveExpiringMap.ConstantTimeToLiveExpirationPolicy<>(60 * 1000 * 10);
- public static PassiveExpiringMap RUNTIME_TO_GRAPH = new PassiveExpiringMap<>(ep);
-
-}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfComponentNameEnum.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfComponentNameEnum.java
index 5e40a6d4..a27732fc 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfComponentNameEnum.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfComponentNameEnum.java
@@ -16,24 +16,14 @@ public enum WfComponentNameEnum {
TONGYI_WANX("Tongyiwanx"),
- DOCUMENT_EXTRACTOR("DocumentExtractor"),
-
- KEYWORD_EXTRACTOR("KeywordExtractor"),
-
FAQ_EXTRACTOR("FaqExtractor"),
KNOWLEDGE_RETRIEVER("KnowledgeRetrieval"),
SWITCHER("Switcher"),
- CLASSIFIER("Classifier"),
-
- TEMPLATE("Template"),
-
GOOGLE_SEARCH("Google"),
- HUMAN_FEEDBACK("HumanFeedback"),
-
MAIL_SEND("MailSend"),
HTTP_REQUEST("HttpRequest");
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfNodeFactory.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfNodeFactory.java
index ce22efca..1b36eabe 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfNodeFactory.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfNodeFactory.java
@@ -4,15 +4,14 @@ import org.ruoyi.workflow.entity.WorkflowComponent;
import org.ruoyi.workflow.entity.WorkflowNode;
import org.ruoyi.workflow.workflow.node.AbstractWfNode;
import org.ruoyi.workflow.workflow.node.EndNode;
-import org.ruoyi.workflow.workflow.node.humanFeedBack.HumanFeedbackNode;
import org.ruoyi.workflow.workflow.node.answer.LLMAnswerNode;
import org.ruoyi.workflow.workflow.node.httpRequest.HttpRequestNode;
import org.ruoyi.workflow.workflow.node.image.ImageNode;
-import org.ruoyi.workflow.workflow.node.keywordExtractor.KeywordExtractorNode;
import org.ruoyi.workflow.workflow.node.knowledgeRetrieval.KnowledgeRetrievalNode;
import org.ruoyi.workflow.workflow.node.mailSend.MailSendNode;
import org.ruoyi.workflow.workflow.node.start.StartNode;
import org.ruoyi.workflow.workflow.node.switcher.SwitcherNode;
+import org.ruoyi.workflow.workflow.node.googleSearch.GoogleSearchNode;
public class WfNodeFactory {
public static AbstractWfNode create(WorkflowComponent wfComponent, WorkflowNode nodeDefinition,
@@ -21,14 +20,13 @@ public class WfNodeFactory {
switch (WfComponentNameEnum.getByName(wfComponent.getName())) {
case START -> wfNode = new StartNode(wfComponent, nodeDefinition, wfState, nodeState);
case LLM_ANSWER -> wfNode = new LLMAnswerNode(wfComponent, nodeDefinition, wfState, nodeState);
- case KEYWORD_EXTRACTOR -> wfNode = new KeywordExtractorNode(wfComponent, nodeDefinition, wfState, nodeState);
case TONGYI_WANX -> wfNode = new ImageNode(wfComponent, nodeDefinition, wfState, nodeState);
case KNOWLEDGE_RETRIEVER -> wfNode = new KnowledgeRetrievalNode(wfComponent, nodeDefinition, wfState, nodeState);
case END -> wfNode = new EndNode(wfComponent, nodeDefinition, wfState, nodeState);
case MAIL_SEND -> wfNode = new MailSendNode(wfComponent, nodeDefinition, wfState, nodeState);
case HTTP_REQUEST -> wfNode = new HttpRequestNode(wfComponent, nodeDefinition, wfState, nodeState);
case SWITCHER -> wfNode = new SwitcherNode(wfComponent, nodeDefinition, wfState, nodeState);
- case HUMAN_FEEDBACK -> wfNode = new HumanFeedbackNode(wfComponent, nodeDefinition, wfState, nodeState);
+ case GOOGLE_SEARCH -> wfNode = new GoogleSearchNode(wfComponent, nodeDefinition, wfState, nodeState);
default -> {
}
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfState.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfState.java
index d0595803..d781d95b 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfState.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WfState.java
@@ -55,11 +55,6 @@ public class WfState {
private List output = new ArrayList<>();
private Integer processStatus = WORKFLOW_PROCESS_STATUS_READY;
- /**
- * 人机交互节点
- */
- private Set interruptNodes = new HashSet<>();
-
public WfState(User user, List input, String uuid, Long userId, String tokenValue, SseEmitter sseEmitter, Long sessionId) {
this.input = input;
this.user = user;
@@ -133,8 +128,4 @@ public class WfState {
.findFirst()
.orElse(null);
}
-
- public void addInterruptNode(String nodeUuid) {
- this.interruptNodes.add(nodeUuid);
- }
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowEngine.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowEngine.java
index 24d0110f..2fecf118 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowEngine.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowEngine.java
@@ -108,82 +108,39 @@ public class WorkflowEngine {
MemorySaver saver = new MemorySaver();
CompileConfig compileConfig = CompileConfig.builder().checkpointSaver(saver)
- .interruptBefore(wfState.getInterruptNodes().toArray(String[]::new))
.build();
app = mainStateGraph.compile(compileConfig);
RunnableConfig invokeConfig = RunnableConfig.builder().build();
- exe(invokeConfig, false);
+ exe(invokeConfig);
} catch (Exception e) {
errorWhenExe(e);
}
}
- private void exe(RunnableConfig invokeConfig, boolean resume) {
+ private void exe(RunnableConfig invokeConfig) {
//不使用langgraph4j state的update相关方法，无需传入input
- AsyncGenerator> outputs = app.stream(resume ? null : Map.of(), invokeConfig);
+ AsyncGenerator> outputs = app.stream(Map.of(), invokeConfig);
streamingResult(wfState, outputs, sseEmitter);
StateSnapshot stateSnapshot = app.getState(invokeConfig);
- String nextNode = stateSnapshot.config().nextNode().orElse("");
- //还有下个节点，表示进入中断状态，等待用户输入后继续执�?
- if (StringUtils.isNotBlank(nextNode) && !nextNode.equalsIgnoreCase(END)) {
- // 获取提示模板
- String nodeMessageTemplate = WorkflowMessageUtil.getNodeMessageTemplate(NodeMessageTemplateEnum.HUMAN_FEED_BACK.getValue());
- // 获取人机交互提示信息
- String intTip = nodeMessageTemplate + WorkflowUtil.getHumanFeedbackTip(nextNode, wfNodes);
- //将等待输入信息[事件与提示词]发送到到客户端
- SSEEmitterHelper.parseAndSendPartialMsg(sseEmitter, "[NODE_WAIT_FEEDBACK_BY_" + nextNode + "]", intTip);
- // 保存提示信息到Chat信息记录中（对话使用）
- WorkflowMessageUtil.saveWorkflowMessage(wfState, intTip);
- InterruptedFlow.RUNTIME_TO_GRAPH.put(wfState.getUuid(), this);
- //更新状�?
- wfState.setProcessStatus(WORKFLOW_PROCESS_STATUS_WAITING_INPUT);
- workflowRuntimeService.updateOutput(wfRuntimeResp.getId(), wfState);
- } else {
- WorkflowRuntime updatedRuntime = workflowRuntimeService.updateOutput(wfRuntimeResp.getId(), wfState);
- // 保存成功会话信息
- wfNodes.stream().filter(item -> stateSnapshot.node().equals(item.getUuid()))
- .findFirst().ifPresent(wfNode -> {
- // 获取节点模板提示词信息
- String nodeMessageTemplate = WorkflowMessageUtil.getNodeMessageTemplate(NodeMessageTemplateEnum.END.getValue());
- // 发送SSE消息驱动事件和保存会话
- WorkflowMessageUtil.notifyAndStoreMessage(wfState, sseEmitter, wfNode, nodeMessageTemplate);
- });
- // 发送结束消息
- sseEmitterHelper.sendComplete(user.getId(), sseEmitter, updatedRuntime.getOutput());
- // 发送驱动消息事件
- InterruptedFlow.RUNTIME_TO_GRAPH.remove(wfState.getUuid());
- }
- }
-
- /**
- * 中断流程等待用户输入时，会进行暂停状态，用户输入后调用本方法执行流程剩余部分
- *
- * @param userInput 用户输入
- */
- public void resume(String userInput) {
- RunnableConfig invokeConfig = RunnableConfig.builder().build();
- try {
- app.updateState(invokeConfig, Map.of(HUMAN_FEEDBACK_KEY, userInput), null);
- exe(invokeConfig, true);
- } catch (Exception e) {
- errorWhenExe(e);
- } finally {
- //有可能多次接收人机交互，待整个流程完全执行后才能删除
- if (wfState.getProcessStatus() != WORKFLOW_PROCESS_STATUS_WAITING_INPUT) {
- InterruptedFlow.RUNTIME_TO_GRAPH.remove(wfState.getUuid());
- }
- }
+ wfState.setProcessStatus(WORKFLOW_PROCESS_STATUS_SUCCESS);
+ WorkflowRuntime updatedRuntime = workflowRuntimeService.updateOutput(wfRuntimeResp.getId(), wfState);
+ wfNodes.stream().filter(item -> stateSnapshot.node().equals(item.getUuid()))
+ .findFirst().ifPresent(wfNode -> {
+ String nodeMessageTemplate = WorkflowMessageUtil.getNodeMessageTemplate(NodeMessageTemplateEnum.END.getValue());
+ WorkflowMessageUtil.notifyAndStoreMessage(wfState, sseEmitter, wfNode, nodeMessageTemplate);
+ });
+ sseEmitterHelper.sendComplete(user.getId(), sseEmitter, updatedRuntime.getOutput());
}
private void errorWhenExe(Exception e) {
log.error("error", e);
String nodeMessageTemplate = WorkflowMessageUtil.getNodeMessageTemplate(NodeMessageTemplateEnum.EXCEPTION.getValue());
String errorMsg = e.getMessage();
- if (errorMsg.contains("parallel node doesn't support conditional branch")) {
+ if (errorMsg != null && errorMsg.contains("parallel node doesn't support conditional branch")) {
errorMsg = "并行节点中不能包含条件分�?";
}
- errorMsg = nodeMessageTemplate + errorMsg;
+ errorMsg = nodeMessageTemplate + (errorMsg != null ? errorMsg : e.getClass().getSimpleName());
// 保存会话信息且发送驱动消息事件
WorkflowMessageUtil.saveWorkflowMessage(wfState, errorMsg);
sseEmitterHelper.sendErrorAndComplete(user.getId(), sseEmitter, errorMsg);
@@ -267,6 +224,10 @@ public class WorkflowEngine {
log.info("node:{},chunk:{}", node, chunk);
SSEEmitterHelper.parseAndSendPartialMsg(sseEmitter, "[NODE_CHUNK_" + node + "]", chunk);
} else {
+ // __END__ 是 langgraph4j 的终止伪节点, 无对应业务节点状态, 跳过
+ if (END.equals(out.node())) {
+ continue;
+ }
AbstractWfNode abstractWfNode = wfState.getCompletedNodes().stream()
.filter(item -> item.getNode().getUuid().endsWith(out.node())).findFirst().orElse(null);
if (null != abstractWfNode) {
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowGraphBuilder.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowGraphBuilder.java
index 55241376..c74e7d2e 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowGraphBuilder.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowGraphBuilder.java
@@ -19,7 +19,6 @@ import static org.bsc.langgraph4j.StateGraph.END;
import static org.bsc.langgraph4j.StateGraph.START;
import static org.bsc.langgraph4j.action.AsyncEdgeAction.edge_async;
import static org.bsc.langgraph4j.action.AsyncNodeAction.node_async;
-import static org.ruoyi.workflow.workflow.WfComponentNameEnum.HUMAN_FEEDBACK;
/**
* 负责构建工作流运行所依赖的状态图�?
@@ -27,7 +26,6 @@ import static org.ruoyi.workflow.workflow.WfComponentNameEnum.HUMAN_FEEDBACK;
@Slf4j
public class WorkflowGraphBuilder {
- private final Map componentIndex;
private final Map nodeIndex;
private final Map> edgesBySource;
private final Map> edgesByTarget;
@@ -46,8 +44,6 @@ public class WorkflowGraphBuilder {
List edges,
WorkflowNodeRunner nodeRunner,
WfState wfState) {
- this.componentIndex = components.stream()
- .collect(Collectors.toMap(WorkflowComponent::getId, Function.identity(), (origin, ignore) -> origin));
this.nodeIndex = nodes.stream()
.collect(Collectors.toMap(WorkflowNode::getUuid, Function.identity(), (origin, ignore) -> origin));
this.edgesBySource = edges.stream().collect(Collectors.groupingBy(WorkflowEdge::getSourceNodeUuid));
@@ -217,14 +213,6 @@ public class WorkflowGraphBuilder {
WorkflowNode wfNode = getNodeByUuid(stateGraphNodeUuid);
stateGraph.addNode(stateGraphNodeUuid, node_async(state -> nodeRunner.run(wfNode, state)));
stateGraphList.add(stateGraph);
-
- WorkflowComponent component = componentIndex.get(wfNode.getWorkflowComponentId());
- if (component == null) {
- throw new BaseException(ErrorEnum.A_PARAMS_ERROR.getInfo());
- }
- if (HUMAN_FEEDBACK.getName().equals(component.getName())) {
- wfState.addInterruptNode(stateGraphNodeUuid);
- }
}
private void addEdgeToStateGraph(StateGraph stateGraph, String source, String target) throws GraphStateException {
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java
index 1a83e95c..4a0db255 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java
@@ -6,9 +6,9 @@ import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.ruoyi.common.chat.entity.User;
import org.ruoyi.common.chat.service.workFlow.IWorkFlowStarterService;
-import org.ruoyi.common.core.exception.base.BaseException;
import org.ruoyi.common.satoken.utils.LoginHelper;
import org.ruoyi.common.sse.core.SseEmitterManager;
+import org.ruoyi.common.tenant.helper.TenantHelper;
import org.ruoyi.workflow.entity.*;
import org.ruoyi.workflow.helper.SSEEmitterHelper;
import org.ruoyi.workflow.service.*;
@@ -58,6 +58,8 @@ public class WorkflowStarter implements IWorkFlowStarterService {
Long userId = LoginHelper.getUserId();
// 获取登录Token（仅透传给 WfState，工作流 SSE 通过 emitter 直发，不串台）
String tokenValue = StpUtil.getTokenValue();
+ // 获取当前租户ID（@Async 线程不继承请求线程的租户上下文，需显式透传）
+ String tenantId = TenantHelper.getTenantId();
// 根据会话ID连接SSE对象（每会话一个连接，避免同用户多会话串台）
SseEmitter sseEmitter = sseEmitterManager.connect(String.valueOf(sessionId));
if (!sseEmitterHelper.checkOrComplete(user, sseEmitter)) {
@@ -71,41 +73,33 @@ public class WorkflowStarter implements IWorkFlowStarterService {
sseEmitterHelper.sendErrorAndComplete(user.getId(), sseEmitter, A_WF_DISABLED.getInfo());
return sseEmitter;
}
- self.asyncRun(user, workflow, userInputs, sseEmitter, userId, tokenValue, sessionId);
+ self.asyncRun(user, workflow, userInputs, sseEmitter, userId, tokenValue, sessionId, tenantId);
return sseEmitter;
}
@Async
- public void asyncRun(User user, Workflow workflow, List userInputs, SseEmitter sseEmitter, Long userId, String tokenValue, Long sessionId) {
- log.info("WorkflowEngine run,userId:{},workflowUuid:{},userInputs:{}", user.getId(), workflow.getUuid(), userInputs);
- List components = workflowComponentService.getAllEnable();
- List nodes = workflowNodeService.lambdaQuery()
- .eq(WorkflowNode::getWorkflowId, workflow.getId())
- .eq(WorkflowNode::getIsDeleted, false)
- .list();
- List edges = workflowEdgeService.lambdaQuery()
- .eq(WorkflowEdge::getWorkflowId, workflow.getId())
- .eq(WorkflowEdge::getIsDeleted, false)
- .list();
- WorkflowEngine workflowEngine = new WorkflowEngine(workflow,
- sseEmitterHelper, components, nodes, edges,
- workflowRuntimeService, workflowRuntimeNodeService);
- workflowEngine.run(user, userInputs, sseEmitter, userId, tokenValue, sessionId);
- }
-
- @Async
- public void resumeFlow(String runtimeUuid, String userInput, SseEmitter sseEmitter) {
- WorkflowEngine workflowEngine = InterruptedFlow.RUNTIME_TO_GRAPH.get(runtimeUuid);
- if (null == workflowEngine) {
- log.error("工作流恢复执行时失败,runtime:{}", runtimeUuid);
- throw new BaseException(A_WF_RESUME_FAIL.getInfo());
+ public void asyncRun(User user, Workflow workflow, List userInputs, SseEmitter sseEmitter, Long userId, String tokenValue, Long sessionId, String tenantId) {
+ // @Async 线程不继承请求线程的租户上下文, 显式设置, 避免租户缓存/隔离逻辑异常
+ if (tenantId != null) {
+ TenantHelper.setDynamic(tenantId);
}
- // 如果SSE连接对象不为空传入该对象（Chat调用工作流对话使用）
- if (null != sseEmitter){
- workflowEngine.setSseEmitter(sseEmitter);
- // 为了让每个节点都可以发送模板消息 保持SSE对象一致（以防出现向已关闭的SSE对象发送消息）
- workflowEngine.getWfState().setSseEmitter(sseEmitter);
+ try {
+ log.info("WorkflowEngine run,userId:{},workflowUuid:{},userInputs:{}", user.getId(), workflow.getUuid(), userInputs);
+ List components = workflowComponentService.getAllEnable();
+ List nodes = workflowNodeService.lambdaQuery()
+ .eq(WorkflowNode::getWorkflowId, workflow.getId())
+ .eq(WorkflowNode::getIsDeleted, false)
+ .list();
+ List edges = workflowEdgeService.lambdaQuery()
+ .eq(WorkflowEdge::getWorkflowId, workflow.getId())
+ .eq(WorkflowEdge::getIsDeleted, false)
+ .list();
+ WorkflowEngine workflowEngine = new WorkflowEngine(workflow,
+ sseEmitterHelper, components, nodes, edges,
+ workflowRuntimeService, workflowRuntimeNodeService);
+ workflowEngine.run(user, userInputs, sseEmitter, userId, tokenValue, sessionId);
+ } finally {
+ TenantHelper.clearDynamic();
}
- workflowEngine.resume(userInput);
}
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowUtil.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowUtil.java
index f1a944be..0f55ed59 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowUtil.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowUtil.java
@@ -2,7 +2,6 @@ package org.ruoyi.workflow.workflow;
import cn.hutool.core.collection.CollStreamUtil;
import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.util.StrUtil;
import dev.langchain4j.data.message.ChatMessage;
import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
@@ -20,7 +19,6 @@ import org.ruoyi.common.chat.factory.ImageServiceFactory;
import org.ruoyi.workflow.base.NodeInputConfigTypeHandler;
import org.ruoyi.workflow.entity.WorkflowNode;
import org.ruoyi.workflow.enums.WfIODataTypeEnum;
-import org.ruoyi.workflow.util.JsonUtil;
import org.ruoyi.workflow.workflow.data.NodeIOData;
import org.ruoyi.workflow.workflow.data.NodeIODataContent;
import org.ruoyi.workflow.workflow.def.WfNodeParamRef;
@@ -88,22 +86,6 @@ public class WorkflowUtil{
return result;
}
- public static String getHumanFeedbackTip(String nodeUuid, List wfNodes) {
- WorkflowNode wfNode = wfNodes.stream()
- .filter(item -> item.getUuid().equals(nodeUuid))
- .findFirst().orElse(null);
- if (null == wfNode) {
- return "";
- }
- String wfNodeNodeConfig = wfNode.getNodeConfig();
- if (StrUtil.isBlank(wfNodeNodeConfig)) {
- return "";
- }
- Map map = JsonUtil.toMap(wfNodeNodeConfig);
- Object tip = map.getOrDefault("tip", "");
- return String.valueOf(tip);
- }
-
public void streamingInvokeLLM(WfState wfState, WfNodeState state, WorkflowNode node, String modelName,
String prompt, String nodeMessageTemplate) {
log.info("stream invoke, modelName: {}", modelName);
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/classifier/ClassifierNodeConfig.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/classifier/ClassifierNodeConfig.java
deleted file mode 100644
index a9432175..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/classifier/ClassifierNodeConfig.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.ruoyi.workflow.workflow.node.classifier;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import lombok.Data;
-
-import java.util.ArrayList;
-import java.util.List;
-
-@Data
-public class ClassifierNodeConfig {
- private List categories = new ArrayList<>();
- @JsonProperty("model_platform")
- private String modelPlatform;
- @JsonProperty("model_name")
- private String modelName;
-}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/enmus/NodeMessageTemplateEnum.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/enmus/NodeMessageTemplateEnum.java
index 1b39443f..3baa74f1 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/enmus/NodeMessageTemplateEnum.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/enmus/NodeMessageTemplateEnum.java
@@ -3,23 +3,44 @@ package org.ruoyi.workflow.workflow.node.enmus;
import lombok.Getter;
/**
- * 节点消息模板ConfigKey枚举
+ * 节点消息模板ConfigKey枚举
+ * 模板优先从 sys_config 读取(可在系统管理-配置管理中自定义), 未配置时回退到 defaultTemplate 内置默认值
*/
@Getter
public enum NodeMessageTemplateEnum {
- HTTP_REQUEST("node.httpRequest.template"),
- MAIL_SEND("node.mailsend.template"),
- IMAGE("node.image.template"),
- HUMAN_FEED_BACK("node.humanFeedback.template"),
- SWITCH("node.switch.template"),
- LLM_RESPONSE("node.llmAnswer.template"),
- KEYWORD_EXTRACTOR("node.keywordExtractor.template"),
- EXCEPTION("node.exception.template"),
- END("node.end.template");
+ HTTP_REQUEST("node.httpRequest.template", "✅ HTTP请求节点：结束响应 - "),
+ MAIL_SEND("node.mailsend.template", "📧 发送邮箱节点：结束响应 - "),
+ IMAGE("node.image.template", "🎨 文生图节点：结束响应 - 图片URL: "),
+ SWITCH("node.switch.template", "🔀 条件分支节点：触发 -> 跳转到节点 "),
+ LLM_RESPONSE("node.llmAnswer.template", "🤖 LLM 节点 生成回答："),
+ GOOGLE_SEARCH("node.googleSearch.template", "🔍 网络搜索节点处理完成："),
+ EXCEPTION("node.exception.template", "🛑 工作流发生异常："),
+ END("node.end.template", "🔚 流程已执行完毕，如果您有其他需求，请随时重新发起请求。");
private final String value;
- NodeMessageTemplateEnum(String value) {
+ /**
+ * 内置默认模板, sys_config 未配置对应键时使用
+ */
+ private final String defaultTemplate;
+
+ NodeMessageTemplateEnum(String value, String defaultTemplate) {
this.value = value;
+ this.defaultTemplate = defaultTemplate;
+ }
+
+ /**
+ * 根据 configKey 获取内置默认模板, 未知键返回空串
+ *
+ * @param configKey sys_config 配置键
+ * @return 内置默认模板
+ */
+ public static String getDefaultTemplate(String configKey) {
+ for (NodeMessageTemplateEnum item : values()) {
+ if (item.value.equals(configKey)) {
+ return item.defaultTemplate;
+ }
+ }
+ return "";
}
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNode.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNode.java
new file mode 100644
index 00000000..571c85be
--- /dev/null
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNode.java
@@ -0,0 +1,91 @@
+package org.ruoyi.workflow.workflow.node.googleSearch;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.ruoyi.workflow.entity.WorkflowComponent;
+import org.ruoyi.workflow.entity.WorkflowNode;
+import org.ruoyi.workflow.util.JsonUtil;
+import org.ruoyi.workflow.util.SpringUtil;
+import org.ruoyi.workflow.workflow.NodeProcessResult;
+import org.ruoyi.workflow.workflow.WfNodeState;
+import org.ruoyi.workflow.workflow.WfState;
+import org.ruoyi.workflow.workflow.WorkflowUtil;
+import org.ruoyi.workflow.workflow.data.NodeIOData;
+import org.ruoyi.workflow.workflow.node.AbstractWfNode;
+import org.ruoyi.workflow.workflow.node.enmus.NodeMessageTemplateEnum;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.ruoyi.workflow.cosntant.AdiConstant.WorkflowConstant.DEFAULT_OUTPUT_PARAM_NAME;
+
+/**
+ * 【扩展节点】网络搜索
+ * 通过智谱 Web Search API 返回适合大模型消费的结构化网页结果。
+ */
+@Slf4j
+public class GoogleSearchNode extends AbstractWfNode {
+
+ public GoogleSearchNode(WorkflowComponent wfComponent, WorkflowNode nodeDef, WfState wfState, WfNodeState nodeState) {
+ super(wfComponent, nodeDef, wfState, nodeState);
+ }
+
+ /**
+ * 处理搜索请求
+ * nodeConfig 格式：
+ * {
+ * "query": "搜索关键词",
+ * "search_engine": "search_std",
+ * "result_count": 10,
+ * "search_domain_filter": "",
+ * "search_recency_filter": "noLimit",
+ * "content_size": "medium",
+ * "include_image": false
+ * }
+ *
+ * @return 搜索结果
+ */
+ @Override
+ public NodeProcessResult onProcess() {
+ GoogleSearchNodeConfig config = checkAndGetConfig(GoogleSearchNodeConfig.class);
+
+ // 获取搜索关键词
+ String searchQuery = WorkflowUtil.renderTemplate(config.getQuery(), state.getInputs());
+ if (StringUtils.isBlank(searchQuery)) {
+ searchQuery = getFirstInputText();
+ }
+
+ if (StringUtils.isBlank(searchQuery)) {
+ throw new IllegalArgumentException("未提供搜索关键词");
+ }
+ searchQuery = searchQuery.trim();
+ if (searchQuery.length() > 70) {
+ throw new IllegalArgumentException("搜索关键词不能超过 70 个字符");
+ }
+
+ log.info("Web search node processing, engine: {}, result_count: {}",
+ config.getSearchEngine(), config.getResultCount());
+
+ String nodeMessageTemplate = getNodeMessageTemplate(NodeMessageTemplateEnum.GOOGLE_SEARCH.getValue());
+ notifyAndStoreMessage(wfState, nodeMessageTemplate);
+
+ ZhipuWebSearchClient searchClient = SpringUtil.getBean(ZhipuWebSearchClient.class);
+ ZhipuWebSearchClient.SearchResponse response = searchClient.search(
+ searchQuery,
+ config,
+ UUID.randomUUID().toString()
+ );
+ String searchResult = JsonUtil.toJson(response);
+ if (searchResult == null) {
+ throw new IllegalStateException("搜索结果序列化失败");
+ }
+
+ log.info("Web search completed, result count: {}", response.count());
+ notifyAndStoreMessage(wfState, nodeMessageTemplate + "返回 " + response.count() + " 条结果");
+
+ List outputs = List.of(
+ NodeIOData.createByText(DEFAULT_OUTPUT_PARAM_NAME, "智谱网络搜索结果", searchResult)
+ );
+ return NodeProcessResult.builder().content(outputs).build();
+ }
+}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNodeConfig.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNodeConfig.java
new file mode 100644
index 00000000..26058042
--- /dev/null
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/GoogleSearchNodeConfig.java
@@ -0,0 +1,44 @@
+package org.ruoyi.workflow.workflow.node.googleSearch;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.Max;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.Pattern;
+import lombok.Data;
+
+@Data
+public class GoogleSearchNodeConfig {
+ /**
+ * 搜索查询关键词
+ */
+ private String query;
+
+ @JsonProperty("search_engine")
+ @Pattern(
+ regexp = "search_std|search_pro|search_pro_sogou|search_pro_quark",
+ message = "搜索引擎参数无效"
+ )
+ private String searchEngine = "search_std";
+
+ @JsonProperty("result_count")
+ @Min(value = 1, message = "搜索结果数量不能小于 1")
+ @Max(value = 50, message = "搜索结果数量不能大于 50")
+ private Integer resultCount = 10;
+
+ @JsonProperty("search_domain_filter")
+ private String searchDomainFilter;
+
+ @JsonProperty("search_recency_filter")
+ @Pattern(
+ regexp = "oneDay|oneWeek|oneMonth|oneYear|noLimit",
+ message = "搜索时间范围参数无效"
+ )
+ private String searchRecencyFilter = "noLimit";
+
+ @JsonProperty("content_size")
+ @Pattern(regexp = "medium|high", message = "网页摘要长度参数无效")
+ private String contentSize = "medium";
+
+ @JsonProperty("include_image")
+ private Boolean includeImage = false;
+}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchClient.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchClient.java
new file mode 100644
index 00000000..a429cace
--- /dev/null
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchClient.java
@@ -0,0 +1,155 @@
+package org.ruoyi.workflow.workflow.node.googleSearch;
+
+import ai.z.openapi.ZhipuAiClient;
+import ai.z.openapi.service.web_search.WebSearchRequest;
+import ai.z.openapi.service.web_search.WebSearchResp;
+import ai.z.openapi.service.web_search.WebSearchResponse;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.ruoyi.common.chat.domain.bo.chat.ChatModelBo;
+import org.ruoyi.common.chat.service.chat.IChatModelService;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 智谱 Web Search 官方 SDK 适配器。
+ */
+@Component
+@RequiredArgsConstructor
+public class ZhipuWebSearchClient {
+
+ private static final String ZHIPU_PROVIDER_CODE = "zhipu";
+ private static final String DEFAULT_BASE_URL = "https://open.bigmodel.cn/api/paas/v4/";
+
+ private final ZhipuWebSearchProperties properties;
+ private final IChatModelService chatModelService;
+
+ public SearchResponse search(String query, GoogleSearchNodeConfig config, String requestId) {
+ Credential credential = resolveCredential();
+ ZhipuAiClient client = createClient(credential);
+ try {
+ WebSearchRequest request = WebSearchRequest.builder()
+ .searchQuery(query)
+ .searchEngine(config.getSearchEngine())
+ .count(config.getResultCount())
+ .searchDomainFilter(blankToNull(config.getSearchDomainFilter()))
+ .searchRecencyFilter(config.getSearchRecencyFilter())
+ .contentSize(config.getContentSize())
+ .includeImage(config.getIncludeImage())
+ .requestId(requestId)
+ .build();
+
+ WebSearchResponse response = client.webSearch().createWebSearch(request);
+ if (response == null || !response.isSuccess() || response.getData() == null) {
+ String message = response == null ? "接口未返回响应" : StringUtils.defaultIfBlank(response.getMsg(), "未知错误");
+ throw new IllegalStateException("智谱 Web Search 调用失败：" + message);
+ }
+
+ List results = response.getData().getWebSearchResp() == null
+ ? List.of()
+ : response.getData().getWebSearchResp().stream()
+ .map(this::toSearchResult)
+ .toList();
+
+ return new SearchResponse(
+ query,
+ config.getSearchEngine(),
+ response.getData().getRequestId(),
+ results.size(),
+ results
+ );
+ } finally {
+ client.close();
+ }
+ }
+
+ private ZhipuAiClient createClient(Credential credential) {
+ int connectTimeout = positiveOrDefault(properties.getConnectTimeout(), 10);
+ int readTimeout = positiveOrDefault(properties.getReadTimeout(), 30);
+ return ZhipuAiClient.builder()
+ .ofZHIPU()
+ .apiKey(credential.apiKey())
+ .baseUrl(credential.baseUrl())
+ .networkConfig(connectTimeout, readTimeout, readTimeout, readTimeout, TimeUnit.SECONDS)
+ .enableTokenCache()
+ .build();
+ }
+
+ private Credential resolveCredential() {
+ if (isUsableApiKey(properties.getApiKey())) {
+ return new Credential(normalizeBaseUrl(properties.getBaseUrl()), properties.getApiKey().trim());
+ }
+
+ ChatModelBo query = new ChatModelBo();
+ query.setProviderCode(ZHIPU_PROVIDER_CODE);
+ return chatModelService.queryList(query).stream()
+ .filter(model -> isUsableApiKey(model.getApiKey()))
+ .findFirst()
+ .map(model -> new Credential(normalizeBaseUrl(model.getApiHost()), model.getApiKey().trim()))
+ .orElseThrow(() -> new IllegalStateException(
+ "未配置智谱 API Key，请设置环境变量 ZAI_API_KEY，或在模型管理中配置 zhipu 厂商密钥"
+ ));
+ }
+
+ private boolean isUsableApiKey(String apiKey) {
+ return StringUtils.isNotBlank(apiKey)
+ && !"sk_xx".equalsIgnoreCase(apiKey.trim())
+ && !"your_api_key".equalsIgnoreCase(apiKey.trim());
+ }
+
+ private String normalizeBaseUrl(String baseUrl) {
+ String normalized = StringUtils.defaultIfBlank(baseUrl, DEFAULT_BASE_URL).trim();
+ normalized = StringUtils.removeEnd(normalized, "/");
+ if (!normalized.endsWith("/api/paas/v4")) {
+ normalized += "/api/paas/v4";
+ }
+ return normalized + "/";
+ }
+
+ private int positiveOrDefault(Integer value, int defaultValue) {
+ return value != null && value > 0 ? value : defaultValue;
+ }
+
+ private String blankToNull(String value) {
+ return StringUtils.isBlank(value) ? null : value.trim();
+ }
+
+ private SearchResult toSearchResult(WebSearchResp result) {
+ return new SearchResult(
+ result.getTitle(),
+ result.getContent(),
+ result.getLink(),
+ result.getMedia(),
+ result.getIcon(),
+ result.getRefer(),
+ result.getPublishDate(),
+ result.getImages()
+ );
+ }
+
+ private record Credential(String baseUrl, String apiKey) {
+ }
+
+ public record SearchResponse(
+ String query,
+ String searchEngine,
+ String requestId,
+ int count,
+ List results
+ ) {
+ }
+
+ public record SearchResult(
+ String title,
+ String content,
+ String link,
+ String media,
+ String icon,
+ String refer,
+ String publishDate,
+ List images
+ ) {
+ }
+}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchProperties.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchProperties.java
new file mode 100644
index 00000000..466acd10
--- /dev/null
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/googleSearch/ZhipuWebSearchProperties.java
@@ -0,0 +1,34 @@
+package org.ruoyi.workflow.workflow.node.googleSearch;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+/**
+ * 智谱 Web Search 配置。
+ */
+@Data
+@Component
+@ConfigurationProperties(prefix = "workflow.web-search.zhipu")
+public class ZhipuWebSearchProperties {
+
+ /**
+ * 智谱国内开放平台 API 根地址。
+ */
+ private String baseUrl = "https://open.bigmodel.cn/api/paas/v4/";
+
+ /**
+ * 智谱 API Key。建议通过环境变量 ZAI_API_KEY 注入。
+ */
+ private String apiKey;
+
+ /**
+ * 连接超时秒数。
+ */
+ private Integer connectTimeout = 10;
+
+ /**
+ * 读取超时秒数。
+ */
+ private Integer readTimeout = 30;
+}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/humanFeedBack/HumanFeedbackNode.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/humanFeedBack/HumanFeedbackNode.java
deleted file mode 100644
index bf14b716..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/humanFeedBack/HumanFeedbackNode.java
+++ /dev/null
@@ -1,56 +0,0 @@
-package org.ruoyi.workflow.workflow.node.humanFeedBack;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.ruoyi.workflow.entity.WorkflowComponent;
-import org.ruoyi.workflow.entity.WorkflowNode;
-import org.ruoyi.workflow.workflow.NodeProcessResult;
-import org.ruoyi.workflow.workflow.WfNodeState;
-import org.ruoyi.workflow.workflow.WfState;
-import org.ruoyi.workflow.workflow.WorkflowUtil;
-import org.ruoyi.workflow.workflow.data.NodeIOData;
-import org.ruoyi.workflow.workflow.node.AbstractWfNode;
-
-import static org.ruoyi.workflow.cosntant.AdiConstant.WorkflowConstant.*;
-
-/**
- * 人机交互节点实现类
- */
-@Slf4j
-public class HumanFeedbackNode extends AbstractWfNode {
-
- public HumanFeedbackNode(WorkflowComponent component, WorkflowNode nodeDefinition, WfState wfState, WfNodeState nodeState) {
- super(component, nodeDefinition, wfState, nodeState);
- }
-
- // 人机交互节点的处理逻辑
- @Override
- public NodeProcessResult onProcess() {
- log.info("Processing HumanFeedback node: {}", node.getTitle());
- // 从状态中获取用户输入数据
- Object humanFeedbackState = state.data().get(HUMAN_FEEDBACK_KEY);
- if (null != humanFeedbackState) {
- String userInput = humanFeedbackState.toString();
- if (StringUtils.isNotBlank(userInput)) {
- // 用户已提供输入，将用户输入添加到节点输入和输出中
- NodeIOData feedbackData = NodeIOData.createByText("output", "default", userInput);
- // 添加到输出列表，这样后续节点可以使用
- state.getOutputs().add(feedbackData);
- // 设置为成功状态
- state.setProcessStatus(NODE_PROCESS_STATUS_SUCCESS);
- log.info("Human feedback processed for node: {}, content: {}", node.getTitle(), userInput);
- } else {
- // 用户输入为空，设置等待状态
- state.setProcessStatus(NODE_PROCESS_STATUS_DOING);
- log.info("Human feedback is empty for node: {}", node.getTitle());
- }
- } else {
- // 没有用户输入，这可能是正常情况（等待用户输入）
- // 但为了确保流程可以继续，我们仍然标记为成功
- state.setProcessStatus(NODE_PROCESS_STATUS_SUCCESS);
- log.info("No human feedback found for node: {}, continuing workflow", node.getTitle());
- }
- return new NodeProcessResult();
- }
-}
-
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNode.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNode.java
deleted file mode 100644
index 79585018..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNode.java
+++ /dev/null
@@ -1,106 +0,0 @@
-package org.ruoyi.workflow.workflow.node.keywordExtractor;
-
-import dev.langchain4j.data.message.SystemMessage;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.ruoyi.workflow.entity.WorkflowComponent;
-import org.ruoyi.workflow.entity.WorkflowNode;
-import org.ruoyi.workflow.util.SpringUtil;
-import org.ruoyi.workflow.util.WorkflowMessageUtil;
-import org.ruoyi.workflow.workflow.NodeProcessResult;
-import org.ruoyi.workflow.workflow.WfNodeState;
-import org.ruoyi.workflow.workflow.WfState;
-import org.ruoyi.workflow.workflow.WorkflowUtil;
-import org.ruoyi.workflow.workflow.data.NodeIOData;
-import org.ruoyi.workflow.workflow.node.AbstractWfNode;
-import org.ruoyi.workflow.workflow.node.enmus.NodeMessageTemplateEnum;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.ruoyi.workflow.cosntant.AdiConstant.WorkflowConstant.DEFAULT_OUTPUT_PARAM_NAME;
-
-/**
- * 【节点】关键词提取节点
- * 使用 LLM 从文本中提取关键词
- */
-@Slf4j
-public class KeywordExtractorNode extends AbstractWfNode {
-
- public KeywordExtractorNode(WorkflowComponent wfComponent, WorkflowNode nodeDef, WfState wfState, WfNodeState nodeState) {
- super(wfComponent, nodeDef, wfState, nodeState);
- }
-
- /**
- * 处理关键词提取
- * nodeConfig 格式：
- * {
- * "model_name": "deepseek-chat",
- * "category": "llm",
- * "top_n": 5,
- * "prompt": "额外的提示词"
- * }
- *
- * @return 提取的关键词列表
- */
- @Override
- public NodeProcessResult onProcess() {
- KeywordExtractorNodeConfig config = checkAndGetConfig(KeywordExtractorNodeConfig.class);
-
- // 获取输入文本
- String inputText = getFirstInputText();
- if (StringUtils.isBlank(inputText)) {
- log.warn("Keyword extractor node has no input text, node: {}", state.getUuid());
- // 返回空结果
- List outputs = new ArrayList<>();
- outputs.add(NodeIOData.createByText(DEFAULT_OUTPUT_PARAM_NAME, "", ""));
- return NodeProcessResult.builder().content(outputs).build();
- }
-
- log.info("Keyword extractor node config: {}", config);
- log.info("Input text length: {}", inputText.length());
-
- // 构建提示词
- String prompt = buildPrompt(config, inputText);
- log.info("Keyword extraction prompt: {}", prompt);
-
- // 调用 LLM 进行关键词提取
- WorkflowUtil workflowUtil = SpringUtil.getBean(WorkflowUtil.class);
- String modelName = config.getModelName();
- // 获取节点模板提示词信息
- String nodeMessageTemplate = WorkflowMessageUtil.getNodeMessageTemplate(NodeMessageTemplateEnum.KEYWORD_EXTRACTOR.getValue());
- // 发送SSE事件消息
- WorkflowMessageUtil.sendEmitterMessage(wfState.getSseEmitter(), node, nodeMessageTemplate);
- // 使用流式调用
- workflowUtil.streamingInvokeLLM(wfState, state, node, modelName, prompt, nodeMessageTemplate);
- return new NodeProcessResult();
- }
-
- /**
- * 构建关键词提取的提示词
- */
- private String buildPrompt(KeywordExtractorNodeConfig config, String inputText) {
- StringBuilder promptBuilder = new StringBuilder();
-
- // 基础提示词
- promptBuilder.append("请从以下文本中提取 ").append(config.getTopN()).append(" 个最重要的关键词。\n\n");
-
- // 添加自定义提示词（如果有）
- if (StringUtils.isNotBlank(config.getPrompt())) {
- promptBuilder.append(config.getPrompt()).append("\n\n");
- }
-
- // 输出格式要求
- promptBuilder.append("要求：\n");
- promptBuilder.append("1. 只返回关键词，每个关键词用逗号分隔\n");
- promptBuilder.append("2. 关键词应该是名词或名词短语\n");
- promptBuilder.append("3. 按重要性从高到低排序\n");
- promptBuilder.append("4. 不要添加任何解释或额外的文字\n\n");
-
- // 原始文本
- promptBuilder.append("文本内容：\n");
- promptBuilder.append(inputText);
-
- return promptBuilder.toString();
- }
-}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNodeConfig.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNodeConfig.java
deleted file mode 100644
index 4e9cc6eb..00000000
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/keywordExtractor/KeywordExtractorNodeConfig.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package org.ruoyi.workflow.workflow.node.keywordExtractor;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.constraints.Max;
-import jakarta.validation.constraints.Min;
-import jakarta.validation.constraints.NotNull;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-/**
- * 关键词提取节点配置
- */
-@EqualsAndHashCode
-@Data
-public class KeywordExtractorNodeConfig {
-
- /**
- * 模型分类（如：llm, embedding 等）
- */
- private String category;
-
- /**
- * 模型名称
- */
- @NotNull
- @JsonProperty("model_name")
- private String modelName;
-
- /**
- * 提取的关键词数量
- */
- @Min(1)
- @Max(50)
- @JsonProperty("top_n")
- private Integer topN = 5;
-
- /**
- * 提示词（可选）
- * 用于指导关键词提取的额外说明
- */
- private String prompt;
-}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/mailSend/MailSendNode.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/mailSend/MailSendNode.java
index c8399aa8..2619b612 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/mailSend/MailSendNode.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/mailSend/MailSendNode.java
@@ -7,6 +7,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.ruoyi.workflow.entity.WorkflowComponent;
import org.ruoyi.workflow.entity.WorkflowNode;
+import org.ruoyi.workflow.util.JsonUtil;
import org.ruoyi.workflow.workflow.NodeProcessResult;
import org.ruoyi.workflow.workflow.WfNodeState;
import org.ruoyi.workflow.workflow.WfState;
@@ -40,15 +41,25 @@ public class MailSendNode extends AbstractWfNode {
String input = getDataFromInput(inputs);
// 判断是否为JSON格式(LLM输出转换 由LLM生成格式)
if (StringUtils.isNotBlank(input) && isJson(input)) {
- // 使用Jackson解析和合并配置
- ObjectMapper objectMapper = new ObjectMapper();
- JsonNode inputJson = objectMapper.readTree(input);
- // 将config转换为JsonNode
- JsonNode configJson = objectMapper.valueToTree(config);
- // 合并两个JSON节点
- JsonNode mergedJson = objectMapper.readerForUpdating(configJson).readValue(inputJson);
- // 转换回config对象
- config = objectMapper.treeToValue(mergedJson, MailSendNodeConfig.class);
+ try {
+ // 使用统一的 JsonUtil 进行解析
+ JsonNode inputJson = JsonUtil.toJsonNode(input);
+ if (inputJson != null) {
+ // 使用 JsonUtil 内部的 ObjectMapper 进行合并
+ ObjectMapper objectMapper = new ObjectMapper();
+ // 将config转换为JsonNode
+ JsonNode configJson = objectMapper.valueToTree(config);
+ // 合并两个JSON节点
+ JsonNode mergedJson = objectMapper.readerForUpdating(configJson).readValue(inputJson);
+ // 转换回config对象
+ config = objectMapper.treeToValue(mergedJson, MailSendNodeConfig.class);
+ } else {
+ log.warn("输入 JSON 解析结果为 null，使用原始配置");
+ }
+ } catch (Exception e) {
+ log.error("合并邮件配置失败，使用原始配置: {}", e.getMessage(), e);
+ // 继续使用原始 config，不中断流程
+ }
}
// 安全获取模板（使用 defaultString 避免 null）
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/switcher/SwitcherNode.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/switcher/SwitcherNode.java
index 4aece033..51a98066 100644
--- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/switcher/SwitcherNode.java
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/node/switcher/SwitcherNode.java
@@ -1,7 +1,6 @@
package org.ruoyi.workflow.workflow.node.switcher;
import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
@@ -9,6 +8,7 @@ import org.ruoyi.common.core.utils.SpringUtils;
import org.ruoyi.workflow.entity.WorkflowComponent;
import org.ruoyi.workflow.entity.WorkflowNode;
import org.ruoyi.workflow.service.WorkflowNodeService;
+import org.ruoyi.workflow.util.JsonUtil;
import org.ruoyi.workflow.workflow.NodeProcessResult;
import org.ruoyi.workflow.workflow.WfNodeState;
import org.ruoyi.workflow.workflow.WfState;
@@ -18,8 +18,6 @@ import org.ruoyi.workflow.workflow.node.enmus.NodeMessageTemplateEnum;
import java.math.BigDecimal;
import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
/**
* 条件分支节点
@@ -339,9 +337,12 @@ public class SwitcherNode extends AbstractWfNode {
log.info("节点 '{}' 的输入配置: {}", nodeUuid, inputConfig);
if (StringUtils.isNotBlank(inputConfig)){
try {
- // 使用Jackson解析输入配置
- ObjectMapper objectMapper = new ObjectMapper();
- JsonNode configJson = objectMapper.readTree(inputConfig);
+ // 使用统一的 JsonUtil 而不是每次创建新的 ObjectMapper
+ JsonNode configJson = JsonUtil.toJsonNode(inputConfig);
+ if (configJson == null) {
+ log.warn("节点 '{}' 的输入配置 JSON 解析结果为 null", nodeUuid);
+ return result;
+ }
// 获取 user_inputs 数组
JsonNode userInputs = configJson.get("user_inputs");
if (userInputs != null && userInputs.isArray()) {
@@ -357,7 +358,8 @@ public class SwitcherNode extends AbstractWfNode {
}
}
} catch (Exception e) {
- log.error("解析节点输入配置失败: {}", nodeUuid, e);
+ log.error("解析节点 '{}' 输入配置失败，参数名: {}, 配置内容: {}", nodeUuid, paramName, inputConfig, e);
+ // 不抛出异常，返回默认结果，避免中断整个流程
}
}
}
diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/resources/sql/workflow_nodes_config.sql b/ruoyi-modules/ruoyi-aiflow/src/main/resources/sql/workflow_nodes_config.sql
new file mode 100644
index 00000000..d03d4d69
--- /dev/null
+++ b/ruoyi-modules/ruoyi-aiflow/src/main/resources/sql/workflow_nodes_config.sql
@@ -0,0 +1,21 @@
+-- =============================================
+-- 流程编排搜索节点配置脚本
+-- =============================================
+-- 说明：本脚本用于添加搜索节点的系统配置
+-- 执行前请确保 sys_config 表存在
+-- =============================================
+
+-- 搜索节点模板配置
+INSERT INTO sys_config (config_name, config_key, config_value, config_type, remark, create_by, create_time, update_by, update_time)
+VALUES ('搜索节点模板', 'node.googleSearch.template', '正在搜索相关内容...', 'Y', '搜索节点的响应模板，用于网络搜索功能', 'admin', NOW(), 'admin', NOW())
+ON DUPLICATE KEY UPDATE config_value = '正在搜索相关内容...', update_time = NOW();
+
+-- =============================================
+-- 验证配置是否添加成功
+-- =============================================
+SELECT config_id, config_name, config_key, config_value, config_type, remark
+FROM sys_config
+WHERE config_key IN (
+ 'node.googleSearch.template'
+)
+ORDER BY config_id;
diff --git a/ruoyi-modules/ruoyi-aiflow/流程编排模块说明.md b/ruoyi-modules/ruoyi-aiflow/流程编排模块说明.md
deleted file mode 100644
index d65a33ba..00000000
--- a/ruoyi-modules/ruoyi-aiflow/流程编排模块说明.md
+++ /dev/null
@@ -1,425 +0,0 @@
-# Ruoyi-AI 流程编排模块详细说明文档
-
-## 概述
-
-Ruoyi-AI 工作流模块是一个基于 LangGraph4j 的智能工作流引擎，支持可视化工作流设计、AI 模型集成、条件分支、人机交互等高级功能。该模块采用微服务架构，提供完整的
-RESTful API 和流式响应支持。
-
-## 模块架构
-
-### 1. 核心依赖
-
-- **LangGraph4j**: 1.5.3 - 工作流图执行引擎
-- **LangChain4j**: 1.11.0 - AI 模型集成框架
-- **Spring Boot**: 3.5.8 - 应用框架
-- **MyBatis Plus**: 数据访问层
-- **Redis**: 缓存和状态管理
-- **OpenAPI**: API 文档
-
-## 核心功能
-
-### 1. 工作流管理
-
-#### 1.1 工作流定义
-
-- **创建工作流**: 支持自定义标题、描述、公开性设置
-- **编辑工作流**: 可视化节点编辑、连接线配置
-- **版本控制**: 支持工作流的版本管理和回滚
-- **权限管理**: 支持公开/私有工作流设置
-
-#### 1.2 工作流执行
-
-- **流式执行**: 基于 SSE 的实时流式响应
-- **状态管理**: 完整的执行状态跟踪
-- **错误处理**: 详细的错误信息和异常处理
-- **中断恢复**: 支持工作流中断和恢复执行
-
-### 2. 节点类型
-
-#### 2.1 基础节点
-
-- **Start**: 开始节点，定义工作流入口
-- **End**: 结束节点，定义工作流出口
-
-
-#### 2.2 AI 模型节点
-
-- **Answer**: 大语言模型问答节点
-- **Dalle3**: DALL-E 3 图像生成
-- **Tongyiwanx**: 通义万相图像生成
-- **Classifier**: 内容分类节点
-
-#### 2.3 数据处理节点
-
-- **DocumentExtractor**: 文档信息提取
-- **KeywordExtractor**: 关键词提取
-- **FaqExtractor**: 常见问题提取
-- **KnowledgeRetrieval**: 知识库检索
-
-#### 2.4 控制流节点
-
-- **Switcher**: 条件分支节点
-- **HumanFeedback**: 人机交互节点
-
-#### 2.5 外部集成节点
-
-- **Google**: Google 搜索集成
-- **MailSend**: 邮件发送
-- **HttpRequest**: HTTP 请求
-- **Template**: 模板转换
-
-### 3. 数据流管理
-
-#### 3.1 输入输出定义
-
-```java
-// 节点输入输出数据结构
-public class NodeIOData {
- private String name; // 参数名称
- private NodeIODataContent content; // 参数内容
-}
-
-// 支持的数据类型
-public enum WfIODataTypeEnum {
- TEXT, // 文本
- NUMBER, // 数字
- BOOLEAN, // 布尔值
- FILES, // 文件
- OPTIONS // 选项
-}
-```
-
-#### 3.2 参数引用
-
-- **节点间引用**: 支持上游节点输出作为下游节点输入
-- **参数映射**: 自动处理参数名称映射
-- **类型转换**: 自动进行数据类型转换
-
-## 数据库设计
-
-### 1. 核心表结构
-
-#### 1.1 工作流定义表 (t_workflow)
-
-```sql
-CREATE TABLE t_workflow (
- id BIGINT AUTO_INCREMENT PRIMARY KEY,
- uuid VARCHAR(32) NOT NULL DEFAULT '',
- title VARCHAR(100) NOT NULL DEFAULT '',
- remark TEXT NOT NULL DEFAULT '',
- user_id BIGINT NOT NULL DEFAULT 0,
- is_public TINYINT(1) NOT NULL DEFAULT 0,
- is_enable TINYINT(1) NOT NULL DEFAULT 1,
- create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
- update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- is_deleted TINYINT(1) NOT NULL DEFAULT 0
-);
-```
-
-#### 1.2 工作流节点表 (t_workflow_node)
-
-```sql
-CREATE TABLE t_workflow_node (
- id BIGINT AUTO_INCREMENT PRIMARY KEY,
- uuid VARCHAR(32) NOT NULL DEFAULT '',
- workflow_id BIGINT NOT NULL DEFAULT 0,
- workflow_component_id BIGINT NOT NULL DEFAULT 0,
- user_id BIGINT NOT NULL DEFAULT 0,
- title VARCHAR(100) NOT NULL DEFAULT '',
- remark VARCHAR(500) NOT NULL DEFAULT '',
- input_config JSON NOT NULL DEFAULT ('{}'),
- node_config JSON NOT NULL DEFAULT ('{}'),
- position_x DOUBLE NOT NULL DEFAULT 0,
- position_y DOUBLE NOT NULL DEFAULT 0,
- create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
- update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- is_deleted TINYINT(1) NOT NULL DEFAULT 0
-);
-```
-
-#### 1.3 工作流边表 (t_workflow_edge)
-
-```sql
-CREATE TABLE t_workflow_edge (
- id BIGINT AUTO_INCREMENT PRIMARY KEY,
- uuid VARCHAR(32) NOT NULL DEFAULT '',
- workflow_id BIGINT NOT NULL DEFAULT 0,
- source_node_uuid VARCHAR(32) NOT NULL DEFAULT '',
- source_handle VARCHAR(32) NOT NULL DEFAULT '',
- target_node_uuid VARCHAR(32) NOT NULL DEFAULT '',
- create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
- update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- is_deleted TINYINT(1) NOT NULL DEFAULT 0
-);
-```
-
-#### 1.4 工作流运行时表 (t_workflow_runtime)
-
-```sql
-CREATE TABLE t_workflow_runtime (
- id BIGINT AUTO_INCREMENT PRIMARY KEY,
- uuid VARCHAR(32) NOT NULL DEFAULT '',
- user_id BIGINT NOT NULL DEFAULT 0,
- workflow_id BIGINT NOT NULL DEFAULT 0,
- input JSON NOT NULL DEFAULT ('{}'),
- output JSON NOT NULL DEFAULT ('{}'),
- status SMALLINT NOT NULL DEFAULT 1,
- status_remark VARCHAR(250) NOT NULL DEFAULT '',
- create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
- update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- is_deleted TINYINT(1) NOT NULL DEFAULT 0
-);
-```
-
-#### 1.5 工作流组件表 (t_workflow_component)
-
-```sql
-CREATE TABLE t_workflow_component (
- id BIGINT AUTO_INCREMENT PRIMARY KEY,
- uuid VARCHAR(32) DEFAULT '' NOT NULL,
- name VARCHAR(32) DEFAULT '' NOT NULL,
- title VARCHAR(100) DEFAULT '' NOT NULL,
- remark TEXT NOT NULL,
- display_order INT DEFAULT 0 NOT NULL,
- is_enable TINYINT(1) DEFAULT 0 NOT NULL,
- create_time DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
- update_time DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
- is_deleted TINYINT(1) DEFAULT 0 NOT NULL
-);
-```
-
-## API 接口
-
-### 1. 工作流管理接口
-
-#### 1.1 基础操作
-
-```http
-# 创建工作流
-POST /workflow/add
-Content-Type: application/json
-{
- "title": "工作流标题",
- "remark": "工作流描述",
- "isPublic": false
-}
-
-# 更新工作流
-POST /workflow/update
-Content-Type: application/json
-{
- "uuid": "工作流UUID",
- "title": "新标题",
- "remark": "新描述"
-}
-
-# 删除工作流
-POST /workflow/del/{uuid}
-
-# 启用/禁用工作流
-POST /workflow/enable/{uuid}?enable=true
-```
-
-#### 1.2 搜索和查询
-
-```http
-# 搜索我的工作流
-GET /workflow/mine/search?keyword=关键词&isPublic=true¤tPage=1&pageSize=10
-
-# 搜索公开工作流
-GET /workflow/public/search?keyword=关键词¤tPage=1&pageSize=10
-
-# 获取工作流组件列表
-GET /workflow/public/component/list
-```
-
-### 2. 工作流执行接口
-
-#### 2.1 流式执行
-
-```http
-# 流式执行工作流
-POST /workflow/run
-Content-Type: application/json
-Accept: text/event-stream
-{
- "uuid": "工作流UUID",
- "inputs": [
- {
- "name": "input",
- "content": {
- "type": 1,
- "textContent": "用户输入内容"
- }
- }
- ]
-}
-```
-
-#### 2.2 运行时管理
-
-```http
-# 恢复中断的工作流
-POST /workflow/runtime/resume/{runtimeUuid}
-Content-Type: application/json
-{
- "feedbackContent": "用户反馈内容"
-}
-
-# 查询工作流执行历史
-GET /workflow/runtime/page?wfUuid=工作流UUID¤tPage=1&pageSize=10
-
-# 查询运行时节点详情
-GET /workflow/runtime/nodes/{runtimeUuid}
-
-# 清理运行时数据
-POST /workflow/runtime/clear?wfUuid=工作流UUID
-```
-
-### 3. 管理端接口
-
-#### 3.1 工作流管理
-
-```http
-# 搜索所有工作流
-POST /admin/workflow/search
-Content-Type: application/json
-{
- "title": "搜索关键词",
- "isPublic": true,
- "isEnable": true
-}
-
-# 启用/禁用工作流
-POST /admin/workflow/enable?uuid=工作流UUID&isEnable=true
-```
-
-## 核心实现
-
-### 1. 工作流引擎 (WorkflowEngine)
-
-工作流引擎是整个模块的核心，负责：
-
-- 工作流图的构建和编译
-- 节点执行调度
-- 状态管理和持久化
-- 流式输出处理
-
-```java
-public class WorkflowEngine {
- // 核心执行方法
- public void run(User user, List userInputs, SseEmitter sseEmitter) {
- // 1. 验证工作流状态
- // 2. 创建运行时实例
- // 3. 构建状态图
- // 4. 执行工作流
- // 5. 处理流式输出
- }
-
- // 恢复执行方法
- public void resume(String userInput) {
- // 1. 更新状态
- // 2. 继续执行
- }
-}
-```
-
-### 2. 节点工厂 (WfNodeFactory)
-
-节点工厂负责根据组件类型创建对应的节点实例：
-
-```java
-public class WfNodeFactory {
- public static AbstractWfNode create(WorkflowComponent component,
- WorkflowNode node,
- WfState wfState,
- WfNodeState nodeState) {
- // 根据组件类型创建对应的节点实例
- switch (component.getName()) {
- case "Answer":
- return new LLMAnswerNode(component, node, wfState, nodeState);
- case "Switcher":
- return new SwitcherNode(component, node, wfState, nodeState);
- // ... 其他节点类型
- }
- }
-}
-```
-
-### 3. 图构建器 (WorkflowGraphBuilder)
-
-图构建器负责将工作流定义转换为可执行的状态图：
-
-```java
-public class WorkflowGraphBuilder {
- public StateGraph build(WorkflowNode startNode) {
- // 1. 构建编译节点树
- // 2. 转换为状态图
- // 3. 添加节点和边
- // 4. 处理条件分支
- // 5. 处理并行执行
- }
-}
-```
-
-## 流式响应机制
-
-### 1. SSE 事件类型
-
-工作流执行过程中会发送多种类型的 SSE 事件：
-
-```javascript
-// 节点开始执行
-[NODE_RUN_节点UUID] - 节点执行开始事件
-
-// 节点输入数据
-[NODE_INPUT_节点UUID] - 节点输入数据事件
-
-// 节点输出数据
-[NODE_OUTPUT_节点UUID] - 节点输出数据事件
-
-// 流式内容块
-[NODE_CHUNK_节点UUID] - 流式内容块事件
-
-// 等待用户输入
-[NODE_WAIT_FEEDBACK_BY_节点UUID] - 等待用户输入事件
-```
-
-### 2. 流式处理流程
-
-1. **初始化**: 创建工作流运行时实例
-2. **节点执行**: 逐个执行工作流节点
-3. **实时输出**: 通过 SSE 实时推送执行结果
-4. **状态更新**: 实时更新节点和工作流状态
-5. **错误处理**: 捕获并处理执行过程中的错误
-
-## 扩展开发
-
-### 1. 自定义节点开发
-
-要开发自定义工作流节点，需要：
-
-1. **创建节点类**：继承 `AbstractWfNode`
-2. **实现处理逻辑**：重写 `onProcess()` 方法
-3. **定义配置类**：创建节点配置类
-4. **注册组件**：在组件表中注册新组件
-
-```java
-public class CustomNode extends AbstractWfNode {
- @Override
- protected NodeProcessResult onProcess() {
- // 实现自定义处理逻辑
- List outputs = new ArrayList<>();
- // ... 处理逻辑
- return NodeProcessResult.success(outputs);
- }
-}
-```
-
-### 2. 自定义组件注册
-
-```sql
--- 在 t_workflow_component 表中添加新组件
-INSERT INTO t_workflow_component (uuid, name, title, remark, is_enable)
-VALUES (REPLACE(UUID(), '-', ''), 'CustomNode', '自定义节点', '自定义节点描述', true);
-```
diff --git a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/controller/chat/ChatController.java b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/controller/chat/ChatController.java
index 5a57c0e0..1580db92 100644
--- a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/controller/chat/ChatController.java
+++ b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/controller/chat/ChatController.java
@@ -1,10 +1,8 @@
package org.ruoyi.controller.chat;
-import jakarta.servlet.http.HttpServletRequest;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.ruoyi.common.chat.domain.dto.request.AgentChatRequest;
import org.ruoyi.common.chat.domain.dto.request.ChatRequest;
import org.ruoyi.service.chat.impl.ChatServiceFacade;
import org.springframework.stereotype.Controller;
diff --git a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java
index 32c86960..bdd89630 100644
--- a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java
+++ b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java
@@ -150,35 +150,59 @@ public class ChatServiceFacade implements IChatService {
* @return SseEmitter
*/
public SseEmitter sseChat(ChatRequest chatRequest) {
-
- // 具体的服务实现
Long userId = LoginHelper.getUserId();
String tokenValue = StpUtil.getTokenValue();
- // 每个会话一个 SSE 连接，避免同用户多会话串台
- SseEmitter emitter = sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
+
+ boolean workflowMode = Boolean.TRUE.equals(chatRequest.getEnableWorkFlow());
+ boolean agentMode = chatRequest.getAgentId() != null;
+ if (workflowMode && agentMode) {
+ throw new IllegalArgumentException("对话模式参数冲突：工作流和智能体不能同时启用");
+ }
+
+ // 工作流模式。工作流引擎负责创建并持有自己的 SSE，必须在普通聊天连接创建前路由。
+ if (workflowMode) {
+ chatMessageService.saveChatMessage(
+ userId,
+ chatRequest.getSessionId(),
+ chatRequest.getContent(),
+ RoleType.USER.getName(),
+ chatRequest.getModel()
+ );
+ return handleWorkflowChat(chatRequest);
+ }
// 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段
AgentVo agentVo = null;
- if (chatRequest.getAgentId() != null) {
+ if (agentMode) {
agentVo = agentService.queryById(chatRequest.getAgentId());
+ if (agentVo == null) {
+ throw new IllegalArgumentException("智能体不存在: " + chatRequest.getAgentId());
+ }
if (agentVo != null && agentVo.getModelId() != null) {
ChatModelVo agentModel = chatModelService.queryById(agentVo.getModelId());
- if (agentModel != null) {
- chatRequest.setModel(agentModel.getModelName());
+ if (agentModel == null) {
+ throw new IllegalArgumentException("智能体绑定的模型不存在: " + agentVo.getModelId());
}
- } else {
- log.warn("智能体不存在或未配置模型，回退到 model 字段: agentId={}", chatRequest.getAgentId());
+ chatRequest.setModel(agentModel.getModelName());
}
}
+ if (StringUtils.isBlank(chatRequest.getModel())) {
+ throw new IllegalArgumentException(
+ agentVo == null ? "对话模式必须指定模型" : "智能体未绑定模型，且请求未提供回退模型"
+ );
+ }
+
// 根据模型名称查询完整配置
ChatModelVo chatModelVo = chatModelService.selectModelByName(chatRequest.getModel());
if (chatModelVo == null) {
throw new IllegalArgumentException("模型不存在: " + chatRequest.getModel());
}
+ // 对话和智能体模式共用按会话隔离的 SSE。
+ SseEmitter emitter = sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
+
// 构建上下文消息列表（系统提示词 + 历史消息 + 当前用户消息）
- // 注意：RAG 检索增强统一在 handleAgentChat 中执行一次，此处不再重复检索
List contextMessages = buildContextMessages(chatRequest, agentVo);
chatRequest.setEmitter(emitter);
@@ -190,43 +214,68 @@ public class ChatServiceFacade implements IChatService {
// 保存用户消息
chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), chatRequest.getContent(), RoleType.USER.getName(), chatRequest.getModel());
- TraceRunHandle traceRun = Boolean.TRUE.equals(chatRequest.getEnableWorkFlow())
- ? null : startRagTraceRun(chatRequest, userId);
- // 3. 路由对话模式：工作流对话 / 智能体对话（两者均返回各自的 SseEmitter）
- return handleSpecialChatModes(chatRequest, agentVo, traceRun);
- }
+ TraceRunHandle traceRun = startRagTraceRun(chatRequest, userId);
- /**
- * 路由对话模式：仅两种情况——工作流对话 / 智能体对话。
- *
- * @param chatRequest 聊天请求
- * @param agentVo 智能体配置（可为 null）
- * @return 对应模式的 SseEmitter
- */
- private SseEmitter handleSpecialChatModes(ChatRequest chatRequest, AgentVo agentVo, TraceRunHandle traceRun) {
- // 模式1：工作流对话（前端应用市场选工作流后携带 workFlowRunner）
- if (Boolean.TRUE.equals(chatRequest.getEnableWorkFlow())) {
- log.info("处理工作流对话,会话: {}", chatRequest.getSessionId());
- WorkFlowRunner runner = chatRequest.getWorkFlowRunner();
- if (ObjectUtils.isEmpty(runner)) {
- log.warn("工作流参数为空");
- }
- return workFlowStarterService.streaming(
- ThreadContext.getCurrentUser(),
- runner.getUuid(),
- runner.getInputs(),
- chatRequest.getSessionId()
- );
+ // 智能体和普通对话互斥：有 agentId 为智能体，否则为普通模型对话。
+ if (agentVo != null) {
+ log.info("处理智能体对话,会话:{},agentId:{}", chatRequest.getSessionId(), chatRequest.getAgentId());
+ return handleAgentChat(chatRequest, agentVo, traceRun);
}
- // 模式2：智能体对话（默认走 Supervisor 多 Agent 编排）
- return handleAgentChat(chatRequest, agentVo, traceRun);
+ log.info("处理普通对话,会话:{},模型:{}", chatRequest.getSessionId(), chatRequest.getModel());
+ return handleModelChat(chatRequest, traceRun);
}
/**
- * 智能体对话模式（默认）：构建 Supervisor 多 Agent 编排并异步执行，结果通过 SSE 推送。
+ * 工作流模式。工作流运行时负责 SSE、节点执行和结束事件。
+ */
+ private SseEmitter handleWorkflowChat(ChatRequest chatRequest) {
+ WorkFlowRunner runner = chatRequest.getWorkFlowRunner();
+ if (ObjectUtils.isEmpty(runner) || StringUtils.isBlank(runner.getUuid())) {
+ throw new IllegalArgumentException("工作流模式必须提供 workFlowRunner.uuid");
+ }
+ log.info("处理工作流对话,会话:{},workflowUuid:{}", chatRequest.getSessionId(), runner.getUuid());
+ return workFlowStarterService.streaming(
+ ThreadContext.getCurrentUser(),
+ runner.getUuid(),
+ runner.getInputs() == null ? List.of() : runner.getInputs(),
+ chatRequest.getSessionId()
+ );
+ }
+
+ /**
+ * 普通对话模式：直接调用选定模型，不装配 Supervisor、MCP、Skills 或专业子 Agent。
+ */
+ private SseEmitter handleModelChat(ChatRequest chatRequest, TraceRunHandle traceRun) {
+ ChatModelVo chatModelVo = chatRequest.getChatModelVo();
+ AbstractChatService chatService = chatServiceFactory.getOriginalService(chatModelVo.getProviderCode());
+ StreamingChatModel streamingChatModel = chatService.buildStreamingChatModel(chatModelVo, chatRequest);
+ List messages = buildModelChatMessages(chatRequest);
+
+ TraceStreamSpan llmSpan = null;
+ try (TraceScope ignored = openTraceScope(traceRun, chatRequest.getUserId())) {
+ llmSpan = startLlmCallSpan(traceRun, chatRequest, "handleModelChat");
+ streamingChatModel.chat(
+ messages,
+ createModelChatResponseHandler(chatRequest, traceRun, llmSpan)
+ );
+ } catch (Exception e) {
+ if (llmSpan != null) {
+ llmSpan.finishError(e);
+ llmSpan.detach();
+ }
+ finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, e);
+ SseMessageUtils.sendError(String.valueOf(chatRequest.getSessionId()), e.getMessage());
+ SseMessageUtils.completeConnection(String.valueOf(chatRequest.getSessionId()));
+ log.error("普通对话执行失败", e);
+ }
+ return chatRequest.getEmitter();
+ }
+
+ /**
+ * 智能体对话模式：构建 Supervisor 多 Agent 编排并异步执行，结果通过 SSE 推送。
*
* @param chatRequest 聊天请求
- * @param agentVo 智能体配置（可为 null，无智能体时用请求 model 兜底）
+ * @param agentVo 智能体配置
*/
private SseEmitter handleAgentChat(ChatRequest chatRequest, AgentVo agentVo, TraceRunHandle traceRun) {
ChatModelVo chatModelVo = chatRequest.getChatModelVo();
@@ -321,7 +370,7 @@ public class ChatServiceFacade implements IChatService {
CompletableFuture.runAsync(() -> {
TraceStreamSpan llmSpan = null;
try (TraceScope ignored = openTraceScope(traceRun, userId)) {
- llmSpan = startLlmCallSpan(traceRun, chatRequest);
+ llmSpan = startLlmCallSpan(traceRun, chatRequest, "handleAgentChat");
String result = supervisor.invoke(prompt);
SseMessageUtils.sendContent(sessionId, result);
SseMessageUtils.sendDone(sessionId);
@@ -386,7 +435,8 @@ public class ChatServiceFacade implements IChatService {
traceRun.businessId, userId, traceRun.tenantId);
}
- private TraceStreamSpan startLlmCallSpan(TraceRunHandle traceRun, ChatRequest chatRequest) {
+ private TraceStreamSpan startLlmCallSpan(TraceRunHandle traceRun, ChatRequest chatRequest,
+ String methodName) {
if (traceRun == null || StringUtils.isBlank(TraceContext.getTraceId())) {
return null;
}
@@ -401,7 +451,7 @@ public class ChatServiceFacade implements IChatService {
node.setNodeName("llm-call");
node.setNodeType(RagTraceNodeTypes.NODE_LLM_CALL);
node.setClassName(ChatServiceFacade.class.getName());
- node.setMethodName("handleAgentChat");
+ node.setMethodName(methodName);
node.setStatus(TraceConstants.STATUS_RUNNING);
node.setStartTime(new Date(startMillis));
node.setInputPayload(RagTracePayloadBuilder.streamInputSummary(chatRequest));
@@ -564,7 +614,11 @@ public class ChatServiceFacade implements IChatService {
Long userId = LoginHelper.getUserId();
// 5. 建立 SSE 连接（用于前端监听，按会话隔离）
- sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
+ // 工作流调用时(externalHandler 非空), SSE 连接由工作流引擎创建并持有(WorkflowStarter#streaming),
+ // connect 为替换语义(关闭同键旧连接), 此处重连会掐断工作流连接, 必须跳过
+ if (externalHandler == null) {
+ sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
+ }
// 保存用户消息
chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), chatRequest.getContent(), RoleType.USER.getName(), chatRequest.getModel());
@@ -645,6 +699,19 @@ public class ChatServiceFacade implements IChatService {
return messages;
}
+ /**
+ * 构建普通对话消息。保留历史上下文，并在请求指定知识库时仅增强当前用户消息。
+ */
+ private List buildModelChatMessages(ChatRequest chatRequest) {
+ List messages = new ArrayList<>(chatRequest.getContextMessages());
+ String augmentedInput = augmentAgentInput(chatRequest, null);
+ int lastIndex = messages.size() - 1;
+ if (lastIndex >= 0 && messages.get(lastIndex) instanceof UserMessage) {
+ messages.set(lastIndex, UserMessage.userMessage(augmentedInput));
+ }
+ return messages;
+ }
+
/**
* 将上下文消息格式化为多轮对话文本（供只接受 String 输入的 Supervisor 使用）。
* 跳过 SystemMessage（系统提示词单独前置）与最后一条当前用户消息（单独做 RAG 增强后拼接）。
@@ -791,6 +858,77 @@ public class ChatServiceFacade implements IChatService {
return queryVectorBo;
}
+ /**
+ * 普通对话响应处理器：推送流式内容、保存助手消息并结束链路追踪。
+ */
+ private StreamingChatResponseHandler createModelChatResponseHandler(ChatRequest chatRequest,
+ TraceRunHandle traceRun,
+ TraceStreamSpan llmSpan) {
+ String sessionId = String.valueOf(chatRequest.getSessionId());
+ return new StreamingChatResponseHandler() {
+
+ private final StringBuilder messageBuffer = new StringBuilder();
+
+ @Override
+ public void onPartialResponse(String partialResponse) {
+ messageBuffer.append(partialResponse);
+ SseMessageUtils.sendContent(sessionId, partialResponse);
+ }
+
+ @Override
+ public void onPartialThinking(PartialThinking partialThinking) {
+ SseMessageUtils.sendReasoning(sessionId, partialThinking.text());
+ }
+
+ @Override
+ public void onCompleteResponse(ChatResponse completeResponse) {
+ try {
+ String fullMessage = messageBuffer.toString();
+ if (StringUtils.isNotBlank(fullMessage)) {
+ chatMessageService.saveChatMessage(
+ chatRequest.getUserId(),
+ chatRequest.getSessionId(),
+ fullMessage,
+ RoleType.ASSISTANT.getName(),
+ chatRequest.getModel()
+ );
+ } else {
+ log.warn("普通对话返回空消息,会话:{}", chatRequest.getSessionId());
+ }
+ if (llmSpan != null) {
+ llmSpan.finishSuccess(RagTracePayloadBuilder.streamOutputSummary(fullMessage.length()));
+ }
+ finishTraceRun(traceRun, TraceConstants.STATUS_SUCCESS, null);
+ SseMessageUtils.sendDone(sessionId);
+ } catch (Exception e) {
+ if (llmSpan != null) {
+ llmSpan.finishError(e);
+ }
+ finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, e);
+ SseMessageUtils.sendError(sessionId, e.getMessage());
+ log.error("普通对话完成处理失败", e);
+ } finally {
+ if (llmSpan != null) {
+ llmSpan.detach();
+ }
+ SseMessageUtils.completeConnection(sessionId);
+ }
+ }
+
+ @Override
+ public void onError(Throwable error) {
+ if (llmSpan != null) {
+ llmSpan.finishError(error);
+ llmSpan.detach();
+ }
+ finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, error);
+ SseMessageUtils.sendError(sessionId, error.getMessage());
+ SseMessageUtils.completeConnection(sessionId);
+ log.error("普通对话流式响应失败", error);
+ }
+ };
+ }
+
/**
* 创建组合响应处理器 - 同时发送到 SSE 和外部 handler
*
@@ -811,7 +949,10 @@ public class ChatServiceFacade implements IChatService {
messageBuffer.append(partialResponse);
// 2. 发送内容事件到 SSE（前端可通过 SSE 监听）
- SseMessageUtils.sendContent(sessionId, partialResponse);
+ // 工作流调用时连接归工作流引擎所有, token 由引擎以 [NODE_CHUNK_] 事件推送, 不走聊天协议
+ if (externalHandler == null) {
+ SseMessageUtils.sendContent(sessionId, partialResponse);
+ }
// 3. 转发给外部 handler（Workflow 等模块可处理）
if (externalHandler != null) {
@@ -821,8 +962,10 @@ public class ChatServiceFacade implements IChatService {
@Override
public void onPartialThinking(PartialThinking partialThinking) {
- // 发送推理内容到 SSE（前端通过 reasoning 事件监听）
- SseMessageUtils.sendReasoning(sessionId, partialThinking.text());
+ // 发送推理内容到 SSE（前端通过 reasoning 事件监听）, 工作流调用时不发送
+ if (externalHandler == null) {
+ SseMessageUtils.sendReasoning(sessionId, partialThinking.text());
+ }
// 转发给外部 handler
if (externalHandler != null) {
@@ -833,11 +976,12 @@ public class ChatServiceFacade implements IChatService {
@Override
public void onCompleteResponse(ChatResponse completeResponse) {
try {
- // 1. 发送完成事件
- SseMessageUtils.sendDone(sessionId);
-
- // 2. 关闭 SSE 连接
- SseMessageUtils.completeConnection(sessionId);
+ // 1&2. 发送完成事件并关闭 SSE 连接
+ // 工作流调用时流程可能还有后续节点, 连接关闭由工作流引擎统一负责, 此处不能关闭
+ if (externalHandler == null) {
+ SseMessageUtils.sendDone(sessionId);
+ SseMessageUtils.completeConnection(sessionId);
+ }
// 3. 转发给外部 handler
if (externalHandler != null) {
@@ -850,8 +994,10 @@ public class ChatServiceFacade implements IChatService {
@Override
public void onError(Throwable error) {
- // 发送错误事件
- SseMessageUtils.sendError(sessionId, error.getMessage());
+ // 发送错误事件（工作流调用时由工作流引擎统一上报）
+ if (externalHandler == null) {
+ SseMessageUtils.sendError(sessionId, error.getMessage());
+ }
log.error("流式响应错误: {}", error.getMessage(), error);
// 转发给外部 handler
-### 企业级AI助手平台
+### Enterprise-Grade AI Assistant Platform
-*开箱即用的全栈AI平台，支持多智能体协同、Supervisor模式编排、多种决策模式、RAG技术和流程编排能力*
+*An out-of-the-box full-stack AI platform supporting multi-agent collaboration, Supervisor mode orchestration, and multiple decision models, with advanced RAG technology and visual workflow orchestration capabilities*
-**[English](README_EN.md)** | **[📖 使用文档](https://doc.ruoyiai.chat/)** |
-**[🚀 在线体验](https://web.ruoyiai.chat/)** | **[🐛 问题反馈](https://github.com/ageerle/ruoyi-ai/issues)** | **[💡 功能建议](https://github.com/ageerle/ruoyi-ai/issues)**
+**[中文](README_ZH.md)** | **[📖 Documentation](https://doc.ruoyiai.chat/)** |
+**[🚀 Live Demo](https://web.ruoyiai.chat/)** | **[🐛 Report Issues](https://github.com/ageerle/ruoyi-ai/issues)** | **[💡 Feature Requests](https://github.com/ageerle/ruoyi-ai/issues)**
-## ✨ 核心亮点
-
-| 模块 | 现有能力
-|:---------:|---
-| **模型管理** | 多模型接入(DeepSeek/智谱/MIMO/百炼/OpenAI)、多模态理解、Coze/DIFY/FastGPT/RAGFlow平台集成
-| **知识管理** | 本地RAG + 向量库(Milvus/Weaviate/Qdrant) + 文档解析
-| **工具管理** | Mcp协议集成、Skills能力 + 可扩展工具生态
-| **流程编排** | 可视化工作流设计器、节点拖拽编排、SSE流式执行,目前已经支持模型调用,邮件发送,人工审核等节点
-| **智能体管理** | 基于Langchain4j的Agent框架、Supervisor模式编排,支持多种决策模型,可以灵活搭配工具,skills
-### 项目源码
+## ✨ Core Features
-| 项目模块 | GitHub 仓库 | Gitee 仓库 | GitCode 仓库 |
+| Module | Current Capabilities |
+|:---:|---|
+| **Model Management** | Multi-model integration (DeepSeek/Zhipu/MIMO/Bailian/OpenAI), multi-modal understanding, Coze/DIFY/FastGPT/RAGFlow platform integration |
+| **Knowledge Management** | Local RAG + Vector DB (Milvus/Weaviate/Qdrant) + Document parsing |
+| **Tool Management** | MCP protocol integration, Skills capability + Extensible tool ecosystem |
+| **Workflow Orchestration** | Visual workflow designer, drag-and-drop node orchestration, SSE streaming execution, currently supports model calls, email sending, manual review, and other nodes |
+| **Multi-Agent** | Agent framework based on Langchain4j, Supervisor mode orchestration, supports multiple decision models, can flexibly combine tools and skills |
+
+### Project Repositories
+
+| Module | GitHub Repository | Gitee Repository | GitCode Repository |
|----------|-------------------------------------------------------|------------------------------------------------------|--------------------------------------------------------|
-| 🔧 后端服务 | [ruoyi-ai](https://github.com/ageerle/ruoyi-ai) | [ruoyi-ai](https://gitee.com/ageerle/ruoyi-ai) | [ruoyi-ai](https://gitcode.com/ageerle/ruoyi-ai) |
-| 🎨 用户前端 | [ruoyi-web](https://github.com/ageerle/ruoyi-web) | [ruoyi-web](https://gitee.com/ageerle/ruoyi-web) | [ruoyi-web](https://gitcode.com/ageerle/ruoyi-web) |
-| 🛠️ 管理后台 | [ruoyi-admin](https://github.com/ageerle/ruoyi-admin) | [ruoyi-admin](https://gitee.com/ageerle/ruoyi-admin) | [ruoyi-admin](https://gitcode.com/ageerle/ruoyi-admin) |
-| 🎬 短剧平台 | [ruoyi-drama](https://github.com/ageerle/ruoyi-drama) | [ruoyi-drama](https://gitee.com/ageerle/ruoyi-drama) | [ruoyi-drama](https://gitcode.com/ageerle/ruoyi-drama) |
-| 🤖 编程助手 | [ruoyi-copilot](https://github.com/ageerle/ruoyi-copilot) | [ruoyi-copilot](https://gitee.com/ageerle/ruoyi-copilot) | [ruoyi-copilot](https://gitcode.com/ageerle/ruoyi-copilot) |
-| 📱 小程序端 | [ruoyi-uniapp](https://github.com/ageerle/ruoyi-uniapp) | [ruoyi-uniapp](https://gitee.com/ageerle/ruoyi-uniapp) | [ruoyi-uniapp](https://gitcode.com/ageerle/ruoyi-uniapp) |
+| 🔧 Backend | [ruoyi-ai](https://github.com/ageerle/ruoyi-ai) | [ruoyi-ai](https://gitee.com/ageerle/ruoyi-ai) | [ruoyi-ai](https://gitcode.com/ageerle/ruoyi-ai) |
+| 🎨 User Frontend | [ruoyi-web](https://github.com/ageerle/ruoyi-web) | [ruoyi-web](https://gitee.com/ageerle/ruoyi-web) | [ruoyi-web](https://gitcode.com/ageerle/ruoyi-web) |
+| 🛠️ Admin Panel | [ruoyi-admin](https://github.com/ageerle/ruoyi-admin) | [ruoyi-admin](https://gitee.com/ageerle/ruoyi-admin) | [ruoyi-admin](https://gitcode.com/ageerle/ruoyi-admin) |
+| 🎬 Drama | [ruoyi-drama](https://github.com/ageerle/ruoyi-drama) | [ruoyi-drama](https://gitee.com/ageerle/ruoyi-drama) | [ruoyi-drama](https://gitcode.com/ageerle/ruoyi-drama) |
+| 🤖 Copilot | [ruoyi-copilot](https://github.com/ageerle/ruoyi-copilot) | [ruoyi-copilot](https://gitee.com/ageerle/ruoyi-copilot) | [ruoyi-copilot](https://gitcode.com/ageerle/ruoyi-copilot) |
+| 📱 Mini-App | [ruoyi-uniapp](https://github.com/ageerle/ruoyi-uniapp) | [ruoyi-uniapp](https://gitee.com/ageerle/ruoyi-uniapp) | [ruoyi-uniapp](https://gitcode.com/ageerle/ruoyi-uniapp) |
-### 合作项目
-| 项目名称 | GitHub 仓库 | Gitee 仓库
+### Partner Projects
+| Project Name | GitHub Repository | Gitee Repository |
|----------------|-------------------------------------------------------|------------------------------------------------------|
-| element-plus-x | [element-plus-x](https://github.com/element-plus-x/Element-Plus-X) | [element-plus-x](https://gitee.com/he-jiayue/element-plus-x) |
+| element-plus-x | [element-plus-x](https://github.com/element-plus-x/Element-Plus-X) | [element-plus-x](https://gitee.com/he-jiayue/element-plus-x) |
-## 🛠️ 技术架构
+## 🛠️ Technical Architecture
-### 核心框架
-- **后端架构**：Spring Boot 3.5.8 + Langchain4j
-- **数据存储**：MySQL 8.0 + Redis + 向量数据库（Milvus/Weaviate/Qdrant）
-- **前端技术**：Vue 3 + Vben Admin + element-plus-x
-- **安全认证**：Sa-Token + JWT 双重保障
-- **文档处理**：PDF、Word、Excel 解析，图像智能分析
-- **实时通信**：WebSocket 实时通信，SSE 流式响应
-- **系统监控**：完善的日志体系、性能监控、服务健康检查
+### Core Framework
+- **Backend**: Spring Boot 3.5.8 + Langchain4j
+- **Data Storage**: MySQL 8.0 + Redis + Vector Databases (Milvus/Weaviate/Qdrant)
+- **Frontend**: Vue 3 + Vben Admin + element-plus-x
+- **Security**: Sa-Token + JWT dual-layer security
+- **Document Processing**: PDF, Word, Excel parsing, intelligent image analysis
+- **Real-time Communication**: WebSocket real-time communication, SSE streaming response
+- **System Monitoring**: Comprehensive logging system, performance monitoring, service health checks
-## 🐳 Docker 部署
+## 🐳 Docker Deployment
-本项目提供两种 Docker 部署方式：
+This project provides two Docker deployment methods:
-### 方式一：一键启动所有服务（推荐）
+### Method 1: One-click Start All Services (Recommended)
-使用 `docker-compose-all.yaml` 可以一键启动所有服务（包括后端、管理端、用户端及依赖服务）：
+Use `docker-compose-all.yaml` to start all services at once (including backend, admin panel, user frontend, and dependencies):
```bash
-# 克隆仓库
+# Clone the repository
git clone https://github.com/ageerle/ruoyi-ai.git
cd ruoyi-ai
-# 启动所有服务（从镜像仓库拉取预构建镜像）
+# Start all services (pull pre-built images from registry)
docker-compose -f docker-compose-all.yaml up -d
-# 查看服务状态
+# Check service status
docker-compose -f docker-compose-all.yaml ps
-# 访问服务
-# 管理端: http://localhost:25666 (admin / admin123)
-# 用户端: http://localhost:25137
-# 后端API: http://localhost:26039
+# Access services
+# Admin Panel: http://localhost:25666 (admin / admin123)
+# User Frontend: http://localhost:25137
+# Backend API: http://localhost:26039
```
-### 方式二：分步部署（源码编译）
+### Method 2: Step-by-step Deployment (Source Build)
-如果您需要从源码构建后端服务，请按照以下步骤操作：
+If you need to build backend services from source, follow these steps:
-#### 第一步：部署后端服务
+#### Step 1: Deploy Backend Service
```bash
-# 进入后端项目目录
+# Enter backend project directory
cd ruoyi-ai
-# 启动后端服务（源码编译构建）
+# Start backend service (build from source)
docker-compose up -d --build
-# 等待后端服务启动完成
+# Wait for backend service to start
docker-compose logs -f backend
```
-#### 第二步：部署管理端
+#### Step 2: Deploy Admin Panel
```bash
-# 进入管理端项目目录
+# Enter admin panel project directory
cd ruoyi-admin
-# 构建并启动管理端
+# Build and start admin panel
docker-compose up -d --build
-# 访问管理端
-# 地址: http://localhost:5666
+# Access admin panel
+# URL: http://localhost:5666
```
-#### 第三步：部署用户端（可选）
+#### Step 3: Deploy User Frontend (Optional)
```bash
-# 进入用户端项目目录
+# Enter user frontend project directory
cd ruoyi-web
-# 构建并启动用户端
+# Build and start user frontend
docker-compose up -d --build
-# 访问用户端
-# 地址: http://localhost:5137
+# Access user frontend
+# URL: http://localhost:5137
```
-### 服务端口说明
+### Service Ports
-| 服务 | 一键启动端口 | 分步部署端口 | 说明 |
+| Service | One-click Port | Step-by-step Port | Description |
|------|-------------|-------------|------|
-| 管理端 | 25666 | 5666 | 管理后台访问地址 |
-| 用户端 | 25137 | 5137 | 用户前端访问地址 |
-| 后端服务 | 26039 | 6039 | 后端 API 服务 |
-| MySQL | 23306 | 23306 | 数据库服务 |
-| Redis | 26379 | 6379 | 缓存服务 |
-| Weaviate | 28080 | 28080 | 向量数据库 |
-| MinIO API | 29000 | 9000 | 对象存储 API |
-| MinIO Console | 29090 | 9090 | 对象存储控制台 |
+| Admin Panel | 25666 | 5666 | Admin backend access |
+| User Frontend | 25137 | 5137 | User frontend access |
+| Backend Service | 26039 | 6039 | Backend API service |
+| MySQL | 23306 | 23306 | Database service |
+| Redis | 26379 | 6379 | Cache service |
+| Weaviate | 28080 | 28080 | Vector database |
+| MinIO API | 29000 | 9000 | Object storage API |
+| MinIO Console | 29090 | 9090 | Object storage console |
-### 镜像仓库
+### Image Registry
-所有镜像托管在阿里云容器镜像服务：
+All images are hosted on Alibaba Cloud Container Registry:
```
crpi-31mraxd99y2gqdgr.cn-beijing.personal.cr.aliyuncs.com/ruoyi_ai
```
-可用镜像：
-- `mysql:v3` - MySQL 数据库（包含初始化 SQL）
-- `redis:6.2` - Redis 缓存
-- `weaviate:1.30.0` - 向量数据库
-- `minio:latest` - 对象存储
-- `ruoyi-ai-backend:latest` - 后端服务
-- `ruoyi-ai-admin:latest` - 管理端前端
-- `ruoyi-ai-web:latest` - 用户端前端
+Available images:
+- `mysql:v3` - MySQL database (includes initialization SQL)
+- `redis:6.2` - Redis cache
+- `weaviate:1.30.0` - Vector database
+- `minio:latest` - Object storage
+- `ruoyi-ai-backend:latest` - Backend service
+- `ruoyi-ai-admin:latest` - Admin frontend
+- `ruoyi-ai-web:latest` - User frontend
-### 常用命令
+### Common Commands
```bash
-# 停止所有服务
+# Stop all services
docker-compose -f docker-compose-all.yaml down
-# 查看服务日志
-docker-compose -f docker-compose-all.yaml logs -f [服务名]
+# View service logs
+docker-compose -f docker-compose-all.yaml logs -f [service-name]
-# 重启某个服务
-docker-compose -f docker-compose-all.yaml restart [服务名]
+# Restart a service
+docker-compose -f docker-compose-all.yaml restart [service-name]
```
-## 📚 使用文档
+## 📚 Documentation
-想要深入了解安装部署、功能配置和二次开发？
+Want to learn more about installation, deployment, configuration, and secondary development?
-**👉 [完整使用文档](https://doc.ruoyiai.chat/)**
+**👉 [Complete Documentation](https://doc.ruoyiai.chat/)**
-## 🤝 参与贡献
+## 🤝 Contributing
-我们热烈欢迎社区贡献！无论您是资深开发者还是初学者，都可以为项目贡献力量 💪
+We warmly welcome community contributions! Whether you are a seasoned developer or just getting started, you can contribute to the project 💪
-### 贡献方式
+### How to Contribute
-1. **Fork** 项目到您的账户
-2. **创建分支** (`git checkout -b feature/新功能名称`)
-3. **提交代码** (`git commit -m '添加某某功能'`)
-4. **推送分支** (`git push origin feature/新功能名称`)
-5. **发起 Pull Request**
+1. **Fork** the project to your account
+2. **Create a branch** (`git checkout -b feature/new-feature-name`)
+3. **Commit your changes** (`git commit -m 'Add new feature'`)
+4. **Push to the branch** (`git push origin feature/new-feature-name`)
+5. **Create a Pull Request**
-> 💡 **小贴士**：建议将 PR 提交到 GitHub，我们会自动同步到其他代码托管平台
+> 💡 **Tip**: We recommend submitting PRs to GitHub, we will automatically sync to other code hosting platforms
-## 📄 开源协议
+## 📄 License
-本项目采用 **MIT 开源协议**，详情请查看 [LICENSE](LICENSE) 文件。
+This project is licensed under the **MIT License**. See the [LICENSE](LICENSE) file for details.
-## 🙏 特别鸣谢
+## 🙏 Acknowledgments
-感谢以下优秀的开源项目为本项目提供支持：
-- [Langchain4j](https://github.com/langchain4j/langchain4j) - 强大的 Java LLM 开发框架
-- [RuoYi-Vue-Plus](https://gitee.com/dromara/RuoYi-Vue-Plus) - 成熟的企业级快速开发框架
-- [Vben Admin](https://github.com/vbenjs/vue-vben-admin) - 现代化的 Vue 后台管理模板
+Thanks to the following excellent open-source projects for their support:
+- [Langchain4j](https://github.com/langchain4j/langchain4j) - Powerful Java LLM development framework
+- [RuoYi-Vue-Plus](https://gitee.com/dromara/RuoYi-Vue-Plus) - Mature enterprise-level rapid development framework
+- [Vben Admin](https://github.com/vbenjs/vue-vben-admin) - Modern Vue admin template
+## 💎 Sponsors
-## 💎 赞助商
-
-**感谢以下赞助商对本项目的支持：**
+**Thanks to the following sponsors for supporting this project:**
+ 
