✨ 新功能实现普通节点的一拖多/多对一的分支并行处理 by swfnotswift · Pull Request #600 · ModelEngine-Group/app-platform

swfnotswift · 2026-03-11T06:01:29Z

🔗 相关问题 / Related Issue

Issue 链接 / Issue Link: #{$IssueNumber} 👈👈

我已经创建了相关 Issue 并进行了讨论 / I have created and discussed the related issue
这是一个微小的修改（如错别字），不需要 Issue / This is a trivial change (like typo fix) that doesn't need an issue

📋 变更类型 / Type of Change

✨ 新功能 / New feature (non-breaking change which adds functionality)

📝 变更目的 / Purpose of the Change

解决 Waterflow 普通节点一拖多、多对一、多输入 LLM 节点执行异常而做的代码修改。

📋 主要变更 / Brief Changelog

1. FlowContext.java

/**

 * 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.copyContext(data);

    context.previous = this.id;

    context.nextPositionId = this.nextPositionId;

    return context;

}

  

/**

 * 用于when.convert数据时候的转换context，除了包裹的数据类型不一样，所有其他信息都一样

 *

 * @param <R> 转换后的数据类型

 * @param data 转换后的数据

 * @param id contextId

 * @return 转换后的context

 */

public <R> FlowContext<R> convertData(R data, String id) {

    FlowContext<R> context = this.copyContext(data);

    context.previous = this.previous;

    context.id = id;

    return context;

}

  

private <R> FlowContext<R> copyContext(R data) {

    FlowContext<R> context = new FlowContext<>(this.streamId, this.rootId, data, this.traceId, this.position,

            this.parallel, this.parallelMode, LocalDateTime.now());

    context.status = this.status;

    context.trans = this.trans;

    context.batchId = this.batchId;

    context.toBatch = this.toBatch;

    context.createAt = this.createAt;

    context.updateAt = this.updateAt;

    context.archivedAt = this.archivedAt;

    return context;

}

修改说明

这一段的目标，是让普通节点一拖多时每条分支都拥有独立的 context 身份。

原来的问题是：同一个上游 context 命中多条出边时，后续链路仍然可能共享同一个 contextId。这样一来，多个分支在持久化、状态推进、批次更新时会互相覆盖，看起来像“只跑了一条分支”或者“后到的分支把先到的分支覆盖掉了”。

本次新增的 fork() 本质上是一个语义化入口，表示“基于当前上下文复制一个新分支”。

convertData(data) 用于生成一个新的 context，对外表现为新分支
新分支会继承原上下文的大部分运行元数据
但它会保留新的 contextId，且 previous 指向原 context
nextPositionId 也会一并继承，确保分支在后续送边时不会丢目标信息

copyContext(...) 被抽出来之后，两个 convertData(...) 的职责就清晰了：

convertData(data) 用于“派生一个新身份”
convertData(data, id) 用于“只替换 data，但保留原身份”

这两个入口分别对应本次修复中的两种完全不同的场景：

一拖多分支复制，需要新身份
When.convert 或执行期临时合并输入，只需要替换 data，不需要换 identity

2. From.java

@Override

public void offer(List<FlowContext<I>> contexts, Consumer<PreSendCallbackInfo<I>> preSendCallback) {

    if (CollectionUtils.isEmpty(contexts)) {

        return;

    }

    if (contexts.stream().map(c -> c.getTrans().getId()).distinct().count() != 1) {

        return;

    }

  

    FlowContext<I> context = contexts.get(0);

    List<FitStream.Subscription<I, ?>> qualifiedWhens = this.getSubscriptions();

  

    List<FitStream.Subscription<I, ?>> sourceWhens = this.getSubscriptions()

            .stream()

            .filter(w -> !context.getStreamId().equals(this.getStreamId()) && w.getTo()

                    .getStreamId()

                    .equals(context.getStreamId()))

            .collect(Collectors.toList());

    if (CollectionUtils.isNotEmpty(sourceWhens)) {

        qualifiedWhens = sourceWhens;

    }

  

    java.util.Map<FitStream.Subscription<I, ?>, List<FlowContext<I>>> matchedContexts = new LinkedHashMap<>();

    Set<FlowContext<I>> matchedContextSet = new HashSet<>();

    List<FlowContext<I>> forkedContexts = new ArrayList<>();

    for (FlowContext<I> contextItem : contexts) {

        List<FitStream.Subscription<I, ?>> matchedSubscriptions = qualifiedWhens.stream()

                .filter(w -> w.getWhether().is(contextItem))

                .collect(Collectors.toList());

        if (CollectionUtils.isEmpty(matchedSubscriptions)) {

            continue;

        }

        matchedContextSet.add(contextItem);

        for (int index = 0; index < matchedSubscriptions.size(); index++) {

            FitStream.Subscription<I, ?> matchedSubscription = matchedSubscriptions.get(index);

            FlowContext<I> branchContext = index == 0 ? contextItem : contextItem.fork();

            branchContext.setNextPositionId(matchedSubscription.getId());

            matchedContexts.computeIfAbsent(matchedSubscription, key -> new ArrayList<>()).add(branchContext);

            if (index > 0) {

                forkedContexts.add(branchContext);

            }

        }

    }

    qualifiedWhens.forEach(w -> matchedContexts.computeIfAbsent(w, key -> new ArrayList<>()));

    List<FlowContext<I>> unMatchedContexts = contexts

            .stream()

            .filter(c -> !matchedContextSet.contains(c))

            .collect(Collectors.toList());

    PreSendCallbackInfo<I> callbackInfo = new PreSendCallbackInfo<>(matchedContexts, unMatchedContexts);

    preSendCallback.accept(callbackInfo);

    persistForkedContexts(forkedContexts, matchedContexts);

    matchedContexts.forEach(FitStream.Subscription::cache);

}

  

