[app-builder] 并行节点特性添加 (ModelEngine-Group#180)

RonnyChan96 · wuayee · commit 7566de24e197 · 2025-05-19T14:47:40.000+08:00
diff --git a/app-builder/jane/plugins/aipp-plugin/src/main/resources/application-prod.yml b/app-builder/jane/plugins/aipp-plugin/src/main/resources/application-prod.yml
@@ -73,6 +73,7 @@ tool:
     TEXTTOIMAGENODESTATE: textToImageNodeState
     FILEEXTRACTNODESTATE: fileExtractionNodeState
     LOOPNODESTATE: loopNodeState
+    PARALLELNODESTATE: parallelNodeState
 export-meta:
   version: 1.0.1
 sensitive:
diff --git a/app-builder/jane/plugins/aipp-plugin/src/main/resources/component/basic_node_en.json b/app-builder/jane/plugins/aipp-plugin/src/main/resources/component/basic_node_en.json
@@ -69,6 +69,11 @@
     "name": "Loop",
     "uniqueName": ""
   },
+  {
+    "type": "parallelNodeState",
+    "name": "Parallel",
+    "uniqueName": ""
+  },
   {
     "type": "manualCheckNodeState",
     "name": "Manual Form",
diff --git a/app-builder/jane/plugins/aipp-plugin/src/main/resources/component/basic_node_zh.json b/app-builder/jane/plugins/aipp-plugin/src/main/resources/component/basic_node_zh.json
@@ -69,6 +69,11 @@
     "name": "循环",
     "uniqueName": ""
   },
