PNProject Notes
返回项目
第四阶段：治理与生产化第 13 章

知识库治理与离线入库发布

以一次完整的知识库发布为主线，串联数据边界、候选版本、增量入库、质量门禁和 active 切换，形成可追踪且可回滚的治理流程。

30 分钟阅读
#KnowForge#RAG#工程实践

发布主线

版本来源约定：在线请求未显式指定历史版本时，只读取 MySQLkb_active_versions.active_kb_version。项目不再通过ACTIVE_KB_VERSION环境变量覆盖数据库指针；发布和回滚统一调用activate_version()，历史版本查询通过请求参数显式指定。

flowchart LR
    Scope["DataScope\ntenant / dataset / visibility / roles"] --> Version["创建候选版本\nSTAGED + version_seq"]
    Version --> Ingest["离线入库\nLoader → normalize → split → embedding"]
    Ingest --> Manifest["IndexManifest\n指纹、chunk、增量决策"]
    Manifest --> Report["入库质量报告\n失败文件、空内容、重复、噪声"]
    Report --> Gate{"质量门禁"}
    Gate -->|通过| Activate["激活版本\n更新 active 指针"]
    Gate -->|不通过| Hold["保留 STAGED\n修复资料后重跑"]
    Activate --> Online["在线检索\n按版本和 DataScope 过滤"]
治理对象 代码中解决的问题 发布时必须保证
DataScope 谁能看哪些租户、数据集、可见性和角色数据 写入和检索使用同一组边界字段
kb_version/version_seq 新旧知识如何并存、切换和回滚 新资料先进入候选版本，不能直接污染 active
IndexManifest 文件是否变化、哪些 chunk 可以复用 增量构建可重复执行，避免无效解析和向量化
质量报告与 Gate 候选版本是否达到发布条件 Gate 失败时不切换 active

离线入库总流程

项目阶段 13保留原有专题分析，但阅读主线要按scripts/rebuild_kb_version.py::main()的 8 个代码阶段来走。也就是说，后面的“文档加载器注册表”“表格入库”“IndexManifest”“FAQ 入库”“复杂图文资料治理”不是独立散点，而是嵌在一次知识库版本发布任务里的局部实现。

flowchart TD
    A[启动 rebuild_kb_version.py] --> B[第1步 解析场景配置并初始化 MySQL schema]
    B --> C[第2步 按需重置 Milvus Collection]
    C --> D[第3步 创建或复用目标 STAGED 版本]
    D --> E[第4步 解析增量构建基准版本]
    E --> F[第5步 FAQ 快照式入库]
    F --> G[第6步 文档引用式增量入库]
    G --> H[第7步 质量报告 + 质量门禁 + 激活版本]
    H --> I[第8步 输出构建摘要]
    I --> J[在线问答只读 active 版本]
入库阶段 对齐代码注释 本阶段要读懂什么 对应内容
第 1 步 解析场景配置并初始化 MySQL schema scenario.toml如何确定数据目录、FAQ 文件、collection、source 白名单；MySQL 控制面表为什么要先准备好 1.3、1.4
第 2 步 按需重置 Milvus Collection --reset-collections只在 schema 不兼容时使用，不能和引用式增量混用 1.3、1.4、9.3
第 3 步 创建或复用目标知识库版本 ensure_version()如何生成 STAGED 候选版本，version_seq为什么必须单调递增 1.3、1.4、5.1
第 4 步 解析增量构建基准版本 --incremental-from active如何确定基准版本，并写入record_incremental_base() 1.3、5.4、5.7
第 5 步 FAQ 入库 FAQ 为什么每次按kb_version快照式重建，为什么page_content放问题、metadata.answer放答案 第二部分
第 6 步 文档入库 Loader、Docling、表格行、metadata 标准化、chunking、Manifest、引用复用和valid_from_seq/valid_to_seq如何串起来 第三、四、五、六、七、八部分
第 7 步 质量报告 + 质量门禁 + 激活版本 质量报告只做试解析和检查；门禁失败保留 STAGED；通过并--activate后才切 active 指针 1.4、3.1.1、9.4
第 8 步 输出构建摘要 如何根据kb_version、faq_records、doc_chunks、activated、incremental_base、quality_report判断本次发布结果 9.1、9.5

读这一讲时建议先顺着上表理解 8 个阶段，再进入后面的专题细节。这样不会把“文档加载器注册表”误认为入库起点，也不会把“Manifest 增量机制”误认为一条独立链路。

第一部分：技术背景 — 离线 vs 在线链路

1.1 清晰的工程边界

离线链路（入库）                      在线链路（问答）
─────────────────                    ─────────────────
定时/手动执行                         每次用户提问时执行
修改 Milvus 数据                      只读 Milvus 数据
可以慢（几分钟到几十分钟）            必须快（秒级响应）
可以重试、可以回滚                    必须一次成功
解析文件、切分、向量化                只做检索、不解析文件

关键原则：在线问答不解析文件、不执行 OCR、不写入知识库。

1.2 为什么分开

如果把文档解析放在在线链路：

  • 用户提问时临时解析 PDF → 首 token 延迟增加 5-10 秒
  • 文件解析失败时用户看到的是"PDF 损坏"而非答案
  • 无法做质量报告（因为解析是即时的，没有机会检查）

如果把向量化放在在线链路：

  • 用户问题需要等 Embedding 模型加载（冷启动 10+ 秒）
  • 无法预热 Embedding 模型

1.3 知识库构建总链路

本项目的离线知识库构建不是单独的“文档切分”或“FAQ 入库”，而是一条完整的版本化构建链路。可以用一句话概括：

离线链路负责把原始资料变成“带版本、带权限、可检索、可回滚”的知识资产；在线链路只读取当前 active 版本来回答问题。

生产、本地验证或资料发布时有两个常用入口：

入口 用途 适合场景
scripts/rebuild_scenarios.py 一次初始化/重建全部 8 个冻结场景 新环境初始化、统一准备、Milvus schema 变更后全量修复
scripts/rebuild_kb_version.py 只重建单个业务场景 只修改了某个场景资料、验证单场景入库、定位某个场景问题

如果在 Docker Compose 里执行入库命令，先确认项目根目录存在.env.compose。仓库只提交.env.compose.example，首次部署需要生成本地配置文件：

if (!(Test-Path .env.compose)) { Copy-Item .env.compose.example .env.compose }
notepad .env.compose

如果是新环境，或者 Milvus collection schema 变更后需要全量修复，优先使用批量脚本：

python scripts/rebuild_scenarios.py --reset-collections

它会对 8 个冻结场景逐个执行“新建版本 → 强制入库 → 质量门禁 → 激活”，并在--reset-collections开启时删除旧 FAQ/Doc collection，确保 Milvus schema 按当前代码重新创建。

在 Docker Compose 模式下，对应命令是：

docker compose --env-file .env.compose up -d mysql etcd minio milvus
docker compose --env-file .env.compose build api
docker compose --env-file .env.compose run --rm api python scripts/rebuild_scenarios.py --reset-collections

如果之前已经存在知识库，只是资料内容变化，批量重建全部 8 个场景时不加--reset-collections：

docker compose --env-file .env.compose run --rm api python scripts/rebuild_scenarios.py

如果只重建一个场景，使用scripts/rebuild_kb_version.py：

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

Docker Compose 模式下，对应命令是：

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

企业中更常见的日常资料更新方式，是“构建阶段增量，查询阶段按有效版本视图读取”。如果当前 active 版本已经存在，且只是少量文件变化，可以创建新候选版本并基于 active 做跨版本增量构建：

python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --incremental-from active --quality-gate --activate

这条命令的语义是：FAQ 仍按新版本重建；文档先读取 active 版本的 MySQL IndexManifest，未变化文件直接引用旧版本 chunk，不复制 Milvus 行；变化文件让旧 chunk 从目标版本开始失效，再重新加载、切分、embedding；删除文件只写失效版本。在线查询按 activeversion_seq解释有效期视图。

--incremental-from不能和--force或--reset-collections同时使用：--force表示全部重算，--reset-collections会删掉可复用的旧向量。

源码第 15 章配套验证实践里提供了一个专门演示引用式增量的脚本：

python scripts\demo_incremental_ingestion.py

这个脚本会在临时目录里构造两个连续版本：第二个版本同时包含未变化、修改、删除和新增文件。输出中的file_transitions会显示unchanged_reused、modified_reembedded、deleted_expired、added_reembedded四类结果；visible_chunks会展示不同active_seq下哪些 chunk 可见。

引用式增量版本

当前项目采用的是引用式增量版本：未变化文件不重新 embedding，也不复制旧 chunk；新版本只记录“从哪个基线版本继承、哪些文件新增、哪些文件修改、哪些文件删除”。查询时通过有效期字段判断当前版本能看到哪些 chunk。

文档检索表达式不再等价于kb_version == active，而是用 active 版本序号解释有效视图：

valid_from_seq <= active_seq
and (valid_to_seq == 0 or valid_to_seq > active_seq)

当前版本中新写入的 chunk 会带上valid_from_seq = active_seq；未变化资料引用的旧 chunk 会保留原来的valid_from_seq，并通过valid_to_seq判断是否失效。因此文档检索只需要有效期窗口，不需要再追加kb_version == active分支。

一种典型 metadata 设计如下：

字段 含义
scenario_id 业务场景
source_type 数据类型，文档为doc
record_type 记录粒度，文档 chunk 为doc_chunk
versioning_mode 文档版本模式，值为reference_incremental
version_filter_mode 文档检索过滤模式，值为validity_window
doc_id 稳定文件 ID
chunk_id 稳定 chunk ID
file_fingerprint 基于绝对路径、修改时间和文件大小生成的本地文件指纹
valid_from_seq 从哪个版本序号开始有效
valid_to_seq 到哪个版本序号前失效，0表示仍然有效

假设 active 版本序号是8，文档检索表达式为：

scenario_id == "enterprise_knowledge"
and valid_from_seq <= 8
and (valid_to_seq == 0 or valid_to_seq > 8)

增量构建时：

文件状态 引用式增量处理方式
未变化 不写 Milvus，只继续引用旧 chunk
新增 插入新 chunk，valid_from_seq = 当前版本序号
修改 旧 chunk 写valid_to_seq = 当前版本序号，新 chunk 写valid_from_seq = 当前版本序号
删除 旧 chunk 写valid_to_seq = 当前版本序号，作为 tombstone 处理

这里的“旧 chunk”只来自本次增量基准版本的同路径 manifest，不是全历史版本扫描。也就是说，某个文件在第二个版本中修改时，只让第一个版本里被新版本继承的那批 chunk 从新版本开始不可见；更早或其他历史版本仍按自己的active_seq查询，不会被物理删除，也不会失去回滚价值。

下面用一个具体例子看引用式增量怎么工作。假设企业知识库里有三份资料：

hr_onboarding.md      入职流程
it_vpn.md             VPN 处理
finance_expense.md    报销流程

首个版本：首次全量入库

第一次入库时，三个文件都写入 Milvus：

chunk_id 文件 内容摘要 valid_from_seq valid_to_seq
hr_c1 hr_onboarding.md 入职需要提交身份证、银行卡、合同信息 1 0
it_c1 it_vpn.md VPN 连不上先检查账号、网络和 MFA 1 0
fin_c1 finance_expense.md 报销流程包括提交单据、审批、财务复核 1 0

此时 active 版本序号是1，查询表达式是：

valid_from_seq <= 1
and (valid_to_seq == 0 or valid_to_seq > 1)

能查到：

hr_c1, it_c1, fin_c1

第二个版本：只修改 VPN 文档

