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:47:54 +08:00
parent 130423edb4
commit 9667a6d262
18 changed files with 793 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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