lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web): 完成子流程节点和代码节点的运行

Browse Source
This commit is contained in:
v-zhangjc9
2025-07-17 18:38:19 +08:00
parent 3a61da054e
commit ad9dade6a1
15 changed files with 299 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
service-ai/service-ai-web/pom.xml
View File

@@ -86,6 +86,11 @@
<groupId>org.noear</groupId>
<artifactId>solon-ai-dialect-openai</artifactId>
</dependency>
<dependency>
<groupId>com.yomahub</groupId>
<artifactId>liteflow-script-graaljs</artifactId>
<version>${liteflow.version}</version>
</dependency>
<dependency>
<groupId>org.hibernate.orm</groupId>

2
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/WebApplication.java
View File

@@ -27,6 +27,8 @@ import org.springframework.scheduling.annotation.EnableScheduling;
public class WebApplication implements ApplicationRunner {
public static void main(String[] args) {
System.setProperty("polyglot.engine.WarnInterpreterOnly", "false");
SpringApplication.run(WebApplication.class, args);
}

14
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/FlowGraphRunner.java
View File

@@ -1,5 +1,6 @@
package com.lanyuanxiaoyao.service.ai.web.engine;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowContext;
@@ -44,7 +45,9 @@ public final class FlowGraphRunner {
flowStore.init(flowGraph);
for (FlowNode node : flowGraph.nodes()) {
executionQueue.offer(node);
if (ObjectUtil.isNull(node.parentId())) {
executionQueue.offer(node);
}
}
while (!executionQueue.isEmpty()) {
var node = executionQueue.poll();
@@ -77,8 +80,17 @@ public final class FlowGraphRunner {
var runner = runnerClazz.getDeclaredConstructor().newInstance();
runner.setNodeId(node.id());
runner.setContext(context);
// 处理子流程节点的逻辑
if (runner instanceof FlowNodeSubflowRunner subflowRunner) {
var subflowNodes = flowGraph.nodes().select(n -> StrUtil.equals(n.parentId(), node.id()));
var subGraph = new FlowGraph(IdUtil.fastUUID(), subflowNodes, flowGraph.edges());
subflowRunner.setSubGraph(subGraph);
}
runner.run();
// 处理选择节点的逻辑
if (runner instanceof FlowNodeOptionalRunner) {
var targetPoint = ((FlowNodeOptionalRunner) runner).getTargetPoint();
for (FlowEdge edge : nodeOutputMap.get(node.id())) {

17
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/FlowNodeSubflowRunner.java Normal file
View File

@@ -0,0 +1,17 @@
package com.lanyuanxiaoyao.service.ai.web.engine;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowGraph;
import lombok.Getter;
import lombok.Setter;
/**
* 包含子流程的流程
*
* @author lanyuanxiaoyao
* @version 20250717
*/
public abstract class FlowNodeSubflowRunner extends FlowNodeRunner {
@Getter
@Setter
private FlowGraph subGraph;
}

3
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/entity/FlowEdge.java
View File

@@ -1,5 +1,6 @@
package com.lanyuanxiaoyao.service.ai.web.engine.entity;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.time.LocalDateTime;
/**
@@ -12,7 +13,9 @@ public record FlowEdge(
String id,
String source,
String target,
@JsonProperty("sourceHandle")
String sourcePoint,
@JsonProperty("targetHandle")
String targetPoint
) {
public enum Status {

3
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/entity/FlowNode.java
View File

@@ -10,7 +10,8 @@ import java.time.LocalDateTime;
*/
public record FlowNode(
String id,
String type
String type,
String parentId
) {
public enum Status {
INITIAL, RUNNING, FINISHED, SKIPPED

33
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/node/CodeNode.java Normal file
View File

@@ -0,0 +1,33 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowHelper;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowNodeRunner;
import com.lanyuanxiaoyao.service.ai.web.engine.node.code.CodeExecutor;
import com.lanyuanxiaoyao.service.ai.web.engine.node.code.JavaScriptCodeExecutor;
import lombok.extern.slf4j.Slf4j;
/**
* @author lanyuanxiaoyao
* @version 20250717
*/
@Slf4j
public class CodeNode extends FlowNodeRunner {
@Override
public void run() {
var inputVariablesMap = FlowHelper.generateInputVariablesMap(getNodeId(), getContext());
var type = this.<String>getData("type");
var script = this.<String>getData("content");
CodeExecutor executor = null;
switch (type) {
case "javascript":
executor = new JavaScriptCodeExecutor();
}
if (ObjectUtil.isNull(executor)) {
throw new RuntimeException(StrUtil.format("Unsupported type: {}", type));
}
var result = executor.execute(script, inputVariablesMap);
result.forEachKeyValue(this::setData);
}
}

18
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/node/LoopNode.java Normal file
View File

@@ -0,0 +1,18 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowNodeSubflowRunner;
import lombok.extern.slf4j.Slf4j;
/**
* 循环节点
*
* @author lanyuanxiaoyao
* @version 20250717
*/
@Slf4j
public class LoopNode extends FlowNodeSubflowRunner {
@Override
public void run() {
log.info("{}", getSubGraph());
}
}

65
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/node/SwitchNode.java Normal file
View File

@@ -0,0 +1,65 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowHelper;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowNodeOptionalRunner;
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
/**
* @author lanyuanxiaoyao
* @version 20250717
*/
@SuppressWarnings("unchecked")
@Slf4j
public class SwitchNode extends FlowNodeOptionalRunner {
@Override
public String runOptional() {
var conditions = this.<List<Map<String, Object>>>getData("conditions");
for (Map<String, Object> item : conditions) {
var condition = (Map<String, Object>) item.getOrDefault("condition", Map.of());
var id = (String) condition.getOrDefault("id", "");
var conjunction = (String) condition.getOrDefault("conjunction", "and");
var conditionChildren = ((List<Map<String, Object>>) condition.getOrDefault("children", List.<Map<String, Object>>of()));
if (
StrUtil.equals(conjunction, "and")
&& conditionChildren.stream().allMatch(this::check)
) {
return id;
} else if (
StrUtil.equals(conjunction, "or")
&& conditionChildren.stream().anyMatch(this::check)
) {
return id;
}
}
return "";
}
private Boolean check(Map<String, Object> condition) {
var leftVariable = (String) BeanUtil.getProperty(condition, "left.field");
var left = FlowHelper.generateVariable(leftVariable, getContext());
var operator = (String) condition.get("op");
var right = condition.get("right");
if (right instanceof Map<?, ?>) {
var rightVariable = (String) BeanUtil.getProperty(condition, "right.field");
if (StrUtil.isNotBlank(rightVariable)) {
right = FlowHelper.generateVariable(rightVariable, getContext());
}
}
return switch (operator) {
case "equal" -> ObjectUtil.equals(left, right);
case "not_equal" -> ObjectUtil.notEqual(left, right);
case "is_empty" -> ObjectUtil.isEmpty(left);
case "is_not_empty" -> ObjectUtil.isNotEmpty(left);
case "like" -> StrUtil.contains((String) left, (String) right);
case "not_like" -> !StrUtil.contains((String) left, (String) right);
case "starts_with" -> StrUtil.startWith((String) left, (String) right);
case "ends_with" -> StrUtil.endWith((String) left, (String) right);
default -> false;
};
}
}

11
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/node/code/CodeExecutor.java Normal file
View File

@@ -0,0 +1,11 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node.code;
import org.eclipse.collections.api.map.ImmutableMap;
/**
* @author lanyuanxiaoyao
* @version 20250717
*/
public interface CodeExecutor {
ImmutableMap<String, Object> execute(String script, ImmutableMap<String, Object> inputVariablesMap);
}

45
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/node/code/JavaScriptCodeExecutor.java Normal file
View File

@@ -0,0 +1,45 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node.code;
import cn.hutool.core.util.StrUtil;
import org.eclipse.collections.api.factory.Maps;
import org.eclipse.collections.api.map.ImmutableMap;
import org.graalvm.polyglot.Context;
import org.graalvm.polyglot.Engine;
import org.graalvm.polyglot.Source;
import org.graalvm.polyglot.TypeLiteral;
/**
* @author lanyuanxiaoyao
* @version 20250717
*/
public class JavaScriptCodeExecutor implements CodeExecutor{
@Override
public ImmutableMap<String, Object> execute(String script, ImmutableMap<String, Object> inputVariablesMap) {
if (StrUtil.isBlank(script)) {
return Maps.immutable.empty();
}
try (var engin = Engine.create()) {
try (
var context = Context.newBuilder()
.allowAllAccess(true)
.engine(engin)
.build()
) {
var bindings = context.getBindings("js");
inputVariablesMap.forEachKeyValue(bindings::putMember);
var result = context.eval(
Source.create(
"js",
"""
function process() {
%s
}
process();
""".formatted(script)
)
);
return Maps.immutable.ofAll(result.as(new TypeLiteral<>() {}));
}
}
}
}

16
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/service/task/FlowTaskService.java
View File

@@ -10,12 +10,16 @@ import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowContext;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowEdge;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowGraph;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowNode;
import com.lanyuanxiaoyao.service.ai.web.engine.node.CodeNode;
import com.lanyuanxiaoyao.service.ai.web.engine.node.LlmNode;
import com.lanyuanxiaoyao.service.ai.web.engine.node.LoopNode;
import com.lanyuanxiaoyao.service.ai.web.engine.node.OutputNode;
import com.lanyuanxiaoyao.service.ai.web.engine.node.SwitchNode;
import com.lanyuanxiaoyao.service.ai.web.engine.store.InMemoryFlowStore;
import com.lanyuanxiaoyao.service.ai.web.entity.FlowTask;
import com.lanyuanxiaoyao.service.ai.web.repository.FlowTaskRepository;
import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.collections.api.factory.Maps;
@@ -38,14 +42,18 @@ public class FlowTaskService extends SimpleServiceSupport<FlowTask> {
var flowTask = detailOrThrow(id);
var graphVo = mapper.readValue(flowTask.getTemplateFlowGraph(), FlowGraphVo.class);
var flowGraph = new FlowGraph(IdUtil.fastUUID(), graphVo.getNodes(), graphVo.getEdges());
log.info("Graph: {}", flowGraph);
var store = new InMemoryFlowStore();
var executor = new FlowExecutor(
store,
Maps.immutable.of(
"output-node", OutputNode.class,
"llm-node", LlmNode.class
)
Maps.immutable.ofAll(Map.of(
"loop-node", LoopNode.class,
"switch-node", SwitchNode.class,
"code-node", CodeNode.class,
"llm-node", LlmNode.class,
"output-node", OutputNode.class
))
);
FlowContext context = new FlowContext();
context.setInput(mapper.readValue(flowTask.getInput(), new TypeReference<>() {}));

18
service-ai/service-ai-web/src/test/java/com/lanyuanxiaoyao/service/ai/web/TestFlow.java
View File

@@ -35,15 +35,15 @@ public class TestFlow {
var graph = new FlowGraph(
"graph-1",
Sets.immutable.of(
new FlowNode("node-1", "plain-node"),
new FlowNode("node-2", "plain-node"),
new FlowNode("node-4", "plain-node"),
new FlowNode("node-6", "plain-node"),
new FlowNode("node-7", "plain-node"),
new FlowNode("node-5", "plain-node"),
new FlowNode("node-8", "option-node"),
new FlowNode("node-9", "plain-node"),
new FlowNode("node-3", "plain-node")
new FlowNode("node-1", "plain-node", null),
new FlowNode("node-2", "plain-node", null),
new FlowNode("node-4", "plain-node", null),
new FlowNode("node-6", "plain-node", null),
new FlowNode("node-7", "plain-node", null),
new FlowNode("node-5", "plain-node", null),
new FlowNode("node-8", "option-node", null),
new FlowNode("node-9", "plain-node", null),
new FlowNode("node-3", "plain-node", null)
),
Sets.immutable.of(
new FlowEdge("edge-1", "node-1", "node-2", null, null),

57
service-ai/service-ai-web/src/test/java/com/lanyuanxiaoyao/service/ai/web/TestJsExecutor.java Normal file
View File

@@ -0,0 +1,57 @@
package com.lanyuanxiaoyao.service.ai.web;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.collections.api.factory.Maps;
import org.eclipse.collections.api.map.ImmutableMap;
import org.graalvm.polyglot.Context;
import org.graalvm.polyglot.Engine;
import org.graalvm.polyglot.Source;
import org.graalvm.polyglot.TypeLiteral;
/**
* @author lanyuanxiaoyao
* @version 20250717
*/
@Slf4j
public class TestJsExecutor {
public static void main(String[] args) {
var result = executeJavascript(
// language=JavaScript
"return 'hello'",
Maps.immutable.of(
"code", 1
)
);
log.info("Result: {}", result);
}
private static ImmutableMap<String, Object> executeJavascript(String script, ImmutableMap<String, Object> inputVariablesMap) {
if (StrUtil.isBlank(script)) {
return Maps.immutable.empty();
}
try (var engin = Engine.create()) {
try (
var context = Context.newBuilder()
.allowAllAccess(true)
.engine(engin)
.build()
) {
var bindings = context.getBindings("js");
inputVariablesMap.forEachKeyValue(bindings::putMember);
var result = context.eval(
Source.create(
"js",
"""
function process() {
%s
}
process();
""".formatted(script)
)
);
return Maps.immutable.ofAll(result.as(new TypeLiteral<>() {}));
}
}
}
}

11
service-web/client/src/components/flow/Helper.tsx
View File

@@ -1,11 +1,11 @@
import {type Edge, getIncomers, type Node} from '@xyflow/react'
import type {Option} from 'amis/lib/Schema'
import {find, has, isEqual, max, min, unique} from 'licia'
import {contain, find, has, isEqual, max, min, unique} from 'licia'
import {type DependencyList, type MouseEvent as ReactMouseEvent, useCallback, useRef} from 'react'
import Queue from 'yocto-queue'
import {originTypeMap} from '../../pages/ai/task/InputSchema.tsx'
import {useFlowStore} from './store/FlowStore.ts'
import {type OutputVariable} from './types.ts'
import {type OutputVariable, type OutputVariableType} from './types.ts'
export const getAllIncomerNodeById: (id: string, nodes: Node[], edges: Edge[]) => string[] = (id, nodes, edges) => {
let queue = new Queue<Node>()
@@ -82,9 +82,12 @@ export const getAllIncomerNodeOutputVariables: (id: string, inputSchema: Record<
]
}
export const generateAllIncomerOutputVariablesFormOptions: (id: string, inputSchema: Record<string, Record<string, any>>, nodes: Node[], edges: Edge[], data: any) => Option[] = (id, inputSchema, nodes, edges, data) => {
export const generateAllIncomerOutputVariablesFormOptions: (id: string, inputSchema: Record<string, Record<string, any>>, nodes: Node[], edges: Edge[], data: any, targetTypes?: OutputVariableType[]) => Option[] = (id, inputSchema, nodes, edges, data, targetTypes) => {
let optionMap: Record<string, Option[]> = {}
for (const item of getAllIncomerNodeOutputVariables(id, inputSchema, nodes, edges, data)) {
if (targetTypes && !contain(targetTypes, item.type)) {
continue
}
if (!optionMap[item.group]) {
optionMap[item.group] = []
}
@@ -142,7 +145,7 @@ export const generateAllIncomerOutputVariablesConditions: (id: string, inputSche
options: [
{label: '真', value: true},
{label: '假', value: false},
]
],
},
defaultOp: booleanDefaultOperator,
operators: booleanOperators,