后来 IT 更新了 VPN 文档，新增了“客户端版本检查”的要求。引用式增量不会复制 HR 和财务 chunk，只处理变化文件：

操作 chunk_id 文件 valid_from_seq valid_to_seq 说明
标记旧 chunk 失效 it_c1 it_vpn.md 1 2 新版本开始不再使用旧 VPN 口径
插入新 chunk it_c2 it_vpn.md 2 0 新 VPN 口径从第二个版本开始有效

Milvus 中现在一共有四条 chunk：

chunk_id 文件 valid_from_seq valid_to_seq
hr_c1 hr_onboarding.md 1 0
it_c1 it_vpn.md 1 2
it_c2 it_vpn.md 2 0
fin_c1 finance_expense.md 1 0

如果 active 版本序号切到2，查询表达式是：

valid_from_seq <= 2
and (valid_to_seq == 0 or valid_to_seq > 2)

能查到：

hr_c1, it_c2, fin_c1

注意：hr_c1和fin_c1没有复制一份到第二个版本，但它们仍然有效，因为valid_to_seq = 0。

第三个版本：删除财务报销文档，新增差旅文档

再后来财务删除旧的报销流程文档，并新增差旅规则文档：

删除：finance_expense.md
新增：finance_travel.md

引用式增量处理如下：

操作 chunk_id 文件 valid_from_seq valid_to_seq 说明
标记旧 chunk 失效 fin_c1 finance_expense.md 1 3 v3 起旧报销资料不再可见
插入新 chunk fin_travel_c1 finance_travel.md 3 0 差旅规则从 v3 起生效

如果 active 版本序号切到3，有效 chunk 是：

hr_c1, it_c2, fin_travel_c1

完整状态表如下：

chunk_id 文件 valid_from_seq valid_to_seq v1 可见 v2 可见 v3 可见
hr_c1 hr_onboarding.md 1 0 是 是 是
it_c1 it_vpn.md 1 2 是 否 否
it_c2 it_vpn.md 2 0 否 是 是
fin_c1 finance_expense.md 1 3 是 是 否
fin_travel_c1 finance_travel.md 3 0 否 否 是

这个例子说明了引用式增量的关键点：

  1. 未变化资料不复制，例如hr_c1从 v1 一直被 v2、v3 复用。
  2. 修改资料不是覆盖旧 chunk，而是让旧 chunk 在新版本前失效，再插入新 chunk。
  3. 删除资料不是立刻物理删除，而是写valid_to_seq，让它从某个版本开始不可见。
  4. 回滚时只需要把 active 版本序号从3切回2，fin_c1又会重新可见。

这种方案节省空间，也更适合大规模知识库。本项目已经把它作为版本治理能力实现：MySQL 版本表提供version_seq，文档 chunk 写入valid_from_seq / valid_to_seq，检索时用 active 版本序号解释有效版本视图。

如果 Milvus Collection 的 schema 发生过变化，例如 sparse 字段从普通 SparseVector 改成 BM25 Function 输出字段，需要加上--reset-collections删除旧集合并重新建表：

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

这里要区分两个参数：--force只是忽略文件指纹、强制把资料重新写入新版本；--reset-collections会删除 Milvus 里的 FAQ/Doc collection，让当前代码重新创建 schema。已有知识库只更新资料时，用--force，不要默认加--reset-collections。

1.4 完整入库链路闭环

这张图不是某一个函数的内部流程，而是从脚本入口到线上可检索、失败可修复、版本可回滚的端到端闭环：

flowchart TD
    Start(["执行 rebuild_kb_version.py<br/>或 rebuild_scenarios.py"]) --> Bootstrap["bootstrap_mysql_schema()<br/>执行 runtime_schema.sql<br/>初始化版本/Manifest/反馈/历史表"]
    Bootstrap --> Scenario["解析 scenario.toml<br/>确定 FAQ/Doc collection、数据目录、source、权限默认值"]
    Scenario --> Reset{"是否传入<br/>--reset-collections？"}
    Reset -->|"是"| Drop["删除旧 FAQ/Doc collection<br/>仅用于 schema 不兼容后的重建"]
    Reset -->|"否"| Version
    Drop --> Version["创建或确认 STAGED 知识库版本<br/>分配 version_seq<br/>线上 active 暂不变化"]

    Version --> FAQ["FAQ 入库<br/>CSV 问答对 -> Document<br/>question=text, answer=metadata"]
    Version --> Docs["普通文档入库<br/>Loader/Docling -> 标准化 -> 父子块切分"]
    Version --> Tables["表格入库<br/>CSV/Excel 每行一个 table_row"]
    Version --> OCR["OCR 复核资料<br/>reviewed Markdown 才进入正式目录"]

    FAQ --> FaqStore["写入 FAQ Milvus collection<br/>dense + sparse + metadata"]
    Docs --> ManifestCheck["读取 MySQL IndexManifest<br/>判断跳过、复用或重建"]
    Tables --> ManifestCheck
    OCR --> ManifestCheck

    ManifestCheck --> Reuse{"文件相对基准版本<br/>是否未变化？"}
    Reuse -->|"未变化"| ReuseChunk["引用旧 chunk<br/>补齐有效期视图<br/>不重复 embedding"]
    Reuse -->|"变化/新增/删除"| Rebuild["失效旧 chunk<br/>重新解析、切分、embedding<br/>写入 Doc Milvus collection"]
    ReuseChunk --> Manifest["更新目标版本 Manifest<br/>记录 fingerprint + chunk_ids"]
    Rebuild --> Manifest
    FaqStore --> Stats["记录 FAQ 入库统计<br/>record_ingest_result()"]
    Manifest --> Stats

    Stats --> Report["生成入库质量报告<br/>空文件、低质量 chunk、重复 FAQ、source 错误、FAQ/文档冲突、OCR 风险"]
    Report --> Gate{"入库质量门禁<br/>evaluate_report_against_gate()<br/>是否通过？"}

    Gate -->|"不通过"| Hold["保留 STAGED 版本<br/>不切换 active<br/>线上继续使用旧版本"]
    Hold --> Diagnose["阅读质量报告<br/>定位 failed_files / duplicate_faq / conflicts / OCR 风险"]
    Diagnose --> Fix["修复资料或场景配置<br/>重新执行入库脚本"]
    Fix --> Start

    Gate -->|"通过但未 --activate"| KeepStaged["保存质量报告<br/>版本继续保持 STAGED<br/>可用于人工复核"]
    Gate -->|"通过 + --activate"| Active["activate_version()<br/>更新 MySQL active 指针<br/>记录激活流水"]
    Active --> Online["在线问答只读 active 版本<br/>FAQ 按 kb_version<br/>文档按 active version_seq 有效视图"]
    Online --> Verify["检索诊断/回归验收<br/>确认新资料可召回、引用可复核"]
    Verify --> Rollback{"发现线上口径问题？"}
    Rollback -->|"是"| Back["回滚到上一个已激活版本<br/>只切 MySQL active 指针"]
    Rollback -->|"否"| Done(["入库发布闭环完成"])
    Back --> Online

    style Bootstrap fill:#EEF2FF,stroke:#4F46E5,stroke-width:2px
    style Version fill:#F8FAFC,stroke:#475569,stroke-width:2px
    style Report fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style Gate fill:#FFF7ED,stroke:#EA580C,stroke-width:2px
    style Hold fill:#FEF2F2,stroke:#DC2626,stroke-width:2px
    style Active fill:#ECFDF5,stroke:#059669,stroke-width:3px
    style Done fill:#ECFDF5,stroke:#059669,stroke-width:3px

这条链路里有几个容易混淆的点：

环节 作用 解释
scenario.toml 定义业务场景、collection、数据目录、默认权限 先确定“这次给哪个业务场景建知识库”
--new-version 创建一个新的 STAGED 版本 新资料先进入候选版本，不直接覆盖线上
--force 强制重新处理数据 用于需要重新入库时跳过增量判断
--incremental-from active 基于当前 active 版本做引用式增量构建 未变化文档引用旧 chunk，变化/删除文档写失效版本，查询按 activeversion_seq解释有效视图
--reset-collections 删除并重建 Milvus collection 只在 schema 变化或旧集合不兼容时使用
FAQ 入库 把 CSV 问答对写入 FAQ collection 适合标准问答、政策口径、固定流程
文档入库 把 PDF/Word/Markdown/表格等切成 chunk 写入 Doc collection 适合长文档、制度、合同、手册
IndexManifest 在 MySQL 中记录文件指纹和 chunk ID 下次入库时判断文件是否变化，避免重复处理
质量报告 统计入库质量问题 把“知识库是否可靠”变成可检查的数据
--quality-gate 质量门禁 可以显式执行门禁；如果同时传入--activate，脚本会自动开启门禁
--activate 激活版本 只有通过门禁的新版本才进入在线链路

所以这一部分中的 FAQ 入库、文档加载、表格行入库、父子块切分、MySQL IndexManifest，都是这条总链路中的局部实现；项目阶段 14负责解释 RAG 质量评测和 Bad Case。

第二部分：FAQ 入库流程

流程位置：第 5 步“FAQ 入库”。FAQ 不走文档切分和引用式增量，而是每个目标kb_version下重新生成一份快照。

2.1 CSV 格式

FAQ 使用 CSV 文件管理，每行一个问答对：

source,question,answer
hr,入职需要准备哪些材料,入职当天需要携带：身份证原件及复印件、学历证书复印件、离职证明、体检报告、银行卡信息...
hr,试用期转正流程是什么,试用期转正流程：1. 员工提交转正申请 2. 直属领导评估 3. HR 审核 4. 部门负责人审批...
it,VPN 连接失败怎么办,请按以下步骤排查：1. 确认账号密码正确 2. 检查网络连接 3. 尝试切换 VPN 节点...
billing,如何申请发票,在订单页面点击"申请发票"，选择发票类型（电子/纸质），填写发票抬头...

2.1.1 FAQ 入库流程图

flowchart TD
    A["第5步 FAQ 入库"] --> B{"是否跳过"}
    B -->|是| Z["跳过 FAQ"]
    B -->|否| C["读取 FAQ CSV"]
    C --> D["逐行校验问答"]
    D --> E["生成 faq_id"]
    E --> F{"faq_id 重复"}
    F -->|是| D
    F -->|否| G["构建 FAQ Document"]
    G --> H["先删旧 FAQ"]
    H --> I["写入新 FAQ"]
    I --> J["记录入库结果"]

这张图要记住 3 个核心点：

  1. FAQ 是快照式重建，不是引用式增量。
  2. 每次入库都是先删旧 FAQ，再写新 FAQ。
  3. FAQ 检索按kb_version精确隔离，不走文档那套valid_from_seq / valid_to_seq。

2.2 入库实现

在完整的rebuild_kb_version.py离线编排路径中，场景、DataScope、目标版本和版本 Store 会在前置阶段统一准备，再传给 FAQ 入库函数复用。faq_documents_from_csv()仍保留独立调用时的兜底解析能力，因此这不是删除原有能力，而是避免同一次构建重复解析。

# qa_core/indexing/faq_ingestion.py

from typing import Any

