From d3f2a310e852df539c38e19ddb913fd3ca0ecb91 Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 7 Apr 2026 07:39:14 +0800 Subject: [PATCH] feat(agent): context pruning, thinking recovery, channel health monitor --- .../context/ConversationWindowManager.java | 83 ++++++++++- .../agent/graph/NodeStreamingChatHelper.java | 61 ++++++++ .../graph/executor/ToolExecutionExecutor.java | 8 +- .../mate/channel/AbstractChannelAdapter.java | 9 ++ .../mate/channel/ChannelHealthMonitor.java | 134 ++++++++++++++++++ .../java/vip/mate/channel/ChannelManager.java | 7 + 6 files changed, 300 insertions(+), 2 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java index fd95698b..35bdb91f 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java @@ -5,6 +5,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.Message; import org.springframework.ai.chat.messages.SystemMessage; +import org.springframework.ai.chat.messages.ToolResponseMessage; import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; @@ -111,9 +112,38 @@ public class ConversationWindowManager { } int splitPoint = messages.size() - preserveCount; - List oldMessages = messages.subList(0, splitPoint); + List oldMessages = new ArrayList<>(messages.subList(0, splitPoint)); // 可变副本 List recentMessages = messages.subList(splitPoint, messages.size()); + // ═══ Phase 1: Soft Trim — 裁剪工具结果（head+tail），避免不必要的 LLM 摘要 ═══ + int softTrimmed = softTrimToolResults(oldMessages); + if (softTrimmed > 0) { + int afterTrimTokens = TokenEstimator.estimateTokens(oldMessages) + TokenEstimator.estimateTokens(recentMessages); + log.info("[ConversationWindow] Soft trim: {} tool results trimmed, tokens now={}, budget={}", + softTrimmed, afterTrimTokens, historyBudget); + if (afterTrimTokens <= historyBudget) { + // Soft trim 够了，跳过 LLM 摘要 + List result = new ArrayList<>(oldMessages); + result.addAll(recentMessages); + return result; + } + } + + // ═══ Phase 2: Hard Clear — 替换所有旧工具结果为占位符 ═══ + int hardCleared = hardClearToolResults(oldMessages); + if (hardCleared > 0) { + int afterClearTokens = TokenEstimator.estimateTokens(oldMessages) + TokenEstimator.estimateTokens(recentMessages); + log.info("[ConversationWindow] Hard clear: {} tool results replaced with placeholder, tokens now={}, budget={}", + hardCleared, afterClearTokens, historyBudget); + if (afterClearTokens <= historyBudget) { + List result = new ArrayList<>(oldMessages); + result.addAll(recentMessages); + return result; + } + } + + // ═══ Phase 3: LLM 摘要（原有逻辑，仅在 Phase 1+2 不够时执行） ═══ + // 检查缓存 String cacheKey = conversationId + ":" + oldMessages.size(); CachedSummary cached = summaryCache.get(cacheKey); @@ -154,6 +184,57 @@ public class ConversationWindowManager { return result; } + // ==================== 工具结果裁剪 ==================== + + /** + * Soft trim：对工具结果做 head+tail 裁剪（保留首尾各 200 字符）。 + * @return 裁剪的工具结果条数 + */ + private int softTrimToolResults(List messages) { + int trimmed = 0; + for (int i = 0; i < messages.size(); i++) { + if (messages.get(i) instanceof ToolResponseMessage trm) { + List newResponses = new ArrayList<>(); + boolean changed = false; + for (ToolResponseMessage.ToolResponse r : trm.getResponses()) { + String data = r.responseData(); + if (data != null && data.length() > 500) { + String head = data.substring(0, 200); + String tail = data.substring(data.length() - 200); + newResponses.add(new ToolResponseMessage.ToolResponse( + r.id(), r.name(), head + "\n...[trimmed " + data.length() + " chars]...\n" + tail)); + changed = true; + } else { + newResponses.add(r); + } + } + if (changed) { + messages.set(i, ToolResponseMessage.builder().responses(newResponses).build()); + trimmed++; + } + } + } + return trimmed; + } + + /** + * Hard clear：将所有工具结果替换为占位符。 + * @return 替换的工具结果条数 + */ + private int hardClearToolResults(List messages) { + int cleared = 0; + for (int i = 0; i < messages.size(); i++) { + if (messages.get(i) instanceof ToolResponseMessage trm) { + List placeholders = trm.getResponses().stream() + .map(r -> new ToolResponseMessage.ToolResponse(r.id(), r.name(), "[tool result removed]")) + .toList(); + messages.set(i, ToolResponseMessage.builder().responses(placeholders).build()); + cleared++; + } + } + return cleared; + } + /** * 二次裁剪：从前往后移除消息直到 token 预算满足。 * 至少保留最后 2 条消息（最近一轮对话）。 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 4d7a1c37..6fe93d15 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 @@ -2,6 +2,7 @@ package vip.mate.agent.graph; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.messages.Message; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; @@ -139,6 +140,12 @@ public class NodeStreamingChatHelper { || msg.contains("Too Many Requests") || msg.contains("engine_overloaded")) { return ErrorType.RATE_LIMIT; } + // Thinking block errors (Anthropic: old thinking blocks cannot be modified) + if (msg.contains("thinking blocks cannot be modified") + || msg.contains("thinking content is not allowed") + || msg.contains("thinking block")) { + return ErrorType.THINKING_BLOCK_ERROR; + } // Client errors (400 Bad Request — unsupported format, invalid params, etc.) — NOT retryable if (msg.contains("400") || msg.contains("Bad Request") || msg.contains("invalid_request_error") || msg.contains("unsupported")) { @@ -194,6 +201,15 @@ public class NodeStreamingChatHelper { if (lastResult.errorType() == ErrorType.CLIENT_ERROR) { return lastResult; } + // THINKING_BLOCK_ERROR: 剥离旧 thinking 块后单次重试 + if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR && attempt == 0) { + log.warn("[{}] Thinking block error detected, stripping old thinking and retrying once", phase); + prompt = stripThinkingFromPrompt(prompt); + continue; // 重试一次 + } + if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR) { + return lastResult; // 已经重试过了 + } // 成功或不可重试 if (lastResult.errorMessage() == null || lastResult.errorType() == ErrorType.NONE) { return lastResult; @@ -498,6 +514,49 @@ public class NodeStreamingChatHelper { } /** 构建纯错误 StreamResult（无任何内容） */ + /** + * 从 Prompt 中剥离旧 AssistantMessage 的 thinking/reasoningContent metadata。 + * 保留最新一条 AssistantMessage 的 thinking（可能是模型需要的签名）。 + */ + private Prompt stripThinkingFromPrompt(Prompt prompt) { + List messages = prompt.getInstructions(); + // 找最后一个 AssistantMessage + int lastAssistantIdx = -1; + for (int i = messages.size() - 1; i >= 0; i--) { + if (messages.get(i) instanceof AssistantMessage) { + lastAssistantIdx = i; + break; + } + } + List cleaned = new ArrayList<>(); + for (int i = 0; i < messages.size(); i++) { + Message msg = messages.get(i); + if (msg instanceof AssistantMessage am && i != lastAssistantIdx) { + Map meta = am.getMetadata(); + if (meta != null && meta.containsKey("reasoningContent")) { + // 用 builder 重建 AssistantMessage，去掉 reasoningContent + Map cleanMeta = new java.util.HashMap<>(meta); + cleanMeta.remove("reasoningContent"); + AssistantMessage.Builder builder = AssistantMessage.builder() + .content(am.getText()) + .properties(cleanMeta); + if (am.getToolCalls() != null && !am.getToolCalls().isEmpty()) { + builder.toolCalls(am.getToolCalls()); + } + if (am.getMedia() != null && !am.getMedia().isEmpty()) { + builder.media(am.getMedia()); + } + cleaned.add(builder.build()); + continue; + } + } + cleaned.add(msg); + } + log.info("[ThinkingRecovery] Stripped thinking blocks from {} messages, last assistant at index {}", + messages.size(), lastAssistantIdx); + return new Prompt(cleaned, prompt.getOptions()); + } + private StreamResult buildErrorResult(String errorMsg, String conversationId, String phase) { log.error("[{}] Building error result for conversation {}: {}", phase, conversationId, errorMsg); if (streamTracker != null && conversationId != null) { @@ -583,6 +642,8 @@ public class NodeStreamingChatHelper { AUTH_ERROR, /** 客户端错误 (400 Bad Request, 不支持的格式等) — 不应重试 */ CLIENT_ERROR, + /** Thinking 块错误（旧消息中的 thinking block 不可修改）— 可剥离后单次重试 */ + THINKING_BLOCK_ERROR, /** 其他未知错误 */ UNKNOWN } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java index 9da4168a..94a0b08d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java @@ -66,10 +66,16 @@ public class ToolExecutionExecutor { int rawLen = result.length(); // 检测尾部 2000 字符是否含错误模式 String tailRegion = result.substring(Math.max(0, rawLen - 2000)); - double headRatio = ERROR_TAIL_PATTERN.matcher(tailRegion).find() ? 0.2 : 0.4; + boolean errorDetected = ERROR_TAIL_PATTERN.matcher(tailRegion).find(); + double headRatio = errorDetected ? 0.2 : 0.4; + if (errorDetected) { + log.info("[ToolExecutor] Error pattern detected in tail, preserving 80% tail (headRatio=0.2)"); + } int headLen = (int) (maxChars * headRatio); int tailLen = maxChars - headLen - 80; if (tailLen <= 0) tailLen = maxChars / 2; + log.info("[ToolExecutor] Truncated tool result from {} to {} chars (headRatio={})", + rawLen, maxChars, headRatio); return result.substring(0, headLen) + "\n\n... [结果已截断，原始 " + rawLen + " 字符，保留首尾关键片段] ...\n\n" + result.substring(rawLen - tailLen); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java index a144e3eb..7b7f95bb 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java @@ -14,6 +14,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; /** @@ -55,6 +56,10 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter { @Getter protected volatile String lastError; + /** 最近一次收到消息或连接成功的时间（用于健康监控） */ + @Getter + protected final AtomicLong lastEventTimeMs = new AtomicLong(System.currentTimeMillis()); + protected ExponentialBackoff backoff = new ExponentialBackoff(); /** 重连调度器（懒初始化，仅 IM 渠道使用） */ @@ -154,6 +159,7 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter { protected void onReconnectSuccess() { backoff.reset(); connectionState.set(ConnectionState.CONNECTED); + lastEventTimeMs.set(System.currentTimeMillis()); lastError = null; log.info("[{}] Reconnected successfully: {} (backoff reset)", getChannelType(), channelEntity.getName()); @@ -176,6 +182,7 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter { try { doStart(); connectionState.set(ConnectionState.CONNECTED); + lastEventTimeMs.set(System.currentTimeMillis()); lastError = null; backoff.reset(); log.info("[{}] Channel started successfully: {}", getChannelType(), channelEntity.getName()); @@ -219,6 +226,8 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter { @Override public void onMessage(ChannelMessage message) { + lastEventTimeMs.set(System.currentTimeMillis()); + // Bot 前缀过滤 if (!shouldProcess(message)) { log.debug("[{}] Message filtered (bot prefix not matched): {}", getChannelType(), message.getContent()); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java new file mode 100644 index 00000000..595a6a3d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java @@ -0,0 +1,134 @@ +package vip.mate.channel; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Instant; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +/** + * 渠道健康监控 + *

