知识库 5 分钟同步生效：文档增量更新、向量热替换与版本回滚的工程实现

运营管理后台
    │ 编辑/新增/删除文档
    ▼
知识库管理服务
    │ 检测变更类型
    ├── 新增文档 → 切分 + Embedding + 写入 Milvus
    ├── 修改文档 → 增量更新流程（核心）
    └── 删除文档 → 标记删除 + 异步清理
    │
    ▼
变更事件 → RocketMQ → 异步处理
    │
    ▼
向量热替换（双版本切换）
    │
    ▼
知识库版本记录（支持回滚）

@Service
public class KnowledgeDocumentService {
​
    public void saveDocument(KnowledgeDocument doc) {
        KnowledgeDocument existing = docMapper.selectById(doc.getId());
​
        if (existing == null) {
            // 新增文档
            doc.setVersion(1);
            doc.setContentHash(DigestUtils.md5Hex(doc.getContent()));
            docMapper.insert(doc);
            publishEvent(new DocChangeEvent(doc.getId(), ChangeType.CREATE));
        } else {
            String newHash = DigestUtils.md5Hex(doc.getContent());
            if (!newHash.equals(existing.getContentHash())) {
                // 内容确实发生了变化
                doc.setVersion(existing.getVersion() + 1);
                doc.setContentHash(newHash);
                docMapper.updateById(doc);
                publishEvent(new DocChangeEvent(doc.getId(), ChangeType.UPDATE));
            }
            // hash 相同说明内容没变（可能只改了标题等元数据），只更新元数据
        }
    }
​
    private void publishEvent(DocChangeEvent event) {
        rocketMQTemplate.syncSend("knowledge-doc-change", event);
    }
}

@RocketMQMessageListener(
    topic = "knowledge-doc-change",
    consumerGroup = "knowledge-update-consumer"
)
@Service
public class KnowledgeUpdateConsumer implements RocketMQListener<DocChangeEvent> {
​
    @Override
    public void onMessage(DocChangeEvent event) {
        log.info("收到知识库变更事件: docId={}, type={}", event.getDocId(), event.getChangeType());
​
        switch (event.getChangeType()) {
            case CREATE -> handleCreate(event.getDocId());
            case UPDATE -> handleUpdate(event.getDocId());
            case DELETE -> handleDelete(event.getDocId());
        }
    }
​
    /**
     * 文档修改：增量更新核心逻辑
     */
    private void handleUpdate(String docId) {
        long startTime = System.currentTimeMillis();
​
        // 1. 加载最新文档
        KnowledgeDocument doc = docMapper.selectById(docId);
​
        // 2. 切分为 chunks
        List<DocumentChunk> newChunks = documentSplitter.split(doc);
​
        // 3. 批量 Embedding
        List<float[]> embeddings = embeddingService.batchEmbed(
            newChunks.stream().map(DocumentChunk::getContent).toList()
        );
        for (int i = 0; i < newChunks.size(); i++) {
            newChunks.get(i).setEmbedding(embeddings.get(i));
            newChunks.get(i).setChunkId(generateChunkId(docId, i));
        }
​
        // 4. 向量热替换（原子操作）
        vectorStoreService.atomicReplace(docId, newChunks);
​
        // 5. 记录版本（支持回滚）
        versionService.recordVersion(docId, doc.getVersion(), newChunks);
​
        long elapsed = System.currentTimeMillis() - startTime;
        log.info("文档 {} 增量更新完成，新chunk数: {}, 耗时: {}ms", docId, newChunks.size(), elapsed);
    }
}

@Service
public class EmbeddingService {
​
    private static final int BATCH_SIZE = 25;
​
    /**
     * 批量 Embedding，自动分批处理
     */
    public List<float[]> batchEmbed(List<String> texts) {
        List<float[]> allEmbeddings = new ArrayList<>();
​
        // 按 BATCH_SIZE 分批
        for (int i = 0; i < texts.size(); i += BATCH_SIZE) {
            List<String> batch = texts.subList(i, Math.min(i + BATCH_SIZE, texts.size()));
​
            List<float[]> batchResult = embeddingModel.embed(batch);
            allEmbeddings.addAll(batchResult);
        }
​
        return allEmbeddings;
    }
}

