跳转至

第14讲：知识库多版本管理

上一讲：应用入口与环境前置校验 下一讲：数据隔离与多租户设计

本讲目标

  • 理解为什么 RAG 系统需要知识库版本管理
  • 掌握版本状态机的设计（STAGED → ACTIVE → ARCHIVED）
  • 理解版本切换的实现（O(1) 操作，不批量更新 Milvus）
  • 理解版本号生成的设计（时间戳 + 配置哈希）
  • 理解入库质量报告/质量门禁为什么是版本激活前置条件

本讲地图

本图对应本讲功能闭环，展示从输入到本讲交付物的主干路径。节点与主项目代码文件和函数保持一致，后续章节消费的能力只作为交付边界出现。

图 1：第 14 讲功能闭环地图

flowchart TD
    C14_GEN["版本号<br/>generate_kb_version()"]
    C14_COMMON["公共时间工具<br/>utc_now() / utc_file_stamp()"]
    C14_MODEL["版本对象<br/>KnowledgeBaseVersion"]
    C14_MYSQL["MySQL 存储基类<br/>_MySqlStore"]
    C14_STORE["版本仓库<br/>KnowledgeBaseVersionStore"]
    C14_ENSURE["确保版本<br/>ensure_version()"]
    C14_RESULT["入库记录<br/>record_ingest_result()"]
    C14_ACTIVE["激活归档<br/>activate_version() / archive_version()"]
    C14_API["版本 API<br/>list/activate/archive payload"]
    C14_QUERY{{"在线接入<br/>active_kb_version"}}
    C14_COMMON --> C14_GEN
    C14_GEN --> C14_MODEL
    C14_MYSQL --> C14_STORE
    C14_MODEL --> C14_STORE
    C14_STORE --> C14_ENSURE
    C14_ENSURE --> C14_RESULT
    C14_RESULT --> C14_ACTIVE
    C14_ACTIVE --> C14_QUERY
    C14_STORE --> C14_API
    C14_API --> C14_ACTIVE
    style C14_GEN fill:#F8FAFC,stroke:#64748B,stroke-width:2px
    style C14_COMMON fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style C14_MODEL fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style C14_MYSQL fill:#FEF3C7,stroke:#D97706,stroke-width:2px
    style C14_STORE fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style C14_ENSURE fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style C14_RESULT fill:#FEF3C7,stroke:#D97706,stroke-width:2px
    style C14_ACTIVE fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style C14_API fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style C14_QUERY fill:#DCFCE7,stroke:#16A34A,stroke-width:2px

节点与代码对齐

节点 对齐文件 函数/对象 本章职责
版本号 qa_core/governance/kb_versions.py generate_kb_version() 生成带场景和时间信息的知识库版本号。
公共时间工具 qa_core/common.py utc_now() / utc_file_stamp() 统一版本时间和文件时间戳格式。
版本对象 qa_core/governance/kb_versions.py KnowledgeBaseVersion 保存版本状态、描述、时间和入库统计。
MySQL 存储基类 qa_core/memory/base.py _MySqlStore 提供 SQLAlchemy 引擎和安全表名校验。
MySQL Schema 初始化 qa_core/storage/mysql_schema.py create_kb_version_tables() 集中维护版本表和 active 指针表的 DDL。
版本仓库 qa_core/governance/kb_versions.py KnowledgeBaseVersionStore 读取和写入 MySQL 版本表与 active 指针。
确保版本 qa_core/governance/kb_versions.py ensure_version() 入库时创建或复用 STAGED 版本。
入库记录 qa_core/governance/kb_versions.py record_ingest_result() 记录 doc/faq 数量和 source。
激活归档 qa_core/governance/kb_versions.py activate_version() / archive_version() 控制当前在线检索可见版本。
版本 API qa_core/api/kb_versions.py list/activate/archive payload 提供轻量版本管理接口。
在线接入 qa_core/pipeline/runtime.py active_kb_version 查询时把版本写入 RAGQueryContext。

第一部分：前置知识 — 为什么 RAG 需要版本管理

1.1 不加版本管理的风险

假设你用一个脚本把 500 个业务文档写入 Milvus。运行完毕后：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
场景 A：一切正常，问答效果好

