RAG 增量更新方案:知识库变更后的索引重建最小化策略

RAG 增量更新方案:知识库变更后的索引重建最小化策略

一、"新增了 10 条 FAQ,重建索引花了 40 分钟"

知识库每周新增约 50 条内容。每次更新后需要重建整个向量索引——5000 条文档全部重新计算 Embedding,耗时 40 分钟。期间用户查询的还是旧索引,新增内容不可用。业务团队问:"能不能只更新新增的那 10 条?"

全量重建的问题在于:用 O(N) 的成本处理 O(1) 的变更。增量更新的目标就是:新增/修改/删除文档时,只处理变更的部分,其他文档的向量保持不变。

二、增量更新的三种策略

策略选择的考量:

  • 追加索引:最简单,直接 insert 新向量。大部分向量数据库原生支持。
  • 覆盖更新:先删后插。需要向量数据库支持按 ID 删除。
  • 标记删除 + 定期合并:删除不立即清理(节省计算),累积到阈值后再重建。

四、增量更新的工程考量

版本一致性。增量更新过程中,一个用户可能看到"部分新、部分旧"的检索结果。如果业务不能接受,需要在更新时加读锁或使用双缓冲(一个版本在读、另一个版本在写,写完切换)。

增量更新的累积碎片。频繁的增删会在向量索引中产生碎片——删除的向量占着位置但不可用,影响检索性能。需要定期做 Compaction(合并),类似数据库的 VACUUM 操作。

Embedding 模型变更时的处理。如果换了 Embedding 模型,新旧模型产生的向量不在同一个语义空间,必须全量重建。这是增量更新的一个硬边界——模型变了,所有向量都得重新算。

监控增量更新的延迟。新增文档从"进入知识库"到"可以被检索到"的时间差,是增量更新的核心 SLA。建议限制在 60 秒以内。

三、Go 实现:增量索引管理器