def faq_documents_from_csv(
    csv_path: str,
    kb_version: str | None = None,
    version_seq: int | None = None,
    scenario_id: str | None = None,
    tenant_id: str | None = None,
    dataset_id: str | None = None,
    visibility: str | None = None,
    allowed_roles: list[str] | None = None,
    *,
    scenario: Any | None = None,
    data_scope: Any | None = None,
) -> tuple[list[Document], list[str]]:
    """把 FAQ CSV 转换为可写入 Milvus 的问题文档。

    FAQ 的 page_content 只放"标准问题"，答案放在 metadata.answer。这样检索时匹配的是
    用户问题和标准问题的相似度；一旦高置信命中，就可以直接返回 metadata.answer。
    """
    scenario = scenario if scenario is not None else resolve_scenario(scenario_id)
    data_scope = (
        data_scope
        if data_scope is not None
        else resolve_data_scope(
            tenant_id=tenant_id,
            dataset_id=dataset_id,
            visibility=visibility,
            user_roles=allowed_roles,
        )
    )
    version_meta = version_metadata(kb_version, scenario.scenario_id, version_seq=version_seq)
    data = pd.read_csv(csv_path, encoding="utf-8")
    docs: list[Document] = []
    ids: list[str] = []
    seen_ids: set[str] = set()
    for _, row in data.iterrows():
        question = str(row.get("问题") or row.get("question") or "").strip()
        answer = str(row.get("答案") or row.get("answer") or "").strip()
        subject = _resolve_csv_source(dict(row))
        if not question or not answer:
            continue

        source = normalize_faq_source(subject, scenario=scenario, question=question)
        faq_id = stable_hash(scenario.scenario_id, kb_version or "", source, question)
        if faq_id in seen_ids:
            faq_id = stable_hash(scenario.scenario_id, kb_version or "", source, question, answer)
        if faq_id in seen_ids:
            continue
        seen_ids.add(faq_id)
        docs.append(
            Document(
                page_content=question,
                metadata={
                    "faq_id": faq_id,
                    "scenario_id": scenario.scenario_id,
                    "source_type": "faq",
                    "record_type": "faq",
                    "versioning_mode": "snapshot",
                    "version_filter_mode": "kb_version_exact",
                    **data_scope.metadata(allowed_roles=allowed_roles),
                    "standard_question": question,
                    "answer": answer,
                    "source": source,
                    "subject_name": subject,
                    "status": "published",
                    **version_meta,
                },
            )
        )
        ids.append(faq_id)
    return docs, ids

存储策略： -page_content= FAQ 标准问题 → 用于向量检索 -metadata.answer= 标准答案 → 检索命中后直接取 metadata 返回 -metadata.source= 当前场景valid_sources中的标准分类 → 用于 Milvus 过滤和数据隔离 -metadata.versioning_mode = snapshot→ FAQ 是按版本快照重建，不做引用式增量复用 -metadata.version_filter_mode = kb_version_exact→ FAQ 检索按kb_version == active_version精确过滤

这样 FAQ 直出时不需要再调用 LLM，直接从 metadata 读取答案即可。

FAQ metadata 里也会出现valid_from_seq/valid_to_seq，这是因为版本字段由统一函数生成，便于质量报告和排障时看到同一套版本上下文。它不表示 FAQ 使用引用式增量；FAQ 的线上可见性由kb_version精确过滤决定，文档 chunk 的线上可见性才由有效期窗口决定。

normalize_faq_source()只依赖当前场景包的valid_sources和source_patterns。如果 CSV 中的分类无法映射到当前场景，系统会直接报错，而不是偷偷写入 Milvus。这样可以保证 FAQ 入库的业务边界和场景配置一致。

第三部分：文档加载器注册表

流程位置：第 6 步“文档入库”。加载器注册表是_rebuild_file_chunks()中load_file(path)的前置能力，负责把不同格式的原始资料统一读成 LangChainDocument。

3.1 本项目采用两层解析策略

这一部分的目标不是只把文件读成字符串，而是把资料读成后续可以治理、切分、检索和引用的Document。因此当前项目把文档解析拆成两层：

层级 负责内容 当前实现
默认解析层 Markdown、TXT、PDF 文本层、DOCX、PPTX、CSV、Excel 项目内 native loader，行为可控
增强解析层 复杂 PDF、复杂 DOCX/PPTX、HTML 等版面资料 Docling 已纳入主依赖，按配置启用后输出 Markdown，再进入同一入库链路

配置项在qa_core/config/settings.py中：

document_parser_backend: str = Field(default="native", validation_alias="DOCUMENT_PARSER_BACKEND")

取值只有两个：

值 含义 适用场景
native 默认解析路径 8 个业务场景的常规资料、稳定验收、快速部署
docling 对 PDF/DOCX/PPTX/HTML 启用 Docling 增强解析 复杂版面 PDF、图文混排、表格版式复杂的资料

注意：CSV/Excel 不会交给 Docling。本项目对业务表格采用行级 Document 设计，每一行都会保留sheet_name、row_number、表头和单元格键值。通用版面解析器可能把表格转成普通 Markdown 文本，反而削弱“按行定位、按行引用、按行回答”的业务能力。

3.2 Docling 是什么

Docling 是一个文档转换与版面理解工具。它的核心价值不是“直接做 RAG”，而是把 PDF、Word、PPT、HTML、图片等不同来源的资料先转换成统一的DoclingDocument，再导出成 Markdown、JSON、HTML 或 DocTags 等格式。官方文档中，DocumentConverter是主要入口；转换结果会包裹一个DoclingDocument，而DoclingDocument可以序列化为 Markdown 等下游更容易处理的格式。

放到 RAG 入库链路里，Docling 所在的位置非常明确：

原始复杂文档
  -> Docling DocumentConverter
  -> DoclingDocument
  -> Markdown 文本
  -> LangChain Document
  -> normalize_documents()
  -> split_documents()
  -> MilvusHybridStore.add_documents()
  -> IndexManifest / 质量报告 / 质量门禁

所以 Docling 在本项目里只负责解析阶段，不负责下面这些事情：

不负责的内容 原因
embedding / rerank 仍由 BGE-M3 和 BGE reranker 负责
Milvus 写入和混合检索 仍由MilvusHybridStore负责
metadata 标准化 仍由normalize_documents()写入 source、scenario、kb_version、DataScope
chunk 切分策略 仍由split_documents()控制父子块、表格行和 Markdown 标题增强
质量报告和质量门禁 仍由qa_core/quality和rebuild_kb_version.py收口
知识库版本激活 仍由KnowledgeBaseVersionStore.activate_version()在门禁通过后执行

这条边界很重要。Docling 能提升复杂资料解析质量，但它不能替代企业 RAG 的版本治理、质量门禁和在线检索链路。

3.3 为什么需要 Docling

项目内置 native loader 已经可以处理常见资料：Markdown、TXT、带文本层 PDF、DOCX、PPTX、CSV、Excel。它轻量、稳定、可控，适合项目默认数据和生产环境中的规范资料。

但企业资料经常不是“干净文本”：

资料问题 native loader 常见风险 Docling 的价值
PDF 多栏排版 读取顺序可能错乱 更关注页面 layout 和 reading order
PDF / Word 中复杂表格 表格可能被压成难读文本 尽量保留表格结构，再导出为 Markdown
PPT 图文混排 文本框、表格、标题层级容易丢失 统一转换成结构化文档表示
HTML 页面 需要保留标题、段落和表格结构 转成统一 Markdown 后进入同一入库链路

因此，本项目把 Docling 作为增强解析后端：默认仍走 native，遇到复杂版面资料时才切换到DOCUMENT_PARSER_BACKEND=docling。这样既能保持主链路简单，也能给企业真实资料留出增强入口。

参考资料：

3.4 注册表设计

# qa_core/indexing/document_loaders.py

DOCLING_SUFFIXES = {".pdf", ".docx", ".pptx", ".html", ".htm"}

DOCUMENT_LOADER_SPECS: tuple[DocumentLoaderSpec, ...] = (
    DocumentLoaderSpec(
        suffixes=(".txt", ".md"),
        factory=_utf8_text_loader,
        description="UTF-8 文本和 Markdown；Markdown 保留原文给标题切分器处理。",
    ),
    DocumentLoaderSpec(
        suffixes=(".pdf",),
        factory=_pdf_loader,
        description="PDF 文本层解析；扫描件 OCR 不默认进入主链路。",
    ),
    DocumentLoaderSpec(
        suffixes=(".docx", ".doc"),
        factory=_word_loader,
        description="Word 文档文本解析。",
    ),
    DocumentLoaderSpec(
        suffixes=(".ppt", ".pptx"),
        factory=_powerpoint_loader,
        description="PowerPoint 文本解析。",
    ),
    DocumentLoaderSpec(
        suffixes=(".html", ".htm"),
        factory=_docling_loader,
        description="HTML 资料解析；需要 DOCUMENT_PARSER_BACKEND=docling。",
    ),
    DocumentLoaderSpec(
        suffixes=(".csv", ".xlsx", ".xls"),
        factory=_table_loader,
        description="CSV/Excel 表格解析；按行保留表头、sheet 和单元格键值。",
    ),
)

load_file()的核心判断是：先按后缀确认文件受支持，再按配置决定是否启用 Docling。

def _use_docling_for(path: Path) -> bool:
    return get_settings().document_parser_backend == "docling" and path.suffix.lower() in DOCLING_SUFFIXES

def load_file(path: Path) -> list[Document]:
    spec = get_document_loader_spec(path)
    if spec is None:
        raise ValueError(f"不支持的文档类型：{path}")
    if _use_docling_for(path):
        return _docling_loader(path).load()
    return spec.create_loader(path).load()

这段代码有两个边界：

  • DOCUMENT_PARSER_BACKEND=docling只影响 PDF/DOCX/PPTX/HTML/HTM；
  • CSV/XLSX/XLS 始终走_table_loader()，保证表格行 metadata 不丢失。

还有两个格式细节要注意：

  • .html和.htm都在DOCLING_SUFFIXES中，启用 Docling 后都走 HTML/Docling 增强解析路径。
  • .doc和.ppt虽然在 loader registry 里注册了后缀，但当前 loader 会提示先转换成.docx/.pptx后入库；质量报告里更可能归为解析失败，而不是“未注册格式”。

3.5 Docling 增强解析如何接入

Docling loader 只负责把复杂版面资料转换成 Markdown。转换后仍然返回 LangChainDocument，后续的normalize_documents()、split_documents()、MilvusHybridStore.add_documents()、IndexManifest 和质量门禁都不需要改。

class DoclingLoader:
    """Docling 增强 loader，把复杂版面资料转换成 Markdown 后进入统一入库链路。"""

    def __init__(self, path: Path) -> None:
        self.path = path

    def load(self) -> list[Document]:
        if (
            importlib.util.find_spec("docling") is None
            or importlib.util.find_spec("docling.document_converter") is None
        ):
            raise RuntimeError("Docling 解析后端不可用：未安装 docling。")

        converter_module = importlib.import_module("docling.document_converter")
        result = converter_module.DocumentConverter().convert(str(self.path))
        content = result.document.export_to_markdown().strip()
        metadata = {
            "file_type": self.path.suffix.lower(),
            "parser_backend": "docling",
            "docling_format": "markdown",
        }
        return [Document(page_content=content, metadata=metadata)] if content else []

实际代码使用importlib做显式动态加载：只有配置为docling且文件类型需要增强解析时，才加载 Docling。这样默认运行路径不会提前初始化 Docling 的模型和依赖；如果当前环境没有按项目主依赖完整安装，会直接报明确错误，不会静默退回 native 路径。

3.6 如何启用 Docling

本机开发环境：

pip install -r requirements.txt
$env:DOCUMENT_PARSER_BACKEND="docling"
python scripts/tools/docling_parser_smoke.py
python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --force --quality-gate --activate --description "docling parser rebuild"

Docker Compose 环境：

# 使用完整项目依赖构建 API 镜像
docker compose --env-file .env.compose build api

# 在 .env.compose 中设置
# DOCUMENT_PARSER_BACKEND=docling

docker compose --env-file .env.compose run --rm api python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --force --quality-gate --activate --description "docling parser rebuild"