场景 B：你发现新的 Embedding 模型（bge-m3-v2）效果更好，想切换
  → 重新入库，覆盖了旧的向量数据
  → 新模型效果反而更差（新的切分策略导致 chunk 太碎）
  → 无法回滚，因为旧数据已经被覆盖了

场景 C：你修改了文档切分参数（chunk_size 从 500 改为 300）
  → 想对比新旧切分方案的效果
  → 没有版本机制，你只能删除旧数据重建，对比无从谈起

版本管理的核心价值：让知识库的更新成为可逆操作。

版本管理还承担一个发布边界：新版本写入 Milvus 只是进入 STAGED，不等于可以上线。上线前必须有质量依据，当前项目通过入库质量报告和质量门禁决定是否允许激活。

1.2 版本管理的典型需求

需求 说明
安全入库 新版本先写入，不影响线上正在使用的版本
灰度验证 新版本入库后先评测，确定没问题再切换
快速回滚 新版本效果不好，一键切回旧版本
对比评测 同一个问题可以分别在新旧版本上验证召回效果
长期保留 历史版本不删除，作为 A/B 测试和故障分析的依据

第二部分：版本状态机

2.1 三种状态

stateDiagram-v2
    [*] --> STAGED : 入库完成

    STAGED --> ACTIVE : 激活版本<br/>仅更新 MySQL active 指针
    STAGED --> ARCHIVED : 直接归档<br/>从未激活的版本

    ACTIVE --> STAGED : 回滚/新版本激活<br/>旧 ACTIVE 转为 STAGED
    ACTIVE --> ARCHIVED : 长期不用后归档

    ARCHIVED --> [*] : Milvus 数据保留<br/>可手动清理

    note right of ACTIVE
        在线检索表达式：
        kb_version == "active_version"
        同一场景只有一个 ACTIVE
    end note

    note right of STAGED
        已写入 Milvus
        评测验证中
        用户检索不可见
    end note

安全入库与激活流程

flowchart TD
    Start(["执行 rebuild_kb_version.py"]) --> Create["1️⃣ 创建 STAGED 版本<br/>版本号含时间戳+配置哈希"]

    Create --> Ingest["2️⃣ 入库存入 STAGED<br/>FAQ 入库 + 文档入库<br/>写入 kb_version=新版本号"]

    Ingest --> Report["3️⃣ 生成入库质量报告<br/>解析失败/低质量chunk/<br/>FAQ空值/冲突检测"]

    Report --> Gate{"4️⃣ 入库质量门禁"}

    Gate -->|"✅ 通过"| Activate["5️⃣ 激活版本<br/>更新 MySQL active 指针<br/>旧 ACTIVE 降为 STAGED<br/>Milvus 数据不更新"]

    Gate -->|"❌ 不通过"| Abort["❌ 终止激活<br/>STAGED 版本保留<br/>线上仍用旧 ACTIVE<br/>用户无感知"]

    Activate --> Online["✅ 新版本上线<br/>下一次检索自动使用<br/>expr: kb_version==新版本号"]

    style Create fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style Gate fill:#FFFBEB,stroke:#D97706,stroke-width:2px
    style Activate fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style Abort fill:#FEF2F2,stroke:#DC2626,stroke-width:2px
    style Online fill:#ECFDF5,stroke:#059669,stroke-width:3px
  • STAGED：版本已写入 Milvus，但线上检索不使用。通常用于新入库的版本，等待评测验证。
  • ACTIVE：当前在线检索使用的版本。同一场景只有一个 ACTIVE 版本。
  • ARCHIVED：不再使用的历史版本。数据和 Milvus chunk 都保留，但不参与在线检索。

2.2 质量门禁与状态转换

第 14 章需要讲清一个边界：activate_version() 本身只负责切换 MySQL active 指针；它不判断资料质量。资料质量由 scripts/rebuild_kb_version.py --quality-gate --activate 串起来：先生成入库质量报告，再执行门禁，门禁通过才调用 activate_version()。

质量报告在本章只作为“是否允许激活”的前置条件出现，不展开检测算法。具体检查项，例如解析失败、低质量 chunk、FAQ 空值、重复问题和 FAQ/文档冲突，放在第 17 讲系统展开。

下面的代码展示的是步骤 5（激活）和归档操作。步骤 1（创建版本）见 ensure_version()，步骤 2（入库写入）见第 16 讲，步骤 3-4（质量报告和门控）见第 17 讲。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# qa_core/governance/kb_versions.py

