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

[swfnotswift]

🔗 相关问题 / 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: