✨ 新功能 实现普通节点的一拖多/多对一的分支并行处理

agent-flow/src/components/base/jadeNode.jsx

@@ -469,7 +469,7 @@ export const jadeNode = (id, x, y, width, height, parent, drawer) => {
    * @returns {number} 连接数。
    */
   self.maxNumToLink = () => {
-    return 1;
+    return self.graph?.connectionLimitDisabled ? 100 : 1;
   };
 
   /**

agent-flow/src/components/base/validator.js

@@ -35,7 +35,9 @@ export class NormalNodeConnectorValidator extends Validator {
   validate() {
     const nextEvents = this.node.getNextRunnableEvents();
     const i18n = this.node.graph.i18n;
-    if (nextEvents.length !== 1) {
+    const isConnectionLimitDisabled = Boolean(this.node.graph?.connectionLimitDisabled);
+    const isValid = isConnectionLimitDisabled ? nextEvents.length >= 1 : nextEvents.length === 1;
+    if (!isValid) {
       return Promise.reject({
         errorFields: [{
           errors: [`${i18n?.t('node') ?? 'node'} ${this.node.text} ${i18n?.t('problemWithConnection') ?? 'problemWithConnection'}`],

agent-flow/src/components/code/codeNodeState.jsx

@@ -93,7 +93,7 @@ export const codeNodeState = (id, x, y, width, height, parent, drawer) => {
      * @override
      */
     self.maxNumToLink = () => {
-        return 10;
+        return self.graph?.connectionLimitDisabled ? 100 : 10;
     };
 
     return self;

agent-flow/src/flow/jadeFlowEntry.jsx

@@ -378,6 +378,10 @@ const jadeFlowAgent = (graph) => {
     graph.destroy();
   };
 
+  self.setConnectionLimitDisabled = (disabled) => {
+    graph.connectionLimitDisabled = Boolean(disabled);
+  };
+
   return self;
 };
 