如果只是确认 Docling 是否能在当前环境解析文件，可以先运行：

python scripts/tools/docling_parser_smoke.py

这个命令会生成一个临时 HTML 样例，只检查load_file()是否能走 Docling 后端，不写 Milvus，也不创建知识库版本。也可以指定真实资料：

python scripts/tools/docling_parser_smoke.py --input 你的复杂版面资料.pdf

如果只是使用 8 个默认业务场景和当前已整理好的多格式资料，保持：

DOCUMENT_PARSER_BACKEND=native

即可。native 路径已经覆盖 Markdown、TXT、带文本层 PDF、DOCX、PPTX、CSV、XLSX、XLS，并且更适合稳定验收。

3.7 为什么不把所有文件都交给 Docling

文件类型 当前选择 原因
Markdown / TXT native 已经是可控文本，保留原始标题和段落结构即可
CSV / Excel native table loader 必须保留表头、sheet、行号、单元格键值，便于行级检索和引用
带文本层 PDF native 或 Docling 普通制度 PDF 可走 native；复杂版面 PDF 可切 Docling
DOCX / PPTX native 或 Docling 普通资料走 native；复杂版式或多表格版面可切 Docling
扫描件 / 图片件 先走 OCR 复核流程 不能把低置信 OCR 结果直接写入 active 知识库

所以当前项目不是“native 和 Docling 二选一”，而是：

默认资料：native，简单、稳定、可控
复杂版面：Docling，增强解析能力
结构化表格：项目行级 loader，保留业务语义
扫描件：OCR 复核治理，避免污染 active 知识库

3.8 为什么没有直接使用 LlamaIndex 入库流水线

LlamaIndex 的SimpleDirectoryReader可以快速读取本地目录文件，IngestionPipeline可以把 transformations、embedding、缓存和向量库写入串起来。这些能力适合快速搭建 RAG 数据接入原型，也适合作为企业项目后续优化方向。

但本项目没有把 LlamaIndex 接入这一部分主代码，原因是这一部分要讲清楚的是企业知识库入库治理，而不只是“把文件变成向量”。

对比点 本项目当前实现 如果直接换成 LlamaIndex
主线 load_file -> normalize_documents -> split_documents -> add_documents -> manifest -> quality report每一步都和代码对齐 会额外引入Document / Node / Index / QueryEngine / IngestionPipeline等概念
版本治理 每个 chunk 显式写入scenario_id / kb_version / source / DataScope 仍需要自己把这些 metadata 接回企业治理链路
增量机制 MySQLIndexManifest明确记录文件指纹和 chunk_id LlamaIndex cache 能减少重复 transformation，但不能直接替代 active 版本发布和质量门禁
表格资料 自定义 table loader 按表头、工作表、行号生成行级 Document 通用 loader 未必能保留本项目需要的业务 metadata 粒度
质量报告 入库后生成空文件、重复 chunk、FAQ 冲突、OCR 风险等报告 质量指标仍然要由项目自己定义和落库

因此当前代码保持显式实现：文档解析使用 native + Docling 可选增强，切分和向量化使用 LangChain/Milvus，版本和质量治理由qa_core自己掌控。这样做的好处是每个字段为什么存在、每个步骤为什么执行，都能和后续检索、评测、回滚闭环对齐。

当前代码和依赖保持一致：requirements.txt不包含llama-index，qa_core主链路也不导入llama_index。

第四部分：文档入库主流程

流程位置：第 6 步“文档入库”。这一部分展开ingest_directory()和_ingest_single_file()，也就是主流程图里最重的文档入库阶段。

4.1 ingest_directory() 完整流程

flowchart TD
    Start(["ingest_directory()<br/>目录路径 + 场景 + 版本"]) --> Version["📋 创建/确认 STAGED KB 版本<br/>KnowledgeBaseVersionStore"]

    Version --> Loop["📂 遍历目录文件"]

    Loop --> Ext{"文件后缀<br/>在注册表中？"}

    Ext -->|"❌"| Skip1["⚠️ 跳过<br/>不支持的文件类型"]
    Ext -->|"✅"| Fingerprint["🔍 计算本地文件指纹<br/>绝对路径 + 修改时间 + 文件大小"]

    Fingerprint --> Check{"Manifest 中<br/>指纹未变化？<br/>且非 force 模式"}

    Check -->|"✅ 未变化"| Skip2["⏭️ 增量跳过<br/>不重复入库"]
    Check -->|"❌ 已变化/新文件"| Load["📄 DocumentLoader 加载<br/>PDF→PyMuPDF<br/>MD→TextLoader<br/>XLSX→TableLoader"]

    Load --> Normalize["🏷️ normalize_documents<br/>补充 source/kb_version/<br/>tenant_id/data_scope"]

    Normalize --> Split["✂️ split_documents<br/>Markdown标题增强<br/>父子块切分"]

    Split --> Delete["🗑️ 删除旧 chunk_ids<br/>(如果存在)"]

    Delete --> Write["💾 Milvus add_documents<br/>BGE-M3 生成 Dense 向量<br/>Milvus 生成 BM25 Sparse"]

    Write --> Manifest["📝 更新 MySQL IndexManifest<br/>记录指纹+chunk_ids"]

    Manifest --> Loop

    Skip1 --> Loop
    Skip2 --> Loop

    Loop --> Done(["✅ 返回文档写入统计<br/>doc_chunks_written"])

    Done --> Script["rebuild_kb_version.py<br/>汇总 FAQ + 文档入库结果"]
    Script --> Report["📊 生成入库质量报告<br/>build_ingestion_quality_report()"]
    Report --> Gate{"🚦 入库质量门禁<br/>evaluate_report_against_gate()"}
    Gate -->|"❌ 不通过"| Staged["保留 STAGED 版本<br/>不切换 active<br/>线上仍用旧版本"]
    Gate -->|"✅ 通过 + --activate"| Active["激活新版本<br/>activate_version()<br/>更新 MySQL active 指针"]
    Gate -->|"✅ 通过但未 --activate"| KeepStaged["保存质量报告<br/>版本继续保持 STAGED"]

    style Load fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style Split fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style Write fill:#FFFBEB,stroke:#D97706,stroke-width:2px
    style Done fill:#ECFDF5,stroke:#059669,stroke-width:3px
    style Report fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style Gate fill:#FFF7ED,stroke:#EA580C,stroke-width:2px
    style Active fill:#ECFDF5,stroke:#059669,stroke-width:3px
    style Staged fill:#FEF2F2,stroke:#DC2626,stroke-width:2px
    style KeepStaged fill:#F8FAFC,stroke:#64748B,stroke-width:2px

代码执行时序图

这张图对应离线入库主入口qa_core/indexing/service.py::ingest_directory()。它最适合按“版本确认 -> 遍历文件 -> 单文件增量判断 -> 写 Milvus -> 回写清单”的顺序阅读。

sequenceDiagram
    autonumber
    participant CLI as 入库脚本
    participant Svc as ingest_directory()
    participant Ver as KnowledgeBaseVersionStore
    participant Man as IndexManifest
    participant Loader as load_file()
    participant Norm as normalize_documents()
    participant Split as split_documents()
    participant Store as doc_store.add_documents()
    participant DB as Milvus/MySQL

    CLI->>Svc: 传入目录 + 已解析的场景、数据域、目标版本
    Svc->>Ver: 复用 version_store/version<br/>独立调用时才 resolve/ensure
    Ver-->>Svc: active_kb_version + version_seq
    Svc->>Man: 创建 manifest
    loop 每个文件
        Svc->>Svc: _ingest_single_file(path, context)
        alt 文件后缀不支持 / 指纹未变化
            Svc-->>CLI: 跳过该文件
        else 需要重建或首次入库
            Svc->>Loader: load_file(path)
            Loader-->>Svc: Documents
            Svc->>Norm: normalize_documents(...)
            Norm-->>Svc: 标准化 Documents
            Svc->>Split: split_documents(...)
            Split-->>Svc: chunks + chunk_ids
            Svc->>DB: delete 旧 chunk_id（如存在）
            Svc->>Store: add_documents(chunks)
            Store-->>Svc: 写入完成
            Svc->>Man: update(record_manifest)
        end
    end
    Svc->>DB: _expire_missing_base_records()
    Svc->>Ver: record_ingest_result(...)
    Ver-->>Svc: 入库统计写回 MySQL

读图重点：

  1. 先确认版本和数据域，再开始遍历目录文件。
  2. 单文件是否重建由指纹和 manifest 决定，不是每次都全量写。
  3. 写入 Milvus 后必须回写 manifest，才能支撑下次跳过和版本回滚。

单独查看：打开可缩放时序图。

FAQ CSV 的链路是另一条更短的路：ingest_faq_csv()直接把 CSV 转成 FAQ 文档，再执行删除旧 FAQ + 批量写入 + 记录版本统计，不走普通文档切分。

4.1.1 质量报告在真实代码中的位置

质量报告不在ingest_directory()内部生成，而是在脚本完成 FAQ 和所有文档入库之后，由rebuild_kb_version.py::main()的第七步统一收口：

# scripts/rebuild_kb_version.py
report = build_ingestion_quality_report(...)
report["actual_ingest"] = {
    "faq_records_written": faq_count,
    "doc_chunks_written": doc_chunks,
    "activated": False,
}

if args.quality_gate:
    gate_result = evaluate_report_against_gate(report, quality_thresholds_from_args(args))
    report["quality_gate"] = gate_result
    if not gate_result["ok"]:
        report_path = save_ingestion_quality_report(report)
        sys.exit(1)

if args.activate:
    version_store.activate_version(kb_version)
    report["actual_ingest"]["activated"] = True

report_path = save_ingestion_quality_report(report)

这里有三个必须讲清楚的边界：

  1. build_ingestion_quality_report()会重新解析和切分候选资料，检查文件、chunk、FAQ、冲突和 OCR 风险，但它本身不写 Milvus；本次真实写入数量通过actual_ingest统计附加到报告中。
  2. evaluate_report_against_gate()只负责把报告中的数量与阈值比较。门禁失败时保存报告并退出，--activate不会执行，线上继续使用旧的 active 版本。
  3. 只有门禁通过且显式传入--activate，才调用activate_version()更新 MySQL 的 active/previous 指针。通过门禁但没有--activate时，版本仍然保持STAGED，用于人工复核。

因此，这一部分的“质量报告”回答的是资料能不能发布，不是“用户问题的召回率和答案是否正确”。后者要在版本激活后由项目阶段 14的 RAG Evaluation 验证，项目阶段 15再用接口和回归测试证明代码、接口和既有功能没有被改坏。

