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

refactor(ai-web): 使用输入节点代替inputSchema

Browse Source 
将输入表单放在流程图中一起定义，方便统一参数的处理，不需要单独为输入表单的变量进行合并和操作
This commit is contained in:
兰缘小妖
2025-07-20 19:27:09 +08:00
parent a5282762ed
commit 267eecbf45
26 changed files with 114 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,7 +89,10 @@
<dependency>
<groupId>com.yomahub</groupId>
<artifactId>liteflow-script-graaljs</artifactId>
<version>${liteflow.version}</version>
</dependency>
<dependency>
<groupId>com.yomahub</groupId>
<artifactId>liteflow-script-python</artifactId>
</dependency>
<dependency>

21
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/controller/task/TaskTemplateController.java
View File

@@ -1,5 +1,6 @@
package com.lanyuanxiaoyao.service.ai.web.controller.task;
import cn.hutool.core.util.ObjectUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -41,6 +42,12 @@ public class TaskTemplateController extends SimpleControllerSupport<FlowTaskTemp
return AmisResponse.responseSuccess(mapper.readValue(template.getInputSchema(), Map.class));
}
@GetMapping("flow_graph/{id}")
public AmisResponse<?> getFlowGraph(@PathVariable("id") Long id) throws JsonProcessingException {
var template = flowTaskTemplateService.detailOrThrow(id);
return AmisResponse.responseSuccess(mapper.readValue(template.getFlowGraph(), Map.class));
}
@PostMapping("update_flow_graph")
public AmisResponse<?> updateFlowGraph(@RequestBody UpdateGraphItem item) throws JsonProcessingException {
flowTaskTemplateService.updateFlowGraph(item.getId(), mapper.writeValueAsString(item.getGraph()));
@@ -49,8 +56,7 @@ public class TaskTemplateController extends SimpleControllerSupport<FlowTaskTemp
@Override
protected SaveItemMapper<FlowTaskTemplate, SaveItem> saveItemMapper() {
var map = Mappers.getMapper(SaveItem.Mapper.class);
return item -> map.from(item, mapper);
return Mappers.getMapper(SaveItem.Mapper.class);
}
@Override
@@ -69,15 +75,9 @@ public class TaskTemplateController extends SimpleControllerSupport<FlowTaskTemp
private Long id;
private String name;
private String description;
private Map<String, Object> inputSchema;
@org.mapstruct.Mapper
public static abstract class Mapper {
public abstract FlowTaskTemplate from(SaveItem saveItem, @Context ObjectMapper mapper) throws Exception;
protected String mapInputSchema(Map<String, Object> inputSchema, @Context ObjectMapper mapper) throws JsonProcessingException {
return mapper.writeValueAsString(inputSchema);
}
public interface Mapper extends SaveItemMapper<FlowTaskTemplate, SaveItem> {
}
}
@@ -105,6 +105,9 @@ public class TaskTemplateController extends SimpleControllerSupport<FlowTaskTemp
public abstract DetailItem from(FlowTaskTemplate template, @Context ObjectMapper mapper) throws Exception;
public Map<String, Object> mapJson(String source, @Context ObjectMapper mapper) throws Exception {
if (ObjectUtil.isNull(source)) {
return null;
}
return mapper.readValue(source, new TypeReference<>() {
});
}

5
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/FlowHelper.java
View File

@@ -47,11 +47,6 @@ public class FlowHelper {
throw new RuntimeException(StrUtil.format("Target node variable not found: {}.{}", targetNodeId, targetVariableName));
}
return targetNodeData.get(targetVariableName);
} else if (context.getInput().containsKey(expression)) {
if (!context.getInput().containsKey(expression)) {
throw new RuntimeException(StrUtil.format("Target variable not found in input {}", expression));
}
return context.getInput().get(expression);
}
return null;
}

2
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/engine/entity/FlowContext.java
View File