+ * 每 5 分钟检查所有活跃渠道适配器的健康状态： + * - 连接状态为 ERROR 超过 5 分钟 → 触发重启 + * - 连接状态为 CONNECTED 但超过 1 小时无事件 → 标记 stale 并重启 + * - 每渠道每小时最多 10 次重启，cooldown 2 分钟 + * + * @author MateClaw Team + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class ChannelHealthMonitor { + + private final ChannelManager channelManager; + + /** 错误状态超过此时间触发重启（毫秒） */ + private static final long ERROR_THRESHOLD_MS = 5 * 60 * 1000; + + /** 连接正常但无事件超过此时间视为 stale（毫秒） */ + private static final long STALE_THRESHOLD_MS = 60 * 60 * 1000; + + /** 每渠道每小时最大重启次数 */ + private static final int MAX_RESTARTS_PER_HOUR = 10; + + /** 同一渠道两次重启最小间隔（毫秒） */ + private static final long COOLDOWN_MS = 2 * 60 * 1000; + + /** 重启历史记录（channelId → 重启时间列表） */ + private final ConcurrentHashMap> restartHistory = new ConcurrentHashMap<>(); + + @Scheduled(fixedRate = 300_000) // 每 5 分钟 + public void checkHealth() { + Collection adapters = channelManager.getActiveAdapters(); + if (adapters.isEmpty()) { + return; + } + + long now = System.currentTimeMillis(); + int checked = 0; + int restarted = 0; + + for (ChannelAdapter adapter : adapters) { + if (!(adapter instanceof AbstractChannelAdapter aca)) { + continue; + } + checked++; + + Long channelId = aca.channelEntity.getId(); + AbstractChannelAdapter.ConnectionState state = aca.getConnectionState().get(); + long lastEvent = aca.getLastEventTimeMs().get(); + long sinceLastEvent = now - lastEvent; + + String reason = null; + + // 检查 1：ERROR 状态超过阈值 + if (state == AbstractChannelAdapter.ConnectionState.ERROR && sinceLastEvent > ERROR_THRESHOLD_MS) { + reason = String.format("ERROR state for %ds", sinceLastEvent / 1000); + } + + // 检查 2：CONNECTED 但长时间无事件（stale） + if (reason == null && state == AbstractChannelAdapter.ConnectionState.CONNECTED + && sinceLastEvent > STALE_THRESHOLD_MS) { + reason = String.format("stale connection, no events for %dm", sinceLastEvent / 60000); + } + + if (reason != null) { + if (canRestart(channelId, now)) { + log.warn("[ChannelHealth] Restarting channel {} ({}): {}", + channelId, aca.getDisplayName(), reason); + try { + channelManager.restartChannel(channelId); + recordRestart(channelId, now); + restarted++; + } catch (Exception e) { + log.error("[ChannelHealth] Failed to restart channel {}: {}", + channelId, e.getMessage()); + } + } else { + log.warn("[ChannelHealth] Channel {} ({}) unhealthy ({}), but restart rate-limited", + channelId, aca.getDisplayName(), reason); + } + } + } + + if (restarted > 0) { + log.info("[ChannelHealth] Check completed: {}/{} channels checked, {} restarted", + checked, adapters.size(), restarted); + } + } + + /** + * 检查是否允许重启（限流 + cooldown） + */ + private boolean canRestart(Long channelId, long nowMs) { + List history = restartHistory.computeIfAbsent(channelId, k -> new ArrayList<>()); + + // 清理 1 小时前的记录 + Instant oneHourAgo = Instant.ofEpochMilli(nowMs - 3600_000); + history.removeIf(t -> t.isBefore(oneHourAgo)); + + // 限流检查 + if (history.size() >= MAX_RESTARTS_PER_HOUR) { + return false; + } + + // cooldown 检查 + if (!history.isEmpty()) { + Instant lastRestart = history.get(history.size() - 1); + if (nowMs - lastRestart.toEpochMilli() < COOLDOWN_MS) { + return false; + } + } + + return true; + } + + private void recordRestart(Long channelId, long nowMs) { + restartHistory.computeIfAbsent(channelId, k -> new ArrayList<>()) + .add(Instant.ofEpochMilli(nowMs)); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java index 5e305101..67cffae1 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java @@ -275,6 +275,13 @@ public class ChannelManager { info.put("connectionState", aca.getConnectionState().get().name()); info.put("lastError", aca.getLastError()); info.put("reconnectAttempts", aca.backoff.getAttempts()); + long lastEventMs = aca.getLastEventTimeMs().get(); + info.put("lastEventTime", lastEventMs > 0 + ? java.time.Instant.ofEpochMilli(lastEventMs).toString() : null); + long silentMs = System.currentTimeMillis() - lastEventMs; + info.put("healthStatus", silentMs > 3600_000 ? "stale" + : aca.getConnectionState().get() == AbstractChannelAdapter.ConnectionState.ERROR ? "error" + : "healthy"); } else { info.put("connectionState", adapter.isRunning() ? "CONNECTED" : "DISCONNECTED"); info.put("lastError", null);