当前位置：首页 > news >正文

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

news 2026/7/16 17:42:25

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 模型升级是增量更新的硬边界——只要换了模型，就必须全量重建。把模型版本号写入索引元数据中，可以避免新旧向量混用导致的检索质量下降。

查看全文

http://www.jsqmd.com/news/1201447/