private void persistForkedContexts(List<FlowContext<I>> forkedContexts,

        java.util.Map<FitStream.Subscription<I, ?>, List<FlowContext<I>>> matchedContexts) {

    if (CollectionUtils.isEmpty(forkedContexts)) {

        return;

    }

    Set<String> forkedIds = forkedContexts.stream().map(FlowContext::getId).collect(Collectors.toSet());

    List<FlowContext<I>> effectiveForkedContexts = matchedContexts.values()

            .stream()

            .flatMap(List::stream)

            .filter(context -> forkedIds.contains(context.getId()))

            .collect(Collectors.toList());

    if (CollectionUtils.isEmpty(effectiveForkedContexts)) {

        return;

    }

    Set<String> traces = effectiveForkedContexts.stream()

            .flatMap(context -> context.getTraceId().stream())

            .collect(Collectors.toSet());

    Lock lock = this.locks.getDistributedLock(this.locks.streamNodeLockKey(this.streamId, this.id,

            "ForkContextPool"));

    lock.lock();

    try {

        this.repo.updateContextPool(effectiveForkedContexts, traces);

        this.repo.save(effectiveForkedContexts);

    } finally {

        lock.unlock();

    }

}

修改说明

这一段的目标，是在真正发往多条边时，把分支拆成独立上下文，而不是只在概念上“一拖多”。

offer(List<FlowContext<I>> ...) 里现在的处理策略是：

如果某个 context 只命中一条边，沿用原 context
如果命中多条边，第一条边仍使用原 context
从第二条边开始，对每条边调用 fork() 生成新的 branch context

这样做有两个好处：

不会把所有分支都强制改成新对象，保留第一条链路的兼容性
又能保证剩余分支拥有独立 contextId，避免互相覆盖

新增的 persistForkedContexts(...) 用来解决第二层问题：即使分支 context 在内存里分出来了，如果不先落库、不先加入 trace 的 contextPool，后续 When.cache(...)、updateStatus(...)、requestAll(...) 仍然会出现查不到、覆盖、丢分支的问题。

它做了两件事：

先 updateContextPool(...)，把新分支 contextId 加入 trace 可见范围
再 save(...)，把这些 fork 出来的 context 持久化

并且这里加了分布式锁，避免并发分支扩散时 contextPool 更新出现竞争覆盖。

3. To.java: fan-in 配置与 request 过滤

private FanInMode fanInMode = FanInMode.ANY;

  

private boolean fanInModeConfigured = false;

  

private Processors.Map<FlowContext<I>, String> mergeKeyGenerator = this::defaultMergeKey;

  

private Processors.Filter<I> requestFilter(Processors.Filter<I> fallbackFilter) {

    if (!FanInMode.ALL.equals(this.fanInMode)) {

        return fallbackFilter;

    }

    return this::selectReadyMergeGroup;

}

  

public void setFanInMode(FanInMode fanInMode) {

    this.fanInMode = Optional.ofNullable(fanInMode).orElse(FanInMode.ANY);

    this.fanInModeConfigured = true;

}

  

public void setMergeKeyGenerator(Processors.Map<FlowContext<I>, String> mergeKeyGenerator) {

    this.mergeKeyGenerator = Optional.ofNullable(mergeKeyGenerator).orElse(this::defaultMergeKey);

}

  

private <T1> String defaultMergeKey(FlowContext<T1> context) {

    String rootId = Optional.ofNullable(context.getRootId()).orElse("");

    String transId = Optional.ofNullable(context.getTrans()).map(trans -> trans.getId()).orElse("");

    String traceIds = context.getTraceId().stream().sorted().collect(Collectors.joining(","));

    return StringUtils.join("|", rootId, transId, traceIds);

}

  

private <T1> String buildMergeKey(FlowContext<T1> context) {

    try {

        String mergeKey = this.mergeKeyGenerator.process(ObjectUtils.cast(context));

        if (StringUtils.isNotEmpty(mergeKey)) {

            return mergeKey;

        }

    } catch (Exception exception) {

        LOG.warn("build merge key failed for context: {}", context.getId(), exception);

    }

    return defaultMergeKey(context);

}

  

@Override

public void onSubscribe(FitStream.Subscription<?, I> subscription) {

    this.froms.add(subscription);

    if (!this.fanInModeConfigured) {

        long fromCount = this.froms.stream().map(Identity::getId).distinct().count();

        this.fanInMode = fromCount > 1 ? FanInMode.ALL : FanInMode.ANY;

    }

}

