用Spring AI＋Qdrant实现可恢复的RAG增量索引服务

原创 24天前 1702

用Spring AI＋Qdrant实现可恢复的RAG增量索引服务

文章摘要

本文实现一个面向企业知识库的增量索引服务：通过文档Checksum和Chunk Hash识别变化，只对新增或修改片段生成Embedding；使用稳定Point ID写入Qdrant；通过版本状态实现新旧索引原子切换；失败任务可重试，删除操作可幂等执行。示例使用Spring Boot、Spring AI和Qdrant，重点展示工程结构、数据模型、任务状态和核心代码。

一、目标架构

上传或数据变更
→ 文档版本记录
→ 解析与分块
→ Chunk差异计算
→ Embedding
→ Qdrant写入READY
→ 完整性校验
→ 版本切换
→ 清理旧版本与缓存

目标：

未变化Chunk不重复Embedding；
新版本未完成前不影响生产；
任务失败可以恢复；
删除操作幂等；
所有Point可以追踪来源；
支持多租户和版本过滤。

二、项目结构

rag-indexing-service
├── domain
│   ├── KnowledgeDocument.java
│   ├── DocumentVersion.java
│   ├── KnowledgeChunk.java
│   └── IndexTask.java
├── application
│   ├── IncrementalIndexService.java
│   ├── DocumentParser.java
│   └── VersionSwitchService.java
├── infrastructure
│   ├── QdrantChunkRepository.java
│   ├── SpringAiEmbeddingGateway.java
│   └── JdbcDocumentRepository.java
└── web
    └── KnowledgeIndexController.java

三、核心数据模型

public enum VersionStatus {
    UPLOADED,
    PARSING,
    INDEXING,
    READY,
    EFFECTIVE,
    EXPIRED,
    FAILED
}

public record DocumentVersion(
        String versionId,
        String logicalDocumentId,
        long version,
        String fileChecksum,
        VersionStatus status,
        Instant createdAt
) {
}

Chunk：

public record KnowledgeChunk(
        String chunkId,
        String logicalDocumentId,
        String versionId,
        String sectionPath,
        int chunkIndex,
        String content,
        String contentHash,
        Map metadata
) {
}

四、稳定Point ID

不要每次生成随机UUID。

public String buildPointId(
        KnowledgeChunk chunk
) {
    String raw = String.join(
            ":",
            chunk.logicalDocumentId(),
            chunk.versionId(),
            chunk.sectionPath(),
            String.valueOf(chunk.chunkIndex())
    );

    return sha256(raw);
}

稳定ID便于：

Upsert；
删除；
重试；
引用；
一致性校验。

五、计算文件和Chunk Hash

public String sha256(String content) {
    try {
        MessageDigest digest =
                MessageDigest.getInstance("SHA-256");

        byte[] bytes = digest.digest(
                content.getBytes(StandardCharsets.UTF_8)
        );

        return HexFormat.of().formatHex(bytes);
    }
    catch (NoSuchAlgorithmException exception) {
        throw new IllegalStateException(exception);
    }
}

解析前先比较文件Hash：

相同
→ 跳过整个文档

不同
→ 解析并比较Chunk Hash

六、差异模型

public record ChunkDiff(
        List added,
        List modified,
        List unchanged,
        List removed
) {
}

比较逻辑：

public ChunkDiff diff(
        List oldChunks,
        List newChunks
) {
    Map oldByPath =
            oldChunks.stream()
                    .collect(Collectors.toMap(
                            this::logicalChunkKey,
                            Function.identity()
                    ));

    Map newByPath =
            newChunks.stream()
                    .collect(Collectors.toMap(
                            this::logicalChunkKey,
                            Function.identity()
                    ));

    // 根据逻辑位置与contentHash分类
    // 示例中省略集合拼装细节
    return calculateDiff(oldByPath, newByPath);
}

逻辑Chunk Key：

section_path + chunk_index

如果分块策略变化，应提升：

chunking_version

并执行全量重建，而不是错误复用旧向量。

七、Embedding网关

public interface EmbeddingGateway {

    List embed(List texts);
}

Spring AI实现：

