[BUG] ReactAgent 在stream状态下调用tool calling时会导致流直接关闭（阻塞状态下则可以有完整的输出结果）

[4zhousheng]

Is there an existing issue for this?

• I have searched the existing issues

Current Behavior

我们以一个小小的例子（就以官网的示例代码）演示一下，这里是阻塞式调用

@SpringBootTest
public class TestAgent {
    @Autowired
//    ChatModel chatModel;
    DashScopeChatModel chatModel;
    @Test
    public void test() throws GraphRunnerException, InterruptedException {
        ToolCallback toolCallback = FunctionToolCallback
                .builder("currentWeather", new WeatherService())
                .description("Get the weather in location")
                .inputType(WeatherRequest.class)
                .build();

        ReactAgent agent = ReactAgent
                .builder()
                .name("my_agent")
                .model(chatModel)
                .tools(toolCallback)
                .saver(new MemorySaver())
                .build();
        AssistantMessage message = agent.call("成都市今天天气怎么样");
        System.out.println(message.getText());
}

他的结果根据当前天气信息，成都市今天的温度是30°C。这是一个比较温暖的天气，建议穿着轻薄透气的衣物，注意防晒和补充水分。

这是流式调用

@SpringBootTest
public class TestAgent {
    @Autowired
//    ChatModel chatModel;
    DashScopeChatModel chatModel;
    @Test
    public void test() throws GraphRunnerException, InterruptedException {
        ToolCallback toolCallback = FunctionToolCallback
                .builder("currentWeather", new WeatherService())
                .description("Get the weather in location")
                .inputType(WeatherRequest.class)
                .build();

        ReactAgent agent = ReactAgent
                .builder()
                .name("my_agent")
                .model(chatModel)
                .tools(toolCallback)
                .saver(new MemorySaver())
                .build();
//        AssistantMessage message = agent.call("成都市今天天气怎么样");
//        System.out.println(message.getText());


        Flux<NodeOutput> stream = agent.stream("成都市今天天气怎么样，单位用摄氏度");
        stream.subscribe(System.out::println);
        Thread.sleep(200000);
    }

他的结果（太冗长了我直接截取最后一段）

NodeOutput{node=__END__, agent=my_agent, tokenUsage=null, state={"OverAllState":{"data":{"input":"成都市今天天气怎么样","messages":[{"messageType":"USER","metadata":{"messageType":"USER"},"media":[],"text":"成都市今天天气怎么样"},{"messageType":"ASSISTANT","metadata":{"finishReason":"STOP","search_info":"","role":"ASSISTANT","id":"64087ac9-647f-4876-8c13-85ad06a5b018","messageType":"ASSISTANT","reasoningContent":""},"toolCalls":[],"media":[],"text":"我需要知道您希望使用哪种温度单位来查看成都市的天气信息。请问您希望使用摄氏度（C）还是华氏度（F）？"}]}}}, subGraph=false}

这里AI被强行降智来回避这个问题，而当你在生产环境（使用redissaver不断地追问的时候）几乎包括deepseek也好，qwen也好，那个什么qlm也好，都会在正在为您（tool calling的函数描述)这一句话停止(是stream就断在这一步了）
就像这样
NodeOutput{node=__END__, agent=my_agent, tokenUsage=null, state={"OverAllState":{"data":{"input":"成都市今天天气怎么样，使用摄氏度","messages":[{"messageType":"USER","metadata":{"messageType":"USER"},"media":[],"text":"成都市今天天气怎么样，使用摄氏度"},{"messageType":"ASSISTANT","metadata":{"finishReason":"","search_info":"","role":"ASSISTANT","id":"e67ecbfa-6499-415e-a61f-e78516bea93a","messageType":"ASSISTANT","reasoningContent":""},"toolCalls":[],"media":[],"text":"我来帮您查询成都市今天的天气情况。\n\n\n"}]}}}, subGraph=false}
就在这个我来帮您查询这句中断了，没有下文了，当你在有上下文存储的场景下使用他只会装傻说“我刚才忘记调用工具了”（其实真实原因看起来更像流在调用工具的这一瞬间被掐死了）

Expected Behavior

希望开发者大大们能早日解决这个bug

Steps To Reproduce

这种情况下只能牺牲用户体验来阻塞式调用（拿到结果之后给前端流式也行吧）

Environment

Spring AI Alibaba version(s):1.1.0.0-M5

Debug logs

No response

Anything else?

No response

[vlooong]

#3138
开发者大大似乎已经解决了，等发新版

[4zhousheng]

#3138 开发者大大似乎已经解决了，等发新版

真的吗，真的吗，真的吗，有参考文献吗

[vlooong]

#3309
#3138

[4zhousheng]

#3309 #3138

感谢大佬不吝赐教🥰🥰🥰

[4zhousheng]

ok,暂时新版本没发布，我再贴个小说明（毕竟我只是一个小小的api调用者，比起开源社区的大佬们用法肯定有所差距），在大佬提供的issue贴以及AI的辅助下，当前我们的解决方案是在当前1.1.0.0-M5项目覆盖com.alibaba.cloud.ai.graph.executor(在自己的项目里创建一个这样的包）的NodeExecutor（这里附上AI生成的【亲测可用的】代码）

package com.alibaba.cloud.ai.graph.executor;

import com.alibaba.cloud.ai.graph.GraphResponse;
import com.alibaba.cloud.ai.graph.GraphRunnerContext;
import com.alibaba.cloud.ai.graph.NodeOutput;
import com.alibaba.cloud.ai.graph.action.AsyncNodeActionWithConfig;
import com.alibaba.cloud.ai.graph.action.Command;
import com.alibaba.cloud.ai.graph.action.InterruptableAction;
import com.alibaba.cloud.ai.graph.action.InterruptionMetadata;
import com.alibaba.cloud.ai.graph.exception.RunnableErrors;
import com.alibaba.cloud.ai.graph.streaming.GraphFlux;
import com.alibaba.cloud.ai.graph.streaming.ParallelGraphFlux;
import com.alibaba.cloud.ai.graph.streaming.StreamingOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.ai.chat.messages.AssistantMessage.ToolCall;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.chat.model.Generation;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

public class NodeExecutor extends BaseGraphExecutor {
    private static final Logger log = LoggerFactory.getLogger(NodeExecutor.class);
    private final MainGraphExecutor mainGraphExecutor;

    public NodeExecutor(MainGraphExecutor mainGraphExecutor) {
        this.mainGraphExecutor = mainGraphExecutor;
    }

    public Flux<GraphResponse<NodeOutput>> execute(GraphRunnerContext context, AtomicReference<Object> resultValue) {
        return this.executeNode(context, resultValue);
    }

    private Flux<GraphResponse<NodeOutput>> executeNode(GraphRunnerContext context, AtomicReference<Object> resultValue) {
        try {
            context.setCurrentNodeId(context.getNextNodeId());
            String currentNodeId = context.getCurrentNodeId();
            AsyncNodeActionWithConfig action = context.getNodeAction(currentNodeId);
            if (action == null) {
                return Flux.just(GraphResponse.error(RunnableErrors.missingNode.exception(new String[]{currentNodeId})));
            } else {
                if (action instanceof InterruptableAction) {
                    context.getConfig().metadata("STATE_UPDATE").ifPresent((updateFromFeedback) -> {
                        if (updateFromFeedback instanceof Map) {
                            context.mergeIntoCurrentState((Map)updateFromFeedback);
                        } else {
                            throw new RuntimeException();
                        }
                    });
                    Optional<InterruptionMetadata> interruptMetadata = ((InterruptableAction)action).interrupt(currentNodeId, context.cloneState(context.getCurrentStateData()), context.getConfig());
                    if (interruptMetadata.isPresent()) {
                        resultValue.set(interruptMetadata.get());
                        return Flux.just(GraphResponse.done(interruptMetadata.get()));
                    }
                }

                context.doListeners("__NODE_BEFORE__", (Exception)null);
                CompletableFuture<Map<String, Object>> future = action.apply(context.getOverallState(), context.getConfig());
                return Mono.fromFuture(future).flatMapMany((updateState) -> this.handleActionResult(context, updateState, resultValue)).onErrorResume((error) -> {
                    context.doListeners("__ERROR__", new Exception(error));
                    return Flux.just(GraphResponse.error(error));
                });
            }
        } catch (Exception e) {
            return Flux.just(GraphResponse.error(e));
        }
    }

    private Flux<GraphResponse<NodeOutput>> handleActionResult(GraphRunnerContext context, Map<String, Object> updateState, AtomicReference<Object> resultValue) {
        try {
            Optional<GraphFlux<?>> embedGraphFlux = this.getEmbedGraphFlux(updateState, context);
            if (embedGraphFlux.isPresent()) {
                return this.handleGraphFlux(context, (GraphFlux)embedGraphFlux.get(), updateState, resultValue);
            } else {
                Optional<ParallelGraphFlux> embedParallelGraphFlux = this.getEmbedParallelGraphFlux(updateState);
                if (embedParallelGraphFlux.isPresent()) {
                    return this.handleParallelGraphFlux(context, (ParallelGraphFlux)embedParallelGraphFlux.get(), updateState, resultValue);
                } else {
                    Optional<Flux<GraphResponse<NodeOutput>>> embedFlux = this.getEmbedFlux(context, updateState);
                    if (embedFlux.isPresent()) {
                        return this.handleEmbeddedFlux(context, (Flux)embedFlux.get(), updateState, resultValue);
                    } else {
                        context.mergeIntoCurrentState(updateState);
                        if (context.getCompiledGraph().compileConfig.interruptBeforeEdge() && context.getCompiledGraph().compileConfig.interruptsAfter().contains(context.getCurrentNodeId())) {
                            context.setNextNodeId("__INTERRUPTED__");
                        } else {
                            Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
                            context.setNextNodeId(nextCommand.gotoNode());
                        }

                        NodeOutput output = context.buildNodeOutputAndAddCheckpoint(updateState);
                        context.doListeners("__NODE_AFTER__", (Exception)null);
                        return Flux.just(GraphResponse.of(output)).concatWith(Flux.defer(() -> this.mainGraphExecutor.execute(context, resultValue)));
                    }
                }
            }
        } catch (Exception e) {
            return Flux.just(GraphResponse.error(e));
        }
    }

    // =========================================================================================
    // 【核心修改区域】 getEmbedFlux
    // =========================================================================================
    @SuppressWarnings("unchecked")
    private Optional<Flux<GraphResponse<NodeOutput>>> getEmbedFlux(GraphRunnerContext context, Map<String, Object> partialState) {
        return partialState.entrySet().stream().filter((e) -> e.getValue() instanceof Flux).findFirst().map((e) -> {
            Flux<?> chatFlux = (Flux)e.getValue();
            AtomicReference<ChatResponse> lastChatResponseRef = new AtomicReference((Object)null);
            AtomicReference<GraphResponse<NodeOutput>> lastGraphResponseRef = new AtomicReference((Object)null);

            Flux<GraphResponse<NodeOutput>> mainFlux = chatFlux.filter((element) -> {
                if (element instanceof ChatResponse response) {
                    return response.getResult() != null;
                } else {
                    return true;
                }
            }).doOnError((error) -> log.error("Error occurred in embedded Flux stream for key '{}': {}", new Object[]{e.getKey(), error.getMessage(), error})).map((element) -> {
                if (element instanceof ChatResponse response) {
                    ChatResponse lastResponse = (ChatResponse)lastChatResponseRef.get();

                    if (lastResponse == null) {
                        // 第一帧，直接初始化
                        AssistantMessage message = response.getResult().getOutput();
                        GraphResponse<NodeOutput> lastGraphResponse = GraphResponse.of(context.buildStreamingOutput(message, response, context.getCurrentNodeId()));
                        lastChatResponseRef.set(response);
                        lastGraphResponseRef.set(lastGraphResponse);
                        return lastGraphResponse;
                    } else {
                        AssistantMessage currentMessage = response.getResult().getOutput();

                        // 1. 【合并 ToolCalls (累积)】
                        // 无论当前帧有没有工具，都应该把历史工具和当前工具合并，确保不丢失
                        List<ToolCall> mergedToolCalls = mergeToolCalls(
                                lastResponse.getResult().getOutput().getToolCalls(),
                                currentMessage.getToolCalls()
                        );

                        // 2. 【计算全量文本 (用于更新内部状态 lastChatResponseRef)】
                        String lastMessageText = lastResponse.getResult().getOutput().getText();
                        if (lastMessageText == null) lastMessageText = "";
                        String currentMessageText = currentMessage.getText();
                        // 这是一个累积的文本，用于最终保存
                        String accumulatedText = currentMessageText != null ? lastMessageText.concat(currentMessageText) : lastMessageText;

                        // 3. 【更新内部状态引用 (State Update)】
                        // 这里必须存全量，否则下一帧合并时会丢失历史
                        AssistantMessage accumulatedMessage = new AssistantMessage(
                                accumulatedText,
                                currentMessage.getMetadata(),
                                mergedToolCalls,
                                currentMessage.getMedia()
                        );
                        Generation newGeneration = new Generation(accumulatedMessage, response.getResult().getMetadata());
                        ChatResponse newResponse = new ChatResponse(List.of(newGeneration), response.getMetadata());
                        lastChatResponseRef.set(newResponse); // <--- 更新引用为全量

                        // 4. 【构造流式输出 (Stream Emission)】
                        // 关键点：发给下游流的 AssistantMessage，文本必须是增量 (currentMessageText)，
                        // 但工具必须是全量 (mergedToolCalls)！
                        // 这样 TTS 读的是增量，但下游解析工具时拿到的是全量。
                        AssistantMessage emissionMessage = new AssistantMessage(
                                currentMessageText, // <--- 保持增量！(Delta)
                                currentMessage.getMetadata(),
                                mergedToolCalls,    // <--- 保持全量！(Accumulated)
                                currentMessage.getMedia()
                        );

                        GraphResponse<NodeOutput> lastGraphResponse = GraphResponse.of(context.buildStreamingOutput(emissionMessage, response, context.getCurrentNodeId()));
                        lastGraphResponseRef.set(lastGraphResponse);
                        return lastGraphResponse;
                    }
                } else if (element instanceof GraphResponse) {
                    GraphResponse<NodeOutput> graphResponse = (GraphResponse)element;
                    lastGraphResponseRef.set(graphResponse);
                    return graphResponse;
                } else {
                    String errorMsg = "Unsupported flux element type: " + (element != null ? element.getClass().getSimpleName() : "null");
                    throw new IllegalArgumentException(errorMsg);
                }
            });

            Mono<GraphResponse<NodeOutput>> completionMono = Mono.defer(() -> {
                if (lastChatResponseRef.get() == null) {
                    GraphResponse<?> lastGraphResponse = (GraphResponse)lastGraphResponseRef.get();
                    if (lastGraphResponse != null && lastGraphResponse.resultValue().isPresent()) {
                        Object result = lastGraphResponse.resultValue().get();
                        if (result instanceof Map) {
                            Map resultMap = (Map)result;
                            if (!resultMap.containsKey(e.getKey()) && resultMap.containsKey("messages")) {
                                List<Object> messages = (List)resultMap.get("messages");
                                Object lastMessage = messages.get(messages.size() - 1);
                                if (lastMessage instanceof AssistantMessage) {
                                    AssistantMessage lastAssistantMessage = (AssistantMessage)lastMessage;
                                    resultMap.put(e.getKey(), lastAssistantMessage.getText());
                                }
                            }
                        }
                        return Mono.just((GraphResponse<NodeOutput>) lastGraphResponse);
                    } else {
                        return Mono.empty();
                    }
                } else {
                    return Mono.fromCallable(() -> {
                        Map<String, Object> completionResult = new HashMap();
                        // 这里使用的是 lastChatResponseRef，它是我们上面步骤3中更新的“全量AccumulatedMessage”
                        // 所以最终状态里保存的是完整的文本，没有丢失。
                        completionResult.put((String)e.getKey(), ((ChatResponse)lastChatResponseRef.get()).getResult().getOutput());
                        if (!((String)e.getKey()).equals("messages")) {
                            completionResult.put("messages", ((ChatResponse)lastChatResponseRef.get()).getResult().getOutput());
                        }
                        GraphResponse<?> done = GraphResponse.done(completionResult);
                        return (GraphResponse<NodeOutput>) done;
                    });
                }
            });

            return mainFlux.concatWith(completionMono);
        });
    }



    private Flux<GraphResponse<NodeOutput>> handleEmbeddedFlux(GraphRunnerContext context, Flux<GraphResponse<NodeOutput>> embedFlux, Map<String, Object> partialState, AtomicReference<Object> resultValue) {
        AtomicReference<GraphResponse<NodeOutput>> lastData = new AtomicReference();
        Flux<GraphResponse<NodeOutput>> processedFlux = embedFlux.map((data) -> {
            if (data.getOutput() != null) {
                NodeOutput output = (NodeOutput)data.getOutput().join();
                output.setSubGraph(true);
                GraphResponse<NodeOutput> newData = GraphResponse.of(output);
                lastData.set(newData);
                return newData;
            } else {
                lastData.set(data);
                return data;
            }
        });
        Mono<Void> updateContextMono = Mono.fromRunnable(() -> {
            GraphResponse<NodeOutput> data = (GraphResponse)lastData.get();
            if (data != null) {
                Optional<Object> nodeResultValue = data.resultValue();
                if (nodeResultValue.isPresent() && nodeResultValue.get() instanceof InterruptionMetadata) {
                    context.setReturnFromEmbedWithValue(nodeResultValue.get());
                } else {
                    Map<String, Object> partialStateWithoutFlux = (Map)partialState.entrySet().stream().filter((ex) -> !(ex.getValue() instanceof Flux)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
                    context.mergeIntoCurrentState(partialStateWithoutFlux);
                    Map<String, Object> updateState = new HashMap();
                    if (nodeResultValue.isPresent()) {
                        Object value = nodeResultValue.get();
                        if (!(value instanceof Map)) {
                            throw new IllegalArgumentException("Node stream must return Map result using Data.done(),");
                        }

                        updateState = (Map)value;
                        context.mergeIntoCurrentState(updateState);
                    }

                    try {
                        Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
                        context.setNextNodeId(nextCommand.gotoNode());
                        context.buildNodeOutputAndAddCheckpoint(updateState);
                        context.doListeners("__NODE_AFTER__", (Exception)null);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        });
        return processedFlux.concatWith(updateContextMono.thenMany(Flux.defer(() -> this.mainGraphExecutor.execute(context, resultValue))));
    }

    private Optional<GraphFlux<?>> getEmbedGraphFlux(Map<String, Object> partialState, GraphRunnerContext context) {
        return partialState.entrySet().stream().filter((e) -> e.getValue() instanceof GraphFlux).findFirst().map((e) -> {
            GraphFlux<Object> graphFlux = (GraphFlux)e.getValue();
            return GraphFlux.of(StringUtils.hasText(graphFlux.getNodeId()) ? graphFlux.getNodeId() : context.getCurrentNodeId(), StringUtils.hasText(graphFlux.getKey()) ? graphFlux.getKey() : (String)e.getKey(), graphFlux.getFlux(), graphFlux.getMapResult(), graphFlux.getChunkResult());
        });
    }

    private Optional<ParallelGraphFlux> getEmbedParallelGraphFlux(Map<String, Object> partialState) {
        return partialState.entrySet().stream().filter((e) -> e.getValue() instanceof ParallelGraphFlux).findFirst().map((e) -> (ParallelGraphFlux)e.getValue());
    }

    @SuppressWarnings("unchecked")
    private Flux<GraphResponse<NodeOutput>> handleGraphFlux(GraphRunnerContext context, GraphFlux<?> graphFlux, Map<String, Object> partialState, AtomicReference<Object> resultValue) {
        String effectiveNodeId = graphFlux.getNodeId();
        AtomicReference<Object> lastDataRef = new AtomicReference();
        Flux<GraphResponse<NodeOutput>> processedFlux = graphFlux.getFlux().map((element) -> {
            lastDataRef.set(graphFlux.hasMapResult() ? graphFlux.getMapResult().apply(element) : element);
            StreamingOutput output = context.buildStreamingOutput(graphFlux, element, effectiveNodeId);
            // [Fix 2]: 显式强转 GraphResponse<StreamingOutput> -> GraphResponse<NodeOutput>
            GraphResponse<?> rawResponse = GraphResponse.of(output);
            return (GraphResponse<NodeOutput>) rawResponse;
        }).onErrorMap((error) -> new RuntimeException("GraphFlux processing error in node: " + effectiveNodeId, error));
        Mono<Void> updateContextMono = Mono.fromRunnable(() -> {
            Object lastData = lastDataRef.get();
            Map<String, Object> resultMap = new HashMap();
            resultMap.put(graphFlux.getKey(), lastData);
            Map<String, Object> partialStateWithoutGraphFlux = (Map)partialState.entrySet().stream().filter((ex) -> !(ex.getValue() instanceof GraphFlux)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
            context.mergeIntoCurrentState(partialStateWithoutGraphFlux);
            if (!resultMap.isEmpty()) {
                context.mergeIntoCurrentState(resultMap);
            }

            try {
                Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
                context.setNextNodeId(nextCommand.gotoNode());
                context.buildNodeOutputAndAddCheckpoint(partialStateWithoutGraphFlux);
                context.doListeners("__NODE_AFTER__", (Exception)null);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        return processedFlux.concatWith(updateContextMono.thenMany(Flux.defer(() -> this.mainGraphExecutor.execute(context, resultValue))));
    }

    @SuppressWarnings("unchecked")
    private Flux<GraphResponse<NodeOutput>> handleParallelGraphFlux(GraphRunnerContext context, ParallelGraphFlux parallelGraphFlux, Map<String, Object> partialState, AtomicReference<Object> resultValue) throws Exception {
        if (parallelGraphFlux.isEmpty()) {
            return this.handleNonStreamingResult(context, partialState, resultValue);
        } else {
            Map<String, AtomicReference<Object>> nodeDataRefs = new HashMap();
            List<Flux<GraphResponse<NodeOutput>>> fluxList = (List)parallelGraphFlux.getGraphFluxes().stream().map((graphFlux) -> {
                String nodeId = graphFlux.getNodeId();
                AtomicReference<Object> nodeDataRef = new AtomicReference();
                nodeDataRefs.put(nodeId, nodeDataRef);
                return graphFlux.getFlux().map((element) -> {
                    nodeDataRef.set(graphFlux.hasMapResult() ? graphFlux.getMapResult().apply(element) : element);
                    StreamingOutput output = context.buildStreamingOutput(graphFlux, element, nodeId);
                    // [Fix 2]: 显式强转 GraphResponse<StreamingOutput> -> GraphResponse<NodeOutput>
                    GraphResponse<?> rawResponse = GraphResponse.of(output);
                    return (GraphResponse<NodeOutput>) rawResponse;
                }).onErrorMap((error) -> new RuntimeException("ParallelGraphFlux processing error in node: " + nodeId, error));
            }).collect(Collectors.toList());
            Flux<GraphResponse<NodeOutput>> mergedFlux = Flux.merge(fluxList);
            Mono<Void> updateContextMono = Mono.fromRunnable(() -> {
                Map<String, Object> combinedResultMap = new HashMap();

                for(GraphFlux<?> graphFlux : parallelGraphFlux.getGraphFluxes()) {
                    String nodeId = graphFlux.getNodeId();
                    Object nodeData = ((AtomicReference)nodeDataRefs.get(nodeId)).get();
                    combinedResultMap.put(graphFlux.getKey(), nodeData);
                }

                Map<String, Object> partialStateWithoutParallelGraphFlux = (Map)partialState.entrySet().stream().filter((ex) -> !(ex.getValue() instanceof ParallelGraphFlux)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
                context.mergeIntoCurrentState(partialStateWithoutParallelGraphFlux);
                if (!combinedResultMap.isEmpty()) {
                    context.mergeIntoCurrentState(combinedResultMap);
                }

                try {
                    Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
                    context.setNextNodeId(nextCommand.gotoNode());
                    context.buildNodeOutputAndAddCheckpoint(partialStateWithoutParallelGraphFlux);
                    context.doListeners("__NODE_AFTER__", (Exception)null);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            });
            return mergedFlux.concatWith(updateContextMono.thenMany(Flux.defer(() -> this.mainGraphExecutor.execute(context, resultValue))));
        }
    }

    private Flux<GraphResponse<NodeOutput>> handleNonStreamingResult(GraphRunnerContext context, Map<String, Object> partialState, AtomicReference<Object> resultValue) throws Exception {
        if (context.getCompiledGraph().compileConfig.interruptBeforeEdge() && context.getCompiledGraph().compileConfig.interruptsAfter().contains(context.getCurrentNodeId())) {
            context.setNextNodeId("__INTERRUPTED__");
        } else {
            Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
            context.setNextNodeId(nextCommand.gotoNode());
        }

        NodeOutput output = context.buildNodeOutputAndAddCheckpoint(partialState);
        return Flux.just(GraphResponse.of(output)).concatWith(Flux.defer(() -> this.mainGraphExecutor.execute(context, resultValue)));
    }

    // =========================================================================
    // 移植自 PR #3138 的 ToolCall 合并逻辑
    // =========================================================================
    private static List<ToolCall> mergeToolCalls(List<ToolCall> existingToolCalls, List<ToolCall> newToolCallChunks) {
        if (CollectionUtils.isEmpty(existingToolCalls)) {
            return !CollectionUtils.isEmpty(newToolCallChunks) ? new ArrayList<>(newToolCallChunks) : new ArrayList<>();
        }
        if (CollectionUtils.isEmpty(newToolCallChunks)) {
            return existingToolCalls;
        }

        Map<String, ToolCall> mergedToolCallMap = existingToolCalls.stream()
                .collect(Collectors.toMap(
                        ToolCall::id,
                        call -> new ToolCall(call.id(), call.type(), call.name(), call.arguments()),
                        (a, b) -> a,
                        LinkedHashMap::new));

        for (ToolCall newChunk : newToolCallChunks) {
            String newId = newChunk.id();

            if (newId != null && mergedToolCallMap.containsKey(newId)) {
                ToolCall existingCall = mergedToolCallMap.get(newId);
                String existingArgs = existingCall.arguments() != null ? existingCall.arguments() : "";
                String newArgsChunk = newChunk.arguments() != null ? newChunk.arguments() : "";

                String mergedArgs = existingArgs + newArgsChunk;
                mergedToolCallMap.put(newId, new ToolCall(newId, existingCall.type(), existingCall.name(), mergedArgs));
            } else {
                String key = newId != null ? newId : UUID.randomUUID().toString();
                mergedToolCallMap.put(key, newChunk);
            }
        }
        return new ArrayList<>(mergedToolCallMap.values());
    }
}

虽然过程有点曲折，但黑猫白猫，能抓到耗子的就是毫猫