feat: ReActAgent流式降级优化——可配置重试、非流式回退包装为流式、周期性SSE心跳

src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java

@@ -5,6 +5,7 @@
 package com.openjiuwen.core.singleagent.agents;
 
 import com.openjiuwen.core.common.constants.Constant;
+import com.openjiuwen.core.common.exception.BaseError;
 import com.openjiuwen.core.common.logging.Loggers;
 import com.openjiuwen.core.common.security.UserConfig;
 import com.openjiuwen.core.context.ContextEngine;
@@ -54,6 +55,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * ReAct paradigm Agent implementation.
@@ -82,6 +89,15 @@ public class ReActAgent extends BaseAgent {
     private static final int IDENTITY_SECTION_PRIORITY = 10;
     private static final int SKILLS_SECTION_PRIORITY = 90;
 
+    // 非流式→流式适配器的分块大小（字符数）
+    private static final int STREAM_CHUNK_SIZE = 200;
+
+    // SSE 心跳间隔（毫秒），在回退非流式调用前发送，防止客户端超时断开
+    private static final long HEARTBEAT_INTERVAL_MS = 5000L;
+
+    // 非流式回退时的心跳提示消息
+    private static final String HEARTBEAT_NON_STREAM_MSG = "正在以非流式模式获取响应，请稍候...";
+
     private ReActAgentConfig config;
     private ContextEngine contextEngine;
     private Model llm;
@@ -1344,10 +1360,9 @@ private Map<String, Object> invokeForStream(Object inputs, Session session, Agen
                 Loggers.AGENT.info("ReAct stream iteration " + (iteration + 1) + "/" + config.getMaxIterations());
 
                 injectPendingSteering(ctx, context);
-                AssistantMessage aiMessage = callModelStream(ctx, context, systemMessages, tools, agentSession);
-                if (aiMessage == null) {
-                    aiMessage = callModel(ctx, context, systemMessages, tools);
-                }
+                // 流式失败时重试（仅空响应）；异常时不重试，空响应回退非流式并包装为流式发送
+                AssistantMessage aiMessage = callModelStreamWithRetry(ctx, context, systemMessages, tools,
+                        agentSession).orElse(null);
                 logLlmResponse(aiMessage);
                 AgentCallbackContext.ForceFinishRequest finishAfterModel = ctx.consumeForceFinish();
                 if (finishAfterModel != null) {
@@ -1356,7 +1371,8 @@ private Map<String, Object> invokeForStream(Object inputs, Session session, Agen
                     return finishAfterModel.getResult();
                 }
                 if (aiMessage == null) {
-                    Map<String, Object> result = buildErrorResult("Model stream skipped without terminal result");
+                    Map<String, Object> result =
+                        buildErrorResult("Model stream failed after retries, no terminal result");
                     invokeInputs.setResult(result);
                     return result;
                 }
@@ -1439,6 +1455,103 @@ private AssistantMessage callModelStream(AgentCallbackContext ctx, ModelContext
         return railedModelStreamCall(ctx, agentSession).orElse(null);
     }
 
+    /**
+     * 带重试的流式模型调用。流式请求返回空时按配置重试；若重试仍为空，
+     * 回退为非流式调用并包装为流式 chunk 发送。流式异常时不重试（避免部分 chunk 重复发送）。
+     *
+     * @param ctx ctx
+     * @param context context
+     * @param systemMessages systemMessages
+     * @param tools tools
+     * @param agentSession agentSession
+     * @return the result, or Optional.empty() if stream error or all retries exhausted
+     * @since 0.1.7
+     */
+    private Optional<AssistantMessage> callModelStreamWithRetry(AgentCallbackContext ctx, ModelContext context,
+            List<BaseMessage> systemMessages, List<ToolInfo> tools, AgentSessionApi agentSession) {
+        int maxRetries = config.getStreamMaxRetries();
+        long retryDelayMs = config.getStreamRetryDelayMs();
+
+        for (int attempt = 0; attempt <= maxRetries; attempt++) {
+            try {
+                AssistantMessage aiMessage = callModelStream(ctx, context, systemMessages, tools, agentSession);
+                if (aiMessage != null) {
+                    return Optional.of(aiMessage);
+                }
+                Loggers.AGENT.warning("ReAct stream returned empty (attempt "
+                    + (attempt + 1) + "/" + (maxRetries + 1) + ")");
+            } catch (IllegalStateException e) {
+                // 异常时已可能有部分 chunk 被发送，不重试避免重复发送
+                Loggers.AGENT.error("ReAct stream error (attempt "
+                    + (attempt + 1) + "/" + (maxRetries + 1) + "), aborting retry: " + e.getMessage());
+                return Optional.empty();
+            }
+
+            if (attempt < maxRetries && retryDelayMs > 0) {
+                // 发送心跳，防止客户端在重试等待期间超时断开
+                writeStreamHeartbeat(agentSession, "流式响应为空，正在重试 (" + (attempt + 2) + "/" + (maxRetries + 1) + ")");
+                try {
+                    Thread.sleep(retryDelayMs);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    return Optional.empty();
+                }
+            }
+        }
+
+        // 流式重试全部返回空（非异常）→ 模型可能不支持流式，回退到非流式并包装为流式发送
+        Loggers.AGENT.warning("ReAct stream returned empty after " + (maxRetries + 1)
+            + " attempts, falling back to non-stream with stream wrapping");
+        return fallbackToNonStreamWithHeartbeat(ctx, context, systemMessages, tools, agentSession);
+    }
+
+    /**
+     * 回退到非流式调用，期间用周期性心跳防止客户端超时，并将结果包装为流式 chunk 发送。
+     *
+     * @param ctx callback context
+     * @param context model context
+     * @param systemMessages system messages
+     * @param tools tools
+     * @param agentSession agent session
+     * @return the result, or Optional.empty() if callModel returns null
+     * @since 0.1.7
+     */
+    private Optional<AssistantMessage> fallbackToNonStreamWithHeartbeat(AgentCallbackContext ctx, ModelContext context,
+            List<BaseMessage> systemMessages, List<ToolInfo> tools, AgentSessionApi agentSession) {
+        // 先同步发送一次心跳，确保客户端立即收到切换通知
+        writeStreamHeartbeat(agentSession, HEARTBEAT_NON_STREAM_MSG);
+        // 周期性心跳：在 callModel 阻塞期间按 HEARTBEAT_INTERVAL_MS 间隔发送，防止客户端超时
+        ThreadFactory defaultFactory = Executors.defaultThreadFactory();
+        ScheduledThreadPoolExecutor heartbeatExecutor = new ScheduledThreadPoolExecutor(
+            1, r -> {
+                Thread t = defaultFactory.newThread(r);
+                t.setName("react-heartbeat-" + agentSession.getSessionId());
+                t.setDaemon(true);
+                t.setUncaughtExceptionHandler((thread, ex) ->
+                    Loggers.AGENT.debug("Heartbeat thread exception", ex));
+                return t;
+            },
+            new ThreadPoolExecutor.AbortPolicy());
+        heartbeatExecutor.setRemoveOnCancelPolicy(true);
+        ScheduledFuture<?> heartbeat = heartbeatExecutor.scheduleAtFixedRate(
+            () -> writeStreamHeartbeat(agentSession, HEARTBEAT_NON_STREAM_MSG),
+            HEARTBEAT_INTERVAL_MS, HEARTBEAT_INTERVAL_MS, TimeUnit.MILLISECONDS);
+        AssistantMessage aiMessage;
+        try {
+            aiMessage = callModel(ctx, context, systemMessages, tools);
+        } finally {
+            heartbeat.cancel(false);
+            heartbeatExecutor.shutdownNow();
+        }
+        if (aiMessage != null) {
+            // 有 tool_call 时：content 必须通过流式发送（因为循环会 continue，writeStreamResult 不会发送此轮 content）
+            // 无 tool_call 时：content 由 writeStreamResult 统一发送，避免重复
+            boolean hasToolCalls = aiMessage.getToolCalls() != null && !aiMessage.getToolCalls().isEmpty();
+            writeNonStreamAsStreamChunks(agentSession, aiMessage, 0, hasToolCalls);
+        }
+        return Optional.ofNullable(aiMessage);
+    }
+
     /**
      * preRunStreamSession.
      * 
@@ -1516,21 +1629,106 @@ private void writeAssistantStreamChunk(AgentSessionApi agentSession, AssistantMe
     }
 
     /**
-     * Write a stream error chunk for a checked/unchecked Exception.
+     * 将非流式 AssistantMessage 的 content 拆分为多个 chunk，通过 SSE 逐步发送。
+     * 确保即使获得非流式响应，用户也能收到流式正文。
+     *
+     * @param agentSession agentSession
+     * @param aiMessage non-stream AssistantMessage
+     * @param startIndex starting chunk index
+     * @param shouldSendContent whether to send content as stream chunks;
+     *                          if false, only tool_calls and usage_metadata are sent
+     * @since 0.1.7
+     */
+    private void writeNonStreamAsStreamChunks(AgentSessionApi agentSession, AssistantMessage aiMessage,
+            int startIndex, boolean shouldSendContent) {
+        if (agentSession == null || aiMessage == null) {
+            return;
+        }
+
+        Object rawContent = aiMessage.getContent();
+        String content = rawContent != null ? toText(rawContent) : null;
+        if (shouldSendContent && content != null && !content.isBlank()) {
+            // 按固定长度分块
+            int chunkIndex = startIndex;
+            for (int i = 0; i < content.length(); i += STREAM_CHUNK_SIZE) {
+                int end = Math.min(i + STREAM_CHUNK_SIZE, content.length());
+                String chunkContent = content.substring(i, end);
+
+                AssistantMessageChunk chunk = AssistantMessageChunk.builder()
+                        .content(chunkContent)
+                        .build();
+                writeAssistantStreamChunk(agentSession, chunk, chunkIndex++);
+            }
+            // 发送 tool_calls 和 usage_metadata，确保正文先于工具调用发送
+            writeToolCallsAndUsage(agentSession, aiMessage, chunkIndex);
+            return;
+        }
+
+        // 不发送 content（由 writeStreamResult 统一负责），仅发送 tool_calls 和 usage_metadata
+        writeToolCallsAndUsage(agentSession, aiMessage, startIndex);
+    }
+
+    /**
+     * 发送 tool_calls 和 usage_metadata 为流式 chunk。
      *
-     * @param agentSession session to write to; ignored when null
-     * @param exception stream failure
+     * @param agentSession agentSession
+     * @param aiMessage non-stream AssistantMessage
+     * @param startIndex tool_calls 的起始 chunk index；usage_metadata 使用 startIndex + 1
+     * @since 0.1.7
+     */
+    private void writeToolCallsAndUsage(AgentSessionApi agentSession, AssistantMessage aiMessage, int startIndex) {
+        if (aiMessage.getToolCalls() != null && !aiMessage.getToolCalls().isEmpty()) {
+            AssistantMessageChunk toolChunk = AssistantMessageChunk.builder()
+                    .toolCalls(aiMessage.getToolCalls())
+                    .build();
+            writeAssistantStreamChunk(agentSession, toolChunk, startIndex);
+        }
+        if (aiMessage.getUsageMetadata() != null) {
+            AssistantMessageChunk usageChunk = AssistantMessageChunk.builder()
+                    .usageMetadata(aiMessage.getUsageMetadata())
+                    .build();
+            writeAssistantStreamChunk(agentSession, usageChunk, startIndex + 1);
+        }
+    }
+
+    /**
+     * writeStreamError.
+     *
+     * @param agentSession agentSession
+     * @param exception exception
      * @since 0.1.7
      */
     private void writeStreamError(AgentSessionApi agentSession, Exception exception) {
         writeStreamThrowable(agentSession, exception);
     }
 
     /**
-     * Write a stream error chunk for any Throwable (including Error from worker threads).
+     * 发送 SSE 心跳事件，防止客户端在长时间等待（重试/非流式回退）期间超时断开。
      *
-     * @param agentSession session to write to; ignored when null
-     * @param throwable stream failure
+     * @param agentSession agentSession
+     * @param message 心跳消息
+     * @since 0.1.7
+     */
+    private void writeStreamHeartbeat(AgentSessionApi agentSession, String message) {
+        if (agentSession == null) {
+            return;
+        }
+        try {
+            Map<String, Object> heartbeat = new HashMap<String, Object>();
+            heartbeat.put("content", message);
+            heartbeat.put("result_type", "progress");
+            agentSession.writeStream(new OutputSchema("progress", 0, heartbeat));
+        } catch (BaseError e) {
+            // emitter 可能已关闭，静默忽略，避免中断心跳定时器线程
+            Loggers.AGENT.debug("Heartbeat write skipped: {}", e.getMessage());
+        }
+    }
+
+    /**
+     * writeStreamThrowable.
+     *
+     * @param agentSession agentSession
+     * @param throwable throwable
      * @since 0.1.7
      */
     private void writeStreamThrowable(AgentSessionApi agentSession, Throwable throwable) {

src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgentConfig.java

@@ -60,6 +60,14 @@ public class ReActAgentConfig {
     @Builder.Default
     private boolean shouldFailTaskOnToolError = false;
 
+    // 流式失败重试次数（不含首次调用）
+    @Builder.Default
+    private int streamMaxRetries = 2;
+
+    // 流式重试间隔（毫秒）
+    @Builder.Default
+    private long streamRetryDelayMs = 1000L;
+
     private ModelClientConfig modelClientConfig;
     private ModelRequestConfig modelConfigObj;
     private String sysOperationId;
@@ -171,6 +179,20 @@ public ReActAgentConfig configureMaxIterations(int maxIterations) {
         return this;
     }
 
+    /**
+     * Set the stream retry parameters for streaming model calls.
+     *
+     * @param maxRetries max retry count (excluding the first attempt)
+     * @param retryDelayMs delay between retries in milliseconds
+     * @return this config
+     * @since 0.1.7
+     */
+    public ReActAgentConfig configureStreamRetry(int maxRetries, long retryDelayMs) {
+        this.streamMaxRetries = maxRetries;
+        this.streamRetryDelayMs = retryDelayMs;
+        return this;
+    }
+
     /**
      * Configure the model client without custom certificate or headers.
      * 

src/main/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilder.java

@@ -148,10 +148,14 @@ public PromptSection getSection(String name) {
      */
     public String build() {
         List<PromptSection> ordered = getSectionsForBuild();
+        ordered.removeIf(section -> section == null);
         ordered.sort(Comparator.comparingInt(PromptSection::getPriority));
 
         List<String> parts = new ArrayList<String>();
         for (PromptSection section : ordered) {
+            if (section == null) {
+                continue;
+            }
             String rendered = section.render(language);
             if (rendered != null && !rendered.trim().isEmpty()) {
                 parts.add(rendered);
@@ -192,6 +196,9 @@ private List<PromptSection> getSectionsForBuild() {
         }
         List<PromptSection> filtered = new ArrayList<PromptSection>();
         for (PromptSection section : sections.values()) {
+            if (section == null) {
+                continue;
+            }
             if (MODE_NONE.equals(mode)) {
                 if ("identity".equals(section.getName())) {
                     filtered.add(section);

src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentReactiveTest.java

@@ -11,6 +11,7 @@
 
 import com.openjiuwen.core.foundation.llm.Model;
 import com.openjiuwen.core.foundation.llm.schema.AssistantMessage;
+import com.openjiuwen.core.foundation.llm.schema.AssistantMessageChunk;
 import com.openjiuwen.core.session.AgentSessionApi;
 import com.openjiuwen.core.session.stream.OutputSchema;
 import com.openjiuwen.core.session.stream.StreamMode;
@@ -48,15 +49,21 @@ void invokeAsyncDelegatesThroughConcreteReActAgentInvoke() throws Exception {
     void streamAsyncDelegatesThroughConcreteReActAgentStream() throws Exception {
         ReActAgent agent = newAgent("reactive-react-stream");
         Model model = mock(Model.class);
-        when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()))
-                .thenReturn(AssistantMessage.builder().content("reactive stream ok").build());
+        when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()))
+                .thenReturn(List.of(
+                        AssistantMessageChunk.builder().content("reactive stream ok").build()
+                ).iterator());
         agent.setLlm(model);
 
         AgentSessionApi session =
             new AgentSessionApi("react-stream-session", null, agent.getCard(), List.of(StreamMode.OUTPUT));
 
         StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT)))
-                .expectNextMatches(item -> item instanceof OutputSchema output && "answer".equals(output.getType())
+                .thenConsumeWhile(item -> !(item instanceof OutputSchema output
+                        && "answer".equals(output.getType())
+                        && String.valueOf(output.getPayload()).contains("reactive stream ok")))
+                .expectNextMatches(item -> item instanceof OutputSchema output
+                        && "answer".equals(output.getType())
                         && String.valueOf(output.getPayload()).contains("reactive stream ok"))
                 .verifyComplete();
     }