Fix/graph observability langfuse (#383)

* fix(graph): 1. correct streamingNode inputKey initialization & remove deprecated KeyStrategy
2. migrate GraphController to reactive execution to resolve Reactor blocking issue

- ensure streamingNode correctly reads inputKey from state
- remove redundant KeyStrategy configuration and references
- refactor controller to return Flux/Mono instead of blocking calls
- align with WebFlux non-blocking architecture
- resolves IllegalStateException caused by block() on reactor-http-nio thread

* refactor(stream): replace deprecated StreamingChatGenerator with FluxConverter for reactive stream building

Author: wangx-wx

spring-ai-alibaba-graph-example/graph-observability-langfuse/src/main/java/com/alibaba/cloud/ai/graph/config/GraphConfiguration.java

@@ -145,7 +145,7 @@ public StateGraph observabilityGraph(ChatClient chatClient) throws GraphStateExc
 		MergeNode mergeNode = new MergeNode(Lists.newArrayList("parallel_output1", "parallel_output2"), "sub_input");
 
 		// Streaming node - real-time AI response
-		StreamingChatNode streamingNode = StreamingChatNode.create("StreamingNode", "final_output", "streaming_output",
+		StreamingChatNode streamingNode = StreamingChatNode.create("StreamingNode", "subgraph_final_output", "streaming_output",
 				chatClient, "Please perform detailed analysis on the subgraph results:");
 
 		// End node - final output formatting
@@ -164,9 +164,7 @@ public StateGraph observabilityGraph(ChatClient chatClient) throws GraphStateExc
 				.addPatternStrategy("sub_input", new ReplaceStrategy())
 				.addPatternStrategy("sub_output1", new ReplaceStrategy())
 				.addPatternStrategy("sub_output2", new ReplaceStrategy())
-				.addPatternStrategy("_subgraph", new ReplaceStrategy())
 				.addPatternStrategy("subgraph_final_output", new ReplaceStrategy())
-				.addPatternStrategy("final_output", new ReplaceStrategy())
 				.addPatternStrategy("streaming_output", new ReplaceStrategy())
 				.addPatternStrategy("summary_output", new ReplaceStrategy())
 				.addPatternStrategy("end_output", new ReplaceStrategy())

spring-ai-alibaba-graph-example/graph-observability-langfuse/src/main/java/com/alibaba/cloud/ai/graph/controller/GraphController.java

@@ -18,11 +18,16 @@
 
 import com.alibaba.cloud.ai.graph.CompiledGraph;
 import com.alibaba.cloud.ai.graph.OverAllState;
+import com.alibaba.cloud.ai.graph.RunnableConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -45,37 +50,41 @@ public class GraphController {
 	@Autowired
 	private CompiledGraph compiledGraph;
 
+	private final static Logger logger = LoggerFactory.getLogger(GraphController.class);
+
 	/**
 	 * Execute graph processing
 	 * @param input the input content to process
 	 * @return processing result with success status and output
 	 */
 	@GetMapping("/execute")
-	public Map<String, Object> execute(@RequestParam(value = "prompt", defaultValue = "Hello World") String input) {
-		try {
-			// Create initial state
-			Map<String, Object> initialState = new HashMap<>();
-			initialState.put("input", input);
-
-			// Execute graph
-			OverAllState result = compiledGraph.call(initialState).get();
+	public Mono<Map<String, Object>> execute(@RequestParam(value = "prompt", defaultValue = "Hello World") String input) {
+		return Mono.fromCallable(() -> {
+					// Create initial state
+					Map<String, Object> initialState = Map.of("input", input);
+					RunnableConfig runnableConfig = RunnableConfig.builder().build();
 
-			// Return result
-			Map<String, Object> response = new HashMap<>();
-			response.put("success", true);
-			response.put("input", input);
-			response.put("output", result.value("end_output").orElse("No output"));
-			response.put("logs", result.value("logs").orElse("No logs"));
+					// Execute graph
+					OverAllState result = compiledGraph.call(initialState, runnableConfig).get();
 
-			return response;
+					// Return result
+					Map<String, Object> response = new HashMap<>();
+					response.put("success", true);
+					response.put("input", input);
+					response.put("output", result.value("end_output").orElse("No output"));
+					response.put("logs", result.value("logs").orElse("No logs"));
 
-		}
-		catch (Exception e) {
-			Map<String, Object> errorResponse = new HashMap<>();
-			errorResponse.put("success", false);
-			errorResponse.put("error", e.getMessage());
-			return errorResponse;
-		}
+					logger.info("分析成功：{}", response);
+					return response;
+				})
+				.subscribeOn(Schedulers.boundedElastic())
+				.onErrorResume(e -> {
+					logger.error("异常结束 [{}]", e.getMessage(), e);
+					Map<String, Object> errorResponse = new HashMap<>();
+					errorResponse.put("success", false);
+					errorResponse.put("error", e.getMessage());
+					return Mono.just(errorResponse);
+				});
 	}
 
 }

spring-ai-alibaba-graph-example/graph-observability-langfuse/src/main/java/com/alibaba/cloud/ai/graph/node/StreamingChatNode.java

@@ -16,11 +16,11 @@
 
 package com.alibaba.cloud.ai.graph.node;
 
-import com.alibaba.cloud.ai.graph.NodeOutput;
+import com.alibaba.cloud.ai.graph.GraphResponse;
 import com.alibaba.cloud.ai.graph.OverAllState;
 import com.alibaba.cloud.ai.graph.action.NodeAction;
-import com.alibaba.cloud.ai.graph.async.AsyncGenerator;
-import com.alibaba.cloud.ai.graph.streaming.StreamingChatGenerator;
+import com.alibaba.cloud.ai.graph.streaming.FluxConverter;
+import com.alibaba.cloud.ai.graph.streaming.StreamingOutput;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.ai.chat.client.ChatClient;
@@ -103,15 +103,15 @@ public Map<String, Object> apply(OverAllState state) throws Exception {
 				});
 
 			// Wrap streaming response with StreamingChatGenerator
-			AsyncGenerator<? extends NodeOutput> generator = StreamingChatGenerator.builder()
-				.startingNode(nodeName + "_stream")
-				.startingState(state)
-				.mapResult(response -> {
-					String content = response.getResult().getOutput().getText();
-					logger.info("{}: mapResult emit chunk: {}", nodeName, content);
-					return Map.of(outputKey, content);
-				})
-				.build(chatResponseFlux);
+			Flux<GraphResponse<StreamingOutput>> generator = FluxConverter.builder()
+					.startingNode(nodeName + "_stream")
+					.startingState(state)
+					.mapResult(resp -> {
+						String content = resp.getResult().getOutput().getText();
+						logger.info("{} mapResult emit content: {}", nodeName, content);
+						return Map.of(outputKey, content);
+					})
+					.build(chatResponseFlux);
 
 			logger.info("{} streaming processing setup completed", nodeName);
 			return Map.of(outputKey, generator);