From fcc36dc699635389cae36a015010853cd3127a65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Fri, 31 Jul 2026 09:40:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(XL13):=20=E6=94=AF=E6=8C=81=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E6=B5=81=E5=A4=A7=E6=A8=A1=E5=9E=8B=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 增加文本与思考增量事件及取消终态收口 - 支持图片输入解析并修复死信定义查找空值 - 补充并发与模型流式回归测试 --- .../com/easyagents/flow/core/chain/Chain.java | 16 +- .../flow/core/chain/event/LlmStreamEvent.java | 126 +++++++ .../core/chain/runtime/ChainExecutor.java | 29 +- .../com/easyagents/flow/core/llm/Llm.java | 19 ++ .../easyagents/flow/core/node/LlmNode.java | 23 +- .../test/ChainExecutorConcurrencyTest.java | 98 ++++++ easy-agents-support/pom.xml | 6 +- .../flow/support/provider/EasyAgentsLlm.java | 286 +++++++++++++++- .../support/provider/ImageInputResolver.java | 16 + .../support/provider/EasyAgentsLlmTest.java | 314 ++++++++++++++++++ 10 files changed, 907 insertions(+), 26 deletions(-) create mode 100644 easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/LlmStreamEvent.java create mode 100644 easy-agents-support/src/main/java/com/easyagents/flow/support/provider/ImageInputResolver.java create mode 100644 easy-agents-support/src/test/java/com/easyagents/flow/support/provider/EasyAgentsLlmTest.java diff --git 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 index df5c58e..9fc1fb1 100644 --- 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(); } diff --git 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 new file mode 100644 index 0000000..bbcdc9f --- /dev/null +++ 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). + *

+ * 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 + *

+ * http://www.gnu.org/licenses/lgpl-3.0.txt + *