def activate_version(self, kb_version: str) -> KnowledgeBaseVersion:
    """把指定版本切为当前在线检索版本。

    激活只更新 MySQL 控制面中的 active 指针，不更新 Milvus 数据。
    """  # → 流程图节点 5️⃣：激活版本
    record = self.get(kb_version)
    if record is None:
        raise ValueError(f"知识库版本不存在：{kb_version}")

    previous = self._active_pointer()[0]
    now = utc_now()

    with self.engine.begin() as conn:
        # 同一场景原 ACTIVE 统一转为 STAGED，保留回滚能力。
        conn.execute(text("UPDATE kb_versions SET status='STAGED' WHERE scenario_id=:scenario_id AND status='ACTIVE'"), ...)
        record.status = "ACTIVE"
        record.activated_at = now
        self._upsert_version_with_conn(conn, record)
        self._set_active_pointer_with_conn(conn, kb_version, previous)
    return self.get(kb_version) or record


def archive_version(self, kb_version: str) -> KnowledgeBaseVersion:
    """归档一个非 active 版本。

    归档不会删除 Milvus 数据，只是状态标记。
    """  # → 流程图未展示的附加操作：归档
    if self.active_version_candidate() == kb_version:
        raise ValueError("不能归档当前 active 知识库版本")

    record = self.get(kb_version)
    record.status = "ARCHIVED"
    record.archived_at = utc_now()
    self._upsert_version(record)
    return record

2.3 激活操作的轻量性

关键设计：激活版本只更新 MySQL 中的一行 active 指针，不碰 Milvus。

1
2
3
4
5
6
激活前的在线检索：
  Milvus expr 包含 kb_version == "v1"

激活后的在线检索：
  Milvus expr 包含 kb_version == "v2"
  v1 的 chunk 数据仍在 Milvus 中，只是不再被查到

如果激活需要修改所有 chunk 的 metadata，一个 10 万条 chunk 的知识库需要很长时间。通过把版本切换放在检索表达式中，版本切换变成了 O(1) 操作。

2.4 为什么采用 STAGED 候选版本 + active 指针切换

多版本入库不要理解成“把资料重新写一遍”，而要理解成一次知识库发布流程。新资料可能出现解析失败、切分异常、FAQ 口径冲突、source 配错、召回退化等问题。如果直接覆盖线上向量数据，一旦新版本质量有问题，用户会立刻受到影响，而且很难快速回滚。

本项目采用的是控制面和数据面分离：

层次 保存内容 主要职责
Milvus 数据面 所有版本的 FAQ 向量和文档 chunk 向量 保存可检索数据，不负责判断哪个版本上线
MySQL 控制面 kb_versions、kb_active_versions 保存版本状态、active 指针和 previous 指针
在线检索表达式 kb_version == active_version 让查询只命中当前 active 版本

所以完整业务逻辑是：

1
2
3
4
5
入库 = 生成一个候选版本
质量报告 = 描述候选版本有哪些风险
质量门禁 = 判断候选版本能不能上线
激活 = 切换 active 指针
回滚 = 把 active 指针切回旧版本

注意两个边界：

  1. 写入 Milvus 不等于上线：STAGED 版本已经在 Milvus 中存在，但线上检索不会命中它，因为线上检索只查 active kb_version。
  2. 质量报告不等于质量门禁：质量报告负责记录事实，例如失败文件、重复 FAQ、低质量 chunk；质量门禁负责根据阈值决定是否阻断激活。

这种设计的好处是：

问题 直接覆盖式入库 多版本发布式入库
新资料有问题 线上立即受影响 STAGED 阶段被拦截
需要回滚 很难恢复旧向量 切回 previous active 指针
发布速度 可能要修改大量向量数据 只更新 MySQL active 指针
排查问题 不知道哪次入库引入问题 每个版本有独立版本号、报告和统计
多场景管理 容易互相影响 每个场景独立 active 指针

企业项目中也经常做跨版本增量构建。当前项目采用的是物理复制式增量：未变化文件不重新 embedding，但会把旧 chunk 和 dense 向量复制到新版本，这样新版本仍然是一个完整快照，线上查询只需要过滤 kb_version == active_version。

大规模企业知识库还可以升级为引用式增量：未变化 chunk 不复制，而是用 valid_from_seq / valid_to_seq 描述 chunk 在哪些版本中有效。两种方案的取舍如下：