# qa_core/indexing/service.py
def ingest_directory(
    directory_path: str,
    source: str | None = None,
    *,
    scenario_id: str | None = None,
    tenant_id: str | None = None,
    dataset_id: str | None = None,
    visibility: str | None = None,
    allowed_roles: list[str] | None = None,
    force: bool = False,
    kb_version: str | None = None,
    create_new_version: bool = False,
    description: str = "",
    incremental_base_kb_version: str | None = None,
    scenario: Any | None = None,
    data_scope: Any | None = None,
    target_version: Any | None = None,
    version_store: Any | None = None,
    incremental_base_version_seq: int | None = None,
) -> int:
    """把目录中的业务文档增量写入 Milvus，逐文件委托 _ingest_single_file 处理。"""

    # Step 1：解析或复用场景、构建或复用数据域、确定业务分类
    scenario = scenario if scenario is not None else resolve_scenario(scenario_id)
    data_scope = (
        data_scope
        if data_scope is not None
        else resolve_data_scope(
            tenant_id=tenant_id,
            dataset_id=dataset_id,
            visibility=visibility,
            user_roles=allowed_roles,
        )
    )
    root = Path(directory_path)
    resolved_source = source or normalize_source_from_path(root)
    if resolved_source not in scenario.valid_sources:
        raise ValueError(f"无效的业务分类：{resolved_source}")

    # Step 2：创建/确认知识库版本
    version_store = (
        version_store
        if version_store is not None
        else get_kb_version_store(scenario.scenario_id)
    )
    version = (
        target_version
        if target_version is not None
        else version_store.ensure_version(
            kb_version,
            create_new=create_new_version,
            description=description,
            created_by="ingest_directory",
        )
    )
    active_kb_version = version.kb_version
    if incremental_base_version_seq is None:
        incremental_base_version_seq = _resolve_incremental_base_version_seq(
            version_store,
            incremental_base_kb_version,
        )

    # Step 3：打开增量清单 + 文档存储，并组装单次入库上下文
    context = DocumentIngestContext(
        source=resolved_source,
        kb_version=active_kb_version,
        kb_version_seq=version.version_seq,
        scenario=scenario,
        data_scope=data_scope,
        allowed_roles=allowed_roles,
        doc_store=get_doc_store(scenario.doc_collection),
        manifest=IndexManifest(),
        chunk_index=ChunkVersionIndex(),
        force=force,
        incremental_base_kb_version=incremental_base_kb_version,
        incremental_base_version_seq=incremental_base_version_seq,
    )

    # Step 4：遍历目录，逐个文件委托给 _ingest_single_file
    stats = DirectoryIngestStats()
    for path in _walk_files(root):
        stats.add(_ingest_single_file(path, context))

    # Step 5：记录入库统计
    version_store.record_ingest_result(
        active_kb_version, content_type="doc",
        count=stats.total_chunks, source=resolved_source,
        extra_stats=stats.as_version_stats(incremental_base_kb_version),
    )

    # Step 6：入库函数只写入和记录统计，不直接激活。
    # 版本发布由 rebuild_kb_version.py 生成质量报告、执行门禁后统一切换 active。

    return stats.total_chunks

_ingest_single_file()负责单个文件的增量入库逻辑，被ingest_directory的循环调用：

def _ingest_single_file(path: Path, context: DocumentIngestContext) -> FileIngestResult:
    """处理单个文件：目标版本跳过、基准版本复用、变化文件重建。"""
    if get_document_loader_spec(path) is None:
        raise ValueError(f"不支持的文档类型：{path}")
    fingerprint = file_fingerprint(path)
    settings = get_settings()
    existing = context.manifest.get(context.source, path, context.kb_version, context.scenario_id)

    if not context.force and _manifest_matches_current_settings(existing, fingerprint, settings):
        return FileIngestResult(skipped=True)

    base_record = None
    if context.incremental_base_kb_version and not context.force:
        base_record = context.manifest.get(
            context.source,
            path,
            context.incremental_base_kb_version,
            context.scenario_id,
        )
    if not existing and base_record and _manifest_matches_current_settings(base_record, fingerprint, settings):
        _record_manifest(context, path, fingerprint, base_record.chunk_ids, settings)
        return FileIngestResult(reused_chunks=len(base_record.chunk_ids))

    expired_chunks = 0
    if base_record:
        expired_chunks = _expire_base_record_for_target(context, base_record)
    return _rebuild_file_chunks(
        context,
        path,
        fingerprint,
        existing,
        settings,
        expired_chunks=expired_chunks,
    )

4.2 normalize_documents 的作用

def normalize_documents(
    documents: list[Document],
    file_path: Path,
    source: str,
    kb_version: str | None = None,
    scenario_id: str | None = None,
    version_seq: int | None = None,
    data_scope: DataScope | None = None,
    allowed_roles: list[str] | None = None,
    *,
    scenario: Any | None = None,
) -> list[Document]:
    """为文档补充项目标准元数据，供过滤和引用使用。"""
    doc_id = file_fingerprint(file_path)
    scenario = scenario if scenario is not None else resolve_scenario(scenario_id)
    scope = data_scope if data_scope is not None else resolve_data_scope()
    version_meta = version_metadata(kb_version, scenario.scenario_id, version_seq=version_seq)
    normalized: list[Document] = []
    for index, doc in enumerate(documents):
        metadata = dict(doc.metadata or {})
        metadata.update(
            {
                "source": source,
                "scenario_id": scenario.scenario_id,
                **scope.metadata(allowed_roles=allowed_roles),
                "file_path": str(file_path),
                "file_name": file_path.name,
                "file_type": file_path.suffix.lower(),
                "doc_id": doc_id,
                "page_index": metadata.get("page", index),
                "content_type": metadata.get("content_type") or "text",
                **version_meta,
            }
        )
        normalized.append(Document(page_content=doc.page_content, metadata=metadata))
    return normalized

4.3 split_documents() 的项目策略

normalize_documents()输出统一 metadata 后，split_documents()根据资料类型选择真实切分路径：

# qa_core/indexing/chunking.py
for doc in documents:
    file_type = str(doc.metadata.get("file_type", "")).lower()

    if is_table_metadata(doc.metadata) or is_reviewed_ocr_metadata(doc.metadata):
        parent_content = doc.page_content.strip()
        if not parent_content:
            continue
        metadata = dict(doc.metadata)
        metadata["parent_content"] = parent_content
        parent_id, chunk_id = chunk_identity(parent_content, metadata)
        metadata.update({"parent_id": parent_id, "chunk_id": chunk_id})
        chunks.append(Document(page_content=parent_content, metadata=metadata))
        ids.append(chunk_id)
        continue

    if file_type == ".md":
        header_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
        )
        header_docs = header_splitter.split_text(doc.page_content)
        for header_doc in header_docs:
            header_doc.metadata.update(doc.metadata)
        parent_docs = parent_splitter.split_documents(header_docs)
    else:
        parent_docs = parent_splitter.split_documents([doc])

    for parent_doc in parent_docs:
        parent_content = parent_doc.page_content
        for child_doc in child_splitter.split_documents([parent_doc]):
            metadata = dict(child_doc.metadata)
            metadata["parent_content"] = parent_content
            parent_id, chunk_id = chunk_identity(child_doc.page_content, metadata)
            metadata.update({"parent_id": parent_id, "chunk_id": chunk_id})
            chunks.append(Document(page_content=child_doc.page_content, metadata=metadata))
            ids.append(chunk_id)

这段代码表达的是项目策略，而不是新的 LangChain 概念：Markdown 先保留标题结构，普通文本执行父子块切分，表格行和已复核 OCR 保持完整；最终为每个块生成稳定的parent_id/chunk_id。完整源码见qa_core/indexing/chunking.py，chunk size 与 overlap 的原理见附录 G。

第五部分：表格 CSV / Excel 专用入库设计

流程位置：第 6 步“文档入库”。表格不是独立发布链路，而是文档入库阶段里一种特殊 loader 和特殊 chunking 策略。

5.1 为什么表格不能按普通文本切分

普通制度、流程、手册是一段段自然语言，适合用 Parent-Child Chunking 按章节和字符长度切分。

但 CSV / Excel 表格不是自然段，而是一条条行记录。一行里多个单元格共同表达一个完整业务事实：

材料名称=施工照片
状态=待补交
责任人=项目经理
截止日期=2026-05-30

如果把表格当普通文本递归切分，可能出现：

  • 检索命中了“施工照片”，但状态被切到另一个 chunk；
  • 检索命中了“金额”，但付款节点、责任人丢失；
  • 两行不同记录被拼到同一个 chunk，答案把 A 行状态说成 B 行状态；
  • 答案引用只能定位到文件，不能定位到工作表和行号。

所以本项目对表格资料的原则是：

一行表格 = 一个完整业务语义单元。

5.2 文件读取策略

表格文件在 Loader 注册表中作为独立类型接入：

# qa_core/indexing/document_loaders.py
DocumentLoaderSpec(
    suffixes=(".csv", ".xlsx", ".xls"),
    factory=_table_loader,
    description="CSV/Excel 表格解析；按行保留表头、sheet 和单元格键值。",
)

读取规则：

文件类型 读取方式 说明
.csv pandas.read_csv(..., encoding="utf-8-sig") 兼容带 BOM 的中文 CSV
.xlsx pandas.read_excel(..., sheet_name=None, engine="openpyxl") 一次读取全部工作表
.xls pandas.read_excel(..., sheet_name=None, engine="xlrd") 兼容旧版 Excel

Excel 会逐个 sheet 处理，避免把多个业务表混成一张表。

5.3 表格清洗

表格入库前先做轻量清洗：

def _normalize_frame(frame: pd.DataFrame) -> pd.DataFrame:
    """清理表格空行空列，并把缺失表头补成稳定列名。"""
    data = frame.dropna(how="all").dropna(axis=1, how="all").fillna("")
    columns = []
    for index, column in enumerate(data.columns, start=1):
        name = str(column).strip()
        if not name or name.lower().startswith("unnamed:"):
            name = f"列{index}"
        columns.append(name)
    data.columns = columns
    return data

清洗目标不是复杂 ETL，而是保证表格行进入 RAG 时不会因为空行、空列表头、Unnamed列名造成检索噪声。

单元格值也会转成适合检索的短文本：

def _cell_text(value: object) -> str:
    text = str(value).strip()
    if text.endswith(".0") and text[:-2].isdigit():
        return text[:-2]
    return text

这样1000.0会变成1000，金额、编号、数量类问题更容易命中。

5.4 每行转换为 Document

表格 loader 会把每一行转换成一个 LangChainDocument：

content = "\n".join(
    [
        f"表格文件：{path.name}",
        f"工作表：{sheet_name}",
        f"表头：{' / '.join(headers)}",
        f"行号：{row_number}",
        "单元格：",
        *cell_lines,
    ]
)

生成后的正文类似：

表格文件：验收清单.xlsx
工作表：材料验收
表头：材料名称 / 状态 / 责任人 / 截止日期
行号：3
单元格：
- 材料名称：施工照片
- 状态：待补交
- 责任人：项目经理
- 截止日期：2026-05-30

这样做有两个好处：

  1. 语义完整：同一行的字段和值不会被拆散；
  2. 适合向量检索和 BM25：既有自然语言标签，也有明确的列名和值。

5.5 metadata 设计

表格行必须携带可追溯 metadata：

metadata={
    "content_type": "table_row",
    "table_id": table_id,
    "sheet_name": str(sheet_name),
    "row_number": row_number,
    "row_count": len(normalized),
    "column_count": len(headers),
    "table_headers": " | ".join(headers),
}

字段含义：

字段 作用
content_type=table_row 告诉切分、质量检测、检索上下文：这是表格行
table_id 标识同一个文件下的同一个工作表
sheet_name 支持答案引用到具体工作表
row_number 支持答案引用到具体行
row_count/column_count 质量报告和容量评估使用
table_headers 帮助回看表结构，也便于后续扩展表头召回

5.6 表格行不再递归切分

split_documents()会识别content_type=table_row：

# qa_core/indexing/chunking.py
if is_table_metadata(doc.metadata):
    parent_docs = [doc]
else:
    parent_docs = parent_splitter.split_documents([doc])

也就是说，表格行不会再进入普通字符切分器。

原因是：表格行已经是完整业务单元，再切一次反而会破坏“列名 -> 单元格值”的关系。

5.7 检索策略中的 prefer_table

表格入库只是第一步。检索时还要识别用户是否在问表格问题。

本项目通过is_table_query()判断问题是否包含表格、清单、台账、字段、行号、工作表、状态、金额、责任人等表达：

prefer_table = is_table_query(compact_query)
params = _apply_table_preference(prefer_table, params["run_doc"], params, settings)

