opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(agent): prevent ReAct degenerate repetition and harden loop termination

Browse Source
This commit is contained in:
matevip 2026-04-05 22:34:03 +08:00
parent 3de643dca9
commit 99befe0d25
14 changed files with 1048 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java
View File

@ -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)

62
mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
View File

@ -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 结果（用户主动停止，有已累积内容） */

112
mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java Normal file
View File

@ -0,0 +1,112 @@
package vip.mate.agent.graph;
import lombok.extern.slf4j.Slf4j;
/**
* 流式输出重复检测器
* <p>
* 检测 LLM 流式输出中的退化重复模式（degenerate repetition），
* 当检测到内容在滑动窗口内高度重复时返回 true，调用方应截断 LLM 流。
* <p>
* 算法：维护一个滑动窗口缓冲区，每次追加新 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;
}
}

48
mateclaw-server/src/main/java/vip/mate/agent/graph/edge/ReasoningDispatcher.java
View File

@ -8,46 +8,74 @@ import vip.mate.agent.graph.state.MateClawStateAccessor;
import static vip.mate.agent.graph.state.MateClawStateKeys.*;
/**
* 推理路由（4 路分支）
* <p>
* 根据 ReasoningNode 产出的状态决定下一步去向：
* 推理路由（分支优先级）
* <ol>
* <li>迭代超限 → limitExceededNode（最高优先级）</li>
* <li>迭代超限 → limitExceededNode</li>
* <li>可直接回答（含 fatal error 自带的错误文案） → finalAnswerNode</li>
* <li>LLM 调用次数超限 → limitExceededNode（仅拦截继续循环的路径）</li>
* <li>需要工具调用 → actionNode</li>
* <li>需要总结压缩 → summarizingNode</li>
* <li>可直接回答 → finalAnswerNode</li>
* <li>兜底 → finalAnswerNode</li>
* </ol>
* <p>
* 注意：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;
}
}

31
mateclaw-server/src/main/java/vip/mate/agent/graph/node/FinalAnswerNode.java
View File

@ -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");
}
}
}

36
mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java
View File

@ -76,6 +76,13 @@ public class ObservationNode implements NodeAction {
List<String> 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<String> 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;
}
}

159
mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java
View File

@ -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<ToolCallback> 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<ToolCallback> 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<String, Object> 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<Message> 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<Message> 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<Message> 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<Message> compactedMessages = conversationWindowManager.compactForRetry(messages);
if (compactedMessages != null && compactedMessages.size() < messages.size()) {
List<Message> 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<Message> compactedMessages = conversationWindowManager.compactForRetry(messages);
if (compactedMessages != null && compactedMessages.size() < messages.size()) {
List<Message> 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")

8
mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java
View File

@ -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));

4
mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java
View File

@ -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";

114
mateclaw-server/src/test/java/vip/mate/agent/graph/RepetitionDetectorTest.java Normal file
View File

@ -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");
}
}

206
mateclaw-server/src/test/java/vip/mate/agent/graph/edge/ReasoningDispatcherLlmCallCountTest.java Normal file
View File

@ -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");
}
}
}

92
mateclaw-server/src/test/java/vip/mate/agent/graph/node/FinalAnswerNodeTest.java Normal file
View File

@ -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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> result = node.apply(state);
assertEquals("未能生成回答，请重试。", result.get(FINAL_ANSWER));
assertEquals("error_fallback", result.get(FINISH_REASON));
}
}

84
mateclaw-server/src/test/java/vip/mate/agent/graph/node/ObservationNodeDuplicateTest.java Normal file
View File

@ -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<String> observationHistory, int iteration, int maxIter) {
Map<String, Object> 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<String> history = List.of("", "");
OverAllState state = buildState(history, 2, 10);
Map<String, Object> 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<String> history = List.of("");
OverAllState state = buildState(history, 1, 10);
Map<String, Object> result = node.apply(state);
assertNull(result.get(ERROR));
}
@Test
@DisplayName("不同的历史观察不触发重复检测")
void shouldNotTriggerWhenHistoryDiffers() throws Exception {
ObservationNode node = createNode();
List<String> history = List.of("result A", "result B");
OverAllState state = buildState(history, 2, 10);
Map<String, Object> 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<String, Object> result = node.apply(state);
assertNull(result.get(ERROR));
}
}

175
mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java Normal file
View File

@ -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 断言测试。
* <p>
* 验证每条退出路径的输出 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> output = createNode().apply(buildStaleState());
assertControlFlagsCleared(output, "stoppedWithPartial");
assertLlmCallCountWritten(output, "stoppedWithPartial");
assertEquals("stopped", output.get(FINISH_REASON));
assertEquals("部分内容", output.get(FINAL_ANSWER));
}
}