@@ -2,12 +2,10 @@ package com.lanyuanxiaoyao.service.ai.web.engine.entity;
import lombok.Data;
import org.eclipse.collections.api.factory.Maps;
import org.eclipse.collections.api.map.ImmutableMap;
import org.eclipse.collections.api.map.MutableMap;
@Data
public class FlowContext {
private ImmutableMap<String, Object> input = Maps.immutable.empty();
private MutableMap<String, MutableMap<String, Object>> data = Maps.mutable.<String, MutableMap<String, Object>>empty().asSynchronized();
public MutableMap<String, Object> get(String key) {

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

@@ -0,0 +1,16 @@
package com.lanyuanxiaoyao.service.ai.web.engine.node;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowNodeRunner;
import lombok.extern.slf4j.Slf4j;
/**
* @author lanyuanxiaoyao
* @version 20250711
*/
@Slf4j
public class InputNode extends FlowNodeRunner {
@Override
public void run() {
// 入参相关的内容在启动流程的时候已经注入了
}
}

14
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/entity/vo/FlowGraphVo.java Normal file
View File

@@ -0,0 +1,14 @@
package com.lanyuanxiaoyao.service.ai.web.entity.vo;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowEdge;
import com.lanyuanxiaoyao.service.ai.web.engine.entity.FlowNode;
import lombok.Data;
import org.eclipse.collections.api.map.MutableMap;
import org.eclipse.collections.api.set.ImmutableSet;
@Data
public class FlowGraphVo {
private ImmutableSet<FlowNode> nodes;
private ImmutableSet<FlowEdge> edges;
private MutableMap<String, MutableMap<String, Object>> data;
}

4
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/service/feedback/FeedbackService.java
View File

@@ -7,9 +7,7 @@ import com.lanyuanxiaoyao.service.ai.web.entity.context.FeedbackContext;
import com.lanyuanxiaoyao.service.ai.web.repository.FeedbackRepository;
import com.yomahub.liteflow.core.FlowExecutor;
import java.util.List;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -24,7 +22,7 @@ import org.springframework.transaction.annotation.Transactional;
this.executor = executor;
}
@Scheduled(initialDelay = 1, fixedDelay = 1, timeUnit = TimeUnit.MINUTES)
// @Scheduled(initialDelay = 1, fixedDelay = 1, timeUnit = TimeUnit.MINUTES)
public void analysis() {
List<Feedback> feedbacks = repository.findAll(
builder -> builder

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

@@ -1,30 +1,29 @@
package com.lanyuanxiaoyao.service.ai.web.service.task;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lanyuanxiaoyao.service.ai.web.base.service.SimpleServiceSupport;
import com.lanyuanxiaoyao.service.ai.web.engine.FlowExecutor;
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.InputNode;
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.entity.vo.FlowGraphVo;
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;
import org.eclipse.collections.api.map.MutableMap;
import org.eclipse.collections.api.set.ImmutableSet;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.stereotype.Service;
@@ -42,7 +41,6 @@ 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(
@@ -52,19 +50,25 @@ public class FlowTaskService extends SimpleServiceSupport<FlowTask> {
"switch-node", SwitchNode.class,
"code-node", CodeNode.class,
"llm-node", LlmNode.class,
"input-node", InputNode.class,
"output-node", OutputNode.class
))
);
FlowContext context = new FlowContext();
context.setInput(mapper.readValue(flowTask.getInput(), new TypeReference<>() {}));
context.setData(graphVo.getData());
// 如果发现输入节点，将相关数据注入到对应节点的上下文中
var inputNode = flowGraph.nodes()
.detectOptional(node -> StrUtil.equals(node.type(), "input-node") && ObjectUtil.isEmpty(node.parentId()))
.orElse(null);
if (ObjectUtil.isNotNull(inputNode)) {
if (!context.getData().containsKey(inputNode.id())) {
context.getData().put(inputNode.id(), Maps.mutable.empty());
}
context.getData().get(inputNode.id()).putAll(mapper.readValue(flowTask.getInput(), new TypeReference<>() {
}));
}
executor.execute(flowGraph, context);
}
@Data
public static final class FlowGraphVo {
private ImmutableSet<FlowNode> nodes;
private ImmutableSet<FlowEdge> edges;
private MutableMap<String, MutableMap<String, Object>> data;
}
}

29
service-ai/service-ai-web/src/main/java/com/lanyuanxiaoyao/service/ai/web/service/task/FlowTaskTemplateService.java
View File

@@ -1,22 +1,47 @@
package com.lanyuanxiaoyao.service.ai.web.service.task;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lanyuanxiaoyao.service.ai.web.base.service.SimpleServiceSupport;
import com.lanyuanxiaoyao.service.ai.web.entity.FlowTaskTemplate;
import com.lanyuanxiaoyao.service.ai.web.entity.vo.FlowGraphVo;
import com.lanyuanxiaoyao.service.ai.web.repository.FlowTaskTemplateRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@Slf4j
@Service
public class FlowTaskTemplateService extends SimpleServiceSupport<FlowTaskTemplate> {
public FlowTaskTemplateService(FlowTaskTemplateRepository flowTaskTemplateRepository) {
private final ObjectMapper mapper;
public FlowTaskTemplateService(FlowTaskTemplateRepository flowTaskTemplateRepository, Jackson2ObjectMapperBuilder builder) {
super(flowTaskTemplateRepository);
this.mapper = builder.build();
}
@Transactional(rollbackFor = Exception.class)
public void updateFlowGraph(Long id, String flowGraph) {
public void updateFlowGraph(Long id, String flowGraph) throws JsonProcessingException {
var template = detailOrThrow(id);
var graph = mapper.readValue(flowGraph, FlowGraphVo.class);
// 如果发现输入节点，就单独提取出来
var inputNode = graph.getNodes()
.detectOptional(node -> StrUtil.equals(node.type(), "input-node") && ObjectUtil.isEmpty(node.parentId()))
.orElse(null);
if (ObjectUtil.isNotNull(inputNode)) {
var nodeId = inputNode.id();
var nodeData = graph.getData().getOrDefault(nodeId, null);
if (ObjectUtil.isNotNull(nodeData) && nodeData.containsKey("inputs")) {
template.setInputSchema(mapper.writeValueAsString(nodeData.get("inputs")));
}
} else {
template.setInputSchema("{}");
}
template.setFlowGraph(flowGraph);
save(template);
}