方案 查询方式 优点 代价
物理复制式增量 kb_version == active_version 查询简单、版本完整、适合中小规模 未变化 chunk 会重复占用空间
引用式增量 valid_from_seq <= active_seq 且未失效 节省空间、适合百万级 chunk 检索表达式、回滚、质量报告和 GC 都更复杂

本课程主项目选择物理复制式增量，是为了先把“候选版本入库 → 质量门禁 → active 指针切换 → 回滚”的主线讲清楚。引用式增量作为企业规模化优化方向，在第 16 讲入库章节补充理解。

第三部分：版本号设计

3.1 版本号生成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def generate_kb_version(prefix="kb", scenario_id=None) -> str:
    """生成一个适合人读和机器过滤的知识库版本号。

    包含：
    - 时间戳：便于肉眼判断版本先后
    - 配置短 hash：把 embedding、reranker、chunk_schema、collection
      等关键配置纳入标识
    """
    settings = get_settings()
    scenario = _resolve_version_scenario(scenario_id)  # 私有 helper，避免 settings -> scenarios -> kb_versions 循环导入

    stamp = utc_file_stamp()  # 如 20260506_103000（年月日_时分秒）
    config_hash = stable_hash(
        scenario.scenario_id,
        settings.embedding_model_version,    # 如 "bge-m3-local-v1"
        settings.reranker_model_version,     # 如 "bge-reranker-v1"
        settings.chunk_schema_version,       # 如 "parent_child_v1"
        scenario.doc_collection,
        scenario.faq_collection,
    )[:8]  # 只取前 8 位

    return f"{prefix}_{scenario.scenario_id}_{stamp}_{config_hash}"
    # 例：kb_enterprise_knowledge_20260506_103000_9f2a1b3c

3.2 为什么版本号包含配置哈希

设计意图：从版本号可以直接判断两个版本是否使用同一套配置。

1
2
3
kb_enterprise_knowledge_20260506_103000_9f2a1b3c
kb_enterprise_knowledge_20260507_150000_7d3e8f1a
                           不同日期 ↑         不同 hash ↑

如果两个版本的 hash 相同但日期不同，说明是同一套配置下的数据更新（新增/修改了文档）。 如果 hash 不同，说明 Embedding 模型、Reranker 模型或 Chunk 方案有变化，需要重点关注召回质量的对比。

第四部分：MySQL 版本控制面

4.1 表结构

当前项目把知识库版本控制面保存到 MySQL，而不是本地 JSON 文件。核心是两张表：

表 作用
kb_versions 保存每个版本的状态、模型配置快照、collection、入库统计
kb_active_versions 保存每个场景当前 active 版本和 previous 版本

kb_versions 的关键字段：

字段 说明
scenario_id 业务场景
kb_version 知识库版本号
status STAGED / ACTIVE / ARCHIVED
doc_collection / faq_collection 对应 Milvus collection
embedding_model_version / reranker_model_version / chunk_schema_version 入库配置快照
sources_json / stats_json 已入库 source 和统计信息

kb_active_versions 的关键字段：

字段 说明
scenario_id 业务场景
active_kb_version 在线检索默认使用的版本
previous_kb_version 上一个 active 版本，用于快速回滚

表结构创建逻辑集中在 qa_core/storage/mysql_schema.py：

1
2
3
4
5
# qa_core/storage/mysql_schema.py
def create_kb_version_tables(engine, *, version_table: str, active_table: str) -> None:
    """创建知识库版本表和 active 指针表。"""
    _execute_ddl(engine, ddl_versions)
    _execute_ddl(engine, ddl_active)

KnowledgeBaseVersionStore.ensure_tables() 只调用 schema 初始化函数，再做版本状态校准。这样表结构初始化和版本状态机读写分开：

模块 职责
qa_core/storage/mysql_schema.py 维护 MySQL 控制面表结构和幂等 schema 补齐逻辑
qa_core/governance/kb_versions.py 负责版本状态机、active 指针、激活、归档和回滚
scripts/rebuild_kb_version.py 负责编排入库、质量门禁，通过后才激活版本

当前项目暂不引入 Alembic。课程一期先讲清控制面/数据面分离和 active 指针切换；生产环境如果需要严格版本化 DDL，可以把 mysql_schema.py 中的表结构迁移成 Alembic revisions。

