[fel] demonstrate data anonymization with example

Author: loveTsong

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

@@ -57,6 +57,23 @@ public <O> AiMatchHappen<O, D, I, RF, F> match(Operators.Whether<I> whether,
                 node -> processor.process(new AiState<>(node, this.flow())).state), this.flow());
     }
 
+    /**
+     * 指定条件和处理器创建条件分支。
+     *
+     * @param whether 表示匹配条件的 {@link Operators.Whether}{@code <}{@link I}{@code >}。
+     * @param processor 表示分支处理器的 {@link Operators.Then}{@code <}{@link O}{@code , }{@link D}{@code ,
+     * }{@link I}{@code , }{@link RF}{@code , }{@link F}{@code >}。
+     * @param <O> 表示第一个条件分支指定的返回类型。
+     * @return 表示条件分支的 {@link AiMatchHappen}{@code <}{@link O}{@code , }{@link D}{@code ,
+     * }{@link I}{@code , }{@link RF}{@code , }{@link F}{@code >}。
+     * @throws IllegalArgumentException 当 {@code processor} 为 {@code null} 时。
+     */
+    public <O> AiWhenHappen<O, D, I, RF, F> when(Operators.Whether<I> whether,
+            Operators.Then<I, O> processor) {
+        Validation.notNull(processor, "Ai branch processor cannot be null.");
+        return new AiWhenHappen<>(this.conditions.when(whether, processor), this.flow());
+    }
+
     /**
      * 指定条件和对应的处理器创建条件跳转分支。
      *

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

@@ -0,0 +1,62 @@
+/*---------------------------------------------------------------------------------------------
+ *  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.fel.engine.activities;
+
+import modelengine.fel.engine.flows.AiFlow;
+import modelengine.fit.waterflow.domain.flow.Flow;
+import modelengine.fit.waterflow.domain.states.WhenHappen;
+import modelengine.fit.waterflow.domain.stream.operators.Operators;
+import modelengine.fitframework.inspection.Validation;
+
+/**
+ * Represents a conditional branch that matches when conditions in an AI processing flow.
+ * This class handles the branching logic when specific conditions are met in the workflow.
+ *
+ * @param <O> The output data type of the current node.
+ * @param <D> The initial data type of the containing flow.
+ * @param <I> The input parameter data type.
+ * @param <RF> The internal flow type, extending {@link Flow}{@code <D>}.
+ * @param <F> The AI flow type, extending {@link AiFlow}{@code <D, RF>}.
+ *
+ * @author 宋永坦
+ * @since 2025-06-12
+ */
+public class AiWhenHappen<O, D, I, RF extends Flow<D>, F extends AiFlow<D, RF>> {
+    private final WhenHappen<O, D, I, RF> matchHappen;
+
+    private final F flow;
+
+    public AiWhenHappen(WhenHappen<O, D, I, RF> matchHappen, F flow) {
+        this.matchHappen = Validation.notNull(matchHappen, "WhenHappen cannot be null.");
+        this.flow = Validation.notNull(flow, "Flow cannot be null.");
+    }
+
+    /**
+     * Creates a conditional branch with the specified predicate and handler.
+     *
+     * @param whether The condition predicate that determines branch activation.
+     * @param processor The transformation handler to execute when condition is met.
+     * @return A new {@link AiWhenHappen} instance representing the conditional branch.
+     * @throws IllegalArgumentException if processor is null.
+     */
+    public AiWhenHappen<O, D, I, RF, F> when(Operators.Whether<I> whether, Operators.Then<I, O> processor) {
+        Validation.notNull(processor, "Ai branch processor cannot be null.");
+        return new AiWhenHappen<>(this.matchHappen.when(whether, processor), this.flow);
+    }
+
+    /**
+     * Provides a default processing logic and terminates the conditional node.
+     *
+     * @param processor The handler to process unmatched inputs.
+     * @return An {@link AiState} representing the terminal node of the conditional flow.
+     * @throws IllegalArgumentException if processor is null
+     */
+    public AiState<O, D, O, RF, F> others(Operators.Then<I, O> processor) {
+        Validation.notNull(processor, "Ai branch processor cannot be null.");
+        return new AiState<>(this.matchHappen.others(processor), this.flow);
+    }
+}

