mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(agent): runtime efficiency — spill oversized tool results, add tool concurrency registry, collect cache metrics
This commit is contained in:
matevip
parent
762218cd32
commit
40bbde1278
@ -126,6 +126,9 @@ public class AgentGraphBuilder {
|
||||
private final WikiContextService wikiContextService;
|
||||
private final vip.mate.workspace.core.service.WorkspaceService workspaceService;
|
||||
private final vip.mate.llm.cache.AnthropicCacheOptionsFactory anthropicCacheOptionsFactory;
|
||||
private final vip.mate.llm.cache.LlmCacheMetricsAggregator llmCacheMetricsAggregator;
|
||||
private final vip.mate.agent.graph.executor.ToolResultStorage toolResultStorage;
|
||||
private final vip.mate.tool.ToolConcurrencyRegistry toolConcurrencyRegistry;
|
||||
|
||||
/**
|
||||
* 根据 AgentEntity 构建完整的 Agent 实例
|
||||
@ -250,8 +253,8 @@ public class AgentGraphBuilder {
|
||||
CompiledGraph buildPlanExecuteGraph(AgentToolSet toolSet, ChatModel chatModel, int maxIterations, String reasoningEffort) {
|
||||
try {
|
||||
ChatModel fallbackModel = buildFallbackModel(chatModel);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel, llmCacheMetricsAggregator);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties, toolResultStorage, toolConcurrencyRegistry);
|
||||
PlanGenerationNode planGenerationNode = new PlanGenerationNode(chatModel, planningService, streamingHelper, conversationWindowManager, toolSet);
|
||||
StepExecutionNode stepExecutionNode = new StepExecutionNode(chatModel, toolSet, executor, planningService, streamTracker, reasoningEffort, streamingHelper, conversationWindowManager);
|
||||
PlanSummaryNode planSummaryNode = new PlanSummaryNode(chatModel, planningService, streamingHelper);
|
||||
@ -344,8 +347,8 @@ public class AgentGraphBuilder {
|
||||
CompiledGraph buildReActGraph(AgentToolSet toolSet, ChatModel chatModel, int maxIterations, String reasoningEffort) {
|
||||
try {
|
||||
ChatModel fallbackModel = buildFallbackModel(chatModel);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties);
|
||||
NodeStreamingChatHelper streamingHelper = new NodeStreamingChatHelper(streamTracker, fallbackModel, llmCacheMetricsAggregator);
|
||||
ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties, toolResultStorage, toolConcurrencyRegistry);
|
||||
ReasoningNode reasoningNode = new ReasoningNode(chatModel, toolSet, reasoningEffort, streamingHelper, conversationWindowManager, streamTracker, 0, wikiContextService);
|
||||
ActionNode actionNode = new ActionNode(executor, streamTracker);
|
||||
ObservationProcessor observationProcessor = new ObservationProcessor(graphObservationProperties);
|
||||
|
||||
@ -44,17 +44,25 @@ public class NodeStreamingChatHelper {
|
||||
|
||||
private final ChatStreamTracker streamTracker;
|
||||
|
||||
/** 备选模型（主模型连续失败后使用） */
|
||||
/** Fallback model, used after consecutive failures of the primary model. */
|
||||
private final ChatModel fallbackModel;
|
||||
|
||||
/** Optional cache-metrics aggregator; {@code null} in tests or when the bean is absent. */
|
||||
private final vip.mate.llm.cache.LlmCacheMetricsAggregator cacheMetrics;
|
||||
|
||||
public NodeStreamingChatHelper(ChatStreamTracker streamTracker) {
|
||||
this.streamTracker = streamTracker;
|
||||
this.fallbackModel = null;
|
||||
this(streamTracker, null, null);
|
||||
}
|
||||
|
||||
public NodeStreamingChatHelper(ChatStreamTracker streamTracker, ChatModel fallbackModel) {
|
||||
this(streamTracker, fallbackModel, null);
|
||||
}
|
||||
|
||||
public NodeStreamingChatHelper(ChatStreamTracker streamTracker, ChatModel fallbackModel,
|
||||
vip.mate.llm.cache.LlmCacheMetricsAggregator cacheMetrics) {
|
||||
this.streamTracker = streamTracker;
|
||||
this.fallbackModel = fallbackModel;
|
||||
this.cacheMetrics = cacheMetrics;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -516,6 +524,7 @@ public class NodeStreamingChatHelper {
|
||||
? AssistantMessage.builder().content(fullContent).toolCalls(finalToolCalls).build()
|
||||
: new AssistantMessage(fullContent);
|
||||
|
||||
recordCacheMetrics(phase, promptTok, completionTok, cacheReadTok, cacheWriteTok);
|
||||
return new StreamResult(fullContent, fullThinking, assembledMessage,
|
||||
finalToolCalls, !finalToolCalls.isEmpty(), promptTok, completionTok,
|
||||
true, null, ErrorType.NONE, true, cacheReadTok, cacheWriteTok);
|
||||
@ -552,11 +561,28 @@ public class NodeStreamingChatHelper {
|
||||
assembledMessage = new AssistantMessage(fullContent);
|
||||
}
|
||||
|
||||
recordCacheMetrics(phase, promptTok, completionTok, cacheReadTok, cacheWriteTok);
|
||||
return new StreamResult(fullContent, fullThinking, assembledMessage,
|
||||
finalToolCalls, !finalToolCalls.isEmpty(), promptTok, completionTok,
|
||||
partial, errorMsg, ErrorType.NONE, false, cacheReadTok, cacheWriteTok);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record token / cache usage to the optional metrics aggregator.
|
||||
* Called only from successful assembly paths ({@link #assembleResult}
|
||||
* and {@link #assembleStoppedResult}) — error paths are excluded because
|
||||
* their token counts are typically zero and would skew the ratio.
|
||||
*/
|
||||
private void recordCacheMetrics(String phase, int promptTok, int completionTok,
|
||||
int cacheReadTok, int cacheWriteTok) {
|
||||
if (cacheMetrics == null) return;
|
||||
// Skip empty-usage records (pure error responses or broken chunks).
|
||||
if (promptTok == 0 && completionTok == 0 && cacheReadTok == 0 && cacheWriteTok == 0) {
|
||||
return;
|
||||
}
|
||||
cacheMetrics.record(phase, promptTok, completionTok, cacheReadTok, cacheWriteTok);
|
||||
}
|
||||
|
||||
/** 构建纯错误 StreamResult（无任何内容） */
|
||||
/**
|
||||
* 从 Prompt 中剥离旧 AssistantMessage 的 thinking/reasoningContent metadata。
|
||||
|
||||
@ -46,7 +46,13 @@ public class ToolExecutionExecutor {
|
||||
return t;
|
||||
});
|
||||
|
||||
/** 默认不安全工具列表（写操作、浏览器交互等） */
|
||||
/**
|
||||
* Legacy hardcoded unsafe set, kept as a fallback when no
|
||||
* {@link vip.mate.tool.ToolConcurrencyRegistry} is wired in (legacy tests,
|
||||
* backwards-compatible constructors). New code should annotate the tool
|
||||
* method with {@link vip.mate.tool.ConcurrencyUnsafe} instead of editing
|
||||
* this list.
|
||||
*/
|
||||
private static final Set<String> DEFAULT_UNSAFE_TOOLS = Set.of(
|
||||
"browser_use", "BrowserUseTool", "write_file", "edit_file"
|
||||
);
|
||||
@ -88,16 +94,35 @@ public class ToolExecutionExecutor {
|
||||
private final ApprovalWorkflowService approvalService;
|
||||
private final ChatStreamTracker streamTracker;
|
||||
private final vip.mate.config.ToolTimeoutProperties toolTimeoutProperties;
|
||||
/** RFC-008 Phase 3 spill store; nullable so legacy constructors keep working. */
|
||||
private final ToolResultStorage resultStorage;
|
||||
/** RFC-008 Phase 4 metadata-driven concurrency classifier; nullable for legacy constructors. */
|
||||
private final vip.mate.tool.ToolConcurrencyRegistry concurrencyRegistry;
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker) {
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, null);
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, null, null, null);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties) {
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, toolTimeoutProperties);
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, toolTimeoutProperties, null, null);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties,
|
||||
ToolResultStorage resultStorage) {
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, toolTimeoutProperties, resultStorage, null);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties,
|
||||
ToolResultStorage resultStorage,
|
||||
vip.mate.tool.ToolConcurrencyRegistry concurrencyRegistry) {
|
||||
this(toolSet, toolGuardService, null, approvalService, streamTracker, toolTimeoutProperties, resultStorage, concurrencyRegistry);
|
||||
}
|
||||
|
||||
public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuard toolGuard,
|
||||
@ -108,18 +133,24 @@ public class ToolExecutionExecutor {
|
||||
this.approvalService = approvalService;
|
||||
this.streamTracker = streamTracker;
|
||||
this.toolTimeoutProperties = null;
|
||||
this.resultStorage = null;
|
||||
this.concurrencyRegistry = null;
|
||||
}
|
||||
|
||||
private ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService,
|
||||
ToolGuard toolGuard, ApprovalWorkflowService approvalService,
|
||||
ChatStreamTracker streamTracker,
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties) {
|
||||
vip.mate.config.ToolTimeoutProperties toolTimeoutProperties,
|
||||
ToolResultStorage resultStorage,
|
||||
vip.mate.tool.ToolConcurrencyRegistry concurrencyRegistry) {
|
||||
this.toolCallbackMap = toolSet.callbackByName();
|
||||
this.toolGuardService = toolGuardService;
|
||||
this.toolGuard = toolGuard;
|
||||
this.approvalService = approvalService;
|
||||
this.streamTracker = streamTracker;
|
||||
this.toolTimeoutProperties = toolTimeoutProperties;
|
||||
this.resultStorage = resultStorage;
|
||||
this.concurrencyRegistry = concurrencyRegistry;
|
||||
}
|
||||
|
||||
private long getToolTimeoutMs(String toolName) {
|
||||
@ -262,9 +293,17 @@ public class ToolExecutionExecutor {
|
||||
executePreparedCalls(preparedCalls, allResponses, events);
|
||||
}
|
||||
|
||||
// 清除 null 占位（不应该有，但防御性处理）
|
||||
// Defensive: drop null placeholders (should never appear in practice).
|
||||
allResponses.removeIf(Objects::isNull);
|
||||
|
||||
// RFC-008 Phase 3 Layer 3: enforce per-turn aggregate budget across all
|
||||
// tool responses for this assistant turn. Spills the largest non-spilled
|
||||
// response in turn until the cumulative size fits the budget.
|
||||
if (resultStorage != null && !allResponses.isEmpty()) {
|
||||
allResponses = new ArrayList<>(resultStorage.enforceTurnBudget(
|
||||
allResponses, conversationId, currentWorkspaceBasePath));
|
||||
}
|
||||
|
||||
boolean hasApprovalPending = barrier != null;
|
||||
return new ToolExecutionResult(allResponses, events, hasApprovalPending,
|
||||
barrier != null ? barrier.pendingId : null,
|
||||
@ -272,11 +311,18 @@ public class ToolExecutionExecutor {
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行预批准的工具调用（用于 StepExecutionNode 的 replay 路径）
|
||||
* Execute a pre-approved tool call (used by StepExecutionNode's replay path
|
||||
* after a user approves a previously-blocked invocation).
|
||||
*
|
||||
* @param conversationId required for per-conversation spill scoping; when
|
||||
* blank, spill files would land in a shared {@code unknown/} directory
|
||||
* and break per-conversation cleanup.
|
||||
* @param workspaceBasePath optional; when blank, spill falls back to tmp.
|
||||
*/
|
||||
public ToolResponseMessage.ToolResponse executePreApproved(
|
||||
AssistantMessage.ToolCall toolCall, String storedArguments,
|
||||
List<GraphEventPublisher.GraphEvent> events) {
|
||||
List<GraphEventPublisher.GraphEvent> events,
|
||||
String conversationId, String workspaceBasePath) {
|
||||
String toolName = toolCall.name();
|
||||
String callArguments = storedArguments != null ? storedArguments : toolCall.arguments();
|
||||
|
||||
@ -291,9 +337,17 @@ public class ToolExecutionExecutor {
|
||||
log.info("[ToolExecutor] Executing pre-approved tool: {}", toolName);
|
||||
String result = callback.call(callArguments);
|
||||
int rawLen = result != null ? result.length() : 0;
|
||||
// Phase 3 Layer 2: spill before truncation when storage is wired.
|
||||
// Use the caller-supplied conversationId so spill files inherit the
|
||||
// same per-conversation directory layout as the non-replay path.
|
||||
if (resultStorage != null && result != null) {
|
||||
String spillConv = conversationId != null && !conversationId.isEmpty() ? conversationId : "unknown";
|
||||
result = resultStorage.persistIfOversized(
|
||||
result, toolName, toolCall.id(), spillConv, workspaceBasePath);
|
||||
}
|
||||
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() + ")" : "");
|
||||
result != null && result.length() < rawLen ? " (now " + result.length() + " after spill/truncate)" : "");
|
||||
events.add(GraphEventPublisher.toolComplete(toolName, result, true));
|
||||
return new ToolResponseMessage.ToolResponse(
|
||||
toolCall.id(), toolName, result != null ? result : "");
|
||||
@ -305,6 +359,21 @@ public class ToolExecutionExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Backwards-compatible overload — replay spills land in a synthetic
|
||||
* {@code unknown/} conversation bucket. New callers must use the
|
||||
* {@link #executePreApproved(AssistantMessage.ToolCall, String, List, String, String)}
|
||||
* variant so spill files are correctly scoped per conversation.
|
||||
*
|
||||
* @deprecated use the 5-arg overload with explicit {@code conversationId}
|
||||
*/
|
||||
@Deprecated
|
||||
public ToolResponseMessage.ToolResponse executePreApproved(
|
||||
AssistantMessage.ToolCall toolCall, String storedArguments,
|
||||
List<GraphEventPublisher.GraphEvent> events) {
|
||||
return executePreApproved(toolCall, storedArguments, events, null, currentWorkspaceBasePath);
|
||||
}
|
||||
|
||||
// ==================== Phase 2: 并发执行 ====================
|
||||
|
||||
private void executePreparedCalls(List<PreparedToolCall> preparedCalls,
|
||||
@ -426,9 +495,17 @@ public class ToolExecutionExecutor {
|
||||
}
|
||||
|
||||
int rawLen = result != null ? result.length() : 0;
|
||||
// RFC-008 Phase 3 Layer 2: spill oversized results to disk and replace
|
||||
// with preview + path. Falls back to truncation when spilling is
|
||||
// disabled or fails. Spill preserves the full output (read_file can
|
||||
// retrieve it); truncation discards the tail.
|
||||
if (resultStorage != null && result != null) {
|
||||
result = resultStorage.persistIfOversized(
|
||||
result, toolName, pc.toolCall.id(), pc.conversationId, pc.workspaceBasePath);
|
||||
}
|
||||
result = truncateToolResult(result, MAX_TOOL_RESULT_CHARS);
|
||||
log.info("[ToolExecutor] Tool {} returned {} chars{}", toolName, rawLen,
|
||||
result != null && result.length() < rawLen ? " (truncated to " + result.length() + ")" : "");
|
||||
result != null && result.length() < rawLen ? " (now " + result.length() + " after spill/truncate)" : "");
|
||||
events.add(GraphEventPublisher.toolComplete(toolName, result, true));
|
||||
if (streamTracker != null) {
|
||||
streamTracker.broadcastObject(pc.conversationId, GraphEventPublisher.EVENT_TOOL_COMPLETE,
|
||||
@ -504,9 +581,15 @@ public class ToolExecutionExecutor {
|
||||
// ==================== 辅助方法 ====================
|
||||
|
||||
/**
|
||||
* 判断工具是否并发安全
|
||||
* Returns true when the tool can run in parallel with other safe tools.
|
||||
* Consults the registry first (annotation-driven, populated at startup);
|
||||
* falls back to the legacy hardcoded set for callers built without a
|
||||
* registry (legacy constructor / unit tests).
|
||||
*/
|
||||
private boolean isConcurrencySafe(String toolName) {
|
||||
if (concurrencyRegistry != null && concurrencyRegistry.isUnsafe(toolName)) {
|
||||
return false;
|
||||
}
|
||||
return !DEFAULT_UNSAFE_TOOLS.contains(toolName);
|
||||
}
|
||||
|
||||
|
||||
@ -0,0 +1,76 @@
|
||||
package vip.mate.agent.graph.executor;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* Configuration for the tool-result three-layer budget (RFC-008 Phase 3).
|
||||
*
|
||||
* <p>Layer 1 — per-tool cap — is implemented inside each tool itself.
|
||||
* Layer 2 — per-result spill — when a single tool result exceeds {@link #perResultThresholdChars}
|
||||
* the full output is written to disk and only a {@link #previewHeadChars} preview
|
||||
* (plus a pointer line) is sent back to the LLM.
|
||||
* Layer 3 — per-turn aggregate budget — after all tools in a turn complete, if
|
||||
* the cumulative response size exceeds {@link #perTurnBudgetChars}, the largest
|
||||
* non-spilled responses are spilled in turn until the aggregate fits.</p>
|
||||
*
|
||||
* <p>Spill files live under {@link #storageBaseDir} when set, otherwise under
|
||||
* {@code <workspaceBasePath>/.mateclaw/tool-results/<conversationId>/} when a
|
||||
* workspace is bound to the agent, otherwise under
|
||||
* {@code ${java.io.tmpdir}/mateclaw/tool-results/<conversationId>/}.</p>
|
||||
*
|
||||
* <pre>
|
||||
* mate:
|
||||
* agent:
|
||||
* tool-result:
|
||||
* enabled: true
|
||||
* per-result-threshold-chars: 4000
|
||||
* per-turn-budget-chars: 16000
|
||||
* preview-head-chars: 800
|
||||
* storage-base-dir:
|
||||
* </pre>
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "mate.agent.tool-result")
|
||||
public class ToolResultProperties {
|
||||
|
||||
/** Master switch. When false, the executor falls back to plain truncation. */
|
||||
private boolean enabled = true;
|
||||
|
||||
/** A single tool result larger than this is spilled to disk. */
|
||||
private int perResultThresholdChars = 4000;
|
||||
|
||||
/** Aggregate cap on combined response size in one tool turn. */
|
||||
private int perTurnBudgetChars = 16000;
|
||||
|
||||
/** Number of leading characters kept inline as a preview after spilling. */
|
||||
private int previewHeadChars = 800;
|
||||
|
||||
/**
|
||||
* Optional absolute path to override the default spill location.
|
||||
* When blank, falls back to {@code <workspace>/.mateclaw/tool-results/} or
|
||||
* {@code ${java.io.tmpdir}/mateclaw/tool-results/}.
|
||||
*/
|
||||
private String storageBaseDir = "";
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
|
||||
public int getPerResultThresholdChars() { return perResultThresholdChars; }
|
||||
public void setPerResultThresholdChars(int perResultThresholdChars) {
|
||||
this.perResultThresholdChars = perResultThresholdChars;
|
||||
}
|
||||
|
||||
public int getPerTurnBudgetChars() { return perTurnBudgetChars; }
|
||||
public void setPerTurnBudgetChars(int perTurnBudgetChars) {
|
||||
this.perTurnBudgetChars = perTurnBudgetChars;
|
||||
}
|
||||
|
||||
public int getPreviewHeadChars() { return previewHeadChars; }
|
||||
public void setPreviewHeadChars(int previewHeadChars) {
|
||||
this.previewHeadChars = previewHeadChars;
|
||||
}
|
||||
|
||||
public String getStorageBaseDir() { return storageBaseDir; }
|
||||
public void setStorageBaseDir(String storageBaseDir) {
|
||||
this.storageBaseDir = storageBaseDir == null ? "" : storageBaseDir;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,215 @@
|
||||
package vip.mate.agent.graph.executor;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.ai.chat.messages.ToolResponseMessage;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Tool-result spill store implementing layers 2 and 3 of the RFC-008 Phase 3
|
||||
* three-layer budget. Layer 1 (per-tool cap) lives inside individual tools.
|
||||
*
|
||||
* <p><b>Layer 2 — per-result spill</b> ({@link #persistIfOversized}): a single
|
||||
* tool result that exceeds the configured threshold is written to disk and
|
||||
* the in-memory copy is replaced with a short preview plus a pointer line so
|
||||
* the LLM can use {@code read_file} to retrieve the full text on demand.</p>
|
||||
*
|
||||
* <p><b>Layer 3 — per-turn aggregate budget</b>
|
||||
* ({@link #enforceTurnBudget}): after every tool in one turn has executed,
|
||||
* if the combined response size still exceeds the turn budget, the largest
|
||||
* non-spilled responses are spilled in turn until the aggregate fits.</p>
|
||||
*
|
||||
* <p>Spill files live under one of, in order:</p>
|
||||
* <ol>
|
||||
* <li>{@code ToolResultProperties.storageBaseDir} when explicitly set</li>
|
||||
* <li>{@code <workspaceBasePath>/.mateclaw/tool-results/<conversationId>/} when a workspace is bound</li>
|
||||
* <li>{@code ${java.io.tmpdir}/mateclaw/tool-results/<conversationId>/} as the universal fallback</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>Failures (disk full, IO error) degrade silently: the original result is
|
||||
* returned unchanged so the agent keeps working. Errors are logged at WARN.</p>
|
||||
*
|
||||
* <p>This class does <b>not</b> manage GC. Spill files accumulate until manually
|
||||
* cleaned. A scheduled cleanup job is tracked as a Phase 3 follow-up.</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(ToolResultProperties.class)
|
||||
public class ToolResultStorage {
|
||||
|
||||
/** Marker placed in the in-context preview so callers and tools can recognize spill output. */
|
||||
public static final String SPILL_MARKER_PREFIX = "[mate-tool-result-spill]";
|
||||
|
||||
private final ToolResultProperties props;
|
||||
|
||||
public ToolResultStorage(ToolResultProperties props) {
|
||||
this.props = props;
|
||||
}
|
||||
|
||||
/**
|
||||
* Layer 2. If {@code result} exceeds the per-result threshold, write the full
|
||||
* text to a spill file and return a preview-plus-pointer string. Otherwise
|
||||
* return the original result unchanged.
|
||||
*
|
||||
* @param result the raw tool output (may be null)
|
||||
* @param toolName used in the preview header so the LLM knows which tool produced it
|
||||
* @param toolUseId unique within a conversation; becomes the spill file's basename
|
||||
* @param conversationId scopes spill files by conversation
|
||||
* @param workspaceBasePath agent's workspace base path; may be null/blank
|
||||
*/
|
||||
public String persistIfOversized(String result, String toolName, String toolUseId,
|
||||
String conversationId, String workspaceBasePath) {
|
||||
if (!props.isEnabled() || result == null) {
|
||||
return result;
|
||||
}
|
||||
if (result.length() <= props.getPerResultThresholdChars()) {
|
||||
return result;
|
||||
}
|
||||
Path file = spillFor(conversationId, toolUseId, workspaceBasePath);
|
||||
if (file == null) {
|
||||
return result;
|
||||
}
|
||||
try {
|
||||
Files.createDirectories(file.getParent());
|
||||
Files.writeString(file, result, StandardCharsets.UTF_8);
|
||||
} catch (IOException ioe) {
|
||||
log.warn("[ToolResultStorage] spill write failed for tool={} convId={} ({}); keeping original",
|
||||
toolName, conversationId, ioe.getMessage());
|
||||
return result;
|
||||
}
|
||||
return buildPreview(result, toolName, file);
|
||||
}
|
||||
|
||||
/**
|
||||
* Layer 3. Walk the responses; if their aggregate length exceeds the turn
|
||||
* budget, spill the largest remaining non-spilled result and recompute.
|
||||
* Mutates the returned list in place by replacing oversized responses.
|
||||
*/
|
||||
public List<ToolResponseMessage.ToolResponse> enforceTurnBudget(
|
||||
List<ToolResponseMessage.ToolResponse> responses,
|
||||
String conversationId,
|
||||
String workspaceBasePath) {
|
||||
if (!props.isEnabled() || responses == null || responses.isEmpty()) {
|
||||
return responses;
|
||||
}
|
||||
int budget = props.getPerTurnBudgetChars();
|
||||
int aggregate = aggregateSize(responses);
|
||||
if (aggregate <= budget) {
|
||||
return responses;
|
||||
}
|
||||
|
||||
log.info("[ToolResultStorage] turn budget exceeded: {} chars > {} (responses={})",
|
||||
aggregate, budget, responses.size());
|
||||
|
||||
List<ToolResponseMessage.ToolResponse> mutable = new ArrayList<>(responses);
|
||||
|
||||
while (aggregate > budget) {
|
||||
// Find the largest response that has not yet been spilled.
|
||||
int targetIdx = -1;
|
||||
int targetLen = -1;
|
||||
for (int i = 0; i < mutable.size(); i++) {
|
||||
String body = mutable.get(i).responseData();
|
||||
if (body == null || body.startsWith(SPILL_MARKER_PREFIX)) continue;
|
||||
if (body.length() > targetLen) {
|
||||
targetLen = body.length();
|
||||
targetIdx = i;
|
||||
}
|
||||
}
|
||||
if (targetIdx < 0) {
|
||||
// Nothing left to spill; aggregate is already as small as we can make it.
|
||||
log.warn("[ToolResultStorage] aggregate still {} chars after spilling everything eligible",
|
||||
aggregate);
|
||||
break;
|
||||
}
|
||||
ToolResponseMessage.ToolResponse target = mutable.get(targetIdx);
|
||||
Path file = spillFor(conversationId, target.id(), workspaceBasePath);
|
||||
if (file == null) {
|
||||
break;
|
||||
}
|
||||
try {
|
||||
Files.createDirectories(file.getParent());
|
||||
Files.writeString(file, target.responseData(), StandardCharsets.UTF_8);
|
||||
} catch (IOException ioe) {
|
||||
log.warn("[ToolResultStorage] spill write failed during turn budget enforcement: {}",
|
||||
ioe.getMessage());
|
||||
break;
|
||||
}
|
||||
String preview = buildPreview(target.responseData(), target.name(), file);
|
||||
mutable.set(targetIdx, new ToolResponseMessage.ToolResponse(target.id(), target.name(), preview));
|
||||
aggregate = aggregateSize(mutable);
|
||||
}
|
||||
return mutable;
|
||||
}
|
||||
|
||||
private static int aggregateSize(List<ToolResponseMessage.ToolResponse> responses) {
|
||||
int sum = 0;
|
||||
for (ToolResponseMessage.ToolResponse r : responses) {
|
||||
if (r.responseData() != null) sum += r.responseData().length();
|
||||
}
|
||||
return sum;
|
||||
}
|
||||
|
||||
private String buildPreview(String fullResult, String toolName, Path spillFile) {
|
||||
int previewLen = Math.min(props.getPreviewHeadChars(), fullResult.length());
|
||||
String head = fullResult.substring(0, previewLen);
|
||||
return SPILL_MARKER_PREFIX
|
||||
+ " tool=" + toolName
|
||||
+ " full_chars=" + fullResult.length()
|
||||
+ " path=" + spillFile.toAbsolutePath()
|
||||
+ "\n[Preview — first " + previewLen + " of " + fullResult.length()
|
||||
+ " chars. Use read_file with the path above to retrieve the rest.]\n"
|
||||
+ head
|
||||
+ "\n…[truncated]";
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the spill file path for a given (conversationId, toolUseId).
|
||||
* Returns {@code null} if no usable directory can be determined.
|
||||
*/
|
||||
private Path spillFor(String conversationId, String toolUseId, String workspaceBasePath) {
|
||||
String safeConv = sanitize(conversationId);
|
||||
String safeId = sanitize(toolUseId);
|
||||
if (safeId.isEmpty()) {
|
||||
safeId = "noid-" + System.nanoTime();
|
||||
}
|
||||
Path base = resolveBaseDir(workspaceBasePath);
|
||||
if (base == null) return null;
|
||||
return base.resolve(safeConv).resolve(safeId + ".txt");
|
||||
}
|
||||
|
||||
private Path resolveBaseDir(String workspaceBasePath) {
|
||||
if (!props.getStorageBaseDir().isEmpty()) {
|
||||
return Paths.get(props.getStorageBaseDir());
|
||||
}
|
||||
if (workspaceBasePath != null && !workspaceBasePath.isBlank()) {
|
||||
return Paths.get(workspaceBasePath, ".mateclaw", "tool-results");
|
||||
}
|
||||
String tmp = System.getProperty("java.io.tmpdir");
|
||||
if (tmp == null || tmp.isEmpty()) return null;
|
||||
return Paths.get(tmp, "mateclaw", "tool-results");
|
||||
}
|
||||
|
||||
/** Strip path separators and reserved characters so user-supplied IDs cannot escape the directory. */
|
||||
private static String sanitize(String s) {
|
||||
if (s == null) return "";
|
||||
return s.replaceAll("[^A-Za-z0-9_.-]", "_");
|
||||
}
|
||||
|
||||
/** Test/admin helper: lexicographic ordering by length, descending. Not used at runtime. */
|
||||
static Comparator<ToolResponseMessage.ToolResponse> byBodyLengthDesc() {
|
||||
return (a, b) -> Integer.compare(
|
||||
b.responseData() == null ? 0 : b.responseData().length(),
|
||||
a.responseData() == null ? 0 : a.responseData().length());
|
||||
}
|
||||
}
|
||||
@ -5,21 +5,24 @@ import com.alibaba.cloud.ai.graph.action.EdgeAction;
|
||||
import vip.mate.agent.graph.plan.state.PlanStateKeys;
|
||||
|
||||
/**
|
||||
* 计划生成后的路由分发器
|
||||
* <p>
|
||||
* 根据 needs_planning 判断：
|
||||
* Routes the graph after the triage node.
|
||||
* <ul>
|
||||
* <li>false → 路由到 DIRECT_ANSWER_NODE（简单问答快速退出）</li>
|
||||
* <li>true → 路由到 STEP_EXECUTION_NODE（开始步骤执行）</li>
|
||||
* <li>{@code needs_planning=false} → {@code DIRECT_ANSWER_NODE} (direct answer, no tools)</li>
|
||||
* <li>{@code needs_planning=true} → {@code STEP_EXECUTION_NODE} (single- or multi-step plan)</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author MateClaw Team
|
||||
* <p>
|
||||
* If the triage key is absent, we default to {@code direct_answer} — an unset
|
||||
* {@code needs_planning} means triage did not run to completion, and Occam's
|
||||
* razor says treat it as "no planning" rather than auto-splitting a task the
|
||||
* system never classified. The previous default ({@code true}) biased every
|
||||
* unresolved request into a multi-step plan, which was the main source of the
|
||||
* "every request splits into subtasks" behavior (see RFC-008).
|
||||
*/
|
||||
public class PlanGenerationDispatcher implements EdgeAction {
|
||||
|
||||
@Override
|
||||
public String apply(OverAllState state) {
|
||||
boolean needsPlanning = state.value(PlanStateKeys.NEEDS_PLANNING, true);
|
||||
boolean needsPlanning = state.value(PlanStateKeys.NEEDS_PLANNING, false);
|
||||
if (!needsPlanning) {
|
||||
return PlanStateKeys.DIRECT_ANSWER_NODE;
|
||||
}
|
||||
|
||||
@ -26,21 +26,23 @@ import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* 计划生成节点
|
||||
* Task triage node for the Plan-Execute graph.
|
||||
* <p>
|
||||
* 职责：
|
||||
* <ol>
|
||||
* <li>判断是否需要规划（简单问答快速退出）</li>
|
||||
* <li>如需规划：生成计划 JSON、解析、校验</li>
|
||||
* <li>调 PlanningService.createPlan() 持久化</li>
|
||||
* <li>发布 plan_created 事件</li>
|
||||
* </ol>
|
||||
* Decides one of three routes for the user's goal and emits a JSON directive:
|
||||
* <ul>
|
||||
* <li>{@code direct_answer} — pure knowledge question, no tools, no planning</li>
|
||||
* <li>single-step plan — needs tools but a single coherent action (steps=1)</li>
|
||||
* <li>multi-step plan — genuinely independent subtasks (2–6 steps)</li>
|
||||
* </ul>
|
||||
* When {@code needs_planning} is false the node streams the direct answer
|
||||
* through {@link NodeStreamingChatHelper} and the graph exits via
|
||||
* {@code DirectAnswerNode}. Otherwise a plan is persisted via
|
||||
* {@link PlanningService} and {@code step_execution} takes over.
|
||||
* <p>
|
||||
* 使用 {@link NodeStreamingChatHelper} 进行流式调用。
|
||||
* 即便最终返回 JSON，也允许模型的 planning 输出以流式产生，最终再聚合解析。
|
||||
* 直接回答路径也通过流式 helper 实时输出给前端。
|
||||
*
|
||||
* @author MateClaw Team
|
||||
* The previous version forced {@code needs_planning=true} whenever any tool
|
||||
* was required, producing multi-step plans for trivial single-hop tasks.
|
||||
* The revised prompt collapses single-hop tool use into a 1-step plan so the
|
||||
* executor can handle it with one ReAct-style iteration (see RFC-008).
|
||||
*/
|
||||
@Slf4j
|
||||
public class PlanGenerationNode implements NodeAction {
|
||||
@ -53,29 +55,31 @@ public class PlanGenerationNode implements NodeAction {
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
private static final String PLANNING_PROMPT = """
|
||||
你是任务规划器，不是聊天助手。
|
||||
你是任务分流器，不是聊天助手。根据用户目标把请求分到三类之一，并只输出一个 JSON 对象。
|
||||
|
||||
你的输出必须满足以下规则：
|
||||
1. 只能返回一个 JSON 对象。
|
||||
2. 不允许输出任何 JSON 之外的文字。
|
||||
3. 不允许使用 markdown 代码块。
|
||||
4. 不要解释，不要寒暄，不要先说"我来...""我先..."。
|
||||
硬性规则：
|
||||
1. 只返回一个 JSON 对象；不允许 markdown 代码块、不允许任何 JSON 以外的文字。
|
||||
2. 不要解释，不要寒暄，不要说"我来...""我先..."。
|
||||
3. 不确定时优先选择"单步"，而不是拆成多步。
|
||||
|
||||
返回格式二选一：
|
||||
三类分流：
|
||||
|
||||
不需要规划时：
|
||||
{"needs_planning": false, "direct_answer": "..."}
|
||||
(A) 直接回答 — 纯知识问答，模型凭自身知识即可回答，不需要任何工具、不需要读文件、不需要查询当前状态。
|
||||
输出：{"needs_planning": false, "direct_answer": "<你的回答>"}
|
||||
|
||||
需要规划时：
|
||||
{"needs_planning": true, "steps": ["步骤1", "步骤2", "步骤3"]}
|
||||
(B) 单步任务 — 需要工具，但本质是一个连贯动作（一次文件读取 / 一次搜索 / 一次命令 / 一次记忆读写 / 一次计算）。
|
||||
执行器会在这一步内部迭代调用多次工具，你**不要**提前拆分。
|
||||
输出：{"needs_planning": true, "steps": ["<将用户目标复述为一句清晰可执行的指令>"]}
|
||||
|
||||
要求：
|
||||
- steps 数量 2 到 6 个。
|
||||
- 每个步骤必须是可执行动作，不要写空话。
|
||||
- 默认不要把 MEMORY.md、PROFILE.md、记忆文件当成独立步骤；但如果用户目标明显依赖历史偏好、长期约束、过往决策或持续上下文，可以加入必要的记忆读取步骤。
|
||||
- 不要把技能文件当成独立步骤，除非用户任务明确要求。
|
||||
- 如果用户目标需要调用任何工具才能完成（包括记忆读写、文件操作、搜索、命令执行等），必须返回 needs_planning: true。只有纯知识问答（不需要调用任何工具的简单问题）才返回 needs_planning: false。
|
||||
- 如果无法确定，也必须返回合法 JSON，不能输出自然语言。
|
||||
(C) 多步任务 — 用户目标包含 2 个及以上明显独立、必须先后完成的子任务（例如"先调研 A 再调研 B 然后对比"、
|
||||
"读配置、迁移数据、验证结果"）。子任务之间如果可以合并，应当合并。
|
||||
输出：{"needs_planning": true, "steps": ["步骤1", "步骤2", ...]}（2 到 6 个步骤）
|
||||
|
||||
关键原则：
|
||||
- 单工具调用绝对不拆成多步。例："读 A 文件并总结" 是单步（B），不是两步。
|
||||
- 默认不要把 MEMORY.md / PROFILE.md / 技能文件读取当成独立步骤；仅当用户明确询问偏好、历史决策或长期约束时才加入。
|
||||
- 每个步骤必须是可执行动作，不写"思考一下""确认一下"之类的空话。
|
||||
- 解析不出来时，视作(B) 单步；宁愿单步也不要无脑拆分。
|
||||
""";
|
||||
|
||||
public PlanGenerationNode(ChatModel chatModel, PlanningService planningService,
|
||||
@ -90,7 +94,7 @@ public class PlanGenerationNode implements NodeAction {
|
||||
}
|
||||
|
||||
/**
|
||||
* @deprecated Use constructor with full parameters
|
||||
* @deprecated use the full-parameter constructor instead
|
||||
*/
|
||||
@Deprecated
|
||||
public PlanGenerationNode(ChatModel chatModel, PlanningService planningService) {
|
||||
@ -110,7 +114,7 @@ public class PlanGenerationNode implements NodeAction {
|
||||
List<GraphEventPublisher.GraphEvent> events = new ArrayList<>();
|
||||
events.add(GraphEventPublisher.phase("planning", Map.of("goal", goal)));
|
||||
|
||||
// Replay 模式：计划已在 state 中（由 chatWithReplayStream 注入），直接跳过 LLM
|
||||
// Replay path: plan is already in state (injected by chatWithReplayStream); skip LLM.
|
||||
Long existingPlanId = state.<Long>value(PlanStateKeys.PLAN_ID).orElse(null);
|
||||
if (existingPlanId != null) {
|
||||
List<String> existingSteps = accessor.planSteps();
|
||||
@ -128,30 +132,32 @@ public class PlanGenerationNode implements NodeAction {
|
||||
}
|
||||
|
||||
try {
|
||||
// 构建 prompt 消息列表：PLANNING_PROMPT 作为独立 system message，
|
||||
// 不拼接完整 systemPrompt（wiki/技能/记忆指南等与规划决策无关，
|
||||
// 拼接后会稀释 PLANNING_PROMPT 的指令优先级）
|
||||
// PLANNING_PROMPT is the sole system message; we deliberately do NOT
|
||||
// concatenate the agent's full systemPrompt (wiki / skill / memory guidance),
|
||||
// which would dilute the triage instructions.
|
||||
List<Message> promptMessages = new ArrayList<>();
|
||||
promptMessages.add(new SystemMessage(PLANNING_PROMPT));
|
||||
// 注入运行时上下文（当前时间 + 工作目录）
|
||||
String workspaceBasePath = state.value(MateClawStateKeys.WORKSPACE_BASE_PATH, "");
|
||||
promptMessages.add(new UserMessage(RuntimeContextInjector.buildContextMessage(workspaceBasePath)));
|
||||
|
||||
// 注入可用工具名称，帮助 LLM 判断用户目标是否需要工具
|
||||
// Advertise available tools so the LLM can recognize when an action is possible,
|
||||
// but do NOT force "any tool usage implies multi-step" — single-hop tool use
|
||||
// should resolve to a 1-step plan, not a multi-step decomposition.
|
||||
if (toolSet != null && !toolSet.callbacks().isEmpty()) {
|
||||
String toolNames = toolSet.callbacks().stream()
|
||||
.map(cb -> cb.getToolDefinition().name())
|
||||
.collect(Collectors.joining(", "));
|
||||
promptMessages.add(new UserMessage(
|
||||
"你可以使用以下工具：" + toolNames
|
||||
+ "\n如果用户目标需要调用任何工具才能完成，必须返回 needs_planning: true。"));
|
||||
"可用工具：" + toolNames
|
||||
+ "\n单次工具调用应归为单步（B），不要拆成多步。"));
|
||||
}
|
||||
|
||||
// 注入 working context（对话历史摘要），让规划能感知之前对话的约束和补充条件
|
||||
// Inject working context (rolling conversation summary) so triage respects
|
||||
// prior constraints without re-reading full history.
|
||||
String workingContext = accessor.workingContext();
|
||||
if (!workingContext.isEmpty()) {
|
||||
promptMessages.add(new UserMessage(
|
||||
"以下是此前对话中用户提出的约束、说明和上下文，请在规划时充分考虑：\n\n"
|
||||
"以下是此前对话中用户提出的约束、说明和上下文，请在分流时参考：\n\n"
|
||||
+ workingContext));
|
||||
}
|
||||
|
||||
@ -159,11 +165,11 @@ public class PlanGenerationNode implements NodeAction {
|
||||
|
||||
Prompt prompt = new Prompt(promptMessages);
|
||||
|
||||
// 静默流式调用 LLM — 返回结构化 JSON，不直接推送给前端
|
||||
// Silent streaming call — structured JSON is parsed below; tokens are not forwarded to the client.
|
||||
NodeStreamingChatHelper.StreamResult result = streamingHelper.streamCallSilent(
|
||||
chatModel, prompt, conversationId, "plan_generation");
|
||||
|
||||
// PTL 处理：压缩后重试
|
||||
// Prompt-too-long handling: compact the conversation window and retry once.
|
||||
if (result.isPromptTooLong() && conversationWindowManager != null) {
|
||||
log.warn("[PlanGeneration] Prompt too long, attempting compaction and retry");
|
||||
List<Message> compactedMessages = conversationWindowManager.compactForRetry(
|
||||
@ -180,20 +186,16 @@ public class PlanGenerationNode implements NodeAction {
|
||||
String llmResponse = result.text();
|
||||
log.debug("[PlanGeneration] LLM response: {}", llmResponse);
|
||||
|
||||
// 清理 markdown 代码块标记
|
||||
String cleanedJson = cleanJsonResponse(llmResponse);
|
||||
|
||||
// 解析 JSON
|
||||
Map<String, Object> parsed = objectMapper.readValue(cleanedJson, new TypeReference<>() {});
|
||||
boolean needsPlanning = Boolean.TRUE.equals(parsed.get("needs_planning"));
|
||||
|
||||
if (!needsPlanning) {
|
||||
// 简单问答快速退出 — 解析出 direct_answer 后手动推送给前端
|
||||
// Category (A): direct answer — push to client and terminate via DirectAnswerNode.
|
||||
String directAnswer = parsed.get("direct_answer") != null
|
||||
? parsed.get("direct_answer").toString() : llmResponse;
|
||||
log.info("[PlanGeneration] Simple question detected, returning direct answer");
|
||||
log.info("[PlanGeneration] Direct-answer route taken (no tools, no planning)");
|
||||
|
||||
// 手动广播 direct_answer 文本（而不是原始 JSON）
|
||||
streamingHelper.broadcastContent(conversationId, directAnswer);
|
||||
|
||||
return PlanStateAccessor.output()
|
||||
@ -207,27 +209,22 @@ public class PlanGenerationNode implements NodeAction {
|
||||
.build();
|
||||
}
|
||||
|
||||
// 需要规划：提取步骤
|
||||
// Categories (B) single-step or (C) multi-step: extract steps.
|
||||
@SuppressWarnings("unchecked")
|
||||
List<String> steps = (List<String>) parsed.get("steps");
|
||||
if (steps == null || steps.isEmpty()) {
|
||||
log.warn("[PlanGeneration] LLM returned needs_planning=true but empty steps, falling back to direct answer");
|
||||
return PlanStateAccessor.output()
|
||||
.needsPlanning(false)
|
||||
.directAnswer(llmResponse)
|
||||
.currentPhase("direct_answer")
|
||||
.contentStreamed(true)
|
||||
.thinkingStreamed(!result.thinking().isEmpty())
|
||||
.mergeUsage(state, result)
|
||||
.events(events)
|
||||
.build();
|
||||
// LLM asked for planning but produced no steps — fall back to a
|
||||
// synthetic 1-step plan using the user's goal so the executor
|
||||
// can still reach the tools. (Previous behavior dropped back to
|
||||
// direct_answer, which silently stripped tool capability.)
|
||||
log.warn("[PlanGeneration] needs_planning=true with empty steps; falling back to single-step plan");
|
||||
steps = List.of(goal);
|
||||
}
|
||||
|
||||
// 持久化计划
|
||||
var plan = planningService.createPlan(agentId, goal, steps);
|
||||
log.info("[PlanGeneration] Plan created: id={}, steps={}", plan.getId(), steps.size());
|
||||
log.info("[PlanGeneration] Plan created: id={}, steps={} ({})",
|
||||
plan.getId(), steps.size(), steps.size() == 1 ? "single-step" : "multi-step");
|
||||
|
||||
// 发布 plan_created 事件
|
||||
events.add(GraphEventPublisher.planCreated(plan.getId(), steps));
|
||||
|
||||
return PlanStateAccessor.output()
|
||||
@ -244,20 +241,39 @@ public class PlanGenerationNode implements NodeAction {
|
||||
.build();
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[PlanGeneration] Failed to generate plan: {}", e.getMessage(), e);
|
||||
// 降级：作为简单问答处理，不向前端暴露内部异常细节
|
||||
return PlanStateAccessor.output()
|
||||
.needsPlanning(false)
|
||||
.directAnswer("抱歉，我暂时无法完成规划，请重试或换一种方式描述任务。")
|
||||
.currentPhase("direct_answer")
|
||||
.events(events)
|
||||
.build();
|
||||
log.error("[PlanGeneration] Triage failed, falling back to single-step plan: {}", e.getMessage(), e);
|
||||
// When the triage LLM fails or returns unparseable output we now fall back to
|
||||
// a single-step plan (the user's goal verbatim) instead of a direct text
|
||||
// answer. This preserves tool access on the failure path; the previous
|
||||
// "direct answer" fallback silently degraded tool-requiring tasks.
|
||||
try {
|
||||
var plan = planningService.createPlan(agentId, goal, List.of(goal));
|
||||
events.add(GraphEventPublisher.planCreated(plan.getId(), List.of(goal)));
|
||||
return PlanStateAccessor.output()
|
||||
.needsPlanning(true)
|
||||
.planId(plan.getId())
|
||||
.planSteps(List.of(goal))
|
||||
.planValid(true)
|
||||
.currentStepIndex(0)
|
||||
.currentPhase("plan_generated")
|
||||
.events(events)
|
||||
.build();
|
||||
} catch (Exception persistErr) {
|
||||
log.error("[PlanGeneration] Single-step fallback persistence also failed: {}", persistErr.getMessage());
|
||||
return PlanStateAccessor.output()
|
||||
.needsPlanning(false)
|
||||
.directAnswer("抱歉，我暂时无法完成任务分流，请重试或换一种方式描述任务。")
|
||||
.currentPhase("direct_answer")
|
||||
.events(events)
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理 LLM 返回的 JSON，移除可能的 markdown 代码块标记。
|
||||
* 若响应中不包含合法的 JSON 对象，抛出异常让调用方走降级路径。
|
||||
* Strip optional markdown code fences and isolate the first JSON object.
|
||||
* Throws if no balanced JSON object is present; the caller treats that as
|
||||
* a triage failure and falls back to a single-step plan.
|
||||
*/
|
||||
private String cleanJsonResponse(String response) {
|
||||
if (response == null) {
|
||||
@ -267,7 +283,6 @@ public class PlanGenerationNode implements NodeAction {
|
||||
if (cleaned.startsWith("```")) {
|
||||
cleaned = cleaned.replaceAll("```json?\\n?", "").replaceAll("```", "").trim();
|
||||
}
|
||||
// 找到第一个 { 和最后一个 }
|
||||
int start = cleaned.indexOf('{');
|
||||
int end = cleaned.lastIndexOf('}');
|
||||
if (start < 0 || end <= start) {
|
||||
|
||||
@ -181,7 +181,7 @@ public class StepExecutionNode implements NodeAction {
|
||||
String storedArguments = extractArgumentsFromPayload(preApprovedPayload);
|
||||
events.add(GraphEventPublisher.toolStart(toolCall.name(), toolCall.arguments()));
|
||||
ToolResponseMessage.ToolResponse response = executor.executePreApproved(
|
||||
toolCall, storedArguments, events);
|
||||
toolCall, storedArguments, events, conversationId, workspaceBasePath);
|
||||
toolResponses.add(response);
|
||||
preApprovedPayload = ""; // 只消费一次
|
||||
} else {
|
||||
@ -267,10 +267,22 @@ public class StepExecutionNode implements NodeAction {
|
||||
stepIndex + 1, steps.size(),
|
||||
finalResult.length() > 100 ? finalResult.substring(0, 100) + "..." : finalResult);
|
||||
|
||||
// 更新 working context：将最新完成的步骤结果纳入摘要
|
||||
List<String> allCompleted = new ArrayList<>(accessor.completedResults());
|
||||
allCompleted.add(formatStepResult(stepIndex, finalResult));
|
||||
String updatedWorkingContext = rebuildWorkingContext(accessor, allCompleted);
|
||||
// RFC-008 P4.2: incremental working-context update.
|
||||
// Previous behavior rebuilt the entire context from history + every
|
||||
// completed result on every step (O(N) per step). On long plans this
|
||||
// re-walks the same conversation history each iteration. Now we take
|
||||
// the previous context as-is (which already encodes earlier history
|
||||
// and earlier completed steps) and append just the freshly-completed
|
||||
// step, then trim from the head if the running total exceeds the cap.
|
||||
// For first-step calls where prior context is empty, fall through to
|
||||
// the original rebuild path so the conversation history seed is still
|
||||
// captured.
|
||||
String prevWorkingContext = accessor.workingContext();
|
||||
String formattedNewStep = formatStepResult(stepIndex, finalResult);
|
||||
String updatedWorkingContext = prevWorkingContext.isEmpty()
|
||||
? rebuildWorkingContext(accessor,
|
||||
appendOne(accessor.completedResults(), formattedNewStep))
|
||||
: appendStepIncremental(prevWorkingContext, formattedNewStep);
|
||||
|
||||
return PlanStateAccessor.output()
|
||||
.currentStepResult(finalResult)
|
||||
@ -414,9 +426,51 @@ public class StepExecutionNode implements NodeAction {
|
||||
}
|
||||
}
|
||||
|
||||
/** Append helper used by the incremental working-context fast path. */
|
||||
private static List<String> appendOne(List<String> previous, String item) {
|
||||
List<String> out = new ArrayList<>(previous);
|
||||
out.add(item);
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据当前 accessor 中的会话历史消息和更新后的已完成步骤结果，
|
||||
* 重建 working context。复用与 StateGraphPlanExecuteAgent.buildWorkingContext 相同的逻辑。
|
||||
* Incrementally extend the previous working context with one new step
|
||||
* result. Cheap O(1) path used for steps 2..N: avoids walking the full
|
||||
* conversation history again. The result is trimmed from the head if it
|
||||
* exceeds the same overall cap that {@link #rebuildWorkingContext}
|
||||
* enforces, so the budget invariant is preserved.
|
||||
*
|
||||
* <p>Per-step truncation: a single step result longer than 800 chars is
|
||||
* abbreviated before append, mirroring the per-step caps in
|
||||
* {@code rebuildWorkingContext}.</p>
|
||||
*/
|
||||
private static String appendStepIncremental(String previousContext, String formattedStepResult) {
|
||||
final int OVERALL_CAP = 6000;
|
||||
final int PER_STEP_CAP = 800;
|
||||
String stepLine = formattedStepResult.length() > PER_STEP_CAP
|
||||
? formattedStepResult.substring(0, PER_STEP_CAP) + "…"
|
||||
: formattedStepResult;
|
||||
String combined = previousContext + "\n" + stepLine + "\n";
|
||||
if (combined.length() <= OVERALL_CAP) {
|
||||
return combined;
|
||||
}
|
||||
// Drop oldest content from the head until we fit. Cut on a newline
|
||||
// boundary so we don't truncate mid-line.
|
||||
int overshoot = combined.length() - OVERALL_CAP;
|
||||
int cutFrom = combined.indexOf('\n', overshoot);
|
||||
if (cutFrom < 0 || cutFrom >= combined.length() - 1) {
|
||||
cutFrom = overshoot;
|
||||
} else {
|
||||
cutFrom += 1; // skip the newline itself
|
||||
}
|
||||
return "…(earlier context truncated)\n" + combined.substring(cutFrom);
|
||||
}
|
||||
|
||||
/**
|
||||
* Full rebuild of working context from conversation history plus all
|
||||
* completed step results. Reused on the cold path (first step, or when
|
||||
* the incremental path can't be applied). Mirrors
|
||||
* {@code StateGraphPlanExecuteAgent.buildWorkingContext}.
|
||||
*/
|
||||
private static String rebuildWorkingContext(PlanStateAccessor accessor, List<String> allCompletedResults) {
|
||||
List<Message> messages = accessor.messages();
|
||||
|
||||
@ -48,7 +48,10 @@ public final class PlanStateAccessor {
|
||||
}
|
||||
|
||||
public boolean needsPlanning() {
|
||||
return state.value(NEEDS_PLANNING, true);
|
||||
// Default to false: an unset triage flag means the request was not
|
||||
// classified as requiring a plan. See PlanGenerationDispatcher for the
|
||||
// rationale and RFC-008 for the full discussion.
|
||||
return state.value(NEEDS_PLANNING, false);
|
||||
}
|
||||
|
||||
// ===== 步骤控制 =====
|
||||
|
||||
134
mateclaw-server/src/main/java/vip/mate/llm/cache/LlmCacheMetricsAggregator.java
vendored
Normal file
134
mateclaw-server/src/main/java/vip/mate/llm/cache/LlmCacheMetricsAggregator.java
vendored
Normal file
@ -0,0 +1,134 @@
|
||||
package vip.mate.llm.cache;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* In-memory, per-phase aggregator for LLM cache usage.
|
||||
*
|
||||
* <p>Complements RFC-014 (Anthropic prompt cache wiring) by exposing live
|
||||
* observability. RFC-014 already persists daily totals in {@code mate_usage_daily};
|
||||
* this aggregator lets operators see the effective cache hit ratio right now,
|
||||
* per graph phase (reasoning / step_execution / plan_generation / ...), without
|
||||
* needing to query the database or install Prometheus/Actuator.</p>
|
||||
*
|
||||
* <p>Thread-safe and lock-free: counters are striped per phase via
|
||||
* {@link ConcurrentHashMap} and each field is an {@link AtomicLong}.</p>
|
||||
*
|
||||
* <p>Periodic log summary: every {@value #LOG_EVERY_N_REQUESTS}-th request for a
|
||||
* given phase, a one-line summary is emitted at INFO level so long-running
|
||||
* processes leave a trail even without a metrics backend.</p>
|
||||
*
|
||||
* <p>If {@code spring-boot-starter-actuator} is later added, this aggregator
|
||||
* can be bridged to a {@code MeterRegistry} by iterating {@link #snapshot()}.</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class LlmCacheMetricsAggregator {
|
||||
|
||||
private static final int LOG_EVERY_N_REQUESTS = 50;
|
||||
|
||||
/** Per-phase counters. {@code phase} is a short label like "reasoning" or "step_execution". */
|
||||
private final Map<String, PhaseCounters> byPhase = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* Record a single LLM call's token usage. Safe to call from streaming
|
||||
* subscriber threads; does not block.
|
||||
*/
|
||||
public void record(String phase,
|
||||
int promptTokens,
|
||||
int completionTokens,
|
||||
int cacheReadTokens,
|
||||
int cacheWriteTokens) {
|
||||
if (phase == null || phase.isEmpty()) {
|
||||
phase = "unknown";
|
||||
}
|
||||
PhaseCounters c = byPhase.computeIfAbsent(phase, p -> new PhaseCounters());
|
||||
long requests = c.requests.incrementAndGet();
|
||||
c.promptTokens.addAndGet(Math.max(0, promptTokens));
|
||||
c.completionTokens.addAndGet(Math.max(0, completionTokens));
|
||||
c.cacheReadTokens.addAndGet(Math.max(0, cacheReadTokens));
|
||||
c.cacheWriteTokens.addAndGet(Math.max(0, cacheWriteTokens));
|
||||
if (cacheReadTokens > 0) {
|
||||
c.cacheHits.incrementAndGet();
|
||||
} else if (cacheWriteTokens > 0) {
|
||||
c.cacheWrites.incrementAndGet();
|
||||
} else {
|
||||
c.cacheMisses.incrementAndGet();
|
||||
}
|
||||
|
||||
if (requests % LOG_EVERY_N_REQUESTS == 0) {
|
||||
log.info("[llm-cache] phase={} requests={} hit_ratio={} prompt_tokens={} cache_read={} cache_write={}",
|
||||
phase, requests, formatHitRatio(c),
|
||||
c.promptTokens.get(), c.cacheReadTokens.get(), c.cacheWriteTokens.get());
|
||||
}
|
||||
}
|
||||
|
||||
/** Point-in-time copy of all counters. Intended for admin endpoints and tests. */
|
||||
public Map<String, PhaseSnapshot> snapshot() {
|
||||
Map<String, PhaseSnapshot> out = new java.util.LinkedHashMap<>();
|
||||
byPhase.forEach((phase, c) -> out.put(phase, c.snapshot()));
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Reset all counters. Useful for tests; not wired to any endpoint. */
|
||||
public void reset() {
|
||||
byPhase.clear();
|
||||
}
|
||||
|
||||
private static String formatHitRatio(PhaseCounters c) {
|
||||
long hits = c.cacheHits.get();
|
||||
long total = c.requests.get();
|
||||
if (total == 0) return "n/a";
|
||||
return String.format("%.1f%%", 100.0 * hits / total);
|
||||
}
|
||||
|
||||
/** Mutable per-phase counters. Package-private so the aggregator owns the state. */
|
||||
private static final class PhaseCounters {
|
||||
final AtomicLong requests = new AtomicLong();
|
||||
final AtomicLong cacheHits = new AtomicLong();
|
||||
final AtomicLong cacheWrites = new AtomicLong();
|
||||
final AtomicLong cacheMisses = new AtomicLong();
|
||||
final AtomicLong promptTokens = new AtomicLong();
|
||||
final AtomicLong completionTokens = new AtomicLong();
|
||||
final AtomicLong cacheReadTokens = new AtomicLong();
|
||||
final AtomicLong cacheWriteTokens = new AtomicLong();
|
||||
|
||||
PhaseSnapshot snapshot() {
|
||||
return new PhaseSnapshot(requests.get(), cacheHits.get(), cacheWrites.get(),
|
||||
cacheMisses.get(), promptTokens.get(), completionTokens.get(),
|
||||
cacheReadTokens.get(), cacheWriteTokens.get());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Immutable snapshot of one phase's counters.
|
||||
*
|
||||
* <p>{@link #hitRatio()} is hits / requests; {@code writes} are counted
|
||||
* separately because the first call on a new cache breakpoint is a
|
||||
* pure write (no hit) and shouldn't pollute the hit ratio denominator.</p>
|
||||
*/
|
||||
public record PhaseSnapshot(long requests,
|
||||
long cacheHits,
|
||||
long cacheWrites,
|
||||
long cacheMisses,
|
||||
long promptTokens,
|
||||
long completionTokens,
|
||||
long cacheReadTokens,
|
||||
long cacheWriteTokens) {
|
||||
|
||||
public double hitRatio() {
|
||||
return requests == 0 ? 0.0 : (double) cacheHits / requests;
|
||||
}
|
||||
|
||||
public double effectiveSavingsRatio() {
|
||||
long paid = promptTokens - cacheReadTokens;
|
||||
if (promptTokens == 0) return 0.0;
|
||||
return 1.0 - ((double) paid / promptTokens);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,46 @@
|
||||
package vip.mate.tool;
|
||||
|
||||
import java.lang.annotation.ElementType;
|
||||
import java.lang.annotation.Retention;
|
||||
import java.lang.annotation.RetentionPolicy;
|
||||
import java.lang.annotation.Target;
|
||||
|
||||
/**
|
||||
* Marks a {@link org.springframework.ai.tool.annotation.Tool}-annotated method
|
||||
* as <b>not safe to run concurrently with itself or with other tools that
|
||||
* touch the same state</b>.
|
||||
*
|
||||
* <p>Read by {@link vip.mate.tool.ToolConcurrencyRegistry} at startup. The
|
||||
* registry returns {@code true} from
|
||||
* {@link ToolConcurrencyRegistry#isUnsafe(String)} for marked tools, which
|
||||
* causes {@code ToolExecutionExecutor} to execute them in their own batch
|
||||
* (no parallelism, no overlap with the surrounding safe batch).</p>
|
||||
*
|
||||
* <p>Use cases:</p>
|
||||
* <ul>
|
||||
* <li>File writes / edits ({@code WriteFileTool}, {@code EditFileTool})</li>
|
||||
* <li>Shell command execution ({@code ShellExecuteTool})</li>
|
||||
* <li>Stateful workspace mutations ({@code WorkspaceMemoryTool}, {@code SkillManageTool})</li>
|
||||
* <li>Persistent operations on shared resources ({@code CronJobTool}, {@code DatasourceTool})</li>
|
||||
* <li>Long-running generative tools where API rate limits forbid parallel calls
|
||||
* ({@code ImageGenerateTool}, {@code VideoGenerateTool})</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Read-only and idempotent tools should remain unannotated; they will be
|
||||
* batched together for parallel execution by the executor.</p>
|
||||
*
|
||||
* <p>For MCP-provided tools the executor will eventually consult the
|
||||
* {@code annotations.readOnlyHint} field from the MCP {@code Tool} schema;
|
||||
* that integration is tracked as a Phase 4 follow-up. Until then MCP tools
|
||||
* default to safe (their pre-existing behavior).</p>
|
||||
*/
|
||||
@Retention(RetentionPolicy.RUNTIME)
|
||||
@Target(ElementType.METHOD)
|
||||
public @interface ConcurrencyUnsafe {
|
||||
|
||||
/**
|
||||
* Optional human-readable reason. Surfaced in startup logs to help
|
||||
* operators audit which tools have been marked unsafe.
|
||||
*/
|
||||
String value() default "";
|
||||
}
|
||||
@ -0,0 +1,129 @@
|
||||
package vip.mate.tool;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.ai.tool.annotation.Tool;
|
||||
import org.springframework.beans.factory.config.BeanDefinition;
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* Startup-scanned registry of tools that must run sequentially
|
||||
* ({@link ConcurrencyUnsafe}-annotated). Replaces the previous hardcoded
|
||||
* {@code DEFAULT_UNSAFE_TOOLS} set in {@code ToolExecutionExecutor}.
|
||||
*
|
||||
* <p>Discovery walks every bean <b>definition</b> and inspects the declared
|
||||
* class's methods for the {@link Tool} + {@link ConcurrencyUnsafe} pair.
|
||||
* Beans are <b>not instantiated</b> by this scan — we only resolve the bean
|
||||
* class name and load it via the class loader, which preserves {@code @Lazy}
|
||||
* semantics and avoids triggering ChatModel / DataSource / MCP-client
|
||||
* construction at registry init.</p>
|
||||
*
|
||||
* <p>Tool name resolution mirrors Spring AI's logic: {@code @Tool#name()}
|
||||
* when set, otherwise the method's simple name.</p>
|
||||
*
|
||||
* <p>The registry is immutable after {@link #scan()}; the unsafe set is
|
||||
* populated once and consulted on every tool execution. MCP tools are not
|
||||
* scanned (their {@link Tool} annotations live inside the MCP framework, not
|
||||
* on user-visible methods); MCP support is tracked as a follow-up.</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class ToolConcurrencyRegistry {
|
||||
|
||||
private final ConfigurableApplicationContext applicationContext;
|
||||
|
||||
/** Populated once at startup; never mutated thereafter. */
|
||||
private volatile Set<String> unsafeNames = Collections.emptySet();
|
||||
|
||||
public ToolConcurrencyRegistry(ConfigurableApplicationContext applicationContext) {
|
||||
this.applicationContext = applicationContext;
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
void scan() {
|
||||
Set<String> discovered = new HashSet<>();
|
||||
ConfigurableListableBeanFactory factory = applicationContext.getBeanFactory();
|
||||
ClassLoader classLoader = applicationContext.getClassLoader();
|
||||
|
||||
for (String beanName : factory.getBeanDefinitionNames()) {
|
||||
Class<?> beanClass = resolveBeanClassWithoutInstantiating(factory, beanName, classLoader);
|
||||
if (beanClass == null) continue;
|
||||
|
||||
// Unwrap CGLIB subclasses (proxies) so we see user-declared methods.
|
||||
Class<?> userClass = ClassUtils.getUserClass(beanClass);
|
||||
for (Method method : userClass.getDeclaredMethods()) {
|
||||
Tool tool = method.getAnnotation(Tool.class);
|
||||
if (tool == null) continue;
|
||||
ConcurrencyUnsafe unsafe = method.getAnnotation(ConcurrencyUnsafe.class);
|
||||
if (unsafe == null) continue;
|
||||
String toolName = tool.name() != null && !tool.name().isEmpty() ? tool.name() : method.getName();
|
||||
discovered.add(toolName);
|
||||
log.info("[ToolConcurrencyRegistry] Marked tool '{}' as unsafe ({}#{}): {}",
|
||||
toolName, userClass.getSimpleName(), method.getName(),
|
||||
unsafe.value().isEmpty() ? "no reason given" : unsafe.value());
|
||||
}
|
||||
}
|
||||
// Keep the legacy hardcoded names so existing deployments without
|
||||
// annotations still see the same behavior. New code should rely on
|
||||
// the @ConcurrencyUnsafe annotation rather than this list.
|
||||
discovered.addAll(Arrays.asList("browser_use", "BrowserUseTool", "write_file", "edit_file"));
|
||||
this.unsafeNames = Collections.unmodifiableSet(discovered);
|
||||
log.info("[ToolConcurrencyRegistry] Concurrency-unsafe tools ({}): {}",
|
||||
unsafeNames.size(), unsafeNames);
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a bean's class without instantiating it.
|
||||
* Preference order:
|
||||
* <ol>
|
||||
* <li>{@link BeanDefinition#getBeanClassName()} → {@link Class#forName} via the context class loader
|
||||
* (works for stereotype-scanned components).</li>
|
||||
* <li>{@code factory.getType(beanName, false)} as a fallback for
|
||||
* {@code @Bean}-defined or programmatically registered beans.
|
||||
* The {@code false} flag forbids FactoryBean initialization.</li>
|
||||
* </ol>
|
||||
* Returns {@code null} when neither path yields a class — for example,
|
||||
* lambda-defined beans without a resolvable class name.
|
||||
*/
|
||||
private static Class<?> resolveBeanClassWithoutInstantiating(ConfigurableListableBeanFactory factory,
|
||||
String beanName,
|
||||
ClassLoader classLoader) {
|
||||
try {
|
||||
BeanDefinition bd = factory.getBeanDefinition(beanName);
|
||||
String className = bd.getBeanClassName();
|
||||
if (className != null && !className.isEmpty()) {
|
||||
try {
|
||||
return Class.forName(className, false, classLoader);
|
||||
} catch (ClassNotFoundException | LinkageError ignored) {
|
||||
// Fall through to factory.getType fallback.
|
||||
}
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
// No bean definition (singleton registered programmatically); fall through.
|
||||
}
|
||||
try {
|
||||
return factory.getType(beanName, false);
|
||||
} catch (Exception ignored) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** {@code true} when the named tool must execute alone (no parallelism). */
|
||||
public boolean isUnsafe(String toolName) {
|
||||
return toolName != null && unsafeNames.contains(toolName);
|
||||
}
|
||||
|
||||
/** Defensive copy for diagnostics / admin endpoints. */
|
||||
public Set<String> snapshot() {
|
||||
return unsafeNames;
|
||||
}
|
||||
}
|
||||
@ -30,6 +30,7 @@ public class CronJobTool {
|
||||
|
||||
private final CronJobService cronJobService;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("cron job creation persists to mate_cron_job; concurrent creates can race on name")
|
||||
@Tool(description = "Create a scheduled task (cron job). The task will run automatically at the specified time "
|
||||
+ "and send the trigger message to the current agent. Use 5-field cron expressions: minute hour day month weekday. "
|
||||
+ "Examples: '0 9 * * *' = daily at 9am, '0 9 * * 1-5' = weekdays at 9am, '*/30 * * * *' = every 30 minutes.")
|
||||
@ -99,6 +100,7 @@ public class CronJobTool {
|
||||
}
|
||||
}
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("toggles row state in mate_cron_job; serialize to keep enabled/disabled deterministic")
|
||||
@Tool(description = "Enable or disable a scheduled task by its job ID. "
|
||||
+ "Use list_cron_jobs first to find the job ID.")
|
||||
public String toggle_cron_job(
|
||||
@ -120,6 +122,7 @@ public class CronJobTool {
|
||||
}
|
||||
}
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("destructive — removes row from mate_cron_job")
|
||||
@Tool(description = "Delete a scheduled task by its job ID. This action requires user approval. "
|
||||
+ "Use list_cron_jobs first to find the job ID.")
|
||||
public String delete_cron_job(
|
||||
|
||||
@ -61,6 +61,7 @@ public class DelegateAgentTool {
|
||||
|
||||
// ==================== 单任务委派 ====================
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("spawns a child agent session and writes to mate_conversation; serialize to keep session graph deterministic")
|
||||
@Tool(description = """
|
||||
Delegate a task to another Agent for multi-agent collaboration. \
|
||||
Target Agent executes in an independent session and returns its final reply. \
|
||||
@ -117,6 +118,7 @@ public class DelegateAgentTool {
|
||||
|
||||
// ==================== 并行委派 ====================
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("internally fans out to its own thread pool; outer executor must not double-parallelize")
|
||||
@Tool(description = """
|
||||
Delegate multiple tasks to different Agents in parallel (max 3). \
|
||||
Each task runs concurrently in an independent child session. \
|
||||
|
||||
@ -33,6 +33,7 @@ public class EditFileTool {
|
||||
|
||||
private final vip.mate.i18n.I18nService i18n;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("in-place file edit — must not race with reads/writes on the same path")
|
||||
@Tool(description = "Edit file content via find-and-replace. Finds exact match of old_text and replaces with new_text. "
|
||||
+ "Returns structured JSON with filePath, replacements count. "
|
||||
+ "Requires user approval. Replaces first occurrence by default; set replaceAll=true for all.")
|
||||
|
||||
@ -29,6 +29,7 @@ public class ImageGenerateTool {
|
||||
private final SystemSettingService systemSettingService;
|
||||
private final AsyncTaskService asyncTaskService;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("creates async tasks and persists generated artifacts; provider rate limits also forbid parallel calls")
|
||||
@Tool(description = "Image generation tool. Supports actions: generate (default), list (show available providers), "
|
||||
+ "status (check task status). Some providers are async (30s-2min), results auto-displayed in conversation.")
|
||||
public String image_generate(
|
||||
|
||||
@ -43,6 +43,7 @@ public class ShellExecuteTool {
|
||||
private static final boolean IS_WINDOWS = System.getProperty("os.name", "")
|
||||
.toLowerCase(Locale.ROOT).contains("win");
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("shell command execution can mutate global state in ways the executor can't reason about")
|
||||
@Tool(description = "Execute a shell command on the local server. For running system commands, viewing files, running scripts. "
|
||||
+ "Uses cmd.exe on Windows, /bin/sh on Linux/macOS. "
|
||||
+ "Dangerous operations trigger security approval. Returns structured result with exitCode, stdout, stderr, timedOut.")
|
||||
|
||||
@ -43,6 +43,7 @@ public class SkillManageTool {
|
||||
/** Skill 内容最大长度（~25K tokens） */
|
||||
private static final int MAX_CONTENT_CHARS = 100_000;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("create/edit/patch/delete on the shared skill registry; concurrent ops on the same skill name race")
|
||||
@Tool(description = """
|
||||
Manage reusable skills: create, edit, patch, or delete skill procedures (SKILL.md format).
|
||||
|
||||
|
||||
@ -28,6 +28,7 @@ public class SkillScriptTool {
|
||||
private final SkillFileAccessPolicy accessPolicy;
|
||||
private final SkillScriptExecutionService executionService;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("script execution can have arbitrary side effects on the host process and filesystem")
|
||||
@Tool(description = """
|
||||
Execute a script from a skill's scripts/ directory.
|
||||
Use this when you need to run skill-provided automation or utilities.
|
||||
|
||||
@ -32,6 +32,7 @@ public class VideoGenerateTool {
|
||||
private final SystemSettingService systemSettingService;
|
||||
private final AsyncTaskService asyncTaskService;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("creates async tasks and persists generated artifacts; provider rate limits also forbid parallel calls")
|
||||
@Tool(description = "视频生成工具，支持以下 action：\n"
|
||||
+ "- generate（默认）：生成视频。提供 prompt 描述视频内容，可选 aspectRatio/duration/imageUrl/model\n"
|
||||
+ "- list：列出所有可用的视频 Provider 及其支持的模型和能力\n"
|
||||
|
||||
@ -100,6 +100,7 @@ public class WorkspaceMemoryTool {
|
||||
return JSONUtil.toJsonPrettyStr(result);
|
||||
}
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("workspace memory write — concurrent writes to the same file would clobber each other")
|
||||
@Tool(description = """
|
||||
创建或覆写指定 Agent 的数据库工作区记忆文件。
|
||||
适用于把提炼后的长期记忆写入 MEMORY.md，或把原始事件写入 memory/YYYY-MM-DD.md。
|
||||
@ -133,6 +134,7 @@ public class WorkspaceMemoryTool {
|
||||
return JSONUtil.toJsonPrettyStr(result);
|
||||
}
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("workspace memory edit — find/replace must serialize per file")
|
||||
@Tool(description = """
|
||||
通过精确查找替换编辑指定 Agent 的数据库工作区记忆文件。
|
||||
适用于在 MEMORY.md 的某个 section 中做增量更新，避免整篇重写。
|
||||
|
||||
@ -33,6 +33,7 @@ public class WriteFileTool {
|
||||
|
||||
private final vip.mate.i18n.I18nService i18n;
|
||||
|
||||
@vip.mate.tool.ConcurrencyUnsafe("file write — must serialize with reads/writes on overlapping paths")
|
||||
@Tool(description = "Write content to a file. Overwrites if exists, creates if not (auto-creates parent directories). "
|
||||
+ "Returns structured JSON with filePath, bytesWritten. "
|
||||
+ "Requires user approval.")
|
||||
|
||||
@ -151,6 +151,17 @@ mate:
|
||||
per-category:
|
||||
shell: 120
|
||||
web: 30
|
||||
# RFC-008 Phase 3: tool-result three-layer budget (per-result spill + per-turn aggregate budget).
|
||||
# Layer 1 (per-tool cap) lives inside individual tools; Layer 2 spills oversized
|
||||
# single results to disk; Layer 3 enforces an aggregate cap on the combined
|
||||
# response size of one tool turn. The full output is preserved on disk and
|
||||
# the in-context preview points the agent at the spill file (read_file tool).
|
||||
tool-result:
|
||||
enabled: true
|
||||
per-result-threshold-chars: 4000
|
||||
per-turn-budget-chars: 16000
|
||||
preview-head-chars: 800
|
||||
storage-base-dir: ""
|
||||
conversation:
|
||||
window:
|
||||
# 测试时临时调低：2000 token ≈ 2000 中文字，3 轮对话即可触发压缩
|
||||
|
||||
Loading…
Reference in New Issue
Block a user