fix: 优化执行步骤获取逻辑和SQL生成流程, 解决重新生成死循环问题 #130 (#140)

* 优化执行步骤获取逻辑和SQL生成流程, 解决重新生成死循环问题

- 重构 `PlanProcessUtil.java` 中的 getCurrentExecutionStep 方法，新增重载版本以支持传入 plan 参数， 解决重新生成的sql 未生效。
- 优化 SQL 生成完成后的返回逻辑，将返回最佳 SQL 改为返回空 Flux 避免死循环

* fix: format code

---------

Co-authored-by: tanjinxiang <[email protected]>

Author: tjx222

spring-ai-alibaba-data-agent-chat/src/main/java/com/alibaba/cloud/ai/dataagent/node/SqlGenerateNode.java

@@ -76,7 +76,8 @@ public Map<String, Object> apply(OverAllState state) throws Exception {
 
 		// Get necessary input parameters
 		Plan plan = PlanProcessUtil.getPlan(state);
-		ExecutionStep executionStep = PlanProcessUtil.getCurrentExecutionStep(state);
+		ExecutionStep executionStep = PlanProcessUtil.getCurrentExecutionStep(plan,
+				PlanProcessUtil.getCurrentStepNumber(state));
 		ExecutionStep.ToolParameters toolParameters = executionStep.getToolParameters();
 
 		// Execute business logic first - determine what needs to be regenerated
@@ -219,11 +220,11 @@ public Flux<String> get() {
 							roundRef.get())
 					.collect(StringBuilder::new, StringBuilder::append)
 					.map(StringBuilder::toString)
-					.flatMapMany(sql -> Flux.just(nl2SqlService.sqlTrim(sql)).expand(newSql -> {
+					.flatMapMany(sql -> Flux.just(nl2SqlService.sqlTrim(sql)).expandDeep(newSql -> {
 						if (checkSqlFunc.apply(newSql) || roundRef.getAndIncrement() > MAX_OPTIMIZATION_ROUNDS) {
 							String bestSql = bestSqlRef.get();
 							bestSqlConsumer.accept(bestSql);
-							return Flux.just(bestSql);
+							return Flux.empty();
 						}
 						else {
 							return this.get();
@@ -243,7 +244,7 @@ public Flux<String> get() {
 				if (checkSqlFunc.apply(newSql) || roundRef.getAndIncrement() > MAX_OPTIMIZATION_ROUNDS) {
 					String bestSql = bestSqlRef.get();
 					bestSqlConsumer.accept(bestSql);
-					return Flux.just(bestSql);
+					return Flux.empty();
 				}
 				else {
 					return reGenerateSupplier.get();

spring-ai-alibaba-data-agent-chat/src/main/java/com/alibaba/cloud/ai/dataagent/util/PlanProcessUtil.java

@@ -60,7 +60,18 @@ private PlanProcessUtil() {
 	public static ExecutionStep getCurrentExecutionStep(OverAllState state) {
 		Plan plan = getPlan(state);
 		int currentStep = getCurrentStepNumber(state);
+		return getCurrentExecutionStep(plan, currentStep);
+	}
 
+	/**
+	 * Get the current execution step from the plan
+	 * @param plan the plan object
+	 * @param currentStep current step
+	 * @return the current execution step
+	 * @throws IllegalStateException if plan output is empty, plan parsing fails, or step
+	 * index is out of range
+	 */
+	public static ExecutionStep getCurrentExecutionStep(Plan plan, Integer currentStep) {
 		List<ExecutionStep> executionPlan = plan.getExecutionPlan();
 		if (executionPlan == null || executionPlan.isEmpty()) {
 			throw new IllegalStateException("执行计划为空");