🧠 图解记忆：用不可变版本构建，验证后原子切换，删除则全链路传播；点击图片可查看原图。
💡 答案要点
为什么版本管理是生产级 RAG 的必备能力？
文档更新场景：
- 用户上传了新版本合同，覆盖旧版本
- 知识库文章修改，但答案还是基于旧内容
- FAQ 更新，Bot 回答还是老答案
→ 如果索引不更新，RAG 会用过期知识回答，误导用户三种文档更新策略：
策略1：基于时间戳的增量更新（适合新闻、博客）
展开 Python 代码示例（37 行）
python
class TimeBasedIndexer:
"""基于文档修改时间戳的增量索引"""
def __init__(self, vector_store, embedding_model):
self.vector_store = vector_store
self.embedding_model = embedding_model
def sync_knowledge_base(self, docs_path: str):
"""
每次同步：
1. 扫描文档目录，记录文件修改时间
2. 对比上次同步记录，找出'新增/修改'的文件
3. 只对变化的文件重新计算 Embedding + 索引
"""
current_state = self._scan_files(docs_path)
previous_state = self._load_previous_state() # {"file": "mtime"}
# 找出增量
changed_files = {
f: mtime for f, mtime in current_state.items()
if f not in previous_state or previous_state[f] != mtime
}
deleted_files = set(previous_state.keys()) - set(current_state.keys())
# 删除已移除文档的索引
for f in deleted_files:
doc_id = self._get_doc_id(f)
self.vector_store.delete(doc_id)
# 只索引变化的文档
for f in changed_files:
chunks = self._parse_and_chunk(f)
vectors = self.embedding_model.encode(chunks)
self.vector_store.add(doc_id=self._get_doc_id(f), vectors=vectors)
self._save_state(current_state)策略2：全文检索 + 向量检索双保险（适合版本敏感场景）
展开 Python 代码示例（39 行）
python
class HybridIndexer:
"""向量检索 + 全文检索双保险，版本敏感场景"""
def __init__(self):
self.vector_store = ChromaDB() # 向量检索
self.fulltext_store = Elasticsearch() # 全文检索
self.metadata_store = SQLite() # 版本元数据
def index_document(self, doc_id: str, content: str, version: str):
# 1. 向量索引（用于语义检索）
vector = self.embedding_model.encode(content)
self.vector_store.upsert(doc_id, vector)
# 2. 全文索引（用于精确关键词检索 + 版本对比）
self.fulltext_store.index(doc_id, content, version=version)
# 3. 元数据记录版本（用于版本溯源）
self.metadata_store.put(doc_id, version=version, indexed_at=datetime.now())
def retrieve(self, query: str, user_version: str = None):
"""
检索时：
- 向量检索保证语义理解
- 全文检索保证版本最新
- 元数据校验版本一致性
"""
# 向量检索 Top-K
vector_results = self.vector_store.search(query, top_k=10)
# 过滤过期版本
if user_version:
fresh_results = []
for r in vector_results:
indexed_version = self.metadata_store.get(r.doc_id)["version"]
if indexed_version == user_version:
fresh_results.append(r)
return fresh_results
return vector_results策略3：事件驱动实时更新（适合高变更频率知识库）
python
# 用 Watchdog 监控文件系统变化，实时触发索引更新
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class KnowledgeBaseWatcher(FileSystemEventHandler):
"""监控知识库目录，文件变化自动触发索引更新"""
def __init__(self, indexer: TimeBasedIndexer):
self.indexer = indexer
def on_modified(self, event):
if event.is_directory:
return
if event.src_path.endswith(('.pdf', '.docx', '.md')):
# 延迟防抖（避免频繁触发）
self.debounce(event.src_path, delay_seconds=2)
def debounce(self, path, delay_seconds):
import threading
timer = threading.Timer(delay_seconds, self._do_index, args=[path])
timer.start()
def _do_index(self, path):
print(f"文档变更检测: {path}，触发索引更新")
self.indexer.sync_single_file(path)生产选型建议：
| 场景 | 推荐策略 | 原因 |
|---|---|---|
| 静态知识库（手册、文档） | 定时增量同步 | 变化少，小时级同步足够 |
| 高变更知识库（FAQ、新闻） | 事件驱动实时 | 变化频繁，需要秒级响应 |
| 版本敏感（合同、法务） | 全文+向量双保险 | 需要版本溯源和对比 |
面试话术：
"文档更新是生产 RAG 的高频痛点。我的经验是'分层更新'：变化频率低的用定时任务（每小时扫一次），变化高的用文件监控实时触发。最关键的是版本溯源——用户问'这份合同什么时候更新的'，你得能回答出来。我用了 SQLite 存元数据（版本号、更新时间），每次检索时校验版本，过期内容直接过滤掉。这是很多 demo 级 RAG 不会考虑但生产必须解决的问题。"