mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(wiki): parallel processing + resilient LLM retry + hash-based skip (RFC-012 M1)
This commit is contained in:
matevip
parent
aa439b6a3a
commit
b9ed8219ba
@ -15,8 +15,30 @@ public class WikiProperties {
|
||||
/** 是否启用 Wiki 知识库功能 */
|
||||
private boolean enabled = true;
|
||||
|
||||
/** LLM 单次处理最大字符数（超过则分块） */
|
||||
private int maxChunkSize = 30000;
|
||||
/**
|
||||
* LLM 单次处理最大字符数（超过则分块）。
|
||||
* <p>
|
||||
* RFC-012：默认从 30000 下调到 15000 —— 单 chunk 输出 tokens 砍半、并行饱和度更高、
|
||||
* 质量也更稳定。中端模型（qwen-plus/claude-sonnet）建议 12000-20000，
|
||||
* 旗舰模型（qwen-max/claude-opus）可调大到 20000-30000。
|
||||
*/
|
||||
private int maxChunkSize = 15000;
|
||||
|
||||
/**
|
||||
* 同一次 processAllPending 下同时处理的原始材料数上限。
|
||||
* <p>
|
||||
* RFC-012 Change 1：保护共享的 @Async 线程池（max=16）不被 wiki 长时间占满，
|
||||
* 同时给材料级并发定一个可控的上限，避免 LLM 提供方触发限流。
|
||||
*/
|
||||
private int maxParallelRawMaterials = 3;
|
||||
|
||||
/**
|
||||
* 单个材料内 chunk 的并行处理数上限。
|
||||
* <p>
|
||||
* RFC-012 Change 1：从硬编码 3 提到 5，并暴露为配置项。默认总并发为
|
||||
* maxParallelRawMaterials × maxParallelChunks = 15，仍在常见 60 RPM 限额下。
|
||||
*/
|
||||
private int maxParallelChunks = 5;
|
||||
|
||||
/** 注入 agent prompt 的最大字符数 */
|
||||
private int maxContextChars = 10000;
|
||||
|
||||
@ -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<Void> 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<Map<String, Object>> processKB(@PathVariable Long kbId,
|
||||
@RequestParam(value = "force", defaultValue = "false") boolean force,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
List<WikiRawMaterialEntity> pending = rawService.listPending(kbId);
|
||||
for (WikiRawMaterialEntity raw : pending) {
|
||||
List<WikiRawMaterialEntity> 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")
|
||||
|
||||
@ -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 处理事件监听器
|
||||
* <p>
|
||||
* 异步处理原始材料消化事件。
|
||||
* RFC-012 Change 1：
|
||||
* <ul>
|
||||
* <li>事件到达后立即提交到 wiki 自己的虚拟线程执行器，不占用全局 @Async 线程池（max=16）；</li>
|
||||
* <li>通过信号量限制同时处理的材料数 ({@link WikiProperties#getMaxParallelRawMaterials()})，
|
||||
* 避免大批量上传瞬间触发 LLM 限流。</li>
|
||||
* </ul>
|
||||
*
|
||||
* @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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -49,6 +49,9 @@ public class WikiRawMaterialEntity {
|
||||
/** 上次处理时间 */
|
||||
private LocalDateTime lastProcessedAt;
|
||||
|
||||
/** 上次成功处理时的 content_hash，用于重处理时的短路判断 */
|
||||
private String lastProcessedHash;
|
||||
|
||||
/** 错误信息 */
|
||||
private String errorMessage;
|
||||
|
||||
|
||||
@ -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 {
|
||||
|
||||
/**
|
||||
* 处理知识库中所有待处理的原始材料
|
||||
* <p>
|
||||
* RFC-012 Change 1：材料级并行，受 {@link WikiProperties#getMaxParallelRawMaterials()} 约束。
|
||||
*/
|
||||
public void processAllPending(Long kbId) {
|
||||
List<WikiRawMaterialEntity> 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<CompletableFuture<Void>> 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，带"任务完成或模型不可用才终止"的重试策略。
|
||||
* <p>
|
||||
* 可重试（一直重试直到成功）：网络抖动、5xx、429 限流、超时、连接中断、内容过滤偶发、JSON 空输出。
|
||||
* <p>
|
||||
* 立即终止（模型不可用）：401/403 认证失败、模型不存在、quota 用尽、非法 API key、
|
||||
* InterruptedException（优雅关停）。
|
||||
* <p>
|
||||
* 使用指数退避（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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断是否为"模型不可用"级别的致命错误（不重试，立即终止）。
|
||||
* <p>
|
||||
* 三类视为 fatal：
|
||||
* <ul>
|
||||
* <li><b>鉴权 / 配额 / 模型不存在</b>：401/403、invalid api key、model not found、quota 用尽</li>
|
||||
* <li><b>prompt 结构性错误</b>：上下文超长、max_tokens 限制、prompt too long（重试也得同样结果）</li>
|
||||
* <li><b>内容审核过滤</b>：content_filter 触发（被 safety 挡下的 prompt 重试也是同样结果）</li>
|
||||
* </ul>
|
||||
* 其余（网络、超时、5xx、429 限流、偶发空响应）均视为瞬时，按指数退避持续重试。
|
||||
* <p>
|
||||
* 说明：关键字启发式在极少数场景可能误判（例如瞬时错误的 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;
|
||||
|
||||
|
||||
@ -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 并发布事件
|
||||
*/
|
||||
|
||||
@ -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;
|
||||
@ -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;
|
||||
Loading…
Reference in New Issue
Block a user