diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 688b44e7..27e7a3c2 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -339,6 +339,7 @@ public class AgentGraphBuilder { .addStrategy(MateClawStateKeys.TOOL_CALLS, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.TOOL_RESULTS, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.TOOL_CALL_COUNT, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.LLM_CALL_COUNT, KeyStrategy.REPLACE) // 控制流 .addStrategy(MateClawStateKeys.FINAL_ANSWER, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.NEEDS_TOOL_CALL, KeyStrategy.REPLACE) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index 19de4fad..d052f1bf 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -7,12 +7,15 @@ import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; import vip.mate.channel.web.ChatStreamTracker; +import reactor.core.Disposable; + import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -236,9 +239,15 @@ public class NodeStreamingChatHelper { AtomicInteger promptTokens = new AtomicInteger(0); AtomicInteger completionTokens = new AtomicInteger(0); + // 重复检测器：检测 LLM 退化输出（如不断重复同一句话） + RepetitionDetector contentRepDetector = new RepetitionDetector(); + RepetitionDetector thinkingRepDetector = new RepetitionDetector(); + // 重复检测触发后设为 true，外层轮询线程据此 dispose 订阅 + AtomicBoolean repetitionTriggered = new AtomicBoolean(false); + CountDownLatch latch = new CountDownLatch(1); - chatModel.stream(prompt) + Disposable subscription = chatModel.stream(prompt) .doOnNext(chatResponse -> { if (chatResponse == null || chatResponse.getResults() == null || chatResponse.getResults().isEmpty()) { return; @@ -247,18 +256,35 @@ public class NodeStreamingChatHelper { AssistantMessage msg = generation.getOutput(); lastAssistantMessage.set(msg); - // 1. 提取 content delta + // 重复已触发 → 跳过一切处理（等外层 dispose） + if (repetitionTriggered.get()) { + return; + } + + // 1. 提取 content delta（含重复检测） String contentDelta = msg.getText(); if (contentDelta != null && !contentDelta.isEmpty()) { + if (contentRepDetector.appendAndCheck(contentDelta)) { + log.warn("[{}] Content repetition detected, will cancel stream " + + "for conversation {}", phase, conversationId); + repetitionTriggered.set(true); + return; + } contentAccum.append(contentDelta); if (broadcast) { broadcastDelta(conversationId, "content_delta", contentDelta); } } - // 2. 提取 thinking delta（从 properties 中的 reasoningContent） + // 2. 提取 thinking delta（含重复检测） String thinkingDelta = extractReasoningContent(msg); if (thinkingDelta != null && !thinkingDelta.isEmpty()) { + if (thinkingRepDetector.appendAndCheck(thinkingDelta)) { + log.warn("[{}] Thinking repetition detected, will cancel stream " + + "for conversation {}", phase, conversationId); + repetitionTriggered.set(true); + return; + } thinkingAccum.append(thinkingDelta); if (broadcast) { broadcastDelta(conversationId, "thinking_delta", thinkingDelta); @@ -287,12 +313,25 @@ public class NodeStreamingChatHelper { latch::countDown ); - // 阻塞等待流完成（节点本身是同步 NodeAction），每 500ms 检查一次停止标志 + // 阻塞等待流完成（节点本身是同步 NodeAction），每 500ms 检查一次停止/重复标志 try { long deadlineMs = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(10); while (!latch.await(500, TimeUnit.MILLISECONDS)) { + // 重复检测触发 → 立即 dispose 上游订阅，停止消耗 tokens + if (repetitionTriggered.get()) { + log.warn("[{}] Repetition detected, disposing upstream subscription " + + "for conversation {}", phase, conversationId); + subscription.dispose(); + if (broadcast) { + broadcastDelta(conversationId, "warning", + buildDeltaJson("检测到模型输出重复，已自动截断")); + } + // dispose 后 latch 可能不会 countDown，直接跳出 + break; + } if (streamTracker.isStopRequested(conversationId)) { - // 不直接抛异常 — 先检查是否已有累积内容，有则返回 partial stopped result + // 用户主动停止 — 也 dispose 上游 + subscription.dispose(); boolean hasContent = !contentAccum.isEmpty() || !thinkingAccum.isEmpty() || !toolCallAccumulators.isEmpty(); if (hasContent) { @@ -309,11 +348,13 @@ public class NodeStreamingChatHelper { throw new CancellationException("Stream stopped by user"); } if (System.currentTimeMillis() > deadlineMs) { + subscription.dispose(); log.warn("[{}] Stream call timed out for conversation {}", phase, conversationId); return buildErrorResult("LLM 调用超时", conversationId, phase); } } } catch (InterruptedException e) { + subscription.dispose(); Thread.currentThread().interrupt(); return buildErrorResult("LLM 调用被中断", conversationId, phase); } @@ -367,9 +408,16 @@ public class NodeStreamingChatHelper { conversationId, phase, errorType); } - // ===== 成功 ===== + // ===== 成功（检查是否因重复被截断） ===== + boolean truncatedByRepetition = repetitionTriggered.get(); + if (truncatedByRepetition) { + log.warn("[{}] LLM output was truncated due to repetition detection for conversation {}", + phase, conversationId); + // warning 已在 dispose 时广播，无需重复 + } return assembleResult(contentAccum, thinkingAccum, toolCallAccumulators, - promptTokens.get(), completionTokens.get(), phase, false, null); + promptTokens.get(), completionTokens.get(), phase, + truncatedByRepetition, truncatedByRepetition ? "output_truncated_repetition" : null); } /** 组装 stopped partial 结果（用户主动停止，有已累积内容） */ diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java new file mode 100644 index 00000000..68d4fd9c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java @@ -0,0 +1,112 @@ +package vip.mate.agent.graph; + +import lombok.extern.slf4j.Slf4j; + +/** + * 流式输出重复检测器 + *

+ * 检测 LLM 流式输出中的退化重复模式（degenerate repetition）， + * 当检测到内容在滑动窗口内高度重复时返回 true，调用方应截断 LLM 流。 + *

+ * 算法：维护一个滑动窗口缓冲区，每次追加新 delta 后， + * 检查窗口尾部是否存在连续重复的 n-gram 模式。 + * + * @author MateClaw Team + */ +@Slf4j +public class RepetitionDetector { + + /** 滑动窗口大小（字符数） */ + private static final int WINDOW_SIZE = 1024; + + /** 最小重复片段长度 */ + private static final int MIN_PATTERN_LEN = 8; + + /** 最大检测的模式长度 */ + private static final int MAX_PATTERN_LEN = 200; + + /** 模式需要连续出现的最小次数才判定为重复 */ + private static final int MIN_REPEATS = 4; + + /** 已累积内容的最小长度才开始检测（避免误判短内容） */ + private static final int MIN_CONTENT_LEN = 200; + + private final StringBuilder buffer = new StringBuilder(); + private boolean repetitionDetected = false; + + /** + * 追加新的 delta 并检测是否存在重复 + * + * @param delta 新增的文本片段 + * @return true 表示检测到退化重复，调用方应截断流 + */ + public boolean appendAndCheck(String delta) { + if (delta == null || delta.isEmpty() || repetitionDetected) { + return repetitionDetected; + } + + buffer.append(delta); + + // 内容太短，不检测 + if (buffer.length() < MIN_CONTENT_LEN) { + return false; + } + + // 保持窗口大小 + if (buffer.length() > WINDOW_SIZE * 2) { + buffer.delete(0, buffer.length() - WINDOW_SIZE); + } + + // 在窗口尾部检测重复模式 + String window = buffer.toString(); + int windowLen = window.length(); + + // 从短模式到长模式扫描 + for (int patternLen = MIN_PATTERN_LEN; + patternLen <= Math.min(MAX_PATTERN_LEN, windowLen / MIN_REPEATS); + patternLen++) { + + // 取窗口末尾的 pattern + String pattern = window.substring(windowLen - patternLen); + + // 向前数这个 pattern 连续出现了几次 + int count = 1; + int pos = windowLen - patternLen * 2; + while (pos >= 0) { + String segment = window.substring(pos, pos + patternLen); + if (segment.equals(pattern)) { + count++; + pos -= patternLen; + } else { + break; + } + } + + if (count >= MIN_REPEATS) { + repetitionDetected = true; + log.warn("[RepetitionDetector] Detected degenerate repetition: " + + "pattern length={}, repeats={}, pattern preview=\"{}\"", + patternLen, count, + pattern.length() > 50 ? pattern.substring(0, 50) + "..." : pattern); + return true; + } + } + + return false; + } + + /** + * 重置检测器状态 + */ + public void reset() { + buffer.setLength(0); + repetitionDetected = false; + } + + /** + * 是否已检测到重复 + */ + public boolean isRepetitionDetected() { + return repetitionDetected; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/edge/ReasoningDispatcher.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/edge/ReasoningDispatcher.java index e963acb7..c01d886c 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/edge/ReasoningDispatcher.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/edge/ReasoningDispatcher.java @@ -8,46 +8,74 @@ import vip.mate.agent.graph.state.MateClawStateAccessor; import static vip.mate.agent.graph.state.MateClawStateKeys.*; /** - * 推理路由（4 路分支） - *

- * 根据 ReasoningNode 产出的状态决定下一步去向： + * 推理路由（分支优先级） *

    - *
  1. 迭代超限 → limitExceededNode（最高优先级）
    2. + *
  2. 迭代超限 → limitExceededNode
    3. + *
  3. 可直接回答（含 fatal error 自带的错误文案） → finalAnswerNode
    4. + *
  4. LLM 调用次数超限 → limitExceededNode（仅拦截继续循环的路径）
    5. *
  5. 需要工具调用 → actionNode
    6. *
  6. 需要总结压缩 → summarizingNode
    7. - *
  7. 可直接回答 → finalAnswerNode
    8. + *
  8. 兜底 → finalAnswerNode
    9. *
+ *

+ * 注意：ReasoningNode 的 fatal error 路径会自行设置 finalAnswer（错误文案）+ + * finishReason(ERROR_FALLBACK)，因此会命中分支 2 直接走 finalAnswerNode， + * 不需要也不应该路由到 LimitExceededNode（后者会再发一次 LLM 调用）。 * * @author MateClaw Team */ @Slf4j public class ReasoningDispatcher implements EdgeAction { + /** LLM 调用次数的安全倍数上限（相对于 maxIterations） */ + private static final int LLM_CALL_MULTIPLIER = 3; + @Override public String apply(OverAllState state) throws Exception { MateClawStateAccessor accessor = new MateClawStateAccessor(state); - // 1. 超限检查优先 + // 1. 迭代超限检查（最高优先级） if (accessor.isLimitReached()) { log.warn("[ReasoningDispatcher] Iteration limit reached ({}/{}), routing to limitExceededNode", accessor.iterationCount(), accessor.maxIterations()); return LIMIT_EXCEEDED_NODE; } - // 2. 工具调用 + // 2. 可直接回答 → finalAnswerNode + // 覆盖以下场景： + // - LLM 正常产出最终回答 (needsToolCall=false, finalAnswer 非空) + // - fatal error (ReasoningNode 设置了 finalAnswer=错误文案 + finishReason=ERROR_FALLBACK) + // - 用户停止无内容 (finalAnswer="" + finishReason=STOPPED) + // 不受 llm_call_count 限制 — 已有结果不应丢弃。 + if (!accessor.needsToolCall() && !accessor.shouldSummarize()) { + log.debug("[ReasoningDispatcher] Routing to finalAnswerNode (direct answer)"); + return FINAL_ANSWER_NODE; + } + + // 3. LLM 调用次数超限 — 仅拦截继续循环（工具调用/总结）的路径 + int llmCallCount = accessor.llmCallCount(); + int llmCallLimit = accessor.maxIterations() * LLM_CALL_MULTIPLIER; + if (llmCallCount >= llmCallLimit) { + log.warn("[ReasoningDispatcher] LLM call count limit reached ({}/{}), " + + "routing to limitExceededNode instead of continuing loop", + llmCallCount, llmCallLimit); + return LIMIT_EXCEEDED_NODE; + } + + // 4. 工具调用 if (accessor.needsToolCall()) { log.debug("[ReasoningDispatcher] Routing to actionNode (tool call needed)"); return ACTION_NODE; } - // 3. 需要总结（上下文过长，最终回答前先压缩） + // 5. 需要总结 if (accessor.shouldSummarize()) { log.info("[ReasoningDispatcher] Routing to summarizingNode (observation context too large)"); return SUMMARIZING_NODE; } - // 4. 直接回答 - log.debug("[ReasoningDispatcher] Routing to finalAnswerNode (direct answer)"); + // 6. 兜底 + log.debug("[ReasoningDispatcher] Routing to finalAnswerNode (fallback)"); return FINAL_ANSWER_NODE; } } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/FinalAnswerNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/FinalAnswerNode.java index 9cb7f52b..7c09fbf3 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/FinalAnswerNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/FinalAnswerNode.java @@ -79,18 +79,27 @@ public class FinalAnswerNode implements NodeAction { finalAnswer.length(), finishReason); } else { - // 异常兜底：使用 summarizedContext - String summary = accessor.summarizedContext(); - if (!summary.isEmpty()) { - finalAnswer = summary; - finalThinking = currentThinking; - finishReason = FinishReason.SUMMARIZED; - log.warn("[FinalAnswerNode] No finalAnswer or draft found, falling back to summarizedContext"); - } else { - finalAnswer = "未能生成回答，请重试。"; + // 无 draft 也无 finalAnswer + // 先检查是否用户主动停止（无内容的 STOPPED 是合法终止，不是错误） + if (parseFinishReason(existingReason) == FinishReason.STOPPED) { + finalAnswer = ""; finalThinking = ""; - finishReason = FinishReason.ERROR_FALLBACK; - log.error("[FinalAnswerNode] No answer source available, returning fallback"); + finishReason = FinishReason.STOPPED; + log.info("[FinalAnswerNode] User stopped before any content was generated, preserving STOPPED"); + } else { + // 异常兜底：使用 summarizedContext + String summary = accessor.summarizedContext(); + if (!summary.isEmpty()) { + finalAnswer = summary; + finalThinking = currentThinking; + finishReason = FinishReason.SUMMARIZED; + log.warn("[FinalAnswerNode] No finalAnswer or draft found, falling back to summarizedContext"); + } else { + finalAnswer = "未能生成回答，请重试。"; + finalThinking = ""; + finishReason = FinishReason.ERROR_FALLBACK; + log.error("[FinalAnswerNode] No answer source available, returning fallback"); + } } } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java index 97c58148..9b439d8c 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java @@ -76,6 +76,13 @@ public class ObservationNode implements NodeAction { List updatedHistory = new ArrayList<>(existingHistory); updatedHistory.add(combinedObservation); + // 检测重复观察：最近 N 条完全相同则强制终止 + boolean duplicateObservation = detectDuplicateObservations(existingHistory, combinedObservation, 3); + if (duplicateObservation) { + log.warn("[ObservationNode] Detected {} consecutive identical observations, " + + "forcing limit exceeded to break loop", 3); + } + // 判断是否需要 summarize boolean shouldSummarize = observationProcessor.needsSummarizing( existingHistory, combinedObservation); @@ -89,11 +96,34 @@ public class ObservationNode implements NodeAction { existingHistory.size(), combinedObservation.length(), newToolCallCount); } - return MateClawStateAccessor.output() + var builder = MateClawStateAccessor.output() .iterationCount(nextIteration) .put(OBSERVATION_HISTORY, updatedHistory) .shouldSummarize(shouldSummarize) - .toolCallCount(newToolCallCount) - .build(); + .toolCallCount(newToolCallCount); + + // 重复观察时标记错误，让 ObservationDispatcher 路由到 limitExceededNode + if (duplicateObservation) { + builder.put(ERROR, "连续 3 次工具调用返回相同结果，已强制终止循环"); + } + + return builder.build(); + } + + /** + * 检测最近 N 条观察是否与当前观察完全相同 + */ + private boolean detectDuplicateObservations(List history, String current, int threshold) { + if (history.size() < threshold - 1 || current == null || current.isEmpty()) { + return false; + } + // 检查 history 的最后 (threshold-1) 条是否都与 current 相同 + int start = history.size() - (threshold - 1); + for (int i = start; i < history.size(); i++) { + if (!current.equals(history.get(i))) { + return false; + } + } + return true; } } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java index fd93f4bd..d948ea21 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java @@ -50,23 +50,36 @@ public class ReasoningNode implements NodeAction { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + /** 单次 LLM 调用的默认最大输出 token 数，防止退化输出无限生成 */ + private static final int DEFAULT_MAX_OUTPUT_TOKENS = 4096; + private final ChatModel chatModel; private final List toolCallbacks; private final String reasoningEffort; private final NodeStreamingChatHelper streamingHelper; private final ConversationWindowManager conversationWindowManager; private final ChatStreamTracker streamTracker; + private final int maxOutputTokens; public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet, String reasoningEffort, NodeStreamingChatHelper streamingHelper, ConversationWindowManager conversationWindowManager, ChatStreamTracker streamTracker) { + this(chatModel, toolSet, reasoningEffort, streamingHelper, conversationWindowManager, + streamTracker, DEFAULT_MAX_OUTPUT_TOKENS); + } + + public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet, String reasoningEffort, + NodeStreamingChatHelper streamingHelper, + ConversationWindowManager conversationWindowManager, + ChatStreamTracker streamTracker, int maxOutputTokens) { this.chatModel = chatModel; this.toolCallbacks = toolSet.callbacks(); this.reasoningEffort = reasoningEffort; this.streamingHelper = streamingHelper; this.conversationWindowManager = conversationWindowManager; this.streamTracker = streamTracker; + this.maxOutputTokens = maxOutputTokens > 0 ? maxOutputTokens : DEFAULT_MAX_OUTPUT_TOKENS; } public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet, String reasoningEffort, @@ -75,25 +88,19 @@ public class ReasoningNode implements NodeAction { this(chatModel, toolSet, reasoningEffort, streamingHelper, conversationWindowManager, null); } - /** - * @deprecated Use {@link #ReasoningNode(ChatModel, AgentToolSet, String, NodeStreamingChatHelper)} instead - */ + /** @deprecated */ @Deprecated public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet, String reasoningEffort) { this(chatModel, toolSet, reasoningEffort, null, null); } - /** - * @deprecated Use {@link #ReasoningNode(ChatModel, AgentToolSet, String, NodeStreamingChatHelper)} instead - */ + /** @deprecated */ @Deprecated public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet) { this(chatModel, toolSet, null, null, null); } - /** - * @deprecated Use {@link #ReasoningNode(ChatModel, AgentToolSet, String, NodeStreamingChatHelper)} instead - */ + /** @deprecated */ @Deprecated public ReasoningNode(ChatModel chatModel, List toolCallbacks) { this.chatModel = chatModel; @@ -102,6 +109,7 @@ public class ReasoningNode implements NodeAction { this.streamingHelper = null; this.conversationWindowManager = null; this.streamTracker = null; + this.maxOutputTokens = DEFAULT_MAX_OUTPUT_TOKENS; } @Override @@ -109,14 +117,14 @@ public class ReasoningNode implements NodeAction { public Map apply(OverAllState state) throws Exception { MateClawStateAccessor accessor = new MateClawStateAccessor(state); - // ======= 取消检查 ======= + // ======= 取消检查（LLM 调用前，尚未计数） ======= String conversationId = accessor.conversationId(); if (streamTracker != null && streamTracker.isStopRequested(conversationId)) { - log.info("[ReasoningNode] Stop requested, aborting LLM call"); + log.info("[ReasoningNode] Stop requested before LLM call, aborting"); throw new CancellationException("Stream stopped by user"); } - // ======= forced_tool_call 检测：审批通过后的重放 ======= + // ======= forced_tool_call 检测：审批通过后的重放（不计入 LLM 调用） ======= String forcedToolCallJson = accessor.forcedToolCall(); if (!forcedToolCallJson.isEmpty()) { try { @@ -124,7 +132,6 @@ public class ReasoningNode implements NodeAction { AssistantMessage.ToolCall toolCall = deserializeToolCall(forcedToolCallJson); - // 构造合成的 AssistantMessage AssistantMessage syntheticMsg = AssistantMessage.builder() .content("") .toolCalls(List.of(toolCall)) @@ -135,10 +142,10 @@ public class ReasoningNode implements NodeAction { .toolCalls(List.of(toolCall)) .messages(List.of((Message) syntheticMsg)) .iterationCount(accessor.iterationCount() + 1) - .forcedToolCall("") // 清空，防止下一轮再触发 + .forcedToolCall("") .currentPhase("forced_replay") - .contentStreamed(true) // 无 content 需要流式推送 - .thinkingStreamed(true) // 无 thinking 需要流式推送 + .contentStreamed(true) + .thinkingStreamed(true) .events(List.of(GraphEventPublisher.phase("forced_replay", Map.of( "toolName", toolCall.name(), "iteration", accessor.iterationCount() + 1)))) @@ -146,18 +153,26 @@ public class ReasoningNode implements NodeAction { } catch (Exception e) { log.error("[ReasoningNode] Failed to deserialize forced_tool_call, falling through to normal LLM: {}", e.getMessage()); - // 不 return，清空 forcedToolCall 后走正常 LLM 流程 } } - // ======= forced_tool_call 检测结束 ======= + // ======= 构建 Prompt ======= String systemPrompt = accessor.systemPrompt(); List messages = accessor.messages(); - // 构建 Prompt，附带工具定义但禁用内部工具执行 + // 消息列表膨胀防护 + final int MAX_LOOP_MESSAGES = 40; + if (messages.size() > MAX_LOOP_MESSAGES) { + log.warn("[ReasoningNode] Messages list too large ({} messages), trimming to {} for conversation {}", + messages.size(), MAX_LOOP_MESSAGES, conversationId); + List trimmed = new ArrayList<>(MAX_LOOP_MESSAGES); + trimmed.addAll(messages.subList(0, Math.min(4, messages.size()))); + trimmed.addAll(messages.subList(messages.size() - (MAX_LOOP_MESSAGES - 4), messages.size())); + messages = trimmed; + } + List promptMessages = new ArrayList<>(); promptMessages.add(new SystemMessage(systemPrompt)); - // 注入运行时上下文（当前时间），让 LLM 在推理阶段即可感知真实日期 promptMessages.add(new UserMessage(RuntimeContextInjector.buildContextMessage())); promptMessages.addAll(messages); @@ -166,6 +181,7 @@ public class ReasoningNode implements NodeAction { OpenAiChatOptions oaiOpts = OpenAiChatOptions.builder() .toolCallbacks(toolCallbacks) .reasoningEffort(reasoningEffort) + .maxTokens(maxOutputTokens) .build(); oaiOpts.setInternalToolExecutionEnabled(false); options = oaiOpts; @@ -173,50 +189,78 @@ public class ReasoningNode implements NodeAction { options = ToolCallingChatOptions.builder() .toolCallbacks(toolCallbacks) .internalToolExecutionEnabled(false) + .maxTokens(maxOutputTokens) .build(); } Prompt prompt = new Prompt(promptMessages, options); - log.debug("[ReasoningNode] Calling LLM with {} messages, {} tool definitions, iteration {}/{}", + // ======= LLM 调用区域 ======= + // nextLlmCallCount 在首次 streamCall 之前计算。 + // 所有退出路径（正常、stopped、fatal error、CancellationException）都必须写回此值。 + // PTL compact retry 会再 +1。 + int nextLlmCallCount = accessor.llmCallCount() + 1; + log.debug("[ReasoningNode] Calling LLM with {} messages, {} tool definitions, iteration {}/{}, llmCallCount={}", promptMessages.size(), toolCallbacks.size(), - accessor.iterationCount(), accessor.maxIterations()); + accessor.iterationCount(), accessor.maxIterations(), nextLlmCallCount); - // 构建 phase 事件 GraphEventPublisher.GraphEvent phaseEvent = GraphEventPublisher.phase("reasoning", Map.of("iteration", accessor.iterationCount())); - // 流式 LLM 调用：content/thinking 增量实时推送给前端 - NodeStreamingChatHelper.StreamResult result = streamingHelper.streamCall( - chatModel, prompt, conversationId, "reasoning"); + NodeStreamingChatHelper.StreamResult result; + try { + result = streamingHelper.streamCall(chatModel, prompt, conversationId, "reasoning"); - // PTL 处理：压缩后重试（由 Node 层负责，因为 helper 不知道哪些消息可压缩） - if (result.isPromptTooLong() && conversationWindowManager != null) { - log.warn("[ReasoningNode] Prompt too long, attempting compaction and retry"); - List compactedMessages = conversationWindowManager.compactForRetry(messages); - if (compactedMessages != null && compactedMessages.size() < messages.size()) { - List retryPromptMessages = new ArrayList<>(); - retryPromptMessages.add(new SystemMessage(systemPrompt)); - retryPromptMessages.addAll(compactedMessages); - Prompt retryPrompt = new Prompt(retryPromptMessages, options); - log.info("[ReasoningNode] Retrying with compacted messages: {} -> {} messages", - messages.size(), compactedMessages.size()); - result = streamingHelper.streamCall(chatModel, retryPrompt, conversationId, "reasoning_compact_retry"); - } else { - log.warn("[ReasoningNode] Compaction did not reduce messages, cannot retry"); + // PTL 处理：压缩后重试 + if (result.isPromptTooLong() && conversationWindowManager != null) { + log.warn("[ReasoningNode] Prompt too long, attempting compaction and retry"); + List compactedMessages = conversationWindowManager.compactForRetry(messages); + if (compactedMessages != null && compactedMessages.size() < messages.size()) { + List retryPromptMessages = new ArrayList<>(); + retryPromptMessages.add(new SystemMessage(systemPrompt)); + retryPromptMessages.add(new UserMessage(RuntimeContextInjector.buildContextMessage())); + retryPromptMessages.addAll(compactedMessages); + Prompt retryPrompt = new Prompt(retryPromptMessages, options); + log.info("[ReasoningNode] Retrying with compacted messages: {} -> {} messages", + messages.size(), compactedMessages.size()); + // compact retry 是第 2 次 LLM 调用，先递增再调用 + nextLlmCallCount++; + result = streamingHelper.streamCall(chatModel, retryPrompt, conversationId, "reasoning_compact_retry"); + } else { + log.warn("[ReasoningNode] Compaction did not reduce messages, cannot retry"); + } } + } catch (CancellationException ce) { + // "调用已发出但尚未产出内容时用户停止" — streamHelper 抛 CancellationException。 + // 返回空 finalAnswer + STOPPED，让 FinalAnswerNode 按 STOPPED 语义处理。 + // 必须显式清零 needsToolCall/shouldSummarize，防止前一轮残留标志导致误路由。 + log.info("[ReasoningNode] CancellationException during LLM call (user stopped before first token), " + + "returning empty answer with STOPPED, llmCallCount={}", nextLlmCallCount); + return MateClawStateAccessor.output() + .finalAnswer("") + .needsToolCall(false) + .shouldSummarize(false) + .llmCallCount(nextLlmCallCount) + .finishReason(FinishReason.STOPPED) + .contentStreamed(true) + .thinkingStreamed(true) + .build(); } - // 用户主动停止且有部分内容：设为 finalAnswer + finalThinking 让 accumulator 持久化 + // ======= 处理 StreamResult ======= + + // 用户主动停止且有部分内容 if (result.stopped() && result.hasAnyContent()) { String partialText = result.text() != null ? result.text() : ""; String partialThinking = result.thinking() != null ? result.thinking() : ""; - log.info("[ReasoningNode] Stop with partial content ({} chars, thinking {} chars), " + - "flushing as final answer", + log.info("[ReasoningNode] Stop with partial content ({} chars, thinking {} chars), flushing as final answer", partialText.length(), partialThinking.length()); var builder = MateClawStateAccessor.output() .finalAnswer(partialText) + .needsToolCall(false) + .shouldSummarize(false) .contentStreamed(true) + .llmCallCount(nextLlmCallCount) .mergeUsage(state, result) .finishReason(FinishReason.STOPPED); if (!partialThinking.isEmpty()) { @@ -226,59 +270,68 @@ public class ReasoningNode implements NodeAction { return builder.build(); } - // 错误处理：无任何可用内容时直接终止图执行。 - // NodeStreamingChatHelper 已广播结构化 error 事件，这里不能再把错误文本当成正常 final answer。 + // Fatal error：直接设置 finalAnswer 为错误文案 + ERROR_FALLBACK， + // 不走 LimitExceededNode（后者会再发一次 LLM 调用，语义不对且对认证/配额错误会再失败）。 + // ReasoningDispatcher 看到 !needsToolCall && !shouldSummarize → finalAnswerNode， + // FinalAnswerNode 检测到 existingAnswer 非空时直接使用，finishReason 保持 ERROR_FALLBACK。 if (result.hasFatalError()) { log.error("[ReasoningNode] Fatal LLM error: {}", result.errorMessage()); - throw new IllegalStateException(result.errorMessage()); + return MateClawStateAccessor.output() + .needsToolCall(false) + .shouldSummarize(false) + .finalAnswer("[错误] " + result.errorMessage()) + .llmCallCount(nextLlmCallCount) + .finishReason(FinishReason.ERROR_FALLBACK) + .contentStreamed(true) + .thinkingStreamed(true) + .mergeUsage(state, result) + .build(); } + if (result.partial()) { - // 有部分内容 — 当作最终回答处理（LLM 已经回答了大部分） log.warn("[ReasoningNode] Partial LLM result ({} chars), treating as final answer", result.text().length()); } if (result.hasToolCalls()) { - // LLM 请求工具调用 log.info("[ReasoningNode] LLM requested {} tool call(s): {}", result.toolCalls().size(), result.toolCalls().stream().map(AssistantMessage.ToolCall::name).toList()); return MateClawStateAccessor.output() .needsToolCall(true) + .shouldSummarize(false) .toolCalls(result.toolCalls()) .messages(List.of((Message) result.assistantMessage())) .currentPhase("reasoning") .currentThinking(result.thinking()) - // 暂存已流式推送的 content/thinking，供 AWAITING_APPROVAL 路径持久化 .streamedContent(result.text() != null ? result.text() : "") .streamedThinking(result.thinking()) .contentStreamed(true) .thinkingStreamed(!result.thinking().isEmpty()) + .llmCallCount(nextLlmCallCount) .mergeUsage(state, result) .events(List.of(phaseEvent)) .build(); } else { - // LLM 给出最终回答 String content = result.text(); log.info("[ReasoningNode] LLM produced final answer ({} chars)", content != null ? content.length() : 0); return MateClawStateAccessor.output() .needsToolCall(false) + .shouldSummarize(false) .finalAnswer(content != null ? content : "") .finalThinking(result.thinking()) .messages(List.of((Message) result.assistantMessage())) .currentPhase("reasoning") .contentStreamed(true) .thinkingStreamed(!result.thinking().isEmpty()) + .llmCallCount(nextLlmCallCount) .mergeUsage(state, result) .events(List.of(phaseEvent)) .build(); } } - /** - * 反序列化 JSON 为 ToolCall - */ private AssistantMessage.ToolCall deserializeToolCall(String json) { try { @SuppressWarnings("unchecked") diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java index 5ba47bf7..777c3a8b 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java @@ -74,6 +74,10 @@ public final class MateClawStateAccessor { return state.value(TOOL_CALL_COUNT, 0); } + public int llmCallCount() { + return state.value(LLM_CALL_COUNT, 0); + } + // ===== 观察历史 ===== @SuppressWarnings("unchecked") @@ -273,6 +277,10 @@ public final class MateClawStateAccessor { return put(TOOL_CALL_COUNT, count); } + public OutputBuilder llmCallCount(int count) { + return put(LLM_CALL_COUNT, count); + } + // ---- 观察 ---- public OutputBuilder observationHistory(String observation) { return put(OBSERVATION_HISTORY, List.of(observation)); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java index ff25567a..edb3267f 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java @@ -127,6 +127,10 @@ public final class MateClawStateKeys { /** 取消标志：外部请求停止时设为 true，各节点在入口处检查 */ public static final String STOP_REQUESTED = "stop_requested"; + // ===== LLM 调用计数（REPLACE 策略）===== + /** 累计 LLM 调用次数（每次 ReasoningNode 调用 LLM 时递增，独立于迭代计数） */ + public static final String LLM_CALL_COUNT = "llm_call_count"; + // ===== Token Usage 累计（REPLACE 策略，节点内累加后写回）===== public static final String PROMPT_TOKENS = "prompt_tokens"; public static final String COMPLETION_TOKENS = "completion_tokens"; diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/RepetitionDetectorTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/RepetitionDetectorTest.java new file mode 100644 index 00000000..7702b74a --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/RepetitionDetectorTest.java @@ -0,0 +1,114 @@ +package vip.mate.agent.graph; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * RepetitionDetector 单元测试 + */ +class RepetitionDetectorTest { + + private RepetitionDetector detector; + + @BeforeEach + void setUp() { + detector = new RepetitionDetector(); + } + + @Test + @DisplayName("正常文本不触发重复检测") + void shouldNotTriggerForNormalText() { + assertFalse(detector.appendAndCheck("Hello, world! This is a normal response. ")); + assertFalse(detector.appendAndCheck("It contains various sentences and ideas. ")); + assertFalse(detector.appendAndCheck("No repetition should be detected here. ")); + assertFalse(detector.appendAndCheck("The detector only flags degenerate patterns. ")); + assertFalse(detector.isRepetitionDetected()); + } + + @Test + @DisplayName("短文本不触发检测（低于最小内容长度）") + void shouldNotTriggerForShortText() { + assertFalse(detector.appendAndCheck("短")); + assertFalse(detector.appendAndCheck("短")); + assertFalse(detector.appendAndCheck("短")); + assertFalse(detector.isRepetitionDetected()); + } + + @Test + @DisplayName("连续重复相同片段触发检测") + void shouldTriggerForRepeatedPattern() { + // 构造足够长的前缀以超过最小检测长度 + StringBuilder sb = new StringBuilder(); + sb.append("这是一段正常的开头文本。".repeat(5)); + detector.appendAndCheck(sb.toString()); + + // 现在重复同一模式多次 + String pattern = "不吃香菜，喝冰美式。"; + boolean triggered = false; + for (int i = 0; i < 20; i++) { + if (detector.appendAndCheck(pattern)) { + triggered = true; + break; + } + } + assertTrue(triggered, "Should detect repetition after many identical appends"); + assertTrue(detector.isRepetitionDetected()); + } + + @Test + @DisplayName("检测到重复后持续返回 true") + void shouldKeepReturningTrueAfterDetection() { + // 直接构造重复内容 + String pattern = "重复片段测试内容。"; + StringBuilder bulk = new StringBuilder(); + bulk.append("正常的前缀内容，长度足够。".repeat(5)); + for (int i = 0; i < 20; i++) { + bulk.append(pattern); + } + detector.appendAndCheck(bulk.toString()); + + // 后续调用应该继续返回 true + assertTrue(detector.appendAndCheck("任何新内容")); + assertTrue(detector.isRepetitionDetected()); + } + + @Test + @DisplayName("reset 后重新检测") + void shouldResetState() { + // 先触发检测 + String pattern = "重复片段测试。"; + StringBuilder bulk = new StringBuilder("前缀".repeat(50)); + for (int i = 0; i < 20; i++) { + bulk.append(pattern); + } + detector.appendAndCheck(bulk.toString()); + + // reset + detector.reset(); + assertFalse(detector.isRepetitionDetected()); + assertFalse(detector.appendAndCheck("正常的新内容")); + } + + @Test + @DisplayName("null 和空字符串不触发也不异常") + void shouldHandleNullAndEmpty() { + assertFalse(detector.appendAndCheck(null)); + assertFalse(detector.appendAndCheck("")); + assertFalse(detector.isRepetitionDetected()); + } + + @Test + @DisplayName("Unicode 中文重复模式正确检测") + void shouldDetectChineseRepetition() { + StringBuilder sb = new StringBuilder("初始化内容填充。".repeat(10)); + String pattern = "已记住。以后涉及点餐时我会提醒你："; + for (int i = 0; i < 20; i++) { + sb.append(pattern); + } + boolean triggered = detector.appendAndCheck(sb.toString()); + assertTrue(triggered, "Should detect Chinese character repetition"); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/edge/ReasoningDispatcherLlmCallCountTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/edge/ReasoningDispatcherLlmCallCountTest.java new file mode 100644 index 00000000..8228ba9d --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/edge/ReasoningDispatcherLlmCallCountTest.java @@ -0,0 +1,206 @@ +package vip.mate.agent.graph.edge; + +import com.alibaba.cloud.ai.graph.OverAllState; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static vip.mate.agent.graph.state.MateClawStateKeys.*; + +/** + * ReasoningDispatcher 扩展测试：LLM 调用计数、fatal error 路由、停止语义 + */ +class ReasoningDispatcherLlmCallCountTest { + + private ReasoningDispatcher dispatcher; + + @BeforeEach + void setUp() { + dispatcher = new ReasoningDispatcher(); + } + + // ===== LLM 调用计数 ===== + + @Nested + @DisplayName("LLM 调用计数限制") + class LlmCallCountLimit { + + @Test + @DisplayName("达到上限但有最终回答 → 放行到 finalAnswerNode") + void shouldAllowFinalAnswerEvenWhenLlmCallLimitReached() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 5, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 30, NEEDS_TOOL_CALL, false + )); + assertEquals(FINAL_ANSWER_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("达到上限且需要工具调用 → limitExceededNode") + void shouldBlockToolCallWhenLlmCallLimitReached() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 5, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 30, NEEDS_TOOL_CALL, true + )); + assertEquals(LIMIT_EXCEEDED_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("达到上限且需要总结 → limitExceededNode") + void shouldBlockSummarizeWhenLlmCallLimitReached() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 5, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 30, + NEEDS_TOOL_CALL, false, SHOULD_SUMMARIZE, true + )); + assertEquals(LIMIT_EXCEEDED_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("未达上限且需要工具调用 → actionNode") + void shouldAllowToolCallWhenUnderLimit() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 5, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 15, NEEDS_TOOL_CALL, true + )); + assertEquals(ACTION_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("计数未设置时默认 0 → 不触发") + void shouldUseDefaultWhenMissing() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 0, MAX_ITERATIONS, 10, + NEEDS_TOOL_CALL, true + )); + assertEquals(ACTION_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("迭代超限优先于 LLM 调用计数") + void iterationLimitTakesPrecedence() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 10, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 5, NEEDS_TOOL_CALL, true + )); + assertEquals(LIMIT_EXCEEDED_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("计数恰好 limit-1 时不触发（边界值）") + void shouldNotTriggerAtLimitMinusOne() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 5, MAX_ITERATIONS, 10, + LLM_CALL_COUNT, 29, NEEDS_TOOL_CALL, true + )); + assertEquals(ACTION_NODE, dispatcher.apply(state)); + } + } + + // ===== Fatal error 路由 ===== + + @Nested + @DisplayName("Fatal error 路由语义") + class FatalErrorRouting { + + @Test + @DisplayName("fatal error 带 finalAnswer 时走 finalAnswerNode（不走 limitExceededNode）") + void fatalErrorWithFinalAnswerGoesToFinalAnswerNode() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 2, MAX_ITERATIONS, 10, + FINAL_ANSWER, "[错误] 认证失败", + NEEDS_TOOL_CALL, false, + SHOULD_SUMMARIZE, false, + FINISH_REASON, "error_fallback" + )); + assertEquals(FINAL_ANSWER_NODE, dispatcher.apply(state), + "Fatal error with finalAnswer should go to finalAnswerNode, not limitExceededNode"); + } + + @Test + @DisplayName("fatal error 叠加旧 SHOULD_SUMMARIZE=true 时仍走 finalAnswerNode（stale-state 防护）") + void fatalErrorWithStaleShouldSummarize() throws Exception { + // 模拟：前一轮 ObservationNode 设了 shouldSummarize=true， + // 但本轮 ReasoningNode fatal error 显式清零了 shouldSummarize=false。 + // 验证不会因为残留标志被送去 summarizingNode / limitExceededNode。 + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 2, MAX_ITERATIONS, 10, + FINAL_ANSWER, "[错误] 模型不可用", + NEEDS_TOOL_CALL, false, + SHOULD_SUMMARIZE, false, // ReasoningNode 显式清零 + FINISH_REASON, "error_fallback" + )); + assertEquals(FINAL_ANSWER_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("如果 SHOULD_SUMMARIZE 未被清零（stale=true），fatal error 会误路由（此为防御性对照测试）") + void demonstrateStaleShouldSummarizeWouldMisroute() throws Exception { + // 对照：如果 shouldSummarize 残留 true，即使有 finalAnswer 也会被送去 summarizingNode + // 这证明 ReasoningNode 必须显式清零 + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 2, MAX_ITERATIONS, 10, + FINAL_ANSWER, "[错误] 模型不可用", + NEEDS_TOOL_CALL, false, + SHOULD_SUMMARIZE, true // 残留！ + )); + // 会走到 llmCallCount check → summarizingNode，不是 finalAnswerNode + assertNotEquals(FINAL_ANSWER_NODE, dispatcher.apply(state), + "Stale shouldSummarize=true would misroute — proves ReasoningNode must clear it"); + } + } + + // ===== 用户停止语义 ===== + + @Nested + @DisplayName("用户停止路由语义") + class StoppedRouting { + + @Test + @DisplayName("用户停止无内容时走 finalAnswerNode（STOPPED 语义不被吞）") + void stoppedWithNoContentGoesToFinalAnswerNode() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 2, MAX_ITERATIONS, 10, + FINAL_ANSWER, "", + NEEDS_TOOL_CALL, false, + SHOULD_SUMMARIZE, false, + FINISH_REASON, "stopped" + )); + assertEquals(FINAL_ANSWER_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("cancel 叠加旧 NEEDS_TOOL_CALL=true 时仍走 finalAnswerNode（stale-state 防护）") + void cancelWithStaleNeedsToolCall() throws Exception { + // 模拟：前一轮 ReasoningNode 设了 needsToolCall=true（请求工具调用）， + // 但本轮 cancel 显式清零了 needsToolCall=false。 + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 3, MAX_ITERATIONS, 10, + FINAL_ANSWER, "", + NEEDS_TOOL_CALL, false, // ReasoningNode 显式清零 + SHOULD_SUMMARIZE, false, // ReasoningNode 显式清零 + FINISH_REASON, "stopped" + )); + assertEquals(FINAL_ANSWER_NODE, dispatcher.apply(state)); + } + + @Test + @DisplayName("如果 NEEDS_TOOL_CALL 未被清零（stale=true），cancel 会误路由（防御性对照测试）") + void demonstrateStaleNeedsToolCallWouldMisroute() throws Exception { + OverAllState state = new OverAllState(Map.of( + CURRENT_ITERATION, 3, MAX_ITERATIONS, 10, + FINAL_ANSWER, "", + NEEDS_TOOL_CALL, true, // 残留！ + SHOULD_SUMMARIZE, false, + FINISH_REASON, "stopped" + )); + // 会走到 actionNode，不是 finalAnswerNode + assertEquals(ACTION_NODE, dispatcher.apply(state), + "Stale needsToolCall=true would misroute to actionNode"); + } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/FinalAnswerNodeTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/FinalAnswerNodeTest.java new file mode 100644 index 00000000..69ad31e3 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/FinalAnswerNodeTest.java @@ -0,0 +1,92 @@ +package vip.mate.agent.graph.node; + +import com.alibaba.cloud.ai.graph.OverAllState; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static vip.mate.agent.graph.state.MateClawStateKeys.*; + +/** + * FinalAnswerNode 单元测试 + */ +class FinalAnswerNodeTest { + + private FinalAnswerNode node; + + @BeforeEach + void setUp() { + node = new FinalAnswerNode(); + } + + @Test + @DisplayName("正常 finalAnswer 直接使用") + void shouldUseExistingFinalAnswer() throws Exception { + OverAllState state = new OverAllState(Map.of( + FINAL_ANSWER, "这是最终回答" + )); + Map result = node.apply(state); + assertEquals("这是最终回答", result.get(FINAL_ANSWER)); + assertEquals("normal", result.get(FINISH_REASON)); + } + + @Test + @DisplayName("finalAnswerDraft 优先于 finalAnswer") + void draftTakesPrecedence() throws Exception { + OverAllState state = new OverAllState(Map.of( + FINAL_ANSWER, "旧回答", + FINAL_ANSWER_DRAFT, "新草稿" + )); + Map result = node.apply(state); + assertEquals("新草稿", result.get(FINAL_ANSWER)); + } + + @Test + @DisplayName("ERROR_FALLBACK finalAnswer 保留错误文案和 finishReason") + void shouldPreserveErrorFallbackAnswer() throws Exception { + OverAllState state = new OverAllState(Map.of( + FINAL_ANSWER, "[错误] 认证失败: Invalid API Key", + FINISH_REASON, "error_fallback" + )); + Map result = node.apply(state); + assertEquals("[错误] 认证失败: Invalid API Key", result.get(FINAL_ANSWER)); + assertEquals("error_fallback", result.get(FINISH_REASON)); + } + + @Test + @DisplayName("STOPPED 且无内容时保留 STOPPED 语义（不降级为 ERROR_FALLBACK）") + void shouldPreserveStoppedWhenNoContent() throws Exception { + OverAllState state = new OverAllState(Map.of( + FINAL_ANSWER, "", + FINISH_REASON, "stopped" + )); + Map result = node.apply(state); + // 关键断言：finishReason 保持 STOPPED，不被改成 ERROR_FALLBACK + assertEquals("stopped", result.get(FINISH_REASON), + "Empty finalAnswer with STOPPED should not become ERROR_FALLBACK"); + // finalAnswer 保持空（用户停止且无内容是合法的） + assertEquals("", result.get(FINAL_ANSWER)); + } + + @Test + @DisplayName("STOPPED 且无 finalAnswer 键时也保留 STOPPED 语义") + void shouldPreserveStoppedWhenNoFinalAnswerKey() throws Exception { + OverAllState state = new OverAllState(Map.of( + FINISH_REASON, "stopped" + )); + Map result = node.apply(state); + assertEquals("stopped", result.get(FINISH_REASON)); + } + + @Test + @DisplayName("无任何内容且非 STOPPED 时降级为 ERROR_FALLBACK") + void shouldFallbackWhenNoContentAndNotStopped() throws Exception { + OverAllState state = new OverAllState(Map.of()); + Map result = node.apply(state); + assertEquals("未能生成回答，请重试。", result.get(FINAL_ANSWER)); + assertEquals("error_fallback", result.get(FINISH_REASON)); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ObservationNodeDuplicateTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ObservationNodeDuplicateTest.java new file mode 100644 index 00000000..12c01a96 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ObservationNodeDuplicateTest.java @@ -0,0 +1,84 @@ +package vip.mate.agent.graph.node; + +import com.alibaba.cloud.ai.graph.OverAllState; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import vip.mate.agent.graph.observation.ObservationProcessor; +import vip.mate.config.GraphObservationProperties; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static vip.mate.agent.graph.state.MateClawStateKeys.*; + +/** + * ObservationNode 重复观察检测单元测试 + */ +class ObservationNodeDuplicateTest { + + private ObservationNode createNode() { + return new ObservationNode(new ObservationProcessor(new GraphObservationProperties())); + } + + /** + * 构建 OverAllState，包含指定的观察历史和工具结果 + */ + private OverAllState buildState(List observationHistory, int iteration, int maxIter) { + Map stateMap = new HashMap<>(); + stateMap.put(CURRENT_ITERATION, iteration); + stateMap.put(MAX_ITERATIONS, maxIter); + stateMap.put(OBSERVATION_HISTORY, new ArrayList<>(observationHistory)); + stateMap.put(TOOL_RESULTS, List.of()); + stateMap.put(TOOL_CALL_COUNT, 0); + return new OverAllState(stateMap); + } + + @Test + @DisplayName("空观察不应触发重复检测（空结果是合法的边界情况）") + void shouldNotTriggerForEmptyObservations() throws Exception { + ObservationNode node = createNode(); + // TOOL_RESULTS 为空时 combinedObservation = ""， + // detectDuplicateObservations 对空字符串返回 false（by design） + List history = List.of("", ""); + OverAllState state = buildState(history, 2, 10); + + Map result = node.apply(state); + assertNull(result.get(ERROR), "Empty observations should not trigger duplicate detection"); + } + + @Test + @DisplayName("观察历史少于阈值时不触发重复检测") + void shouldNotTriggerWhenHistoryTooShort() throws Exception { + ObservationNode node = createNode(); + // 只有 1 条历史，threshold=3 需要至少 2 条匹配 + List history = List.of(""); + OverAllState state = buildState(history, 1, 10); + + Map result = node.apply(state); + assertNull(result.get(ERROR)); + } + + @Test + @DisplayName("不同的历史观察不触发重复检测") + void shouldNotTriggerWhenHistoryDiffers() throws Exception { + ObservationNode node = createNode(); + List history = List.of("result A", "result B"); + OverAllState state = buildState(history, 2, 10); + + Map result = node.apply(state); + assertNull(result.get(ERROR)); + } + + @Test + @DisplayName("空历史不触发重复检测") + void shouldNotTriggerWithEmptyHistory() throws Exception { + ObservationNode node = createNode(); + OverAllState state = buildState(List.of(), 0, 10); + + Map result = node.apply(state); + assertNull(result.get(ERROR)); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java new file mode 100644 index 00000000..e5961f58 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java @@ -0,0 +1,175 @@ +package vip.mate.agent.graph.node; + +import com.alibaba.cloud.ai.graph.OverAllState; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.tool.ToolCallback; +import vip.mate.agent.AgentToolSet; +import vip.mate.agent.graph.NodeStreamingChatHelper; +import vip.mate.channel.web.ChatStreamTracker; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CancellationException; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; +import static vip.mate.agent.graph.state.MateClawStateKeys.*; + +/** + * ReasoningNode 输出 map 断言测试。 + *

+ * 验证每条退出路径的输出 map 都包含 needsToolCall + shouldSummarize 的显式值， + * 防止 stale-state 导致 ReasoningDispatcher 误路由。 + */ +class ReasoningNodeOutputTest { + + private ChatModel chatModel; + private NodeStreamingChatHelper streamingHelper; + private ChatStreamTracker streamTracker; + private AgentToolSet toolSet; + + @BeforeEach + void setUp() { + chatModel = mock(ChatModel.class); + streamingHelper = mock(NodeStreamingChatHelper.class); + streamTracker = mock(ChatStreamTracker.class); + toolSet = mock(AgentToolSet.class); + when(toolSet.callbacks()).thenReturn(List.of()); + when(streamTracker.isStopRequested(anyString())).thenReturn(false); + } + + private ReasoningNode createNode() { + return new ReasoningNode(chatModel, toolSet, null, streamingHelper, null, streamTracker); + } + + /** 构建包含前一轮残留标志的 stale state */ + private OverAllState buildStaleState() { + Map map = new HashMap<>(); + map.put(CONVERSATION_ID, "test-conv"); + map.put(SYSTEM_PROMPT, "you are a helper"); + map.put(USER_MESSAGE, "hello"); + map.put(MESSAGES, List.of()); + map.put(CURRENT_ITERATION, 3); + map.put(MAX_ITERATIONS, 10); + map.put(LLM_CALL_COUNT, 5); + map.put(FORCED_TOOL_CALL, ""); + // 前一轮残留标志 + map.put(NEEDS_TOOL_CALL, true); + map.put(SHOULD_SUMMARIZE, true); + return new OverAllState(map); + } + + // ===== 控制流标志断言辅助 ===== + + private void assertControlFlagsCleared(Map output, String path) { + assertEquals(false, output.get(NEEDS_TOOL_CALL), + path + ": needsToolCall should be explicitly false"); + assertEquals(false, output.get(SHOULD_SUMMARIZE), + path + ": shouldSummarize should be explicitly false"); + } + + private void assertLlmCallCountWritten(Map output, String path) { + assertNotNull(output.get(LLM_CALL_COUNT), + path + ": llmCallCount should be written"); + assertTrue((int) output.get(LLM_CALL_COUNT) > 0, + path + ": llmCallCount should be positive"); + } + + // ===== 正常 final answer ===== + + @Test + @DisplayName("正常 final answer 路径：output 包含 needsToolCall=false + shouldSummarize=false") + void normalFinalAnswer_clearsControlFlags() throws Exception { + NodeStreamingChatHelper.StreamResult result = new NodeStreamingChatHelper.StreamResult( + "回答内容", "", new AssistantMessage("回答内容"), + List.of(), false, 100, 50); + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result); + + Map output = createNode().apply(buildStaleState()); + + assertControlFlagsCleared(output, "normalFinalAnswer"); + assertLlmCallCountWritten(output, "normalFinalAnswer"); + assertEquals("回答内容", output.get(FINAL_ANSWER)); + } + + // ===== 工具调用 ===== + + @Test + @DisplayName("tool call 路径：output 包含 needsToolCall=true + shouldSummarize=false") + void toolCall_clearsShouldSummarize() throws Exception { + AssistantMessage.ToolCall tc = new AssistantMessage.ToolCall("id1", "function", "search", "{}"); + AssistantMessage msg = AssistantMessage.builder().content("").toolCalls(List.of(tc)).build(); + NodeStreamingChatHelper.StreamResult result = new NodeStreamingChatHelper.StreamResult( + "", "", msg, List.of(tc), true, 100, 50); + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result); + + Map output = createNode().apply(buildStaleState()); + + assertEquals(true, output.get(NEEDS_TOOL_CALL)); + assertEquals(false, output.get(SHOULD_SUMMARIZE), + "toolCall: shouldSummarize should be explicitly false"); + assertLlmCallCountWritten(output, "toolCall"); + } + + // ===== Fatal error ===== + + @Test + @DisplayName("fatal error 路径：output 包含 needsToolCall=false + shouldSummarize=false + finalAnswer=错误文案") + void fatalError_clearsControlFlags() throws Exception { + NodeStreamingChatHelper.StreamResult result = new NodeStreamingChatHelper.StreamResult( + "", "", new AssistantMessage(""), + List.of(), false, 0, 0, false, "认证失败: Invalid API Key", + NodeStreamingChatHelper.ErrorType.AUTH_ERROR); + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result); + + Map output = createNode().apply(buildStaleState()); + + assertControlFlagsCleared(output, "fatalError"); + assertLlmCallCountWritten(output, "fatalError"); + String answer = (String) output.get(FINAL_ANSWER); + assertNotNull(answer); + assertTrue(answer.contains("认证失败"), "Fatal error answer should contain error message"); + assertEquals("error_fallback", output.get(FINISH_REASON)); + } + + // ===== CancellationException (no content stop) ===== + + @Test + @DisplayName("CancellationException 路径：output 包含 needsToolCall=false + shouldSummarize=false + STOPPED") + void cancellation_clearsControlFlags() throws Exception { + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())) + .thenThrow(new CancellationException("Stream stopped by user")); + + Map output = createNode().apply(buildStaleState()); + + assertControlFlagsCleared(output, "cancellation"); + assertLlmCallCountWritten(output, "cancellation"); + assertEquals("stopped", output.get(FINISH_REASON)); + assertEquals("", output.get(FINAL_ANSWER)); + } + + // ===== Stopped with partial content ===== + + @Test + @DisplayName("stopped-with-partial 路径：output 包含 needsToolCall=false + shouldSummarize=false + STOPPED") + void stoppedWithPartial_clearsControlFlags() throws Exception { + NodeStreamingChatHelper.StreamResult result = new NodeStreamingChatHelper.StreamResult( + "部分内容", "部分思考", new AssistantMessage("部分内容"), + List.of(), false, 100, 30, true, null, + NodeStreamingChatHelper.ErrorType.NONE, true); + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result); + + Map output = createNode().apply(buildStaleState()); + + assertControlFlagsCleared(output, "stoppedWithPartial"); + assertLlmCallCountWritten(output, "stoppedWithPartial"); + assertEquals("stopped", output.get(FINISH_REASON)); + assertEquals("部分内容", output.get(FINAL_ANSWER)); + } +}