修改说明

这一段的目标，是给多对一节点建立一套明确的“汇聚判定规则”。

本次引入了三个关键概念：

fanInMode
fanInModeConfigured
mergeKeyGenerator

含义分别是：

fanInMode 决定多输入节点是“来一条就处理”还是“必须全部到齐再处理”
fanInModeConfigured 用于区分“用户主动配置”和“框架自动推断”
mergeKeyGenerator 用于定义哪些输入属于同一组汇聚数据

默认 mergeKey 使用 rootId + transId + traceId 集合 组合生成，目的是尽量把同一次流程实例里的同一组分支归到一个 key 下。

onSubscribe(...) 里增加的自动切换逻辑表示：

如果当前节点只有一个上游来源，默认 ANY
如果当前节点接入多个 distinct from，默认 ALL

这保证普通多输入节点在未显式配置时，也能先按“需要汇聚”处理。

4. To.java: 多输入 ready 判断与完整汇聚组选择

private <T1> List<FlowContext<T1>> filterReadyByFanIn(List<FlowContext<T1>> candidates) {

    if (CollectionUtils.isEmpty(candidates)) {

        return Collections.emptyList();

    }

    if (FanInMode.ANY.equals(this.fanInMode)) {

        return candidates;

    }

  

    long expectedInputs = this.froms.stream().map(Identity::getId).distinct().count();

    if (expectedInputs <= 1) {

        return candidates;

    }

  

    Map<String, List<FlowContext<T1>>> grouped = candidates.stream().collect(Collectors.groupingBy(this::buildMergeKey));

    Set<String> qualifiedMergeKeys = grouped.entrySet()

            .stream()

            .filter(entry -> entry.getValue()

                    .stream()

                    .map(FlowContext::getPosition)

                    .filter(StringUtils::isNotEmpty)

                    .distinct()

                    .count() >= expectedInputs)

            .map(Map.Entry::getKey)

            .collect(Collectors.toSet());

    return candidates.stream().filter(context -> qualifiedMergeKeys.contains(buildMergeKey(context))).collect(

            Collectors.toList());

}

  

private <T1> List<FlowContext<T1>> selectReadyMergeGroup(List<FlowContext<T1>> candidates) {

    if (CollectionUtils.isEmpty(candidates)) {

        return Collections.emptyList();

    }

    long expectedInputs = this.froms.stream().map(Identity::getId).distinct().count();

    if (expectedInputs <= 1) {

        return candidates;

    }

    Map<String, List<FlowContext<T1>>> grouped = new LinkedHashMap<>();

    candidates.forEach(context -> grouped.computeIfAbsent(buildMergeKey(context), key -> new ArrayList<>()).add(

            context));

    return grouped.values()

            .stream()

            .filter(group -> group.stream()

                    .map(FlowContext::getPosition)

                    .filter(StringUtils::isNotEmpty)

                    .distinct()

                    .count() >= expectedInputs)

            .findFirst()

            .orElseGet(Collections::emptyList);

}

  

private <T1> List<FlowContext<T1>> markReady(List<FlowContext<T1>> contexts) {

    this.introduceToProcess(contexts);

    return contexts.stream().filter(context -> context.getStatus() == FlowNodeStatus.READY).collect(

            Collectors.toList());

}

修改说明

这一段解决的是多对一节点 request 阶段的两个问题：

候选数据虽然查出来了，但不知道哪些已经真正“到齐”
不能把所有 pending 混在一起直接处理，否则会把不同批次、不同分支组混起来

filterReadyByFanIn(...) 的职责是做“到齐判定”。

它按 mergeKey 分组，然后统计每个 group 里出现了多少个不同的 position。当 distinct position 数量大于等于当前节点的上游输入数量时，说明这一组输入已经凑齐。

这里用 position 而不是直接用 context 数量，是因为我们真正关心的是“是否来自不同上游来源”，而不是“来了几条数据”。

selectReadyMergeGroup(...) 的职责是做“请求期截取”。

它不再像默认 filter 那样按 batch 简单取一批，而是：

先按 mergeKey 归组
再找出第一组已经满足完整输入数量的 group
只返回这一组

这样可以避免多组 pending 混在一起被同时拉入一次处理。

markReady(...) 则把原来 ready 标记和 ready 过滤的动作收口成一个独立步骤，保证 fan-in 分组判断在前，状态推进在后，顺序更稳定。

5. To.java: 处理阶段合并多输入上下文

@Override

