fix(framework): 修复 Fork.join() 并发空指针异常

framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java

@@ -118,6 +118,9 @@ public Tip addAll(Map<String, Content> args) {
      * @return 表示当前的 {@link Tip}。
      */
     public Tip merge(Tip other) {
+        if (other == null) {
+            return this;
+        }
         return this.addAll(other.values);
     }
 

framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java

@@ -592,10 +592,7 @@ public final AiState<Tip, D, Tip, RF, F> runnableParallel(Pattern<O, Tip>... pat
                     .orElseGet(() -> new AiParallel<>(this.start.parallel(), mineFlow).fork(branchProcessor));
         }
 
-        AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> {
-            acc.merge(data);
-            return acc;
-        });
+        AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> acc.merge(data));
         ((Processor<?, ?>) state.publisher()).displayAs("runnableParallel");
         return state;
     }

framework/fel/java/fel-flow/src/test/java/modelengine/fel/engine/operators/PatternTest.java

@@ -42,6 +42,7 @@
 import modelengine.fitframework.util.StringUtils;
 
 import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import java.util.Collection;
@@ -97,6 +98,23 @@ void shouldOkWhenAiFlowWithExampleSelector() {
         assertThat(converse.offer("1+2").await().text()).isEqualTo("2+2=4\n2+3=5\n1+2=");
     }
 
+    @RepeatedTest(1000)
+    @DisplayName("测试 RunnableParallel 并发稳定性")
+    void shouldStableWhenRunnableParallelUnderConcurrency() {
+        Example[] examples = {new DefaultExample("2+2", "4"), new DefaultExample("2+3", "5")};
+        Conversation<String, Prompt> converse = AiFlows.<String>create()
+                .runnableParallel(question(),
+                        fewShot(ExampleSelector.builder()
+                                .template("{{q}}={{a}}", "q", "a")
+                                .delimiter("\n")
+                                .example(examples)
+                                .build()))
+                .prompt(Prompts.human("{{examples}}\n{{question}}="))
+                .close()
+                .converse();
+        assertThat(converse.offer("1+2").await().text()).isEqualTo("2+2=4\n2+3=5\n1+2=");
+    }
+
     @Test
     @DisplayName("测试 Retriever")
     void shouldOkWhenAiFlowWithRetriever() {

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/Fork.java

@@ -93,7 +93,13 @@ public synchronized R process(FlowContext<O> input) {
                         acc = Tuple.from((R) "", 0);
                     }
                 }
-                acc = Tuple.from(processor.process(acc.first(), input.getData()), acc.second() + 1);
+
+                O inputData = input.getData();
+                if (inputData == null) {
+                    return null;
+                }
+                R processedResult = processor.process(acc.first(), inputData);
+                acc = Tuple.from(processedResult, acc.second() + 1);
                 accs.put(key, acc);
 
                 if (acc.second() == forkNumber.get()) {

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/To.java

@@ -937,7 +937,9 @@ public <T1, R1> List<FlowContext<R1>> process(To<T1, R1> to, List<FlowContext<T1
                             nextSession.getWindow().complete();
                         }
                     } else {
-                        peekedToken.finishConsume();
+                        if (peekedToken != null) {
+                            peekedToken.finishConsume();
+                        }
                         if (window.isDone()) {
                             window.tryFinish();
                         }