@Component
public class SpringAiEmbeddingGateway
        implements EmbeddingGateway {

    private final EmbeddingModel embeddingModel;

    public SpringAiEmbeddingGateway(
            EmbeddingModel embeddingModel
    ) {
        this.embeddingModel = embeddingModel;
    }

    @Override
    public List embed(
            List texts
    ) {
        return texts.stream()
                .map(embeddingModel::embed)
                .toList();
    }
}

生产环境应批量调用并限制：

批大小；
并发；
超时；
重试；
Token；
单任务成本。

八、Qdrant Payload设计

{
  "tenant_id": "T001",
  "logical_document_id": "TRAVEL-POLICY",
  "version_id": "V4",
  "source_version": 4,
  "status": "READY",
  "section_path": "4.2",
  "content_hash": "...",
  "embedding_model": "...",
  "chunking_version": "v2"
}

生产检索只允许：

status = EFFECTIVE

九、增量索引核心服务

@Service
public class IncrementalIndexService {

    private final DocumentRepository documents;
    private final DocumentParser parser;
    private final EmbeddingGateway embeddings;
    private final ChunkVectorRepository vectors;
    private final VersionSwitchService switchService;

    @Transactional
    public IndexResult index(IndexCommand command) {
        DocumentVersion version =
                documents.createVersion(command);

        try {
            documents.updateStatus(
                    version.versionId(),
                    VersionStatus.PARSING
            );

            List newChunks =
                    parser.parse(command.file());

            List oldChunks =
                    documents.findEffectiveChunks(
                            command.logicalDocumentId()
                    );

            ChunkDiff diff = diff(oldChunks, newChunks);

            documents.updateStatus(
                    version.versionId(),
                    VersionStatus.INDEXING
            );

            indexChangedChunks(version, diff);

            verify(version, newChunks.size());

            documents.updateStatus(
                    version.versionId(),
                    VersionStatus.READY
            );

            switchService.activate(version);

            return IndexResult.success(
                    version.versionId(),
                    diff
            );
        }
        catch (RuntimeException exception) {
            documents.markFailed(
                    version.versionId(),
                    exception.getMessage()
            );
            throw exception;
        }
    }
}

真正项目中不要将长时间Embedding放在单个数据库事务内。上面主要展示流程，生产实现应使用任务状态和短事务。

十、批量写入Qdrant

private void indexChangedChunks(
        DocumentVersion version,
        ChunkDiff diff
) {
    List changed =
            Stream.concat(
                    diff.added().stream(),
                    diff.modified().stream()
            ).toList();

    for (
        List batch
        : BatchUtils.partition(changed, 64)
    ) {
        List vectorBatch =
                embeddings.embed(
                        batch.stream()
                                .map(KnowledgeChunk::content)
                                .toList()
                );

        vectors.upsertReady(
                version,
                batch,
                vectorBatch
        );
    }
}

写入状态先使用：

READY

不要直接设置EFFECTIVE。

十一、版本切换

@Service
public class VersionSwitchService {

    @Transactional
    public void activate(DocumentVersion version) {
        repository.expireCurrentVersion(
                version.logicalDocumentId()
        );

        repository.activateVersion(
                version.versionId()
        );

        vectorRepository.updateStatus(
                version.logicalDocumentId(),
                version.versionId(),
                "EFFECTIVE"
        );

        cacheVersionRepository.increment(
                version.logicalDocumentId()
        );
    }
}

理想情况下，数据库状态和向量Payload更新需要设计补偿机制，因为它们不能参与同一个本地事务。

可以采用：

状态机
＋Outbox
＋幂等补偿任务

十二、删除旧版本

版本切换后先逻辑失效：

旧版本 → EXPIRED

再异步清理：

public void cleanupExpired(
        String logicalDocumentId,
        Duration retention
) {
    List expiredVersions =
            repository.findExpiredBefore(
                    logicalDocumentId,
                    Instant.now().minus(retention)
            );

    for (String versionId : expiredVersions) {
        vectorRepository.deleteVersion(versionId);
    }
}

十三、失败恢复

索引任务记录：

task_id
version_id
stage
batch_no
retry_count
error_message
updated_at

恢复时从失败批次继续，而不是重跑全部文档。

重复写入依赖稳定Point ID，所以Upsert是幂等的。

十四、完整性校验