framework/fel/java/fel-flow/src/test/java/modelengine/fel/engine/AiFlowCaseTest.java

@@ -16,6 +16,7 @@
 import modelengine.fel.engine.operators.models.ChatFlowModel;
 import modelengine.fel.engine.operators.prompts.Prompts;
 import modelengine.fit.waterflow.domain.context.FlowSession;
+import modelengine.fit.waterflow.domain.context.StateContext;
 import modelengine.fit.waterflow.domain.utils.SleepUtil;
 import modelengine.fitframework.flowable.Choir;
 
@@ -33,6 +34,82 @@
  * @since 2025-06-11
  */
 public class AiFlowCaseTest {
+    @Nested
+    class DesensitizeCase {
+        private final ChatFlowModel model = new ChatFlowModel((prompt, chatOption) -> Choir.create(emitter -> {
+            emitter.emit(new AiMessage("<think>"));
+            for (int i = 0; i < 10; i++) {
+                emitter.emit(new AiMessage(String.valueOf(i)));
+                SleepUtil.sleep(100);
+            }
+            emitter.emit(new AiMessage("</think>"));
+            for (int i = 100; i < 110; i++) {
+                emitter.emit(new AiMessage(String.valueOf(i)));
+                SleepUtil.sleep(100);
+            }
+            emitter.complete();
+        }), ChatOption.custom().model("modelName").stream(true).build());
+
+        private final AiProcessFlow<Tip, String> flow = AiFlows.<Tip>create()
+                .prompt(Prompts.human("{{0}}"))
+                .generate(model)
+                .map(this::classic)
+                .conditions()
+                .when(chunk -> chunk.isThinkContent, input -> input)
+                .others(input -> {
+                    this.log(input);
+                    return input;
+                })
+                .map(this::mockDesensitize)
+                .close();
+
+        @Test
+        void run() {
+            AtomicInteger counter = new AtomicInteger(0);
+            long startTime = System.currentTimeMillis();
+            System.out.printf("time:%s, start.\n", startTime);
+            ConverseLatch<String> result = flow.converse(new FlowSession(true)).doOnConsume(answer -> {
+                System.out.printf("time:%s, chunk=%s\n", System.currentTimeMillis(), answer);
+                counter.incrementAndGet();
+            }).offer(Tip.fromArray("hi"));
+            result.await();
+            System.out.printf("time:%s, cost=%s\n", System.currentTimeMillis(), System.currentTimeMillis() - startTime);
+            Assertions.assertEquals(22, counter.get());
+        }
+
+        private Chunk classic(ChatMessage message, StateContext ctx) {
+            if (message.text().trim().equals("<think>")) {
+                ctx.setState("isThinking", true);
+                return new Chunk(true, message.text());
+            }
+            if (message.text().trim().equals("</think>")) {
+                ctx.setState("isThinking", false);
+                return new Chunk(true, message.text());
+            }
+            if (Boolean.TRUE.equals(ctx.getState("isThinking"))) {
+                return new Chunk(true, message.text());
+            }
+            return new Chunk(false, message.text());
+        }
+
+        private String mockDesensitize(Chunk chunk) {
+            return chunk.content.replace("3", "*");
+        }
+
+        private void log(Chunk chunk) {
+            System.out.println("log content:" + chunk.content);
+        }
+
+        private static class Chunk {
+            private final boolean isThinkContent;
+            private final String content;
+
+            private Chunk(boolean isThinkContent, String content) {this.isThinkContent = isThinkContent;
+                this.content = content;
+            }
+        }
+    }
+
     /**
      * Simulates a backpressure scenario where:
      * <ol>