public void onProcess(List<FlowContext<I>> pre) {

    if (CollectionUtils.isEmpty(pre)) {

        return;

    }

    try {

        if (!isOwnTrace(pre)) {

            LOG.warn("[BeforeProcess] The trace is not belong to this node, traceId={}.",

                    String.join(",", pre.get(0).getTraceId()));

            return;

        }

        beforeProcess(pre);

        if (pre.size() == 1 && pre.get(0).getData() == null) {

            this.afterProcess(pre, new ArrayList<>());

            return;

        }

        List<FlowContext<I>> processInputs = mergeProcessInputs(pre);

        if (this.isAsyncJob) {

            beforeAsyncProcess(pre);

            this.getProcessMode().process(this, processInputs);

            return;

        }

        List<FlowContext<O>> after = this.getProcessMode().process(this, processInputs);

        if (!isOwnTrace(pre)) {

            LOG.warn("[AfterProcess] The trace is not belong to this node, traceId={}.",

                    String.join(",", pre.get(0).getTraceId()));

            return;

        }

        this.afterProcess(pre, after);

    } catch (Exception ex) {

        LOG.error("node process exception stream-id: {}, node-id: {}, position-id: {}, traceId: {}. errors: {}",

                this.streamId, this.id, pre.get(0).getPosition(), pre.get(0).getTraceId(), ex.getMessage());

        LOG.error("node process exception details: ", ex);

        setFailed(pre, ex);

    } finally {

        updateConcurrency(-1);

    }

}

  

private List<FlowContext<I>> mergeProcessInputs(List<FlowContext<I>> pre) {

    if (!FanInMode.ALL.equals(this.fanInMode) || pre.size() <= 1) {

        return pre;

    }

    if (!(ProcessMode.MAPPING.equals(this.processMode)

            || ProcessMode.FLATMAPPING.equals(this.processMode)

            || ProcessMode.PRODUCING.equals(this.processMode))) {

        return pre;

    }

    if (pre.stream().anyMatch(context -> !(context.getData() instanceof FlowData))) {

        return pre;

    }

    FlowContext<I> baseContext = pre.get(0);

    FlowData mergedFlowData = mergeFlowData(pre);

    return Collections.singletonList(baseContext.convertData(ObjectUtils.cast(mergedFlowData), baseContext.getId()));

}

  

private FlowData mergeFlowData(List<FlowContext<I>> pre) {

    FlowData first = ObjectUtils.cast(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).map(ObjectUtils::<FlowData>cast).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();

}

修改说明

这一段是本次修复里最关键的一层。

前面的 request 修复只能保证“多对一节点能捞到多条上游输入”，但还不能保证“节点执行时真的把这些输入当作一组来消费”。

原问题正是出在这里：

request 阶段已经能看到两条输入
但普通节点的 MAPPING 或 LLM 节点底层的 PRODUCING 执行时，仍然按单条 context 分别处理
于是某一条 context 里只包含自己这一侧的数据，另一侧引用自然就是 null

所以 onProcess(...) 现在增加了一步：

先调用 mergeProcessInputs(pre)
再把 processInputs 交给真正的处理器执行

mergeProcessInputs(...) 当前只在以下条件下生效：

fanInMode == ALL
输入条数大于 1
当前处理模式属于 MAPPING、FLATMAPPING 或 PRODUCING
输入 data 都是 FlowData

其中把 PRODUCING 也纳入是这次后续补充的关键，因为普通 llmNodeState 实际走的是 FlowStateNode -> Node(... this::stateProduce ...) -> To.PRODUCING 这条链路，而不是 MAPPING。

mergeFlowData(...) 则负责把多条 FlowContext 中的三类核心数据拼起来：

businessData
contextData
passData

合并时使用 FlowUtil.mergeMaps(...)，这样下游节点最终看到的是一个完整的、同时包含多个上游输出的 FlowData。

最终效果是：对于需要同时引用多个上游节点输出的 LLM 节点、代码节点或普通 map 类节点，执行器真正拿到的是“合并后的完整输入”，而不是某一侧的单独 context。

6. To.java: 各处理模式接入新的 request 逻辑

PRODUCING {

    @Override

    protected <T1, R1> List<FlowContext<T1>> requestAll(To<T1, R1> to) {

        return to.repo.requestProducingContext(to.streamId,

                to.froms.stream().map(Identity::getId).collect(Collectors.toList()),

                to.requestFilter(to.postFilter()));

    }

},

  

MAPPING {

    @Override

    protected <T1, R1> List<FlowContext<T1>> requestAll(To<T1, R1> to) {

        return to.repo.requestMappingContext(to.streamId,

                to.froms.stream().map(Identity::getId).collect(Collectors.toList()),

                to.requestFilter(to.defaultFilter()),

                to.validator);

    }

},

  

private <T1, R1> List<FlowContext<T1>> filterReady(To<T1, R1> to, List<FlowContext<T1>> pre) {

    List<FlowContext<T1>> grouped = to.filterReadyByFanIn(pre);

    return to.markReady(grouped);

}

修改说明

这一段解决的是“不同处理模式下 request 路径不一致”的问题。

如果只在某一个模式里改 request 过滤，其它模式仍然会沿用旧逻辑，那么：

reduce 节点可能正确
producing 节点仍然不对
mapping 节点还是只能捞到一边

所以现在统一改成：

PRODUCING.requestAll(...) 调用 requestFilter(to.postFilter())
MAPPING.requestAll(...) 调用 requestFilter(to.defaultFilter())
FLATMAPPING、REDUCING 继续复用前者逻辑

这样不同处理模式都会在 ALL 场景下走同一套“完整汇聚组筛选”策略。

