EasyFlow-Tech-Dev/Easy-Agents
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 统一工作流失败原因与执行状态

Browse Source 
- 关联 EASY-2，补齐模型错误分类、节点归属与执行日志

- 覆盖流式失败、重试状态及模型错误映射回归
This commit is contained in:
陈子默
2026-09-08 10:48:17 +08:00
parent 45c708a212
commit 913a432ee7
18 changed files with 793 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
easy-agents-core/src/main/java/com/easyagents/core/model/client/BaseStreamClientListener.java
View File

@@ -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);

19
easy-agents-core/src/main/java/com/easyagents/core/model/client/impl/Util.java
View File

@@ -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;

32
easy-agents-core/src/main/java/com/easyagents/core/model/exception/ModelException.java
View File

@@ -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

52
easy-agents-core/src/test/java/com/easyagents/core/model/client/StreamFailureLifecycleTest.java Normal file
View File

@@ -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<String> events = new ArrayList<>();
final StreamClient client = new StreamClient() {
public void start(String url, Map<String, String> 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;
});
}
}

39
easy-agents-core/src/test/java/com/easyagents/core/model/client/impl/ModelHttpFailureTest.java Normal file
View File

@@ -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[]{"<html>Not Found</html>", "{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);
}
}

41
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java
View File

@@ -612,6 +612,8 @@ public class Chain {
Map<String, Object> 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<ChainStateField> 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;
}

110
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainState.java
View File

@@ -596,63 +596,71 @@ public class ChainState implements Serializable {
List<Parameter> suspendParameters = null;
List<Map<String, Object>> 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()) {

36
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ExceptionSummary.java
View File

@@ -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<Throwable> 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<Throwable> 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;
}

37
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowErrorReason.java Normal file
View File

@@ -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;
}
}

30
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/WorkflowExecutionException.java Normal file
View File

@@ -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; }
}

7
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/event/NodeEndEvent.java
View File

@@ -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<String, Object> 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; }
/**
* 获取事件创建时捕获的节点终态。
*

10
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
View File

@@ -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());

29
easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LlmNode.java
View File

@@ -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<Map<String, Object>> 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<Map<String, Object>> values) {
try {
return TextTemplate.of(template).formatToString(values);
} catch (RuntimeException failure) {
throw new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "Cannot resolve prompt variables", failure);
}
}
/**
* 移除 ``` 或者 ```json 等

161
easy-agents-flow/src/test/java/com/easyagents/flow/core/test/WorkflowFailureContractTest.java Normal file
View File

@@ -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<String> 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<String, Object> 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(); }
}
}

82
easy-agents-support/src/main/java/com/easyagents/flow/support/provider/EasyAgentsLlm.java
View File

@@ -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<Throwable> 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,
"当前模型不支持图片输入，请选择支持视觉能力的模型");
}
}
}

33
easy-agents-support/src/test/java/com/easyagents/flow/support/provider/EasyAgentsLlmTest.java
View File

@@ -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 节点时，每次调用拥有独立流标识且增量不会被覆盖。
*/

45
easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelFailureTest.java Normal file
View File

@@ -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());
}
}

93
easy-agents-support/src/test/java/com/easyagents/flow/support/provider/WorkflowModelHttpFailureTest.java Normal file
View File

@@ -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, "<html>Not Found</html>", 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);
}
}
}