diff --git a/easy-agents-core/src/main/java/com/easyagents/core/model/client/BaseStreamClientListener.java b/easy-agents-core/src/main/java/com/easyagents/core/model/client/BaseStreamClientListener.java index 1edf2d2..3f40ca4 100644 --- a/easy-agents-core/src/main/java/com/easyagents/core/model/client/BaseStreamClientListener.java +++ b/easy-agents-core/src/main/java/com/easyagents/core/model/client/BaseStreamClientListener.java @@ -20,6 +20,7 @@ import com.easyagents.core.model.chat.ChatContext; import com.easyagents.core.model.chat.ChatModel; import com.easyagents.core.model.chat.StreamResponseListener; import com.easyagents.core.model.chat.response.AiMessageResponse; +import com.easyagents.core.model.exception.ModelException; import com.easyagents.core.parser.AiMessageParser; import com.easyagents.core.util.StringUtil; import com.alibaba.fastjson2.JSON; @@ -57,6 +58,7 @@ public class BaseStreamClientListener implements StreamClientListener { @Override public void onMessage(StreamClient client, String response) { + if (isFailure.get() || stoppedFlag.get()) return; if (StringUtil.noText(response) || "[DONE]".equalsIgnoreCase(response.trim()) || finishedFlag.get()) { notifyLastMessageAndStop(response); return; @@ -64,6 +66,12 @@ public class BaseStreamClientListener implements StreamClientListener { try { JSONObject jsonObject = JSON.parseObject(response); + if (jsonObject != null && jsonObject.get("error") != null) { + JSONObject error = jsonObject.get("error") instanceof JSONObject value ? value : new JSONObject(); + String code = error.getString("code"); + throw new ModelException(200, "Model stream error: " + response, null, + code, error.getString("type"), error.getString("message")); + } AiMessage delta = messageParser.parse(jsonObject, chatContext); //合并 增量 delta 到 fullMessage @@ -77,8 +85,15 @@ public class BaseStreamClientListener implements StreamClientListener { AiMessageResponse resp = new AiMessageResponse(chatContext, response, delta); streamResponseListener.onMessage(context, resp); } catch (Exception err) { - onFailure(this.context.getClient(), err); - onStop(this.context.getClient()); + try { + onFailure(client, err); + } finally { + try { + client.stop(); + } finally { + onStop(client); + } + } } } @@ -133,6 +148,7 @@ public class BaseStreamClientListener implements StreamClientListener { @Override public void onFailure(StreamClient client, Throwable throwable) { + if (stoppedFlag.get() || finishedFlag.get()) return; if (isFailure.compareAndSet(false, true)) { context.setThrowable(throwable); streamResponseListener.onFailure(context, throwable); diff --git a/easy-agents-core/src/main/java/com/easyagents/core/model/client/impl/Util.java b/easy-agents-core/src/main/java/com/easyagents/core/model/client/impl/Util.java index e01a672..dcd7480 100644 --- a/easy-agents-core/src/main/java/com/easyagents/core/model/client/impl/Util.java +++ b/easy-agents-core/src/main/java/com/easyagents/core/model/client/impl/Util.java @@ -15,6 +15,8 @@ */ package com.easyagents.core.model.client.impl; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; import com.easyagents.core.model.exception.ModelException; import com.easyagents.core.util.StringUtil; import okhttp3.Response; @@ -25,11 +27,14 @@ import java.io.IOException; class Util { public static Throwable getFailureThrowable(Throwable t, Response response) { - if (t != null) { + if (t != null && response == null) { return t; } if (response != null) { + String errorCode = null; + String errorType = null; + String errorMessage = null; String errMessage = "Response code: " + response.code(); String message = response.message(); if (StringUtil.hasText(message)) { @@ -40,12 +45,22 @@ class Util { String string = body.string(); if (StringUtil.hasText(string)) { errMessage += ", body: " + string; + try { + JSONObject payload = JSON.parseObject(string); + if (payload != null && payload.get("error") instanceof JSONObject error) { + errorCode = error.getString("code"); + errorType = error.getString("type"); + errorMessage = error.getString("message"); + } + } catch (RuntimeException ignored) { + // 网关可能返回 HTML 或不完整 JSON；仍保留 HTTP 状态和原始诊断。 + } } } } catch (IOException e) { // ignore } - t = new ModelException(errMessage); + t = new ModelException(response.code(), errMessage, t, errorCode, errorType, errorMessage); } return t; diff --git a/easy-agents-core/src/main/java/com/easyagents/core/model/exception/ModelException.java b/easy-agents-core/src/main/java/com/easyagents/core/model/exception/ModelException.java index 7691f71..c8f9c35 100644 --- a/easy-agents-core/src/main/java/com/easyagents/core/model/exception/ModelException.java +++ b/easy-agents-core/src/main/java/com/easyagents/core/model/exception/ModelException.java @@ -16,6 +16,38 @@ package com.easyagents.core.model.exception; public class ModelException extends RuntimeException { + private Integer statusCode; + private String errorCode; + private String errorType; + private String errorMessage; + + public ModelException(int statusCode, String message, Throwable cause) { + this(statusCode, message, cause, null, null, null); + } + + public ModelException(int statusCode, String message, Throwable cause, String errorCode, String errorType, String errorMessage) { + super(message, cause); + this.statusCode = statusCode; + this.errorCode = errorCode; + this.errorType = errorType; + this.errorMessage = errorMessage; + } + + public Integer getStatusCode() { + return statusCode; + } + + public String getErrorCode() { + return errorCode; + } + + public String getErrorMessage() { + return errorMessage; + } + + public String getErrorType() { + return errorType; + } /** * Constructs a new runtime exception with {@code null} as its diff --git a/easy-agents-core/src/test/java/com/easyagents/core/model/client/StreamFailureLifecycleTest.java b/easy-agents-core/src/test/java/com/easyagents/core/model/client/StreamFailureLifecycleTest.java new file mode 100644 index 0000000..92ab327 --- /dev/null +++ b/easy-agents-core/src/test/java/com/easyagents/core/model/client/StreamFailureLifecycleTest.java @@ -0,0 +1,52 @@ +package com.easyagents.core.model.client; + +import com.easyagents.core.message.AiMessage; +import com.easyagents.core.model.chat.ChatConfig; +import com.easyagents.core.model.chat.ChatContext; +import com.easyagents.core.model.chat.StreamResponseListener; +import com.easyagents.core.model.chat.response.AiMessageResponse; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public class StreamFailureLifecycleTest { + @Test + public void completedStreamMustIgnoreLateFailureAndData() { + Fixture fixture = new Fixture(); + fixture.listener.onMessage(fixture.client, "{}"); + fixture.listener.onMessage(fixture.client, "[DONE]"); + fixture.listener.onFailure(fixture.client, new IllegalStateException("late transport error")); + fixture.listener.onMessage(fixture.client, "{}"); + Assert.assertEquals(List.of("message", "message", "stop"), fixture.events); + } + + @Test + public void failedStreamMustStopWithoutLaterSuccess() { + Fixture fixture = new Fixture(); + fixture.listener.onMessage(fixture.client, "{}"); + fixture.listener.onMessage(fixture.client, "{\"error\":{\"code\":\"rate_limit_exceeded\"}}"); + fixture.listener.onMessage(fixture.client, "[DONE]"); + fixture.listener.onMessage(fixture.client, "{}"); + fixture.listener.onFailure(fixture.client, new IllegalStateException("cancelled")); + Assert.assertEquals(List.of("message", "failure", "transport-stop", "stop"), fixture.events); + } + + private static class Fixture { + final List events = new ArrayList<>(); + final StreamClient client = new StreamClient() { + public void start(String url, Map headers, String payload, StreamClientListener listener, ChatConfig config) { } + public void stop() { events.add("transport-stop"); } + }; + final BaseStreamClientListener listener = new BaseStreamClientListener(null, new ChatContext(), client, + new StreamResponseListener() { + public void onMessage(StreamContext context, AiMessageResponse response) { events.add("message"); } + public void onFailure(StreamContext context, Throwable error) { events.add("failure"); } + public void onStop(StreamContext context) { events.add("stop"); } + }, (json, context) -> { + AiMessage message = new AiMessage(); message.setContent("partial"); return message; + }); + } +} diff --git a/easy-agents-core/src/test/java/com/easyagents/core/model/client/impl/ModelHttpFailureTest.java b/easy-agents-core/src/test/java/com/easyagents/core/model/client/impl/ModelHttpFailureTest.java new file mode 100644 index 0000000..ab0c797 --- /dev/null +++ b/easy-agents-core/src/test/java/com/easyagents/core/model/client/impl/ModelHttpFailureTest.java @@ -0,0 +1,39 @@ +package com.easyagents.core.model.client.impl; + +import com.easyagents.core.model.exception.ModelException; +import okhttp3.MediaType; +import okhttp3.Protocol; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.ResponseBody; +import org.junit.Assert; +import org.junit.Test; + +public class ModelHttpFailureTest { + @Test + public void shouldPreserveStructuredErrorFromActualProviderShape() { + ModelException error = failure("{\"error\":{\"message\":\"Model not found\",\"code\":404,\"type\":\"NotFound\"}}"); + Assert.assertEquals(Integer.valueOf(404), error.getStatusCode()); + Assert.assertEquals("404", error.getErrorCode()); + Assert.assertEquals("NotFound", error.getErrorType()); + Assert.assertEquals("Model not found", error.getErrorMessage()); + Assert.assertTrue(error.getMessage().contains("Model not found")); + } + + @Test + public void shouldPreserveHttpStatusForNonModelGatewayErrors() { + for (String body : new String[]{"Not Found", "{broken", "{\"error\":\"not found\"}", "null"}) { + ModelException error = failure(body); + Assert.assertEquals(Integer.valueOf(404), error.getStatusCode()); + Assert.assertNull(error.getErrorCode()); + Assert.assertNull(error.getErrorMessage()); + } + } + + private ModelException failure(String body) { + Response response = new Response.Builder().request(new Request.Builder().url("http://localhost/chat").build()) + .protocol(Protocol.HTTP_1_1).code(404).message("Not Found") + .body(ResponseBody.create(MediaType.get("application/json"), body)).build(); + return (ModelException) Util.getFailureThrowable(null, response); + } +} 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 b9a5edf..9a76b1c 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 @@ -612,6 +612,8 @@ public class Chain { Map nodeResult = null; Throwable error = null; String executionAttemptKey = null; + String auditInstanceId = StringUtil.hasText(chainState.getAuditInstanceId()) + ? chainState.getAuditInstanceId() : stateInstanceId; try { AtomicBoolean nodeStarted = new AtomicBoolean(); String candidateAttemptKey = @@ -647,12 +649,13 @@ public class Chain { activeNodeState .getExecutionAttemptKey(); if (nodeStarted.get()) { + auditInstanceId = getAuditInstanceId(); notifyEvent(new NodeStartEvent( this, node, executionAttemptKey, activeNodeState.getStatus(), - getAuditInstanceId())); + auditInstanceId)); } ChainState nodeExecutionState = updateStateSafely(state -> { @@ -672,7 +675,10 @@ public class Chain { } catch (TriggerClaimLostException | RetryableTriggerException claimLost) { throw claimLost; } catch (Throwable throwable) { - log.error("Node execute error", throwable); + if (!(throwable instanceof ChainSuspendException)) { + log.error("Node execute error, executeId={}, chainInstanceId={}, nodeId={}, nodeName={}, attemptKey={}", + auditInstanceId, stateInstanceId, node.getId(), node.getName(), executionAttemptKey, throwable); + } error = throwable; } // 结果提交入口会在实例锁内重读状态，统一拦截取消、超时和其他终态。 @@ -882,6 +888,10 @@ public class Chain { NodeStatus finalNodeStatus = null; try { if (error == null) { + updateNodeStateSafely(node.id, state -> { + state.setError(null); + return EnumSet.of(NodeStateField.ERROR); + }); // 更新 state 数据 updateStateSafely(state -> { EnumSet fields = EnumSet.of(ChainStateField.EXECUTE_RESULT); @@ -941,9 +951,10 @@ public class Chain { } // 失败 else { + finalNodeStatus = NodeStatus.ERROR; NodeState newState = updateNodeStateSafely(node.getId(), s -> { s.setStatus(NodeStatus.ERROR); - s.setError(new ExceptionSummary(error)); + s.setError(new ExceptionSummary(error, stateInstanceId, node.getId(), node.getName())); return EnumSet.of(NodeStateField.ERROR, NodeStateField.STATUS); }); @@ -960,7 +971,8 @@ public class Chain { scheduleNode(node, triggerEdgeId, TriggerType.RETRY, node.getRetryIntervalMs()); } else { - finalChainStatus = handleNodeError(node.id, error); + finalNodeStatus = NodeStatus.FAILED; + finalChainStatus = handleNodeError(node, error); } } } @@ -992,10 +1004,15 @@ public class Chain { // 更新父级链的状态 if (!finalChainStatus.isSuccess()) { ChainState currentState = getState(); + ExceptionSummary originError = currentState.getError(); ChainStatus currentStatus = finalChainStatus; while (currentState != null && StringUtil.hasText(currentState.getParentInstanceId())) { updateStateSafely(currentState.getParentInstanceId(), state -> { state.setStatus(currentStatus); + if (state.getError() == null && originError != null) { + state.setError(originError); + return EnumSet.of(ChainStateField.STATUS, ChainStateField.ERROR); + } return EnumSet.of(ChainStateField.STATUS); }); setStatusAndNotifyEvent(currentState.getParentInstanceId(), currentStatus); @@ -1661,7 +1678,7 @@ public class Chain { } before.set(state.getStatus()); state.setStatus(ChainStatus.FAILED); - state.setError(new ExceptionSummary(failure)); + state.setError(new ExceptionSummary(failure, stateInstanceId, null, null)); state.setMessage(failure.getMessage()); changed.set(true); return EnumSet.of( @@ -1693,19 +1710,23 @@ public class Chain { } - private ChainStatus handleNodeError(String nodeId, Throwable throwable) { - updateNodeStateSafely(nodeId, s -> { + private ChainStatus handleNodeError(Node node, Throwable throwable) { + ExceptionSummary summary = new ExceptionSummary(throwable, stateInstanceId, node.getId(), node.getName()); + updateNodeStateSafely(node.getId(), s -> { s.setStatus(NodeStatus.FAILED); - s.setError(new ExceptionSummary(throwable)); + s.setError(summary); return EnumSet.of(NodeStateField.ERROR, NodeStateField.STATUS); }); updateStateSafely(state -> { - state.setError(new ExceptionSummary(throwable)); + if (state.getError() != null) { + return null; + } + state.setError(summary); return EnumSet.of(ChainStateField.ERROR); }); - setStatusAndNotifyEvent(ChainStatus.FAILED); + // 节点结束事件先于工作流终态，终态订阅者关闭连接前能收到失败节点。 deferOrRun(() -> eventManager.notifyChainError(throwable, this)); return ChainStatus.FAILED; } diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainState.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainState.java index 1370749..d3ef200 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainState.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainState.java @@ -596,63 +596,71 @@ public class ChainState implements Serializable { List suspendParameters = null; List> templateRootMaps = null; for (Parameter parameter : parameters) { - RefType refType = parameter.getRefType(); - Object value = null; - if (refType == RefType.FIXED) { - if (templateRootMaps == null) { - templateRootMaps = - preserveDirectReferences - ? buildLazyTemplateRootMaps( - formatArgs) - : buildTemplateRootMaps( - formatArgs); - } - value = TextTemplate.of(parameter.getValue()) - .formatToString(templateRootMaps); - } else if (refType == RefType.REF) { - value = this.resolveValue( - parameter.getRef(), - preserveDirectReferences); - } - // 单节点执行时，参数只会传入 name 内容。 - if (value == null) { - value = this.resolveValue( - parameter.getName(), - preserveDirectReferences); - } - - if (value == null && parameter.getDefaultValue() != null) { - value = parameter.getDefaultValue(); - } - - if (refType == RefType.INPUT && isNullOrBlank(value)) { - if (!ignoreRequired && parameter.isRequired()) { - if (suspendParameters == null) { - suspendParameters = new ArrayList<>(); + try { + RefType refType = parameter.getRefType(); + Object value = null; + if (refType == RefType.FIXED) { + if (templateRootMaps == null) { + templateRootMaps = + preserveDirectReferences + ? buildLazyTemplateRootMaps( + formatArgs) + : buildTemplateRootMaps( + formatArgs); } - suspendParameters.add(parameter); - continue; + value = TextTemplate.of(parameter.getValue()) + .formatToString(templateRootMaps); + } else if (refType == RefType.REF) { + value = this.resolveValue( + parameter.getRef(), + preserveDirectReferences); } - } - - if (parameter.isRequired() && isNullOrBlank(value)) { - if (!ignoreRequired) { - throw new ChainException(node.getName() + " Missing required parameter:" + parameter.getName()); + // 单节点执行时，参数只会传入 name 内容。 + if (value == null) { + value = this.resolveValue( + parameter.getName(), + preserveDirectReferences); } - } - if (value instanceof String) { - value = ((String) value).trim(); - if (parameter.getDataType() == DataType.Boolean) { - value = "true".equalsIgnoreCase((String) value) || "1".equalsIgnoreCase((String) value); - } else if (parameter.getDataType() == DataType.Number) { - value = Long.parseLong((String) value); - } else if (parameter.getDataType() == DataType.Array) { - value = JSON.parseArray((String) value); + if (value == null && parameter.getDefaultValue() != null) { + value = parameter.getDefaultValue(); } - } - variables.put(parameter.getName(), value); + if (refType == RefType.INPUT && isNullOrBlank(value)) { + if (!ignoreRequired && parameter.isRequired()) { + if (suspendParameters == null) { + suspendParameters = new ArrayList<>(); + } + suspendParameters.add(parameter); + continue; + } + } + + if (parameter.isRequired() && isNullOrBlank(value)) { + if (!ignoreRequired) { + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, + node.getName() + " Missing required parameter:" + parameter.getName()); + } + } + + if (value instanceof String) { + value = ((String) value).trim(); + if (parameter.getDataType() == DataType.Boolean) { + value = "true".equalsIgnoreCase((String) value) || "1".equalsIgnoreCase((String) value); + } else if (parameter.getDataType() == DataType.Number) { + value = Long.parseLong((String) value); + } else if (parameter.getDataType() == DataType.Array) { + value = JSON.parseArray((String) value); + } + } + + variables.put(parameter.getName(), value); + } catch (WorkflowExecutionException failure) { + throw failure; + } catch (RuntimeException failure) { + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, + "Cannot resolve parameter: " + parameter.getName(), failure); + } } if (suspendParameters != null && !suspendParameters.isEmpty()) { diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ExceptionSummary.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ExceptionSummary.java index 8438cd5..4b1b6df 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ExceptionSummary.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ExceptionSummary.java @@ -18,6 +18,9 @@ package com.easyagents.flow.core.chain; import java.io.PrintWriter; import java.io.Serializable; import java.io.StringWriter; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; public class ExceptionSummary implements Serializable { private static final long serialVersionUID = 1L; @@ -31,12 +34,20 @@ public class ExceptionSummary implements Serializable { private String chainId; private String nodeId; + private String nodeName; private String errorCode; // 可选 private long timestamp; public ExceptionSummary(Throwable error) { + this(error, null, null, null); + } + + public ExceptionSummary(Throwable error, String chainId, String nodeId, String nodeName) { + this.chainId = chainId; + this.nodeId = nodeId; + this.nodeName = nodeName; this.exceptionClass = error.getClass().getName(); this.message = error.getMessage(); this.stackTrace = getStackTraceAsString(error); @@ -46,11 +57,30 @@ public class ExceptionSummary implements Serializable { this.rootCauseMessage = root.getMessage(); this.timestamp = System.currentTimeMillis(); + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable cause = error; cause != null && seen.add(cause); cause = cause.getCause()) { + if (cause instanceof WorkflowExecutionException failure) { + if (this.errorCode == null && failure.getReason() != null) { + this.errorCode = failure.getReason().getCode(); + } + if (failure.getNodeId() != null) { + this.chainId = failure.getChainId(); + this.nodeId = failure.getNodeId(); + this.nodeName = failure.getNodeName(); + } + } + } + if (this.errorCode == null) { + this.errorCode = (this.nodeId == null ? WorkflowErrorReason.WORKFLOW_INTERNAL_ERROR + : WorkflowErrorReason.NODE_EXECUTION_FAILED).getCode(); + } } private static Throwable getRootCause(Throwable t) { Throwable result = t; - while (result.getCause() != null) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + seen.add(result); + while (result.getCause() != null && seen.add(result.getCause())) { result = result.getCause(); } return result; @@ -123,6 +153,10 @@ public class ExceptionSummary implements Serializable { return errorCode; } + public String getNodeName() { return nodeName; } + + public void setNodeName(String nodeName) { this.nodeName = nodeName; } + public void setErrorCode(String errorCode) { this.errorCode = errorCode; } diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowErrorReason.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowErrorReason.java new file mode 100644 index 0000000..f7168d7 --- /dev/null +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowErrorReason.java @@ -0,0 +1,37 @@ +package com.easyagents.flow.core.chain; + +/** 工作流对外稳定的失败原因；文案不包含底层响应、凭据或业务输入。 */ +public enum WorkflowErrorReason { + INPUT_INVALID("输入参数无效，请检查必填参数和变量引用"), + MODEL_NOT_FOUND("模型不存在，请检查模型名称或重新选择模型"), + MODEL_RATE_LIMITED("模型调用受到限流，请稍后重试"), + MODEL_AUTH_FAILED("模型鉴权失败，请检查模型凭据和访问权限"), + MODEL_UNAVAILABLE("模型服务暂不可用，请稍后重试"), + MODEL_TIMEOUT("模型响应超时，请稍后重试"), + NODE_OUTPUT_INVALID("节点输出为空或格式不符合要求，请检查输出配置"), + NODE_EXECUTION_FAILED("节点执行失败"), + WORKFLOW_INTERNAL_ERROR("工作流内部执行异常"); + + private final String defaultMessage; + + WorkflowErrorReason(String defaultMessage) { + this.defaultMessage = defaultMessage; + } + + public String getCode() { + return name(); + } + + public String getDefaultMessage() { + return defaultMessage; + } + + public static WorkflowErrorReason fromCode(String code) { + for (WorkflowErrorReason reason : values()) { + if (reason.getCode().equals(code)) { + return reason; + } + } + return null; + } +} diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowExecutionException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowExecutionException.java new file mode 100644 index 0000000..be7cc8f --- /dev/null +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowExecutionException.java @@ -0,0 +1,30 @@ +package com.easyagents.flow.core.chain; + +/** 保留原始 cause，并在已知业务边界标注失败原因。 */ +public class WorkflowExecutionException extends ChainException { + private final WorkflowErrorReason reason; + private String chainId; + private String nodeId; + private String nodeName; + + public WorkflowExecutionException(WorkflowErrorReason reason, String message) { + this(reason, message, null); + } + + public WorkflowExecutionException(WorkflowErrorReason reason, String message, Throwable cause) { + super(message, cause); + this.reason = reason; + } + + public WorkflowExecutionException withContext(String chainId, String nodeId, String nodeName) { + this.chainId = chainId; + this.nodeId = nodeId; + this.nodeName = nodeName; + return this; + } + + public WorkflowErrorReason getReason() { return reason; } + public String getChainId() { return chainId; } + public String getNodeId() { return nodeId; } + public String getNodeName() { return nodeName; } +} diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/NodeEndEvent.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/NodeEndEvent.java index cfecc8e..0188aa7 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/NodeEndEvent.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/NodeEndEvent.java @@ -19,6 +19,8 @@ package com.easyagents.flow.core.chain.event; import com.easyagents.flow.core.chain.Chain; import com.easyagents.flow.core.chain.Node; import com.easyagents.flow.core.chain.NodeStatus; +import com.easyagents.flow.core.chain.ExceptionSummary; +import com.easyagents.flow.core.chain.ChainSuspendException; import java.util.Map; @@ -30,6 +32,7 @@ public class NodeEndEvent extends BaseEvent { private final Node node; private final Map result; private final Throwable error; + private final ExceptionSummary errorSummary; private final NodeStatus status; private final String executionAttemptKey; @@ -88,6 +91,8 @@ public class NodeEndEvent extends BaseEvent { this.node = node; this.result = result; this.error = error; + this.errorSummary = error == null || error instanceof ChainSuspendException ? null + : new ExceptionSummary(error, chain.getStateInstanceId(), node.getId(), node.getName()); this.status = status; this.executionAttemptKey = executionAttemptKey; } @@ -119,6 +124,8 @@ public class NodeEndEvent extends BaseEvent { return error; } + public ExceptionSummary getErrorSummary() { return errorSummary; } + /** * 获取事件创建时捕获的节点终态。 * 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 ba130f3..79575c1 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 @@ -898,6 +898,16 @@ public class ChainExecutor { }); } return node.execute(temp); + } catch (ChainSuspendException | TriggerClaimLostException | RetryableTriggerException control) { + throw control; + } catch (RuntimeException failure) { + log.error("Single node execute error, executeId={}, chainInstanceId={}, nodeId={}, nodeName={}, attemptKey={}", + temp.getStateInstanceId(), temp.getStateInstanceId(), nodeId, node.getName(), + temp.getStateInstanceId() + ":" + nodeId + ":single", failure); + ExceptionSummary summary = new ExceptionSummary(failure, temp.getStateInstanceId(), node.getId(), node.getName()); + throw new WorkflowExecutionException(WorkflowErrorReason.fromCode(summary.getErrorCode()), + "Single node execution failed", failure) + .withContext(temp.getStateInstanceId(), summary.getNodeId(), summary.getNodeName()); } finally { activeDefinitions.remove(temp.getStateInstanceId()); definitionSnapshotRepository.remove(temp.getStateInstanceId()); 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 f2e3e0f..4e4f9d0 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 @@ -19,6 +19,8 @@ import com.alibaba.fastjson.JSON; import com.easyagents.flow.core.chain.Chain; import com.easyagents.flow.core.chain.ChainState; import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; import com.easyagents.flow.core.llm.Llm; import com.easyagents.flow.core.llm.LlmManager; import com.easyagents.flow.core.util.*; @@ -96,23 +98,21 @@ public class LlmNode extends BaseNode { chainState.resolveParameters(this); if (StringUtil.noText(userPrompt)) { - throw new RuntimeException("Can not find user prompt"); + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "Can not find user prompt"); } List> templateRootMaps = chainState.buildTemplateRootMaps( parameterValues); - String userPromptString = TextTemplate.of(userPrompt) - .formatToString(templateRootMaps); + String userPromptString = formatPrompt(userPrompt, templateRootMaps); Llm llm = LlmManager.getInstance().getChatModel(this.llmId); if (llm == null) { - throw new RuntimeException("Can not find llm: " + this.llmId); + throw new WorkflowExecutionException(WorkflowErrorReason.MODEL_NOT_FOUND, "Can not find llm: " + this.llmId); } - String systemPromptString = TextTemplate.of(this.systemPrompt) - .formatToString(templateRootMaps); + String systemPromptString = formatPrompt(this.systemPrompt, templateRootMaps); Llm.MessageInfo messageInfo = new Llm.MessageInfo(); messageInfo.setMessage(userPromptString); @@ -130,7 +130,7 @@ public class LlmNode extends BaseNode { if (!(value instanceof String) && !(value instanceof java.io.File) && !(value instanceof Map)) { - throw new IllegalArgumentException( + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "Unsupported image input for parameter '" + name + "': " + value.getClass().getName()); } @@ -143,7 +143,7 @@ public class LlmNode extends BaseNode { String responseContent = llm.chat(messageInfo, chatOptions, this, chain); if (StringUtil.noText(responseContent)) { - throw new RuntimeException("Can not get response from llm"); + throw new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "Can not get response from llm"); } else { responseContent = responseContent.trim(); } @@ -154,7 +154,10 @@ public class LlmNode extends BaseNode { try { jsonObjectOrArray = JSON.parse(unWrapMarkdown(responseContent)); } catch (Exception e) { - throw new RuntimeException("Can not parse json: " + responseContent + " " + e.getMessage()); + throw new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "Can not parse model JSON output", e); + } + if (jsonObjectOrArray == null) { + throw new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "Model JSON output is empty"); } if (CollectionUtil.noItems(this.outputDefs)) { @@ -173,6 +176,14 @@ public class LlmNode extends BaseNode { } } + private String formatPrompt(String template, List> values) { + try { + return TextTemplate.of(template).formatToString(values); + } catch (RuntimeException failure) { + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "Cannot resolve prompt variables", failure); + } + } + /** * 移除 ``` 或者 ```json 等 diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/WorkflowFailureContractTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/WorkflowFailureContractTest.java new file mode 100644 index 0000000..1764dc9 --- /dev/null +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/WorkflowFailureContractTest.java @@ -0,0 +1,161 @@ +package com.easyagents.flow.core.test; + +import com.easyagents.flow.core.chain.*; +import com.easyagents.flow.core.chain.event.*; +import com.easyagents.flow.core.chain.repository.*; +import com.easyagents.flow.core.chain.runtime.*; +import com.easyagents.flow.core.node.StartNode; +import com.easyagents.flow.core.node.LlmNode; +import com.easyagents.flow.core.llm.LlmProvider; +import com.easyagents.flow.core.llm.LlmManager; +import com.easyagents.flow.core.node.EndNode; +import org.junit.Assert; +import org.junit.Test; + +import java.io.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +public class WorkflowFailureContractTest { + @Test + public void failedNodeMustEndBeforeSingleTerminalEvent() throws Exception { + try (Fixture fixture = new Fixture(Integer.MAX_VALUE, false)) { + fixture.run(); + Assert.assertEquals(ChainStatus.FAILED, fixture.states.load(fixture.id).getStatus()); + Assert.assertEquals(List.of("FAILED", "terminal"), fixture.events); + ExceptionSummary error = fixture.states.load(fixture.id).getError(); + Assert.assertEquals("worker", error.getNodeId()); + Assert.assertEquals("模型分析", error.getNodeName()); + Assert.assertEquals(WorkflowErrorReason.MODEL_RATE_LIMITED.getCode(), error.getErrorCode()); + Assert.assertEquals(NodeStatus.FAILED, fixture.nodes.load(fixture.id, "worker").getStatus()); + } + } + + @Test + public void retrySuccessMustClearErrorAndKeepAttemptStates() throws Exception { + try (Fixture fixture = new Fixture(1, true)) { + fixture.run(); + Assert.assertEquals(ChainStatus.SUCCEEDED, fixture.states.load(fixture.id).getStatus()); + Assert.assertEquals(List.of("ERROR", "SUCCEEDED", "terminal"), fixture.events); + Assert.assertNull(fixture.nodes.load(fixture.id, "worker").getError()); + Assert.assertNull(fixture.states.load(fixture.id).getError()); + } + } + + @Test + public void retryExhaustionMustFailWithoutSuccessEvent() throws Exception { + try (Fixture fixture = new Fixture(Integer.MAX_VALUE, true)) { + fixture.run(); + Assert.assertEquals(List.of("ERROR", "FAILED", "terminal"), fixture.events); + } + } + + @Test + public void summaryMustPreserveTypedCauseContextAndSerialization() throws Exception { + for (WorkflowErrorReason reason : WorkflowErrorReason.values()) { + ExceptionSummary summary = new ExceptionSummary(new RuntimeException(new WorkflowExecutionException( + reason, "internal", new IOException("raw body"))), "chain", "node", "分析"); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream output = new ObjectOutputStream(bytes)) { output.writeObject(summary); } + try (ObjectInputStream input = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + ExceptionSummary restored = (ExceptionSummary) input.readObject(); + Assert.assertEquals(reason.getCode(), restored.getErrorCode()); + Assert.assertEquals("node", restored.getNodeId()); + Assert.assertEquals("分析", restored.getNodeName()); + Assert.assertEquals(IOException.class.getName(), restored.getRootCauseClass()); + } + } + Assert.assertEquals("WORKFLOW_INTERNAL_ERROR", new ExceptionSummary(new IllegalStateException()).getErrorCode()); + Assert.assertEquals("NODE_EXECUTION_FAILED", new ExceptionSummary(new IllegalStateException(), "c", "n", "N").getErrorCode()); + } + + @Test + public void singleNodeFailureMustCarryReasonAndNode() { + try (Fixture fixture = new Fixture(1, false)) { + WorkflowExecutionException failure = Assert.assertThrows(WorkflowExecutionException.class, + () -> fixture.executor.executeNode("test", "worker", Map.of())); + Assert.assertEquals(WorkflowErrorReason.MODEL_RATE_LIMITED, failure.getReason()); + Assert.assertEquals("worker", failure.getNodeId()); + } + } + + @Test + public void inputParsingAndModelOutputParsingMustHaveDifferentReasons() { + Chain chain = new Chain(new ChainDefinition(), "input-output-test"); + chain.setChainStateRepository(new InMemoryChainStateRepository()); + chain.setNodeStateRepository(new InMemoryNodeStateRepository()); + ChainState state = chain.initializeState(); + LlmNode node = new LlmNode(); node.setId("llm"); node.setName("模型分析"); + Parameter parameter = new Parameter("items"); parameter.setRefType(RefType.REF); + parameter.setRef("items"); parameter.setDataType(DataType.Array); + state.getMemory().put("items", "not-json"); + WorkflowExecutionException input = Assert.assertThrows(WorkflowExecutionException.class, + () -> state.resolveParameters(node, List.of(parameter))); + Assert.assertEquals(WorkflowErrorReason.INPUT_INVALID, input.getReason()); + node.setUserPrompt("test"); node.setLlmId("error-contract-model"); node.setOutType("json"); + LlmProvider provider = id -> "error-contract-model".equals(id) + ? (message, options, n, c) -> "not-json" : null; + LlmManager.getInstance().registerProvider(provider); + try { + WorkflowExecutionException output = Assert.assertThrows(WorkflowExecutionException.class, () -> node.execute(chain)); + Assert.assertEquals(WorkflowErrorReason.NODE_OUTPUT_INVALID, output.getReason()); + Assert.assertNotNull(output.getCause()); + } finally { LlmManager.getInstance().removeProvider(provider); } + } + + @Test + public void missingRegisteredModelMustHaveItsOwnReason() { + Chain chain = new Chain(new ChainDefinition(), "missing-model-test"); + chain.setChainStateRepository(new InMemoryChainStateRepository()); + chain.setNodeStateRepository(new InMemoryNodeStateRepository()); + chain.initializeState(); + LlmNode node = new LlmNode(); + node.setId("llm"); node.setUserPrompt("test"); node.setLlmId("absent-model-" + UUID.randomUUID()); + WorkflowExecutionException error = Assert.assertThrows(WorkflowExecutionException.class, () -> node.execute(chain)); + Assert.assertEquals(WorkflowErrorReason.MODEL_NOT_FOUND, error.getReason()); + } + + private static class Fixture implements AutoCloseable { + final InMemoryChainStateRepository states = new InMemoryChainStateRepository(); + final InMemoryNodeStateRepository nodes = new InMemoryNodeStateRepository(); + final TriggerScheduler scheduler = new TriggerScheduler(new InMemoryTriggerStore(), + Executors.newSingleThreadScheduledExecutor(), Executors.newFixedThreadPool(3), 1000); + final List events = new CopyOnWriteArrayList<>(); + final CountDownLatch ended = new CountDownLatch(1); + final ChainExecutor executor; + String id; + + Fixture(int failCount, boolean retry) { + ChainDefinition definition = new ChainDefinition(); + definition.setId("test"); + StartNode start = new StartNode(); start.setId("start"); definition.addNode(start); + Node worker = new Node() { + final AtomicInteger attempts = new AtomicInteger(); + @Override public Map execute(Chain chain) { + if (attempts.incrementAndGet() <= failCount) { + throw new WorkflowExecutionException(WorkflowErrorReason.MODEL_RATE_LIMITED, "internal provider body"); + } + return Map.of("output", "ok"); + } + }; + worker.setId("worker"); worker.setName("模型分析"); worker.setRetryEnable(retry); + worker.setMaxRetryCount(1); worker.setRetryIntervalMs(5); definition.addNode(worker); + EndNode end = new EndNode(); end.setId("end"); definition.addNode(end); + for (String[] pair : List.of(new String[]{"start", "worker"}, new String[]{"worker", "end"})) { + Edge edge = new Edge(); edge.setId(pair[0] + pair[1]); edge.setSource(pair[0]); edge.setTarget(pair[1]); definition.addEdge(edge); + } + executor = new ChainExecutor(ignored -> definition, states, nodes, scheduler); + executor.addEventListener((event, chain) -> { + if (event instanceof NodeEndEvent node && node.getNode().getId().equals("worker")) events.add(node.getStatus().name()); + if (event instanceof ChainStatusChangeEvent status && status.getStatus().isTerminal()) events.add("terminal"); + if (event instanceof ChainEndEvent) ended.countDown(); + }); + } + void run() throws InterruptedException { + id = executor.executeAsync("test", Map.of()); + Assert.assertTrue("workflow should finish", ended.await(5, TimeUnit.SECONDS)); + } + @Override public void close() { scheduler.shutdown(); } + } +} 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 d09d295..ad61d85 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 @@ -5,12 +5,15 @@ 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.exception.ModelException; 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.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent; import com.easyagents.flow.core.chain.event.LlmStreamEvent; import com.easyagents.flow.core.chain.listener.ChainEventListener; @@ -23,14 +26,25 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.UUID; +import java.util.IdentityHashMap; +import java.util.Set; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.util.concurrent.TimeoutException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; /** * 基于 Easy-Agents 聊天模型实现工作流 LLM 调用。 */ public class EasyAgentsLlm implements Llm { + private static final Pattern MODEL_NOT_FOUND_MESSAGE = Pattern.compile( + "^(?:the\\s+)?model(?:\\s+([`'\"])[^\\r\\n]+\\1)?\\s+(?:not found|does not exist)[.!]?$", + Pattern.CASE_INSENSITIVE); + private ChatModel chatModel; private ImageInputResolver imageInputResolver; @@ -163,7 +177,7 @@ public class EasyAgentsLlm implements Llm { if (message == null || StringUtil.noText(message.getFullContent())) { failure.compareAndSet( null, - new IllegalStateException( + new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "EasyAgentsLlm can not get aiMessage!")); } else { result.set(message.getFullContent()); @@ -191,6 +205,8 @@ public class EasyAgentsLlm implements Llm { } }, chatOptions); awaitCompletion(completion, streamContext); + } catch (RuntimeException exception) { + throw modelFailure(exception); } finally { chain.getEventManager().removeEventListener( ChainStatusChangeEvent.class, cancellationListener); @@ -198,14 +214,66 @@ public class EasyAgentsLlm implements Llm { Throwable throwable = failure.get(); if (throwable != null) { - throw new RuntimeException("EasyAgentsLlm stream failed", throwable); + throw modelFailure(throwable); } if (StringUtil.noText(result.get())) { - throw new RuntimeException("EasyAgentsLlm can not get response!"); + throw new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "EasyAgentsLlm can not get response!"); } return result.get(); } + static WorkflowExecutionException modelFailure(Throwable error) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + WorkflowErrorReason reason = WorkflowErrorReason.NODE_EXECUTION_FAILED; + for (Throwable cause = error; cause != null && seen.add(cause); cause = cause.getCause()) { + if (cause instanceof WorkflowExecutionException known) { + return known; + } + if (cause instanceof ModelException model && model.getStatusCode() != null) { + int status = model.getStatusCode(); + if (status == 429) reason = WorkflowErrorReason.MODEL_RATE_LIMITED; + else if (status == 401 || status == 403) reason = WorkflowErrorReason.MODEL_AUTH_FAILED; + else if (status == 408 || status == 504) reason = WorkflowErrorReason.MODEL_TIMEOUT; + else { + reason = reasonFromModelCode(model.getErrorCode()); + if (reason == WorkflowErrorReason.NODE_EXECUTION_FAILED) reason = reasonFromModelCode(model.getErrorType()); + if (reason == WorkflowErrorReason.NODE_EXECUTION_FAILED) { + if ((status == 200 || status == 400 || status == 404) && isModelNotFound(model)) { + reason = WorkflowErrorReason.MODEL_NOT_FOUND; + } else if (status >= 500 && status <= 599) { + reason = WorkflowErrorReason.MODEL_UNAVAILABLE; + } + } + } + } else if (cause instanceof SocketTimeoutException || cause instanceof TimeoutException) { + reason = WorkflowErrorReason.MODEL_TIMEOUT; + } else if (cause instanceof SocketException || cause instanceof UnknownHostException) { + reason = WorkflowErrorReason.MODEL_UNAVAILABLE; + } + if (reason != WorkflowErrorReason.NODE_EXECUTION_FAILED) break; + } + return new WorkflowExecutionException(reason, "EasyAgentsLlm stream failed", error); + } + + private static WorkflowErrorReason reasonFromModelCode(String code) { + if (code == null) return WorkflowErrorReason.NODE_EXECUTION_FAILED; + return switch (code.toLowerCase(java.util.Locale.ROOT)) { + case "model_not_found" -> WorkflowErrorReason.MODEL_NOT_FOUND; + case "rate_limit_exceeded", "rate_limit_error" -> WorkflowErrorReason.MODEL_RATE_LIMITED; + case "invalid_api_key", "authentication_error", "permission_denied", "permission_error" -> WorkflowErrorReason.MODEL_AUTH_FAILED; + case "service_unavailable", "overloaded_error" -> WorkflowErrorReason.MODEL_UNAVAILABLE; + case "request_timeout", "timeout" -> WorkflowErrorReason.MODEL_TIMEOUT; + default -> WorkflowErrorReason.NODE_EXECUTION_FAILED; + }; + } + + private static boolean isModelNotFound(ModelException error) { + if ("model_not_found".equalsIgnoreCase(error.getErrorCode())) return true; + String message = error.getErrorMessage(); + // 只识别模型服务结构化错误中的明确语义，普通路由 404 不等于模型不存在。 + return message != null && message.length() <= 512 && MODEL_NOT_FOUND_MESSAGE.matcher(message.trim()).matches(); + } + /** * 构建模型提示词，并解析图片输入。 * @@ -292,7 +360,8 @@ public class EasyAgentsLlm implements Llm { } String resolvedImage = resolveImage(input); if (StringUtil.noText(resolvedImage)) { - throw new IllegalArgumentException("Resolved image input must not be blank"); + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, + "Resolved image input must not be blank"); } resolvedImages.add(resolvedImage); } @@ -315,7 +384,7 @@ public class EasyAgentsLlm implements Llm { if (imageInput instanceof File file) { return ImageUtil.imageFileToDataUri(file); } - throw new IllegalArgumentException( + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "Unsupported image input type: " + imageInput.getClass().getName()); } @@ -325,7 +394,8 @@ public class EasyAgentsLlm implements Llm { private void assertImageSupported() { if (chatModel instanceof BaseChatModel baseChatModel && Boolean.FALSE.equals(baseChatModel.getConfig().getSupportImage())) { - throw new IllegalArgumentException("当前模型不支持图片输入，请选择支持视觉能力的模型"); + throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, + "当前模型不支持图片输入，请选择支持视觉能力的模型"); } } } 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 index ea51d0e..7164277 100644 --- 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 @@ -12,6 +12,8 @@ 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.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; import com.easyagents.flow.core.chain.event.LlmStreamEvent; import com.easyagents.flow.core.llm.Llm; import com.easyagents.flow.core.node.LlmNode; @@ -101,14 +103,41 @@ public class EasyAgentsLlmTest { try { llm.chat(messageInfo, new Llm.ChatOptions(), null, null); - Assert.fail("expected IllegalArgumentException"); - } catch (IllegalArgumentException exception) { + Assert.fail("expected WorkflowExecutionException"); + } catch (WorkflowExecutionException exception) { + Assert.assertEquals(WorkflowErrorReason.INPUT_INVALID, exception.getReason()); Assert.assertEquals( "当前模型不支持图片输入，请选择支持视觉能力的模型", exception.getMessage()); } } + @Test + public void shouldClassifyExplicitImageInputValidation() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + Llm.MessageInfo message = new Llm.MessageInfo(); + message.setImageInputs(List.of(123)); + WorkflowExecutionException invalidType = Assert.assertThrows(WorkflowExecutionException.class, + () -> llm.resolveImages(message)); + Assert.assertEquals(WorkflowErrorReason.INPUT_INVALID, invalidType.getReason()); + + llm.setImageInputResolver(input -> " "); + WorkflowExecutionException blank = Assert.assertThrows(WorkflowExecutionException.class, + () -> llm.resolveImages(message)); + Assert.assertEquals(WorkflowErrorReason.INPUT_INVALID, blank.getReason()); + } + + @Test + public void shouldNotReclassifyUnknownImageResolverFailures() { + EasyAgentsLlm llm = new EasyAgentsLlm(); + IllegalStateException original = new IllegalStateException("synthetic resolver failure"); + llm.setImageInputResolver(input -> { throw original; }); + Llm.MessageInfo message = new Llm.MessageInfo(); + message.setImageInputs(List.of("image")); + Assert.assertSame(original, Assert.assertThrows(IllegalStateException.class, + () -> llm.resolveImages(message))); + } + /** * 验证重复执行同一 LLM 节点时，每次调用拥有独立流标识且增量不会被覆盖。 */ diff --git a/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelFailureTest.java b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelFailureTest.java new file mode 100644 index 0000000..d55430a --- /dev/null +++ b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelFailureTest.java @@ -0,0 +1,45 @@ +package com.easyagents.flow.support.provider; + +import com.easyagents.core.model.exception.ModelException; +import com.easyagents.flow.core.chain.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; +import org.junit.Assert; +import org.junit.Test; +import java.net.ConnectException; +import java.net.SocketTimeoutException; +import java.net.SocketException; + +public class WorkflowModelFailureTest { + @Test + public void shouldClassifyTypedModelFailuresWithoutParsingMessages() { + assertReason(WorkflowErrorReason.MODEL_RATE_LIMITED, new ModelException(429, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_AUTH_FAILED, new ModelException(401, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_AUTH_FAILED, new ModelException(403, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_UNAVAILABLE, new ModelException(503, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_TIMEOUT, new ModelException(408, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_TIMEOUT, new ModelException(504, "opaque", null)); + assertReason(WorkflowErrorReason.MODEL_TIMEOUT, new RuntimeException(new SocketTimeoutException())); + assertReason(WorkflowErrorReason.MODEL_UNAVAILABLE, new ConnectException()); + assertReason(WorkflowErrorReason.MODEL_UNAVAILABLE, new SocketException("Connection reset")); + assertReason(WorkflowErrorReason.NODE_EXECUTION_FAILED, new RuntimeException("429 timeout 鉴权失败")); + assertReason(WorkflowErrorReason.NODE_OUTPUT_INVALID, new WorkflowExecutionException(WorkflowErrorReason.NODE_OUTPUT_INVALID, "empty")); + } + @Test + public void shouldRecognizeMissingModelWithoutMisclassifyingRoute404() { + assertReason(WorkflowErrorReason.MODEL_NOT_FOUND, + new ModelException(404, "raw response", null, "404", null, "Model not found")); + assertReason(WorkflowErrorReason.MODEL_NOT_FOUND, + new ModelException(404, "raw response", null, null, null, "The model `test` does not exist.")); + assertReason(WorkflowErrorReason.MODEL_NOT_FOUND, + new ModelException(400, "raw response", null, "model_not_found", null, "opaque")); + assertReason(WorkflowErrorReason.NODE_EXECUTION_FAILED, + new ModelException(404, "raw response", null, "404", null, "Not Found")); + assertReason(WorkflowErrorReason.NODE_EXECUTION_FAILED, + new ModelException(404, "raw response", null, "404", null, "Model endpoint not found")); + assertReason(WorkflowErrorReason.MODEL_AUTH_FAILED, + new ModelException(403, "raw response", null, "model_not_found", null, "Model not found")); + } + private void assertReason(WorkflowErrorReason reason, Throwable error) { + Assert.assertEquals(reason, EasyAgentsLlm.modelFailure(error).getReason()); + } +} diff --git a/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelHttpFailureTest.java b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelHttpFailureTest.java new file mode 100644 index 0000000..2a4f4da --- /dev/null +++ b/easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelHttpFailureTest.java @@ -0,0 +1,93 @@ +package com.easyagents.flow.support.provider; + +import com.easyagents.core.model.chat.ChatConfig; +import com.easyagents.core.model.chat.OpenAICompatibleChatModel; +import com.easyagents.flow.core.chain.*; +import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository; +import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository; +import com.easyagents.flow.core.llm.LlmManager; +import com.easyagents.flow.core.llm.Llm; +import com.easyagents.flow.core.llm.LlmProvider; +import com.easyagents.flow.core.node.LlmNode; +import com.sun.net.httpserver.HttpServer; +import org.junit.Assert; +import org.junit.Test; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.UUID; + +/** 真实 HTTP/SSE 到 LlmNode 的分类回归，不直接注入原因码。 */ +public class WorkflowModelHttpFailureTest { + @Test + public void shouldClassifyHttpErrorsAtModelBoundary() throws Exception { + assertFailure(429, "{}", false, WorkflowErrorReason.MODEL_RATE_LIMITED); + assertFailure(401, "{}", false, WorkflowErrorReason.MODEL_AUTH_FAILED); + assertFailure(503, "{}", false, WorkflowErrorReason.MODEL_UNAVAILABLE); + assertFailure(408, "{}", false, WorkflowErrorReason.MODEL_TIMEOUT); + assertFailure(504, "{}", false, WorkflowErrorReason.MODEL_TIMEOUT); + assertFailure(404, "{\"error\":{\"message\":\"Model not found\",\"code\":404,\"type\":\"NotFound\"}}", + false, WorkflowErrorReason.MODEL_NOT_FOUND); + assertFailure(404, "Not Found", false, WorkflowErrorReason.NODE_EXECUTION_FAILED); + } + + @Test + public void streamErrorAfterPartialOutputMustFailInsteadOfSucceeding() throws Exception { + for (String[] error : new String[][]{ + {"rate_limit_exceeded", "MODEL_RATE_LIMITED"}, + {"authentication_error", "MODEL_AUTH_FAILED"}, + {"overloaded_error", "MODEL_UNAVAILABLE"}, + {"request_timeout", "MODEL_TIMEOUT"}, + {"model_not_found", "MODEL_NOT_FOUND"}}) { + String body = "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n" + + "data: {\"error\":{\"code\":\"upstream_error\",\"type\":\"" + error[0] + "\",\"message\":\"private upstream body\"}}\n\n" + + "data: [DONE]\n\n"; + assertFailure(200, body, false, WorkflowErrorReason.valueOf(error[1])); + } + } + + @Test + public void blankOrInvalidJsonOutputMustHaveOutputReason() throws Exception { + assertFailure(200, "data: [DONE]\n\n", false, WorkflowErrorReason.NODE_OUTPUT_INVALID); + for (String output : new String[]{"not-json", "```json\\n\\n```", "null"}) { + assertFailure(200, "data: {\"choices\":[{\"delta\":{\"content\":\"" + output + + "\"}}]}\n\ndata: [DONE]\n\n", true, WorkflowErrorReason.NODE_OUTPUT_INVALID); + } + } + + private void assertFailure(int status, String body, boolean json, WorkflowErrorReason reason) throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/chat", exchange -> { + exchange.getRequestBody().readAllBytes(); + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", status == 200 ? "text/event-stream" : "application/json"); + exchange.sendResponseHeaders(status, bytes.length); + try (var output = exchange.getResponseBody()) { output.write(bytes); } + }); + server.start(); + String id = UUID.randomUUID().toString(); + ChatConfig config = new ChatConfig(); + config.setEndpoint("http://127.0.0.1:" + server.getAddress().getPort()); + config.setRequestPath("/chat"); config.setModel("missing-test"); config.setApiKey("synthetic"); + config.setLogEnabled(false); config.setObservabilityEnabled(false); config.setRetryEnabled(false); + EasyAgentsLlm llm = new EasyAgentsLlm(); + llm.setChatModel(new OpenAICompatibleChatModel<>(config)); + LlmProvider provider = modelId -> id.equals(modelId) ? llm : null; + LlmManager.getInstance().registerProvider(provider); + try { + Chain chain = new Chain(new ChainDefinition(), id); + chain.setEventManager(new EventManager()); + chain.setChainStateRepository(new InMemoryChainStateRepository()); + chain.setNodeStateRepository(new InMemoryNodeStateRepository()); + chain.initializeState(); + LlmNode node = new LlmNode(); node.setId("llm"); node.setLlmId(id); node.setUserPrompt("test"); + node.setChatOptions(new Llm.ChatOptions()); + node.setOutType(json ? "json" : "text"); + WorkflowExecutionException failure = Assert.assertThrows(WorkflowExecutionException.class, () -> node.execute(chain)); + Assert.assertEquals(reason, failure.getReason()); + } finally { + LlmManager.getInstance().removeProvider(provider); + server.stop(0); + } + } +}