filterReady(...) 也被调整为先 filterReadyByFanIn(...) 再 markReady(...)，避免旧逻辑里“先标 ready、后分组”造成的状态不一致问题。

7. To.java: 处理线程退出前的并发冲突补偿

private <T1, R1> void handleProcessConcurrentConflict(To<T1, R1> to) {

    List<FlowContext<T1>> pending = requestAll(to).stream()

            .filter(context -> !context.getParallelMode().equals(ParallelMode.EITHER.name())

                    || context.isJoined() || !to.isParallelJoined(context))

            .collect(Collectors.toList());

    if (CollectionUtils.isEmpty(pending) || to.inParallelMode(pending)) {

        return;

    }

    List<FlowContext<T1>> ready = filterReady(to, pending);

    if (CollectionUtils.isEmpty(ready)) {

        return;

    }

    LOG.info("[{}] process thread conflict happens for stream-id: {}, node-id: {}",

            to.getThreadName(To.PROCESS_T_NAME_PREFIX), to.streamId, to.id);

    to.accept(ProcessType.PROCESS, pending);

}

修改说明

这一段解决的是处理线程退出窗口期的竞态问题。

原场景是：

处理线程判断当前没有 ready 数据，准备退出
就在退出前后，新数据恰好写入边上
外部又因为线程标记未及时变化，没有重新拉起处理

结果就是：新数据挂在边上，但没有线程再去消费。

handleProcessConcurrentConflict(...) 的做法是：

在线程准备退出时，再检查一次当前 pending
对这些 pending 再走一遍 ready 判定
只有当确实存在 ready 数据时，才重新触发 accept(ProcessType.PROCESS, pending)

这个改动的重点在于“只对真正 ready 的数据重新拉起处理”，避免之前那种无条件自唤醒导致的空转或死循环。

🧪 验证变更 / Verifying this Change

测试步骤 / Test Steps

测试覆盖 / Test Coverage

📸 截图 / Screenshots

✅ 贡献者检查清单 / Contributor Checklist

请确保你的 Pull Request 符合以下要求 / Please ensure your Pull Request meets the following requirements:

基本要求 / Basic Requirements:

确保有 GitHub Issue 对应这个变更（微小变更如错别字除外）/ Make sure there is a Github issue filed for the change (trivial changes like typos excluded)
你的 Pull Request 只解决一个 Issue，没有包含其他不相关的变更 / Your PR addresses just this issue, without pulling in other changes - one PR resolves one issue
PR 中的每个 commit 都有有意义的主题行和描述 / Each commit in the PR has a meaningful subject line and body

代码质量 / Code Quality:

我的代码遵循项目的代码规范 / My code follows the project's coding standards
我已经进行了自我代码审查 / I have performed a self-review of my code
我已经为复杂的代码添加了必要的注释 / I have commented my code, particularly in hard-to-understand areas

测试要求 / Testing Requirements:

我已经编写了必要的单元测试来验证逻辑正确性 / I have written necessary unit-tests to verify the logic correction
当存在跨模块依赖时，我尽量使用了 mock / I have used mocks when cross-module dependencies exist
基础检查通过：mvn -B clean package -Dmaven.test.skip=true，npm install --force && npm run build:pro / Basic checks pass
单元测试通过：mvn clean install / Unit tests pass

文档和兼容性 / Documentation and Compatibility:

我已经更新了相应的文档 / I have made corresponding changes to the documentation
如果有破坏性变更，我已经在 PR 描述中详细说明 / If there are breaking changes, I have documented them in detail
我已经考虑了向后兼容性 / I have considered backward compatibility

📋 附加信息 / Additional Notes

审查者注意事项 / Reviewer Notes:

同步

loveTsong · 2026-03-17T06:57:43Z

...lugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/domains/business/RunContext.java

        RunContext runContext = new RunContext(new HashMap<>(), context);
        runContext.putAllToBusiness(requestContext.getUserContext());
+
+//        runContext请求操作


无意义的注释，删除

loveTsong · 2026-03-17T06:58:27Z

...ce/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

    /**
     * 通过from.offer(data)而不是.offer(context)发起的数据会新增一个trace，这个trace会延续到flow end
     */
+


不应该存在空行

loveTsong · 2026-03-17T07:00:33Z

...ce/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

-        context.previous = this.previous;
        context.status = this.status;
        context.trans = this.trans;
-        context.id = id;


如果是copyContext的含义，不应该删除previous和id的赋值

loveTsong · 2026-03-17T07:03:45Z

...ce/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