4.2 KnowledgeBaseVersionStore 类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class KnowledgeBaseVersionStore(_MySqlStore):
    """知识库版本状态机的 MySQL 存储实现。

    每个场景在 kb_versions 中有多条版本记录，在 kb_active_versions
    中有一条 active 指针记录。这样多场景可以独立激活、回滚和对比。
    """

    def __init__(self, scenario_id=None):
        self.scenario = _resolve_version_scenario(scenario_id)
        self.settings = get_settings()
        self._engine = None

    def resolve_active_version(self, requested=None) -> str:
        """解析一次检索应该使用的知识库版本。

        优先级：
        1. 请求显式传入的 kb_version（用于评测/灰度）
        2. 环境变量 ACTIVE_KB_VERSION
        3. MySQL active 指针表中的 active_kb_version

        当前在线检索必须带版本过滤。没有 active 版本直接报错。
        """
        if requested:
            if not self.exists(requested):
                raise ValueError(f"请求的知识库版本不存在：{requested}")
            return requested

        active = self.active_version_candidate()
        if not active:
            raise ValueError(
                f"场景 {self.scenario.scenario_id} 没有 active 知识库版本"
            )
        if not self.exists(active):
            raise ValueError(
                f"active 知识库版本不存在于版本表：{active}"
            )
        return active

第五部分：与 Milvus 检索的集成

5.1 写入时携带版本信息

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# qa_core/governance/kb_versions.py
def version_metadata(kb_version, scenario_id=None):
    """构建写入每个 FAQ/chunk metadata 的版本字段。"""
    settings = get_settings()
    scenario = _resolve_version_scenario(scenario_id)
    return {
        "scenario_id": scenario.scenario_id,
        "kb_version": kb_version,
        "embedding_model_version": settings.embedding_model_version,
        "reranker_model_version": settings.reranker_model_version,
        "chunk_schema_version": settings.chunk_schema_version,
    }

每条 FAQ 和 chunk 入库时，这些字段都会被写入 metadata：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
chunk = Document(
    page_content="入职流程包括以下步骤...",
    metadata={
        "source": "hr",
        "chunk_id": "abc123",
        "kb_version": "kb_enterprise_knowledge_20260507_150000_7d3e8f1a",
        "embedding_model_version": "bge-m3-local-v1",
        ...
    }
)

5.2 检索时过滤版本

1
2
3
4
5
6
7
8
9
# 在线检索
active_version = resolve_active_kb_version(requested, scenario.scenario_id)
# active_version = "kb_enterprise_knowledge_20260507_150000_7d3e8f1a"

# 拼入 Milvus 表达式
expr = f'kb_version == "{active_version}" and source == "hr" and ...'

# 只检索 active 版本的 chunk
results = milvus_store.similarity_search(query, expr=expr)

5.3 评测用历史版本

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 评测脚本可以显式指定历史版本
service.debug_retrieval(
    query="入职流程有哪些步骤",
    kb_version="kb_enterprise_knowledge_20260506_103000_9f2a1b3c",  # 旧版本
    ...
)

# 对比两个版本对同一批问题的召回效果
for question in eval_set:
    old_result = service.debug_retrieval(query=question, kb_version=old_version)
    new_result = service.debug_retrieval(query=question, kb_version=new_version)
    compare(old_result, new_result)

第六部分：全量重建的安全流程

这一部分是第 14 章需要和第 16、17 章衔接的地方：第 16 章负责把资料写入新版本，第 17 章负责解释质量报告如何检查；第 14 章负责解释为什么质量报告不通过时不能激活版本。

1
python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --force --quality-gate --activate

执行顺序：

1
2
3
4
5
6
7
1. 创建 STAGED 版本（version = "kb_...20260507_150000_xxxx"）
2. FAQ 入库（写入 STAGED 版本的 kb_version）
3. 文档入库（写入 STAGED 版本的 kb_version）
4. 生成入库质量报告
5. 执行入库质量门禁
   ├─ 通过 → 调用 activate_version()，将 STAGED 切换为 ACTIVE
   └─ 不通过 → 终止流程，STAGED 版本仍保留（不激活）

对应真实代码在 scripts/rebuild_kb_version.py：