+ * 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; + } +} diff --git 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 index 130d1a8..c2491e2 100644 --- 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 variables) { + return executeAsync(definitionId, variables, null); + } + + /** + * 异步启动工作流，并在首个节点开始前暴露执行实例 ID。 + * + *

回调用于提前注册流式事件接收器，保证高速工作流不会在调用方拿到执行 ID + * 之前丢失开始事件或首批输出。

+ * + * @param definitionId 工作流定义 ID + * @param variables 工作流输入变量 + * @param beforeStart 启动前回调；可为 {@code null} + * @return 执行实例 ID + */ + public String executeAsync( + String definitionId, + Map variables, + Consumer 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; diff --git 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 index d681115..1bcf699 100644 --- 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 images; + private List imageInputs; public String getMessage() { return message; @@ -74,6 +75,24 @@ public interface Llm { public void setImages(List images) { this.images = images; } + + /** + * 获取尚未转换为模型图片 URL 的原始图片输入。 + * + * @return 原始图片输入列表 + */ + public List getImageInputs() { + return imageInputs; + } + + /** + * 设置原始图片输入，供模型调用前按运行环境解析。 + * + * @param imageInputs 原始图片输入列表 + */ + public void setImageInputs(List imageInputs) { + this.imageInputs = imageInputs; + } } /** diff --git 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 index e62e31b..f2e3e0f 100644 --- 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 filesMap = chainState.resolveParameters( this, images); - List 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 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); } diff --git 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 index f1547eb..b9d402d 100644 --- 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 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(); diff --git a/easy-agents-support/pom.xml b/easy-agents-support/pom.xml index b25b315..0dfe000 100644 --- a/easy-agents-support/pom.xml +++ b/easy-agents-support/pom.xml @@ -28,7 +28,11 @@ easy-agents-core - + + junit + junit + test + diff --git a/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/EasyAgentsLlm.java b/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/EasyAgentsLlm.java index 9ab2422..d09d295 100644 --- a/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/EasyAgentsLlm.java +++ b/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/EasyAgentsLlm.java @@ -2,46 +2,240 @@ package com.easyagents.flow.support.provider; import com.easyagents.core.message.AiMessage; import com.easyagents.core.message.SystemMessage; +import com.easyagents.core.model.chat.BaseChatModel; import com.easyagents.core.model.chat.ChatModel; +import com.easyagents.core.model.chat.StreamResponseListener; +import com.easyagents.core.model.client.StreamContext; import com.easyagents.core.model.chat.response.AiMessageResponse; import com.easyagents.core.prompt.SimplePrompt; +import com.easyagents.core.util.ImageUtil; import com.easyagents.flow.core.chain.Chain; +import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent; +import com.easyagents.flow.core.chain.event.LlmStreamEvent; +import com.easyagents.flow.core.chain.listener.ChainEventListener; import com.easyagents.flow.core.llm.Llm; import com.easyagents.flow.core.node.LlmNode; import com.easyagents.flow.core.util.StringUtil; +import java.io.File; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; +/** + * 基于 Easy-Agents 聊天模型实现工作流 LLM 调用。 + */ public class EasyAgentsLlm implements Llm { private ChatModel chatModel; + private ImageInputResolver imageInputResolver; + /** + * 获取聊天模型。 + * + * @return 聊天模型 + */ public ChatModel getChatModel() { return chatModel; } + /** + * 设置聊天模型。 + * + * @param chatModel 聊天模型 + */ public void setChatModel(ChatModel chatModel) { this.chatModel = chatModel; } + /** + * 获取图片输入解析器。 + * + * @return 图片输入解析器 + */ + public ImageInputResolver getImageInputResolver() { + return imageInputResolver; + } + + /** + * 设置图片输入解析器。 + * + * @param imageInputResolver 图片输入解析器 + */ + public void setImageInputResolver(ImageInputResolver imageInputResolver) { + this.imageInputResolver = imageInputResolver; + } + + /** + * 调用聊天模型并返回文本结果。 + * + * @param messageInfo 消息信息 + * @param options 模型参数 + * @param llmNode 当前 LLM 节点 + * @param chain 工作流链 + * @return 模型文本结果 + */ @Override public String chat(MessageInfo messageInfo, ChatOptions options, LlmNode llmNode, Chain chain) { + SimplePrompt prompt = buildPrompt(messageInfo); + com.easyagents.core.model.chat.ChatOptions chatOptions = buildChatOptions(options); + CountDownLatch completion = new CountDownLatch(1); + AtomicReference streamContext = new AtomicReference<>(); + AtomicReference result = new AtomicReference<>(); + AtomicReference failure = new AtomicReference<>(); + String streamId = UUID.randomUUID().toString(); + + ChainEventListener cancellationListener = (event, eventChain) -> { + if (!(event instanceof ChainStatusChangeEvent statusEvent) + || statusEvent.getStatus() != ChainStatus.CANCELLED + || !chain.getStateInstanceId().equals(eventChain.getStateInstanceId())) { + return; + } + StreamContext context = streamContext.get(); + if (context != null) { + context.getClient().stop(); + } + }; + chain.getEventManager().addEventListener( + ChainStatusChangeEvent.class, cancellationListener); + + try { + chatModel.chatStream(prompt, new StreamResponseListener() { + /** + * 记录流客户端，供工作流取消时立即关闭模型连接。 + * + * @param context 流上下文 + */ + @Override + public void onStart(StreamContext context) { + streamContext.set(context); + } + + /** + * 将模型文本增量转发为工作流事件。 + * + * @param context 流上下文 + * @param response 本次模型响应 + */ + @Override + public void onMessage(StreamContext context, AiMessageResponse response) { + AiMessage message = response == null ? null : response.getMessage(); + if (message == null) { + return; + } + String reasoningDelta = message.getReasoningContent(); + if (StringUtil.hasText(reasoningDelta)) { + chain.notifyEvent(new LlmStreamEvent( + chain, + llmNode, + streamId, + reasoningDelta, + LlmStreamEvent.ContentType.REASONING + )); + } + String textDelta = message.getContent(); + if (StringUtil.hasText(textDelta)) { + chain.notifyEvent(new LlmStreamEvent( + chain, + llmNode, + streamId, + textDelta, + LlmStreamEvent.ContentType.TEXT + )); + } + } + + /** + * 收集完整响应并结束同步节点等待。 + * + * @param context 流上下文 + */ + @Override + public void onStop(StreamContext context) { + try { + if (failure.get() == null) { + AiMessage message = context.getFullMessage(); + if (message == null || StringUtil.noText(message.getFullContent())) { + failure.compareAndSet( + null, + new IllegalStateException( + "EasyAgentsLlm can not get aiMessage!")); + } else { + result.set(message.getFullContent()); + } + } + } finally { + completion.countDown(); + } + } + + /** + * 记录模型流异常并结束等待。 + * + * @param context 流上下文 + * @param throwable 模型异常 + */ + @Override + public void onFailure(StreamContext context, Throwable throwable) { + failure.compareAndSet( + null, + throwable == null + ? new IllegalStateException("EasyAgentsLlm stream failed") + : throwable); + completion.countDown(); + } + }, chatOptions); + awaitCompletion(completion, streamContext); + } finally { + chain.getEventManager().removeEventListener( + ChainStatusChangeEvent.class, cancellationListener); + } + + Throwable throwable = failure.get(); + if (throwable != null) { + throw new RuntimeException("EasyAgentsLlm stream failed", throwable); + } + if (StringUtil.noText(result.get())) { + throw new RuntimeException("EasyAgentsLlm can not get response!"); + } + return result.get(); + } + + /** + * 构建模型提示词，并解析图片输入。 + * + * @param messageInfo 消息信息 + * @return 模型提示词 + */ + private SimplePrompt buildPrompt(MessageInfo messageInfo) { SimplePrompt prompt = new SimplePrompt(messageInfo.getMessage()); - // 系统提示词 if (StringUtil.hasText(messageInfo.getSystemMessage())) { prompt.setSystemMessage(SystemMessage.of(messageInfo.getSystemMessage())); } - // 图片 - List images = messageInfo.getImages(); + List images = resolveImages(messageInfo); if (images != null && !images.isEmpty()) { + assertImageSupported(); for (String image : images) { prompt.addImageUrl(image); } } + return prompt; + } + /** + * 构建 Easy-Agents 模型参数。 + * + * @param options 工作流模型参数 + * @return Easy-Agents 模型参数 + */ + private com.easyagents.core.model.chat.ChatOptions buildChatOptions(ChatOptions options) { com.easyagents.core.model.chat.ChatOptions chatOptions = new com.easyagents.core.model.chat.ChatOptions(); chatOptions.setSeed(options.getSeed()); chatOptions.setTemperature(options.getTemperature()); @@ -49,21 +243,89 @@ public class EasyAgentsLlm implements Llm { chatOptions.setTopK(options.getTopK()); chatOptions.setMaxTokens(options.getMaxTokens()); chatOptions.setStop(options.getStop()); + return chatOptions; + } - AiMessageResponse response = chatModel.chat(prompt, chatOptions); - if (response == null) { - throw new RuntimeException("EasyAgentsLlm can not get response!"); + /** + * 等待异步模型流结束。 + * + * @param completion 完成信号 + * @param streamContext 当前模型流上下文 + */ + private void awaitCompletion( + CountDownLatch completion, + AtomicReference streamContext) { + try { + completion.await(); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + StreamContext context = streamContext.get(); + if (context != null) { + context.getClient().stop(); + } + throw new RuntimeException("EasyAgentsLlm stream interrupted", exception); + } + } + + /** + * 将原始图片输入解析为模型图片 URL。 + * + * @param messageInfo 消息信息 + * @return 已解析图片 URL 列表 + */ + List resolveImages(MessageInfo messageInfo) { + if (messageInfo == null) { + return Collections.emptyList(); + } + List inputs = messageInfo.getImageInputs(); + if (inputs == null || inputs.isEmpty()) { + inputs = messageInfo.getImages(); + } + if (inputs == null || inputs.isEmpty()) { + return Collections.emptyList(); } - if (response.isError()) { - throw new RuntimeException("EasyAgentsLlm error: " + response.getErrorMessage()); + List resolvedImages = new ArrayList<>(inputs.size()); + for (Object input : inputs) { + if (input == null) { + continue; + } + String resolvedImage = resolveImage(input); + if (StringUtil.noText(resolvedImage)) { + throw new IllegalArgumentException("Resolved image input must not be blank"); + } + resolvedImages.add(resolvedImage); } + return resolvedImages; + } - AiMessage aiMessage = response.getMessage(); - if (aiMessage != null) { - return aiMessage.getTextContent(); + /** + * 解析单个图片输入。 + * + * @param imageInput 原始图片输入 + * @return 图片 URL 或 Data URI + */ + private String resolveImage(Object imageInput) { + if (imageInputResolver != null) { + return imageInputResolver.resolve(imageInput); } + if (imageInput instanceof String value) { + return value; + } + if (imageInput instanceof File file) { + return ImageUtil.imageFileToDataUri(file); + } + throw new IllegalArgumentException( + "Unsupported image input type: " + imageInput.getClass().getName()); + } - throw new RuntimeException("EasyAgentsLlm can not get aiMessage!"); + /** + * 校验当前聊天模型是否明确支持图片。 + */ + private void assertImageSupported() { + if (chatModel instanceof BaseChatModel baseChatModel + && Boolean.FALSE.equals(baseChatModel.getConfig().getSupportImage())) { + throw new IllegalArgumentException("当前模型不支持图片输入，请选择支持视觉能力的模型"); + } } } diff --git a/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/ImageInputResolver.java b/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/ImageInputResolver.java new file mode 100644 index 0000000..9bdf14d --- /dev/null +++ b/easy-agents-support/src/main/java/com/easyagents/flow/support/provider/ImageInputResolver.java @@ -0,0 +1,16 @@ +package com.easyagents.flow.support.provider; + +/** + * 将工作流运行态图片输入解析为模型可消费的图片 URL 或 Data URI。 + */ +@FunctionalInterface +public interface ImageInputResolver { + + /** + * 解析单个图片输入。 + * + * @param imageInput 原始图片输入 + * @return 图片 URL 或带 MIME 的 Data URI + */ + String resolve(Object imageInput); +} diff --git a/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/EasyAgentsLlmTest.java b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/EasyAgentsLlmTest.java new file mode 100644 index 0000000..ea51d0e --- /dev/null +++ b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/EasyAgentsLlmTest.java @@ -0,0 +1,314 @@ +package com.easyagents.flow.support.provider; + +import com.easyagents.core.model.chat.BaseChatModel; +import com.easyagents.core.model.chat.ChatConfig; +import com.easyagents.core.model.chat.ChatModel; +import com.easyagents.core.model.chat.ChatOptions; +import com.easyagents.core.model.chat.StreamResponseListener; +import com.easyagents.core.model.chat.response.AiMessageResponse; +import com.easyagents.core.model.client.StreamClient; +import com.easyagents.core.model.client.StreamContext; +import com.easyagents.core.prompt.Prompt; +import com.easyagents.flow.core.chain.Chain; +import com.easyagents.flow.core.chain.ChainDefinition; +import com.easyagents.flow.core.chain.EventManager; +import com.easyagents.flow.core.chain.event.LlmStreamEvent; +import com.easyagents.flow.core.llm.Llm; +import com.easyagents.flow.core.node.LlmNode; +import org.junit.Assert; +import org.junit.Test; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * 工作流 LLM 图片输入解析测试。 + */ +public class EasyAgentsLlmTest { + + /** + * 验证结构化图片对象通过注入的解析器转换。 + */ + @Test + public void shouldResolveRawImageInput() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + llm.setImageInputResolver(input -> { + Assert.assertTrue(input instanceof Map); + return "data:image/png;base64,AQID"; + }); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setImageInputs(List.of(Map.of( + "sourceType", "upload", + "fileName", "image.png", + "filePath", "workflow/image.png"))); + + Assert.assertEquals( + List.of("data:image/png;base64,AQID"), + llm.resolveImages(messageInfo)); + } + + /** + * 验证旧版图片字符串仍然可直接传递。 + */ + @Test + public void shouldKeepLegacyImageString() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setImages(List.of("https://example.com/image.png")); + + Assert.assertEquals( + List.of("https://example.com/image.png"), + llm.resolveImages(messageInfo)); + } + + /** + * 验证 File 图片会补全 MIME 和 Data URI 前缀。 + * + * @throws Exception 临时文件写入失败 + */ + @Test + public void shouldConvertFileToCompleteDataUri() throws Exception { + Path image = Files.createTempFile("easy-agents-image-", ".png"); + try { + Files.write(image, new byte[]{1, 2, 3}); + EasyAgentsLlm llm = new EasyAgentsLlm(); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setImageInputs(List.of(image.toFile())); + + Assert.assertEquals( + List.of("data:image/png;base64,AQID"), + llm.resolveImages(messageInfo)); + } finally { + Files.deleteIfExists(image); + } + } + + /** + * 验证明确不支持图片的模型会在发送请求前失败。 + */ + @Test + public void shouldRejectImageForUnsupportedModel() { + ChatConfig config = new ChatConfig(); + config.setSupportImage(Boolean.FALSE); + EasyAgentsLlm llm = new EasyAgentsLlm(); + llm.setChatModel(new BaseChatModel<>(config) { + }); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setImages(List.of("data:image/png;base64,AQID")); + + try { + llm.chat(messageInfo, new Llm.ChatOptions(), null, null); + Assert.fail("expected IllegalArgumentException"); + } catch (IllegalArgumentException exception) { + Assert.assertEquals( + "当前模型不支持图片输入，请选择支持视觉能力的模型", + exception.getMessage()); + } + } + + /** + * 验证重复执行同一 LLM 节点时，每次调用拥有独立流标识且增量不会被覆盖。 + */ + @Test + public void shouldEmitIndependentStreamForRepeatedNodeInvocation() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + llm.setChatModel(new DeterministicStreamChatModel()); + + Chain chain = new Chain(new ChainDefinition(), "stream-test"); + EventManager eventManager = new EventManager(); + chain.setEventManager(eventManager); + List events = new ArrayList<>(); + eventManager.addEventListener( + LlmStreamEvent.class, + (event, currentChain) -> events.add((LlmStreamEvent) event)); + + LlmNode node = new LlmNode(); + node.setId("llm-1"); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setMessage("请回答"); + + Assert.assertEquals( + "你好", + llm.chat(messageInfo, new Llm.ChatOptions(), node, chain)); + Assert.assertEquals( + "你好", + llm.chat(messageInfo, new Llm.ChatOptions(), node, chain)); + + Assert.assertEquals(4, events.size()); + Assert.assertEquals(List.of("你", "好", "你", "好"), events.stream() + .map(LlmStreamEvent::getDelta) + .toList()); + Assert.assertTrue(events.stream().noneMatch(LlmStreamEvent::isReasoning)); + Assert.assertEquals(events.get(0).getStreamId(), events.get(1).getStreamId()); + Assert.assertEquals(events.get(2).getStreamId(), events.get(3).getStreamId()); + Assert.assertNotEquals(events.get(0).getStreamId(), events.get(2).getStreamId()); + } + + /** + * 验证模型思考与正式回答使用同一流标识并按内容类型分别发布。 + */ + @Test + public void shouldEmitReasoningAndTextDeltasSeparately() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + llm.setChatModel(new ReasoningStreamChatModel()); + + Chain chain = new Chain(new ChainDefinition(), "reasoning-stream-test"); + EventManager eventManager = new EventManager(); + chain.setEventManager(eventManager); + List events = new ArrayList<>(); + eventManager.addEventListener( + LlmStreamEvent.class, + (event, currentChain) -> events.add((LlmStreamEvent) event)); + + LlmNode node = new LlmNode(); + node.setId("llm-reasoning"); + Llm.MessageInfo messageInfo = new Llm.MessageInfo(); + messageInfo.setMessage("请回答"); + + Assert.assertEquals( + "答案", + llm.chat(messageInfo, new Llm.ChatOptions(), node, chain)); + Assert.assertEquals(4, events.size()); + Assert.assertEquals( + List.of("先", "想", "答", "案"), + events.stream().map(LlmStreamEvent::getDelta).toList()); + Assert.assertEquals( + List.of( + LlmStreamEvent.ContentType.REASONING, + LlmStreamEvent.ContentType.REASONING, + LlmStreamEvent.ContentType.TEXT, + LlmStreamEvent.ContentType.TEXT + ), + events.stream().map(LlmStreamEvent::getContentType).toList()); + Assert.assertEquals( + 1, + events.stream().map(LlmStreamEvent::getStreamId).distinct().count()); + } + + /** + * 固定输出两个文本增量的测试聊天模型。 + */ + private static final class DeterministicStreamChatModel implements ChatModel { + + /** + * 同步聊天接口不参与本测试。 + * + * @param prompt 提示词 + * @param options 模型参数 + * @return 无 + */ + @Override + public AiMessageResponse chat(Prompt prompt, ChatOptions options) { + throw new UnsupportedOperationException("sync chat is not used"); + } + + /** + * 连续发送两个增量及完整消息。 + * + * @param prompt 提示词 + * @param listener 流监听器 + * @param options 模型参数 + */ + @Override + public void chatStream( + Prompt prompt, + StreamResponseListener listener, + ChatOptions options) { + StreamContext context = new StreamContext(this, null, new NoopStreamClient()); + listener.onStart(context); + for (String content : List.of("你", "好")) { + com.easyagents.core.message.AiMessage delta = + new com.easyagents.core.message.AiMessage(); + delta.setContent(content); + listener.onMessage(context, new AiMessageResponse(null, content, delta)); + } + context.setFullMessage(new com.easyagents.core.message.AiMessage("你好")); + listener.onStop(context); + } + } + + /** + * 固定输出思考与回答增量的测试聊天模型。 + */ + private static final class ReasoningStreamChatModel implements ChatModel { + + /** + * 同步聊天接口不参与本测试。 + * + * @param prompt 提示词 + * @param options 模型参数 + * @return 无 + */ + @Override + public AiMessageResponse chat(Prompt prompt, ChatOptions options) { + throw new UnsupportedOperationException("sync chat is not used"); + } + + /** + * 依次发送思考增量、回答增量及完整消息。 + * + * @param prompt 提示词 + * @param listener 流监听器 + * @param options 模型参数 + */ + @Override + public void chatStream( + Prompt prompt, + StreamResponseListener listener, + ChatOptions options) { + StreamContext context = new StreamContext(this, null, new NoopStreamClient()); + listener.onStart(context); + for (String reasoning : List.of("先", "想")) { + com.easyagents.core.message.AiMessage delta = + new com.easyagents.core.message.AiMessage(); + delta.setReasoningContent(reasoning); + listener.onMessage(context, new AiMessageResponse(null, reasoning, delta)); + } + for (String content : List.of("答", "案")) { + com.easyagents.core.message.AiMessage delta = + new com.easyagents.core.message.AiMessage(); + delta.setContent(content); + listener.onMessage(context, new AiMessageResponse(null, content, delta)); + } + com.easyagents.core.message.AiMessage fullMessage = + new com.easyagents.core.message.AiMessage("答案"); + fullMessage.setFullReasoningContent("先想"); + context.setFullMessage(fullMessage); + listener.onStop(context); + } + } + + /** + * 测试使用的空流客户端。 + */ + private static final class NoopStreamClient implements StreamClient { + + /** + * 测试模型直接分发事件，无需启动网络请求。 + * + * @param url 请求地址 + * @param headers 请求头 + * @param payload 请求体 + * @param listener 客户端监听器 + * @param config 模型配置 + */ + @Override + public void start( + String url, + Map headers, + String payload, + com.easyagents.core.model.client.StreamClientListener listener, + ChatConfig config) { + } + + /** + * 测试客户端没有需要关闭的网络资源。 + */ + @Override + public void stop() { + } + } +}