+     */
+    public <R> FlowContext<R> convertData(R data) {
+        FlowContext<R> context = this.copyContext(data);
+        context.previous = this.id;


作为fork生成数据的previous不应该是他自己，应该是保持一致的

loveTsong · 2026-03-17T07:07:45Z

...ce/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

     */
    public <R> FlowContext<R> convertData(R data, String id) {
+        FlowContext<R> context = this.copyContext(data);
+        context.previous = this.previous;


看下copyContext整体调整后，这里应该不再需要为previous赋值

loveTsong · 2026-03-17T07:09:43Z

...main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/FlowDefinition.java

                    //  startNode不能出现在event的to属性, endNode不能出现在event的from属性
+
                    FlowNode toNode = nodeMap.get(event.getTo());
+


不合适的空行

loveTsong · 2026-03-19T02:50:00Z

frontend/src/pages/addFlow/components/addflow-header.tsx

              <img src={timeImg} />
            </span>
          }
+          <Button


不要增加按钮体现在界面上，功能未放开前，用户不需要感知这个存在，直接内部变量默认不开启

loveTsong · 2026-03-20T02:45:55Z

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

     */
    self.maxNumToLink = () => {
-        return 10;
+        return self.graph?.connectionLimitDisabled ? 100 : 10;


这个是聚合连线的数量，当前可以默认为10个，先不要放到100这个量级

loveTsong · 2026-03-20T02:46:09Z

agent-flow/src/components/variableAggregation/variableAggregationNodeState.jsx

   */
  self.maxNumToLink = () => {
-    return 10;
+    return self.graph?.connectionLimitDisabled ? 100 : 10;


这个是聚合连线的数量，当前可以默认为10个，先不要放到100这个量级

loveTsong · 2026-03-20T03:10:10Z

...w-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/From.java

+            return;
+        }
+        Set<String> forkedIds = forkedContexts.stream().map(FlowContext::getId).collect(Collectors.toSet());
+        List<FlowContext<I>> effectiveForkedContexts = matchedContexts.values()


这里为什么还要再计算一次effectiveForkedContexts ，理论上forkContexts都是effectiveForkedContexts 才对

loveTsong · 2026-03-20T03:12:09Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

        if (type == ProcessType.PROCESS && (processT == null || !processRunning)) {
            processRunning = true;
            String threadName = getThreadName(PROCESS_T_NAME_PREFIX);
+


变量和使用这种需要紧凑，不要隔空行

loveTsong · 2026-03-20T03:16:00Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

        this.froms.add(subscription); // 将该节点的from的event加入
+        if (!this.fanInModeConfigured) {
+            long fromCount = this.froms.stream().map(Identity::getId).distinct().count();
+            this.fanInMode = fromCount > 1 ? FanInMode.ALL : FanInMode.ANY;


看起来fanInModeConfigured没有必要存在，概念已经出现重叠了，直接能够配置FanInMode即可，ANY就是非聚合场景，ALL就是聚合的场景

loveTsong · 2026-03-20T03:29:36Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

                this.afterProcess(pre, new ArrayList<>());
                return;
            }
+            List<FlowContext<I>> processInputs = mergeProcessInputs(pre);


1. mergeProcessInputs的实现建议通过再FlowStateNode中构造Node时注入，提炼为一个通用机制，当前在To的内部直接识别FlowData处理，结构不太合理 2. merge的动作应该在beforeProcess前就做，后续都使用merge后的数据处理，这样保证数据处理的一致性和机制的明确性

loveTsong · 2026-03-20T03:30:37Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

        Optional.ofNullable(this.globalErrorHandler).ifPresent(handler -> handler.handle(ex, retryable, pre));
    }

+    private List<FlowContext<I>> mergeProcessInputs(List<FlowContext<I>> pre) {


如上述描述，这块逻辑考虑放到上层实现，to里只是提供这个机制调用

loveTsong · 2026-03-20T06:11:06Z

docker/dev-app-builder.sh

 echo "Copying shared libraries..."
 docker cp "$SHARED_DIR"/. app-builder-tmp:/opt/fit-framework/shared/

+docker exec app-builder-tmp bash -c "rm -f /opt/fit-framework/plugins/authentication-oauth2-client-1.0.0-SNAPSHOT.jar"


调试脚本，这个不要上传

loveTsong · 2026-03-20T06:11:33Z

frontend/src/locale/en_US.json

  "plsEnterNumber": "Enter a number",
  "debugRun": "Test Running",
+  "disableConnectionLimit": "Lift Link Limits",
+  "restoreConnectionLimit": "Restore Link Limits",


loveTsong · 2026-03-20T06:11:48Z

frontend/src/locale/zh_CN.json

  "createWorkflow": "创建工作流",
  "debugRun": "测试运行",
+  "disableConnectionLimit": "放开连线限制",
+  "restoreConnectionLimit": "恢复连线限制",


loveTsong · 2026-03-20T06:12:27Z

frontend/src/pages/addFlow/components/addflow-header.tsx

  const dispatch = useAppDispatch();
  const { t } = useTranslation();
-  const { handleDebugClick, workFlow, types, saveTime, updateAippCallBack } = props;
+  const {


这里不需要修改

loveTsong · 2026-03-20T06:14:07Z

frontend/src/pages/addFlow/index.tsx

+  const isConnectionLimitDisabled = typeof props.isConnectionLimitDisabled === 'boolean'
+    ? props.isConnectionLimitDisabled
+    : localConnectionLimitDisabled;
+  const toggleConnectionLimitDisabled = props.toggleConnectionLimitDisabled


无需该变更

loveTsong · 2026-03-20T06:14:35Z

frontend/src/pages/addFlow/index.tsx

            showDebug={showDebug}
            saveTime={saveTime}
            setShowDebug={setShowDebug}
+            toggleConnectionLimitDisabled={toggleConnectionLimitDisabled}


无需该变更

loveTsong · 2026-03-20T06:37:08Z

frontend/src/pages/components/header.tsx

            </span>
          }
        </div>
+        {showElsa && <Button


清理该按钮相关内容

loveTsong · 2026-03-20T06:39:08Z

frontend/src/pages/aippIndex/index.tsx

                  setSaveTime={setSaveTime}
                  showFlowChangeWarning={showFlowChangeWarning}
                  setShowFlowChangeWarning={setShowFlowChangeWarning}
+                  toggleConnectionLimitDisabled={() => setIsConnectionLimitDisabled((prev) => !prev)}


不需要在界面控制，变量写死控制即可

变量进行放开

loveTsong · 2026-04-02T02:59:14Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

+                || ProcessMode.PRODUCING.equals(this.processMode))) {
+            return pre;
+        }
+//        if (this.merger == null) {


不要上传调试注释的代码

loveTsong · 2026-04-02T03:02:51Z

...ice/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/nodes/Node.java

+     * @param inputType 输入数据类型，用于从 Registry 获取对应的 Merger
+     */
+    protected void autoInjectMerger(Class<?> inputType) {
+        Processors.Merger<T> registered = ObjectUtils.cast(MergerRegistry.getInstance().getMerger(inputType));


实现复杂了，没有必要再实现一个mergerRegistry，同时这样就限制死一个类型只有有一种处理方式。建议构造的时候直接通过类似lambda的方式传入就行，简单直接也够灵活。

loveTsong · 2026-04-02T03:03:37Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

+    }
+
+    private Processors.Merger<I> detectMerger(List<FlowContext<I>> pre) {
+        if (pre == null || pre.isEmpty()) {


调试后无用的代码需要及时清理

loveTsong · 2026-04-02T03:17:05Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

    private Processors.Validator<I> validator = (i, all) -> true;
+    private FanInMode fanInMode = FanInMode.ANY;
+    private Processors.Map<FlowContext<I>, String> mergeKeyGenerator = this::defaultMergeKey;
+    private Processors.Merger<I> merger;


新增的merge机制，需要增加测试用例。加在WaterFlowsTest中，参考其中节点操作，多连线后，给指定节点设置merge动作，验证merge的执行和该节点获取merge后的结果

loveTsong · 2026-04-09T02:30:31Z

存在用例不通过的情况，需要执行全量用例，检测对应问题

loveTsong · 2026-04-13T07:41:55Z

...service/src/test/java/modelengine/fit/waterflow/flowsengine/domain/flows/WaterFlowsTest.java

-            assertTestData(new TestData(0, 0, 13), output);
-        }
+//        @Test
+//        @DisplayName("流程实例condition节点以及match节点以及others节点流转逻辑")


不要直接注释代码，如果只是临时屏蔽，使用disable注解，同时在注解中编写具体原因和背景。另外，这种场景应该默认使用ANY模式运行，只有flowdefinitaion层才显式启用all, 这样更合适

loveTsong · 2026-04-13T07:45:39Z

.../java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContextPersistTest.java

-            assertEquals(data3.getBusinessData(), result.get(0).getBusinessData());
-        }
+//        @Test
+//        @DisplayName("测试带有condition节点流程实例持久化成功")


不要直接注释代码，使用disable注解，并在注解中描述具体的原因和背景。

loveTsong · 2026-04-13T07:48:53Z

frontend/src/pages/components/styles/header.scss

        margin-right: 5px;
      }
  }