当prefer_table=True时：

  • 扩大doc_top_k，多召回一些候选表格行；
  • 扩大final_context_top_n，给表格证据更多上下文空间；
  • 设置faq_direct_exact_only=True，禁止相似 FAQ 直接回答；
  • 上下文构建时把表格行排在普通正文前。

为什么要禁用相似 FAQ 直出？

用户问：验收材料清单里测试报告那一行是什么状态？
相似 FAQ：验收需要提交哪些材料？

这两个问题都包含“验收”“材料”“测试报告”，相似度可能不低。
但 FAQ 回答的是材料范围，用户问的是某一行字段值。
所以表格类问题只允许精确 FAQ 直出，相似 FAQ 必须让位给文档 RAG。

5.8 答案引用和兜底

表格资料的答案必须能回到原始证据。当前项目在来源标签中追加工作表和行号：

[1] 验收清单.xlsx / 工作表：材料验收 / 第 3 行

另外，表格类问题经常涉及状态、金额、责任人、日期等精确值。LLM 有时会概括回答而漏掉某个关键单元格，所以项目里增加了表格行兜底：

def enforce_table_row_details(answer: str, context_docs: list[Document]) -> str:
    """确保表格类答案在模型遗漏关键单元格时，确定性追加表格行要点。"""

如果模型回答没有覆盖表格行里的核心字段，系统会追加：

表格行要点：状态：待补交；责任人：项目经理 [1]

这不是替代 LLM，而是对表格精确字段的一层确定性保护。

5.9 表格入库设计要点

Excel 和 CSV 入库可以概括为：

Excel 和 CSV 不能按普通文本切分。我们把每一行转成一个带表头、工作表、行号和单元格键值的 LangChain Document，并写入content_type=table_row。切分阶段识别到表格行后不会再递归切分；检索阶段如果问题命中表格、清单、台账、金额、状态等关键词，会启用prefer_table，扩大文档召回并优先保留表格行。答案引用会展示文件、工作表和行号，如果模型漏掉关键单元格，系统会追加表格行要点，保证表格类问题能追溯、能复核、字段不丢。

5.10 表格入库完整示例

下面用最小数据演示“表格读取 → 行级 Document → 检索偏好 → 答案引用”的完整处理方式。

准备一个最小 CSV：

材料名称,状态,责任人,截止日期,备注
施工图纸,已提交,设计负责人,2026-05-10,版本为 V3
隐蔽工程照片,待补交,项目经理,2026-05-18,缺少二层西侧照片
验收测试报告,已通过,质量负责人,2026-05-20,检测编号 QA-2026-021

下面这段代码用于在本地快速验证表格 loader 和切分策略。它不连接 Milvus，也不会改动线上知识库，只检查三件事：

  • load_table_file()是否把 CSV 每一行转换成一个Document；
  • split_documents()是否保持表格行完整，不再递归切分；
  • is_table_query()是否能把表格类问题识别为prefer_table=True。
from pathlib import Path

from qa_core.indexing.chunking import split_documents
from qa_core.indexing.table_documents import load_table_file
from qa_core.intent.question_category import is_table_query

csv_path = Path("reports/table_practice/acceptance_material_checklist.csv")
csv_path.parent.mkdir(parents=True, exist_ok=True)
csv_path.write_text(
    "\n".join(
        [
            "材料名称,状态,责任人,截止日期,备注",
            "施工图纸,已提交,设计负责人,2026-05-10,版本为 V3",
            "隐蔽工程照片,待补交,项目经理,2026-05-18,缺少二层西侧照片",
            "验收测试报告,已通过,质量负责人,2026-05-20,检测编号 QA-2026-021",
        ]
    ),
    encoding="utf-8-sig",
)

documents = load_table_file(csv_path)
print("行级 Document 数量：", len(documents))
print("第一条 Document 正文：")
print(documents[0].page_content)
print("第一条 Document metadata：")
print(documents[0].metadata)

chunks, ids = split_documents(documents)
print("切分后 chunk 数量：", len(chunks))
print("chunk_id 示例：", ids[0])
print("第二条 chunk 正文：")
print(chunks[1].page_content)
print("第二条 chunk metadata：")
print(chunks[1].metadata)

query = "验收清单里隐蔽工程照片是什么状态，责任人是谁？"
print("是否表格类问题：", is_table_query(query))

运行时应该看到：

行级 Document 数量： 3
切分后 chunk 数量： 3
是否表格类问题： True

第二条 chunk 的正文应该仍然保留完整行记录，类似：

表格文件：acceptance_material_checklist.csv
工作表：csv
表头：材料名称 / 状态 / 责任人 / 截止日期 / 备注
行号：2
单元格：
- 材料名称：隐蔽工程照片
- 状态：待补交
- 责任人：项目经理
- 截止日期：2026-05-18
- 备注：缺少二层西侧照片

metadata 中至少要看到这些字段：

{
  "content_type": "table_row",
  "sheet_name": "csv",
  "row_number": 2,
  "row_count": 3,
  "column_count": 5
}

如果要把这个 CSV 真正放进知识库，可以把文件移动到某个场景的数据目录，例如：

scenarios/engineering_project_qa/data/quality_data/acceptance_material_checklist.csv

然后执行单场景重建：

python scripts/rebuild_kb_version.py --scenario engineering_project_qa --new-version --force --quality-gate --activate --description "table row ingestion practice"

入库后可以用检索诊断或页面提问：

验收清单里隐蔽工程照片是什么状态，责任人是谁？

期望链路是：

问题命中 prefer_table=True
  ↓
文档检索优先保留 table_row
  ↓
答案引用能定位到 CSV 文件、工作表 csv、第 2 行
  ↓
如果模型漏掉状态或责任人，后处理追加表格行要点

建议把它放到工程项目资料问答场景的数据目录中，并按常规知识库重建流程入库。重点观察四件事：

检查点 期望结果 为什么检查
入库后的 metadata 包含content_type=table_row、sheet_name、row_number 证明表格行没有被当成普通正文。
chunk 数量 每个有效数据行生成一个可检索Document 证明行级证据粒度正确。
检索计划 表格类问题命中prefer_table=True 证明检索策略知道当前问题更适合查表格。
答案来源 来源中能看到文件、工作表、行号 证明答案可以回到原始证据复核。

可以在页面或接口中提问：

验收清单里隐蔽工程照片是什么状态，责任人是谁？

理想回答应该包含：

  • 状态是“待补交”；
  • 责任人是“项目经理”；
  • 引用来源能定位到 CSV/Excel 的对应行；
  • 如果模型遗漏状态或责任人，系统会追加“表格行要点”。

这个验证实践的目的不是测试模型文采，而是验证表格证据没有在切分和生成阶段丢失。

5.11 当前边界

一期表格入库只覆盖“规范二维表”。复杂 Excel 能力不能无边界扩散，否则会把 RAG 项目变成 Office 解析项目。

边界场景 一期处理策略 推荐做法
合并单元格 不默认还原层级语义 入库前整理成普通二维表。
多级表头 不自动推断复杂表头关系 人工扁平化字段名，比如“合同-金额”“合同-付款节点”。
公式单元格 读取解析后的单元格值，不重新计算业务公式 关键计算逻辑应在业务系统或数据准备阶段完成。
图表 不把柱状图、折线图直接转成结构化证据 导出图表背后的原始数据表再入库。
截图表格 不走 CSV/Excel 表格 loader 进入 OCR/VLM 图文资料治理链路。
超大 Excel 不在一期做复杂分布式解析 拆分工作表、拆分文件，或按业务周期归档。
隐藏行列和批注 不作为可信主证据 重要内容必须整理成显式列。
透视表 不直接作为原始证据 导出明细表或汇总表后再入库。

这部分能力可以概括为：

我们一期支持的是规范 CSV/Excel 的行级语义入库，不追求解析所有复杂 Office 特性。这样做是为了保证 RAG 主链路清晰可控：表格行能召回、字段能引用、来源能复核。合并单元格、截图表格、图表解释这类复杂资料会进入后续 OCR/VLM 和资料治理链路，而不是塞进普通表格 loader 里。

第六部分：MySQL IndexManifest 增量机制

流程位置：第 4 步和第 6 步。第 4 步确定增量基准版本，第 6 步用IndexManifest判断单文件应该跳过、复用、重建还是过期。

6.1 先分清两个“版本”

分析增量入库时，最容易混淆的是“知识库版本”和“Manifest 记录”。

概念 保存位置 作用
知识库版本kb_version MySQLkb_versions/kb_active_versions 控制线上当前查哪个知识库版本，例如kb_enterprise_knowledge_20260618_xxx。
Manifest 记录 MySQLkb_document_manifests 记录某个文件在某个kb_version下生成了哪些 chunk，用于判断下次是否可以跳过、复用或重建。
Milvus chunk Milvus collection 保存真正用于检索的文本、向量和 metadata。

一句话概括：

kb_version决定线上查哪一版；IndexManifest记录每一版里每个文件对应哪些 chunk。

当前项目的线上检索始终只查一个 active 版本：

valid_from_seq <= active_seq
and (valid_to_seq == 0 or valid_to_seq > active_seq)

它不是查询时把“旧版本 + 增量版本”拼起来查。增量发生在离线构建阶段，最终仍然产出一个完整的新kb_version。

6.2 为什么需要 Manifest

假设一个业务场景有 500 个文档，只改了其中 1 个文件。如果每次都全量重建，会有三个问题：

问题 后果
时间浪费 499 个未变化文件重复解析、切分、embedding。
成本浪费 embedding 和 Milvus 写入重复执行。
发布风险 全量重建过程中任何一步失败，都可能影响新版本发布。

Manifest 要解决的是三个判断问题：

判断问题 处理动作
目标版本里这个文件已经入过库，而且文件、embedding 模型、切分策略都没变 直接跳过。
新建目标版本时，基准版本里这个文件没变，而且模型和切分策略也没变 目标版本 manifest 引用旧 chunk，不复制、不重新 embedding。
文件内容变化、新增文件、embedding 模型变化、切分策略变化 重新加载、切分、embedding、写入 Milvus。

所以它不是“简单记录文件名”，而是离线入库的决策依据。

6.3 Manifest 表结构

当前项目把文档增量清单保存到 MySQL 表kb_document_manifests。这张表记录“某个文件在某个场景、某个知识库版本下生成了哪些 chunk”。

字段 说明
manifest_key scenario_id + source + kb_version + 文件绝对路径的稳定 hash
scenario_id 业务场景
source 业务分类
path 本地文件绝对路径
fingerprint 文件指纹，用于判断是否变化
chunk_ids_json 该文件写入 Milvus 后生成的 chunk id 列表
kb_version 所属知识库版本
embedding_model_version/chunk_schema_version 入库配置快照
updated_at 最近入库时间

这里有两个字段尤其关键：

  • embedding_model_version：同一个文件如果换了 embedding 模型，旧向量不能继续复用。
  • chunk_schema_version：同一个文件如果切分策略或 chunk metadata 契约变了，旧 chunk 结构不能继续复用。

这里采用严格等值匹配，不做“字段种类没少就算兼容”的宽松判断。valid_from_seq / valid_to_seq是引用式增量有效期窗口，属于 chunk metadata 契约；引入这类影响检索过滤语义的字段时，要同步提升CHUNK_SCHEMA_VERSION，让旧 manifest 自动失配并重建。

Manifest 表结构集中在qa_core/storage/runtime_schema.sql：

CREATE TABLE IF NOT EXISTS {{INDEX_MANIFEST_TABLE}} (...);

qa_core/indexing/manifest.py不再直接维护大段CREATE TABLE字符串，而是只负责 Manifest 的读写动作：

