mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(agent): smart truncation, stale stream cleanup, configurable tool timeouts, and new indexes
This commit is contained in:
matevip
parent
e6be23a040
commit
5fc60ec513
@ -110,6 +110,7 @@ public class AgentGraphBuilder {
|
||||
private final ObjectProvider<WebClient.Builder> webClientBuilderProvider;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final GraphObservationProperties graphObservationProperties;
|
||||
private final vip.mate.config.ToolTimeoutProperties toolTimeoutProperties;
|
||||
private final WorkspaceFileService workspaceFileService;
|
||||
private final vip.mate.agent.context.ConversationWindowManager conversationWindowManager;
|
||||
|
||||
@ -221,7 +222,7 @@ public class AgentGraphBuilder {
|
||||
try {
|
||||
ChatModel fallbackModel = buildFallbackModel(chatModel);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties);
|
||||
PlanGenerationNode planGenerationNode = new PlanGenerationNode(chatModel, planningService, streamingHelper, conversationWindowManager);
|
||||
StepExecutionNode stepExecutionNode = new StepExecutionNode(chatModel, toolSet, executor, planningService, streamTracker, reasoningEffort, streamingHelper, conversationWindowManager);
|
||||
PlanSummaryNode planSummaryNode = new PlanSummaryNode(chatModel, planningService, streamingHelper);
|
||||
@ -315,7 +316,7 @@ public class AgentGraphBuilder {
|
||||
try {
|
||||
ChatModel fallbackModel = buildFallbackModel(chatModel);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties);
|
||||
ReasoningNode reasoningNode = new ReasoningNode(chatModel, toolSet, reasoningEffort, streamingHelper, conversationWindowManager, streamTracker);
|
||||
ActionNode actionNode = new ActionNode(executor, streamTracker);
|
||||
ObservationProcessor observationProcessor = new ObservationProcessor(graphObservationProperties);
|
||||
|
||||
@ -53,19 +53,44 @@ public class ToolExecutionExecutor {
|
||||
/** 工具结果最大字符数（防止超长结果膨胀 ToolResponseMessage → 撑爆 LLM 上下文） */
|
||||
private static final int MAX_TOOL_RESULT_CHARS = 8000;
|
||||
|
||||
/** 尾部错误模式检测 */
|
||||
private static final java.util.regex.Pattern ERROR_TAIL_PATTERN = java.util.regex.Pattern.compile(
|
||||
"(?i)\\b(error|exception|traceback|failed|fatal|panic|stack.?trace|errno)\\b");
|
||||
|
||||
/**
|
||||
* 智能截断工具结果：检测尾部是否含错误信息，动态调整 head/tail 比例。
|
||||
* 错误信息在尾部时保留 80% tail，确保 agent 能看到错误原因。
|
||||
*/
|
||||
static String truncateToolResult(String result, int maxChars) {
|
||||
if (result == null || result.length() <= maxChars) return result;
|
||||
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;
|
||||
int headLen = (int) (maxChars * headRatio);
|
||||
int tailLen = maxChars - headLen - 80;
|
||||
if (tailLen <= 0) tailLen = maxChars / 2;
|
||||
return result.substring(0, headLen)
|
||||
+ "\n\n... [结果已截断，原始 " + rawLen + " 字符，保留首尾关键片段] ...\n\n"
|
||||
+ result.substring(rawLen - tailLen);
|
||||
}
|
||||
|
||||
private final Map<String, ToolCallback> toolCallbackMap;
|
||||
private final ToolGuardService toolGuardService;
|
||||
private final ToolGuard toolGuard; // legacy fallback
|
||||
private final ApprovalWorkflowService approvalService;
|
||||
private final ChatStreamTracker streamTracker;
|
||||
private final vip.mate.config.ToolTimeoutProperties toolTimeoutProperties;
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker) {
|
||||
this.toolCallbackMap = toolSet.callbackByName();
|
||||
this.toolGuardService = toolGuardService;
|
||||
this.toolGuard = null;
|
||||
this.approvalService = approvalService;
|
||||
this.streamTracker = streamTracker;
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, null);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties) {
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, toolTimeoutProperties);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuard toolGuard,
|
||||
@ -75,6 +100,26 @@ public class ToolExecutionExecutor {
|
||||
this.toolGuard = toolGuard;
|
||||
this.approvalService = approvalService;
|
||||
this.streamTracker = streamTracker;
|
||||
this.toolTimeoutProperties = null;
|
||||
}
|
||||
|
||||
private ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ToolGuard toolGuard, ApprovalWorkflowService approvalService,
|
||||
ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties) {
|
||||
this.toolCallbackMap = toolSet.callbackByName();
|
||||
this.toolGuardService = toolGuardService;
|
||||
this.toolGuard = toolGuard;
|
||||
this.approvalService = approvalService;
|
||||
this.streamTracker = streamTracker;
|
||||
this.toolTimeoutProperties = toolTimeoutProperties;
|
||||
}
|
||||
|
||||
private long getToolTimeoutMs(String toolName) {
|
||||
if (toolTimeoutProperties != null) {
|
||||
return toolTimeoutProperties.getTimeoutSeconds(toolName) * 1000L;
|
||||
}
|
||||
return 5 * 60 * 1000L; // default 5 min
|
||||
}
|
||||
|
||||
/**
|
||||
@ -211,14 +256,9 @@ public class ToolExecutionExecutor {
|
||||
log.info("[ToolExecutor] Executing pre-approved tool: {}", toolName);
|
||||
String result = callback.call(callArguments);
|
||||
int rawLen = result != null ? result.length() : 0;
|
||||
if (result != null && result.length() > MAX_TOOL_RESULT_CHARS) {
|
||||
int headLen = (int) (MAX_TOOL_RESULT_CHARS * 0.4);
|
||||
int tailLen = MAX_TOOL_RESULT_CHARS - headLen - 80;
|
||||
result = result.substring(0, headLen)
|
||||
+ "\n\n... [结果已截断，原始 " + rawLen + " 字符] ...\n\n"
|
||||
+ result.substring(rawLen - tailLen);
|
||||
}
|
||||
log.info("[ToolExecutor] Pre-approved tool {} returned {} chars", toolName, rawLen);
|
||||
result = truncateToolResult(result, MAX_TOOL_RESULT_CHARS);
|
||||
log.info("[ToolExecutor] Pre-approved tool {} returned {} chars{}", toolName, rawLen,
|
||||
result != null && result.length() < rawLen ? " (truncated to " + result.length() + ")" : "");
|
||||
events.add(GraphEventPublisher.toolComplete(toolName, result, true));
|
||||
return new ToolResponseMessage.ToolResponse(
|
||||
toolCall.id(), toolName, result != null ? result : "");
|
||||
@ -305,7 +345,11 @@ public class ToolExecutionExecutor {
|
||||
// 等待所有并行工具完成，按原始顺序填入结果
|
||||
for (var entry : futures.entrySet()) {
|
||||
try {
|
||||
ToolResponseMessage.ToolResponse response = entry.getValue().get(5, TimeUnit.MINUTES);
|
||||
// 按工具名查找配置的超时时间
|
||||
PreparedToolCall matchedPc = batch.stream()
|
||||
.filter(p -> p.resultIndex == entry.getKey()).findFirst().orElse(null);
|
||||
long timeoutMs = getToolTimeoutMs(matchedPc != null ? matchedPc.toolCall.name() : null);
|
||||
ToolResponseMessage.ToolResponse response = entry.getValue().get(timeoutMs, TimeUnit.MILLISECONDS);
|
||||
allResponses.set(entry.getKey(), response);
|
||||
} catch (Exception e) {
|
||||
// 超时或异常 — 填入错误响应
|
||||
@ -338,18 +382,9 @@ public class ToolExecutionExecutor {
|
||||
? pc.arguments.substring(0, 200) + "..." : pc.arguments);
|
||||
String result = pc.callback.call(pc.arguments);
|
||||
int rawLen = result != null ? result.length() : 0;
|
||||
// 截断过长结果，防止 ToolResponseMessage 撑爆 LLM 上下文
|
||||
if (result != null && result.length() > MAX_TOOL_RESULT_CHARS) {
|
||||
int headLen = (int) (MAX_TOOL_RESULT_CHARS * 0.4);
|
||||
int tailLen = MAX_TOOL_RESULT_CHARS - headLen - 80;
|
||||
result = result.substring(0, headLen)
|
||||
+ "\n\n... [结果已截断，原始 " + rawLen + " 字符，保留首尾关键片段] ...\n\n"
|
||||
+ result.substring(rawLen - tailLen);
|
||||
log.info("[ToolExecutor] Tool {} returned {} chars, truncated to {} chars",
|
||||
toolName, rawLen, result.length());
|
||||
} else {
|
||||
log.info("[ToolExecutor] Tool {} returned {} chars", toolName, rawLen);
|
||||
}
|
||||
result = truncateToolResult(result, MAX_TOOL_RESULT_CHARS);
|
||||
log.info("[ToolExecutor] Tool {} returned {} chars{}", toolName, rawLen,
|
||||
result != null && result.length() < rawLen ? " (truncated to " + result.length() + ")" : "");
|
||||
events.add(GraphEventPublisher.toolComplete(toolName, result, true));
|
||||
if (streamTracker != null) {
|
||||
streamTracker.broadcastObject(pc.conversationId, GraphEventPublisher.EVENT_TOOL_COMPLETE,
|
||||
|
||||
@ -4,6 +4,7 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import vip.mate.config.GraphObservationProperties;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/**
|
||||
* 观察结果处理器
|
||||
@ -46,10 +47,16 @@ public class ObservationProcessor {
|
||||
return String.format("[%s] %s", toolName, trimmed);
|
||||
}
|
||||
|
||||
/** 用于检测尾部是否包含错误信息 */
|
||||
private static final Pattern ERROR_TAIL_PATTERN = Pattern.compile(
|
||||
"(?i)\\b(error|exception|traceback|failed|fatal|panic|stack.?trace|errno)\\b");
|
||||
|
||||
/**
|
||||
* 截断大文本，保留首尾关键片段
|
||||
* 截断大文本，保留首尾关键片段。
|
||||
* <p>
|
||||
* 保留前 40% 和后 60% 扣除标记长度后的内容。
|
||||
* 如果尾部 2000 字符内检测到错误模式（error, exception, traceback 等），
|
||||
* 自动提升 tail 保留比例（默认从 0.6 → 0.8），确保错误信息不被截掉。
|
||||
* 同时保证截断后至少保留 minKeepChars 字符。
|
||||
*
|
||||
* @param text 原始文本
|
||||
* @param maxLen 最大允许长度
|
||||
@ -60,6 +67,14 @@ public class ObservationProcessor {
|
||||
return text;
|
||||
}
|
||||
|
||||
// 最少保留保证
|
||||
if (maxLen < properties.getMinKeepChars()) {
|
||||
maxLen = properties.getMinKeepChars();
|
||||
if (text.length() <= maxLen) {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
|
||||
int originalLen = text.length();
|
||||
String marker = String.format(properties.getTruncationMarker(), originalLen);
|
||||
int available = maxLen - marker.length();
|
||||
@ -67,13 +82,23 @@ public class ObservationProcessor {
|
||||
return text.substring(0, maxLen);
|
||||
}
|
||||
|
||||
int headLen = (int) (available * properties.getHeadRatio());
|
||||
// 检测尾部是否含错误信息 → 动态调整 head/tail 比例
|
||||
double effectiveHeadRatio = properties.getHeadRatio();
|
||||
String tailRegion = text.substring(Math.max(0, originalLen - 2000));
|
||||
if (ERROR_TAIL_PATTERN.matcher(tailRegion).find()) {
|
||||
effectiveHeadRatio = 1.0 - properties.getErrorTailRatio(); // 0.2（保留 80% 给 tail）
|
||||
log.info("[Observation] Error pattern detected in tail, preserving tail (ratio={})",
|
||||
properties.getErrorTailRatio());
|
||||
}
|
||||
|
||||
int headLen = (int) (available * effectiveHeadRatio);
|
||||
int tailLen = available - headLen;
|
||||
|
||||
String head = text.substring(0, headLen);
|
||||
String tail = text.substring(originalLen - tailLen);
|
||||
|
||||
log.info("[Observation] Truncated from {} to {} chars (limit={})", originalLen, head.length() + tail.length(), maxLen);
|
||||
log.info("[Observation] Truncated from {} to {} chars (limit={}, headRatio={})",
|
||||
originalLen, head.length() + tail.length(), maxLen, effectiveHeadRatio);
|
||||
return head + marker + tail;
|
||||
}
|
||||
|
||||
|
||||
@ -91,6 +91,9 @@ public class ChatStreamTracker {
|
||||
/** 已广播的 pending approval ID 集合（用于幂等去重） */
|
||||
final java.util.Set<String> broadcastedApprovalIds = java.util.concurrent.ConcurrentHashMap.newKeySet();
|
||||
|
||||
/** 创建时间（用于 stale 检测和清理） */
|
||||
final long createdAt = System.currentTimeMillis();
|
||||
|
||||
RunState(String conversationId) {
|
||||
this.conversationId = conversationId;
|
||||
}
|
||||
@ -739,4 +742,58 @@ public class ChatStreamTracker {
|
||||
sb.append("\"}");
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
// ==================== Stale RunState 清理 ====================
|
||||
|
||||
/** 已完成的 RunState 保留时间（5 分钟） */
|
||||
private static final long DONE_RETENTION_MS = 5 * 60 * 1000;
|
||||
/** RunState 最大存活时间（30 分钟，防止挂起的流永远占内存） */
|
||||
private static final long MAX_LIFETIME_MS = 30 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* 定期清理过期的 RunState，防止内存泄漏。
|
||||
* - 已完成超过 5 分钟的 → 移除
|
||||
* - 存活超过 30 分钟的（无论是否完成）→ 强制移除
|
||||
*/
|
||||
@org.springframework.scheduling.annotation.Scheduled(fixedRate = 600_000)
|
||||
public void cleanupStaleRuns() {
|
||||
long now = System.currentTimeMillis();
|
||||
int evicted = 0;
|
||||
|
||||
var iterator = runs.entrySet().iterator();
|
||||
while (iterator.hasNext()) {
|
||||
var entry = iterator.next();
|
||||
RunState state = entry.getValue();
|
||||
long age = now - state.createdAt;
|
||||
|
||||
boolean shouldEvict = false;
|
||||
String reason = null;
|
||||
|
||||
if (state.done && age > DONE_RETENTION_MS) {
|
||||
shouldEvict = true;
|
||||
reason = "completed and expired";
|
||||
} else if (age > MAX_LIFETIME_MS) {
|
||||
shouldEvict = true;
|
||||
reason = "exceeded max lifetime (" + (age / 1000) + "s)";
|
||||
}
|
||||
|
||||
if (shouldEvict) {
|
||||
// 先清理资源再移除
|
||||
stopHeartbeat(entry.getKey());
|
||||
Disposable d = state.disposable;
|
||||
if (d != null && !d.isDisposed()) {
|
||||
d.dispose();
|
||||
}
|
||||
iterator.remove();
|
||||
evicted++;
|
||||
log.warn("[SSE] Evicted stale RunState for conversation={}: {}",
|
||||
entry.getKey(), reason);
|
||||
}
|
||||
}
|
||||
|
||||
if (evicted > 0) {
|
||||
log.info("[SSE] Cleanup completed: evicted {} stale RunState entries, {} remaining",
|
||||
evicted, runs.size());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -29,4 +29,10 @@ public class GraphObservationProperties {
|
||||
|
||||
/** 截断省略标记（%d 会被替换为原始字符数） */
|
||||
private String truncationMarker = "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n";
|
||||
|
||||
/** 检测到尾部错误模式时的 tail 保留比例（默认 0.8，优先保留错误信息） */
|
||||
private double errorTailRatio = 0.8;
|
||||
|
||||
/** 截断时最少保留字符数（避免过度截断导致信息完全丢失） */
|
||||
private int minKeepChars = 2000;
|
||||
}
|
||||
|
||||
@ -0,0 +1,64 @@
|
||||
package vip.mate.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 工具执行超时配置
|
||||
* <p>
|
||||
* 支持三级配置：per-tool → per-category → default。
|
||||
* 查找优先级：先精确匹配工具名，再匹配类别，最后用默认值。
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Data
|
||||
@ConfigurationProperties(prefix = "mate.agent.tool.timeout")
|
||||
public class ToolTimeoutProperties {
|
||||
|
||||
/** 默认超时（秒） */
|
||||
private int defaultTimeoutSeconds = 300;
|
||||
|
||||
/** 按工具类别的超时（秒）。key: shell/web/mcp/file */
|
||||
private Map<String, Integer> perCategory = new HashMap<>();
|
||||
|
||||
/** 按工具名的超时（秒）。key: 工具名（如 web_fetch） */
|
||||
private Map<String, Integer> perTool = new HashMap<>();
|
||||
|
||||
// 内置类别映射
|
||||
private static final Map<String, String> TOOL_CATEGORY_MAP = Map.of(
|
||||
"execute_bash", "shell",
|
||||
"run_command", "shell",
|
||||
"web_fetch", "web",
|
||||
"url_fetch", "web",
|
||||
"write_file", "file",
|
||||
"edit_file", "file",
|
||||
"read_file", "file"
|
||||
);
|
||||
|
||||
/**
|
||||
* 获取指定工具的超时时间（秒）
|
||||
* 查找顺序：per-tool → per-category → default
|
||||
*/
|
||||
public int getTimeoutSeconds(String toolName) {
|
||||
// 1. 精确匹配工具名
|
||||
if (toolName != null && perTool.containsKey(toolName)) {
|
||||
return perTool.get(toolName);
|
||||
}
|
||||
// 2. 匹配类别
|
||||
if (toolName != null) {
|
||||
String category = TOOL_CATEGORY_MAP.get(toolName);
|
||||
// MCP 工具通常以 mcp_ 开头
|
||||
if (category == null && toolName.startsWith("mcp_")) {
|
||||
category = "mcp";
|
||||
}
|
||||
if (category != null && perCategory.containsKey(category)) {
|
||||
return perCategory.get(category);
|
||||
}
|
||||
}
|
||||
// 3. 默认值
|
||||
return defaultTimeoutSeconds;
|
||||
}
|
||||
}
|
||||
@ -11,7 +11,7 @@ import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({GraphObservationProperties.class, ConversationWindowProperties.class})
|
||||
@EnableConfigurationProperties({GraphObservationProperties.class, ConversationWindowProperties.class, ToolTimeoutProperties.class})
|
||||
public class WebMvcConfig implements WebMvcConfigurer {
|
||||
|
||||
@Override
|
||||
|
||||
@ -99,6 +99,12 @@ mate:
|
||||
min-rounds-for-summarize: 3
|
||||
head-ratio: 0.4
|
||||
truncation-marker: "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n"
|
||||
tool:
|
||||
timeout:
|
||||
default-timeout-seconds: 300
|
||||
per-category:
|
||||
shell: 120
|
||||
web: 30
|
||||
conversation:
|
||||
window:
|
||||
# 测试时临时调低：2000 token ≈ 2000 中文字，3 轮对话即可触发压缩
|
||||
|
||||
@ -431,3 +431,8 @@ CREATE TABLE IF NOT EXISTS mate_memory_recall (
|
||||
CREATE INDEX IF NOT EXISTS idx_memory_recall_agent ON mate_memory_recall(agent_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_memory_recall_agent_file ON mate_memory_recall(agent_id, filename);
|
||||
CREATE INDEX IF NOT EXISTS idx_memory_recall_score ON mate_memory_recall(agent_id, score);
|
||||
CREATE INDEX IF NOT EXISTS idx_memory_recall_candidates ON mate_memory_recall(agent_id, promoted, deleted);
|
||||
|
||||
-- 补充复合索引（高频查询优化）
|
||||
CREATE INDEX IF NOT EXISTS idx_message_conv_time ON mate_message(conversation_id, create_time);
|
||||
CREATE INDEX IF NOT EXISTS idx_workspace_file_agent_enabled ON mate_workspace_file(agent_id, enabled);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user