+  .link-limit-btn-active {


需要清理

loveTsong · 2026-04-13T07:49:02Z

frontend/src/styles/workSpace.scss

    color: rgba(0, 0, 0, 0.25);
    border-color: #d9d9d9;
  }
+  .link-limit-btn-active {


需要清理

loveTsong · 2026-04-13T08:35:48Z

...st/java/modelengine/fit/waterflow/flowsengine/domain/flows/validators/FlowValidatorTest.java

            WaterflowParamException exception = assertThrows(WaterflowParamException.class,
                    () -> flowNodeValidator.validate(flowDefinition));
-            assertEquals(INVALID_START_NODE_EVENT_SIZE.getErrorCode(), exception.getCode());
+            assertEquals(INVALID_EVENT_CONFIG.getErrorCode(), exception.getCode());


行为已经改变，这么修改不合适，应该配置对应场景的流程，变成校验通过

loveTsong · 2026-04-13T08:35:54Z

...st/java/modelengine/fit/waterflow/flowsengine/domain/flows/validators/FlowValidatorTest.java

            WaterflowParamException exception = assertThrows(WaterflowParamException.class,
                    () -> flowNodeValidator.validate(flowDefinition));
-            assertEquals(INVALID_STATE_NODE_EVENT_SIZE.getErrorCode(), exception.getCode());
+            assertEquals( INVALID_EVENT_CONFIG.getErrorCode(), exception.getCode());


行为已经改变，这么修改不合适，应该配置对应场景的流程，变成校验通过

loveTsong · 2026-04-14T06:29:01Z

...main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowNode.java

    public void subscribe(String streamId, FlowEnv flowEnv, FlowNode toNode, FlowEvent event) {
+        FitStream.Subscriber<FlowData, FlowData> toSubscriber = getTo(streamId, flowEnv.getRepo(),
+                flowEnv.getMessenger(), flowEnv.getLocks(), toNode);
+        toSubscriber.setFromFlowDefinition(true);


命名太生硬不合适，直接设置模式就行

loveTsong · 2026-04-14T06:35:31Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java


    private Processors.Validator<I> validator = (i, all) -> true;
    private FanInMode fanInMode = FanInMode.ANY;
+    private boolean fromFlowDefinition = false;


直接设置fanInMode 就行，不需要fromFlowDefinition

loveTsong · 2026-04-14T06:35:33Z

...vice/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/FitStream.java

+     *
+     * @param fromFlowDefinition 是否来自流程定义
+     */
+    default void setFromFlowDefinition(boolean fromFlowDefinition) {


提供设置模式的接口即可，不需要再中转一层含义，同时flowDefition的概念为上层，不应该让下层感知这个概念

loveTsong · 2026-04-14T06:35:39Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

        this.froms.add(subscription); // 将该节点的from的event加入
+        if (this.fromFlowDefinition) {
            long fromCount = this.froms.stream().map(Identity::getId).distinct().count();
            this.fanInMode = fromCount > 1 ? FanInMode.ALL : FanInMode.ANY;


不用在这里动态判断，设置什么就是什么

loveTsong · 2026-04-14T06:38:54Z

...low-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java

+     *
+     * @param fromFlowDefinition 是否来自流程定义
+     */
+    public void setFromFlowDefinition(boolean fromFlowDefinition) {


替换为设置fanInMode

swfnotswift and others added 5 commits January 15, 2026 21:59

Merge pull request #1 from ModelEngine-Group/main

66993f7

同步

修改前端代码

6a6942e

修改后端代码，实现条件流程校验阶段对多节点的支持

3dd9244

修改后端代码，实现条件流程校验阶段对多节点的支持

9d7fc0f

修改前端代码，新增放开连线限制按钮

851f414

loveTsong reviewed Mar 17, 2026

View reviewed changes

loveTsong reviewed Mar 19, 2026

View reviewed changes

loveTsong reviewed Mar 20, 2026

View reviewed changes

qq_62395577 added 3 commits March 24, 2026 19:41

根据PR修复部分代码

35a7bf2

重构merge节点逻辑

793d668

重写前端页面，用户无法通过按钮放开连线逻辑，可通过elsa-stage中 connectionLimitDisabled: false,

aa2ba28

变量进行放开

loveTsong reviewed Apr 2, 2026

View reviewed changes

loveTsong changed the base branch from main to hzjh-main April 2, 2026 06:45

qq_62395577 added 2 commits April 7, 2026 16:05

根据修改意见进行更改，并且调整了merge的逻辑

d338eb7

补充测试

33558d1

loveTsong closed this Apr 8, 2026

loveTsong reopened this Apr 8, 2026

qq_62395577 added 6 commits April 9, 2026 17:37

debug修改1

0bdfdf6

bug修改

5c45e25

bug修改3

371c93c

测试修改

51c2854

修改2

24f4e71

修改校验测试文件

5dcc8e0

loveTsong reviewed Apr 13, 2026

View reviewed changes

qq_62395577 added 3 commits April 13, 2026 19:25

修改完成合法性校验

1d93d7b

修改all模式的注入机制

690b8aa

清楚按钮相关代码

b06d21c

loveTsong reviewed Apr 14, 2026

View reviewed changes

		// startNode不能出现在event的to属性, endNode不能出现在event的from属性

		FlowNode toNode = nodeMap.get(event.getTo());

Conversation

swfnotswift commented Mar 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🔗 相关问题 / Related Issue

📋 变更类型 / Type of Change

📝 变更目的 / Purpose of the Change

📋 主要变更 / Brief Changelog

1. FlowContext.java

修改说明

2. From.java

修改说明

3. To.java: fan-in 配置与 request 过滤

修改说明

4. To.java: 多输入 ready 判断与完整汇聚组选择

修改说明

5. To.java: 处理阶段合并多输入上下文

修改说明

6. To.java: 各处理模式接入新的 request 逻辑

修改说明

7. To.java: 处理线程退出前的并发冲突补偿

修改说明

🧪 验证变更 / Verifying this Change

测试步骤 / Test Steps

测试覆盖 / Test Coverage

📸 截图 / Screenshots

✅ 贡献者检查清单 / Contributor Checklist

📋 附加信息 / Additional Notes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

loveTsong Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

loveTsong Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

loveTsong Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

swfnotswift commented Mar 11, 2026 •

edited

Loading

loveTsong Mar 17, 2026 •

edited

Loading

loveTsong Mar 17, 2026 •

edited

Loading

loveTsong Mar 17, 2026 •

edited

Loading

loveTsong Apr 2, 2026 •

edited

Loading

loveTsong Apr 13, 2026 •

edited

Loading

loveTsong Apr 13, 2026 •

edited

Loading