模块 职责
qa_core/storage/runtime_schema.sql 定义kb_document_manifests表结构
qa_core/storage/bootstrap.py 在 API 或脚本入口显式初始化 MySQL schema
qa_core/indexing/manifest.py 查询、更新、删除 Manifest 记录
qa_core/indexing/service.py 在入库流程中根据 Manifest 判断跳过、复用或重建

这种拆分不是为了增加文件数量，而是让“数据库初始化”和“入库业务判断”分开。这一部分先理解 Manifest 怎么做增量决策；需要看表结构时，再进入runtime_schema.sql。

因此“文件没变”还不够，必须同时满足：

fingerprint 相同
embedding_model_version 相同
chunk_schema_version 相同

6.4 两种增量：同版本跳过与跨版本复用

项目里实际有两类增量场景，它们的处理动作不同。

场景 A：同一个目标版本重复执行入库

比如某次入库中断了，重新执行同一个目标kb_version。这时目标版本里可能已经有部分文件写入成功。

处理逻辑是：

读取目标版本 manifest
  ↓
当前文件 fingerprint / embedding_model_version / chunk_schema_version 都一致
  ↓
直接跳过，不重复写入 Milvus

这就是_ingest_single_file()里第一层判断：

existing = manifest.get(source, path, target_kb_version, scenario_id)

if (
    existing
    and existing.fingerprint == fingerprint
    and existing.embedding_model_version == settings.embedding_model_version
    and existing.chunk_schema_version == settings.chunk_schema_version
):
    return 0, True, 0

场景 B：基于 active 版本创建一个新版本

这是更常见的企业发布方式：

python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --incremental-from active --quality-gate --activate

这时目标版本是新的，例如：

旧 active 版本：kb_v1
新 staged 版本：kb_v2

如果某个文件在kb_v1里已经入过库，并且文件内容、embedding 模型、切分策略都没变，目标版本也必须把它纳入自己的版本清单。文档检索不是按kb_version == kb_v2精确过滤，而是按 active 版本序号解释有效期窗口：

valid_from_seq <= active_seq
and (valid_to_seq == 0 or valid_to_seq > active_seq)

所以未变化旧 chunk 能在新版本中继续可见，依赖的是valid_from_seq/valid_to_seq有效期窗口；目标版本 manifest 继续记录旧chunk_ids，是为了说明kb_v2继承了这个文件，也让下一次以kb_v2为基准的增量构建还能继续判断跳过、复用、重建或过期。

因此跨版本增量的动作是：

读取基准版本 kb_v1 的 manifest
  ↓
确认文件未变化，模型和切分策略也未变化
  ↓
目标版本 manifest 直接记录旧 chunk_ids
  ↓
查询时由 active version_seq 决定这些 chunk 是否可见

这段逻辑现在直接写在_ingest_single_file()的跨版本复用分支里：

if not existing and base_record and _manifest_matches_current_settings(base_record, fingerprint, settings):
    _record_manifest(context, path, fingerprint, base_record.chunk_ids, settings)
    return FileIngestResult(reused_chunks=len(base_record.chunk_ids))

这里复用的是旧版本已经存在的 chunk 和向量，不复制 Milvus 行，也不重新调用 embedding 模型。

6.5 完整决策流程

对每一个文件，文档入库服务会按下面顺序判断：

顺序 判断 动作
1 文件类型不支持 报错，进入质量检查问题。
2 目标版本已有相同 manifest，且文件、模型、切分策略都未变化 跳过，不重复写入。
3 指定了--incremental-from，基准版本有相同文件，且文件、模型、切分策略都未变化 目标版本 manifest 引用基准版本 chunk，不复制 Milvus 行。
4 文件新增、文件变化、模型变化、切分策略变化 旧 chunk 写失效序号后，重新加载、标准化、切分、写入 Milvus。
5 新目录中已经没有某个旧文件 旧 chunk 写valid_to_seq = 目标版本序号，激活新版本后不再可见。

这也是为什么--incremental-from不能和下面两个参数同时使用：

参数 不能共用的原因
--force --force表示全部重算，和引用旧版本 chunk 的目标冲突。
--reset-collections 重置 collection 会删除旧向量，引用式增量没有可继承的数据来源。

6.6 具体示例

假设当前 active 版本是kb_v1，包含三份资料：

文件 kb_v1Manifest 状态
hr/onboarding.md chunk[hr_1, hr_2] 未变化
finance/expense.md chunk[fin_1, fin_2] 内容修改
finance/budget_preapproval_matrix.xlsx chunk[table_1, table_2] 文件被删除

现在执行：

python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --incremental-from active --quality-gate --activate

系统创建新版本kb_v2，处理结果如下：

文件 处理方式 kb_v2结果
hr/onboarding.md kb_v2manifest 引用kb_v1的旧 chunk kb_v2也能查到入职资料，且不重新 embedding、不复制向量。
finance/expense.md 旧 chunk 写valid_to_seq=2，新内容重新加载、切分、embedding kb_v2使用新的报销资料内容。
finance/budget_preapproval_matrix.xlsx 旧 chunk 写valid_to_seq=2 kb_v2激活后查不到这份已删除资料。
新增it/vpn.md 重新加载、切分、embedding kb_v2新增 VPN 资料。

最终线上激活后，文档检索按 active 版本序号解释有效期视图：

valid_from_seq <= active_seq
and (valid_to_seq == 0 or valid_to_seq > active_seq)

所以用户看到的是一套完整的新知识库视图：

  • 未变化文件仍然能查到，因为kb_v2的 manifest 会继续引用旧 chunk，且旧 chunk 的有效期覆盖 active_seq；
  • 变化文件查到的是新内容；
  • 删除文件不会再被召回；
  • 旧版本kb_v1仍可保留，用于回滚。

6.7 引用式增量的运行语义

当前项目采用的是引用式增量版本：

未变化文件不重新 embedding、不复制 Milvus 行，只通过有效期字段继续可见。

这个设计有三个直接结果：

结果 说明
存储更省 未变化 chunk 只保存一份。
构建更轻 日常更新只处理新增、修改、删除文件。
回滚清晰 MySQL active 指针切回旧版本，检索重新解释对应version_seq的有效视图。

线上查询、质量报告、调试诊断和垃圾回收都围绕同一套“有效版本视图”工作。

对比点 引用式增量版本
核心思路 新版本只记录变化，未变化 chunk 继续引用旧版本中仍然有效的 chunk。
未变化文件 不写新的 Milvus 记录，只通过valid_from_seq / valid_to_seq继续可见。
变化文件 旧 chunk 标记失效，新 chunk 从当前版本开始生效。
删除文件 给旧 chunk 写失效版本，例如valid_to_seq = 当前版本序号。
线上查询表达式 valid_from_seq <= active_seq and (valid_to_seq == 0 or valid_to_seq > active_seq)。
新版本完整性 active 版本是一张“有效版本视图”，由历史 chunk 和当前变化共同组成。
存储占用 较低，未变化 chunk 只保存一份。
构建耗时 较低，只处理新增、修改、删除文件。
回滚方式 切换 MySQL active 指针到旧kb_version，重新解释有效版本视图。
垃圾回收 需要确认没有任何版本继续引用旧 chunk，才能物理清理。

用一句话记：

引用式增量方案：离线只记录变化，线上按 active version_seq 解释 chunk 有效期。

6.8 核心方法

class IndexManifest(_MySqlStore):
    @staticmethod
    def key(source, file_path, kb_version=None, scenario_id=None):
        """根据来源、路径、版本和场景生成稳定清单键。"""
        return stable_hash(scenario_id or "", source, kb_version or "", str(Path(file_path).resolve()))

    def get(self, source, file_path, kb_version=None, scenario_id=None):
        """按 manifest_key 从 MySQL 读取文件入库记录。"""
        row = conn.execute(
            text("SELECT ... FROM kb_document_manifests WHERE manifest_key=:key"),
            {"key": self.key(source, file_path, kb_version, scenario_id)},
        ).mappings().fetchone()
        return ManifestRecord.from_row(row) if row else None

    def update(self, source, file_path, fingerprint, chunk_ids, *, scenario_id="", kb_version="", ...):
        """Milvus 写入成功后，把新 fingerprint 和 chunk_ids upsert 到 MySQL。"""
        conn.execute(text("INSERT INTO kb_document_manifests (...) VALUES (...) ON DUPLICATE KEY UPDATE ..."), params)

    def iter_records(self, *, scenario_id=None, source=None, kb_version=None):
        """按条件列出清单记录，用于缺失文件清理和治理报告。"""
        rows = conn.execute(text("SELECT ... FROM kb_document_manifests WHERE ..."), params)
        return [ManifestRecord.from_row(row) for row in rows]

方法和职责可以这样记：

方法 职责
IndexManifest.get() 查询某个文件在某个版本下是否已经入库。
IndexManifest.update() 文件成功写入或成功引用后，记录新的 fingerprint 和 chunk_ids。
IndexManifest.iter_records() 按场景、source、版本列出清单，用于治理和诊断。
IndexManifest.remove_by_key() 清理某条 manifest 记录。
expire_documents_for_version() 修改或删除文件时，把旧 chunk 的valid_to_seq写成目标版本序号。

6.9 常见误解

误解 正确理解
Manifest 是向量库 Manifest 只保存文件指纹和 chunk id，不保存正文和向量。
增量版本查询时要查旧版本加新版本 当前项目不是手工拼接多个版本，而是用 activeversion_seq解释 chunk 有效期视图。
文件内容没变就一定能复用 还要检查 embedding 模型版本和 chunk schema 版本。
跨版本增量就是跳过未变化文件 不是。目标版本 manifest 会引用旧 chunk，检索表达式保证这些 chunk 在新版本仍然可见。
删除文件会立即删除旧版本数据 不会。旧版本数据保留用于回滚；新版本通过valid_to_seq让它不可见。

第七部分：清理与维护

流程位置：第 6 步之后的维护动作。删除文件在增量入库时先通过valid_to_seq让旧 chunk 对新版本不可见，物理清理属于后续治理，不是发布主流程的必要步骤。

7.1 清理已删除的本地文件

当本地文档被删除时，Milvus 中的旧 chunk 不会自动消失。需要运行清理脚本：

# 预览将要清理的内容（默认 dry-run）
python scripts/kb/cleanup_missing_docs.py --scenario enterprise_knowledge

# 实际执行清理
python scripts/kb/cleanup_missing_docs.py --scenario enterprise_knowledge --no-dry-run

7.2 cleanup_missing_document_chunks 原理

def cleanup_missing_document_chunks(
    *,
    scenario_id: str | None = None,
    source: str | None = None,
    kb_version: str | None = None,
    dry_run: bool = True,
) -> dict[str, Any]:
    """清理 MySQL manifest 中已不存在本地文件的文档 chunk。

    该操作会删除 Milvus 数据，默认 dry-run 先预览再执行。
    """
    scenario = resolve_scenario(scenario_id)
    manifest = IndexManifest()
    records = manifest.iter_records(
        scenario_id=scenario.scenario_id,
        source=source,
        kb_version=kb_version,
    )
    missing = [r for r in records if r.path and not Path(r.path).exists()]

    if dry_run:
        return {
            "dry_run": True,
            "missing_file_count": len(missing),
            "affected_chunk_count": sum(len(r.chunk_ids) for r in missing),
            "missing_files": [
                {"path": r.path, "chunk_count": len(r.chunk_ids)}
                for r in missing
            ],
        }

    # 实际删除
    doc_store = get_doc_store(scenario.doc_collection)
    for record in missing:
        doc_store.delete_ids(record.chunk_ids)
        manifest.remove_by_key(record.key)

    return {
        "dry_run": False,
        "deleted_chunk_count": sum(len(r.chunk_ids) for r in missing),
        "deleted_file_count": len(missing),
    }