+  {
+    "type": "parallelNodeState",
+    "name": "并行",
+    "uniqueName": ""
+  },
   {
     "type": "manualCheckNodeState",
     "name": "人工表单",
diff --git a/app-builder/plugins/aipp-parallel-tool/pom.xml b/app-builder/plugins/aipp-parallel-tool/pom.xml
@@ -135,7 +135,7 @@
                         <configuration>
                             <target>
                                 <copy file="${project.build.directory}/${project.build.finalName}.jar"
-                                      todir="../../../framework/fit/java/target/plugins"/>
+                                      todir="../../../build/plugins"/>
                             </target>
                         </configuration>
                         <goals>
diff --git a/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/domain/BatchRequest.java b/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/domain/BatchRequest.java
@@ -123,13 +123,10 @@ private boolean postUndoTask() {
                         task.getIndex());
 
                 this.doingToolCallTasks.put(task.getIndex(), task);
-                Map<String, Object> toolArgs = task.getToolCall()
-                        .getArgs()
-                        .stream()
-                        .collect(Collectors.toMap(Argument::getName, Argument::getValue));
+                String jsonArgs = JSONObject.toJSONString(task.getToolCall().getArgs(),
+                        SerializerFeature.WriteMapNullValue);
                 this.complete(task,
-                        JSONArray.parse(this.syncToolCall.call(task.getToolCall().getUniqueName(),
-                                JSONObject.toJSONString(toolArgs, SerializerFeature.WriteMapNullValue))));
+                        JSONArray.parse(this.syncToolCall.call(task.getToolCall().getUniqueName(), jsonArgs)));
             } catch (Throwable ex) {
                 this.setException(task, ex);
             } finally {
diff --git a/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/entities/ToolCall.java b/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/entities/ToolCall.java
@@ -11,7 +11,7 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-import java.util.List;
+import java.util.Map;
 
 /**
  * 并行工具调用信息。
@@ -25,6 +25,6 @@
 @AllArgsConstructor
 public class ToolCall {
     private String uniqueName;
-    private List<Argument> args;
+    private Map<String, Object> args;
     private String outputName;
 }
diff --git a/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/service/impl/ParallelToolServiceImpl.java b/app-builder/plugins/aipp-parallel-tool/src/main/java/modelengine/fit/jade/aipp/tool/parallel/service/impl/ParallelToolServiceImpl.java
@@ -94,11 +94,11 @@ private void validateToolCall(ToolCall toolCall) {
         Validation.isTrue(StringUtils.isNotEmpty(toolCall.getUniqueName()),
                 "The tool unique name should not be empty.");
         Validation.notNull(toolCall.getArgs(), "The tool args should not be null.");
-        toolCall.getArgs().forEach(arg -> {
+        toolCall.getArgs().entrySet().forEach(arg -> {
             Validation.notNull(arg,
                     StringUtils.format("The tool arg should not be null. [toolUniqueName={0}]",
                             toolCall.getUniqueName()));
-            Validation.isTrue(StringUtils.isNotEmpty(arg.getName()),
+            Validation.isTrue(StringUtils.isNotEmpty(arg.getKey()),
                     StringUtils.format("The tool arg name should not be empty. [toolUniqueName={0}]",
                             toolCall.getUniqueName()));
         });
diff --git a/app-builder/plugins/aipp-parallel-tool/src/main/resources/sql/data/aipp_parallel_tool.sql b/app-builder/plugins/aipp-parallel-tool/src/main/resources/sql/data/aipp_parallel_tool.sql
@@ -0,0 +1,12 @@
+INSERT INTO "public"."store_plugin" ("plugin_id", "plugin_name", "extension", "deploy_status", "is_builtin", "source", "icon") VALUES ('d44a239ed854ef94af0f032a526907e20ba8a56ebb4f851cc6956c0172a144e7', '并行', '{"uniqueness.groupId":"modelengine.fit.jade.plugin","pluginFullName":"aipp-parallel-tool-1.0.0-SNAPSHOT_1745841582889.jar","checksum":"42266c0ada1415feb0b689f288b4063b9406506ff7da16eca0332b91a01df20e","name":"并行","description":"并行执行工具","uniqueness.artifactId":"aipp-parallel-tool","type":"java"}', 'DEPLOYED', 't', '', NULL) ON CONFLICT ("plugin_id") DO NOTHING;
+
+INSERT INTO "public"."store_plugin_tool" ("tool_name", "plugin_id", "tool_unique_name") VALUES ('parallelToolDefault', 'd44a239ed854ef94af0f032a526907e20ba8a56ebb4f851cc6956c0172a144e7', '1d0c8164-1aea-4264-879f-2c7898d13fb9') ON CONFLICT ("plugin_id", "tool_unique_name") DO NOTHING;
+
+INSERT INTO "public"."store_definition" ("name", "schema", "definition_group_name") VALUES ('parallelTool', '{"name":"parallelTool","description":"用于并行执行工具","parameters":{"type":"object","properties":{"toolCalls":{"defaultValue":"","description":"并行调用的工具列表","name":"toolCalls","type":"array","items":{"type":"object","properties":{"uniqueName":{"type":"string"},"args":{"type":"object"},"outputName":{"type":"string"}}},"examples":"","required":[]},"config":{"defaultValue":"","description":"并行调用的配置","name":"config","type":"object","properties":{"concurrency":{"type":"integer"}},"examples":"","required":[]},"context":{"defaultValue":"","description":"调用时的上下文信息","name":"context","type":"object","examples":"","required":[]}},"required":["toolCalls"]},"order":["toolCalls","config","context"],"parameterExtensions":null,"return":{"type":"object","convertor":""}}', 'ParallelTool') ON CONFLICT ("definition_group_name", "name") DO NOTHING;
+
+INSERT INTO "public"."store_tag" ("tool_unique_name", "name") VALUES ('1d0c8164-1aea-4264-879f-2c7898d13fb9', 'FIT') ON CONFLICT ("tool_unique_name", "name") DO NOTHING;
+INSERT INTO "public"."store_tag" ("tool_unique_name", "name") VALUES ('1d0c8164-1aea-4264-879f-2c7898d13fb9', 'PARALLELNODESTATE') ON CONFLICT ("tool_unique_name", "name") DO NOTHING;
+INSERT INTO "public"."store_tag" ("tool_unique_name", "name") VALUES ('1d0c8164-1aea-4264-879f-2c7898d13fb9', 'BASIC') ON CONFLICT ("tool_unique_name", "name") DO NOTHING;
+
+
+INSERT INTO "public"."store_tool" ("name", "schema", "runnables", "extensions", "unique_name", "version", "is_latest", "group_name", "definition_name", "definition_group_name") VALUES ('parallelToolDefault', '{"name":"parallelTool","description":"用于并行执行工具","parameters":{"type":"object","properties":{"toolCalls":{"defaultValue":"","description":"并行调用的工具列表","name":"toolCalls","type":"array","items":{"type":"object","properties":{"uniqueName":{"type":"string"},"args":{"type":"object"},"outputName":{"type":"string"}}},"examples":"","required":[]},"config":{"defaultValue":"","description":"并行调用的配置","name":"config","type":"object","properties":{"concurrency":{"type":"integer"}},"examples":"","required":[]},"context":{"defaultValue":"","description":"调用时的上下文信息","name":"context","type":"object","examples":"","required":[]}},"required":["toolCalls"]},"order":["toolCalls","config","context"],"parameterExtensions":null,"return":{"type":"object","convertor":""}}', '{"FIT":{"genericableId":"modelengine.fit.jade.aipp.tool.parallel","fitableId":"default"}}', '{"tags":["FIT","BASIC","PARALLELNODESTATE"]}', '1d0c8164-1aea-4264-879f-2c7898d13fb9', '1.0.0', 't', 'ParallelToolImpl', 'parallelTool', 'ParallelTool')  ON CONFLICT ("unique_name", "version") DO NOTHING;
diff --git a/app-builder/plugins/aipp-parallel-tool/src/test/java/modelengine/fit/jade/aipp/tool/parallel/domain/BatchRequestTest.java b/app-builder/plugins/aipp-parallel-tool/src/test/java/modelengine/fit/jade/aipp/tool/parallel/domain/BatchRequestTest.java
@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -47,9 +48,9 @@ class BatchRequestTest {
 
     @Test
     void shouldCallExecutorByConcurrencyWhenPostGivenToolCall() {
-        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new ArrayList<>()).build(),
-                ToolCall.builder().uniqueName("u2").args(new ArrayList<>()).build(),
-                ToolCall.builder().uniqueName("u3").args(new ArrayList<>()).build());
+        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new HashMap<>()).build(),
+                ToolCall.builder().uniqueName("u2").args(new HashMap<>()).build(),
+                ToolCall.builder().uniqueName("u3").args(new HashMap<>()).build());
         Config config = Config.builder().concurrency(2).build();
         Mockito.doNothing().when(this.taskExecutor).post(Mockito.any());
 
@@ -67,10 +68,10 @@ void shouldCallExecutorByConcurrencyWhenPostGivenToolCall() {
     @Test
     void shouldGetResultWhenAwaitGivenToolCallSuccessfully() {
         List<ToolCall> toolCalls =
-                Arrays.asList(ToolCall.builder().uniqueName("u1").args(new ArrayList<>()).outputName("1").build(),
+                Arrays.asList(ToolCall.builder().uniqueName("u1").args(new HashMap<>()).outputName("1").build(),
                         ToolCall.builder()
                                 .uniqueName("u2")
-                                .args(Collections.singletonList(Argument.builder().name("a").value(1).build()))
+                                .args(MapBuilder.<String, Object>get().put("a", 1).build())
                                 .outputName("2")
                                 .build());
         Config config = Config.builder().concurrency(1).build();
@@ -104,8 +105,8 @@ void shouldGetResultWhenAwaitGivenToolCallSuccessfully() {
 
     @Test
     void shouldThrowExceptionWhenAwaitGivenToolCallException() {
-        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new ArrayList<>()).build(),
-                ToolCall.builder().uniqueName("u2").args(new ArrayList<>()).build());
+        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new HashMap<>()).build(),
+                ToolCall.builder().uniqueName("u2").args(new HashMap<>()).build());
         Config config = Config.builder().concurrency(1).build();
         Mockito.doAnswer((Answer<Void>) invocation -> {
             Runnable runnable = invocation.getArgument(0);
@@ -130,8 +131,8 @@ void shouldThrowExceptionWhenAwaitGivenToolCallException() {
 
     @Test
     void shouldNotExecuteRemainToolWhenAwaitGivenInstanceNotRunning() {
-        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new ArrayList<>()).build(),
-                ToolCall.builder().uniqueName("u2").args(new ArrayList<>()).build());
+        List<ToolCall> toolCalls = Arrays.asList(ToolCall.builder().uniqueName("u1").args(new HashMap<>()).build(),
+                ToolCall.builder().uniqueName("u2").args(new HashMap<>()).build());
         Config config = Config.builder().concurrency(1).build();
         Mockito.doAnswer((Answer<Void>) invocation -> {
             Runnable runnable = invocation.getArgument(0);
diff --git a/app-engine/frontend/src/locale/en_US.json b/app-engine/frontend/src/locale/en_US.json
@@ -951,5 +951,6 @@
   "reorganizeNodes": "Organize nodes",
   "zoomOut": "Zoom Out",
   "zoomIn": "Zoom In",
-  "formItemFieldTypeCannotBeEmpty": "Field type is required"
+  "formItemFieldTypeCannotBeEmpty": "Field type is required",
+  "addParallelTask": "Add Parallel Tasks"
 }
diff --git a/app-engine/frontend/src/locale/zh_CN.json b/app-engine/frontend/src/locale/zh_CN.json
@@ -951,5 +951,6 @@
   "reorganizeNodes": "整理节点",
   "zoomOut": "缩小",
   "zoomIn": "放大",
-  "formItemFieldTypeCannotBeEmpty": "表单项类型不能为空"
+  "formItemFieldTypeCannotBeEmpty": "表单项类型不能为空",
+  "addParallelTask": "添加并行任务"
 }
diff --git a/app-engine/frontend/src/pages/addFlow/components/elsa-stage.tsx b/app-engine/frontend/src/pages/addFlow/components/elsa-stage.tsx
@@ -218,6 +218,12 @@ const Stage = (props) => {
         setShowTools(true);
         setModalTypes('loop');
       });
+      // 并行节点
+      agent.onParallelSelect(({ onSelect }) => {
+        pluginCallback.current = onSelect;
+        setShowTools(true);
+        setModalTypes('parallel');
+      });
     }).catch(() => {
       setSpinning && setSpinning(false);
     });
@@ -263,13 +269,17 @@ const Stage = (props) => {
     let obj = {};
     let uniqueName = '';
     let loopObj = {};
+    let parallelObj = {};
     item.forEach((e) => {
       obj = e.schema;
       uniqueName = e.uniqueName;
       loopObj = e;
+      parallelObj = e;
     });
     if (modalTypes === 'loop') {
       pluginCallback.current(loopObj);
+    } else if (modalTypes === 'parallel') {
+      pluginCallback.current(parallelObj);
     } else if (modalTypes === 'llmTool') {
       pluginCallback.current(uniqueName);
     } else {

Original file line number	Diff line number	Diff line change
`@@ -951,5 +951,6 @@`
`951`	`951`	`"reorganizeNodes": "Organize nodes",`
`952`	`952`	`"zoomOut": "Zoom Out",`
`953`	`953`	`"zoomIn": "Zoom In",`
`954`		`- "formItemFieldTypeCannotBeEmpty": "Field type is required"`
	`954`	`+ "formItemFieldTypeCannotBeEmpty": "Field type is required",`
	`955`	`+ "addParallelTask": "Add Parallel Tasks"`
`955`	`956`	`}`