Skip to content

Commit 217cee8

Browse files
authored
fix(workflow): accumulate multi-step SQL results (#575)
* fix(workflow): accumulate multi-step SQL results * refactor: separate multi-step SQL result preparation * chore: remove PR validation illustration * test: clarify multi-step Python input regression
1 parent 15398a3 commit 217cee8

8 files changed

Lines changed: 275 additions & 58 deletions

File tree

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2024-2026 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.alibaba.cloud.ai.dataagent.util;
17+
18+
import com.alibaba.cloud.ai.dataagent.bo.schema.ResultSetBO;
19+
import java.util.Comparator;
20+
import java.util.List;
21+
import java.util.Map;
22+
import java.util.Objects;
23+
import lombok.extern.slf4j.Slf4j;
24+
25+
/**
26+
* Converts the SQL execution history into ordered result sets for downstream Python
27+
* nodes.
28+
*/
29+
@Slf4j
30+
public final class SqlResultSetExtractor {
31+
32+
private static final String STEP_PREFIX = "step_";
33+
34+
private SqlResultSetExtractor() {
35+
}
36+
37+
public static List<List<Map<String, String>>> extractAll(Map<String, String> executionResults) {
38+
return extract(executionResults, null);
39+
}
40+
41+
public static List<List<Map<String, String>>> extractSamples(Map<String, String> executionResults,
42+
int maxRowsPerStep) {
43+
if (maxRowsPerStep <= 0) {
44+
throw new IllegalArgumentException("maxRowsPerStep must be greater than zero");
45+
}
46+
return extract(executionResults, maxRowsPerStep);
47+
}
48+
49+
private static List<List<Map<String, String>>> extract(Map<String, String> executionResults,
50+
Integer maxRowsPerStep) {
51+
if (executionResults == null || executionResults.isEmpty()) {
52+
return List.of();
53+
}
54+
return executionResults.entrySet()
55+
.stream()
56+
.map(SqlResultSetExtractor::toStepResult)
57+
.filter(Objects::nonNull)
58+
.sorted(Comparator.comparingInt(StepResult::stepNumber))
59+
.map(stepResult -> parseRows(stepResult, maxRowsPerStep))
60+
.filter(Objects::nonNull)
61+
.toList();
62+
}
63+
64+
private static StepResult toStepResult(Map.Entry<String, String> entry) {
65+
String key = entry.getKey();
66+
if (key == null || !key.startsWith(STEP_PREFIX)) {
67+
return null;
68+
}
69+
try {
70+
return new StepResult(Integer.parseInt(key.substring(STEP_PREFIX.length())), entry.getValue());
71+
}
72+
catch (NumberFormatException ex) {
73+
log.warn("Ignoring SQL execution result with invalid step key: {}", key);
74+
return null;
75+
}
76+
}
77+
78+
private static List<Map<String, String>> parseRows(StepResult stepResult, Integer maxRowsPerStep) {
79+
try {
80+
ResultSetBO resultSet = JsonUtil.getObjectMapper().readValue(stepResult.json(), ResultSetBO.class);
81+
List<Map<String, String>> rows = resultSet.getData();
82+
if (rows == null || rows.isEmpty()) {
83+
return List.of();
84+
}
85+
return maxRowsPerStep == null ? List.copyOf(rows) : rows.stream().limit(maxRowsPerStep).toList();
86+
}
87+
catch (Exception ex) {
88+
log.warn("Ignoring invalid SQL execution result for step_{}: {}", stepResult.stepNumber(), ex.getMessage());
89+
return null;
90+
}
91+
}
92+
93+
private record StepResult(int stepNumber, String json) {
94+
}
95+
96+
}

‎data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/workflow/node/PythonExecuteNode.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,14 +28,14 @@
2828
import com.alibaba.cloud.ai.dataagent.util.ChatResponseUtil;
2929
import com.alibaba.cloud.ai.dataagent.util.FluxUtil;
3030
import com.alibaba.cloud.ai.dataagent.util.JsonUtil;
31+
import com.alibaba.cloud.ai.dataagent.util.SqlResultSetExtractor;
3132
import com.alibaba.cloud.ai.dataagent.util.StateUtil;
3233
import com.fasterxml.jackson.databind.ObjectMapper;
3334
import lombok.extern.slf4j.Slf4j;
3435
import org.springframework.ai.chat.model.ChatResponse;
3536
import org.springframework.stereotype.Component;
3637
import reactor.core.publisher.Flux;
3738

38-
import java.util.ArrayList;
3939
import java.util.List;
4040
import java.util.Map;
4141

@@ -77,8 +77,9 @@ public Map<String, Object> apply(OverAllState state) throws Exception {
7777
try {
7878
// Get context
7979
String pythonCode = StateUtil.getStringValue(state, PYTHON_GENERATE_NODE_OUTPUT);
80-
List<Map<String, String>> sqlResults = StateUtil.hasValue(state, SQL_RESULT_LIST_MEMORY)
81-
? StateUtil.getListValue(state, SQL_RESULT_LIST_MEMORY) : new ArrayList<>();
80+
Map<String, String> executionResults = StateUtil.getObjectValue(state, SQL_EXECUTE_NODE_OUTPUT, Map.class,
81+
Map.of());
82+
List<List<Map<String, String>>> sqlResults = SqlResultSetExtractor.extractAll(executionResults);
8283

8384
// 检查重试次数
8485
int triesCount = StateUtil.getObjectValue(state, PYTHON_TRIES_COUNT, Integer.class, 0);

‎data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/workflow/node/PythonGenerateNode.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import com.alibaba.cloud.ai.dataagent.util.FluxUtil;
3030
import com.alibaba.cloud.ai.dataagent.util.MarkdownParserUtil;
3131
import com.alibaba.cloud.ai.dataagent.util.PlanProcessUtil;
32+
import com.alibaba.cloud.ai.dataagent.util.SqlResultSetExtractor;
3233
import com.alibaba.cloud.ai.dataagent.util.StateUtil;
3334
import com.fasterxml.jackson.annotation.JsonInclude;
3435
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -37,7 +38,6 @@
3738
import org.springframework.stereotype.Component;
3839
import reactor.core.publisher.Flux;
3940

40-
import java.util.ArrayList;
4141
import java.util.List;
4242
import java.util.Map;
4343

@@ -72,8 +72,10 @@ public Map<String, Object> apply(OverAllState state) throws Exception {
7272

7373
// Get context
7474
SchemaDTO schemaDTO = StateUtil.getObjectValue(state, TABLE_RELATION_OUTPUT, SchemaDTO.class);
75-
List<Map<String, String>> sqlResults = StateUtil.hasValue(state, SQL_RESULT_LIST_MEMORY)
76-
? StateUtil.getListValue(state, SQL_RESULT_LIST_MEMORY) : new ArrayList<>();
75+
Map<String, String> executionResults = StateUtil.getObjectValue(state, SQL_EXECUTE_NODE_OUTPUT, Map.class,
76+
Map.of());
77+
List<List<Map<String, String>>> sqlResults = SqlResultSetExtractor.extractSamples(executionResults,
78+
SAMPLE_DATA_NUMBER);
7779
boolean codeRunSuccess = StateUtil.getObjectValue(state, PYTHON_IS_SUCCESS, Boolean.class, true);
7880
int triesCount = StateUtil.getObjectValue(state, PYTHON_TRIES_COUNT, Integer.class, 0);
7981

@@ -105,8 +107,8 @@ public Map<String, Object> apply(OverAllState state) throws Exception {
105107
.render(Map.of("python_memory", codeExecutorProperties.getLimitMemory().toString(), "python_timeout",
106108
codeExecutorProperties.getCodeTimeout().toSeconds() + "s", "database_schema",
107109
objectMapper.writeValueAsString(schemaDTO), "sample_input",
108-
objectMapper.writeValueAsString(sqlResults.stream().limit(SAMPLE_DATA_NUMBER).toList()),
109-
"plan_description", objectMapper.writeValueAsString(toolParameters)));
110+
objectMapper.writeValueAsString(sqlResults), "plan_description",
111+
objectMapper.writeValueAsString(toolParameters)));
110112

111113
Flux<ChatResponse> pythonGenerateFlux = llmService.call(systemPrompt, userPrompt);
112114

‎data-agent-management/src/main/resources/prompts/python-generator.txt‎

Lines changed: 34 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
**请严格遵循以下规范生成代码**:
66
1. **纯净输出**:只输出可执行的Python代码,禁止包含任何额外说明或自然语言。在代码内部需要有适量的注释方便阅读。
77
- **特别注意**:模型输出的文本直接接入Python解释器运行,因此不要添加任何额外符号,**比如Markdown的代码块标记符号**!
8-
2. **输入规范**:从`sys.stdin`读取JSON数据(List[Dict]),使用`json.load(sys.stdin)`。
8+
2. **输入规范**:从`sys.stdin`读取JSON数据(List[List[Dict]]),外层数组按执行顺序保存每个SQL步骤的结果集。使用`json.load(sys.stdin)`,根据分析需求分别处理或关联各结果集,不要把不同结构的结果直接混合。
99
3. **输出规范**:最终结果必须是JSON对象(Dict),通过`print(json.dumps(result, ensure_ascii=False))`输出,JSON字段可以自定义,但要满足用户需求。
1010
4. **错误处理**:使用以下结构捕获所有异常:
1111
```python
@@ -49,32 +49,32 @@ try:
4949
# 从stdin读取输入数据
5050
input_data = json.load(sys.stdin)
5151

52-
# 将输入数据转换为DataFrame以便于分析
53-
df = pd.DataFrame(input_data)
54-
55-
# 自动类型推断:将字符串形式的数值转换为实际数值类型(解决后端统一返回字符串的问题)
56-
for col in df.columns:
57-
df[col] = pd.to_numeric(df[col], errors='ignore')
52+
# 每个元素对应一个SQL步骤的结果集,分别转换为DataFrame
53+
dataframes = []
54+
for step_data in input_data:
55+
df = pd.DataFrame(step_data)
56+
# 自动类型推断:将字符串形式的数值转换为实际数值类型
57+
for col in df.columns:
58+
df[col] = pd.to_numeric(df[col], errors='ignore')
59+
dataframes.append(df)
5860

5961
# 动态分析逻辑
6062
# 示例:计算某些统计指标
6163
result = \{
62-
"summary": \{\},
63-
"details": []
64+
"step_summaries": []
6465
\}
6566

66-
# 示例逻辑:计算每列的平均值(可根据需求调整)
67-
for column in df.columns:
68-
if pd.api.types.is_numeric_dtype(df[column]):
69-
result["summary"][column] = \{
70-
"mean": df[column].mean(),
71-
"min": df[column].min(),
72-
"max": df[column].max()
73-
\}
74-
75-
# 示例逻辑:将原始数据分组并统计(可根据需求调整)
76-
grouped = df.groupby(list(df.columns[:2])).size().reset_index(name="count")
77-
result["details"] = grouped.to_dict(orient="records")
67+
# 示例逻辑:分别统计每个SQL步骤的数值字段
68+
for step_number, df in enumerate(dataframes, start=1):
69+
summary = \{"step_number": step_number, "statistics": \{\}\}
70+
for column in df.columns:
71+
if pd.api.types.is_numeric_dtype(df[column]):
72+
summary["statistics"][column] = \{
73+
"mean": df[column].mean(),
74+
"min": df[column].min(),
75+
"max": df[column].max()
76+
\}
77+
result["step_summaries"].append(summary)
7878

7979
# 输出结果为JSON对象
8080
print(json.dumps(result, ensure_ascii=False))
@@ -89,7 +89,7 @@ except Exception:
8989

9090
# 注意事项
9191

92-
1. **输入验证**:确保代码能够正确处理空输入或格式不正确的输入,并在异常时提供清晰的错误信息。**处理的数据必须来自`json.load(sys.stdin)`**。
92+
1. **输入验证**:确保代码能够正确处理空输入、空结果集或格式不正确的输入,并在异常时提供清晰的错误信息。**处理的数据必须来自`json.load(sys.stdin)`,外层数组中的每个元素代表一个SQL步骤的结果集**。
9393
2. **类型转换**:由于后端SQL查询结果统一转换为字符串格式,**必须在DataFrame创建后进行类型推断**,使用`pd.to_numeric()`将数值字符串转换为实际数值类型,以便进行数学运算和统计分析。
9494
3. **性能优化**:尽量减少不必要的计算和内存占用,确保代码在性能约束内高效运行。
9595
4. **结果完整性**:输出的JSON对象应全面反映分析结果,且字段命名清晰易懂。
@@ -114,26 +114,21 @@ try:
114114
# 从stdin读取输入数据
115115
input_data = json.load(sys.stdin)
116116

117-
# 转换为DataFrame
118-
df = pd.DataFrame(input_data)
119-
120-
# 自动类型推断:将字符串形式的数值转换为实际数值类型
121-
for col in df.columns:
122-
df[col] = pd.to_numeric(df[col], errors='ignore')
117+
# 分别读取两个SQL步骤的结果
118+
channel_df = pd.DataFrame(input_data[0]) if len(input_data) > 0 else pd.DataFrame()
119+
conversion_df = pd.DataFrame(input_data[1]) if len(input_data) > 1 else pd.DataFrame()
123120

124-
# 动态分析逻辑
125-
result = \{
126-
"channel_stats": []
127-
\}
121+
for df in (channel_df, conversion_df):
122+
for col in df.columns:
123+
df[col] = pd.to_numeric(df[col], errors='ignore')
128124

129-
# 计算每个渠道的线索数量和转化率
130-
if "channel" in df.columns and "conversion" in df.columns:
131-
grouped = df.groupby("channel").agg(
132-
lead_count=("conversion", "size"),
133-
conversion_rate=("conversion", "mean")
134-
).reset_index()
125+
# 根据共同维度关联两个步骤的数据,再完成综合分析
126+
result = \{"channel_stats": []\}
127+
common_columns = list(set(channel_df.columns) & set(conversion_df.columns))
128+
if common_columns:
129+
merged = channel_df.merge(conversion_df, on=common_columns, how="inner")
130+
result["channel_stats"] = merged.to_dict(orient="records")
135131

136-
result["channel_stats"] = grouped.to_dict(orient="records")
137132

138133
# 输出结果为JSON对象
139134
print(json.dumps(result, ensure_ascii=False))
Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Copyright 2024-2026 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.alibaba.cloud.ai.dataagent.util;
17+
18+
import static org.junit.jupiter.api.Assertions.assertEquals;
19+
import static org.junit.jupiter.api.Assertions.assertThrows;
20+
21+
import java.util.List;
22+
import java.util.Map;
23+
import org.junit.jupiter.api.Test;
24+
25+
class SqlResultSetExtractorTest {
26+
27+
@Test
28+
void extractAll_ordersStepsAndPreservesResultSetBoundaries() {
29+
Map<String, String> executionResults = Map.of("step_10", "{\"data\":[{\"department\":\"support\"}]}", "step_2",
30+
"{\"data\":[{\"department\":\"engineering\"}]}", "step_1", "{\"data\":[{\"department\":\"sales\"}]}");
31+
32+
List<List<Map<String, String>>> result = SqlResultSetExtractor.extractAll(executionResults);
33+
34+
assertEquals(List.of(List.of(Map.of("department", "sales")), List.of(Map.of("department", "engineering")),
35+
List.of(Map.of("department", "support"))), result);
36+
}
37+
38+
@Test
39+
void extractSamples_limitsEachStepIndependently() {
40+
Map<String, String> executionResults = Map.of("step_1", "{\"data\":[{\"id\":\"1\"},{\"id\":\"2\"}]}", "step_2",
41+
"{\"data\":[{\"id\":\"3\"},{\"id\":\"4\"}]}");
42+
43+
List<List<Map<String, String>>> result = SqlResultSetExtractor.extractSamples(executionResults, 1);
44+
45+
assertEquals(List.of(List.of(Map.of("id", "1")), List.of(Map.of("id", "3"))), result);
46+
}
47+
48+
@Test
49+
void extractAll_skipsInvalidEntriesAndKeepsEmptyResultSets() {
50+
Map<String, String> executionResults = Map.of("not_a_step", "{\"data\":[{\"id\":\"0\"}]}", "step_bad", "{}",
51+
"step_1", "{\"data\":[]}", "step_2", "not-json");
52+
53+
List<List<Map<String, String>>> result = SqlResultSetExtractor.extractAll(executionResults);
54+
55+
assertEquals(List.of(List.of()), result);
56+
}
57+
58+
@Test
59+
void extractSamples_rejectsNonPositiveLimit() {
60+
assertThrows(IllegalArgumentException.class, () -> SqlResultSetExtractor.extractSamples(Map.of(), 0));
61+
}
62+
63+
}

‎data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/workflow/node/python/PythonExecuteNodeTest.java‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@
1919
import static org.junit.jupiter.api.Assertions.*;
2020
import static org.mockito.ArgumentMatchers.any;
2121
import static org.mockito.ArgumentMatchers.anyString;
22+
import static org.mockito.Mockito.verify;
2223
import static org.mockito.Mockito.when;
2324

24-
import java.util.ArrayList;
2525
import java.util.List;
2626
import java.util.Map;
2727

@@ -77,6 +77,7 @@ private OverAllState createTestState() {
7777
state.registerKeyAndStrategy(PYTHON_TRIES_COUNT, new ReplaceStrategy());
7878
state.registerKeyAndStrategy(PYTHON_FALLBACK_MODE, new ReplaceStrategy());
7979
state.registerKeyAndStrategy(SQL_RESULT_LIST_MEMORY, new ReplaceStrategy());
80+
state.registerKeyAndStrategy(SQL_EXECUTE_NODE_OUTPUT, new ReplaceStrategy());
8081
return state;
8182
}
8283

@@ -194,13 +195,13 @@ void apply_emptyOutput_setsEmptyResult() throws Exception {
194195
}
195196

196197
@Test
197-
void apply_withSqlResultsAsCSV_passesCorrectly() throws Exception {
198+
void apply_withMultipleSqlResults_passesOrderedResultSetsToPython() throws Exception {
198199
OverAllState state = createTestState();
199200
state.updateState(Map.of(PYTHON_GENERATE_NODE_OUTPUT, "print('processed')"));
200-
List<Map<String, String>> sqlResults = new ArrayList<>();
201-
sqlResults.add(Map.of("name", "Alice", "amount", "100"));
202-
sqlResults.add(Map.of("name", "Bob", "amount", "200"));
203-
state.updateState(Map.of(SQL_RESULT_LIST_MEMORY, sqlResults));
201+
state.updateState(Map.of(SQL_RESULT_LIST_MEMORY,
202+
List.of(Map.of("department", "engineering", "headcount", "20")), SQL_EXECUTE_NODE_OUTPUT,
203+
Map.of("step_2", "{\"data\":[{\"department\":\"engineering\",\"headcount\":\"20\"}]}", "step_1",
204+
"{\"data\":[{\"department\":\"sales\",\"revenue\":\"100\"}]}")));
204205

205206
when(pythonCodeExecutor.runTask(any())).thenReturn(PythonCodeExecutorService.TaskResponse.success("processed"));
206207
when(jsonParseUtil.tryConvertToObject(anyString(), any(Class.class))).thenReturn(null);
@@ -209,6 +210,15 @@ void apply_withSqlResultsAsCSV_passesCorrectly() throws Exception {
209210
assertNotNull(result);
210211
assertTrue(result.containsKey(PYTHON_EXECUTE_NODE_OUTPUT));
211212
assertNotNull(result.get(PYTHON_EXECUTE_NODE_OUTPUT));
213+
214+
ArgumentCaptor<CodePoolExecutorService.TaskRequest> request = ArgumentCaptor
215+
.forClass(CodePoolExecutorService.TaskRequest.class);
216+
verify(codePoolExecutor).runTask(request.capture());
217+
String input = request.getValue().input();
218+
String message = "Expected ordered sales and engineering result sets, but Python received: " + input;
219+
assertTrue(input.startsWith("[[{"), message);
220+
assertTrue(input.indexOf("sales") < input.indexOf("engineering"), message);
221+
assertTrue(input.endsWith("}]]"), message);
212222
}
213223

214224
@Test

0 commit comments

Comments
 (0)