extensions();
/**
diff --git a/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/tool/support/DefaultToolCallChunk.java b/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/tool/support/DefaultToolCallChunk.java
new file mode 100644
index 00000000..f3005303
--- /dev/null
+++ b/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/tool/support/DefaultToolCallChunk.java
@@ -0,0 +1,71 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.core.tool.support;
+
+import modelengine.fel.core.tool.ToolCall;
+import modelengine.fel.core.tool.ToolCallChunk;
+import modelengine.fitframework.inspection.Validation;
+import modelengine.fitframework.util.LazyLoader;
+import modelengine.fitframework.util.ObjectUtils;
+import modelengine.fitframework.util.StringUtils;
+
+/**
+ * 表示工具调用请求的实体片段默认实现。
+ *
+ * 该实现不保证流式片段聚合的线程安全，需要外部使用方保证线程安全。
+ *
+ *
+ * @author 刘信宏
+ * @since 2024-12-23
+ */
+public class DefaultToolCallChunk implements ToolCallChunk {
+ private final String id;
+ private final String name;
+ private final String arguments;
+ private final LazyLoader argumentsBuffer;
+
+ /**
+ * 使用 {@link ToolCall} 构造一个新的 {@link ToolCallChunk}。
+ *
+ * @param toolCall 表示工具调用的 {@link ToolCall}。
+ */
+ public DefaultToolCallChunk(ToolCall toolCall) {
+ Validation.notNull(toolCall, "The tool call cannot be null.");
+ this.id = Validation.notNull(toolCall.id(), "The tool call id cannot be null.");
+ this.name = toolCall.name();
+ this.arguments = toolCall.arguments();
+ this.argumentsBuffer =
+ new LazyLoader<>(() -> new StringBuilder(ObjectUtils.nullIf(this.arguments, StringUtils.EMPTY)));
+ }
+
+ @Override
+ public String id() {
+ return this.id;
+ }
+
+ @Override
+ public Integer index() {
+ // 工具调用的片段不需要index字段。
+ return null;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ @Override
+ public String arguments() {
+ return this.argumentsBuffer.get().toString();
+ }
+
+ @Override
+ public void merge(ToolCall toolCall) {
+ Validation.notNull(toolCall, "The tool call cannot be null.");
+ this.argumentsBuffer.get().append(ObjectUtils.nullIf(toolCall.arguments(), StringUtils.EMPTY));
+ }
+}
diff --git a/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java b/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java
index 7a1c2d60..c5a51fa4 100644
--- a/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java
+++ b/framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java
@@ -25,6 +25,11 @@ public class Tip {
private final Map values = new HashMap<>();
private int index = 0;
+ @Override
+ public String toString() {
+ return this.values.toString();
+ }
+
/**
* 从键值对创建 {@link Tip} 的实例。
*
diff --git a/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RerankDocumentProcessorTest.java b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RerankDocumentProcessorTest.java
new file mode 100644
index 00000000..8a08a5e7
--- /dev/null
+++ b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RerankDocumentProcessorTest.java
@@ -0,0 +1,108 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.core.document.support;
+
+import static modelengine.fel.core.document.support.TestRerankModelController.FAIL_ENDPOINT;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+import modelengine.fel.core.document.Document;
+import modelengine.fel.core.document.MeasurableDocument;
+import modelengine.fit.http.client.HttpClassicClientFactory;
+import modelengine.fitframework.annotation.Fit;
+import modelengine.fitframework.exception.FitException;
+import modelengine.fitframework.test.annotation.MvcTest;
+import modelengine.fitframework.test.domain.mvc.MockMvc;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * ReRank 客户端服务测试。
+ *
+ * @author 马朝阳
+ * @since 2024-09-14
+ */
+@MvcTest(classes = TestRerankModelController.class)
+public class RerankDocumentProcessorTest {
+ private static final String[] DOCS = new String[] {"Burgers", "Carson", "Shanghai", "Beijing", "Test"};
+
+ private RerankDocumentProcessor client;
+
+ @Fit
+ private HttpClassicClientFactory httpClientFactory;
+
+ @Fit
+ private MockMvc mockMvc;
+
+ @BeforeEach
+ public void setUp() {
+ this.client = new RerankDocumentProcessor(httpClientFactory,
+ RerankOption.custom()
+ .baseUri("http://localhost:" + mockMvc.getPort())
+ .model("rerank1")
+ .query("What is the capital of the united states?")
+ .topN(3)
+ .build());
+ }
+
+ @Test
+ @DisplayName("测试 Rerank 接口调用响应成功")
+ public void testWhenCallRerankModelThenSuccess() {
+ List texts = Arrays.asList(DOCS[3], DOCS[4], DOCS[0]);
+ List scores = Arrays.asList(0.999071, 0.7867867, 0.32713068);
+ List docs = this.client.process(this.getRequest());
+ assertThat(docs).extracting(MeasurableDocument::text).isEqualTo(texts);
+ assertThat(docs).extracting(MeasurableDocument::score).isEqualTo(scores);
+ }
+
+ @Test
+ @DisplayName("测试 Rerank 接口调用响应异常")
+ public void testWhenCallRerankModelThenResponseException() {
+ RerankDocumentProcessor client1 = new RerankDocumentProcessor(httpClientFactory,
+ RerankOption.custom().baseUri("http://localhost:" + mockMvc.getPort() + FAIL_ENDPOINT).build());
+ assertThatThrownBy(() -> client1.process(this.getRequest())).isInstanceOf(FitException.class);
+ }
+
+ @Test
+ @DisplayName("测试 Rerank 接口参数为空响应异常")
+ public void testWhenCallRerankModelNullParamThenResponseException() {
+ assertThatThrownBy(() -> new RerankDocumentProcessor(this.httpClientFactory, null)).isInstanceOf(
+ IllegalArgumentException.class);
+ assertThatThrownBy(() -> new RerankDocumentProcessor(null, RerankOption.custom().build())).isInstanceOf(
+ IllegalArgumentException.class);
+ }
+
+ @Test
+ @DisplayName("测试 Rerank 接口请求参数为空响应异常")
+ public void testWhenCallRerankModelNullRequestParamThenResponseException() {
+ assertThat(this.client.process(new ArrayList<>())).isEqualTo(Collections.emptyList());
+ assertThat(this.client.process(null)).isEqualTo(Collections.emptyList());
+ }
+
+ private List getRequest() {
+ List documents = new ArrayList<>();
+ Arrays.stream(DOCS)
+ .forEach(doc -> documents.add(new MeasurableDocument(Document.custom()
+ .text(doc)
+ .metadata(new HashMap<>())
+ .build(), -1)));
+ return documents;
+ }
+
+ private String getMockReRankResponseBody() {
+ return "{\"results\":[{\"index\":3,\"relevance_score\":0.999071},{\"index\":4,\"relevance_score\":0.7867867},"
+ + "{\"index\":0,\"relevance_score\":0.32713068}]}";
+ }
+}
\ No newline at end of file
diff --git a/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RrfPostProcessorTest.java b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RrfPostProcessorTest.java
new file mode 100644
index 00000000..1b6777c6
--- /dev/null
+++ b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/RrfPostProcessorTest.java
@@ -0,0 +1,90 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.core.document.support;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+import modelengine.fel.core.document.Document;
+import modelengine.fel.core.document.MeasurableDocument;
+import modelengine.fel.core.document.support.postprocessor.RrfPostProcessor;
+import modelengine.fel.core.document.support.postprocessor.RrfScoreStrategyEnum;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * RRF 算法测试。
+ *
+ * @author 马朝阳
+ * @since 2024-09-29
+ */
+public class RrfPostProcessorTest {
+ private static final String[] DOCS = new String[] {"A", "B", "C", "D", "E"};
+
+ @Test
+ @DisplayName("测试 RFF 算法最大值策略成功")
+ public void testWhenCallRRFMaxThenSuccess() {
+ RrfPostProcessor rrf = new RrfPostProcessor();
+ List process = rrf.process(getDocumentList());
+ assertThat(process).map(MeasurableDocument::score).containsExactly(0.94, 0.69, 0.36, 0.52, 0.32);
+ assertThat(process).map(MeasurableDocument::id).containsExactly("1", "4", "2", "5", "3");
+ }
+
+ @Test
+ @DisplayName("测试 RFF 算法平均值策略成功")
+ public void testWhenCallRRFAvgThenSuccess() {
+ RrfPostProcessor rrf = new RrfPostProcessor(RrfScoreStrategyEnum.AVG);
+ List process = rrf.process(getDocumentList());
+ assertThat(process).map(MeasurableDocument::score).containsExactly(0.84, 0.655, 0.36, 0.52, 0.32);
+ assertThat(process).map(MeasurableDocument::id).containsExactly("1", "4", "2", "5", "3");
+ }
+
+ @Test
+ @DisplayName("测试 RFF 算法倒数系数")
+ public void testWhenCallRRFFactorThenSuccess() {
+ RrfPostProcessor rrf = new RrfPostProcessor(RrfScoreStrategyEnum.AVG, 100);
+ List process = rrf.process(getDocumentList());
+ assertThat(process).map(MeasurableDocument::score).containsExactly(0.84, 0.655, 0.36, 0.52, 0.32);
+ assertThat(process).map(MeasurableDocument::id).containsExactly("1", "4", "2", "5", "3");
+ }
+
+ @Test
+ @DisplayName("测试 RFF 算法策略失败")
+ public void testWhenCallRRFArgNullThenFail() {
+ assertThatThrownBy(() -> new RrfPostProcessor(null, 60)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new RrfPostProcessor(RrfScoreStrategyEnum.AVG, -1)).isInstanceOf(
+ IllegalArgumentException.class);
+ }
+
+ private List getDocumentList() {
+ List res = new ArrayList<>();
+ res.addAll(getGroup("1", new int[] {1, 3, 4}, new double[] {0.74, 0.32, 0.69}));
+ res.addAll(getGroup("2", new int[] {1, 5}, new double[] {0.94, 0.52}));
+ res.addAll(getGroup("3", new int[] {2, 4}, new double[] {0.36, 0.62}));
+
+ return res;
+ }
+
+ private List getGroup(String groupId, int[] ids, double[] scores) {
+ List documents = new ArrayList<>();
+ int scoreId = 0;
+ for (int id : ids) {
+ documents.add(new MeasurableDocument(Document.custom()
+ .text(DOCS[id - 1])
+ .id(String.valueOf(id))
+ .metadata(new HashMap<>())
+ .build(), scores[scoreId], groupId));
+ scoreId++;
+ }
+ return documents;
+ }
+}
diff --git a/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/TestRerankModelController.java b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/TestRerankModelController.java
new file mode 100644
index 00000000..74e1fd23
--- /dev/null
+++ b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/document/support/TestRerankModelController.java
@@ -0,0 +1,55 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.core.document.support;
+
+import modelengine.fit.http.annotation.PostMapping;
+import modelengine.fitframework.annotation.Component;
+import modelengine.fitframework.serialization.ObjectSerializer;
+
+/**
+ * 表示测试使用的 Rerank 接口。
+ *
+ * @author 马朝阳
+ * @since 2024-09-27
+ */
+@Component
+public class TestRerankModelController {
+ /**
+ * Rerank 接口失败调用端口。
+ */
+ public static final String FAIL_ENDPOINT = "/fail";
+
+ private final ObjectSerializer serializer;
+
+ TestRerankModelController(ObjectSerializer serializer) {
+ this.serializer = serializer;
+ }
+
+ /**
+ * 测试成功用 Rerank 接口。
+ *
+ * @return 表示流式返回结果的 {@link String}。
+ */
+ @PostMapping(RerankApi.RERANK_ENDPOINT)
+ public RerankResponse rerankSuccess() {
+ String json =
+ "{\"results\":[{\"index\":3,\"relevance_score\":0.999071},{\"index\":4,\"relevance_score\":0.7867867},"
+ + "{\"index\":0,\"relevance_score\":0.32713068}]}";
+ return this.serializer.deserialize(json, RerankResponse.class);
+ }
+
+ /**
+ * 测试用 Rerank 接口。
+ *
+ * @return 表示流式返回结果的 {@link String}。
+ */
+ @PostMapping(FAIL_ENDPOINT + RerankApi.RERANK_ENDPOINT)
+ public RerankResponse rerankFail() {
+ String json = "wrong json";
+ return this.serializer.deserialize(json, RerankResponse.class);
+ }
+}
diff --git a/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/splitter/support/SimpleTokenizer.java b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/splitter/support/SimpleTokenizer.java
index b4489b55..48a29378 100644
--- a/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/splitter/support/SimpleTokenizer.java
+++ b/framework/fel/java/fel-core/src/test/java/modelengine/fel/core/splitter/support/SimpleTokenizer.java
@@ -6,6 +6,8 @@
package modelengine.fel.core.splitter.support;
+import static modelengine.fitframework.inspection.Validation.notNull;
+
import modelengine.fel.core.tokenizer.Tokenizer;
import java.util.ArrayList;
@@ -36,4 +38,10 @@ public String decode(List tokens) {
}
return new String(charArray);
}
+
+ @Override
+ public int countToken(String text) {
+ notNull(text, "Text cannot be null.");
+ return text.length();
+ }
}
\ No newline at end of file
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java
index 8fc2f794..0bd7962c 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java
@@ -12,6 +12,7 @@
import modelengine.fel.core.document.Content;
import modelengine.fel.core.document.Document;
import modelengine.fel.core.document.Measurable;
+import modelengine.fel.core.model.BlockModel;
import modelengine.fel.core.pattern.Parser;
import modelengine.fel.core.pattern.Pattern;
import modelengine.fel.core.pattern.PostProcessor;
@@ -28,7 +29,9 @@
import modelengine.fel.engine.flows.Conversation;
import modelengine.fel.engine.operators.models.FlowModel;
import modelengine.fel.engine.operators.patterns.AbstractFlowPattern;
+import modelengine.fel.engine.operators.patterns.FlowNodeSupportable;
import modelengine.fel.engine.operators.patterns.FlowPattern;
+import modelengine.fel.engine.operators.patterns.FlowSupportable;
import modelengine.fel.engine.operators.patterns.SimpleFlowPattern;
import modelengine.fel.engine.operators.prompts.PromptTemplate;
import modelengine.fel.engine.util.AiFlowSession;
@@ -46,7 +49,6 @@
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.function.Supplier;
@@ -419,17 +421,30 @@ public AiState parse(Parser parser) {
*/
public AiState delegate(Pattern pattern) {
Validation.notNull(pattern, "Pattern operator cannot be null.");
- FlowPattern flowPattern = this.castFlowPattern(pattern);
+ return this.delegate(new SimpleFlowPattern<>(pattern));
+ }
+
+ /**
+ * 将数据委托给 {@link FlowPattern}{@code <}{@link O}{@code , }{@link R}{@code >}
+ * 处理，然后自身放弃处理数据。处理后的数据会发送回该节点，作为该节点的处理结果。
+ *
+ * @param pattern 表示异步委托单元的 {@link FlowPattern}{@code <}{@link O}{@code , }{@link R}{@code >}。
+ * @param 表示委托节点的输出数据类型。
+ * @return 表示委托节点的 {@link AiState}{@code <}{@link R}{@code , }{@link D}{@code , }{@link O}{@code ,
+ * }{@link RF}{@code , }{@link F}{@code >}。
+ * @throws IllegalArgumentException 当 {@code pattern} 为 {@code null} 时。
+ */
+ public AiState delegate(FlowPattern pattern) {
+ Validation.notNull(pattern, "Pattern operator cannot be null.");
Processor orProcessor = this.publisher().flatMap(input -> {
- FlowEmitter cachedEmitter = FlowEmitter.from(flowPattern);
- AiFlowSession.applyPattern(flowPattern, input.getData(), input.getSession());
- return Flows.source(cachedEmitter);
+ FlowEmitter emitter = AiFlowSession.applyPattern(pattern, input.getData(), input.getSession());
+ return Flows.source(emitter);
}, null);
this.displayPatternProcessor(pattern, orProcessor);
return new AiState<>(new State<>(orProcessor, this.flow().origin()), this.flow());
}
- private void displayPatternProcessor(Pattern pattern, Processor processor) {
+ private void displayPatternProcessor(FlowPattern pattern, Processor processor) {
if (pattern instanceof AbstractFlowPattern) {
Flow originFlow = ObjectUtils.>cast(pattern).origin();
processor.displayAs("delegate to flow", originFlow, originFlow.start().getId());
@@ -477,13 +492,7 @@ public AiState delegate(Operators.ProcessMap operator)
*/
public AiState delegate(AiProcessFlow aiFlow) {
Validation.notNull(aiFlow, "Flow cannot be null.");
- Processor processor = this.publisher().map(input -> {
- aiFlow.converse(input.getSession()).offer(input.getData());
- return (R) null;
- }, null).displayAs("delegate to flow", aiFlow.origin(), aiFlow.origin().start().getId());
- AiState state = new AiState<>(new State<>(processor, this.flow().origin()), this.flow());
- state.offer(aiFlow);
- return state;
+ return this.delegate(new FlowSupportable<>(aiFlow));
}
/**
@@ -503,14 +512,7 @@ public AiState delegate(AiProcessFlow aiFlow) {
public AiState delegate(AiProcessFlow aiFlow, String nodeId) {
Validation.notNull(aiFlow, "Flow cannot be null.");
Validation.notBlank(nodeId, "Node id cannot be blank.");
- Processor processor = this.publisher().map(input -> {
- aiFlow.converse(input.getSession()).offer(nodeId, Collections.singletonList(input.getData()));
- return (R) null;
- }, null).displayAs("delegate to node", aiFlow.origin(), nodeId);
-
- AiState state = new AiState<>(new State<>(processor, this.flow().origin()), this.flow());
- state.offer(aiFlow);
- return state;
+ return this.delegate(new FlowNodeSupportable<>(aiFlow, nodeId));
}
/**
@@ -531,6 +533,22 @@ public final AiState prompt(PromptTemplate... templates)
}, null).displayAs("prompt"), this.flow().origin()), this.flow());
}
+ /**
+ * 生成大模型阻塞调用节点。
+ *
+ * @param model 表示模型算子实现的 {@link BlockModel}{@code <}{@link M}{@code >}。
+ * @param 表示模型节点的输入数据类型。
+ * @return 表示大模型阻塞调用节点的 {@link AiState}{@code <}{@link ChatMessage}{@code , }{@link D}{@code ,
+ * }{@link O}{@code , }{@link RF}{@code , }{@link F}{@code >}。
+ * @throws IllegalArgumentException 当 {@code model} 为 {@code null} 时。
+ */
+ public AiState generate(BlockModel model) {
+ Validation.notNull(model, "Model operator cannot be null.");
+ return new AiState<>(new State<>(this.publisher()
+ .map(input -> AiFlowSession.applyPattern(model, input.getData(), input.getSession()), null)
+ .displayAs("generate"), this.flow().origin()), this.flow());
+ }
+
/**
* 生成大模型流式调用节点。
*
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiState.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiState.java
index bb697ed5..1f175ac5 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiState.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiState.java
@@ -103,9 +103,12 @@ public Publisher publisher() {
@Override
public void register(EmitterListener handler) {
- if (handler != null) {
- this.state.register(handler);
- }
+ this.state.register(handler);
+ }
+
+ @Override
+ public void unregister(EmitterListener listener) {
+ this.state.unregister(listener);
}
@Override
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiFlows.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiFlows.java
index fff32fc5..49ace2aa 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiFlows.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiFlows.java
@@ -8,6 +8,8 @@
import modelengine.fel.engine.activities.AiDataStart;
import modelengine.fel.engine.activities.AiStart;
+import modelengine.fit.waterflow.domain.context.FlowSession;
+import modelengine.fit.waterflow.domain.emitters.Emitter;
import modelengine.fit.waterflow.domain.flow.Flows;
import modelengine.fit.waterflow.domain.flow.ProcessFlow;
import modelengine.fit.waterflow.domain.states.Start;
@@ -56,4 +58,16 @@ public static AiDataStart flux(D... data) {
AiStart, AiProcessFlow> start = AiFlows.create();
return new AiDataStart<>(start, data);
}
+
+ /**
+ * 通过指定的发射源来构造一个数据前置流。
+ *
+ * @param emitter 表示数据源的 {@link Emitter}{@code <}{@link D}{@code , }{@link FlowSession}{@code >}。
+ * @param 表示数据类型。
+ * @return 表示数据前置流的 {@link AiDataStart}{@code <}{@link D}{@code , }{@link D}{@code , }{@link D}{@code >}。
+ */
+ public static AiDataStart source(Emitter emitter) {
+ AiStart, AiProcessFlow> start = AiFlows.create();
+ return new AiDataStart<>(start, emitter);
+ }
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiProcessFlow.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiProcessFlow.java
index 40e5abd9..c7d132e9 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiProcessFlow.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/AiProcessFlow.java
@@ -16,6 +16,9 @@
import modelengine.fit.waterflow.domain.stream.reactive.Publisher;
import modelengine.fitframework.util.ObjectUtils;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* AI 数据处理流程，在 {@link AiFlow} 的基础上增加流程间的数据流转能力，并对外提供对话语义。
*
@@ -26,6 +29,9 @@
*/
public class AiProcessFlow extends AiFlow>
implements EmitterListener, Emitter {
+ private final Map, EmitterListener> listeners =
+ new ConcurrentHashMap<>();
+
public AiProcessFlow(ProcessFlow flow) {
super(flow);
}
@@ -38,7 +44,18 @@ public void handle(D data, FlowSession session) {
@Override
public void register(EmitterListener listener) {
if (listener != null) {
- this.origin().register((data, token) -> listener.handle(ObjectUtils.cast(data), new FlowSession(token)));
+ EmitterListener wrapperHandler =
+ (data, session) -> listener.handle(ObjectUtils.cast(data), session);
+ this.listeners.put(listener, wrapperHandler);
+ this.origin().register(wrapperHandler);
+ }
+ }
+
+ @Override
+ public void unregister(EmitterListener listener) {
+ EmitterListener target = this.listeners.remove(listener);
+ if (target != null) {
+ this.origin().unregister(target);
}
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/Conversation.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/Conversation.java
index 1a741843..48703e5e 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/Conversation.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/flows/Conversation.java
@@ -6,10 +6,13 @@
package modelengine.fel.engine.flows;
+import modelengine.fel.core.chat.ChatMessage;
import modelengine.fel.core.chat.ChatOption;
import modelengine.fel.core.memory.Memory;
import modelengine.fel.engine.activities.AiStart;
import modelengine.fel.engine.activities.FlowCallBack;
+import modelengine.fel.engine.operators.models.StreamingConsumer;
+import modelengine.fel.engine.operators.sources.Source;
import modelengine.fel.engine.util.StateKey;
import modelengine.fit.waterflow.domain.context.FlowSession;
import modelengine.fit.waterflow.domain.stream.operators.Operators;
@@ -45,7 +48,7 @@ public class Conversation {
public Conversation(AiProcessFlow flow, FlowSession session) {
this.flow = Validation.notNull(flow, "Flow cannot be null.");
this.session =
- (session == null) ? this.setConverseListener(new FlowSession()) : this.setSubConverseListener(session);
+ (session == null) ? this.setConverseListener(new FlowSession(true)) : this.setSubConverseListener(session);
this.session.begin();
this.callBackBuilder = FlowCallBack.builder();
}
@@ -60,7 +63,8 @@ public Conversation(AiProcessFlow flow, FlowSession session) {
@SafeVarargs
public final ConverseLatch offer(D... data) {
ConverseLatch latch = setListener(this.flow);
- FlowSession newSession = new FlowSession(this.session);
+ FlowSession newSession = FlowSession.newRootSession(this.session, this.session.preserved());
+ newSession.getWindow().setFrom(null);
this.flow.start().offer(data, newSession);
newSession.getWindow().complete();
return latch;
@@ -79,6 +83,7 @@ public ConverseLatch offer(String nodeId, List data) {
Validation.notBlank(nodeId, "invalid nodeId.");
ConverseLatch latch = setListener(this.flow);
FlowSession newSession = new FlowSession(this.session);
+ newSession.getWindow().setFrom(null);
this.flow.origin().offer(nodeId, data.toArray(new Object[0]), newSession);
newSession.getWindow().complete();
return latch;
@@ -110,6 +115,20 @@ public Conversation bind(Memory memory) {
return this;
}
+ /**
+ * 绑定流式响应信息消费者到对话上下文，用于消费流程流转过程中的流式信息。
+ *
+ * @param consumer 表示流式响应信息消费者的 {@link StreamingConsumer}{@code <}{@link ChatMessage}{@code ,
+ * }{@link ChatMessage}{@code >}。
+ * @return 表示绑定了流式响应信息消费者的对话对象的 {@link Conversation}{@code <}{@link D}{@code , }{@link R}{@code >}。
+ * @throws IllegalArgumentException 当 {@code consumer} 为 {@code null} 时。
+ */
+ public Conversation bind(StreamingConsumer consumer) {
+ Validation.notNull(consumer, "Streaming consumer cannot be null.");
+ this.session.setInnerState(StateKey.STREAMING_CONSUMER, consumer);
+ return this;
+ }
+
/**
* 绑定自定义参数到对话上下文，后续可以在流程中的如下节点获取：
*
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatBlockModel.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatBlockModel.java
new file mode 100644
index 00000000..84da4294
--- /dev/null
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatBlockModel.java
@@ -0,0 +1,90 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.operators.models;
+
+import modelengine.fel.core.chat.ChatMessage;
+import modelengine.fel.core.chat.ChatOption;
+import modelengine.fel.core.chat.MessageType;
+import modelengine.fel.core.chat.Prompt;
+import modelengine.fel.core.chat.support.AiMessage;
+import modelengine.fel.core.chat.ChatModel;
+import modelengine.fel.core.memory.Memory;
+import modelengine.fel.core.model.BlockModel;
+import modelengine.fel.engine.util.AiFlowSession;
+import modelengine.fel.engine.util.StateKey;
+import modelengine.fit.waterflow.domain.context.FlowSession;
+import modelengine.fitframework.inspection.Validation;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * 阻塞对话模型实现。
+ *
+ * @author 刘信宏
+ * @since 2024-04-16
+ */
+public class ChatBlockModel implements BlockModel {
+ private final ChatModel provider;
+ private final ChatOption option;
+
+ public ChatBlockModel(ChatModel provider) {
+ this(provider, ChatOption.custom().build());
+ }
+
+ public ChatBlockModel(ChatModel provider, ChatOption option) {
+ this.provider = Validation.notNull(provider, "The model provider cannot be null.");
+ this.option = Validation.notNull(option, "The chat options cannot be null.");
+ }
+
+ @Override
+ public ChatMessage invoke(Prompt input) {
+ Validation.notNull(input, "The model input data cannot be null.");
+ ChatOption dynamicOptions = AiFlowSession.get()
+ .map(state -> state.getInnerState(StateKey.CHAT_OPTION))
+ .orElse(this.option);
+ List chatMessages = this.provider.generate(input, dynamicOptions).blockAll();
+ Validation.notEmpty(chatMessages, "The model chat messages can not be empty.");
+ ChatMessage message = chatMessages.get(0);
+ Validation.equals(message.type(),
+ MessageType.AI,
+ "The message type must be {0}. [actualMessageType={1}]",
+ MessageType.AI,
+ message.type());
+ AiMessage answer = new AiMessage(message.text(), message.toolCalls());
+ this.updateMemory(answer);
+ return answer;
+ }
+
+ /**
+ * 绑定模型超参数。
+ *
+ * @param option 表示模型超参数的 {@link ChatOption}。
+ * @return 表示绑定了超参数的 {@link ChatBlockModel}。
+ * @throws IllegalArgumentException 当 {@code options} 为 {@code null} 时。
+ */
+ public ChatBlockModel bind(ChatOption option) {
+ Validation.notNull(option, "The chat options cannot be null.");
+ return new ChatBlockModel(this.provider, option);
+ }
+
+ private void updateMemory(AiMessage answer) {
+ if (answer.isToolCall()) {
+ return;
+ }
+ Optional session = AiFlowSession.get();
+ if (!session.isPresent()) {
+ return;
+ }
+ Memory memory = session.get().getInnerState(StateKey.HISTORY);
+ if (memory == null) {
+ return;
+ }
+ memory.add(session.get().getInnerState(StateKey.HISTORY_INPUT));
+ memory.add(answer);
+ }
+}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatChunk.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatChunk.java
new file mode 100644
index 00000000..e75e21e1
--- /dev/null
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatChunk.java
@@ -0,0 +1,103 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.operators.models;
+
+import modelengine.fel.core.chat.ChatMessage;
+import modelengine.fel.core.chat.MessageType;
+import modelengine.fel.core.tool.ToolCall;
+import modelengine.fel.core.tool.ToolCallChunk;
+import modelengine.fel.core.tool.support.DefaultToolCallChunk;
+import modelengine.fitframework.inspection.Validation;
+import modelengine.fitframework.util.CollectionUtils;
+import modelengine.fitframework.util.ObjectUtils;
+import modelengine.fitframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * 大模型流式响应内容片段。
+ *
+ * @author 刘信宏
+ * @since 2024-05-16
+ */
+public class ChatChunk implements ChatMessage {
+ private final StringBuilder text = new StringBuilder();
+ private final List toolCalls = new ArrayList<>();
+
+ public ChatChunk() {
+ }
+ /**
+ * 使用文本数据、媒体数据和工具请求初始化 {@link ChatChunk}。
+ *
+ * @param text 表示字符串数据的 {@link String}。
+ * @param toolCalls 表示工具请求的 {@link List}{@code <}{@link ToolCall}{@code >}。
+ */
+ public ChatChunk(String text, List toolCalls) {
+ this.text.append(ObjectUtils.nullIf(text, StringUtils.EMPTY));
+ this.toolCalls.addAll(ObjectUtils.getIfNull(toolCalls, Collections::emptyList));
+ }
+
+ /**
+ * 聚合流式响应内容片段。
+ *
+ * @param message 表示大模型流式响应内容片段的 {@link ChatMessage}。
+ */
+ public void merge(ChatMessage message) {
+ Validation.notNull(message, "Chat message can not be null.");
+ this.merge(message.text(), message.toolCalls());
+ }
+
+ @Override
+ public MessageType type() {
+ return MessageType.AI;
+ }
+
+ @Override
+ public String text() {
+ return this.text.toString();
+ }
+
+ @Override
+ public List toolCalls() {
+ return this.toolCalls;
+ }
+
+ @Override
+ public String toString() {
+ String textVal = this.toolCalls.isEmpty() ? this.text() : this.toolCalls.toString();
+ return this.type().getRole() + ": " + textVal;
+ }
+
+ /**
+ * 合并文本数据、媒体数据和工具请求。
+ *
+ * @param text 表示字符串数据的 {@link String}。
+ * @param toolCalls 表示工具请求的 {@link List}{@code <}{@link ToolCall}{@code >}。
+ */
+ private void merge(String text, List toolCalls) {
+ this.text.append(ObjectUtils.nullIf(text, StringUtils.EMPTY));
+ if (CollectionUtils.isEmpty(toolCalls)) {
+ return;
+ }
+ toolCalls.stream().filter(Objects::nonNull).forEach(toolCall -> {
+ if (StringUtils.isNotBlank(toolCall.id())) {
+ this.toolCalls.add(new DefaultToolCallChunk(toolCall));
+ return;
+ }
+ if (toolCall.index() == null || this.toolCalls.size() <= toolCall.index()) {
+ return;
+ }
+ ToolCall tarToolCall = this.toolCalls.get(toolCall.index());
+ if (tarToolCall instanceof ToolCallChunk) {
+ ObjectUtils.cast(tarToolCall).merge(toolCall);
+ }
+ });
+ }
+}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatFlowModel.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatFlowModel.java
index c4e664a4..1a00e827 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatFlowModel.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/ChatFlowModel.java
@@ -33,7 +33,7 @@ public class ChatFlowModel implements FlowModel {
public ChatFlowModel(ChatModel chatModel, ChatOption option) {
this.chatModel = notNull(chatModel, "The model provider can not be null.");
- this.option = notNull(option, "The chat options can not be null.");
+ this.option = option;
}
/**
@@ -54,7 +54,8 @@ public FitBoundedEmitter invoke(Prompt input) {
FlowSession session =
AiFlowSession.get().orElseThrow(() -> new IllegalStateException("The ai session cannot be empty."));
ChatOption dynamicOption = nullIf(session.getInnerState(StateKey.CHAT_OPTION), this.option);
+ notNull(dynamicOption, "The chat options can not be null.");
Choir choir = ObjectUtils.cast(this.chatModel.generate(input, dynamicOption));
- return new LlmEmitter<>(choir);
+ return new LlmEmitter<>(choir, input, session);
}
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/LlmEmitter.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/LlmEmitter.java
index 4b93424b..c13d0f67 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/LlmEmitter.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/LlmEmitter.java
@@ -7,8 +7,13 @@
package modelengine.fel.engine.operators.models;
import modelengine.fel.core.chat.ChatMessage;
+import modelengine.fel.core.chat.Prompt;
+import modelengine.fel.engine.util.StateKey;
import modelengine.fit.waterflow.bridge.fitflow.FitBoundedEmitter;
+import modelengine.fit.waterflow.domain.context.FlowSession;
import modelengine.fitframework.flowable.Publisher;
+import modelengine.fitframework.inspection.Validation;
+import modelengine.fitframework.util.ObjectUtils;
/**
* 流式模型发射器。
@@ -17,12 +22,28 @@
* @since 2024-05-16
*/
public class LlmEmitter extends FitBoundedEmitter {
+ private static final StreamingConsumer EMPTY_CONSUMER = (acc, chunk) -> {};
+
+ private final ChatChunk chunkAcc = new ChatChunk();
+ private final StreamingConsumer consumer;
+
/**
* 初始化 {@link LlmEmitter}。
*
* @param publisher 表示数据发布者的 {@link Publisher}{@code <}{@link O}{@code >}。
+ * @param prompt 表示模型输入的 {@link Prompt}， 用于获取默认用户问题。
+ * @param session 表示流程实例运行标识的 {@link FlowSession}。
*/
- public LlmEmitter(Publisher publisher) {
+ public LlmEmitter(Publisher publisher, Prompt prompt, FlowSession session) {
super(publisher, data -> data);
+ Validation.notNull(session, "The session cannot be null.");
+ this.consumer = ObjectUtils.nullIf(session.getInnerState(StateKey.STREAMING_CONSUMER), EMPTY_CONSUMER);
+ }
+
+ @Override
+ public void emit(ChatMessage data, FlowSession trans) {
+ super.emit(data, this.flowSession);
+ this.chunkAcc.merge(data);
+ this.consumer.accept(this.chunkAcc, data);
}
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/StreamingConsumer.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/StreamingConsumer.java
new file mode 100644
index 00000000..73170cd8
--- /dev/null
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/models/StreamingConsumer.java
@@ -0,0 +1,24 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.operators.models;
+
+/**
+ * 流式响应信息消费者。
+ *
+ * @author 刘信宏
+ * @since 2024-05-17
+ */
+@FunctionalInterface
+public interface StreamingConsumer {
+ /**
+ * 消费流式响应数据。
+ *
+ * @param acc 表示聚合信息的 {@link T}。
+ * @param chunk 表示单次流式响应信息 {@link U}。
+ */
+ void accept(T acc, U chunk);
+}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractAgent.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractAgent.java
index eb1cf831..f12bcce1 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractAgent.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractAgent.java
@@ -9,6 +9,7 @@
import static modelengine.fitframework.inspection.Validation.notBlank;
import static modelengine.fitframework.inspection.Validation.notNull;
+import lombok.Getter;
import modelengine.fel.core.chat.ChatMessage;
import modelengine.fel.core.chat.Prompt;
import modelengine.fel.core.chat.support.AiMessage;
@@ -62,13 +63,23 @@ protected AbstractAgent(ChatFlowModel flowModel, String memoryId) {
this.memoryId = notBlank(memoryId, "The agent message key cannot be blank.");
}
+ /**
+ * 获取配置的模型对象。
+ *
+ * @return 配置的模型对象。
+ */
+ public ChatFlowModel getModel() {
+ return model;
+ }
+
/**
* 执行工具调用。
*
* @param toolCalls 表示工具调用的 {@link List}{@code <}{@link ToolCall}{@code >}。
+ * @param ctx 表示工具调用上下文的 {@link StateContext}。
* @return 表示工具调用结果的 {@link Prompt}。
*/
- protected abstract Prompt doToolCall(List toolCalls);
+ protected abstract Prompt doToolCall(List toolCalls, StateContext ctx);
@Override
protected AiProcessFlow buildFlow() {
@@ -93,6 +104,6 @@ protected AiProcessFlow buildFlow() {
private void handleTool(ChatMessage message, StateContext ctx) {
ChatMessages lastRequest = ctx.getState(this.memoryId);
lastRequest.add(message);
- lastRequest.addAll(this.doToolCall(message.toolCalls()).messages());
+ lastRequest.addAll(this.doToolCall(message.toolCalls(), ctx).messages());
}
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractFlowPattern.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractFlowPattern.java
index 4fe00642..ec475432 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractFlowPattern.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/AbstractFlowPattern.java
@@ -13,9 +13,11 @@
import modelengine.fit.waterflow.domain.context.FlowSession;
import modelengine.fit.waterflow.domain.context.Window;
import modelengine.fit.waterflow.domain.emitters.EmitterListener;
+import modelengine.fit.waterflow.domain.emitters.FlowEmitter;
import modelengine.fit.waterflow.domain.flow.Flow;
import modelengine.fitframework.inspection.Validation;
import modelengine.fitframework.util.LazyLoader;
+import modelengine.fitframework.util.ObjectUtils;
/**
* 流程委托单元。
@@ -24,10 +26,25 @@
* @since 2024-06-04
*/
public abstract class AbstractFlowPattern implements FlowPattern {
+ private static final String RESULT_ACTION_KEY = "resultAction";
+ private static final String PARENT_SESSION_ID_KEY = "parentSessionId";
+
private final LazyLoader> flowSupplier;
+ private final EmitterListener dataDispatcher = (data, session) -> {
+ Object rawResultAction = session.getInnerState(RESULT_ACTION_KEY);
+ if (rawResultAction == null) {
+ return;
+ }
+ ResultAction resultAction = ObjectUtils.cast(rawResultAction);
+ resultAction.process(data, session);
+ };
protected AbstractFlowPattern() {
- this.flowSupplier = LazyLoader.of(this::buildFlow);
+ this.flowSupplier = LazyLoader.of(() -> {
+ AiProcessFlow flow = buildFlow();
+ flow.register(this.dataDispatcher);
+ return flow;
+ });
}
/**
@@ -39,21 +56,25 @@ protected AbstractFlowPattern() {
@Override
public void register(EmitterListener handler) {
- if (handler != null) {
- this.getFlow().register(handler);
- }
+ this.getFlow().register(handler);
+ }
+
+ @Override
+ public void unregister(EmitterListener listener) {
+ this.getFlow().unregister(listener);
}
@Override
public void emit(O data, FlowSession session) {
- FlowSession flowSession = new FlowSession(session);
- this.getFlow().emit(data, flowSession);
+ this.getFlow().emit(data, session);
}
@Override
- public O invoke(I data) {
- this.getFlow().converse(AiFlowSession.require()).offer(data);
- return null;
+ public FlowEmitter invoke(I data) {
+ FlowEmitter emitter = new FlowEmitter.AutoCompleteEmitter<>();
+ FlowSession flowSession = buildFlowSession(emitter);
+ this.getFlow().converse(flowSession).offer(data);
+ return emitter;
}
/**
@@ -65,7 +86,7 @@ public O invoke(I data) {
public Pattern sync() {
return new SimplePattern<>(data -> {
FlowSession require = AiFlowSession.require();
- FlowSession session = new FlowSession();
+ FlowSession session = new FlowSession(true);
Window window = session.begin();
session.copySessionState(require);
ConverseLatch conversation = this.getFlow().converse(session).offer(data);
@@ -83,7 +104,39 @@ public Flow origin() {
return this.getFlow().origin();
}
+ /**
+ * Built the flow session for starting the conversation.
+ *
+ * @param emitter The {@link FlowEmitter}{@code <}{@link O}{@code >} representing output emitter.
+ * @return The new {@link FlowSession}.
+ * @param The output data type.
+ */
+ protected static FlowSession buildFlowSession(FlowEmitter emitter) {
+ FlowSession mainSession = AiFlowSession.require();
+ FlowSession flowSession = FlowSession.newRootSession(mainSession, true);
+ flowSession.setInnerState(PARENT_SESSION_ID_KEY, mainSession.getId());
+ ResultAction resultAction = emitter::emit;
+ flowSession.setInnerState(RESULT_ACTION_KEY, resultAction);
+ return flowSession;
+ }
+
private AiProcessFlow getFlow() {
return Validation.notNull(this.flowSupplier.get(), "The flow cannot be null.");
}
+
+ /**
+ * A functional interface defining an action to be performed with processed results.
+ * Implementations handle both the result data and its associated flow session context.
+ *
+ * @param The type of result data to be processed.
+ */
+ protected interface ResultAction {
+ /**
+ * Process the result.
+ *
+ * @param data The result of {@link O}.
+ * @param flowSession The result flow session of {@link FlowSession}.
+ */
+ void process(O data, FlowSession flowSession);
+ }
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowNodeSupportable.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowNodeSupportable.java
new file mode 100644
index 00000000..70f61528
--- /dev/null
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowNodeSupportable.java
@@ -0,0 +1,52 @@
+/*---------------------------------------------------------------------------------------------
+ * Copyright (c) 2024 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.operators.patterns;
+
+import modelengine.fel.engine.flows.AiProcessFlow;
+import modelengine.fit.waterflow.domain.context.FlowSession;
+import modelengine.fit.waterflow.domain.emitters.FlowEmitter;
+import modelengine.fitframework.inspection.Validation;
+
+import java.util.Collections;
+
+/**
+ * 指定流程节点的异步委托单元的流程实现。
+ *
+ * @param 表示输入数据的类型。
+ * @param 表示流程处理完成的数据类型。
+ * @author 宋永坦
+ * @since 2025-05-16
+ */
+public class FlowNodeSupportable extends AbstractFlowPattern {
+ private final AiProcessFlow flow;
+ private final String nodeId;
+
+ /**
+ * 通过 AI 流程初始化 {@link FlowNodeSupportable}{@code <}{@link I}{@code , }{@link O}{@code >}。
+ *
+ * @param flow 表示 AI 流程的 {@link AiProcessFlow}{@code <}{@link I}{@code , }{@link O}{@code >}。
+ * @param nodeId 表示流程节点标识的 {@link String}。
+ * @throws IllegalArgumentException 当 {@code flow} 为 {@code null} 时。
+ */
+ public FlowNodeSupportable(AiProcessFlow flow, String nodeId) {
+ this.flow = Validation.notNull(flow, "The flow cannot be null.");
+ this.nodeId = Validation.notBlank(nodeId, "The node id cannot be null.");
+ }
+
+ @Override
+ protected AiProcessFlow buildFlow() {
+ return this.flow;
+ }
+
+ @Override
+ public FlowEmitter invoke(I data) {
+ FlowEmitter emitter = new FlowEmitter.AutoCompleteEmitter<>();
+ FlowSession flowSession = buildFlowSession(emitter);
+ this.flow.converse(flowSession).offer(this.nodeId, Collections.singletonList(data));
+ return emitter;
+ }
+}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowPattern.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowPattern.java
index 7a8d787f..dbd4199f 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowPattern.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/FlowPattern.java
@@ -9,6 +9,7 @@
import modelengine.fel.core.pattern.Pattern;
import modelengine.fit.waterflow.domain.context.FlowSession;
import modelengine.fit.waterflow.domain.emitters.Emitter;
+import modelengine.fit.waterflow.domain.emitters.FlowEmitter;
/**
* 流程委托单元。
@@ -18,4 +19,4 @@
* @author 刘信宏
* @since 2024-04-22
*/
-public interface FlowPattern extends Pattern, Emitter {}
+public interface FlowPattern extends Pattern>, Emitter {}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/SimpleFlowPattern.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/SimpleFlowPattern.java
index cba97274..4b42ca40 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/SimpleFlowPattern.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/SimpleFlowPattern.java
@@ -10,6 +10,7 @@
import modelengine.fel.engine.util.AiFlowSession;
import modelengine.fit.waterflow.domain.context.FlowSession;
import modelengine.fit.waterflow.domain.emitters.EmitterListener;
+import modelengine.fit.waterflow.domain.emitters.FlowEmitter;
import modelengine.fit.waterflow.domain.stream.operators.Operators;
import modelengine.fitframework.inspection.Validation;
import modelengine.fitframework.util.ObjectUtils;
@@ -25,7 +26,7 @@
* @since 2024-04-22
*/
public class SimpleFlowPattern implements FlowPattern {
- private EmitterListener handler;
+ private final FlowEmitter emitter = new FlowEmitter<>();
private final Operators.ProcessMap processor;
/**
@@ -35,38 +36,39 @@ public class SimpleFlowPattern implements FlowPattern {
* @throws IllegalArgumentException 当 {@code processor} 为 {@code null} 时。
*/
public SimpleFlowPattern(Operators.ProcessMap processor) {
- this(processor, null);
+ this.processor = Validation.notNull(processor, "The processor cannot be null.");
}
+ /**
+ * 使用委托单元初始化 {@link SimpleFlowPattern}{@code <}{@link I}{@code , }{@link O}{@code >}。
+ *
+ * @param pattern 表示委托单元的 {@link Pattern}{@code <}{@link I}{@code , }{@link O}{@code >}。
+ * @throws IllegalArgumentException 当 {@code processor} 为 {@code null} 时。
+ */
public SimpleFlowPattern(Pattern pattern) {
- this((data, ctx) -> AiFlowSession.applyPattern(pattern, data, ObjectUtils.cast(ctx)), null);
- }
-
- private SimpleFlowPattern(Operators.ProcessMap processor, EmitterListener handler) {
- this.processor = Validation.notNull(processor, "The processor cannot be null.");
- this.handler = handler;
+ this((data, ctx) -> AiFlowSession.applyPattern(pattern, data, ObjectUtils.cast(ctx)));
}
@Override
- public O invoke(I data) {
+ public FlowEmitter invoke(I data) {
FlowSession session = AiFlowSession.require();
- this.emit(this.processor.process(data, session), session);
- session.getWindow().complete();
- return null;
+ this.emitter.emit(this.processor.process(data, session));
+ this.emitter.complete();
+ return this.emitter;
}
@Override
public void register(EmitterListener handler) {
- if (handler != null) {
- this.handler = handler;
- }
+ this.emitter.register(handler);
+ }
+
+ @Override
+ public void unregister(EmitterListener handler) {
+ this.emitter.unregister(handler);
}
@Override
public void emit(O data, FlowSession session) {
- if (this.handler == null) {
- return;
- }
- this.handler.handle(data, session);
+ this.emitter.emit(data, session);
}
}
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/support/DefaultAgent.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/support/DefaultAgent.java
index af40e238..1ca3742e 100644
--- a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/support/DefaultAgent.java
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/patterns/support/DefaultAgent.java
@@ -17,6 +17,7 @@
import modelengine.fel.engine.operators.models.ChatFlowModel;
import modelengine.fel.engine.operators.patterns.AbstractAgent;
import modelengine.fel.tool.service.ToolExecuteService;
+import modelengine.fit.waterflow.domain.context.StateContext;
import java.util.List;
import java.util.stream.Collectors;
@@ -38,7 +39,7 @@ public DefaultAgent(ChatFlowModel flowModel, String namespace, ToolExecuteServic
}
@Override
- protected Prompt doToolCall(List toolCalls) {
+ protected Prompt doToolCall(List toolCalls, StateContext ctx) {
return toolCalls.stream().map(toolCall -> {
String text = this.toolExecuteService.execute(this.namespace, toolCall.name(), toolCall.arguments());
return (ChatMessage) new ToolMessage(toolCall.id(), text);
diff --git a/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/sources/Source.java b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/sources/Source.java
new file mode 100644
index 00000000..57e0e260
--- /dev/null
+++ b/framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/operators/sources/Source.java
@@ -0,0 +1,57 @@
+/*---------------------------------------------------------------------------------------------
+ * 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.operators.sources;
+
+import modelengine.fit.waterflow.domain.context.FlowSession;
+import modelengine.fit.waterflow.domain.emitters.Emitter;
+import modelengine.fit.waterflow.domain.emitters.EmitterListener;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 数据发射源实现，支持注册多个监听器。仅支持主动逐个发射数据。
+ *
+ * @param 表示数据源的业务数据类型。
+ * @author 刘信宏
+ * @since 2024-04-28
+ */
+public class Source implements Emitter {
+ private static final Integer MAXIMUM_POOL_SIZE = 50;
+ private static final ExecutorService THREAD_POOL = new ThreadPoolExecutor(0, MAXIMUM_POOL_SIZE,
+ 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
+
+ private final List