* 核心管线：将原始材料通过 LLM 消化为结构化 Wiki 页面。 * * @author MateClaw Team */ @Slf4j @Service @RequiredArgsConstructor public class WikiProcessingService { private final WikiKnowledgeBaseService kbService; private final WikiRawMaterialService rawService; private final WikiPageService pageService; private final WikiProperties properties; private final ModelConfigService modelConfigService; private final AgentGraphBuilder agentGraphBuilder; private final ObjectMapper objectMapper; private final WikiProgressBus progressBus; /** 并行 chunk / 材料处理执行器（JDK 21 虚拟线程）；Listener 跨包需要引用，故 public */ public static final ExecutorService WIKI_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); /** * RFC-012 M2 v2 UI v2：单 raw 的进度计数器，多个并行 chunk 的 {@code processChunkTwoPhase} * 共享同一份 atomic 计数，避免 6 个 chunk 各写各的 progress 字段时互相覆盖（导致 UI 永远 preparing）。 *
* 生命周期：{@code processRawMaterial} 入口 put，try/finally 出口 remove。 */ private static final class ProgressCounter { final AtomicInteger total = new AtomicInteger(0); final AtomicInteger done = new AtomicInteger(0); /** Page-level 失败计数（chunk 内单页 create / merge 抛异常）。 * 注：DuplicateKeyException 触发的 fallback-to-update 不算 failure， * 内容仍合入了同 slug page。仅 LLM 调用爆炸、JSON 解析失败、内容为空等真失败才递增。 */ final AtomicInteger failed = new AtomicInteger(0); final AtomicBoolean phaseBStarted = new AtomicBoolean(false); /** * 跨 chunk slug 抢占表：canonical slug → 第一个声明该概念的实际 slug。 *
* 解决 LLM 在并行 chunk 中给同一概念起不同 slug 拼写（按词分组 vs 按字分隔）的问题。
* 使用 {@link ConcurrentHashMap#computeIfAbsent} 实现原子抢占：先到的 chunk 把自己的
* slug 注册为 winner，后到的 chunk 看到 winner 后会把内容写入 winner 对应的 page。
*/
final ConcurrentHashMap
* RFC-012 Change 1：材料级并行，受 {@link WikiProperties#getMaxParallelRawMaterials()} 约束。
*/
public void processAllPending(Long kbId) {
List
* 阶段 A（route）：一次 LLM 调用决定要 create 哪些新页 + 要 update 哪些已有页（仅 slug 列表）。
* 输入小、输出短，单次稳定在 30s 内返回。
*
* 阶段 B（merge）：对 update 列表里的每个 slug 单独发 LLM 调用，输入只塞这一页的现有正文 + 当前
* chunk 文本，输出该页 merge 后的完整内容。每次调用单页规模，远不会触发 nginx 60s 超时。
*
* 新建页直接落库；merge 页因互不依赖，可在当前 chunk 的 virtual thread 内顺序处理（chunk 之间
* 已通过 maxParallelChunks Semaphore 拿到了并行度）。
*
* @return 创建+更新的页面数
*/
private int processChunkTwoPhase(WikiKnowledgeBaseEntity kb, WikiRawMaterialEntity raw,
String textContent, String existingPagesIndex) {
Long kbId = kb.getId();
Long rawId = raw.getId();
String configContent = kb.getConfigContent() != null ? kb.getConfigContent() : "";
String rawTitle = raw.getTitle();
// RFC-012 M2 v2 UI v2：取共享进度计数器（processRawMaterial 入口已 put）。
// 多 chunk 并行时所有 chunk 共享同一份 atomic 计数，避免互相覆盖把 UI 拉回 preparing。
ProgressCounter pc = progressCounters.get(rawId);
// ─── 阶段 A：路由 ───
String routeSystem = PromptLoader.loadPrompt("wiki/route-system");
String routeUserTemplate = PromptLoader.loadPrompt("wiki/route-user");
String routeUser = routeUserTemplate
.replace("{config}", configContent)
.replace("{existing_pages}", existingPagesIndex)
.replace("{raw_title}", rawTitle)
.replace("{raw_content}", textContent);
Prompt routePrompt = new Prompt(List.of(
new SystemMessage(routeSystem),
new UserMessage(routeUser)
));
String routeResponse = callLlmWithResilientRetry(routePrompt, "route chunk of raw=" + rawId);
JsonNode routeJson = parseJsonResponse(routeResponse);
if (routeJson == null) {
log.warn("[Wiki] Route phase: failed to parse JSON for kbId={}, rawId={}, responseLen={}, first200={}",
kbId, rawId, routeResponse != null ? routeResponse.length() : 0,
routeResponse != null ? routeResponse.substring(0, Math.min(200, routeResponse.length())) : "null");
return 0;
}
// RFC-012 follow-up #3：phase B 现在并行执行，计数必须是 atomic
AtomicInteger created = new AtomicInteger(0);
AtomicInteger updated = new AtomicInteger(0);
// ─── 收集 route 输出（仅 metadata，无 content） ───
List
* 输入仅几 KB（chunk 主题片段 + metadata + 已有页索引），输出仅一页 markdown，
* 单次调用稳稳 ≤ 60 秒，避免 nginx 60s 网关。
*
* 兜底：如果 slug 已存在（route 误判），改走 update 路径。
*
* @return true 表示成功 create 一页（或兜底 update 一页时返回 false 以让上层归到 update 计数）
*/
private boolean createOnePage(WikiKnowledgeBaseEntity kb, WikiRawMaterialEntity raw,
String chunkText, String existingPagesIndex, JsonNode meta) {
Long kbId = kb.getId();
Long rawId = raw.getId();
String slug = meta.path("slug").asText("");
String title = meta.path("title").asText("");
String summary = meta.path("summary").asText("");
if (slug.isBlank() || title.isBlank()) return false;
String configContent = kb.getConfigContent() != null ? kb.getConfigContent() : "";
String createSystem = PromptLoader.loadPrompt("wiki/create-page-system");
String createUserTemplate = PromptLoader.loadPrompt("wiki/create-page-user");
String createUser = createUserTemplate
.replace("{config}", configContent)
.replace("{existing_pages}", existingPagesIndex)
.replace("{page_slug}", slug)
.replace("{page_title}", title)
.replace("{page_summary}", summary)
.replace("{raw_title}", raw.getTitle())
.replace("{raw_content}", chunkText);
Prompt prompt = new Prompt(List.of(
new SystemMessage(createSystem),
new UserMessage(createUser)
));
String response = callLlmWithResilientRetry(prompt,
"create page slug=" + slug + " of raw=" + rawId);
JsonNode pageJson = parseJsonResponse(response);
if (pageJson == null) {
log.warn("[Wiki] Phase B create page slug='{}' returned unparseable JSON, skipping", slug);
return false;
}
String content = pageJson.path("content").asText("");
String pageSummary = pageJson.path("summary").asText("");
if (pageSummary.isBlank()) pageSummary = summary;
if (content.isBlank()) {
log.warn("[Wiki] Phase B create page slug='{}' returned blank content, skipping", slug);
return false;
}
// 兜底 0：跨拼写 canonical 匹配（DB 已有 page，但 slug 拼写不同）
// 覆盖场景：之前上传的 raw 已经创建了同概念 page，本次 LLM 给了不同拼写
WikiPageEntity existingByCanonical = pageService.findByCanonicalSlug(kbId, slug);
if (existingByCanonical != null && !existingByCanonical.getSlug().equals(slug)) {
String actualSlug = existingByCanonical.getSlug();
pageService.updatePageByAi(kbId, actualSlug, content, pageSummary, rawId);
log.info("[Wiki] Phase B create slug='{}' canonical-matches existing '{}', updated",
slug, actualSlug);
return false;
}
// 兜底 0.5：跨 chunk in-flight slug 抢占（同一 raw 的另一并发 chunk 已声明同概念）
// computeIfAbsent 是原子操作，先到的 chunk 把自己 slug 注册为 winner
ProgressCounter pcLocal = progressCounters.get(rawId);
String canonical = WikiPageService.canonicalSlug(slug);
if (pcLocal != null && !canonical.isEmpty()) {
// lambda 要求 effectively final，用 finalSlug 副本
final String routedSlug = slug;
String winnerSlug = pcLocal.slugClaims.computeIfAbsent(canonical, k -> routedSlug);
if (!winnerSlug.equals(slug)) {
// 另一 chunk 先 claim 了同 canonical，但用了不同 slug 拼写
WikiPageEntity winner = pageService.getBySlug(kbId, winnerSlug);
if (winner != null) {
// winner 已 INSERT 进 DB → 直接 update
pageService.updatePageByAi(kbId, winnerSlug, content, pageSummary, rawId);
log.info("[Wiki] Phase B create slug='{}' lost slug-claim race to '{}', updated",
slug, winnerSlug);
return false;
}
// winner claim 早于 INSERT（claim 是 in-memory，INSERT 是 DB IO）
// → 用 winnerSlug 继续走下面的 INSERT 路径，DuplicateKey fallback 会兜住实际 race
log.info("[Wiki] Phase B create slug='{}' redirects to in-flight winner '{}'",
slug, winnerSlug);
slug = winnerSlug;
}
}
// 兜底 1：如果 slug 已存在（route 误判 / 上一次成功 INSERT），走 update 而不是 create
WikiPageEntity existing = pageService.getBySlug(kbId, slug);
if (existing != null) {
pageService.updatePageByAi(kbId, slug, content, pageSummary, rawId);
log.info("[Wiki] Phase B create page slug='{}' done (updated existing)", slug);
return false; // 不计入 created
}
String sourceRawIds = "[" + rawId + "]";
try {
pageService.createPage(kbId, slug, title, content, pageSummary, sourceRawIds);
log.info("[Wiki] Phase B create page slug='{}' done (created)", slug);
return true;
} catch (org.springframework.dao.DuplicateKeyException e) {
// 兜底 2：select-then-create 在并发下不是原子操作。当 N 个 chunk 同时
// route 出相同 slug，只有第一个 INSERT 能成功，其余都会触发 H2/MySQL
// unique key violation。本次 chunk 的 LLM 输出仍有价值——降级为 update，
// 把内容合并进已存在的 page，而不是丢弃。
pageService.updatePageByAi(kbId, slug, content, pageSummary, rawId);
log.info("[Wiki] Phase B create page slug='{}' lost INSERT race -> updated existing", slug);
return false; // 不计入 created
}
}
/**
* RFC-012 M2 v2 — 阶段 B 单页 merge：把 chunk 文本合并进一个已有页面。
*
* 输入仅几 KB（该页现有 content + chunk 主题片段），输出仅一页 markdown。
*
* @return true 表示成功 update 一页
*/
private boolean mergeOnePage(WikiKnowledgeBaseEntity kb, WikiRawMaterialEntity raw,
String chunkText, String slug) {
Long kbId = kb.getId();
Long rawId = raw.getId();
WikiPageEntity existing = pageService.getBySlug(kbId, slug);
if (existing == null) {
// 兜底：跨拼写 canonical 匹配——LLM 给的 slug 在 DB 里找不到，
// 但 canonical 形式（去连字符）对得上某个已有 page（典型场景：
// route 输出 `zhong-yao-qi-qing-pei-wu`，DB 存 `zhongyao-qiqing-peiwu`）
existing = pageService.findByCanonicalSlug(kbId, slug);
if (existing != null && !existing.getSlug().equals(slug)) {
log.info("[Wiki] Phase B merge slug='{}' canonical-matches existing '{}', using canonical slug for LLM call",
slug, existing.getSlug());
slug = existing.getSlug();
} else {
log.warn("[Wiki] Phase B merge page slug='{}' planned for update but not found in DB (even by canonical), skipping", slug);
return false;
}
}
String configContent = kb.getConfigContent() != null ? kb.getConfigContent() : "";
String mergeSystem = PromptLoader.loadPrompt("wiki/merge-page-system");
String mergeUserTemplate = PromptLoader.loadPrompt("wiki/merge-page-user");
String mergeUser = mergeUserTemplate
.replace("{config}", configContent)
.replace("{page_slug}", existing.getSlug() != null ? existing.getSlug() : slug)
.replace("{page_title}", existing.getTitle() != null ? existing.getTitle() : "")
.replace("{page_last_updated_by}", existing.getLastUpdatedBy() != null ? existing.getLastUpdatedBy() : "ai")
.replace("{page_content}", existing.getContent() != null ? existing.getContent() : "")
.replace("{raw_title}", raw.getTitle())
.replace("{raw_content}", chunkText);
Prompt prompt = new Prompt(List.of(
new SystemMessage(mergeSystem),
new UserMessage(mergeUser)
));
String response = callLlmWithResilientRetry(prompt,
"merge page slug=" + slug + " of raw=" + rawId);
JsonNode mergeJson = parseJsonResponse(response);
if (mergeJson == null) {
log.warn("[Wiki] Phase B merge page slug='{}' returned unparseable JSON, skipping", slug);
return false;
}
String content = mergeJson.path("content").asText("");
String summary = mergeJson.path("summary").asText("");
if (content.isBlank()) {
log.warn("[Wiki] Phase B merge page slug='{}' returned blank content, skipping", slug);
return false;
}
pageService.updatePageByAi(kbId, slug, content, summary, rawId);
log.info("[Wiki] Phase B merge page slug='{}' done", slug);
return true;
}
/**
* 解析 LLM 响应并创建/更新 Wiki 页面
*
* @return 创建+更新的页面总数
*/
private int applyLlmResponse(Long kbId, Long rawId, String llmResponse) {
JsonNode root = parseJsonResponse(llmResponse);
if (root == null) {
log.warn("[Wiki] Failed to parse LLM response for kbId={}, rawId={}, responseLen={}, first200={}",
kbId, rawId, llmResponse != null ? llmResponse.length() : 0,
llmResponse != null ? llmResponse.substring(0, Math.min(200, llmResponse.length())) : "null");
return 0;
}
// 结构校验：必须有 pages 数组
if (!root.has("pages") || !root.get("pages").isArray()) {
log.warn("[Wiki] LLM response missing 'pages' array for kbId={}, rawId={}", kbId, rawId);
return 0;
}
String sourceRawIds = "[" + rawId + "]";
int created = 0;
int updated = 0;
// 新页面
JsonNode pagesNode = root.path("pages");
if (pagesNode.isArray()) {
for (JsonNode pageNode : pagesNode) {
String slug = pageNode.path("slug").asText("");
String title = pageNode.path("title").asText("");
String content = pageNode.path("content").asText("");
String summary = pageNode.path("summary").asText("");
if (slug.isBlank() || title.isBlank()) continue;
// 检查是否已存在（LLM 可能将已有页面误判为新页面）
WikiPageEntity existing = pageService.getBySlug(kbId, slug);
if (existing != null) {
pageService.updatePageByAi(kbId, slug, content, summary, rawId);
updated++;
} else {
pageService.createPage(kbId, slug, title, content, summary, sourceRawIds);
created++;
}
}
}
// 更新的页面
JsonNode updatedPagesNode = root.path("updated_pages");
if (updatedPagesNode.isArray()) {
for (JsonNode pageNode : updatedPagesNode) {
String slug = pageNode.path("slug").asText("");
String content = pageNode.path("content").asText("");
String summary = pageNode.path("summary").asText("");
if (slug.isBlank()) continue;
WikiPageEntity existing = pageService.getBySlug(kbId, slug);
if (existing != null) {
// 保护手动编辑的页面：仍然更新，但 LLM 已在 prompt 中被告知要保留手动内容
pageService.updatePageByAi(kbId, slug, content, summary, rawId);
updated++;
}
}
}
log.info("[Wiki] Applied LLM response: kbId={}, rawId={}, created={}, updated={}",
kbId, rawId, created, updated);
return created + updated;
}
/**
* 构建已有 Wiki 页面索引（供 LLM 参考）
*/
private String buildExistingPagesIndex(Long kbId) {
List
* 可重试（一直重试直到成功）：网络抖动、5xx、429 限流、超时、连接中断、内容过滤偶发、JSON 空输出。
*
* 立即终止（模型不可用）：401/403 认证失败、模型不存在、quota 用尽、非法 API key、
* InterruptedException（优雅关停）。
*
* 使用指数退避（1s → 2s → 4s → ... → 封顶 60s）。
*
* RFC-012 M1：加入 maxAttempts 与 maxTotalDurationMs 双重上限，避免 nginx 504 这种
* 反复瞬时错误把单 chunk 卡到永远；buildChatModel 提到循环外，所有重试复用同一实例。
*/
private String callLlmWithResilientRetry(Prompt prompt, String ctx) {
long backoffMs = 1000;
final long maxBackoffMs = 60_000;
final int maxAttempts = Math.max(1, properties.getLlmMaxAttempts());
final long maxTotalDurationMs = Math.max(1_000L, properties.getLlmMaxTotalDurationMs());
final long startNanos = System.nanoTime();
final ChatModel chatModel = buildChatModel();
int attempt = 0;
while (true) {
attempt++;
try {
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);
}
String rootInfo = summarizeRoot(t);
if (isFatalModelError(t)) {
log.error("[Wiki] LLM unavailable (fatal) for {} after {} attempts (rootCause={}): {}",
ctx, attempt, rootInfo, t.getMessage());
throw new RuntimeException("LLM unavailable (rootCause=" + rootInfo + "): " + t.getMessage(), t);
}
long elapsedMs = (System.nanoTime() - startNanos) / 1_000_000L;
if (attempt >= maxAttempts || elapsedMs >= maxTotalDurationMs) {
log.error("[Wiki] LLM exhausted for {} after {} attempts in {}ms (limits: maxAttempts={}, maxTotalDurationMs={}, rootCause={}): {}",
ctx, attempt, elapsedMs, maxAttempts, maxTotalDurationMs, rootInfo, t.getMessage());
throw new RuntimeException("LLM exhausted after " + attempt + " attempts in " + elapsedMs
+ "ms (rootCause=" + rootInfo + "): " + t.getMessage(), t);
}
long sleepMs = Math.min(backoffMs, Math.max(0L, maxTotalDurationMs - elapsedMs));
log.warn("[Wiki] LLM transient failure for {} attempt={}/{} elapsed={}ms, retrying in {}ms (rootCause={}): {}",
ctx, attempt, maxAttempts, elapsedMs, sleepMs, rootInfo, t.getMessage());
try {
Thread.sleep(sleepMs);
} 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) {
// 按异常类型直接判 fatal —— DNS / 连接拒绝 / TLS 问题重试都是浪费
String className = cur.getClass().getSimpleName();
if ("UnknownHostException".equals(className)
|| "SSLHandshakeException".equals(className)
|| "CertificateException".equals(className)
|| "SSLPeerUnverifiedException".equals(className)) {
return true;
}
if ("ConnectException".equals(className) && cur.getMessage() != null
&& cur.getMessage().toLowerCase().contains("refused")) {
return true;
}
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;
}
// 基础设施类永久错误：关键字兜底（和类名判断互补，跨语言 SDK 也能抓到）
if (m.contains("unknown host") || m.contains("no such host")
|| m.contains("connection refused")
|| m.contains("pkix path building failed")
|| m.contains("certificate verify failed")
|| m.contains("certificate_unknown")
|| m.contains("ssl handshake")) {
return true;
}
}
cur = cur.getCause();
depth++;
}
return false;
}
/**
* 沿 getCause() 遍历到最深，返回根因异常的 "SimpleName: message" 形式。
*
* Spring RestClient 会把 HTTP 层异常包装成 ResourceAccessException，外层消息统一是
* "I/O error on POST request for ...:
* 拼进最终抛出的 RuntimeException 消息里，UI 就算截断也能在前几十字看见类名。
*/
private String summarizeRoot(Throwable t) {
Throwable cur = t;
int depth = 0;
while (cur != null && cur.getCause() != null && cur.getCause() != cur && depth < 8) {
cur = cur.getCause();
depth++;
}
String cls = cur != null ? cur.getClass().getSimpleName() : "Unknown";
String msg = cur != null ? cur.getMessage() : null;
if (msg == null) return cls;
// 截短 message 避免把整段 HTML 错误页塞进异常链
String trimmed = msg.replaceAll("\\s+", " ").trim();
if (trimmed.length() > 200) trimmed = trimmed.substring(0, 200) + "...";
return cls + ": " + trimmed;
}
/** 瞬时错误的内部标记异常，确保空响应也能走重试路径 */
private static class TransientLlmException extends RuntimeException {
TransientLlmException(String msg) { super(msg); }
}
private JsonNode parseJsonResponse(String response) {
if (response == null || response.isBlank()) return null;
String cleaned = response.trim();
// 1. 剥离 markdown 代码块标记
if (cleaned.startsWith("```json")) {
cleaned = cleaned.substring(7);
} else if (cleaned.startsWith("```")) {
cleaned = cleaned.substring(3);
}
if (cleaned.endsWith("```")) {
cleaned = cleaned.substring(0, cleaned.length() - 3);
}
cleaned = cleaned.trim();
// 2. 清洗控制字符（保留 \n \r \t），防止 LLM 输出含不可见字符导致 JSON 解析失败
cleaned = cleaned.replaceAll("[\\x00-\\x08\\x0B\\x0C\\x0E-\\x1F]", "");
// 3. 第一次尝试直接解析
try {
return objectMapper.readTree(cleaned);
} catch (Exception e) {
// 4. 如果整体不是 JSON，尝试提取第一个 JSON 对象块（LLM 可能在 JSON 前后加了说明文字）
int jsonStart = cleaned.indexOf("{");
int jsonEnd = cleaned.lastIndexOf("}");
if (jsonStart >= 0 && jsonEnd > jsonStart) {
String extracted = cleaned.substring(jsonStart, jsonEnd + 1);
try {
return objectMapper.readTree(extracted);
} catch (Exception e2) {
log.warn("[Wiki] Failed to parse extracted JSON block: {}", e2.getMessage());
}
}
log.warn("[Wiki] Failed to parse JSON response: {}", e.getMessage());
return null;
}
}
}
*
* 其余（网络、超时、5xx、429 限流、偶发空响应）均视为瞬时，按指数退避持续重试。
*