private void verify(
        DocumentVersion version,
        int expectedCount
) {
    long actual = vectors.countByVersion(
            version.versionId()
    );

    if (actual != expectedCount) {
        throw new IllegalStateException(
                "Chunk数量不一致：expected="
                + expectedCount
                + ", actual="
                + actual
        );
    }
}

还可以抽样验证：

文本Hash；
向量维度；
Payload字段；
检索结果；
引用映射。

十五、查询过滤

must:
  tenant_id = 当前租户
  status = EFFECTIVE

不要只按logical_document_id查询，也不要让前端直接传任意tenantId。

租户信息应来自认证上下文。

十六、接口设计

@RestController
@RequestMapping("/api/knowledge-index")
public class KnowledgeIndexController {

    private final IndexTaskService tasks;

    @PostMapping
    public IndexTaskResponse submit(
            @RequestBody IndexRequest request
    ) {
        return tasks.submit(request);
    }

    @GetMapping("/{taskId}")
    public IndexTaskStatus status(
            @PathVariable String taskId
    ) {
        return tasks.status(taskId);
    }
}

大文档使用异步任务，不要让HTTP连接等待整个Embedding过程。

十七、监控指标

index_task_success_rate
index_task_duration
changed_chunk_count
embedding_reuse_rate
embedding_batch_failure_count
index_version_switch_failure_count
ready_version_age
stale_effective_version_count
source_to_searchable_latency

十八、生产环境继续补齐

分布式锁；
同一文档并发更新；
任务队列；
断点续跑；
限流；
成本预算；
回归评测；
删除传播；
多Collection切换；
Embedding模型迁移。

总结

可恢复的RAG增量索引服务需要同时具备：

稳定ID
＋文件与Chunk Hash
＋差异计算
＋批量Embedding
＋READY/EFFECTIVE状态
＋原子版本切换
＋幂等重试
＋完整性验证

只做“新文件重新Embedding并Upsert”还不足以支撑企业知识库长期运行。

延伸阅读

如果你正在关注企业级 AI 应用、RAG、Agent、MCP 与大模型工程化落地，欢迎访问 智元界：

https://www.zyentor.com/

智元界将持续分享可运行的技术实战、架构设计、问题排查与企业应用案例。

持续研究设计方法手册 2026/07/19 08:00:25

确实，模型能力再强，落地的坑全在工程细节上。我们之前做金融场景的合规审计，光是把私有数据安全地喂进API就折腾了两个月，驻场团队能直接调底层接口，这效率确实没法比。不过有点好奇，OpenAI这么重的人力投入，会不会只是阶段性策略？等他们沉淀出标准化工具包，可能又会把咨询业务剥离出去。

一线深度学习观察室 2026/07/18 18:05:57

确实，部署才是真正的深水区，模型再强落不了地也是白搭。

运维频道 2026/07/17 22:29:40

确实，部署这块才是真正劝退企业的硬骨头。我们之前试过在医疗场景落地，模型准确率达标了，但数据合规和延迟问题折腾了三个月，OpenAI直接派工程师驻场等于把最头疼的环节包圆了。不过这种模式对中小企业成本会不会太高？毕竟模型蒸馏和私有化部署的定制化程度不低，最后可能还是只有大厂玩得起。

星河做实验记 2026/07/17 18:41:53

确实，部署环节的坑远比想象的多，我们做AI客服对接时，光是让模型适配不同部门的权限和数据格式就折腾了两个月，OpenAI这招等于把最脏最累的活标准化了。不过这么一来，以后中小企业可能更依赖他们的全栈方案，自主调优的空间反而被压缩了，是福是祸还真不好说。

用Spring AI＋Qdrant实现可恢复的RAG增量索引服务

用Spring AI＋Qdrant实现可恢复的RAG增量索引服务

文章摘要

一、目标架构

二、项目结构

三、核心数据模型

四、稳定Point ID

五、计算文件和Chunk Hash

六、差异模型

七、Embedding网关

八、Qdrant Payload设计

九、增量索引核心服务

十、批量写入Qdrant

十一、版本切换

十二、删除旧版本

十三、失败恢复

十四、完整性校验

十五、查询过滤

十六、接口设计

十七、监控指标

十八、生产环境继续补齐

总结

延伸阅读

评论 4

📖 关联推荐

精选推荐