@@ -432,6 +436,7 @@ export const JadeFlow = (() => {
                        div,
                        tenant,
                        appId,
+                       connectionLimitDisabled = false,
                        flowConfigData,
                        configs,
                        i18n,
@@ -440,7 +445,7 @@ export const JadeFlow = (() => {
                      }) => {
     const graphDom = getGraphDom(div);
     const g = jadeFlowGraph(div, 'jadeFlow');
-    await configGraph(g, tenant, appId, flowConfigData, configs, i18n, importStatements);
+    await configGraph(g, tenant, appId, flowConfigData, configs, i18n, importStatements, connectionLimitDisabled);
     g.flowType = flowType;
     const pageData = g.getPageData(0);
     await g.editFlow(0, graphDom, pageData.id);
@@ -470,8 +475,9 @@ export const JadeFlow = (() => {
     return jadeFlowAgent(g);
   };
 
-  const configGraph = async (g, tenant, appId, flowConfigData, configs, i18n, importStatements) => {
+  const configGraph = async (g, tenant, appId, flowConfigData, configs, i18n, importStatements, connectionLimitDisabled = false) => {
     g.collaboration.mute = true;
+    g.connectionLimitDisabled = Boolean(connectionLimitDisabled);
     g.configs = configs;
     g.i18n = i18n;
     for (let i = 0; i < importStatements.length; i++) {

agent-flow/src/flow/jadeFlowGraph.js

@@ -76,6 +76,7 @@ export const jadeFlowGraph = (div, title) => {
   const self = defaultGraph(div, title);
   self.type = 'jadeFlowGraph';
   self.pageType = 'jadeFlowPage';
+  self.connectionLimitDisabled = false;
   self.enableText = false;
   self.flowMeta = {
     exceptionFitables: ['modelengine.fit.jober.aipp.fitable.AippFlowExceptionHandler'],

agent-flow/src/flow/jadeFlowPage.js

@@ -40,6 +40,8 @@ export const jadeFlowPage = (div, graph, name, id) => {
   self.addEventListener('COPY_SHAPE', shapeChangeListener);
   self.addEventListener('DELETE_SHAPE', shapeChangeListener);
 
+  const isConnectionLimitDisabled = () => Boolean(self.graph?.connectionLimitDisabled);
+
   /**
    * @override
    */
@@ -305,7 +307,7 @@ export const jadeFlowPage = (div, graph, name, id) => {
    */
   self.canDragOut = (node, connector) => {
     const lines = self.getEvents().filter(s => s.fromShape === node.id && s.getDefinedFromConnector() === connector);
-    return lines && lines.length < 1;
+    return lines.length < (isConnectionLimitDisabled() ? 10 : 1);
   };
 
   /**
@@ -330,7 +332,9 @@ export const jadeFlowPage = (div, graph, name, id) => {
       }
     };
 
-    return jadeEvent.fromShape !== jadeEvent.toShape && isConnectorAllowToLink() && isConnectorWithinLimit();
+    return jadeEvent.fromShape !== jadeEvent.toShape
+      && isConnectorAllowToLink()
+      && (isConnectionLimitDisabled() || isConnectorWithinLimit());
   };
 
   /**

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

@@ -265,6 +265,29 @@ public <R> List<FlowContext<R>> generate(List<R> data, String position) {
         return data.stream().map(d -> this.generate(d, position, LocalDateTime.now())).collect(Collectors.toList());
     }
 
+    /**
+     * fork一个新的context用于一拖多分支，继承当前context的运行元数据，但生成新的contextId。
+     *
+     * @return 新的分支context
+     */
+    public FlowContext<T> fork() {
+        return this.convertData(this.data);
+    }
+
+    /**
+     * convertData
+     *
+     * @param <R> 转换后的数据类型
+     * @param data 转换后的数据
+     * @return 转换后的context
+     */
+    public <R> FlowContext<R> convertData(R data) {
+        FlowContext<R> context = this.copyContextWithoutID(data);
+        context.previous = this.previous;
+        context.nextPositionId = this.nextPositionId;
+        return context;
+    }
+
     /**
      * 用于when.convert数据时候的转换context，除了包裹的数据类型不一样，所有其他信息都一样
      *
@@ -274,12 +297,17 @@ public <R> List<FlowContext<R>> generate(List<R> data, String position) {
      * @return 转换后的context
      */
     public <R> FlowContext<R> convertData(R data, String id) {
+        FlowContext<R> context = this.copyContextWithoutID(data);
+        context.previous = this.previous;
+        context.id = id;
+        return context;
+    }
+
+    private <R> FlowContext<R> copyContextWithoutID(R data) {
         FlowContext<R> context = new FlowContext<>(this.streamId, this.rootId, data, this.traceId, this.position,
                 this.parallel, this.parallelMode, LocalDateTime.now());
-        context.previous = this.previous;
         context.status = this.status;
         context.trans = this.trans;
-        context.id = id;
         context.batchId = this.batchId;
         context.toBatch = this.toBatch;
         context.createAt = this.createAt;

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextRepo.java

@@ -92,16 +92,16 @@ default void update(List<FlowContext<T>> contexts) {
     }
 
     /**
-     * updateToSent
+     * 更新context状态为SENT
      *
-     * @param contexts contexts
+     * @param contexts 上下文列表
      */
     void updateToSent(List<FlowContext<T>> contexts);
 
     /**
-     * updateToReady
+     * 更新context状态为READY
      *
-     * @param contexts contexts
+     * @param contexts 上下文列表
      */
     void updateToReady(List<FlowContext<T>> contexts);
 

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/FlowDefinition.java

@@ -179,6 +179,7 @@ public FitStream.Publisher<FlowData> convertToFlow(FlowContextRepo<FlowData> rep
                     //  startNode不能出现在event的to属性, endNode不能出现在event的from属性
                     FlowNode toNode = nodeMap.get(event.getTo());
                     fromNode.subscribe(streamId, flowEnv, toNode, event);
+
                 });
             });
             return getFlowNode(FlowNodeType.START).getPublisher(streamId, repo, messenger, locks);

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowForkNode.java

@@ -46,7 +46,8 @@ public class FlowForkNode extends FlowNode {
     public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, FlowContextRepo<FlowData> repo,
             FlowContextMessenger messenger, FlowLocks locks) {
         if (!Optional.ofNullable(processor).isPresent()) {
-            this.processor = new Node<>(streamId, this.metaId, this::forkJuster, repo, messenger, locks, this.type);
+            this.processor = new Node<>(streamId, this.metaId, this::forkJuster, repo, messenger, locks, this.type,
+                    FlowData.class);
             this.processor.onError(errorHandler(streamId));
         }
         return this.processor;

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowStateNode.java

@@ -20,8 +20,13 @@
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.FitStream;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes.Blocks;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes.Node;
+import modelengine.fit.waterflow.flowsengine.utils.FlowUtil;
 import modelengine.fitframework.log.Logger;
+import modelengine.fitframework.util.CollectionUtils;
+import modelengine.fitframework.util.ObjectUtils;
+import modelengine.fitframework.util.StringUtils;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -54,10 +59,11 @@ public class FlowStateNode extends FlowNode {
      */
     @Override
     public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, FlowContextRepo<FlowData> repo,
-            FlowContextMessenger messenger, FlowLocks locks) {
+                                                                FlowContextMessenger messenger, FlowLocks locks) {
         if (!Optional.ofNullable(this.processor).isPresent()) {
             Node<FlowData, FlowData> node = new Node<>(streamId, this.metaId, this::stateProduce, repo, messenger,
-                    locks, this.type);
+                    locks, this.type, FlowData.class);
+
             if (!Objects.isNull(this.jober)) {
                 node.setIsAsyncJob(this.jober.isAsync());
             }
@@ -74,6 +80,50 @@ public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, Flo
         return this.processor;
     }
 
+    private List<FlowContext<FlowData>> mergeProcessInputs(List<FlowContext<FlowData>> pre) {
+        if (CollectionUtils.isEmpty(pre) || pre.size() <= 1) {
+            return pre;
+        }
+        if (pre.stream().anyMatch(context -> !(context.getData() instanceof FlowData))) {
+            return pre;
+        }
+        if (pre.stream().map(FlowContext::getPosition).filter(StringUtils::isNotEmpty).distinct().count() <= 1) {
+            return pre;
+        }
+        FlowContext<FlowData> baseContext = pre.get(0);
+        FlowData mergedFlowData = mergeFlowData(pre, baseContext.getId());
+        return Collections.singletonList(
+                baseContext.convertData(ObjectUtils.cast(mergedFlowData), baseContext.getId()));
+    }
+
+    private FlowData mergeFlowData(List<FlowContext<FlowData>> pre, String baseContextId) {
+        FlowData first = pre.get(0).getData();
+        Map<String, Object> businessData = new HashMap<>(
+                Optional.ofNullable(first.getBusinessData()).orElseGet(HashMap::new));
+        Map<String, Object> contextData = new HashMap<>(
+                Optional.ofNullable(first.getContextData()).orElseGet(HashMap::new));
+        Map<String, Object> passData = new HashMap<>(Optional.ofNullable(first.getPassData()).orElseGet(HashMap::new));
+
+        pre.stream().skip(1).map(FlowContext::getData).forEach(flowData -> {
+            businessData.putAll(FlowUtil.mergeMaps(businessData,
+                    Optional.ofNullable(flowData.getBusinessData()).orElseGet(HashMap::new)));
+            contextData.putAll(FlowUtil.mergeMaps(contextData,
+                    Optional.ofNullable(flowData.getContextData()).orElseGet(HashMap::new)));
+            passData.putAll(FlowUtil.mergeMaps(passData,
+                    Optional.ofNullable(flowData.getPassData()).orElseGet(HashMap::new)));
+        });
+            contextData.put(Constant.CONTEXT_ID, baseContextId);
+        return FlowData.builder()
+                .operator(first.getOperator())
+                .startTime(first.getStartTime())
+                .businessData(businessData)
+                .contextData(contextData)
+                .passData(passData)
+                .errorMessage(first.getErrorMessage())
+                .errorInfo(first.getErrorInfo())
+                .build();
+    }
+
     private List<FlowData> stateProduce(List<FlowContext<FlowData>> inputs) {
         addContextData(inputs);
         List<FlowData> flowDataList = inputs.stream().map(FlowContext::getData).collect(Collectors.toList());

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/FlowDataMerger.java

@@ -0,0 +1,64 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *  This file is a part of the ModelEngine Project.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+package modelengine.fit.waterflow.flowsengine.domain.flows.streams;
+
+import modelengine.fit.waterflow.flowsengine.domain.flows.context.FlowContext;
+import modelengine.fit.waterflow.flowsengine.domain.flows.context.FlowData;
+import modelengine.fit.waterflow.flowsengine.utils.FlowUtil;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * FlowData 类型数据的多输入合并器
+ * 用于 fan-in 场景下将多条 FlowData 输入合并为单条处理
+ *
+ * @author 高诗意
+ * @since 2023/08/14
+ */
+public class FlowDataMerger implements Processors.Merger<FlowData> {
+
+    @Override
+    public FlowContext<FlowData> merge(List<FlowContext<FlowData>> contexts) {
+        if (contexts == null || contexts.isEmpty()) {
+            return null;
+        }
+        FlowContext<FlowData> baseContext = contexts.get(0);
+        FlowData mergedFlowData = mergeFlowData(contexts);
+        return baseContext.convertData(mergedFlowData, baseContext.getId());
+    }
+
+    private FlowData mergeFlowData(List<FlowContext<FlowData>> contexts) {
+        FlowData first = contexts.get(0).getData();
+        Map<String, Object> businessData = new HashMap<>(
+                Optional.ofNullable(first.getBusinessData()).orElseGet(HashMap::new));
+        Map<String, Object> contextData = new HashMap<>(
+                Optional.ofNullable(first.getContextData()).orElseGet(HashMap::new));
+        Map<String, Object> passData = new HashMap<>(
+                Optional.ofNullable(first.getPassData()).orElseGet(HashMap::new));
+
+        contexts.stream().skip(1).map(FlowContext::getData).forEach(flowData -> {
+            businessData.putAll(FlowUtil.mergeMaps(businessData,
+                    Optional.ofNullable(flowData.getBusinessData()).orElseGet(HashMap::new)));
+            contextData.putAll(FlowUtil.mergeMaps(contextData,
+                    Optional.ofNullable(flowData.getContextData()).orElseGet(HashMap::new)));
+            passData.putAll(FlowUtil.mergeMaps(passData,
+                    Optional.ofNullable(flowData.getPassData()).orElseGet(HashMap::new)));
+        });
+        return FlowData.builder()
+                .operator(first.getOperator())
+                .startTime(first.getStartTime())
+                .businessData(businessData)
+                .contextData(contextData)
+                .passData(passData)
+                .errorMessage(first.getErrorMessage())
+                .errorInfo(first.getErrorInfo())
+                .build();
+    }
+}