From 320e13b975770fa081ee821756dfd8cf317122ad Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 22 Apr 2026 10:13:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent):=20review=20fixes=20for=20Lane=20D?= =?UTF-8?q?=20=E2=80=94=20D-2=20strategy=20split,=20D-4=20naming,=20D-5=20?= =?UTF-8?q?docs,=20D-6=20instrumentation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../vip/mate/agent/GraphEventPublisher.java | 18 ++++++++ .../agent/graph/NodeStreamingChatHelper.java | 45 +++++++++++++++---- .../graph/executor/ToolExecutionExecutor.java | 29 +++++++++++- .../graph/executor/ToolResultProperties.java | 13 +++++- .../graph/executor/ToolResultStorage.java | 10 +++++ .../graph/plan/node/PlanGenerationNode.java | 11 +++++ 6 files changed, 114 insertions(+), 12 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java b/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java index 837e2a22..a398e6b9 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java @@ -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 metrics) { + long ts = System.currentTimeMillis(); + Map data = new java.util.HashMap<>(metrics); + data.put("phase", phase); + data.put("timestamp", ts); + return new GraphEvent(EVENT_PERF_SUMMARY, Map.copyOf(data), ts); + } + // ===== 提取方法 ===== /** diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index bbd205ce..2a0780a5 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -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 触发重试 + } } // 不可重试或已耗尽重试 diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java index c00c08b7..7cf06e7a 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java @@ -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. + * + *

Two-level budget chain (RFC-008 / RFC-06 D-5): + * 

+     *   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
+     *
+ * 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 preparedCalls, List allResponses, List 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() + ))); } /** diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java index 638e05a7..fe95415e 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java @@ -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. */ diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java index 07a5c3bf..98f7f38e 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java @@ -55,11 +55,19 @@ public class ToolResultStorage { /** Cached at construction; refreshed lazily if the underlying list mutates (rare). */ private volatile java.util.Set 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); } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java index ab7dbe8b..c900cef3 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java @@ -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 parsed = objectMapper.readValue(cleanedJson, new TypeReference<>() {}); boolean needsPlanning = Boolean.TRUE.equals(parsed.get("needs_planning"));