_{记忆点：版本化构建、原子切换、可验证、可回滚。}

💡 答案要点

为什么需要动态知识更新？

RAG 知识库不更新的问题：
  - 产品手册更新了，但 RAG 还在回答旧版本
  - 新政策发布，但 RAG 回答的是旧政策
  - 用户质量反馈：答案过时，导致客诉

三种更新策略：

策略一：批处理更新（Batch Update）

展开 Python 代码示例（40 行）

import schedule
from datetime import datetime

class BatchKnowledgeUpdater:
    def __init__(self, vector_store, doc_source):
        self.vector_store = vector_store
        self.doc_source = doc_source

    def full_rebuild(self):
        """全量重建（简单但停机时间长）"""
        # 蓝绿部署：先建新库，切换后删旧库
        new_collection = self.vector_store.create_collection(
            f"knowledge_v{datetime.now().strftime('%Y%m%d')}"
        )
        # 导入所有最新文档
        all_docs = self.doc_source.get_all_docs()
        new_collection.add_documents(all_docs)

        # 原子切换（无缝切换，零停机）
        self.vector_store.switch_active_collection(new_collection.name)
        # 保留旧库 24 小时作为回滚备份

    def incremental_update(self, changed_docs: list):
        """增量更新（只处理变化的文档）"""
        for doc in changed_docs:
            if doc["action"] == "add":
                self.vector_store.add_documents([doc])
            elif doc["action"] == "update":
                # 先删旧的，再插新的
                self.vector_store.delete(ids=[doc["id"]])
                self.vector_store.add_documents([doc])
            elif doc["action"] == "delete":
                self.vector_store.delete(ids=[doc["id"]])

# 每天凌晨 2 点增量更新
schedule.every().day.at("02:00").do(
    lambda: batch_updater.incremental_update(
        get_changed_docs_since_last_update()
    )
)

策略二：实时更新（Real-time Update）

展开 Python 代码示例（45 行）

from kafka import KafkaConsumer
import threading

class RealTimeKnowledgeUpdater:
    def __init__(self, vector_store, kafka_topic):
        self.vector_store = vector_store
        self.consumer = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )

    def start_listening(self):
        """监听文档变更事件，实时更新向量库"""
        def consume():
            for message in self.consumer:
                event = message.value
                self.handle_change_event(event)

        # 后台线程消费 Kafka 消息
        thread = threading.Thread(target=consume, daemon=True)
        thread.start()

    def handle_change_event(self, event: dict):
        """处理文档变更事件"""
        doc_id = event["doc_id"]
        action = event["action"]

        if action in ("create", "update"):
            # 异步向量化并更新
            doc = fetch_latest_document(doc_id)
            chunks = split_document(doc)
            embeddings = embed_batch(chunks)

            if action == "update":
                # 删除旧版本（按文档ID过滤删除）
                self.vector_store.delete(
                    filter={"metadata.doc_id": doc_id}
                )
            self.vector_store.add_documents(chunks, embeddings)

        elif action == "delete":
            self.vector_store.delete(
                filter={"metadata.doc_id": doc_id}
            )

策略三：版本化知识库（适合需要回滚的场景）

展开 Python 代码示例（30 行）

class VersionedKnowledgeBase:
    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.versions = {}
        self.active_version = "v1.0"

    def create_version(self, version: str, documents: list):
        """创建新版本知识库"""
        collection_name = f"knowledge_{version}"
        collection = self.vector_store.create_collection(collection_name)
        collection.add_documents(documents)
        self.versions[version] = collection_name
        return collection

    def activate_version(self, version: str):
        """灰度发布：逐步切换流量到新版本"""
        # 金丝雀发布：先 10% 流量走新版本
        self.active_version = version
        print(f"切换到版本 {version}")

    def rollback(self, version: str):
        """快速回滚到指定版本"""
        if version in self.versions:
            self.active_version = version
            print(f"回滚到版本 {version}")

    def search(self, query: str, k: int = 5) -> list:
        """查询时自动走当前激活版本"""
        collection_name = self.versions[self.active_version]
        return self.vector_store.get_collection(collection_name).search(query, k)

三种策略对比：

策略	更新延迟	实现复杂度	停机风险	适用场景
批处理（全量）	每天/每周	低	有（蓝绿部署可避免）	低频更新的知识库
批处理（增量）	每小时	中	无	中频更新
实时	秒级	高（Kafka+异步）	无	高频更新、实时性要求高

知识库版本管理最佳实践：

# 知识库更新时的一致性保证
class AtomicKnowledgeUpdate:
    def update_document(self, doc_id: str, new_content: str):
        """原子更新：先插入新版本，再删除旧版本"""
        new_doc_id = f"{doc_id}_v{int(time.time())}"

        # 1. 插入新版本
        new_chunks = split_and_embed(new_content)
        self.vector_store.add_documents(
            new_chunks,
            metadata={"original_doc_id": doc_id, "version_id": new_doc_id}
        )

        # 2. 原子切换（两步操作中间不会有空窗期）
        # 如果 Qdrant，可以用 collection aliases 做原子切换

        # 3. 删除旧版本
        self.vector_store.delete(
            filter={"metadata.original_doc_id": doc_id,
                    "metadata.version_id": {"$ne": new_doc_id}}
        )

面试话术：

"动态知识更新有三种策略：批处理（每天凌晨全量或增量重建）、实时（Kafka 监听变更事件）、版本化（支持快速回滚）。关键点是'无缝切换'——用蓝绿部署或 collection aliases 做原子切换，新库建好后瞬间切换，不会有查到旧数据的空窗期。我做客服知识库用批处理增量更新，每小时扫描变更文档，先删旧向量再插新向量，P99 延迟在非高峰时段 500ms 以内完成单文档更新。"

📚 参考：LlamaIndex：Data Ingestion（知识库更新管道）

---

版本: v3.134 | 更新: 2026-08-10 | 补充多模态RAG、Parent-Document Retrieval、动态知识更新