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

fix(agent): review fixes for Lane D — D-2 strategy split, D-4 naming, D-5 docs, D-6 instrumentation

Browse Source
This commit is contained in:
matevip 2026-04-22 10:13:13 +08:00
parent 23133ea45d
commit 320e13b975
6 changed files with 114 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java
View File

@ -27,6 +27,8 @@ public final class GraphEventPublisher {
public static final String EVENT_STEP_STARTED = "plan_step_started";
public static final String EVENT_STEP_COMPLETED = "plan_step_completed";
public static final String EVENT_TOOL_APPROVAL_REQUESTED = "tool_approval_requested";
/** RFC-06 D-6: lightweight performance summary emitted per-phase. */
public static final String EVENT_PERF_SUMMARY = "perf_summary";
/**
* 事件记录
@ -121,6 +123,22 @@ public final class GraphEventPublisher {
return new GraphEvent(EVENT_TOOL_APPROVAL_REQUESTED, Map.copyOf(data), ts);
}
/**
* RFC-06 D-6: emit a lightweight performance summary for a phase.
* Consumers (dashboard, audit, _usage_final) can aggregate these
* to reconstruct per-turn latency profiles without full tracing.
*
* @param phase e.g. "triage", "reasoning", "tool_execution"
* @param metrics arbitrary key-value pairs (e.g. "retry_count", "backoff_wait_ms")
*/
public static GraphEvent perfSummary(String phase, Map<String, Object> metrics) {
long ts = System.currentTimeMillis();
Map<String, Object> data = new java.util.HashMap<>(metrics);
data.put("phase", phase);
data.put("timestamp", ts);
return new GraphEvent(EVENT_PERF_SUMMARY, Map.copyOf(data), ts);
}
// ===== 提取方法 =====
/**

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

@ -271,7 +271,9 @@ public class NodeStreamingChatHelper {
// ==================== 重试配置 ====================
private static final int MAX_RETRIES = 5;
// For rate-limit and server-error: fail fast to failover chain.
// RATE_LIMIT: fail fast to failover chain — staying on the same
// provider during a rate-limit window wastes time without recovery.
// SERVER_ERROR keeps MAX_RETRIES (upstream flaps often self-heal).
private static final int MAX_RETRIES_RATE_LIMIT = 2;
private static final long BACKOFF_BASE_MS = 3000;
private static final long BACKOFF_CAP_MS = 60_000;
@ -406,9 +408,18 @@ public class NodeStreamingChatHelper {
}
}
// D-6: performance counters
int retryCount = 0;
long totalBackoffMs = 0;
int failoverCount = 0;
int llmCallCount = 0;
long callStartMs = System.currentTimeMillis();
// 主模型重试循环
StreamResult lastResult = null;
if (!primarySkipped) for (int attempt = 0; attempt <= MAX_RETRIES; attempt++) {
llmCallCount++;
if (attempt > 0) retryCount++;
lastResult = doStreamCall(chatModel, prompt, conversationId, phase, broadcast, attempt);
if (lastResult != null) {
// PTL: 不重试，直接返回给上层 Node 处理
@ -462,6 +473,7 @@ public class NodeStreamingChatHelper {
if (lastResult.errorMessage() == null || lastResult.errorType() == ErrorType.NONE) {
recordPrimary(true);
addToPool(primaryProviderId);
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult;
}
// Any other non-null errored result with a classified type that doStreamCall
@ -469,6 +481,7 @@ public class NodeStreamingChatHelper {
// must exit — otherwise we silently spin through attempts and waste seconds
// per turn on unrecoverable errors like DashScope's "url error" / unknown model.
recordPrimary(false);
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult;
}
// lastResult == null 表示需要重试
@ -509,6 +522,8 @@ public class NodeStreamingChatHelper {
broadcastDelta(conversationId, "warning",
buildDeltaJson("主模型不可用，正在切换到备选模型 (" + (i + 1) + "/" + fallbackChain.size() + ")..."));
}
failoverCount++;
llmCallCount++;
StreamResult fallbackResult = doStreamCall(fallback, prompt, conversationId,
phase + "_fallback_" + (i + 1), broadcast, 0);
// Accept only fully successful fallbacks. Non-successful results (auth
@ -519,6 +534,7 @@ public class NodeStreamingChatHelper {
&& fallbackResult.errorMessage() == null) {
if (healthTracker != null) healthTracker.recordSuccess(entry.providerId());
addToPool(entry.providerId());
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return fallbackResult;
}
if (healthTracker != null) healthTracker.recordFailure(entry.providerId());
@ -531,10 +547,19 @@ public class NodeStreamingChatHelper {
}
}
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult != null ? lastResult
: buildErrorResult("LLM 调用失败，已达最大重试次数", conversationId, phase);
}
/** D-6: log a structured performance summary for the LLM call phase. */
private void logPerfSummary(String phase, String conversationId, long startMs,
int llmCallCount, int retryCount, int failoverCount) {
long totalMs = System.currentTimeMillis() - startMs;
log.info("[{}] perf_summary: conversationId={} total_ms={} llm_call_count={} retry_count={} failover_count={}",
phase, conversationId, totalMs, llmCallCount, retryCount, failoverCount);
}
/**
* 单次流式调用尝试。
* @return StreamResult 如果成功/降级/不可重试；null 如果应该重试
@ -756,13 +781,17 @@ public class NodeStreamingChatHelper {
conversationId, phase, errorType);
}
// Rate limit / Server error: retry with reduced limit for rate-limit/server-error
boolean isRateLimitOrServerError = errorType == ErrorType.RATE_LIMIT || errorType == ErrorType.SERVER_ERROR;
int effectiveMaxRetries = isRateLimitOrServerError ? MAX_RETRIES_RATE_LIMIT : MAX_RETRIES;
if (attempt < effectiveMaxRetries && isRateLimitOrServerError) {
log.warn("[{}] Retryable error (attempt {}/{}, type={}): {}",
phase, attempt, effectiveMaxRetries, errorType, error.getMessage());
return null; // 返回 null 触发重试
// Rate limit / Server error: retryable, but with different budgets.
// RATE_LIMIT: cap at 2 retries then failover (RFC 06 D-2).
// SERVER_ERROR: keep full MAX_RETRIES — upstream flaps often self-heal.
if (errorType == ErrorType.RATE_LIMIT || errorType == ErrorType.SERVER_ERROR) {
int effectiveMaxRetries = (errorType == ErrorType.RATE_LIMIT)
? MAX_RETRIES_RATE_LIMIT : MAX_RETRIES;
if (attempt < effectiveMaxRetries) {
log.warn("[{}] Retryable error (attempt {}/{}, type={}): {}",
phase, attempt, effectiveMaxRetries, errorType, error.getMessage());
return null; // 返回 null 触发重试
}
}
// 不可重试或已耗尽重试

29
mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java
View File

@ -40,8 +40,10 @@ public class ToolExecutionExecutor {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
// JDK 21 virtual threads: no blocking stall for I/O-bound tools.
// Each tool invocation gets its own lightweight carrier thread.
// Named threads (matching HookDispatcher convention) for log traceability.
private static final ExecutorService TOOL_EXECUTOR =
Executors.newVirtualThreadPerTaskExecutor();
Executors.newThreadPerTaskExecutor(
Thread.ofVirtual().name("tool-executor-", 0).factory());
/**
* Legacy hardcoded unsafe set, kept as a fallback when no
@ -54,7 +56,21 @@ public class ToolExecutionExecutor {
"browser_use", "BrowserUseTool", "write_file", "edit_file"
);
/** 工具结果最大字符数（防止超长结果膨胀 ToolResponseMessage → 撑爆 LLM 上下文） */
/**
* Layer 1 — hard truncation cap applied to every tool result before it
* reaches ToolResultStorage (Layer 2 spill) or the LLM prompt.
*
* <p>Two-level budget chain (RFC-008 / RFC-06 D-5):
* <pre>
* raw tool result
* → truncateToolResult(..., MAX_TOOL_RESULT_CHARS=8000) // Layer 1: hard cap
* → persistIfOversized(..., perResultThresholdChars=16000) // Layer 2: spill to disk
* → enforceTurnBudget(..., perTurnBudgetChars=32000) // Layer 3: per-turn aggregate
* </pre>
* Layer 1 runs first and is intentionally kept at 8000 to prevent oversized
* results from inflating the prompt. Layers 2/3 thresholds are configured in
* {@link ToolResultProperties} and application.yml.
*/
private static final int MAX_TOOL_RESULT_CHARS = 8000;
/** 尾部错误模式检测 */
@ -376,6 +392,7 @@ public class ToolExecutionExecutor {
private void executePreparedCalls(List<PreparedToolCall> preparedCalls,
List<ToolResponseMessage.ToolResponse> allResponses,
List<GraphEventPublisher.GraphEvent> events) {
long execStartMs = System.currentTimeMillis();
if (!preparedCalls.isEmpty() && streamTracker != null) {
String conversationId = preparedCalls.get(0).conversationId;
String phase = classifyBatchPhase(preparedCalls);
@ -398,6 +415,14 @@ public class ToolExecutionExecutor {
executeParallelBatch(batch, allResponses, events);
}
}
// D-6: emit tool execution perf summary
long toolExecMs = System.currentTimeMillis() - execStartMs;
events.add(GraphEventPublisher.perfSummary("tool_execution", Map.of(
"tool_exec_ms", toolExecMs,
"tool_count", preparedCalls.size(),
"batch_count", batches.size()
)));
}
/**

13
mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java
View File

@ -38,10 +38,19 @@ 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. */
/**
* Layer 2 — a single tool result larger than this is spilled to disk.
* Note: Layer 1 hard truncation ({@code MAX_TOOL_RESULT_CHARS=8000} in
* {@link ToolExecutionExecutor}) runs before this threshold is evaluated,
* so only results that survive Layer 1 can trigger a spill.
*/
private int perResultThresholdChars = 16000; // was 4000 — prevents WebSearch spill-to-disk
/** Aggregate cap on combined response size in one tool turn. */
/**
* Layer 3 — aggregate cap on combined response size in one tool turn.
* After all tools complete, the largest non-spilled responses are spilled
* in turn until the cumulative size fits this budget.
*/
private int perTurnBudgetChars = 32000; // was 16000 — headroom for multi-tool turns
/** Number of leading characters kept inline as a preview after spilling. */

10
mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java
View File

@ -55,11 +55,19 @@ public class ToolResultStorage {
/** Cached at construction; refreshed lazily if the underlying list mutates (rare). */
private volatile java.util.Set<String> excludedToolsSnapshot;
/** D-6: monotonically increasing spill counter for observability. */
private final java.util.concurrent.atomic.AtomicLong spillCount = new java.util.concurrent.atomic.AtomicLong();
public ToolResultStorage(ToolResultProperties props) {
this.props = props;
this.excludedToolsSnapshot = props.excludedToolsSet();
}
/** D-6: current cumulative spill count (monotonically increasing). */
public long getSpillCount() {
return spillCount.get();
}
/**
* Returns true when {@code toolName} is in the configured exclusion list.
* Excluded tools (typically retrieval tools like {@code read_file}) are
@ -112,6 +120,8 @@ public class ToolResultStorage {
toolName, conversationId, ioe.getMessage());
return result;
}
long count = spillCount.incrementAndGet();
log.info("[ToolResultStorage] spill #{}: tool={} chars={} convId={}", count, toolName, result.length(), conversationId);
return buildPreview(result, toolName, file);
}

11
mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java
View File

@ -172,6 +172,7 @@ public class PlanGenerationNode implements NodeAction {
}
// Silent streaming call — structured JSON is parsed below; tokens are not forwarded to the client.
long triageStartMs = System.currentTimeMillis();
NodeStreamingChatHelper.StreamResult result = streamingHelper.streamCallSilent(
chatModel, prompt, conversationId, "plan_generation");
@ -189,9 +190,19 @@ public class PlanGenerationNode implements NodeAction {
}
}
long triageMs = System.currentTimeMillis() - triageStartMs;
String llmResponse = result.text();
log.info("[PlanGeneration] Triage completed in {}ms", triageMs);
log.debug("[PlanGeneration] LLM response: {}", llmResponse);
// D-6: emit triage perf summary
events.add(GraphEventPublisher.perfSummary("triage", Map.of(
"triage_ms", triageMs,
"prompt_tokens", result.promptTokens(),
"completion_tokens", result.completionTokens()
)));
String cleanedJson = cleanJsonResponse(llmResponse);
Map<String, Object> parsed = objectMapper.readValue(cleanedJson, new TypeReference<>() {});
boolean needsPlanning = Boolean.TRUE.equals(parsed.get("needs_planning"));