mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(wiki): PR-1a infra — content hash split, chunk metadata columns, kb-default model
This commit is contained in:
matevip
parent
49a3ebafcf
commit
50d9ff2b3d
@ -0,0 +1,26 @@
|
||||
package vip.mate.wiki.dto;
|
||||
|
||||
/**
|
||||
* RFC-051 PR-1a: a chunk-to-be along with the structural metadata produced by
|
||||
* {@code DocumentPreprocessService} / {@code WikiContentNormalizer}.
|
||||
* <p>
|
||||
* This carries everything {@link vip.mate.wiki.service.WikiChunkService} needs
|
||||
* to persist a chunk row, including the new metadata columns added in V39:
|
||||
* {@code page_number}, {@code token_count}, {@code header_breadcrumb},
|
||||
* {@code source_section}. The structural fields are nullable because not every
|
||||
* source format yields each piece of metadata (e.g. plain text has no page
|
||||
* number; HTML has no slide id).
|
||||
* <p>
|
||||
* No production callers in PR-1a — wiring lands in PR-1c. The type is added
|
||||
* here so the persistence overload in PR-1a can be unit-tested independently.
|
||||
*/
|
||||
public record WikiChunkDraft(
|
||||
String content,
|
||||
int startOffset,
|
||||
int endOffset,
|
||||
Integer pageNumber,
|
||||
Integer tokenCount,
|
||||
String headerBreadcrumb,
|
||||
String sourceSection
|
||||
) {
|
||||
}
|
||||
@ -0,0 +1,77 @@
|
||||
package vip.mate.wiki.job;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
import vip.mate.wiki.repository.WikiChunkMapper;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* RFC-051 PR-1a: low-frequency backfill job that fills the new
|
||||
* {@code token_count} column on existing chunks. Runs in the background after
|
||||
* the V39 migration adds the column with all-NULL data.
|
||||
* <p>
|
||||
* Strategy:
|
||||
* <ul>
|
||||
* <li>Pick at most {@value #BATCH_SIZE} chunks where {@code token_count IS NULL}.</li>
|
||||
* <li>Estimate tokens as {@code ceil(charCount / 4.0)} — a rough approximation
|
||||
* that PR-1c will replace with a real tokenizer once Tika and the
|
||||
* preprocessing pipeline are in place.</li>
|
||||
* <li>Persist and stop. Subsequent ticks pick up the next batch.</li>
|
||||
* <li>Any failure is logged at {@code warn} and swallowed so the main
|
||||
* ingest pipeline is never affected.</li>
|
||||
* </ul>
|
||||
*
|
||||
* The job is scheduled at a half-hour cron so a fresh upgrade does not
|
||||
* thrash the database.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class WikiChunkTokenBackfillJob {
|
||||
|
||||
private static final int BATCH_SIZE = 500;
|
||||
|
||||
private final WikiChunkMapper chunkMapper;
|
||||
|
||||
@Async
|
||||
@Scheduled(cron = "${mate.wiki.chunk-token-backfill-cron:0 */30 * * * ?}")
|
||||
public void runOnce() {
|
||||
try {
|
||||
List<WikiChunkEntity> batch = chunkMapper.selectList(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.isNull(WikiChunkEntity::getTokenCount)
|
||||
.last("LIMIT " + BATCH_SIZE));
|
||||
if (batch.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
int updated = 0;
|
||||
for (WikiChunkEntity chunk : batch) {
|
||||
Integer chars = chunk.getCharCount();
|
||||
if (chars == null || chars <= 0) {
|
||||
chunk.setTokenCount(0);
|
||||
} else {
|
||||
chunk.setTokenCount((int) Math.ceil(chars / 4.0));
|
||||
}
|
||||
try {
|
||||
chunkMapper.updateById(chunk);
|
||||
updated++;
|
||||
} catch (Exception inner) {
|
||||
log.warn("[WikiChunkTokenBackfill] Failed to update chunk={}: {}",
|
||||
chunk.getId(), inner.getMessage());
|
||||
}
|
||||
}
|
||||
log.info("[WikiChunkTokenBackfill] Backfilled token_count for {}/{} chunks",
|
||||
updated, batch.size());
|
||||
} catch (Exception e) {
|
||||
log.warn("[WikiChunkTokenBackfill] Backfill batch failed (will retry next tick): {}",
|
||||
e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -12,6 +12,25 @@ import java.util.Map;
|
||||
@Data
|
||||
public class WikiKbConfig {
|
||||
|
||||
/**
|
||||
* RFC-051: ingest pipeline mode for this KB. {@code "lazy"} skips page
|
||||
* generation on upload (chunk + embed only); {@code "eager"} runs the
|
||||
* legacy heavy ingest pipeline. {@code null} means caller should apply
|
||||
* its own default. PR-1a only adds the field; the lazy branch lands
|
||||
* in PR-1b.
|
||||
*/
|
||||
private String ingestMode;
|
||||
|
||||
/**
|
||||
* RFC-051: KB-level default chat model. Used by routing as the
|
||||
* intermediate fallback between {@link #stepModels} and the system
|
||||
* default. {@code null} means the caller should fall through to the
|
||||
* system default. The frontend already writes this field; before
|
||||
* PR-1a it had no Java field to deserialize into and was silently
|
||||
* dropped.
|
||||
*/
|
||||
private Long wikiDefaultModelId;
|
||||
|
||||
/** Per-step model overrides: "heavy_ingest.create_page" → modelId */
|
||||
private Map<String, Long> stepModels;
|
||||
|
||||
|
||||
@ -14,7 +14,7 @@ import vip.mate.wiki.model.WikiKnowledgeBaseEntity;
|
||||
* strong steps (CREATE_PAGE, MERGE_PAGE) use the default model.
|
||||
*/
|
||||
@Component
|
||||
@Order(2)
|
||||
@Order(3)
|
||||
@RequiredArgsConstructor
|
||||
public class GlobalDefaultStepModelStrategy implements WikiStepModelStrategy {
|
||||
|
||||
|
||||
@ -0,0 +1,49 @@
|
||||
package vip.mate.wiki.job.strategy;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.wiki.job.WikiJobStep;
|
||||
import vip.mate.wiki.job.WikiKbConfig;
|
||||
import vip.mate.wiki.job.model.WikiProcessingJobEntity;
|
||||
import vip.mate.wiki.model.WikiKnowledgeBaseEntity;
|
||||
|
||||
/**
|
||||
* RFC-051 PR-1a: middle-priority strategy that resolves the KB-level default
|
||||
* chat model ({@link WikiKbConfig#getWikiDefaultModelId()}). Sits between the
|
||||
* per-step override ({@link KbConfigStepModelStrategy}, Order 1) and the
|
||||
* system-wide default ({@link GlobalDefaultStepModelStrategy}, Order 3),
|
||||
* yielding the chain prescribed by RFC-051 §10.1:
|
||||
*
|
||||
* <pre>
|
||||
* stepModels[step] -> wikiDefaultModelId -> system default
|
||||
* </pre>
|
||||
*
|
||||
* The frontend has long written {@code wikiDefaultModelId} into the KB config
|
||||
* JSON, but no Java code consumed it before this strategy existed.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@Order(2)
|
||||
@RequiredArgsConstructor
|
||||
public class KbDefaultModelStrategy implements WikiStepModelStrategy {
|
||||
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
@Override
|
||||
public boolean supports(WikiJobStep step) { return true; }
|
||||
|
||||
@Override
|
||||
public Long selectModelId(WikiProcessingJobEntity job, WikiKnowledgeBaseEntity kb, WikiJobStep step) {
|
||||
if (kb == null || kb.getConfigContent() == null) return null;
|
||||
try {
|
||||
WikiKbConfig config = objectMapper.readValue(kb.getConfigContent(), WikiKbConfig.class);
|
||||
return config.getWikiDefaultModelId();
|
||||
} catch (Exception e) {
|
||||
log.debug("[KbDefaultModelStrategy] Failed to parse KB config: {}", e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -51,6 +51,18 @@ public class WikiChunkEntity {
|
||||
/** RFC-011：生成该 embedding 的模型名称（切模型时需全量重嵌） */
|
||||
private String embeddingModel;
|
||||
|
||||
/** RFC-051: source page number (PDF/PPTX) when known; null otherwise. */
|
||||
private Integer pageNumber;
|
||||
|
||||
/** RFC-051: estimated token count; null until populated by chunker or backfill job. */
|
||||
private Integer tokenCount;
|
||||
|
||||
/** RFC-051: header path leading to this chunk, e.g. "Intro / Setup / Linux". */
|
||||
private String headerBreadcrumb;
|
||||
|
||||
/** RFC-051: short identifier of the source section (slide id, sheet name, heading). */
|
||||
private String sourceSection;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
|
||||
|
||||
@ -5,6 +5,7 @@ import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import vip.mate.wiki.dto.WikiChunkDraft;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
import vip.mate.wiki.repository.WikiChunkMapper;
|
||||
|
||||
@ -55,6 +56,125 @@ public class WikiChunkService {
|
||||
return reconcile(kbId, rawId, chunks, offsets, existing);
|
||||
}
|
||||
|
||||
/**
|
||||
* RFC-051 PR-1a: persist chunks along with their structural metadata
|
||||
* (page number, token count, header breadcrumb, source section).
|
||||
* <p>
|
||||
* Behaves like {@link #persistChunks(Long, Long, List, List)} — full insert
|
||||
* when no existing chunks for the raw, otherwise hash-based reconcile so
|
||||
* unchanged chunks keep their embeddings. Reconcile updates the metadata
|
||||
* columns on retained rows so re-running ingest with a richer normalizer
|
||||
* can fill in fields that were null on the previous pass.
|
||||
*
|
||||
* @param drafts chunks-to-persist with metadata (ordered, ordinal == index)
|
||||
* @return persisted chunk IDs (same order as {@code drafts})
|
||||
*/
|
||||
@Transactional
|
||||
public List<Long> persistChunks(Long kbId, Long rawId, List<WikiChunkDraft> drafts) {
|
||||
List<WikiChunkEntity> existing = listByRawId(rawId);
|
||||
|
||||
if (existing.isEmpty()) {
|
||||
return insertAllDrafts(kbId, rawId, drafts);
|
||||
}
|
||||
return reconcileDrafts(kbId, rawId, drafts, existing);
|
||||
}
|
||||
|
||||
private List<Long> insertAllDrafts(Long kbId, Long rawId, List<WikiChunkDraft> drafts) {
|
||||
List<Long> ids = new ArrayList<>(drafts.size());
|
||||
for (int i = 0; i < drafts.size(); i++) {
|
||||
WikiChunkDraft draft = drafts.get(i);
|
||||
String hash = computeHash(draft.content());
|
||||
WikiChunkEntity entity = buildEntityFromDraft(kbId, rawId, i, draft, hash);
|
||||
chunkMapper.insert(entity);
|
||||
ids.add(entity.getId());
|
||||
}
|
||||
log.info("[WikiChunk] Inserted {} drafts for raw={}", drafts.size(), rawId);
|
||||
return ids;
|
||||
}
|
||||
|
||||
private List<Long> reconcileDrafts(Long kbId, Long rawId, List<WikiChunkDraft> drafts,
|
||||
List<WikiChunkEntity> existing) {
|
||||
Map<Integer, WikiChunkEntity> oldByOrdinal = new HashMap<>();
|
||||
for (WikiChunkEntity e : existing) {
|
||||
oldByOrdinal.put(e.getOrdinal(), e);
|
||||
}
|
||||
|
||||
List<Long> resultIds = new ArrayList<>(drafts.size());
|
||||
Set<Long> retainedIds = new HashSet<>();
|
||||
int retained = 0, rebuilt = 0;
|
||||
|
||||
for (int i = 0; i < drafts.size(); i++) {
|
||||
WikiChunkDraft draft = drafts.get(i);
|
||||
String hash = computeHash(draft.content());
|
||||
WikiChunkEntity old = oldByOrdinal.get(i);
|
||||
|
||||
if (old != null && hash.equals(old.getContentHash())) {
|
||||
// Same content: keep existing row (and its embedding). Refresh offsets and
|
||||
// metadata so a re-run with a smarter normalizer can fill gaps.
|
||||
boolean changed = false;
|
||||
if (!Objects.equals(old.getStartOffset(), draft.startOffset())) {
|
||||
old.setStartOffset(draft.startOffset()); changed = true;
|
||||
}
|
||||
if (!Objects.equals(old.getEndOffset(), draft.endOffset())) {
|
||||
old.setEndOffset(draft.endOffset()); changed = true;
|
||||
}
|
||||
if (!Objects.equals(old.getPageNumber(), draft.pageNumber())) {
|
||||
old.setPageNumber(draft.pageNumber()); changed = true;
|
||||
}
|
||||
if (!Objects.equals(old.getTokenCount(), draft.tokenCount())) {
|
||||
old.setTokenCount(draft.tokenCount()); changed = true;
|
||||
}
|
||||
if (!Objects.equals(old.getHeaderBreadcrumb(), draft.headerBreadcrumb())) {
|
||||
old.setHeaderBreadcrumb(draft.headerBreadcrumb()); changed = true;
|
||||
}
|
||||
if (!Objects.equals(old.getSourceSection(), draft.sourceSection())) {
|
||||
old.setSourceSection(draft.sourceSection()); changed = true;
|
||||
}
|
||||
if (changed) {
|
||||
chunkMapper.updateById(old);
|
||||
}
|
||||
resultIds.add(old.getId());
|
||||
retainedIds.add(old.getId());
|
||||
retained++;
|
||||
} else {
|
||||
WikiChunkEntity entity = buildEntityFromDraft(kbId, rawId, i, draft, hash);
|
||||
chunkMapper.insert(entity);
|
||||
resultIds.add(entity.getId());
|
||||
rebuilt++;
|
||||
}
|
||||
}
|
||||
|
||||
int deleted = 0;
|
||||
for (WikiChunkEntity old : existing) {
|
||||
if (!retainedIds.contains(old.getId()) && !resultIds.contains(old.getId())) {
|
||||
chunkMapper.deleteById(old.getId());
|
||||
deleted++;
|
||||
}
|
||||
}
|
||||
|
||||
log.info("[WikiChunk] Reconciled drafts raw={}: retained={}, rebuilt={}, deleted={}",
|
||||
rawId, retained, rebuilt, deleted);
|
||||
return resultIds;
|
||||
}
|
||||
|
||||
private WikiChunkEntity buildEntityFromDraft(Long kbId, Long rawId, int ordinal,
|
||||
WikiChunkDraft draft, String hash) {
|
||||
WikiChunkEntity entity = new WikiChunkEntity();
|
||||
entity.setKbId(kbId);
|
||||
entity.setRawId(rawId);
|
||||
entity.setOrdinal(ordinal);
|
||||
entity.setContent(draft.content());
|
||||
entity.setCharCount(draft.content().length());
|
||||
entity.setStartOffset(draft.startOffset());
|
||||
entity.setEndOffset(draft.endOffset());
|
||||
entity.setContentHash(hash);
|
||||
entity.setPageNumber(draft.pageNumber());
|
||||
entity.setTokenCount(draft.tokenCount());
|
||||
entity.setHeaderBreadcrumb(draft.headerBreadcrumb());
|
||||
entity.setSourceSection(draft.sourceSection());
|
||||
return entity;
|
||||
}
|
||||
|
||||
/**
|
||||
* 增量对账：比对 hash，保留不变的 chunk（保护未来的 embedding），重建变化的。
|
||||
*
|
||||
|
||||
@ -132,10 +132,13 @@ public class WikiRawMaterialService {
|
||||
entity.setFileSize(fileSize);
|
||||
entity.setProcessingStatus("pending");
|
||||
|
||||
// 计算文件内容 hash（用于上传去重）
|
||||
// Compute hash of original upload bytes (for dedup). RFC-051: hash raw bytes
|
||||
// directly — the previous `new String(bytes, UTF_8)` round-trip produced unstable
|
||||
// hashes for binary files (PDF/Office) because invalid UTF-8 sequences become
|
||||
// replacement characters, collapsing distinct files into the same hash.
|
||||
try {
|
||||
byte[] bytes = java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(sourcePath));
|
||||
entity.setContentHash(computeHash(new String(bytes, java.nio.charset.StandardCharsets.UTF_8)));
|
||||
entity.setContentHash(computeHashOfBytes(bytes));
|
||||
} catch (Exception e) {
|
||||
log.warn("[Wiki] Could not compute file hash for dedup: {}", e.getMessage());
|
||||
}
|
||||
@ -219,12 +222,21 @@ public class WikiRawMaterialService {
|
||||
rawMapper.updateById(entity);
|
||||
}
|
||||
|
||||
/**
|
||||
* Cache the extracted text for a raw material.
|
||||
* <p>
|
||||
* RFC-051: this method no longer touches {@code contentHash}. The previous
|
||||
* behavior overwrote the original-upload hash with an extracted-text hash,
|
||||
* which broke upload dedup (re-uploading the same file would compute a hash
|
||||
* over raw bytes but find a row whose hash had been replaced with extracted
|
||||
* text). The {@code contentHash} field is now an immutable identity for the
|
||||
* uploaded artifact; downstream short-circuiting uses {@code lastProcessedHash}.
|
||||
*/
|
||||
@Transactional
|
||||
public void updateExtractedText(Long id, String extractedText, String contentHash) {
|
||||
public void updateExtractedText(Long id, String extractedText) {
|
||||
WikiRawMaterialEntity entity = rawMapper.selectById(id);
|
||||
if (entity == null) return;
|
||||
entity.setExtractedText(extractedText);
|
||||
entity.setContentHash(contentHash);
|
||||
rawMapper.updateById(entity);
|
||||
}
|
||||
|
||||
@ -316,8 +328,8 @@ public class WikiRawMaterialService {
|
||||
log.warn("[Wiki] Extracted text truncated at {} chars for: {} (full document may be larger)",
|
||||
text.length(), entity.getSourcePath());
|
||||
} else {
|
||||
// 完整提取结果：缓存以避免重复提取
|
||||
updateExtractedText(entity.getId(), text, computeHash(text));
|
||||
// Full extraction: cache to avoid re-extracting on subsequent calls.
|
||||
updateExtractedText(entity.getId(), text);
|
||||
}
|
||||
log.info("[Wiki] Extracted text from {}: {} chars, method={}, truncated={}",
|
||||
entity.getSourcePath(), text.length(), json.getStr("method"), truncated);
|
||||
@ -401,4 +413,19 @@ public class WikiRawMaterialService {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* SHA-256 over raw bytes. Used for file uploads so that PDF/Office binaries
|
||||
* produce a stable identity hash regardless of UTF-8 round-tripping.
|
||||
*/
|
||||
private String computeHashOfBytes(byte[] bytes) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
byte[] hash = digest.digest(bytes);
|
||||
return HexFormat.of().formatHex(hash);
|
||||
} catch (Exception e) {
|
||||
log.warn("[Wiki] Failed to compute byte hash: {}", e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,7 @@
|
||||
-- V39: RFC-051 PR-1a — chunk structural metadata.
|
||||
-- All columns are nullable; NULL means "unknown" for legacy chunks. token_count
|
||||
-- backfill happens asynchronously via WikiChunkTokenBackfillJob.
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS page_number INT DEFAULT NULL;
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS token_count INT DEFAULT NULL;
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS header_breadcrumb VARCHAR(1024) DEFAULT NULL;
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS source_section VARCHAR(512) DEFAULT NULL;
|
||||
@ -0,0 +1,17 @@
|
||||
-- V39: RFC-051 PR-1a — chunk structural metadata.
|
||||
-- MySQL lacks `ADD COLUMN IF NOT EXISTS`; use INFORMATION_SCHEMA guard instead.
|
||||
SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_wiki_chunk' AND COLUMN_NAME = 'page_number');
|
||||
SET @s := IF(@c = 0, 'ALTER TABLE mate_wiki_chunk ADD COLUMN page_number INT DEFAULT NULL', 'SELECT 1');
|
||||
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
|
||||
|
||||
SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_wiki_chunk' AND COLUMN_NAME = 'token_count');
|
||||
SET @s := IF(@c = 0, 'ALTER TABLE mate_wiki_chunk ADD COLUMN token_count INT DEFAULT NULL', 'SELECT 1');
|
||||
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
|
||||
|
||||
SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_wiki_chunk' AND COLUMN_NAME = 'header_breadcrumb');
|
||||
SET @s := IF(@c = 0, 'ALTER TABLE mate_wiki_chunk ADD COLUMN header_breadcrumb VARCHAR(1024) DEFAULT NULL', 'SELECT 1');
|
||||
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
|
||||
|
||||
SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_wiki_chunk' AND COLUMN_NAME = 'source_section');
|
||||
SET @s := IF(@c = 0, 'ALTER TABLE mate_wiki_chunk ADD COLUMN source_section VARCHAR(512) DEFAULT NULL', 'SELECT 1');
|
||||
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
|
||||
Loading…
Reference in New Issue
Block a user