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