package incrementalrag import ( "context" "crypto/sha256" "fmt" "log" "sync" "time" ) // ========== 文档与向量 ========== type Document struct { ID string Content string Vector []float32 Status string // active, deleted, updating UpdatedAt time.Time Version int // 版本号,用于冲突检测 } // ========== 增量索引管理器 ========== type IncrementalIndexManager struct { vectorStore VectorStore embedder Embedder mu sync.RWMutex // 变更追踪 changeLog []ChangeRecord lastMergeAt time.Time // 配置 mergeThreshold float64 // 删除比例超过此值触发合并 } type ChangeRecord struct { DocID string ChangeType string // insert, update, delete Timestamp time.Time } type VectorStore interface { Upsert(ctx context.Context, docs []Document) error Delete(ctx context.Context, ids []string) error Search(ctx context.Context, vec []float32, topK int) ([]Document, error) Count(ctx context.Context) (int, error) GetDeletedCount(ctx context.Context) (int, error) } type Embedder interface { Embed(ctx context.Context, texts []string) ([][]float32, error) } func NewIncrementalIndexManager(vs VectorStore, emb Embedder) *IncrementalIndexManager { return &IncrementalIndexManager{ vectorStore: vs, embedder: emb, changeLog: make([]ChangeRecord, 0), lastMergeAt: time.Now(), mergeThreshold: 0.2, // 20% } } // InsertDocuments 增量添加文档 func (im *IncrementalIndexManager) InsertDocuments( ctx context.Context, docs []Document, ) error { if len(docs) == 0 { return nil } // 1. 批量计算 Embedding texts := make([]string, len(docs)) for i, doc := range docs { texts[i] = doc.Content } vectors, err := im.embedder.Embed(ctx, texts) if err != nil { return fmt.Errorf("计算 Embedding 失败: %w", err) } // 2. 注入向量 for i := range docs { docs[i].Vector = vectors[i] docs[i].Status = "active" docs[i].UpdatedAt = time.Now() docs[i].Version = 1 } // 3. 插入向量库 im.mu.Lock() defer im.mu.Unlock() if err := im.vectorStore.Upsert(ctx, docs); err != nil { return fmt.Errorf("插入向量库失败: %w", err) } // 4. 记录变更 for _, doc := range docs { im.changeLog = append(im.changeLog, ChangeRecord{ DocID: doc.ID, ChangeType: "insert", Timestamp: time.Now(), }) } log.Printf("[Incremental] 已添加 %d 篇文档", len(docs)) return nil } // UpdateDocument 增量更新文档 func (im *IncrementalIndexManager) UpdateDocument( ctx context.Context, doc Document, ) error { // 1. 计算新 Embedding vectors, err := im.embedder.Embed(ctx, []string{doc.Content}) if err != nil { return fmt.Errorf("计算 Embedding 失败: %w", err) } doc.Vector = vectors[0] doc.Status = "active" doc.UpdatedAt = time.Now() doc.Version++ // 2. Upsert(覆盖旧版本) im.mu.Lock() defer im.mu.Unlock() if err := im.vectorStore.Upsert(ctx, []Document{doc}); err != nil { return fmt.Errorf("更新向量库失败: %w", err) } // 3. 记录变更 im.changeLog = append(im.changeLog, ChangeRecord{ DocID: doc.ID, ChangeType: "update", Timestamp: time.Now(), }) return nil } // DeleteDocument 标记删除文档 func (im *IncrementalIndexManager) DeleteDocument( ctx context.Context, docID string, ) error { im.mu.Lock() defer im.mu.Unlock() // 标记删除(不在索引中移除,只是标记) if err := im.vectorStore.Delete(ctx, []string{docID}); err != nil { return fmt.Errorf("标记删除失败: %w", err) } im.changeLog = append(im.changeLog, ChangeRecord{ DocID: docID, ChangeType: "delete", Timestamp: time.Now(), }) // 检查是否需要合并 go im.checkAndMerge(context.Background()) return nil } // checkAndMerge 检查是否达到了合并阈值 func (im *IncrementalIndexManager) checkAndMerge(ctx context.Context) { im.mu.RLock() deletedCount, err := im.vectorStore.GetDeletedCount(ctx) totalCount, _ := im.vectorStore.Count(ctx) im.mu.RUnlock() if err != nil || totalCount == 0 { return } deletedRatio := float64(deletedCount) / float64(totalCount) if deletedRatio > im.mergeThreshold { log.Printf("[Incremental] 删除比例 %.1f%% > 阈值 %.1f%%, 触发合并", deletedRatio*100, im.mergeThreshold*100) im.ForceMerge(ctx) } } // ForceMerge 强制合并(清理被标记删除的向量) func (im *IncrementalIndexManager) ForceMerge(ctx context.Context) error { im.mu.Lock() defer im.mu.Unlock() log.Println("[Incremental] 开始全量合并...") start := time.Now() // 1. 导出所有 active 文档 // 2. 重建索引 // 3. 替换旧索引 // 实际实现依赖向量数据库的 compact/merge API // Milvus: collection.compact() // Qdrant: 通过 delete + vacuum 实现 im.lastMergeAt = time.Now() im.changeLog = im.changeLog[:0] // 清空变更日志 log.Printf("[Incremental] 合并完成, 耗时 %v", time.Since(start)) return nil } // ========== 变更日志的版本追踪 ========== // GetSince 获取指定时间之后的变更 func (im *IncrementalIndexManager) GetSince(ctx context.Context, since time.Time) []ChangeRecord { im.mu.RLock() defer im.mu.RUnlock() var changes []ChangeRecord for _, c := range im.changeLog { if c.Timestamp.After(since) { changes = append(changes, c) } } return changes } // ========== 内容哈希:检测内容是否真正变更 ========== func contentHash(content string) string { h := sha256.Sum256([]byte(content)) return fmt.Sprintf("%x", h[:16]) } func (im *IncrementalIndexManager) HasContentChanged( ctx context.Context, docID string, newContent string, ) (bool, error) { newHash := contentHash(newContent) // 查询旧版本的内容哈希 // 如果相同,跳过 Embedding 计算和更新 _ = docID _ = newHash return true, nil }

五、总结

增量更新的核心策略:新增追加(最低成本)、修改覆盖(计算新向量 + Upsert)、删除标记+定期合并(延迟清理成本)。关键指标是"变更可检索延迟"——目标应控制在 60 秒以内。实施建议:先从追加模式开始,等删除积累到影响检索性能时再上合并策略。一个容易被忽略的细节:变更日志要持久化到数据库或文件,进程重启后能恢复最后的位置,否则会丢失增量信息。另外,Embedding 模型升级是增量更新的硬边界——只要换了模型,就必须全量重建。把模型版本号写入索引元数据中,可以避免新旧向量混用导致的检索质量下降。