feat(XL13): 支持工作流大模型流式输出

- 增加文本与思考增量事件及取消终态收口 - 支持图片输入解析并修复死信定义查找空值 - 补充并发与模型流式回归测试
2026-07-31 09:40:38 +08:00
parent e74d229de2
commit fcc36dc699
10 changed files with 907 additions and 26 deletions
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java
@@ -778,7 +778,19 @@ public class Chain {
        if (state == null) {
            throw new IllegalStateException("Unable to initialize chain state: " + stateInstanceId);
        }
-        return state;
+        if (StringUtil.hasText(state.getChainDefinitionId())
+                || definition == null
+                || StringUtil.noText(definition.getId())) {
+            return state;
+        }
+        // 定义 ID 必须先于可重放入口触发器持久化，避免恢复时无法定位定义快照。
+        return updateStateSafely(current -> {
+            if (StringUtil.hasText(current.getChainDefinitionId())) {
+                return null;
+            }
+            current.setChainDefinitionId(definition.getId());
+            return EnumSet.of(ChainStateField.CHAIN_DEFINITION_ID);
+        });
    }

    private boolean shouldSkipNode(Node node, String edgeId) {
@@ -1613,6 +1625,8 @@ public class Chain {
        if (changed.get()) {
            notifyEvent(new ChainStatusChangeEvent(
                    this, ChainStatus.CANCELLED, before.get()));
+            // 取消属于工作流终态，统一发布结束事件供审计、清理等监听器收口。
+            notifyEvent(new ChainEndEvent(this));
        }
        return changed.get();
    }
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/LlmStreamEvent.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/LlmStreamEvent.java
@@ -0,0 +1,126 @@
+/**
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ * <p>
+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.gnu.org/licenses/lgpl-3.0.txt
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.easyagents.flow.core.chain.event;
+
+import com.easyagents.flow.core.chain.Chain;
+import com.easyagents.flow.core.chain.Node;
+
+/**
+ * LLM 节点生成文本时发布的增量事件。
+ */
+public class LlmStreamEvent extends BaseEvent {
+
+    private final Node node;
+    private final String streamId;
+    private final String delta;
+    private final ContentType contentType;
+
+    /**
+     * LLM 流式内容类型。
+     */
+    public enum ContentType {
+        /**
+         * 模型正式回答。
+         */
+        TEXT,
+        /**
+         * 模型思考过程。
+         */
+        REASONING
+    }
+
+    /**
+     * 创建 LLM 文本增量事件。
+     *
+     * @param chain 当前工作流
+     * @param node 当前 LLM 节点
+     * @param streamId 当前节点本次调用的流标识
+     * @param delta 本次新增文本
+     */
+    public LlmStreamEvent(Chain chain, Node node, String streamId, String delta) {
+        this(chain, node, streamId, delta, ContentType.TEXT);
+    }
+
+    /**
+     * 创建指定内容类型的 LLM 增量事件。
+     *
+     * @param chain 当前工作流
+     * @param node 当前 LLM 节点
+     * @param streamId 当前节点本次调用的流标识
+     * @param delta 本次新增内容
+     * @param contentType 增量内容类型
+     */
+    public LlmStreamEvent(
+            Chain chain,
+            Node node,
+            String streamId,
+            String delta,
+            ContentType contentType
+    ) {
+        super(chain);
+        this.node = node;
+        this.streamId = streamId;
+        this.delta = delta;
+        this.contentType = contentType == null
+                ? ContentType.TEXT
+                : contentType;
+    }
+
+    /**
+     * 获取当前 LLM 节点。
+     *
+     * @return 当前节点
+     */
+    public Node getNode() {
+        return node;
+    }
+
+    /**
+     * 获取当前节点本次调用的流标识。
+     *
+     * @return 流标识
+     */
+    public String getStreamId() {
+        return streamId;
+    }
+
+    /**
+     * 获取本次新增文本。
+     *
+     * @return 文本增量
+     */
+    public String getDelta() {
+        return delta;
+    }
+
+    /**
+     * 获取本次增量的内容类型。
+     *
+     * @return 内容类型
+     */
+    public ContentType getContentType() {
+        return contentType;
+    }
+
+    /**
+     * 判断本次增量是否为模型思考内容。
+     *
+     * @return {@code true} 表示思考内容
+     */
+    public boolean isReasoning() {
+        return contentType == ContentType.REASONING;
+    }
+}
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
 import java.io.Serializable;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.function.Consumer;

 /**
 * TinyFlow 最新 ChainExecutor
@@ -397,8 +398,29 @@ public class ChainExecutor {
    }

    public String executeAsync(String definitionId, Map<String, Object> variables) {
+        return executeAsync(definitionId, variables, null);
+    }
+
+    /**
+     * 异步启动工作流，并在首个节点开始前暴露执行实例 ID。
+     *
+     * <p>回调用于提前注册流式事件接收器，保证高速工作流不会在调用方拿到执行 ID
+     * 之前丢失开始事件或首批输出。</p>
+     *
+     * @param definitionId 工作流定义 ID
+     * @param variables 工作流输入变量
+     * @param beforeStart 启动前回调；可为 {@code null}
+     * @return 执行实例 ID
+     */
+    public String executeAsync(
+            String definitionId,
+            Map<String, Object> variables,
+            Consumer<String> beforeStart) {
        Chain chain = createChain(definitionId);
        try {
+            if (beforeStart != null) {
+                beforeStart.accept(chain.getStateInstanceId());
+            }
            chain.start(variables);
            return chain.getStateInstanceId();
        } catch (RuntimeException | Error error) {
@@ -962,9 +984,12 @@ public class ChainExecutor {
            return definition;
        }
        ChainDefinition loaded = definitionSnapshotRepository.load(stateInstanceId);
-        if (loaded == null) {
+        String definitionId = state.getChainDefinitionId();
+        if (loaded == null
+                && definitionId != null
+                && !definitionId.isBlank()) {
            // 兼容升级前已经启动、尚未持久化定义快照的实例。
-            loaded = definitionRepository.getChainDefinitionById(state.getChainDefinitionId());
+            loaded = definitionRepository.getChainDefinitionById(definitionId);
        }
        if (loaded == null) {
            return null;
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/llm/Llm.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/llm/Llm.java
@@ -50,6 +50,7 @@ public interface Llm {
        private String message;
        private String systemMessage;
        private List<String> images;
+        private List<Object> imageInputs;

        public String getMessage() {
            return message;
@@ -74,6 +75,24 @@ public interface Llm {
        public void setImages(List<String> images) {
            this.images = images;
        }
+
+        /**
+         * 获取尚未转换为模型图片 URL 的原始图片输入。
+         *
+         * @return 原始图片输入列表
+         */
+        public List<Object> getImageInputs() {
+            return imageInputs;
+        }
+
+        /**
+         * 设置原始图片输入，供模型调用前按运行环境解析。
+         *
+         * @param imageInputs 原始图片输入列表
+         */
+        public void setImageInputs(List<Object> imageInputs) {
+            this.imageInputs = imageInputs;
+        }
    }

    /**
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LlmNode.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LlmNode.java
@@ -23,7 +23,6 @@ import com.easyagents.flow.core.llm.Llm;
 import com.easyagents.flow.core.llm.LlmManager;
 import com.easyagents.flow.core.util.*;

-import java.io.File;
 import java.util.*;

 public class LlmNode extends BaseNode {
@@ -123,17 +122,21 @@ public class LlmNode extends BaseNode {
            Map<String, Object> filesMap =
                    chainState.resolveParameters(
                            this, images);
-            List<String> imagesUrls = new ArrayList<>();
-            filesMap.forEach((s, o) -> {
-                if (o instanceof String) {
-                    imagesUrls.add((String) o);
-                } else if (o instanceof File) {
-                    byte[] bytes = IOUtil.readBytes((File) o);
-                    String base64 = Base64.getEncoder().encodeToString(bytes);
-                    imagesUrls.add(base64);
+            List<Object> imageInputs = new ArrayList<>(filesMap.size());
+            filesMap.forEach((name, value) -> {
+                if (value == null) {
+                    return;
                }
+                if (!(value instanceof String)
+                        && !(value instanceof java.io.File)
+                        && !(value instanceof Map<?, ?>)) {
+                    throw new IllegalArgumentException(
+                            "Unsupported image input for parameter '" + name + "': "
+                                    + value.getClass().getName());
+                }
+                imageInputs.add(value);
            });
-            messageInfo.setImages(imagesUrls);
+            messageInfo.setImageInputs(imageInputs);
        }


--- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java
+++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java
@@ -20,6 +20,7 @@ import com.easyagents.flow.core.chain.ChainDefinition;
 import com.easyagents.flow.core.chain.ChainStatus;
 import com.easyagents.flow.core.chain.Edge;
 import com.easyagents.flow.core.chain.ChainState;
+import com.easyagents.flow.core.chain.event.ChainEndEvent;
 import com.easyagents.flow.core.chain.repository.ChainDefinitionSnapshotRepository;
 import com.easyagents.flow.core.chain.repository.ChainStateField;
 import com.easyagents.flow.core.chain.repository.ChainStateRepository;
@@ -28,6 +29,7 @@ import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository;
 import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
 import com.easyagents.flow.core.chain.runtime.ChainExecutor;
 import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
+import com.easyagents.flow.core.chain.runtime.Trigger;
 import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
 import com.easyagents.flow.core.node.EndNode;
 import com.easyagents.flow.core.node.BaseNode;
@@ -36,6 +38,7 @@ import org.junit.Assert;
 import org.junit.Test;

 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumSet;
@@ -49,12 +52,100 @@ import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;

 /**
 * {@link ChainExecutor} 并发同步执行测试。
 */
 public class ChainExecutorConcurrencyTest {

+    /**
+     * 验证实例初始化会在入口触发器创建前持久化工作流定义 ID。
+     *
+     * @throws Exception 执行器初始化或异步启动失败时抛出
+     */
+    @Test
+    public void shouldPersistDefinitionIdBeforeStartingWorkflow() throws Exception {
+        ScheduledExecutorService schedulerPool = Executors.newSingleThreadScheduledExecutor();
+        ExecutorService workerPool = Executors.newFixedThreadPool(2);
+        TriggerScheduler triggerScheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L);
+        ChainDefinition definition = createDefinition();
+        InMemoryChainStateRepository stateRepository =
+                new InMemoryChainStateRepository();
+        ChainExecutor chainExecutor = new ChainExecutor(
+                id -> definition,
+                stateRepository,
+                new InMemoryNodeStateRepository(),
+                triggerScheduler);
+        AtomicReference<String> persistedDefinitionId = new AtomicReference<>();
+
+        try {
+            chainExecutor.executeAsync(
+                    definition.getId(),
+                    Collections.emptyMap(),
+                    executeId -> persistedDefinitionId.set(
+                            stateRepository.load(executeId).getChainDefinitionId())
+            );
+
+            Assert.assertEquals(
+                    definition.getId(),
+                    persistedDefinitionId.get());
+        } finally {
+            triggerScheduler.shutdown();
+        }
+    }
+
+    /**
+     * 验证历史异常实例缺少定义 ID 时，死信触发器仍能收敛为失败终态。
+     *
+     * @throws Exception 反射调用死信收口逻辑失败时抛出
+     */
+    @Test
+    public void shouldFailDeadLetteredWorkflowWithoutDefinitionId()
+            throws Exception {
+        ScheduledExecutorService schedulerPool = Executors.newSingleThreadScheduledExecutor();
+        ExecutorService workerPool = Executors.newFixedThreadPool(2);
+        TriggerScheduler triggerScheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L);
+        InMemoryChainStateRepository stateRepository =
+                new InMemoryChainStateRepository();
+        AtomicInteger definitionLoadCount = new AtomicInteger();
+        ChainExecutor chainExecutor = new ChainExecutor(
+                id -> {
+                    definitionLoadCount.incrementAndGet();
+                    return null;
+                },
+                stateRepository,
+                new InMemoryNodeStateRepository(),
+                triggerScheduler);
+        String instanceId = "dead-letter-missing-definition";
+        stateRepository.create(instanceId);
+        Trigger trigger = new Trigger();
+        trigger.setId("dead-letter-trigger");
+        trigger.setStateInstanceId(instanceId);
+        Method failDeadLetteredTrigger = ChainExecutor.class.getDeclaredMethod(
+                "failDeadLetteredTrigger",
+                Trigger.class,
+                Throwable.class);
+        failDeadLetteredTrigger.setAccessible(true);
+
+        try {
+            boolean finalized = (boolean) failDeadLetteredTrigger.invoke(
+                    chainExecutor,
+                    trigger,
+                    new IllegalStateException("delivery attempts exhausted"));
+
+            Assert.assertTrue(finalized);
+            Assert.assertEquals(
+                    ChainStatus.FAILED,
+                    stateRepository.load(instanceId).getStatus());
+            Assert.assertEquals(0, definitionLoadCount.get());
+        } finally {
+            triggerScheduler.shutdown();
+        }
+    }
+
    /**
     * 验证多个同步调用可以通过实例 ID 独立接收执行结果。
     *
@@ -143,6 +234,7 @@ public class ChainExecutorConcurrencyTest {
        CountDownLatch ioStarted = new CountDownLatch(1);
        CountDownLatch allowIoCompletion = new CountDownLatch(1);
        AtomicInteger downstreamExecutions = new AtomicInteger();
+        AtomicInteger chainEndEvents = new AtomicInteger();
        ChainDefinition definition = createCancellationDefinition(
                ioStarted, allowIoCompletion, downstreamExecutions);
        ChainExecutor chainExecutor = new ChainExecutor(
@@ -150,6 +242,11 @@ public class ChainExecutorConcurrencyTest {
                chainStateRepository,
                new InMemoryNodeStateRepository(),
                triggerScheduler);
+        chainExecutor.addEventListener((event, chain) -> {
+            if (event instanceof ChainEndEvent) {
+                chainEndEvents.incrementAndGet();
+            }
+        });

        try {
            String instanceId = chainExecutor.executeAsync(
@@ -163,6 +260,7 @@ public class ChainExecutorConcurrencyTest {
            Assert.assertEquals(
                    ChainStatus.CANCELLED,
                    chainStateRepository.load(instanceId).getStatus());
+            Assert.assertEquals(1, chainEndEvents.get());
        } finally {
            allowIoCompletion.countDown();
            triggerScheduler.shutdown();