@Service
public class VectorStoreService {
​
    /**
     * 原子替换某个文档的所有 chunks
     * 策略：先插入新版本 → 更新路由 → 再删除旧版本
     */
    public void atomicReplace(String docId, List<DocumentChunk> newChunks) {
        // 1. 查出旧版本的所有 chunk IDs
        List<String> oldChunkIds = getChunkIdsByDocId(docId);
​
        // 2. 插入新版本的 chunks（此时新旧共存）
        insertChunks(newChunks);
​
        // 3. 更新文档版本映射表（原子操作，此刻起检索会命中新版本）
        docVersionMapping.put(docId, newChunks.stream()
            .map(DocumentChunk::getChunkId)
            .toList());
​
        // 4. 异步删除旧版本 chunks（等待一段时间确保没有进行中的查询引用旧数据）
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(3000); // 等 3 秒，确保进行中的查询已完成
                deleteChunks(oldChunkIds);
                log.info("文档 {} 旧版本 {} 个chunks已清理", docId, oldChunkIds.size());
            } catch (Exception e) {
                log.error("清理旧chunks失败，docId={}", docId, e);
                // 清理失败不影响业务，下次更新时会再尝试清理
            }
        });
    }
}

private String generateChunkId(String docId, int chunkIndex) {
    int version = docMapper.selectById(docId).getVersion();
    return String.format("%s_v%d_c%d", docId, version, chunkIndex);
    // 例：doc_refund_policy_001_v3_c2
    // 文档 refund_policy_001 的第3个版本的第2个chunk
}

// 检索时只查最新版本的 chunks
String versionFilter = String.format("doc_version == %d", latestVersion);

CREATE TABLE `knowledge_doc_version` (
    `id` BIGINT PRIMARY KEY AUTO_INCREMENT,
    `doc_id` VARCHAR(64) NOT NULL,
    `version` INT NOT NULL,
    `content_snapshot` MEDIUMTEXT NOT NULL COMMENT '文档内容快照',
    `chunk_count` INT NOT NULL,
    `chunk_ids` TEXT NOT NULL COMMENT '该版本对应的chunk ID列表(JSON)',
    `operator` VARCHAR(64) COMMENT '操作人',
    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY `uk_doc_version` (`doc_id`, `version`)
);

@Service
public class KnowledgeVersionService {

    /**
     * 回滚到指定版本
     */
    public void rollback(String docId, int targetVersion) {
        // 1. 查出目标版本的文档快照
        KnowledgeDocVersion versionRecord = versionMapper.selectByDocVersion(docId, targetVersion);
        if (versionRecord == null) {
            throw new BusinessException("版本不存在");
        }

        // 2. 用快照内容覆盖当前文档
        KnowledgeDocument doc = docMapper.selectById(docId);
        doc.setContent(versionRecord.getContentSnapshot());
        doc.setVersion(doc.getVersion() + 1); // 回滚也是一次新版本
        doc.setContentHash(DigestUtils.md5Hex(doc.getContent()));
        docMapper.updateById(doc);

        // 3. 触发增量更新（和正常编辑走同一条链路）
        publishEvent(new DocChangeEvent(docId, ChangeType.UPDATE));

        log.info("文档 {} 回滚到版本 {}，新版本号 {}", docId, targetVersion, doc.getVersion());
    }
}

public void fullRebuild() {
    String newCollection = "knowledge_chunks_" + System.currentTimeMillis();

    // 1. 创建新 Collection
    milvusClient.createCollection(newCollection, schema);

    // 2. 遍历所有文档，切分 + Embedding + 写入新 Collection
    List<KnowledgeDocument> allDocs = docMapper.selectAll();
    for (KnowledgeDocument doc : allDocs) {
        List<DocumentChunk> chunks = documentSplitter.split(doc);
        List<float[]> embeddings = embeddingService.batchEmbed(
            chunks.stream().map(DocumentChunk::getContent).toList());
        // ... 写入 newCollection
    }

    // 3. 原子切换 Collection 别名
    milvusClient.createAlias(newCollection, "knowledge_active");
    milvusClient.dropCollection(oldCollection);

    log.info("全量重建完成，新Collection: {}", newCollection);
}