From b9ed8219ba10d7c3140da1ca480380d02ab88e51 Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 14 Apr 2026 15:18:03 +0800 Subject: [PATCH] feat(wiki): parallel processing + resilient LLM retry + hash-based skip (RFC-012 M1) --- .../java/vip/mate/wiki/WikiProperties.java | 26 ++- .../mate/wiki/controller/WikiController.java | 26 ++- .../wiki/event/WikiProcessingListener.java | 40 +++- .../wiki/model/WikiRawMaterialEntity.java | 3 + .../wiki/service/WikiProcessingService.java | 190 ++++++++++++++++-- .../wiki/service/WikiRawMaterialService.java | 11 + .../h2/V7__wiki_last_processed_hash.sql | 3 + .../mysql/V7__wiki_last_processed_hash.sql | 3 + 8 files changed, 268 insertions(+), 34 deletions(-) create mode 100644 mateclaw-server/src/main/resources/db/migration/h2/V7__wiki_last_processed_hash.sql create mode 100644 mateclaw-server/src/main/resources/db/migration/mysql/V7__wiki_last_processed_hash.sql diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java index 9aa9a51d..7b23dd04 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java @@ -15,8 +15,30 @@ public class WikiProperties { /** 是否启用 Wiki 知识库功能 */ private boolean enabled = true; - /** LLM 单次处理最大字符数（超过则分块） */ - private int maxChunkSize = 30000; + /** + * LLM 单次处理最大字符数（超过则分块）。 + *

+ * RFC-012：默认从 30000 下调到 15000 —— 单 chunk 输出 tokens 砍半、并行饱和度更高、 + * 质量也更稳定。中端模型（qwen-plus/claude-sonnet）建议 12000-20000， + * 旗舰模型（qwen-max/claude-opus）可调大到 20000-30000。 + */ + private int maxChunkSize = 15000; + + /** + * 同一次 processAllPending 下同时处理的原始材料数上限。 + *

+ * RFC-012 Change 1：保护共享的 @Async 线程池（max=16）不被 wiki 长时间占满， + * 同时给材料级并发定一个可控的上限，避免 LLM 提供方触发限流。 + */ + private int maxParallelRawMaterials = 3; + + /** + * 单个材料内 chunk 的并行处理数上限。 + *

+ * RFC-012 Change 1：从硬编码 3 提到 5，并暴露为配置项。默认总并发为 + * maxParallelRawMaterials × maxParallelChunks = 15，仍在常见 60 RPM 限额下。 + */ + private int maxParallelChunks = 5; /** 注入 agent prompt 的最大字符数 */ private int maxContextChars = 10000; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java index 8e3bb318..547d2fa9 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java @@ -240,15 +240,20 @@ public class WikiController { } @RequireWorkspaceRole("member") - @Operation(summary = "重新处理原始材料") + @Operation(summary = "重新处理原始材料（force=true 时绕过 content_hash 短路）") @PostMapping("/knowledge-bases/{kbId}/raw/{rawId}/reprocess") public R reprocessRaw(@PathVariable Long kbId, @PathVariable Long rawId, + @RequestParam(value = "force", defaultValue = "false") boolean force, @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { verifyKBWorkspace(kbId, workspaceId); WikiRawMaterialEntity raw = rawService.getById(rawId); if (raw == null || !kbId.equals(raw.getKbId())) { return R.fail("Raw material not found in this knowledge base"); } + // RFC-012 Change 5：force=true 时清空 last_processed_hash，让下一次处理必然执行完整管线 + if (force) { + rawService.setLastProcessedHash(rawId, null); + } rawService.reprocess(rawId); return R.ok(); } @@ -320,16 +325,27 @@ public class WikiController { // ==================== Processing ==================== @RequireWorkspaceRole("member") - @Operation(summary = "触发知识库处理（异步）") + @Operation(summary = "触发知识库处理（异步）；force=true 时清空所有 last_processed_hash 并重新入队全部材料") @PostMapping("/knowledge-bases/{kbId}/process") public R> processKB(@PathVariable Long kbId, + @RequestParam(value = "force", defaultValue = "false") boolean force, @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { verifyKBWorkspace(kbId, workspaceId); - List pending = rawService.listPending(kbId); - for (WikiRawMaterialEntity raw : pending) { + List targets; + if (force) { + // 强制重处理：所有非 pending 的材料重置为 pending，并清空 hash 短路 + targets = rawService.listByKbId(kbId); + for (WikiRawMaterialEntity r : targets) { + rawService.setLastProcessedHash(r.getId(), null); + rawService.reprocess(r.getId()); // reprocess 会把状态设为 pending 并发布事件 + } + return R.ok(Map.of("queued", targets.size(), "force", true)); + } + targets = rawService.listPending(kbId); + for (WikiRawMaterialEntity raw : targets) { eventPublisher.publishEvent(new WikiProcessingEvent(this, raw.getId(), kbId)); } - return R.ok(Map.of("queued", pending.size())); + return R.ok(Map.of("queued", targets.size(), "force", false)); } @RequireWorkspaceRole("viewer") diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiProcessingListener.java b/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiProcessingListener.java index b95cdc10..cf762ce2 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiProcessingListener.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiProcessingListener.java @@ -1,34 +1,60 @@ package vip.mate.wiki.event; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.context.event.EventListener; -import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; +import vip.mate.wiki.WikiProperties; import vip.mate.wiki.service.WikiProcessingService; +import java.util.concurrent.Semaphore; + /** * Wiki 处理事件监听器 *

- * 异步处理原始材料消化事件。 + * RFC-012 Change 1： + *

* * @author MateClaw Team */ @Slf4j @Component -@RequiredArgsConstructor public class WikiProcessingListener { private final WikiProcessingService processingService; + private final Semaphore rawMaterialSemaphore; + + public WikiProcessingListener(WikiProcessingService processingService, WikiProperties properties) { + this.processingService = processingService; + int parallel = Math.max(1, properties.getMaxParallelRawMaterials()); + this.rawMaterialSemaphore = new Semaphore(parallel); + log.info("[Wiki] Listener initialized with rawMaterialSemaphore permits={}", parallel); + } - @Async @EventListener public void onWikiProcessing(WikiProcessingEvent event) { log.info("[Wiki] Processing event received: rawId={}, kbId={}", event.getRawMaterialId(), event.getKbId()); + // 立即把实际处理派发到 wiki 的虚拟线程池，释放事件发布者线程和共享的 @Async 池 + WikiProcessingService.WIKI_EXECUTOR.submit(() -> runWithSemaphore(event.getRawMaterialId())); + } + + private void runWithSemaphore(Long rawId) { try { - processingService.processRawMaterial(event.getRawMaterialId()); + rawMaterialSemaphore.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("[Wiki] Interrupted while waiting for processing slot, rawId={}", rawId); + return; + } + try { + processingService.processRawMaterial(rawId); } catch (Exception e) { - log.error("[Wiki] Async processing failed for rawId={}: {}", event.getRawMaterialId(), e.getMessage(), e); + log.error("[Wiki] Async processing failed for rawId={}: {}", rawId, e.getMessage(), e); + } finally { + rawMaterialSemaphore.release(); } } } diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiRawMaterialEntity.java b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiRawMaterialEntity.java index a6e8d136..5010712b 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiRawMaterialEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiRawMaterialEntity.java @@ -49,6 +49,9 @@ public class WikiRawMaterialEntity { /** 上次处理时间 */ private LocalDateTime lastProcessedAt; + /** 上次成功处理时的 content_hash，用于重处理时的短路判断 */ + private String lastProcessedHash; + /** 错误信息 */ private String errorMessage; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java index 829a14d8..41a0ccd0 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java @@ -47,15 +47,23 @@ public class WikiProcessingService { private final AgentGraphBuilder agentGraphBuilder; private final ObjectMapper objectMapper; - /** 并行 chunk 处理执行器（JDK 21 虚拟线程） */ - private static final ExecutorService WIKI_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); - /** 最大并行 chunk 数 */ - private static final int MAX_PARALLEL_CHUNKS = 3; + /** 并行 chunk / 材料处理执行器（JDK 21 虚拟线程）；Listener 跨包需要引用，故 public */ + public static final ExecutorService WIKI_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); /** * 处理单个原始材料 */ public void processRawMaterial(Long rawId) { + processRawMaterial(rawId, false); + } + + /** + * 处理单个原始材料（支持强制重跑） + * + * @param rawId 材料 ID + * @param force 为 true 时忽略 content_hash 短路（RFC-012 Change 5），用于模型/提示词变更后的强制重跑 + */ + public void processRawMaterial(Long rawId, boolean force) { // CAS 式抢占：防止并发重复处理 if (!rawService.claimForProcessing(rawId)) { log.debug("[Wiki] Raw material {} already claimed or not pending, skipping", rawId); @@ -68,6 +76,15 @@ public class WikiProcessingService { return; } + // RFC-012 Change 5：若 content_hash 与上次成功处理时一致，直接短路 + if (!force + && raw.getContentHash() != null + && raw.getContentHash().equals(raw.getLastProcessedHash())) { + rawService.updateProcessingStatus(rawId, "completed", "Skipped: content unchanged since last processing"); + log.info("[Wiki] Skip reprocessing raw={} (content unchanged, hash={})", rawId, raw.getContentHash()); + return; + } + WikiKnowledgeBaseEntity kb = kbService.getById(raw.getKbId()); if (kb == null) { log.warn("[Wiki] Knowledge base not found for raw material: kbId={}", raw.getKbId()); @@ -115,8 +132,14 @@ public class WikiProcessingService { // 部分成功：有些 chunk 失败但有些产出了页面 rawService.updateProcessingStatus(rawId, "partial", failedChunks + " of " + totalChunks + " chunks failed, " + totalPages + " pages generated"); + // 【Review Bug 1】partial 不写 lastProcessedHash：partial 的语义就是"还有失败、需要再跑"， + // 写了会导致下次用户点"重新处理"被 hash 短路直接跳过，永远没机会修失败的 chunk。 } else { rawService.updateProcessingStatus(rawId, "completed", null); + // RFC-012 Change 5：记录本次成功处理时的 hash，供下次短路判断 + if (raw.getContentHash() != null) { + rawService.setLastProcessedHash(rawId, raw.getContentHash()); + } } int pageCount = pageService.countByKbId(kb.getId()); kbService.setPageCount(kb.getId(), pageCount); @@ -134,6 +157,8 @@ public class WikiProcessingService { /** * 处理知识库中所有待处理的原始材料 + *

+ * RFC-012 Change 1：材料级并行，受 {@link WikiProperties#getMaxParallelRawMaterials()} 约束。 */ public void processAllPending(Long kbId) { List pendingList = rawService.listPending(kbId); @@ -141,10 +166,29 @@ public class WikiProcessingService { log.info("[Wiki] No pending raw materials for kbId={}", kbId); return; } - log.info("[Wiki] Processing {} pending raw materials for kbId={}", pendingList.size(), kbId); + int parallel = Math.max(1, properties.getMaxParallelRawMaterials()); + log.info("[Wiki] Processing {} pending raw materials for kbId={} with parallelism={}", + pendingList.size(), kbId, parallel); + + Semaphore rawSem = new Semaphore(parallel); + List> futures = new ArrayList<>(pendingList.size()); for (WikiRawMaterialEntity raw : pendingList) { - processRawMaterial(raw.getId()); + final Long rawId = raw.getId(); + futures.add(CompletableFuture.runAsync(() -> { + try { + rawSem.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + try { + processRawMaterial(rawId); + } finally { + rawSem.release(); + } + }, WIKI_EXECUTOR)); } + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); } /** @@ -171,7 +215,8 @@ public class WikiProcessingService { } // Phase 2: 并行处理（Semaphore 限制并发数） - Semaphore semaphore = new Semaphore(MAX_PARALLEL_CHUNKS); + int parallelChunks = Math.max(1, properties.getMaxParallelChunks()); + Semaphore semaphore = new Semaphore(parallelChunks); AtomicInteger totalPages = new AtomicInteger(0); AtomicInteger failedChunks = new AtomicInteger(0); @@ -292,19 +337,12 @@ public class WikiProcessingService { .replace("{raw_title}", raw.getTitle()) .replace("{raw_content}", textContent); - // 调用 LLM - String llmResponse; - try { - ChatModel chatModel = buildChatModel(); - Prompt prompt = new Prompt(List.of( - new SystemMessage(systemPrompt), - new UserMessage(userPrompt) - )); - ChatResponse response = chatModel.call(prompt); - llmResponse = response.getResult().getOutput().getText(); - } catch (Exception e) { - throw new RuntimeException("LLM call failed: " + e.getMessage(), e); - } + // 调用 LLM（带无限重试，仅在模型不可用时终止） + Prompt prompt = new Prompt(List.of( + new SystemMessage(systemPrompt), + new UserMessage(userPrompt) + )); + String llmResponse = callLlmWithResilientRetry(prompt, "chunk of raw=" + raw.getId()); // 解析并持久化页面 return applyLlmResponse(kb.getId(), raw.getId(), llmResponse); @@ -408,6 +446,118 @@ public class WikiProcessingService { return agentGraphBuilder.buildRuntimeChatModel(defaultModel); } + /** + * 调用 LLM，带"任务完成或模型不可用才终止"的重试策略。 + *

+ * 可重试（一直重试直到成功）：网络抖动、5xx、429 限流、超时、连接中断、内容过滤偶发、JSON 空输出。 + *

+ * 立即终止（模型不可用）：401/403 认证失败、模型不存在、quota 用尽、非法 API key、 + * InterruptedException（优雅关停）。 + *

+ * 使用指数退避（1s → 2s → 4s → ... → 封顶 60s），无最大尝试次数。 + */ + private String callLlmWithResilientRetry(Prompt prompt, String ctx) { + long backoffMs = 1000; + final long maxBackoffMs = 60_000; + int attempt = 0; + while (true) { + attempt++; + try { + ChatModel chatModel = buildChatModel(); + ChatResponse response = chatModel.call(prompt); + if (response == null || response.getResult() == null + || response.getResult().getOutput() == null + || response.getResult().getOutput().getText() == null + || response.getResult().getOutput().getText().isBlank()) { + throw new TransientLlmException("Empty response from model"); + } + if (attempt > 1) { + log.info("[Wiki] LLM call for {} succeeded on attempt {}", ctx, attempt); + } + return response.getResult().getOutput().getText(); + } catch (Throwable t) { + if (Thread.currentThread().isInterrupted()) { + throw new RuntimeException("LLM call interrupted for " + ctx, t); + } + if (isFatalModelError(t)) { + log.error("[Wiki] LLM unavailable (fatal) for {} after {} attempts: {}", + ctx, attempt, t.getMessage()); + throw new RuntimeException("LLM unavailable: " + t.getMessage(), t); + } + log.warn("[Wiki] LLM transient failure for {} attempt={}, retrying in {}ms: {}", + ctx, attempt, backoffMs, t.getMessage()); + try { + Thread.sleep(backoffMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new RuntimeException("LLM retry interrupted for " + ctx, ie); + } + backoffMs = Math.min(maxBackoffMs, backoffMs * 2); + } + } + } + + /** + * 判断是否为"模型不可用"级别的致命错误（不重试，立即终止）。 + *

+ * 三类视为 fatal： + *

+ * 其余（网络、超时、5xx、429 限流、偶发空响应）均视为瞬时，按指数退避持续重试。 + *

+ * 说明：关键字启发式在极少数场景可能误判（例如瞬时错误的 message 恰好含 "authentication"）， + * 但实际云厂商 SDK 的错误消息规范度较高，这个风险可接受。 + */ + private boolean isFatalModelError(Throwable t) { + Throwable cur = t; + int depth = 0; + while (cur != null && depth < 8) { + String msg = cur.getMessage(); + if (msg != null) { + String m = msg.toLowerCase(); + // 鉴权 / 配额 / 模型不存在（HTTP 401/403 + 提供方错误字段） + if (m.contains("401") || m.contains("unauthorized") + || m.contains("403") || m.contains("forbidden") + || m.contains("invalid api key") || m.contains("invalid_api_key") + || m.contains("authentication") || m.contains("api key not valid") + || m.contains("model not found") || m.contains("model_not_found") + || m.contains("invalidapikey") || m.contains("invalid_request_error") + || m.contains("quota") || m.contains("insufficient_quota") + || m.contains("no default model") || m.contains("model configuration")) { + return true; + } + // 【Review Bug 3】prompt 结构性错误：重试也得同样结果，立即终止 + if (m.contains("context_length_exceeded") + || m.contains("context length") + || m.contains("maximum context") + || m.contains("max_tokens") + || m.contains("prompt too long") + || m.contains("input is too long") + || m.contains("token limit")) { + return true; + } + // 【Review Bug 2】内容审核过滤：被 safety 挡下的 prompt 重试也是同样结果 + if (m.contains("content_filter") + || m.contains("content filter") + || m.contains("data_inspection_failed") + || (m.contains("safety") && m.contains("block"))) { + return true; + } + } + cur = cur.getCause(); + depth++; + } + return false; + } + + /** 瞬时错误的内部标记异常，确保空响应也能走重试路径 */ + private static class TransientLlmException extends RuntimeException { + TransientLlmException(String msg) { super(msg); } + } + private JsonNode parseJsonResponse(String response) { if (response == null || response.isBlank()) return null; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java index a81e57f6..274eaa18 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java @@ -190,6 +190,17 @@ public class WikiRawMaterialService { rawMapper.updateById(entity); } + /** + * 记录本次成功处理时的 content_hash（RFC-012 Change 5 的短路依据）。 + */ + @Transactional + public void setLastProcessedHash(Long id, String hash) { + WikiRawMaterialEntity entity = rawMapper.selectById(id); + if (entity == null) return; + entity.setLastProcessedHash(hash); + rawMapper.updateById(entity); + } + /** * 重新处理：重置状态为 pending 并发布事件 */ diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V7__wiki_last_processed_hash.sql b/mateclaw-server/src/main/resources/db/migration/h2/V7__wiki_last_processed_hash.sql new file mode 100644 index 00000000..05dc0de2 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V7__wiki_last_processed_hash.sql @@ -0,0 +1,3 @@ +-- V7: Add last_processed_hash to mate_wiki_raw_material for skip-if-unchanged optimization +-- RFC-012 Change 5: when reprocessing, skip LLM pipeline if content_hash == last_processed_hash +ALTER TABLE mate_wiki_raw_material ADD COLUMN IF NOT EXISTS last_processed_hash VARCHAR(64) DEFAULT NULL; diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V7__wiki_last_processed_hash.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V7__wiki_last_processed_hash.sql new file mode 100644 index 00000000..05dc0de2 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V7__wiki_last_processed_hash.sql @@ -0,0 +1,3 @@ +-- V7: Add last_processed_hash to mate_wiki_raw_material for skip-if-unchanged optimization +-- RFC-012 Change 5: when reprocessing, skip LLM pipeline if content_hash == last_processed_hash +ALTER TABLE mate_wiki_raw_material ADD COLUMN IF NOT EXISTS last_processed_hash VARCHAR(64) DEFAULT NULL;