Merge branch 'main' into hyq_update

Author: Aias00

spring-ai-alibaba-graph/spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/CompiledGraph.java

@@ -26,6 +26,7 @@
 import com.alibaba.cloud.ai.graph.internal.edge.EdgeValue;
 import com.alibaba.cloud.ai.graph.internal.node.ParallelNode;
 import com.alibaba.cloud.ai.graph.state.StateSnapshot;
+import org.apache.commons.lang3.StringUtils;
 import org.bsc.async.AsyncGenerator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -668,6 +669,10 @@ private Optional<Data<Output>> getEmbedGenerator(Map<String, Object> partialStat
 						if (data != null) {
 
 							if (data instanceof Map<?, ?>) {
+								// FIX #102
+								// Assume that the whatever used appender channel doesn't
+								// accept duplicates
+								// FIX #104: remove generator
 								var partialStateWithoutGenerator = partialState.entrySet()
 									.stream()
 									.filter(e -> !Objects.equals(e.getKey(), generatorEntry.getKey()))
@@ -678,7 +683,8 @@ private Optional<Data<Output>> getEmbedGenerator(Map<String, Object> partialStat
 
 								currentState = OverAllState.updateState(intermediateState, (Map<String, Object>) data,
 										keyStrategyMap);
-								overAllState.updateState(intermediateState);
+
+								overAllState.updateState((Map<String, Object>) data);
 							}
 							else {
 								throw new IllegalArgumentException("Embedded generator must return a Map");
@@ -707,7 +713,7 @@ private CompletableFuture<Data<Output>> evaluateAction(AsyncNodeActionWithConfig
 					}
 
 					this.currentState = OverAllState.updateState(currentState, updateState, keyStrategyMap);
-					overAllState.updateState(updateState);
+					this.overAllState.updateState(updateState);
 					var nextNodeCommand = nextNodeId(currentNodeId, overAllState, currentState, config);
 					nextNodeId = nextNodeCommand.gotoNode();
 					this.currentState = nextNodeCommand.update();

spring-ai-alibaba-graph/spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/OverAllState.java

@@ -23,6 +23,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiConsumer;
 import java.util.function.BinaryOperator;
 import java.util.function.Function;
 import java.util.stream.Collector;
@@ -325,6 +326,23 @@ public Map<String, Object> updateState(Map<String, Object> partialState) {
 		return data();
 	}
 
+	/**
+	 * Updates the internal state based on a schema-defined strategy.
+	 * <p>
+	 * This method first validates the input state, then updates the partial state
+	 * according to the provided key strategies. The updated state is formed by merging
+	 * the original state and the modified partial state, removing any null values in the
+	 * process. The resulting entries are then used to update the internal data map.
+	 * @param state the base state to update; must not be null
+	 * @param partialState the partial state containing updates; may be null or empty
+	 * @param keyStrategies the mapping of keys to update strategies; used to transform
+	 * values
+	 */
+	public void updateStateBySchema(Map<String, Object> state, Map<String, Object> partialState,
+			Map<String, KeyStrategy> keyStrategies) {
+		updateState(updateState(state, partialState, keyStrategies));
+	}
+
 	/**
 	 * Key verify boolean.
 	 * @return the boolean

spring-ai-alibaba-graph/spring-ai-alibaba-graph-example/src/main/java/com/alibaba/cloud/ai/example/graph/stream/LLmSearchStreamController.java

@@ -17,6 +17,7 @@
 
 import com.alibaba.cloud.ai.example.graph.stream.node.BaiduSearchNode;
 import com.alibaba.cloud.ai.example.graph.stream.node.LLmNode;
+import com.alibaba.cloud.ai.example.graph.stream.node.ResultNode;
 import com.alibaba.cloud.ai.example.graph.stream.node.TavilySearchNode;
 import com.alibaba.cloud.ai.graph.*;
 import com.alibaba.cloud.ai.graph.exception.GraphStateException;
@@ -64,19 +65,24 @@ public class LLmSearchStreamController {
 	@Autowired
 	private TavilySearchNode tavilySearchNode;
 
+	@Autowired
+	private ResultNode resultNode;
+
 	@PostConstruct
 	public void init() throws GraphStateException {
 		workflow = new StateGraph(
 				() -> new OverAllState().registerKeyAndStrategy("parallel_result", new AppendStrategy())
 					.registerKeyAndStrategy("messages", new AppendStrategy()))
 			.addNode("baiduSearchNode", node_async(baiduSearchNode))
 			.addNode("tavilySearchNode", node_async(tavilySearchNode))
+			.addNode("resultNode", node_async(resultNode))
 			.addNode("llmNode", node_async(lLmNode))
 			.addEdge(START, "baiduSearchNode")
 			.addEdge(START, "tavilySearchNode")
 			.addEdge("baiduSearchNode", "llmNode")
 			.addEdge("tavilySearchNode", "llmNode")
-			.addEdge("llmNode", END);
+			.addEdge("llmNode", "resultNode")
+			.addEdge("resultNode", END);
 
 	}
 
@@ -98,6 +104,7 @@ public void searchChat(HttpServletRequest request, HttpServletResponse response,
 		CompletableFuture.runAsync(() -> {
 			try (PrintWriter writer = response.getWriter()) {
 				generator.forEachAsync(output -> {
+					System.out.println("output = " + output);
 					try {
 						if (output instanceof StreamingOutput) {
 							writer.write("data: " + ((StreamingOutput) output).chunk() + "\n\n");

spring-ai-alibaba-graph/spring-ai-alibaba-graph-example/src/main/java/com/alibaba/cloud/ai/example/graph/stream/node/ResultNode.java

@@ -0,0 +1,18 @@
+package com.alibaba.cloud.ai.example.graph.stream.node;
+
+import com.alibaba.cloud.ai.graph.OverAllState;
+import com.alibaba.cloud.ai.graph.action.NodeAction;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+@Component
+public class ResultNode implements NodeAction {
+
+	@Override
+	public Map<String, Object> apply(OverAllState state) throws Exception {
+		System.out.println("messages = " + state.value("messages").get());
+		return Map.of();
+	}
+
+}