1
2
3
4
5
6
7
8
report = build_ingestion_quality_report(...)
gate_result = evaluate_report_against_gate(report, thresholds)
if not gate_result["ok"]:
    save_ingestion_quality_report(report)
    sys.exit(1)

if args.activate:
    version_store.activate_version(kb_version)

这段逻辑表达的是：版本已经入库不代表可以上线；只有质量门禁通过，才允许切换 active 指针。

关键安全点：即使新的 STAGED 版本已经写入了 Milvus，只要没有执行激活步骤，线上检索仍然使用旧的 ACTIVE 版本。用户完全无感知。

本讲实践闭环

项目 内容
本讲类型 工程治理
实践产物 KnowledgeBaseVersionStore、版本状态机、激活和回滚能力、激活前质量门禁边界
是否进入最终项目 是
验收方式 创建 staged 版本，生成入库质量报告，门禁通过后激活为 active，再归档旧版本
后续落点 第 16 讲入库生成版本，第 17 讲质量门禁控制激活

通过标准：任意时刻每个场景只有一个 active 版本，版本可追踪、可回滚；质量门禁失败时不会切换线上版本。

本讲从 0 到 1 实现闭环

这一讲的核心是把“重建知识库”变成可追踪、可回滚的发布动作。实现顺序如下：

stateDiagram-v2
    [*] --> STAGED: 新建版本并入库
    STAGED --> ACTIVE: 质量门禁通过 + activate
    ACTIVE --> ARCHIVED: 新版本激活后旧版本归档
    ARCHIVED --> ACTIVE: 回滚到历史版本
    STAGED --> ARCHIVED: 构建失败或废弃
  1. 先定义版本状态：STAGED、ACTIVE、ARCHIVED。
  2. 再实现版本清单 store，把每个场景的版本列表和 active 指针持久化。
  3. 然后在入库脚本里先创建 staged 版本，入库后生成质量报告。
  4. 质量门禁通过后再激活，失败则保留 staged 且线上仍用旧 active。
  5. 最后保证同一个场景任意时刻只有一个 active 版本。

实现完成后，相关代码结构应该是下面这张图：

flowchart LR
    VersionStore["qa_core/governance/kb_versions.py<br/>版本记录/状态机/active 指针"] --> Rebuild["scripts/rebuild_kb_version.py<br/>创建 STAGED<br/>质量通过后激活"]
    VersionStore --> Filters["qa_core/retrieval/filters.py<br/>按 active kb_version 检索"]
    VersionStore --> Admin["qa_core/api/kb_versions.py<br/>版本查询/管理接口"]
    Rebuild --> Reports["reports/<br/>构建与质量报告"]

来源：真实代码逻辑压缩版，对应 qa_core/governance/kb_versions.py::KnowledgeBaseVersion。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
@dataclass
class KnowledgeBaseVersion:
    kb_version: str
    scenario_id: str = ""
    status: str = "STAGED"
    description: str = ""
    created_at: str = field(default_factory=utc_now)
    activated_at: str | None = None
    archived_at: str | None = None
    doc_collection: str = ""
    faq_collection: str = ""
    embedding_model_version: str = ""
    reranker_model_version: str = ""
    chunk_schema_version: str = ""
    created_by: str = "local"
    sources: list[str] = field(default_factory=list)
    stats: dict[str, Any] = field(default_factory=dict)

激活版本不是改 Milvus 数据，而是修改 MySQL 控制面里的 active 指针。这样切换速度快，也能随时回滚。

来源：真实代码逻辑压缩版，对应 qa_core/governance/kb_versions.py::activate_version()。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
def activate_version(self, kb_version: str) -> KnowledgeBaseVersion:
    record = self.get(kb_version)
    if record is None:
        raise ValueError(f"知识库版本不存在：{kb_version}")

    with self.engine.begin() as conn:
        previous = self._active_pointer_with_conn(conn)[0]
        conn.execute(text("UPDATE kb_versions SET status='STAGED' WHERE scenario_id=:scenario_id AND status='ACTIVE'"), ...)
        record.status = "ACTIVE"
        record.activated_at = utc_now()
        self._upsert_version_with_conn(conn, record)
        self._set_active_pointer_with_conn(conn, kb_version, previous)
    return self.get(kb_version) or record

