+ * 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
- * 异步处理原始材料消化事件。
+ * RFC-012 Change 1：
+ *
+ * RFC-012 Change 1：材料级并行，受 {@link WikiProperties#getMaxParallelRawMaterials()} 约束。
*/
public void processAllPending(Long kbId) {
List
+ * 可重试（一直重试直到成功）：网络抖动、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：
+ *
+ * 说明：关键字启发式在极少数场景可能误判（例如瞬时错误的 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;
+ *
*
* @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 {
/**
* 处理知识库中所有待处理的原始材料
+ *
+ *
+ * 其余（网络、超时、5xx、429 限流、偶发空响应）均视为瞬时，按指数退避持续重试。
+ *