diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java index 97f4bc9..379629f 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java @@ -38,6 +38,10 @@ public class Parameter implements Serializable, Cloneable { protected String value; protected boolean required; protected String defaultValue; + /** + * 是否在循环节点完成后将各轮数组输出合并一层。 + */ + protected boolean flattenAggregation; protected List children; /** @@ -167,6 +171,24 @@ public class Parameter implements Serializable, Cloneable { this.defaultValue = defaultValue; } + /** + * 判断循环输出是否启用一层扁平聚合。 + * + * @return 启用时返回 {@code true} + */ + public boolean isFlattenAggregation() { + return flattenAggregation; + } + + /** + * 设置循环输出的一层扁平聚合开关。 + * + * @param flattenAggregation 是否启用 + */ + public void setFlattenAggregation(boolean flattenAggregation) { + this.flattenAggregation = flattenAggregation; + } + public boolean isRequired() { return required; } @@ -273,6 +295,7 @@ public class Parameter implements Serializable, Cloneable { ", value='" + value + '\'' + ", required=" + required + ", defaultValue='" + defaultValue + '\'' + + ", flattenAggregation=" + flattenAggregation + ", children=" + children + ", enums=" + enums + ", formType='" + formType + '\'' + diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java index cd1a740..9a2ea51 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java @@ -15,6 +15,7 @@ public final class LoopResultReference implements Serializable { private final String resultId; private final int iterationCount; private final String outputName; + private final boolean flattenAggregation; /** * 创建循环结果引用。 @@ -24,23 +25,64 @@ public final class LoopResultReference implements Serializable { * @param outputName 输出名称 */ public LoopResultReference(String resultId, int iterationCount, String outputName) { + this(resultId, iterationCount, outputName, false); + } + + /** + * 创建带聚合策略的循环结果引用。 + * + * @param resultId 循环结果 ID + * @param iterationCount 迭代次数 + * @param outputName 输出名称 + * @param flattenAggregation 是否将各轮数组合并一层 + */ + public LoopResultReference( + String resultId, + int iterationCount, + String outputName, + boolean flattenAggregation) { this.resultId = Objects.requireNonNull(resultId, "resultId must not be null"); this.iterationCount = iterationCount; this.outputName = Objects.requireNonNull(outputName, "outputName must not be null"); + this.flattenAggregation = flattenAggregation; } + /** + * 获取循环结果 ID。 + * + * @return 循环结果 ID + */ public String getResultId() { return resultId; } + /** + * 获取累计迭代次数。 + * + * @return 累计迭代次数 + */ public int getIterationCount() { return iterationCount; } + /** + * 获取输出名称。 + * + * @return 输出名称 + */ public String getOutputName() { return outputName; } + /** + * 判断是否将各轮数组输出合并一层。 + * + * @return 启用时返回 {@code true} + */ + public boolean isFlattenAggregation() { + return flattenAggregation; + } + /** * 获取跨异步审计边界使用的稳定引用类型。 * diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java index 55cd40c..a9776f2 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java @@ -21,6 +21,7 @@ import java.util.Iterator; import java.util.List; import java.util.LinkedHashMap; import java.util.Map; +import java.util.Set; import java.util.function.Consumer; /** @@ -295,12 +296,38 @@ public interface LoopResultRepository { */ default Map references( String resultId, int iterationCount, List outputNames) { + return references( + resultId, + iterationCount, + outputNames, + java.util.Collections.emptySet()); + } + + /** + * 为每个循环输出创建带聚合策略的轻量引用。 + * + * @param resultId 循环结果 ID + * @param iterationCount 已累计迭代数 + * @param outputNames 输出名称 + * @param flattenedOutputNames 启用一层扁平聚合的输出名称 + * @return 输出名称到轻量引用的映射 + */ + default Map references( + String resultId, + int iterationCount, + List outputNames, + Set flattenedOutputNames) { Map references = new LinkedHashMap<>(); if (outputNames != null) { for (String outputName : outputNames) { references.put( outputName, - new LoopResultReference(resultId, iterationCount, outputName)); + new LoopResultReference( + resultId, + iterationCount, + outputName, + flattenedOutputNames != null + && flattenedOutputNames.contains(outputName))); } } return references; @@ -310,14 +337,15 @@ public interface LoopResultRepository { * 解析单个循环输出引用。 * * @param reference 循环输出引用 - * @return 与旧实现相同的累计列表 + * @return 未启用时返回按轮次累计的列表，启用时返回只扁平一层的数组 */ default Object resolve(LoopResultReference reference) { - return load( + Object output = load( reference.getResultId(), reference.getIterationCount(), List.of(reference.getOutputName())) .get(reference.getOutputName()); + return ReferenceResolver.applyAggregation(reference, output); } /** @@ -420,7 +448,10 @@ public interface LoopResultRepository { LoopResultReference reference = (LoopResultReference) value; Map outputs = loaded.get(new GroupKey( reference.getResultId(), reference.getIterationCount())); - return outputs == null ? null : outputs.get(reference.getOutputName()); + Object output = outputs == null + ? null + : outputs.get(reference.getOutputName()); + return applyAggregation(reference, output); } if (value instanceof LoopInputReference) { return loadedInputs.get( @@ -451,6 +482,91 @@ public interface LoopResultRepository { return value; } + /** + * 根据循环结果引用对累计输出执行一次线性聚合。 + * + * @param reference 循环结果引用 + * @param output 按轮次累计的原始输出 + * @return 原始输出或只扁平一层后的数组 + * @throws IllegalStateException 启用扁平聚合但某轮值不是数组 + */ + private static Object applyAggregation( + LoopResultReference reference, Object output) { + if (!reference.isFlattenAggregation()) { + return output; + } + if (!(output instanceof List)) { + throw invalidFlattenValue(reference, -1, output); + } + + List iterationValues = (List) output; + int flattenedSize = 0; + int iterationIndex = 0; + // 使用顺序迭代兼容链表，避免按索引读取退化为 O(n²)。 + for (Object iterationValue : iterationValues) { + int currentSize; + if (iterationValue instanceof List) { + currentSize = ((List) iterationValue).size(); + } else if (iterationValue != null + && iterationValue.getClass().isArray()) { + currentSize = java.lang.reflect.Array.getLength(iterationValue); + } else { + throw invalidFlattenValue( + reference, iterationIndex, iterationValue); + } + try { + flattenedSize = Math.addExact(flattenedSize, currentSize); + } catch (ArithmeticException exception) { + throw new IllegalStateException( + "Loop output '" + reference.getOutputName() + + "' is too large to flatten", + exception); + } + iterationIndex++; + } + + // 先统计容量再顺序追加，避免大数组扩容复制，整体复杂度保持 O(n)。 + java.util.ArrayList flattened = + new java.util.ArrayList<>(flattenedSize); + for (Object iterationValue : iterationValues) { + if (iterationValue instanceof List) { + flattened.addAll((List) iterationValue); + continue; + } + int length = java.lang.reflect.Array.getLength(iterationValue); + for (int index = 0; index < length; index++) { + flattened.add( + java.lang.reflect.Array.get(iterationValue, index)); + } + } + return flattened; + } + + /** + * 构造扁平聚合类型错误。 + * + * @param reference 循环结果引用 + * @param iterationIndex 错误轮次；负数表示累计输出结构错误 + * @param value 实际值 + * @return 类型错误 + */ + private static IllegalStateException invalidFlattenValue( + LoopResultReference reference, + int iterationIndex, + Object value) { + String actualType = value == null + ? "null" + : value.getClass().getName(); + String iteration = iterationIndex < 0 + ? "" + : ", iteration " + iterationIndex; + return new IllegalStateException( + "Loop output '" + reference.getOutputName() + + "' requires an array for flatten aggregation" + + iteration + + ", but got " + actualType); + } + /** * 循环结果批量读取分组键。 */ diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java index 7d5895f..0238105 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java @@ -401,7 +401,10 @@ public class LoopNode extends BaseNode { prevTrigger.getLoopCursors().remove(this.id); } Map completedResult = chain.getLoopResultRepository().references( - loopContext.resultId, loopContext.currentIndex, getOutputNames()); + loopContext.resultId, + loopContext.currentIndex, + getOutputNames(), + getFlattenedOutputNames()); chain.getLoopResultRepository().releaseActiveCache( loopContext.resultId); if (!loopContext.inputExternalized) { @@ -1041,6 +1044,24 @@ public class LoopNode extends BaseNode { return outputNames; } + /** + * 获取启用一层扁平聚合的输出名称。 + * + * @return 保持定义顺序的输出名称集合 + */ + private Set getFlattenedOutputNames() { + Set outputNames = new LinkedHashSet<>(); + List outputDefs = getOutputDefs(); + if (outputDefs != null) { + for (Parameter outputDef : outputDefs) { + if (outputDef.isFlattenAggregation()) { + outputNames.add(outputDef.getName()); + } + } + } + return outputNames; + } + /** * 估算本轮输出占用字节数，用于宽松的累计结果失控保护。 * diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java index 046bb32..cbe5eab 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java @@ -76,6 +76,8 @@ public abstract class BaseNodeParser implements NodeParser", + "ref", + "knowledge.documents.content", + null); + output.put("flattenAggregation", true); + outputDefs.add(output); + data.put("outputDefs", outputDefs); + JSONObject nodeJson = new JSONObject(); + nodeJson.put("id", "loop"); + nodeJson.put("type", "loopNode"); + nodeJson.put("data", data); + + LoopNode loopNode = new LoopNodeParser().parse( + nodeJson, new JSONObject(), null); + + Assert.assertEquals(1, loopNode.getOutputDefs().size()); + Assert.assertTrue( + loopNode.getOutputDefs().get(0).isFlattenAggregation()); + } + /** * 构造参数 JSON。 * diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopResultReferenceResolverTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopResultReferenceResolverTest.java index b60318a..ebf49fd 100644 --- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopResultReferenceResolverTest.java +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopResultReferenceResolverTest.java @@ -12,6 +12,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Iterator; +import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; /** @@ -42,6 +43,98 @@ public class LoopResultReferenceResolverTest { Assert.assertEquals(1, repository.getLoadCount()); } + /** + * 验证数组输出按轮次和轮内顺序只扁平一层。 + */ + @Test + public void shouldFlattenArrayOutputOnceInStableOrder() { + InMemoryLoopResultRepository repository = + new InMemoryLoopResultRepository(); + repository.append( + "result-flatten", + 0, + Map.of( + "items", + List.of( + List.of("a"), + List.of("b")))); + repository.append( + "result-flatten", + 1, + Map.of( + "items", + List.of( + List.of("c")))); + + Object resolved = repository.resolve( + new LoopResultReference( + "result-flatten", + 2, + "items", + true)); + + Assert.assertEquals( + List.of( + List.of("a"), + List.of("b"), + List.of("c")), + resolved); + } + + /** + * 验证启用扁平聚合后拒绝某轮标量结果。 + */ + @Test + public void shouldRejectScalarIterationWhenFlattening() { + InMemoryLoopResultRepository repository = + new InMemoryLoopResultRepository(); + repository.append( + "result-invalid", + 0, + Map.of("items", List.of("a"))); + repository.append( + "result-invalid", + 1, + Map.of("items", "scalar")); + + try { + repository.resolve( + new LoopResultReference( + "result-invalid", + 2, + "items", + true)); + Assert.fail("scalar iteration should be rejected"); + } catch (IllegalStateException expected) { + Assert.assertTrue( + expected.getMessage().contains("items")); + Assert.assertTrue( + expected.getMessage().contains("iteration 1")); + } + } + + /** + * 验证循环完成时只给指定输出携带扁平聚合策略。 + */ + @Test + public void shouldCreateReferencesWithPerOutputAggregation() { + InMemoryLoopResultRepository repository = + new InMemoryLoopResultRepository(); + + Map references = repository.references( + "result-references", + 2, + List.of("grouped", "flattened"), + Set.of("flattened")); + + Assert.assertFalse( + ((LoopResultReference) references.get("grouped")) + .isFlattenAggregation()); + Assert.assertTrue( + ((LoopResultReference) references.get("flattened")) + .isFlattenAggregation()); + } + /** * 验证无限 Iterable 在达到预算后立即停止物化。 */