知识库检索质量修复：「风险评估问卷怎么评分」从必然转人工 → 直接答

## 根因（两个问题叠加）

1. 灌库脚本 tools/build_knowledge_chunks.py 的 expand_table_rows() 用「第一个非分隔行」
   当表头且 header 从不重置 → 同一节里第 2 张表格起**表头被当成数据行**。
   《个人投资者适当性管理指南》第九条下有 16 张问卷表格 → 15 条正文逐字相同的
   19 字零信息量碎片；同类共 19 条。
2. 检索层去重只按 doc_id，接不住「同一父块的兄弟子块」（POL-AST-009-07 与 -12 是
   不同 doc_id、同一父块）→ 15 条碎片全留在候选里互相打平，把 top1/次优差压到 0.002，
   而客服判定要求中置信必须领先 ≥0.07（MIN_GAP）→ 恒判并列转人工。

## 为什么没有重灌知识库

用 Milvus 现成向量离线复算四种做法（同一问句、同一批向量）：
  现状                   gap 0.002  转人工
  只修 bug（=重灌全部收益） gap 0.001  仍转人工，且更糟
  只加父块归并            gap 0.093  可答，但答案是 19 字碎片
  两处都改                gap 0.076  可答且有内容
原因：第九条被切成 94 块，删掉 15 条表头碎片后，剩下 79 条数据行碎片依然互相打平。
重灌解决不了，却要停机 3–10 分钟并丢掉上传路径的内容（含 R1–R5 那条）。

## 改了什么

① app/service/knowledge_search_service.py：新增 _merge_sibling_subblocks()
   同一父块的兄弟子块只保留最高分那条；父块自身与 FAQ/政策这类本身就是细粒度答案
   的块一律不合并（它们之间打平是真的多个候选）。6 个单测守着。
   被丢掉的只是同节其它细节，本节完整内容由 _parent_hits 带回的父块兜底。

② tools/build_knowledge_chunks.py：修表头识别（markdown 表格只有紧邻 |---| 之前的
   那一行才是表头）+ 新增 assert_no_duplicate_contents() 自带守卫 ——
   正文完全相同的块必须为 0，否则中止且不写 jsonl。该脚本是一次性灌库脚本、
   原来没有单测覆盖，这正是该 bug 活下来的原因，所以守卫放在它自己的执行路径上。
   块数 636 → 617；正文完全相同的组 1 组 15 块 → 0 组。
   负向验证：换回旧逻辑跑，退出码 1 并报出那 15 份碎片，且未覆盖 jsonl。

③ tools/drop_table_header_vectors.py：清掉已灌进 Milvus 的 19 条历史碎片。
   判定可复现：语义 id（非纯数字）且正文不在修正后产物里。只动语义 id 是因为
   纯数字 id 来自上传路径、本来就不在 jsonl 里（实测冒烟 A 线的 top1 就是数字 id 179）；
   不按长度判是因为短块本身是设计的一部分（「评审标准：管理人资质 15%」13 字是有效答案）。
   实删 policy 16 + product 3，faq 0；dry-run 逐条核对过，全部是「标签 + 表头词」形态。
   这是唯一一次绕过 Outbox 的删除：事件消费侧要回读 MySQL 行，而这批是无元数据种子向量。

## 验证

真实链路 10 问句回归 10/10 与预期一致，0 个变坏：
  风险评估问卷怎么评分  0.7156 gap 0.1473 → 答（原 0.002 转人工），top1 变成真实答案行
  场内基金的管理费率    0.8114 gap 0.0977 → 直接答（冒烟 A 线）
  南方季季盈90天起投金额 0.8631 gap 0.3387 → 直接答（行级子块精确命中能力未受影响）
  今天天气怎么样        仍正确转人工
端到端：ask_customer_service.py「风险评估问卷怎么评分」→ 直接答，返回完整评分标准；
e2e_smoke_test.py 44/44（含 A4 知识库覆盖未转人工）；
pytest tests/unit tests/contract 1506 passed / 0 failed；
对账 重复正文 15 → 0，孤儿/死向量/缺向量仍全为 0。
（冒烟首跑 42/43 的唯一失败 B6 买入下单 503 是行情过期，补刷后 44/44，与本次无关。）

## 顺带查明

Milvus 的 delete() 是标记删除，约 1 秒后才在 query/search 中不可见（隔离临时集合实测：
删完立即查仍看得到，+1s 起消失）。清理工具第一版"删完立刻复核"因此谎报 19 条残留，
已改为轮询复核并把文案改成实测依据。get_collection_stats().row_count 在删除后仍显示旧值，
这也是对账工具坚持用 query 实际行数的原因。

## 文档

新增 docs/演示用/知识库检索质量修复-2026-09-15.md（根因、四方案对比、改动、全部证据）；
并对 docs/演示用/知识库向量对账与清理-2026-09-15.md 做更正 —— 451 条短块里 429 条是
设计内的有效数据行、只有 19 条是 bug 产物；"按长度合并短块 + 重建重灌"的建议已被实测否决。

## 仍未解决（记录在案）

「高净值客户有什么权益」gap 0.0075 仍转人工：根因是不同父块之间同族内容打平
（金卡 vs 白金权益），父块归并救不了也不该救，要从内容侧或业务口径入手。
knowledge/_chunks.jsonl（新 617 块、编号连续）与 Milvus（旧编号、含 19 个空洞）目前不一致；
不重灌无影响，但下次重灌必须 drop 集合重建而不是 upsert，否则两套编号会共存。

tools/build_knowledge_chunks.py

