diff --git a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java b/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java index 49d02e76f..8cff05253 100644 --- a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java +++ b/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 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 invokeForStream(Object inputs, Session session, Agen return finishAfterModel.getResult(); } if (aiMessage == null) { - Map result = buildErrorResult("Model stream skipped without terminal result"); + Map 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 callModelStreamWithRetry(AgentCallbackContext ctx, ModelContext context, + List systemMessages, List 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 fallbackToNonStreamWithHeartbeat(AgentCallbackContext ctx, ModelContext context, + List systemMessages, List 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,10 +1629,73 @@ 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) { @@ -1527,10 +1703,32 @@ private void writeStreamError(AgentSessionApi agentSession, Exception 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 heartbeat = new HashMap(); + 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) { diff --git a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgentConfig.java b/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgentConfig.java index 719ad30df..ff30ab274 100644 --- a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgentConfig.java +++ b/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. * diff --git a/src/main/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilder.java b/src/main/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilder.java index 828955c60..294dc6045 100644 --- a/src/main/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilder.java +++ b/src/main/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilder.java @@ -148,10 +148,14 @@ public PromptSection getSection(String name) { */ public String build() { List ordered = getSectionsForBuild(); + ordered.removeIf(section -> section == null); ordered.sort(Comparator.comparingInt(PromptSection::getPriority)); List parts = new ArrayList(); 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 getSectionsForBuild() { } List filtered = new ArrayList(); for (PromptSection section : sections.values()) { + if (section == null) { + continue; + } if (MODE_NONE.equals(mode)) { if ("identity".equals(section.getName())) { filtered.add(section); diff --git a/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentReactiveTest.java b/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentReactiveTest.java index 02a313ef2..124b621d4 100644 --- a/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentReactiveTest.java +++ b/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(); } diff --git a/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentStreamDegradationTest.java b/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentStreamDegradationTest.java new file mode 100644 index 000000000..e7c6a17a0 --- /dev/null +++ b/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentStreamDegradationTest.java @@ -0,0 +1,652 @@ +/* + * Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved. + */ + +package com.openjiuwen.core.singleagent.agents; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +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.foundation.llm.schema.ToolCall; +import com.openjiuwen.core.session.AgentSessionApi; +import com.openjiuwen.core.session.stream.OutputSchema; +import com.openjiuwen.core.session.stream.StreamMode; +import com.openjiuwen.core.singleagent.schema.AgentCard; + +import org.junit.jupiter.api.Test; + +import reactor.test.StepVerifier; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * Tests for ReActAgent streaming degradation removal and retry behavior. + *

+ * Verifies that: + *

    + *
  • Stream failures retry instead of silently degrading to non-stream
    • + *
  • Empty stream responses fall back to non-stream with content wrapped as stream chunks
    • + *
  • Stream exceptions (network errors) return errors without falling back
    • + *
  • Normal streaming behavior is unaffected
    • + *
+ */ +class ReActAgentStreamDegradationTest { + + @Test + void streamRetryOnEmptyStreamThenSucceeds() throws Exception { + ReActAgent agent = newAgent("retry-empty-then-success"); + agent.configure(ReActAgentConfig.builder() + .maxIterations(3) + .streamMaxRetries(1) + .streamRetryDelayMs(0) + .build()); + Model model = mock(Model.class); + // 第一次返回空 Iterator（触发重试），第二次返回正常 chunks + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()) + .thenReturn(List.of( + AssistantMessageChunk.builder().content("recovered content").build() + ).iterator()); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("retry-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> !(item instanceof OutputSchema output + && "answer".equals(output.getType()) + && String.valueOf(output.getPayload()).contains("recovered content"))) + .expectNextMatches(item -> item instanceof OutputSchema output + && "answer".equals(output.getType()) + && String.valueOf(output.getPayload()).contains("recovered content")) + .verifyComplete(); + + // 验证 model.invoke() 从未被调用（不降级） + verify(model, never()).invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + } + + @Test + void streamEmptyResponseFallsBackToNonStreamWithWrapping() throws Exception { + ReActAgent agent = newAgent("empty-fallback"); + Model model = mock(Model.class); + // stream 始终返回空（模拟模型返回非 SSE 格式） + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + // invoke 返回有 content 的响应（作为回退） + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(AssistantMessage.builder().content("non-stream content").build()); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("empty-fallback-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 验证 content 通过 answer 类型发送了一次 + long answerCount = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "answer".equals(output.getType())) + .filter(output -> String.valueOf(output.getPayload()).contains("non-stream content")) + .count(); + assertThat(answerCount).isEqualTo(1); + + // 验证 content 没有通过 llm_output 类型重复发送 + long llmOutputCount = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "llm_output".equals(output.getType())) + .filter(output -> String.valueOf(output.getPayload()).contains("non-stream content")) + .count(); + assertThat(llmOutputCount).isEqualTo(0); + + // 验证 model.invoke() 被调用了一次（作为回退） + verify(model, times(1)) + .invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + } + + @Test + void streamAllRetriesExhaustedWithError() throws Exception { + ReActAgent agent = newAgent("retries-exhausted"); + Model model = mock(Model.class); + // stream 抛异常（异常时不重试，避免部分 chunk 重复发送） + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenThrow(new IllegalStateException("connection timeout")); + + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("exhausted-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()) + && String.valueOf(output.getPayload()).contains("failed")) + .verifyComplete(); + + // 验证 model.invoke() 从未被调用 + verify(model, never()).invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + + // 验证 stream 仅被调用一次（异常时不重试） + verify(model, times(1)) + .stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + } + + @Test + void normalStreamNotAffected() throws Exception { + ReActAgent agent = newAgent("normal-stream"); + Model model = mock(Model.class); + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(List.of( + AssistantMessageChunk.builder().content("Hello ").build(), + AssistantMessageChunk.builder().content("World").build() + ).iterator()); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("normal-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + // 收集所有流式输出 + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 验证流式输出了 content chunks + String allContent = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .map(output -> String.valueOf(output.getPayload())) + .reduce("", (a, b) -> a + b); + assertThat(allContent).contains("Hello"); + assertThat(allContent).contains("World"); + + // 验证 model.invoke() 从未被调用 + verify(model, never()).invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + } + + @Test + void streamWithContentAndToolCallsSendsContentFirst() throws Exception { + ReActAgent agent = newAgent("content-then-tools"); + Model model = mock(Model.class); + // 返回包含 content 和 tool_calls 的单 chunk + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(List.of( + AssistantMessageChunk.builder() + .content("analysis result") + .toolCalls(List.of(ToolCall.builder() + .name("todo_modify") + .arguments("{\"status\": \"completed\"}") + .build())) + .build() + ).iterator()); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("content-tools-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 验证 content 在 tool_calls 之前被发送 + int contentIndex = -1; + int toolCallIndex = -1; + for (int i = 0; i < collected.size(); i++) { + if (collected.get(i) instanceof OutputSchema output) { + String payloadStr = String.valueOf(output.getPayload()); + if (payloadStr.contains("analysis result") && contentIndex == -1) { + contentIndex = i; + } + if (payloadStr.contains("tool_calls") && toolCallIndex == -1) { + toolCallIndex = i; + } + } + } + assertThat(contentIndex).as("content should be streamed").isGreaterThanOrEqualTo(0); + // tool_calls 可能不存在于最终输出中（因为工具执行可能失败），但如果存在，应在 content 之后 + if (toolCallIndex >= 0) { + assertThat(toolCallIndex).as("tool_calls should come after content").isGreaterThan(contentIndex); + } + } + + /** + * 模拟非流式响应（包含 content 和 tool_call），验证 content 被正确包装为流式 SSE 发送。 + * + * 场景：流式请求返回空 → 回退非流式 → 第一次 invoke 返回 content + todo_modify → + * content 通过 llm_output 分块发送 → tool_call 通过 llm_output 发送 → 工具执行后继续 → + * 第二次 invoke 返回最终答案 → 通过 answer 类型发送 + */ + @Test + void nonStreamResponseWithContentAndToolCallWrappedAsStream() throws Exception { + ReActAgent agent = newAgent("non-stream-content-toolcall"); + Model model = mock(Model.class); + + // stream 始终返回空（模拟模型不支持流式） + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + + // 第一次 invoke：返回对比正文 + todo_modify 工具调用 + String comparisonContent = "乔丹 vs 詹姆斯：乔丹6冠6FMVP，詹姆斯4冠4FMVP。梅西 vs C罗：梅西8金球，C罗5金球。"; + AssistantMessage firstResponse = AssistantMessage.builder() + .content(comparisonContent) + .toolCalls(List.of(ToolCall.builder() + .name("todo_modify") + .arguments("{\"task_id\": \"5\", \"status\": \"completed\"}") + .build())) + .build(); + + // 第二次 invoke：返回最终摘要（无工具调用） + AssistantMessage secondResponse = AssistantMessage.builder() + .content("对比分析任务已完成。") + .build(); + + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(firstResponse) + .thenReturn(secondResponse); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("content-toolcall-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "GOAT对比"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 提取所有 OutputSchema + List outputs = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .toList(); + + // 1. 验证对比正文通过 llm_output 类型发送（被包装为流式） + String llmOutputContent = outputs.stream() + .filter(o -> "llm_output".equals(o.getType())) + .map(o -> String.valueOf(o.getPayload())) + .filter(s -> s.contains("乔丹") || s.contains("梅西")) + .reduce("", (a, b) -> a + b); + assertThat(llmOutputContent).as("对比正文应通过 llm_output 流式发送").contains("乔丹"); + assertThat(llmOutputContent).as("对比正文应通过 llm_output 流式发送").contains("梅西"); + + // 2. 验证 todo_modify 工具调用通过 llm_output 类型发送 + String toolCallPayload = outputs.stream() + .filter(o -> "llm_output".equals(o.getType())) + .map(o -> String.valueOf(o.getPayload())) + .filter(s -> s.contains("tool_calls")) + .reduce("", (a, b) -> a + b); + assertThat(toolCallPayload).as("tool_call 应通过 llm_output 发送").contains("tool_calls"); + assertThat(toolCallPayload).as("tool_call 应包含 ToolCall 对象").contains("ToolCall"); + + // 3. 验证 content 在 tool_call 之前发送 + int contentIdx = -1; + int toolCallIdx = -1; + for (int i = 0; i < outputs.size(); i++) { + String payload = String.valueOf(outputs.get(i).getPayload()); + if (contentIdx == -1 && payload.contains("乔丹")) { + contentIdx = i; + } + if (toolCallIdx == -1 && payload.contains("tool_calls") && payload.contains("ToolCall")) { + toolCallIdx = i; + } + } + assertThat(contentIdx).as("对比正文应被发送").isGreaterThanOrEqualTo(0); + assertThat(toolCallIdx).as("tool_call 应被发送").isGreaterThanOrEqualTo(0); + assertThat(toolCallIdx).as("content 必须在 tool_call 之前发送").isGreaterThan(contentIdx); + + // 4. 验证最终答案通过 answer 类型发送 + String answerPayload = outputs.stream() + .filter(o -> "answer".equals(o.getType())) + .map(o -> String.valueOf(o.getPayload())) + .reduce("", (a, b) -> a + b); + assertThat(answerPayload).as("最终答案应通过 answer 发送").contains("对比分析任务已完成"); + + // 5. 验证对比正文没有被 answer 类型重复发送 + boolean answerHasComparison = outputs.stream() + .filter(o -> "answer".equals(o.getType())) + .anyMatch(o -> String.valueOf(o.getPayload()).contains("乔丹")); + assertThat(answerHasComparison).as("对比正文不应通过 answer 重复发送").isFalse(); + + // 6. 验证 model.invoke 被调用两次（第一次 content+tool_call，第二次最终答案） + verify(model, times(2)) + .invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any()); + } + + /** + * 模拟长耗时（100秒）非流式回退场景，验证心跳机制是否在回退期间发送 progress 事件， + * 防止客户端因长时间无响应而超时断开。 + * + * 场景：流式返回空 → 回退非流式 → callModel 阻塞 100 秒 → 响应到达 → 包装为流式发送 + * 验证：在 callModel 前收到了 progress 类型的 SSE 心跳事件 + */ + @Test + void heartbeatSentBeforeLongRunningNonStreamFallback() throws Exception { + ReActAgent agent = newAgent("long-fallback"); + agent.configure(ReActAgentConfig.builder() + .maxIterations(3) + .streamMaxRetries(0) + .streamRetryDelayMs(0) + .build()); + Model model = mock(Model.class); + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + + // 模拟 callModel 长耗时：用 latch 阻塞 invoke，直到测试释放 + java.util.concurrent.CountDownLatch invokeStarted = new java.util.concurrent.CountDownLatch(1); + java.util.concurrent.CountDownLatch invokeProceed = new java.util.concurrent.CountDownLatch(1); + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + invokeStarted.countDown(); + // 阻塞等待测试放行，模拟长耗时 + invokeProceed.await(); + return AssistantMessage.builder().content("delayed content").build(); + }); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("long-fallback-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + // 使用线程安全集合收集结果 + java.util.List collected = java.util.Collections.synchronizedList(new ArrayList<>()); + + // 后台线程消费流输出（streamOutput 会阻塞直到流结束） + java.util.concurrent.CountDownLatch consumerReady = new java.util.concurrent.CountDownLatch(1); + Thread consumerThread = new Thread(() -> { + consumerReady.countDown(); + session.streamOutput(collected::add); + }, "test-consumer"); + consumerThread.setDaemon(true); + consumerThread.start(); + // 确保 consumer 已开始等待 + consumerReady.await(); + + // 触发流式执行（stream 方法会启动后台线程，并返回迭代器——但我们不消费它） + agent.stream(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT)); + + // 等待 callModel 被调用（说明心跳已经发送完毕） + assertThat(invokeStarted.await(10, java.util.concurrent.TimeUnit.SECONDS)) + .as("callModel should be invoked within 10s").isTrue(); + + // 给消费回调一点时间处理已入队的心跳事件 + Thread.sleep(500); + + // 在 callModel 阻塞期间，验证已收到 progress 心跳事件 + long heartbeatCount = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "progress".equals(output.getType())) + .count(); + assertThat(heartbeatCount).as("应在 callModel 前发送至少一个 progress 心跳").isGreaterThanOrEqualTo(1); + + // 验证心跳内容包含非流式切换提示 + boolean hasFallbackMessage = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "progress".equals(output.getType())) + .anyMatch(output -> String.valueOf(output.getPayload()).contains("非流式")); + assertThat(hasFallbackMessage).as("心跳应包含非流式切换提示").isTrue(); + + // 释放 callModel 阻塞，让流完成 + invokeProceed.countDown(); + + // 等待流完成（轮询检查 answer 事件是否到达） + long deadline = System.currentTimeMillis() + 10000; + boolean found = false; + while (System.currentTimeMillis() < deadline) { + found = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "answer".equals(output.getType())) + .anyMatch(output -> String.valueOf(output.getPayload()).contains("delayed content")); + if (found) { + break; + } + Thread.sleep(100); + } + assertThat(found).as("应在 10s 内收到最终 content").isTrue(); + + // 验证最终 content 也被正确发送 + long answerCount = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "answer".equals(output.getType())) + .filter(output -> String.valueOf(output.getPayload()).contains("delayed content")) + .count(); + assertThat(answerCount).as("最终 content 应通过 answer 发送").isEqualTo(1); + } + + /** + * 模拟重试期间的心跳验证：流式返回空 → 心跳 → 重试 → 心跳 → 回退非流式 → 心跳。 + * 验证每次重试等待前和回退前都发送了 progress 心跳。 + */ + @Test + void heartbeatSentDuringRetriesAndFallback() throws Exception { + ReActAgent agent = newAgent("retry-heartbeat"); + agent.configure(ReActAgentConfig.builder() + .maxIterations(3) + .streamMaxRetries(2) + .streamRetryDelayMs(50) // 短延迟，测试快速完成 + .build()); + Model model = mock(Model.class); + // stream 始终返回空（触发重试 + 回退） + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(AssistantMessage.builder().content("final content").build()); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("retry-heartbeat-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 提取所有 progress 心跳事件 + List heartbeats = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(output -> "progress".equals(output.getType())) + .toList(); + + // streamMaxRetries=2，所以首次 + 2 次重试 = 3 次 stream 调用 + // 重试前心跳：attempt 0→1, attempt 1→2，共 2 次重试心跳 + // 回退前心跳：1 次 + // 总计：3 次 progress 事件 + assertThat(heartbeats).as("应有 3 个心跳（2 次重试 + 1 次回退）").hasSize(3); + + // 验证心跳内容 + String firstMsg = String.valueOf(heartbeats.get(0).getPayload()); + assertThat(firstMsg).contains("重试"); + String lastMsg = String.valueOf(heartbeats.get(2).getPayload()); + assertThat(lastMsg).contains("非流式"); + } + + /** + * 模拟长耗时（12秒）非流式回退场景，验证心跳按 HEARTBEAT_INTERVAL_MS(5s) 周期性发送。 + * + * 场景：流式返回空 → 回退非流式 → callModel 阻塞 12 秒 → 响应到达 + * 验证：阻塞期间收到 ≥2 个心跳，且相邻心跳间隔在 [3s, 8s] 范围内（容忍调度抖动） + */ + @Test + void heartbeatSentPeriodicallyDuringLongCallModel() throws Exception { + ReActAgent agent = newAgent("periodic-heartbeat"); + agent.configure(ReActAgentConfig.builder() + .maxIterations(3) + .streamMaxRetries(0) + .streamRetryDelayMs(0) + .build()); + Model model = mock(Model.class); + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + + // 用 latch 阻塞 invoke，模拟 12 秒长耗时 + java.util.concurrent.CountDownLatch invokeStarted = new java.util.concurrent.CountDownLatch(1); + java.util.concurrent.CountDownLatch invokeProceed = new java.util.concurrent.CountDownLatch(1); + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + invokeStarted.countDown(); + invokeProceed.await(); + return AssistantMessage.builder().content("periodic content").build(); + }); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("periodic-heartbeat-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + // 线程安全收集：记录每个 progress 事件的接收时间戳 + java.util.List heartbeatTimestamps = java.util.Collections.synchronizedList(new ArrayList<>()); + java.util.concurrent.CountDownLatch consumerReady = new java.util.concurrent.CountDownLatch(1); + Thread consumerThread = new Thread(() -> { + consumerReady.countDown(); + session.streamOutput(item -> { + if (item instanceof OutputSchema output && "progress".equals(output.getType())) { + // 记录心跳到达的相对时间戳（毫秒） + heartbeatTimestamps.add(System.currentTimeMillis()); + } + }); + }, "test-consumer-periodic"); + consumerThread.setDaemon(true); + consumerThread.start(); + consumerReady.await(); + + // 触发流式执行 + agent.stream(Map.of("query", "hello"), session, List.of(StreamMode.OUTPUT)); + + // 等待 callModel 被调用（说明回退已开始，心跳已启动） + assertThat(invokeStarted.await(10, java.util.concurrent.TimeUnit.SECONDS)) + .as("callModel should be invoked within 10s").isTrue(); + + // 等待 12 秒，让周期性心跳发送 2-3 次（5s 间隔） + Thread.sleep(12000); + + // 释放 callModel 阻塞，让流完成 + invokeProceed.countDown(); + + // 等待流结束 + long deadline = System.currentTimeMillis() + 10000; + boolean found = false; + while (System.currentTimeMillis() < deadline) { + found = heartbeatTimestamps.size() >= 2; + if (found) { + break; + } + Thread.sleep(100); + } + + // 验证至少收到 2 个心跳（12 秒内，5 秒间隔应发 2-3 次） + assertThat(heartbeatTimestamps.size()) + .as("12秒内应收到至少2个心跳，实际收到 " + heartbeatTimestamps.size() + " 个") + .isGreaterThanOrEqualTo(2); + + // 验证相邻心跳间隔在 [3s, 8s] 范围内（5s ± 调度抖动） + for (int i = 1; i < heartbeatTimestamps.size(); i++) { + long intervalMs = heartbeatTimestamps.get(i) - heartbeatTimestamps.get(i - 1); + assertThat(intervalMs) + .as("第 " + i + " 个心跳与上一个的间隔应在 [3000, 8000]ms 范围内，实际 " + intervalMs + "ms") + .isBetween(3000L, 8000L); + } + } + + /** + * 验证回退非流式时，若 AssistantMessage 含 tool_calls 但 content 为 null， + * 不会将字面量 "null" 作为流式 chunk 发送给客户端。 + * + * 场景：LLM 调用工具时 content 通常为 null，toText(null) 返回 "null" 字符串 + * 验证：收集所有 llm_output 类型的输出，确认不包含 "null" 字面量 + */ + @Test + void nullContentWithToolCallsDoesNotSendLiteralNull() throws Exception { + ReActAgent agent = newAgent("null-content-toolcall"); + Model model = mock(Model.class); + + // stream 始终返回空（触发回退非流式） + when(model.stream(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Collections.emptyIterator()); + + // 第一次 invoke：返回 content=null + tool_call（LLM 常见行为） + AssistantMessage firstResponse = AssistantMessage.builder() + .content(null) + .toolCalls(List.of(ToolCall.builder() + .name("todo_modify") + .arguments("{\"task_id\": \"1\", \"status\": \"completed\"}") + .build())) + .build(); + + // 第二次 invoke：返回最终摘要 + AssistantMessage secondResponse = AssistantMessage.builder() + .content("done") + .build(); + + when(model.invoke(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(firstResponse) + .thenReturn(secondResponse); + agent.setLlm(model); + + AgentSessionApi session = + new AgentSessionApi("null-content-session", null, agent.getCard(), List.of(StreamMode.OUTPUT)); + + List collected = new ArrayList<>(); + StepVerifier.create(agent.streamAsync(Map.of("query", "test"), session, List.of(StreamMode.OUTPUT))) + .thenConsumeWhile(item -> { + collected.add(item); + return true; + }) + .verifyComplete(); + + // 提取所有 llm_output 类型的输出文本 + String allLlmOutput = collected.stream() + .filter(item -> item instanceof OutputSchema) + .map(item -> (OutputSchema) item) + .filter(o -> "llm_output".equals(o.getType())) + .map(o -> String.valueOf(o.getPayload())) + .reduce("", (a, b) -> a + b); + + // 核心断言：不包含独立的 "null" 字面量 + assertThat(allLlmOutput).as("content 为 null 时不应发送字面量 null").doesNotContain("null"); + + // tool_call 应正常发送 + assertThat(allLlmOutput).as("tool_call 应正常发送").contains("tool_calls"); + } + + private static ReActAgent newAgent(String id) { + ReActAgent agent = new ReActAgent(AgentCard.builder().id(id).name(id).description(id).build()); + agent.configure(ReActAgentConfig.builder() + .maxIterations(3) + .streamMaxRetries(0) + .streamRetryDelayMs(0) + .build()); + return agent; + } +} diff --git a/src/test/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilderTest.java b/src/test/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilderTest.java index 114d1c94f..107bc90f4 100644 --- a/src/test/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilderTest.java +++ b/src/test/java/com/openjiuwen/core/singleagent/prompts/SystemPromptBuilderTest.java @@ -5,6 +5,8 @@ import org.junit.jupiter.api.Test; +import java.lang.reflect.Field; +import java.util.LinkedHashMap; import java.util.Map; class SystemPromptBuilderTest { @@ -79,4 +81,55 @@ void invalidPromptModeFallsBackToFull() { assertThat(builder.getMode()).isEqualTo("full"); assertThat(builder.build()).isEqualTo("identity\n\nmemory file"); } + + /** + * 模拟 sections map 中混入 null 值（可能由并发修改或反射注入导致）， + * 验证 build() 不会抛出 NPE，而是跳过 null section 继续构建。 + */ + @Test + @SuppressWarnings("unchecked") + void buildSkipsNullSectionWithoutNpe() throws Exception { + SystemPromptBuilder builder = new SystemPromptBuilder("cn", "full"); + builder.addSection(new PromptSection("identity", Map.of("cn", "identity"), 10)); + builder.addSection(new PromptSection("tools", Map.of("cn", "tools"), 20)); + + // 通过反射注入 null 值到 sections map + Field sectionsField = SystemPromptBuilder.class.getDeclaredField("sections"); + sectionsField.setAccessible(true); + Map sections = + (Map) sectionsField.get(builder); + // 使用新的 LinkedHashMap 模拟并发场景下可能出现的 null 值 + Map tampered = new LinkedHashMap(); + tampered.put("identity", new PromptSection("identity", Map.of("cn", "identity"), 10)); + tampered.put("broken", null); // 注入 null + tampered.put("tools", new PromptSection("tools", Map.of("cn", "tools"), 20)); + sectionsField.set(builder, tampered); + + // build() 应跳过 null section，不抛出 NPE + String result = builder.build(); + assertThat(result).isEqualTo("identity\n\ntools"); + } + + /** + * 模拟 minimal 模式下 sections map 中混入 null 值， + * 验证 getSectionsForBuild() 不会抛出 NPE。 + */ + @Test + @SuppressWarnings("unchecked") + void minimalModeSkipsNullSectionWithoutNpe() throws Exception { + SystemPromptBuilder builder = new SystemPromptBuilder("cn", "minimal"); + builder.addSection(new PromptSection("identity", Map.of("cn", "identity"), 10)); + + // 通过反射注入 null 值 + Field sectionsField = SystemPromptBuilder.class.getDeclaredField("sections"); + sectionsField.setAccessible(true); + Map tampered = new LinkedHashMap(); + tampered.put("identity", new PromptSection("identity", Map.of("cn", "identity"), 10)); + tampered.put("broken", null); + sectionsField.set(builder, tampered); + + // minimal 模式下应跳过 null，仅输出 identity + String result = builder.build(); + assertThat(result).isEqualTo("identity"); + } }