默认 dry-run：先预览再执行，防止误删。

第八部分：复杂图文资料入库闭环

流程位置：第 6 步和第 7 步之间。复杂图文资料必须先治理成可复核文本，再进入文档入库；质量报告和门禁负责阻断未治理完成的风险资料。

本部分边界

本部分是当前项目对图片资料的最小闭环：识别图片风险、阻断未复核图片资料、支持 OCR 复核后入库。它不是实时多模态问答，也不宣称模型已经能理解流程图、照片和截图里的全部视觉语义。

8.1 这属于多模态吗

导入文档中同时存在文字、图片、截图、扫描页、流程图、设备照片时，本质上已经进入了多模态资料处理范围。

但在当前一期项目里，它应该被定位为：

多模态入库治理，不是多模态在线问答。

两者区别如下：

类型 做什么 当前一期定位
多模态入库治理 离线识别图片风险、扫描件和图文 PDF，把确认后的内容转成可复核文本 当前已做最小闭环
多模态在线问答 用户实时上传图片，模型现场看图回答 不放一期主链路
多模态检索 同时存文本向量和图片向量，用 CLIP/VLM 做跨模态召回 更适合二期或三期

这样设计的原因是：在线问答必须稳定、低延迟、可追踪；图片解析、OCR、VLM 描述成本高且失败率高，如果直接塞进在线链路，会让 RAG 主流程变慢、变重、变不可控。

8.2 为什么不能“图片 OCR 一下就入库”

真实企业资料中的图片经常包含：

  • 合同扫描件；
  • 审批截图；
  • 设备告警截图；
  • 流程图；
  • 验收照片；
  • 表格截图；
  • 盖章文件；
  • 票据和单证照片。

这些内容的风险不只是“能不能识别出文字”，而是：

风险 示例
OCR 识别错误 金额8000被识别成B000
上下文断裂 图片中的“处理步骤”脱离前后正文后无法理解
来源不可追溯 回答引用了图片内容，但不知道来自第几页第几张图
证据未确认 扫描件内容未经人工复核，不能作为正式制度口径
图中信息不全 流程图箭头、颜色、图例无法仅靠 OCR 还原

所以复杂图文资料不能简单走“OCR -> 普通文本切分 -> 入库”。当前项目先实现下面这个闭环：

图文资料
  -> analyze_image_risk() 识别图片、扫描页、独立图片文件
  -> 入库质量报告写入 image_risk_files
  -> image_risk_blocking_files_count > 0 时质量门禁失败
  -> OCR 生成候选 Markdown
  -> 人工复核
  -> 提升为已复核 OCR Markdown
  -> 入库质量检查
  -> 新知识库版本激活

这里的关键点是：图片风险先可见，再决定能不能上线。普通文档里的 logo 或配图不会直接阻断；独立图片、图片型 PDF、文本层不足的扫描页必须先走 OCR 复核。

如果后续引入 VLM 和图文块，再升级为：

图片/流程图/设备照片
  -> OCR 或 VLM 生成候选说明
  -> 绑定附近正文、页码、图片编号
  -> 人工复核
  -> 生成 image_text_block
  -> 入库质量检查
  -> 新知识库版本激活

8.3 三类资料的处理策略

资料类型 处理方式 是否直接进入 active 知识库
有文本层的 PDF / Word / PPT 正文先按普通文档入库，图片进入风险报告 正文可以，图片不直接进
扫描件 / 图片 PDF 进入离线 OCR，生成待复核 Markdown；复核后提升为ocr_reviewed_text 复核后才可以进
独立图片文件 质量报告标记为severity=block，门禁阻断 不可以，必须 OCR/复核
图片和正文强相关资料 当前先做风险报告和 OCR 复核；后续可生成image_text_block 当前只允许复核后的文本入库

当前项目已有离线 OCR 脚本：

python scripts/ocr/run_offline_ocr.py --input-dir incoming_scans --output-dir reports/ocr/batch_001
python scripts/ocr/promote_ocr_candidates.py --input-dir reports/ocr/batch_001 --scenario engineering_project_qa --source quality --apply

第一条命令只生成待复核资料，第二条命令才把复核后的 Markdown 提升到场景资料目录。提升后仍然要执行知识库版本重建、入库质量检查和 RAG 回归验收。

当前已经落地的最小闭环是：

qa_core/indexing/image_risk.py
  -> 识别 standalone image / image-only PDF / 图文混排文档
  -> build_ingestion_quality_report() 记录 image_risk_files
  -> check_ingestion_quality_gate.py 阻断 blocking 图片资料
  -> run_offline_ocr.py
  -> 输出 OCR Markdown 和报告
  -> 人工把复核状态改为“已复核”或加入 review_status: reviewed
  -> promote_ocr_candidates.py 复制到场景资料目录
  -> 文本 loader 标记 content_type=ocr_reviewed_text
  -> 入库质量门禁不再按未复核 OCR 风险拦截
  -> rebuild_kb_version.py 生成候选知识库版本

质量报告中的关键字段：

字段 含义
image_risk_files 本次发现的图片、扫描页或图文混排风险文件
image_risk_files_count 图片风险文件总数
image_risk_blocking_files_count 必须 OCR/人工复核、不能直接激活的文件数
severity=review 文档含图片，但已有足够文本层；需要确认图片是否承载业务信息
severity=block 独立图片、图片型 PDF 或文本层不足，必须先 OCR/复核

8.4 多模态能力边界

这里的“多模态”指的是文档里同时存在文本、图片、截图、扫描件、表格、流程图。当前项目不是做在线看图问答，而是先把多模态资料纳入入库治理。

当前已经闭环的部分是：

  • 独立图片、扫描件、图文混排资料先做风险识别；
  • 风险文件进入入库质量报告；
  • 阻断级图片必须先 OCR / 人工复核；
  • 复核后的文本再以ocr_reviewed_text进入场景资料目录；
  • 版本激活和评测仍然沿用统一入库链路。

这意味着，当前项目已经能处理企业资料里的多模态内容，但方式是治理式闭环，不是实时视觉理解。

8.5 后续增强

image_text_block、页码级引用和图文专用检索是后续增强项，不在当前主链路展开。后续如果接入 VLM，再把图片说明、页码、附近正文和置信度整理成图文块即可。

第九部分：入库失败排查手册（排障附录）

流程位置：围绕第 2 步、第 7 步和第 8 步排障。先确认 active 指针和 collection，再看质量报告，最后根据构建摘要判断本次任务停在哪个阶段。

排障附录

本部分是问题定位清单，不是新的入库流程。只有在重建失败、质量门禁失败、active 版本异常或页面仍显示旧答案时，再按这里的顺序排查。

入库链路牵涉 MySQL、Milvus、Embedding、Reranker、场景配置、质量门禁和版本激活。排查时不要直接猜原因，按下面顺序查，速度最快。

9.1 先确认当前 active 版本

页面提示“信息不足”、检索结果为空、或者刚重建后仍然回答旧内容时，先查 active 版本：

docker compose --env-file .env.compose run --rm api python -c "from qa_core.config.settings import get_settings; from qa_core.scenarios.registry import resolve_scenario; from qa_core.governance.kb_versions import get_kb_version_store; s=get_settings(); sc=resolve_scenario(s.active_scenario_id); store=get_kb_version_store(sc.scenario_id); print(sc.scenario_id); print(store.resolve_active_version())"

判断：

现象 含义 处理
active=None 没有激活版本，在线问答不知道查哪批数据 重新执行rebuild_kb_version.py --quality-gate --activate
active 不是刚构建的版本 新版本停留在 staged 或 gate 失败 查看质量报告，修复后重新激活
active 是新版本但仍没答案 继续查 collection 和过滤条件 看 10.2/10.3

9.2 再确认 Milvus collection 是否存在且有数据

docker compose --env-file .env.compose run --rm api python -c "from pymilvus import MilvusClient; from qa_core.config.settings import get_settings; c=MilvusClient(uri=get_settings().milvus_uri); print(c.list_collections())"

如果 collection 不存在，说明入库没有真正写到 Milvus；如果 collection 存在但实体数量很少或为 0，需要回看入库日志。

常见原因：

现象 原因 处理
collection 不存在 场景配置里的 collection 名和实际不一致，或入库任务失败 检查scenario.toml，重新构建
只有 FAQ 没有 Doc 文档目录为空，或--skip-docs被使用 检查scenarios/<id>/docs
只有 Doc 没有 FAQ FAQ CSV 不存在，或--skip-faq被使用 检查faq.csv

9.3 schema 不兼容时使用 reset-collections

如果日志出现：

sparse 字段不是 BM25 Function 输出字段
nq [0] is invalid
BM25 Function / sparse 字段不兼容

通常表示复用了旧 schema collection。处理方式是删除旧 collection 并重建：

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

排查口径：

只手动删除 collection 不会自动生成新知识库版本。必须重新跑入库脚本，让脚本重新创建 collection、写入 FAQ/Doc、生成质量报告并激活版本。

9.4 质量门禁失败先看报告，不要直接跳过

--quality-gate失败时，说明资料里可能存在空文件、重复 FAQ、source 无效、FAQ/正文冲突或低质量 chunk。

排查顺序：

查看 reports/ingestion/<scenario_id>/
  -> 找到对应 scenario 和 kb_version 的报告
  -> 先修复 failed_files / unsupported_files / empty_files
  -> 再修复 duplicate_faq_questions / invalid_sources
  -> 最后再考虑调整阈值

报告也可以通过受 admin token 保护的管理接口查看：

GET /api/admin/ingestion_reports?scenario_id=<scenario_id>
GET /api/admin/ingestion_report_detail?path=<report_path>

列表接口返回报告摘要，详情接口返回完整 JSON；接口只允许访问reports/ingestion/目录内的报告文件。

不要为了让命令通过就尝试绕过质量门禁。当前脚本已把--activate和质量报告、质量门禁绑定在一起：要激活就必须先通过门禁，否则低质量资料只能停留在 STAGED，不会进入 active 知识库。

9.5 重建后页面还是旧答案

按这个顺序检查：

  1. 页面右侧当前状态里的知识库版本是否变成新版本。
  2. .env.compose中ACTIVE_SCENARIO_ID是否是你刚重建的场景。
  3. API 容器是否重新加载了.env.compose：
docker compose --env-file .env.compose up -d --force-recreate api
docker logs -f knowforge-api
  1. 是否有多个 Milvus 实例：宿主机脚本连的是127.0.0.1:19530，容器内脚本连的是http://milvus:19530。要确认两者指向同一个 Docker Compose 服务。

9.6 八场景全量初始化的推荐命令

如果需要在新环境中一次性把全部 8 个场景初始化到可运行状态，使用：

if (!(Test-Path .env.compose)) { Copy-Item .env.compose.example .env.compose }
notepad .env.compose
docker compose --env-file .env.compose up -d mysql etcd minio milvus
docker compose --env-file .env.compose build api
docker compose --env-file .env.compose run --rm api python scripts/rebuild_scenarios.py --reset-collections

如果之前已经存在知识库，只是资料内容变化，重建全部 8 个场景时不要删除 collection：

docker compose --env-file .env.compose run --rm api python scripts/rebuild_scenarios.py

如果容器镜像里还没有最新脚本，执行docker compose --env-file .env.compose build api后再运行入库命令。

讨论

留下你的想法

评论功能尚未配置。启用 Giscus 后，这里会显示基于 GitHub Discussions 的评论区。

图片预览

100%