@@ -147,25 +147,28 @@ def expand_table_rows(chunk: dict[str, str], parent_id: str) -> list[dict[str, o
 
     小块必须**自解释**：只回「1万元」客户不知道说的是哪个产品，所以带上产品名与行标签。
     父块保留，客户问「这个产品怎么样」时仍要能拿到完整一节。
+
+    ## 2026-09-15 修掉的 bug：表头被当成数据行
+
+    原实现用「**第一个非分隔行**」当表头（`if not header: header = cells`），而 `header`
+    从不重置。于是**同一节里出现第二张表格时，它的表头行被当成数据行**，产出形如
+    「第九条 问卷内容及评分标准：选项 分值」的**零信息量碎片**：
+    《个人投资者适当性管理指南》第九条下有 16 张问卷表格 → 15 条碎片，且**正文逐字相同**。
+    检索时它们必然互相打平（实测把「风险评估问卷怎么评分」的 top1/次优差压到 **0.002**，
+    客服按"中置信需领先 ≥0.07"判并列 → 转人工），把真正有内容的块挤到第 5 名。
+
+    修法：markdown 表格的表头**只可能是紧邻分隔行 `|---|---|` 之前的那一行**，所以按分隔行
+    认表头，用 `prev` 延迟一行判断。修后块数 636 → 617，正文完全相同的组从 1 组 15 块降到 0。
     """
     blocks: list[dict[str, object]] = []
-    header: list[str] = []
     name = LEADING_NUMBER.sub("", chunk["section"]).strip() or chunk["section"]
-    for line in chunk["content"].splitlines():
-        match = TABLE_ROW.match(line.strip())
-        if not match:
-            continue
-        cells = [cell.strip() for cell in match.group(1).split("|")]
-        if all(set(cell) <= {"-", ":", " "} for cell in cells):
-            continue  # 分隔行 |---|
-        if not header:
-            header = cells
-            continue
+
+    def emit(cells: list[str]) -> None:
         if len(cells) < 2:
-            continue
+            return
         label, value = cells[0], cells[1]
         if not label or not value:
-            continue
+            return
         blocks.append({
             "title": f"{chunk['title']} · {label}",
             "chapter": chunk["chapter"],
@@ -173,9 +176,54 @@ def expand_table_rows(chunk: dict[str, str], parent_id: str) -> list[dict[str, o
             "content": f"{name}：{label} {value}",
             "parent_id": parent_id,
         })
+
+    prev: list[str] | None = None
+    for line in chunk["content"].splitlines():
+        match = TABLE_ROW.match(line.strip())
+        if not match:
+            continue
+        cells = [cell.strip() for cell in match.group(1).split("|")]
+        if all(set(cell) <= {"-", ":", " "} for cell in cells):
+            # 分隔行 |---|：紧邻它之前的那一行（`prev`）是**表头**，不能当数据行 → 丢弃。
+            prev = None
+            continue
+        if prev is not None:
+            emit(prev)  # 没被分隔行认领为表头的行 = 数据行
+        prev = cells
+    if prev is not None:
+        emit(prev)  # 收尾：最后一行也要处理（没有分隔行收尾的表格）
     return blocks
 
 
+def assert_no_duplicate_contents(records: list[dict[str, object]]) -> None:
+    """自带守卫：**正文完全相同的块必须为 0**，否则直接失败退出、不生成 jsonl。
+
+    为什么用这一条当守卫：表头被误当数据行时的直接后果就是"**多张表格产出逐字相同的块**"
+    （第九条那 16 张问卷表 → 15 条一模一样的「…：选项 分值」）。这类块在检索里必然互相
+    打平，把 top1/次优差压到 0.07 门槛之下（实测 0.002）→ 客服判并列转人工。
+    本脚本是**一次性灌库脚本**、没有单测覆盖（这正是该 bug 活下来的原因），
+    所以把守卫放在脚本自己的执行路径上：**每次重灌都会跑一遍**。
+
+    为什么不是"块长度下限"：短块本身是设计的一部分（「评审标准：管理人资质 15%」13 字，
+    但它是真实的数据行、是有效答案）。**内容逐字重复**才是缺陷特征，长度不是。
+    """
+    seen: dict[str, list[str]] = {}
+    for record in records:
+        seen.setdefault(str(record["content"]), []).append(str(record["doc_id"]))
+    duplicated = {content: ids for content, ids in seen.items() if len(ids) > 1}
+    if duplicated:
+        detail = "\n".join(
+            f"  {len(ids)} 份：{content[:60]!r} → {ids[:6]}"
+            for content, ids in list(duplicated.items())[:5]
+        )
+        raise SystemExit(
+            f"知识块自检失败：有 {len(duplicated)} 组正文完全相同的块。\n"
+            "这类块在检索里必然互相打平（把 top1/次优差压到 0.07 之下 → 客服转人工），"
+            "通常是**表格表头被当成了数据行**（见 `expand_table_rows` 的 docstring）。\n"
+            f"{detail}\n已中止，未写入 jsonl。"
+        )
+
+
 records: list[dict[str, object]] = []
 for relative, config in SOURCES.items():
     text = (Path("knowledge") / relative).read_text(encoding="utf-8")
@@ -240,6 +288,8 @@ for order, chunk in enumerate(
         "chars": len(chunk["content"]),
     })
 
+assert_no_duplicate_contents(records)
+
 (Path("knowledge") / "_chunks.jsonl").write_text(
     "\n".join(json.dumps(record, ensure_ascii=False) for record in records), encoding="utf-8"
 )