注意：真实代码激活新版本时，旧 ACTIVE 会降为 STAGED，不是自动归档为 ARCHIVED。归档是单独的 archive_version() 操作，并且不能归档当前 active 版本。

入库时，每个 chunk 都要写入 scenario_id 和 kb_version。线上检索通过 Milvus expr 只查当前 active 版本。

来源：真实代码调用点，见 qa_core/governance/kb_versions.py 和 qa_core/retrieval/filters.py。

1
2
3
4
5
metadata = {
    "scenario_id": scenario_id,
    "kb_version": kb_version,
    "source": source,
}

验收时先创建 staged，再激活，再检查旧 active 是否归档。

来源：命令行验收，对应 scripts/rebuild_kb_version.py。 Docker Compose 模式下执行前，先确认项目根目录已经存在 .env.compose。

1
docker compose --env-file .env.compose run --rm api python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --force --quality-gate --activate

闭环验证重点：

验证项 验证方式 期望结果
新建版本 执行 --new-version 生成 STAGED 版本
入库质量报告 默认生成，或显式使用 --quality-gate 报告写入 reports/ingestion/，包含当前 kb_version
激活版本 加 --quality-gate --activate 门禁通过后新版本变 ACTIVE
旧版本转为 STAGED 多次激活 旧 ACTIVE 变 STAGED，保留回滚能力
检索过滤 查看 expr 包含当前 kb_version
回滚能力 指定历史版本激活 active 指针切回历史版本
手动归档 调用 archive 非 active 版本变 ARCHIVED

验收重点：知识库更新必须可追踪、可回滚，不能直接覆盖线上数据。

重点掌握

优先级 内容 原因
★★★ 必会 版本状态机：STAGED（已入库待验证）→ ACTIVE（在线检索使用）→ ARCHIVED（归档保留） 知识库版本管理的核心模型
★★★ 必会 O(1) 版本切换：激活只更新 MySQL active 指针，不更新 Milvus 数据 理解版本切换为什么是即时的
★★★ 必会 控制面 / 数据面分离：Milvus 保存所有版本数据，MySQL 控制哪个版本上线 理解为什么 STAGED 数据不会污染线上
★★★ 必会 版本号设计：kb_{scenario_id}_{timestamp}_{config_hash}，配置哈希体现版本差异 从版本号直接判断配置是否变化
★★ 理解 resolve_active_version() 的三级优先级：请求传入 > 环境变量 > MySQL active 指针 理解版本解析流程
★★ 理解 version_metadata() 将版本信息写入每个 chunk 的 metadata，检索时通过 expr 过滤 写入和检索的版本关联机制
★★ 理解 安全入库流程：创建 STAGED → 入库 → 生成质量报告 → 质量门禁 → 激活 生产环境的完整安全操作
★★ 理解 质量报告记录事实，质量门禁给出是否允许激活的结论 避免把“发现问题”和“阻断上线”混成一个概念
★★ 理解 质量报告在第 14 章是版本激活前置条件，具体检查算法在第 17 章展开 避免把版本治理和质量评估混成一章
★ 了解 跨版本增量构建仍然产出完整的新 kb_version 理解企业项目如何减少重复 embedding，同时保持版本确定性
★ 了解 评测时可以显式指定历史版本做对比 了解版本管理的扩展用途

本讲小结

  • 版本管理让知识库更新成为可逆操作：入库 → 评测 → 激活 →（效果不好）→ 回滚
  • 版本状态机：STAGED（待验证）→ ACTIVE（在线使用）→ ARCHIVED（归档保留）
  • 版本切换是 O(1) 操作：只更新 MySQL 中的 active 指针，不更新 Milvus 数据
  • 控制面和数据面分离：Milvus 保存各版本数据，MySQL 决定哪个版本在线
  • 版本号 = 时间戳 + 配置哈希，可以肉眼判断先后和配置差异
  • 每个场景独立 active 指针，不同行业场景可以独立管理知识库版本
  • 入库质量报告是激活前置条件：报告或门禁失败时，新版本保持 STAGED，线上仍使用旧 ACTIVE
  • 质量报告记录问题，质量门禁决定是否激活，两者职责不能混淆
  • 增量构建也要产出完整版本，线上始终只查一个 active kb_version
  • 评测脚本可以显式指定历史版本，实现新老版本对比

下一讲：数据隔离与多租户 — 租户/数据集/角色隔离、Milvus 表达式过滤