fix: 完善数据中枢导入与查询链路

- 支持常见 Excel 表头、工作簿校验及数据可靠落库 - 修复大整数 ID 精度和逻辑表 SQL 解析问题 - 为查询数据节点注入结构化上下文并兼容 SQL 代码块
2026-08-03 11:39:14 +08:00
parent e3228837f4
commit 19dac5146c
26 changed files with 1399 additions and 82 deletions
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowDatacenterContentService.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowDatacenterContentService.java
@@ -37,6 +37,7 @@ public class WorkflowDatacenterContentService {
    public static final String SEARCH_SQL_MISSING_MESSAGE = "查询数据节点未设置 SQL";
    public static final String SAVE_EXPIRED_MESSAGE = "写入数据节点配置已过期，请重新选择已接入表";
    public static final String INVALID_QUERY_CONTEXT_MESSAGE = "查询上下文配置无效，请重新选择查询数据节点";
+    private static final String QUERY_DATA_CONTEXT_PLACEHOLDER = "{{" + QUERY_DATA_CONTEXT + "}}";
    private static final String QUERY_CONTEXT_PROMPT = """
            你是为工作流中的查询数据节点生成只读 SQL 的生成器，你的职责是返回可直接执行的 SQL，并且你只能输出 SQL。

@@ -47,7 +48,9 @@ public class WorkflowDatacenterContentService {
            4. 只能生成只读 SELECT SQL，允许 WITH、JOIN、子查询、聚合、分组、排序。
            5. 不要生成 INSERT、UPDATE、DELETE、DDL、多语句、存储过程调用。
            6. 优先使用逻辑表名和逻辑字段名，不要输出物理表名、JDBC、驱动信息。
-            7. 如果存在重名表，请使用 catalog.table 形式消除歧义。
+            7. 表名和字段名默认不要加引号，禁止使用双引号包裹标识符。
+            8. 只有摘要中存在重名表时才使用 catalog.table；没有重名时只输出 tableName，不要添加 sourceName 或 catalogName 前缀。
+            9. 当字段名无法体现业务含义、字段描述提供了明确含义时，在 SELECT 中为该字段添加简短清晰的英文 snake_case 别名，例如 token AS input_price；别名不要加引号。

            以下是可用的连接摘要：
            """;
@@ -211,6 +214,7 @@ public class WorkflowDatacenterContentService {
        JSONArray nodeIds = data.getJSONArray("queryContextNodeIds");
        if (nodeIds == null || nodeIds.isEmpty()) {
            removeQueryDataContextParameter(data);
+            removeQueryDataContextPlaceholder(data);
            return;
        }
        Map<BigInteger, JSONObject> sourceSummaries = new LinkedHashMap<>();
@@ -229,6 +233,37 @@ public class WorkflowDatacenterContentService {
        }
        String contextValue = QUERY_CONTEXT_PROMPT + "\n" + JSON.toJSONString(new ArrayList<>(sourceSummaries.values()));
        upsertQueryDataContextParameter(data, contextValue);
+        appendQueryDataContextPlaceholder(data);
+    }
+
+    /**
+     * 将查询上下文参数追加到大模型系统提示词，避免覆盖用户配置的提示词。
+     *
+     * @param data 大模型节点数据
+     */
+    private void appendQueryDataContextPlaceholder(JSONObject data) {
+        String systemPrompt = data.getString("systemPrompt");
+        if (StringUtils.hasText(systemPrompt) && systemPrompt.contains(QUERY_DATA_CONTEXT_PLACEHOLDER)) {
+            return;
+        }
+        if (!StringUtils.hasText(systemPrompt)) {
+            data.put("systemPrompt", QUERY_DATA_CONTEXT_PLACEHOLDER);
+            return;
+        }
+        data.put("systemPrompt", systemPrompt.stripTrailing() + "\n\n" + QUERY_DATA_CONTEXT_PLACEHOLDER);
+    }
+
+    /**
+     * 查询上下文关闭后移除自动绑定的提示词占位符。
+     *
+     * @param data 大模型节点数据
+     */
+    private void removeQueryDataContextPlaceholder(JSONObject data) {
+        String systemPrompt = data.getString("systemPrompt");
+        if (!StringUtils.hasText(systemPrompt) || !systemPrompt.contains(QUERY_DATA_CONTEXT_PLACEHOLDER)) {
+            return;
+        }
+        data.put("systemPrompt", systemPrompt.replace(QUERY_DATA_CONTEXT_PLACEHOLDER, "").trim());
    }

    private String resolveFieldType(DatacenterTableField field) {
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/node/SearchDatasetNode.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/node/SearchDatasetNode.java
@@ -26,6 +26,9 @@ public class SearchDatasetNode extends BaseNode {


    private static final Pattern PARAM_PATTERN = Pattern.compile("\\{\\{(.+?)\\}\\}");
+    private static final Pattern SQL_CODE_BLOCK_PATTERN = Pattern.compile(
+            "\\A```(?:sql)?\\s*([\\s\\S]*?)\\s*```\\z",
+            Pattern.CASE_INSENSITIVE);
    private static final int QUERY_PAGE_SIZE = Math.max(
            1,
            Integer.getInteger(
@@ -101,13 +104,28 @@ public class SearchDatasetNode extends BaseNode {
    }

    private String resolveQuerySql(Map<String, Object> params) {
-        String sql = resolveTemplateString(querySql, params);
+        String sql = normalizeSqlCodeBlock(resolveTemplateString(querySql, params));
        if (!StringUtil.hasText(sql)) {
            throw new BusinessException("查询数据节点未设置 SQL");
        }
        return sql.trim();
    }

+    /**
+     * 去除完整单个 SQL Markdown 代码块的边界标记。
+     *
+     * @param sql 原始 SQL 文本
+     * @return 可交给 SQL 解析器处理的文本
+     */
+    private String normalizeSqlCodeBlock(String sql) {
+        if (!StringUtil.hasText(sql)) {
+            return sql;
+        }
+        String trimmed = sql.trim();
+        Matcher matcher = SQL_CODE_BLOCK_PATTERN.matcher(trimmed);
+        return matcher.matches() ? matcher.group(1).trim() : trimmed;
+    }
+
    private DatasetRef copyDatasetRef() {
        DatasetRef copy = new DatasetRef();
        copy.setSourceId(datasetRef == null ? null : datasetRef.getSourceId());
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowDatacenterContentServiceTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowDatacenterContentServiceTest.java
@@ -0,0 +1,217 @@
+package tech.easyflow.ai.easyagentsflow.service;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import tech.easyflow.datacenter.entity.DatacenterTable;
+import tech.easyflow.datacenter.entity.DatacenterTableField;
+import tech.easyflow.datacenter.meta.entity.DatacenterSource;
+import tech.easyflow.datacenter.meta.service.DatacenterDatasetRegistryService;
+
+import java.lang.reflect.Field;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 工作流数据中枢内容准备服务测试。
+ */
+public class WorkflowDatacenterContentServiceTest {
+
+    private static final BigInteger SOURCE_ID = BigInteger.valueOf(1001L);
+    private static final BigInteger TABLE_ID = BigInteger.valueOf(2001L);
+
+    private WorkflowDatacenterContentService service;
+    private DatacenterDatasetRegistryService registryService;
+
+    /**
+     * 初始化服务及数据源元数据桩。
+     *
+     * @throws Exception 注入测试依赖失败时抛出
+     */
+    @Before
+    public void setUp() throws Exception {
+        service = new WorkflowDatacenterContentService();
+        registryService = Mockito.mock(DatacenterDatasetRegistryService.class);
+        injectField(service, "registryService", registryService);
+
+        DatacenterSource source = Mockito.mock(DatacenterSource.class);
+        Mockito.when(source.getSourceName()).thenReturn("ama 实验基线模型预算");
+        Mockito.when(source.getSourceType()).thenReturn("EXCEL");
+
+        DatacenterTableField modelId = mockField("col_id", "模型ID", "VARCHAR");
+        DatacenterTableField inputPrice = mockField("token", "输入价格", "DECIMAL");
+        DatacenterTable table = Mockito.mock(DatacenterTable.class);
+        Mockito.when(table.getId()).thenReturn(TABLE_ID);
+        Mockito.when(table.getTableName()).thenReturn("Sheet1");
+        Mockito.when(table.getFields()).thenReturn(List.of(modelId, inputPrice));
+
+        Mockito.when(registryService.getSourceRequired(SOURCE_ID)).thenReturn(source);
+        Mockito.when(registryService.listManagedTables(SOURCE_ID, null))
+                .thenReturn(new ArrayList<>(List.of(table)));
+        Mockito.when(registryService.getTableWithFields(TABLE_ID)).thenReturn(table);
+    }
+
+    /**
+     * 验证查询上下文会自动且幂等地绑定到系统提示词。
+     */
+    @Test
+    public void testPrepareRootShouldBindQueryContextToSystemPromptIdempotently() {
+        JSONObject root = buildWorkflowRoot();
+        JSONObject llmData = root.getJSONArray("nodes").getJSONObject(1).getJSONObject("data");
+
+        service.prepareRoot(root);
+        service.prepareRoot(root);
+
+        String systemPrompt = llmData.getString("systemPrompt");
+        Assert.assertEquals("请根据问题生成查询语句\n\n{{queryDataContext}}", systemPrompt);
+        Assert.assertEquals(1, countOccurrences(systemPrompt, "{{queryDataContext}}"));
+        JSONObject contextParameter = findParameter(llmData, "queryDataContext");
+        Assert.assertNotNull(contextParameter);
+        String contextValue = contextParameter.getString("value");
+        Assert.assertTrue(contextValue.contains("只输出 SQL"));
+        Assert.assertTrue(contextValue.contains("Sheet1"));
+        Assert.assertTrue(contextValue.contains("col_id"));
+        Assert.assertTrue(contextValue.contains("模型ID"));
+        Assert.assertTrue(contextValue.contains("token AS input_price"));
+    }
+
+    /**
+     * 验证关闭查询上下文后会同步清理参数和系统提示词占位符。
+     */
+    @Test
+    public void testPrepareRootShouldRemoveQueryContextBindingWhenDisabled() {
+        JSONObject root = buildWorkflowRoot();
+        JSONObject llmData = root.getJSONArray("nodes").getJSONObject(1).getJSONObject("data");
+        service.prepareRoot(root);
+        llmData.put("queryContextNodeIds", new JSONArray());
+
+        service.prepareRoot(root);
+
+        Assert.assertEquals("请根据问题生成查询语句", llmData.getString("systemPrompt"));
+        Assert.assertNull(findParameter(llmData, "queryDataContext"));
+    }
+
+    /**
+     * 构造包含查询节点和大模型节点的最小工作流。
+     *
+     * @return 工作流根对象
+     */
+    private JSONObject buildWorkflowRoot() {
+        JSONObject datasetRef = new JSONObject();
+        datasetRef.put("sourceId", SOURCE_ID);
+
+        JSONObject queryData = new JSONObject();
+        queryData.put("datasetRef", datasetRef);
+        queryData.put("querySql", "{{query}}");
+        JSONObject queryNode = buildNode(
+                "query-node",
+                WorkflowDatacenterContentService.SEARCH_NODE_TYPE,
+                queryData);
+
+        JSONArray queryContextNodeIds = new JSONArray();
+        queryContextNodeIds.add("query-node");
+        JSONObject llmData = new JSONObject();
+        llmData.put("systemPrompt", "请根据问题生成查询语句");
+        llmData.put("queryContextNodeIds", queryContextNodeIds);
+        llmData.put("parameters", new JSONArray());
+        JSONObject llmNode = buildNode(
+                "llm-node",
+                WorkflowDatacenterContentService.LLM_NODE_TYPE,
+                llmData);
+
+        JSONArray nodes = new JSONArray();
+        nodes.add(queryNode);
+        nodes.add(llmNode);
+        JSONObject root = new JSONObject();
+        root.put("nodes", nodes);
+        return root;
+    }
+
+    /**
+     * 构造工作流节点。
+     *
+     * @param id 节点标识
+     * @param type 节点类型
+     * @param data 节点数据
+     * @return 工作流节点
+     */
+    private JSONObject buildNode(String id, String type, JSONObject data) {
+        JSONObject node = new JSONObject();
+        node.put("id", id);
+        node.put("type", type);
+        node.put("data", data);
+        return node;
+    }
+
+    /**
+     * 构造字段元数据桩。
+     *
+     * @param fieldName 字段名
+     * @param fieldDesc 字段描述
+     * @param jdbcType JDBC 类型
+     * @return 字段元数据
+     */
+    private DatacenterTableField mockField(String fieldName, String fieldDesc, String jdbcType) {
+        DatacenterTableField field = Mockito.mock(DatacenterTableField.class);
+        Mockito.when(field.getFieldName()).thenReturn(fieldName);
+        Mockito.when(field.getFieldDesc()).thenReturn(fieldDesc);
+        Mockito.when(field.getJdbcType()).thenReturn(jdbcType);
+        return field;
+    }
+
+    /**
+     * 查找指定名称的节点参数。
+     *
+     * @param data 节点数据
+     * @param name 参数名
+     * @return 参数对象，不存在时返回 {@code null}
+     */
+    private JSONObject findParameter(JSONObject data, String name) {
+        JSONArray parameters = data.getJSONArray("parameters");
+        if (parameters == null) {
+            return null;
+        }
+        for (int i = 0; i < parameters.size(); i++) {
+            JSONObject parameter = parameters.getJSONObject(i);
+            if (parameter != null && name.equals(parameter.getString("name"))) {
+                return parameter;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 统计子串出现次数。
+     *
+     * @param value 原始文本
+     * @param target 目标子串
+     * @return 出现次数
+     */
+    private int countOccurrences(String value, String target) {
+        int count = 0;
+        int index = 0;
+        while ((index = value.indexOf(target, index)) >= 0) {
+            count++;
+            index += target.length();
+        }
+        return count;
+    }
+
+    /**
+     * 注入服务私有依赖。
+     *
+     * @param target 目标对象
+     * @param fieldName 字段名
+     * @param value 字段值
+     * @throws Exception 反射注入失败时抛出
+     */
+    private void injectField(Object target, String fieldName, Object value) throws Exception {
+        Field field = target.getClass().getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(target, value);
+    }
+}
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/node/SearchDatasetNodeTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/node/SearchDatasetNodeTest.java
@@ -65,6 +65,43 @@ public class SearchDatasetNodeTest {
        }
    }

+    /**
+     * 验证查询节点可接收大模型返回的单个 SQL Markdown 代码块。
+     *
+     * @throws Exception 反射调用失败时抛出
+     */
+    @Test
+    public void testResolveQuerySqlShouldUnwrapMarkdownSqlCodeBlock() throws Exception {
+        DatasetRef datasetRef = new DatasetRef();
+        datasetRef.setSourceId(BigInteger.valueOf(4004L));
+        SearchDatasetNode node = new SearchDatasetNode(datasetRef, "{{query}}");
+        Map<String, Object> params = new HashMap<>();
+        params.put("query", "```sql SELECT token FROM Sheet1 WHERE col_id = 'deepseek-v4-pro'; ```");
+
+        String sql = invokeResolveQuerySql(node, params);
+
+        Assert.assertEquals("SELECT token FROM Sheet1 WHERE col_id = 'deepseek-v4-pro';", sql);
+    }
+
+    /**
+     * 验证包含说明文本的模型输出不会被宽松清洗，从而继续由 SQL 解析器拒绝。
+     *
+     * @throws Exception 反射调用失败时抛出
+     */
+    @Test
+    public void testResolveQuerySqlShouldKeepTextOutsideSqlCodeBlock() throws Exception {
+        DatasetRef datasetRef = new DatasetRef();
+        datasetRef.setSourceId(BigInteger.valueOf(5005L));
+        SearchDatasetNode node = new SearchDatasetNode(datasetRef, "{{query}}");
+        Map<String, Object> params = new HashMap<>();
+        String generated = "查询语句如下：\n```sql\nSELECT token FROM Sheet1;\n```";
+        params.put("query", generated);
+
+        String sql = invokeResolveQuerySql(node, params);
+
+        Assert.assertEquals(generated, sql);
+    }
+
    private String invokeResolveQuerySql(SearchDatasetNode node, Map<String, Object> params) throws Exception {
        Method method = SearchDatasetNode.class.getDeclaredMethod("resolveQuerySql", Map.class);
        method.setAccessible(true);