iT邦幫忙

2026 iThome 鐵人賽

DAY 17
0
佛心分享-SideProject30

30 天打造公開資料版急診檢傷系統:Side Project 與實驗計畫系列 第 17

Day 17|Chunk 不是越大越好:切塊策略與階層 Metadata

  • 分享至 

  • xImage
  •  

Day 16 已把兩個公開來源整理成九筆可以追溯的韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)知識單元,也另外留下五項目前無法回答的知識缺口。今天不增加新來源,而是處理檢索前的下一個問題:一筆知識應該用多大的單位進入搜尋?

假設原始內容先介紹第一級到第五級,接著說明主要與次要考量。如果程式每隔固定字數就切一刀,切點可能剛好落在「血糖與脫水程度屬於次要考量」這句話中間。前半塊只剩例子,後半塊只剩分類名稱;即使兩塊都有文字,任何一塊單獨被找回時都可能缺少必要脈絡。

切塊(Chunking)就是把長文件拆成較小搜尋單位的過程;拆完的每一小塊稱為區塊(Chunk)。區塊不是越大越好,也不是越小越精準。太大會把多個主題混在一起,太小會切斷一個完整主張。真正要問的是:切塊邊界有沒有保留回答問題所需的最小完整脈絡?

下圖是概念示意。中央是同一條原始知識帶;左側使用固定刻度裁切,右側則依完整概念整理卡片,再把相關卡片放進父子資料夾。

同一份知識在固定邊界下出現概念切斷,在規則導向切塊下保留完整卡片並建立父子層次的概念示意

上圖不是在宣稱規則導向切塊一定有較高檢索分數,而是先建立本篇的工程問題:切塊方式必須留下可檢查的邊界與結構,後續才有可能公平評估。

今天不會建立向量、不會呼叫生成模型,也不需要啟動 Ollama。已下載的 qwen3-embedding:4b 會從 Day 18 才開始使用。


今天要完成什麼?

讀完並跟著操作後,你會完成以下六件事:

  1. 分辨固定字元、固定 token、段落與規則單元四種切塊方式。
  2. 用 140 字元、重疊 30 字元做一次固定邊界診斷。
  3. 建立九筆平面搜尋候選,以及內容完全相同的九筆階層 child 候選。
  4. 把九筆 child 分到三個 parent,命中後才補回同群組脈絡。
  5. 為每筆候選建立 11 個必要中繼資料欄位,未知資訊一律不推測。
  6. 建立八筆不使用病患紀錄與標籤的檢索測試案例,供後續評估重複使用。

本篇會用到的名詞

本篇的新名詞較多,先用表格建立快速索引。正文仍會在第一次實際使用時補上用途與例子。

中文名稱 英文全名/縮寫 本篇用途
檢索增強生成 Retrieval-Augmented Generation, RAG 先從外部知識找出相關內容,再把內容交給生成模型;本篇只準備其中的搜尋單位
區塊 Chunk 一次可以成為搜尋候選的文字單位
切塊 Chunking 把來源文件轉成多個 chunk 的過程
詞元 Token 模型文字處理器切出的計算單位,不一定等於一個中文字或英文單字
詞元化器 Tokenizer 依特定模型規則把文字轉成 token 的元件
重疊 Overlap 相鄰固定視窗重複保留的範圍,用來降低邊界切斷風險
中繼資料 Metadata 描述 chunk 來源、主題、適用級數與完整度等資訊
平面知識庫 Flat Knowledge Store 每個 chunk 獨立存在,命中後只回傳該筆內容
階層知識庫 Hierarchical Knowledge Store child chunk 另外連到 parent 脈絡,命中 child 後可以回取同群組內容
檢索測試案例 Retrieval Probe 預先寫好的查詢與預期行為,用來檢查檢索系統是否找對內容或承認缺口
關聯性判定 Relevance Judgment 指定某個查詢應該對應哪些知識單元的人工判定
安全雜湊演算法 256 位元 Secure Hash Algorithm 256-bit, SHA-256 偵測來源或搜尋文字是否改變,確保比較使用相同內容

為什麼不能先選一個「大家常用」的大小?

檢索增強生成(Retrieval-Augmented Generation, RAG)會先找外部知識,再讓大型語言模型根據找到的內容產生回答。最早的 RAG 工作之一把外部知識放入稠密向量索引,讓模型能取回相關內容;但「索引裡的一筆內容該有多大」仍然是建置者必須明確決定的工程條件。Lewis 等人的 RAG 研究

網路教學常見「每塊 1,000 字元、重疊 200 字元」之類的起始值。這組數字可以是大型長文件的候選設定,卻不是跨資料集通用的答案。Day 16 目前只有九筆短摘要,其中多數本來就是一個完整主張。如果直接使用 1,000 字元,數筆不同主題很可能被包進同一塊;如果每筆摘要個別處理,又可能完全切不到第二塊,無法觀察邊界問題。

所以今天把兩個目的拆開:

  1. 主要結構比較使用 Day 16 的原子規則單元。每個 rule_id 就是一筆搜尋候選,不再任意從句子中間切開。
  2. 固定邊界診斷把同群組規則串起來,再使用 140 字元視窗與 30 字元 overlap。這個較短設定是為了讓目前的小型語料真的出現邊界,觀察規則是否被切斷;產物明確標記為 diagnostic_only,不加入主要檢索比較。

這樣可以避免把「為了展示問題而刻意設置的視窗」誤寫成「已經選出的最佳切塊大小」。

四種邊界各自回答不同需求

切塊方式的核心差異,是邊界從哪裡來。下圖比較四種常見做法;請特別留意下方標記,只有固定字元診斷與規則單元會在今天實際產生檔案。

固定字元、固定 token、段落與規則單元四種切塊方式的邊界來源、取捨與本日採用狀態

上圖顯示,切塊大小只是其中一個參數;邊界由字元、模型、作者排版或知識結構決定,會帶來不同限制。

固定字元切塊:容易重現,也可能切斷語意

固定字元切塊(Fixed-Character Chunking)每隔固定字元數建立一個視窗。它不需要模型,容易實作,也容易在不同電腦重跑。

本篇設定:

  • 視窗長度 window_characters = 140
  • 相鄰視窗重疊 overlap_characters = 30
  • 每次向前移動 140 - 30 = 110 個字元。

假設某個 parent 串接後有 158 個字元,第一個視窗涵蓋位置 0 到 139,第二個視窗從位置 110 開始。兩個視窗會共享 30 個字元。Overlap 能降低邊界附近資訊完全消失的機率,卻不能保證一條長規則一定完整落在某個視窗內。

程式會保存每個視窗碰到哪些 rule_id,以及哪些規則完整包含在該視窗。只要一筆規則沒有被任何單一視窗完整保留,就把它列入 fragmented_rule_ids。這是結構診斷,不是模型答錯,也不是臨床風險結果。

固定 token 切塊:貼近模型限制,但必須先鎖定模型

固定詞元切塊(Fixed-Token Chunking)依 token 數量建立視窗。Token 是模型實際處理文字時使用的單位;詞元化器(Tokenizer)則負責把文字切成 token。同一句繁體中文,在不同 tokenizer 下可能得到不同 token 數量。

Token 視窗適合控制嵌入模型與生成模型的輸入上限,但今天不執行。原因不是這個方法較差,而是 Day 18 才會正式鎖定向量嵌入(Embedding)的輸入格式與 qwen3-embedding:4b 實際版本。現在先算 token,之後若更換 tokenizer,切塊邊界也會一起改變,無法穩定重現。

段落切塊:尊重作者結構,但品質依賴來源排版

段落切塊(Paragraph-Based Chunking)沿著標題、換行或清單邊界拆分。它通常比固定字元更能保留作者原本的敘事單位,但段落長度可能差異很大;網頁抽取錯誤時,也可能把整頁合成一段,或把同一句拆成多段。

Day 16 沒有保存完整來源網頁,而是保存自行撰寫的短摘要,因此今天不額外宣稱完成段落切塊比較。

規則單元切塊:保留主張,但需要人工治理

規則單元切塊(Rule-Unit Chunking)依知識主張的完整邊界建立 chunk。Day 16 的每筆 rule_id 已具備標題、完整內容、來源與限制,所以今天直接把它當成原子規則單元(Atomic Rule Unit)。「原子」在這裡不是指內容極短,而是指它只承擔一個可以獨立理解與追溯的主要主張。

這種方式能保留來源與限制,代價是需要人工審查。如果規則單元一開始就把三個不同主題塞在一起,程式不會自動把內容變正確。因此,自動測試負責守住結構契約,人工審查仍負責語意品質。

search_text 到底放什麼?

每筆主要搜尋候選的 search_text 都使用同一個簡單格式:

知識單元標題
知識單元內容

例如,第一級公開定義會把「KTAS 第一級公開定義」標題與對應短摘要接在一起。標題補充主題線索,內容保留完整主張;來源網址、完整度與級數則放在 metadata,不重複塞進搜尋文字。

中繼資料(Metadata)是「描述內容的資料」。它可以在顯示證據時提供來源,也能在未來做過濾或稽核,但不應偷偷改寫原始內容。Day 17 每個主要候選固定包含 11 個欄位:

欄位 意義 為什麼需要
rule_id 本系列建立的穩定規則識別碼 連回 Day 16 知識單元與測試答案
topic 知識主題 支援稽核與未來分組分析
ktas_levels 公開摘要明確涉及的級數 保留適用範圍,不擴張到其他級數
source_id 來源識別碼 找回來源登錄資料
source_url 來源網址 讓證據可以追溯
source_locator 來源頁面中的定位描述 說明摘要根據哪個段落或表格列
claim_kind 主張類型 區分來源摘要與作者推論
completeness 這筆公開主張完整到什麼程度 防止部分摘要被當成完整規則
age_group 年齡群 未公開時保留未知,不自行猜測
symptom_family 主訴家族 未公開時保留未知,不自行猜測
source_content_sha256 Day 16 知識內容雜湊 偵測上游內容是否改變

Day 16 的九筆公開摘要沒有逐筆說明年齡群與完整主訴家族,因此九筆 age_group 與九筆 symptom_family 全部寫成 unspecified_in_public_summary。這個值表示「目前公開摘要沒有指定」,不是「適用所有年齡或主訴」。

為什麼要同時做 flat 與 hierarchical?

平面知識庫(Flat Knowledge Store)把每筆 chunk 當成獨立候選。搜尋命中哪一筆,就只回傳哪一筆。它結構簡單,也容易看出查詢實際找到了什麼。

階層知識庫(Hierarchical Knowledge Store)則讓較小的 child chunk 連到較大的 parent context。搜尋仍在 child 上進行;命中 child 後,系統才依 parent_id 補回同群組脈絡。例如命中第三級公開定義後,可以再取得包含五級定義的 parent,讓後續模型理解相鄰級別關係。

階層檢索不是只有一種實作。RAPTOR會遞迴建立摘要樹,並在不同抽象層級取回內容。Day 17 的設計更簡單:沒有使用大型語言模型產生摘要,也沒有讓 parent 參與搜尋;三個 parent 只是人工固定的群組脈絡。引用 RAPTOR 是為了說明階層檢索這個研究方向,不代表本篇重現了 RAPTOR。

公平比較:搜尋候選一個字都不能偷改

若平面版搜尋九筆短文字,階層版卻搜尋三筆較長摘要,即使未來分數不同,也無法判斷差異來自階層、候選數量或文字內容。今天先建立公平比較契約:

  1. Flat 與 hierarchical 都搜尋相同的九個 rule_id
  2. 每一對候選的 search_text 必須逐字相同。
  3. 每一對候選的 search_text_sha256 必須相同。
  4. 每一對候選的 metadata 必須相同。
  5. 唯一允許的差異,是 hierarchical child 多一個非空 parent_id,命中後可以回取 parent。

三個 parent 分別是:

  • ktas-parent-system-overview:制度定位與高階流程,共兩個 child。
  • ktas-parent-level-definitions:第一級到第五級公開定義,共五個 child。
  • ktas-parent-considerations:主要與次要考量,共兩個 child。

下圖把公平條件畫在同一張圖中。圖上只列出五個 child 作為版面示意,實際輸出仍是九筆 flat 與九筆 hierarchical child;右側三個 parent 不參與今天的搜尋候選。

平面與階層知識庫使用相同九筆搜尋候選,階層版只增加 child 到三個 parent 的連結

上圖的紫色箭頭標示唯一改變。未來若兩種方法使用相同嵌入模型、相同查詢與相同前 k 筆候選數,差異才有機會合理歸因於「命中後是否補回 parent 脈絡」,而不是候選文字被偷偷改過。

不用病患測試標籤挑切塊大小

Day 15 已建立 KTAS_expert 專家參考標籤,未來會用來評估五級分類;但不能反覆查看測試折分類分數,再選擇讓分數最高的 chunk size。那會把測試資料變成調參資料,使最終結果過度樂觀。

Day 17 先建立檢索測試案例(Retrieval Probe)。每一筆 probe 包含:

  • query:人工撰寫的查詢。
  • expected_behavior:應該取回知識,或應該揭露缺口。
  • expected_rule_ids:可回答時的關聯性判定(Relevance Judgment)。
  • expected_gap_id:不可回答時應指出哪一個既有知識缺口。

這八筆 probe 全部由 Day 16 公開知識與覆蓋報告撰寫,不來自 Kaggle 病患紀錄,也不使用 KTAS_RNKTAS_expert。其中五筆預期取回規則,三筆預期揭露成人完整主訴目錄、逐主訴完整數值門檻與官方版本紀錄的缺口。

大型檢索評估集合通常會包含大量查詢、語料與關聯性判定,例如異質資訊檢索基準(Benchmarking Information Retrieval, BEIR)用多個資料集評估檢索模型的跨領域表現。Day 17 的八筆 probe 只是小型工程契約,不是統計充分的檢索基準;它的作用是讓後續系統至少能被一致地檢查,而不是宣稱哪種方法較好。

本篇會建立哪些檔案?

開始貼程式碼前,先確認每個資料夾與檔案的責任。讀者不需要前往其他網站尋找任何必要程式。

configs/knowledge/:保存可審查的切塊決策

configs/ 是設定資料夾,knowledge/ 子目錄保存知識建置政策。configs/knowledge/day-17-chunking-and-metadata.json 會讀取 Day 16 的九筆知識單元與覆蓋報告,並固定:

  • 上游檔案路徑與 SHA-256。
  • 140/30 固定字元診斷參數。
  • 11 個 metadata 必要欄位與禁止欄位。
  • 九筆 rule 到三個 parent 的對應。
  • 八筆 retrieval probes。
  • 五份 JSONL 與一份公開摘要的輸出路徑。

src/triage_rag/knowledge_base/:保存可測試的核心邏輯

src/ 是 Python 套件原始碼資料夾;triage_rag/knowledge_base/ 集中知識庫建置功能。

  • __init__.py 是 Day 16 已建立的套件入口,本篇不修改,讓只完成 Day 16 的讀者仍可獨立執行。
  • chunking.py 是本篇新增的核心模組。它接收設定、九筆知識單元與覆蓋報告,驗證版本、禁止欄位與父子關係,再回傳五組產物與公開摘要。Day 17 的命令列程式會直接從這個模組匯入建置函式。

核心模組只回傳資料,不自行決定命令列參數或輸出資料夾,因此測試可以直接傳入故意破壞的內容。

scripts/:保存讀者實際執行的入口

scripts/ 是自動化程式資料夾。scripts/build_day17_chunks.py 是 Python 命令列程式,會先比對 Day 16 兩個輸入檔的 SHA-256,再呼叫 chunking.py,寫出以下內容:

  1. flat-chunks.jsonl:九筆平面搜尋候選。
  2. hierarchical-children.jsonl:九筆階層 child 搜尋候選。
  3. hierarchical-parents.jsonl:三筆只供命中後回取的 parent context。
  4. fixed-window-diagnostic.jsonl:七個只作邊界診斷的固定字元視窗。
  5. retrieval-probes.jsonl:五筆可回答與三筆應揭露缺口的測試案例。
  6. results/public/day-17-chunking-and-metadata.json:不含病患資料的聚合摘要。
  7. results/runs/day-17/<run_id>/run-manifest.json:本次輸入、輸出、Git 狀態與執行參數。

data/knowledge/ 保存公開知識的機器可讀產物,不是病患資料目錄;results/public/ 保存可以公開的聚合檢查;results/runs/ 則保存每次本機執行的追溯紀錄。

tests/:確認契約真的會阻擋錯誤

tests/test_chunking.py 使用 Python 內建的單元測試框架 unittest。它會檢查正常輸出可重現,也會故意製造重複 parent 與病患標籤欄位,確認程式會拒絕:

  • Flat 與 hierarchical child 的搜尋文字和 metadata 完全相同。
  • 每個 rule_id 恰好屬於一個 parent。
  • 未知年齡與主訴不被自行推測。
  • 固定視窗只標記為診斷產物。
  • Probes 不使用病患紀錄或參考標籤。
  • KTAS_expert 等禁止欄位一旦混入知識就會失敗。

執行前的前置條件

本篇沿用 Day 13 建立的 Poetry 環境。Poetry 是 Python 的相依套件與虛擬環境管理工具;請先在專案根目錄執行:

poetry --version
poetry install

第一個命令應顯示 Poetry 版本,第二個命令會依 pyproject.tomlpoetry.lock 安裝環境。本篇沒有新增第三方執行相依,也不會下載模型。

你還需要先完成 Day 16 的正式建置步驟,確認以下兩個檔案存在:

test -f data/knowledge/ktas-public-v1/knowledge-units.jsonl && echo "知識單元已存在"
test -f results/public/day-16-knowledge-coverage.json && echo "覆蓋報告已存在"

兩行都應顯示對應的「已存在」。Day 17 設定還會核對兩個檔案的 SHA-256;如果你修改過 Day 16 設定,必須先重新建置 Day 16,再把新雜湊寫入 Day 17 設定,不能略過版本檢查。

先在自己的專案資料夾建立本篇完整檔案

接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾,依下列順序建立檔案;每個程式碼區塊都是該檔案的完整內容,不含省略號。

本篇沿用 Day 16 已完整建立的九筆知識單元、覆蓋報告,以及 Day 13 的 Poetry 與重現性工具。以下是 Day 17 新增或修改後的完整設定、套件入口、切塊模組、執行入口與測試;五份 JSONL、公開摘要與 run manifest 都由程式自動產生,不需要手動建立。

先從專案根目錄建立需要的資料夾:

mkdir -p configs/knowledge src/triage_rag/knowledge_base scripts tests data/knowledge/ktas-public-v1/day-17 results/public

如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案,把對應區塊完整貼入後儲存。

檔案 1:建立 configs/knowledge/day-17-chunking-and-metadata.json

鎖定主要切塊比較、固定字元診斷、中繼資料政策、三個父層群組、八筆檢索測試案例與輸出路徑。

請在文字編輯器建立 configs/knowledge/day-17-chunking-and-metadata.json,貼入以下完整內容並儲存:

{
  "schema_version": 1,
  "experiment_id": "day-17-chunking-and-metadata",
  "scope": "knowledge_structure_engineering_not_retrieval_quality_result",
  "source": {
    "knowledge_units_path": "data/knowledge/ktas-public-v1/knowledge-units.jsonl",
    "knowledge_units_sha256": "1c682c4c8d9e20a656a4609a5cbade8ceb9b3bdb828fcaa0eb9c3424dbe757e6",
    "coverage_report_path": "results/public/day-16-knowledge-coverage.json",
    "coverage_report_sha256": "1e1a50c01b18a4cac4c7e2c87d37cb49dc2fac1ed2475259d914d1c933967588",
    "required_kb_scope": "public_partial_knowledge_base_not_official_manual"
  },
  "strategies": {
    "primary_comparison": {
      "candidate_unit": "atomic_rule_unit",
      "flat_store": "每筆 rule_id 是一個獨立搜尋候選,命中後只回傳該筆內容。",
      "hierarchical_store": "搜尋候選與 flat 完全相同,命中 child 後再依 parent_id 取得同群組脈絡。",
      "fairness_checks": [
        "same_rule_ids",
        "same_search_text",
        "same_search_text_sha256",
        "same_chunk_metadata"
      ]
    },
    "fixed_character_diagnostic": {
      "window_characters": 140,
      "overlap_characters": 30,
      "retrieval_eligible": false,
      "purpose": "只用來觀察固定邊界是否切斷規則,不進入主要 flat 與 hierarchical 比較。"
    }
  },
  "metadata_policy": {
    "required_fields": [
      "rule_id",
      "topic",
      "ktas_levels",
      "source_id",
      "source_url",
      "source_locator",
      "claim_kind",
      "completeness",
      "age_group",
      "symptom_family",
      "source_content_sha256"
    ],
    "age_group_default": "unspecified_in_public_summary",
    "symptom_family_default": "unspecified_in_public_summary",
    "do_not_infer_missing_metadata": true,
    "forbidden_fields": [
      "record_index",
      "KTAS_RN",
      "KTAS_expert",
      "Group",
      "diagnosis",
      "disposition",
      "length_of_stay",
      "Error_group",
      "mistriage"
    ]
  },
  "parent_groups": [
    {
      "parent_id": "ktas-parent-system-overview",
      "title": "KTAS 公開定位與高階流程",
      "rule_ids": [
        "ktas-public-system-purpose-001",
        "ktas-public-workflow-001"
      ]
    },
    {
      "parent_id": "ktas-parent-level-definitions",
      "title": "KTAS 五級公開定義",
      "rule_ids": [
        "ktas-public-level-1-001",
        "ktas-public-level-2-001",
        "ktas-public-level-3-001",
        "ktas-public-level-4-001",
        "ktas-public-level-5-001"
      ]
    },
    {
      "parent_id": "ktas-parent-considerations",
      "title": "公開描述的主要與次要考量",
      "rule_ids": [
        "ktas-public-primary-considerations-001",
        "ktas-public-secondary-considerations-001"
      ]
    }
  ],
  "retrieval_probes": [
    {
      "probe_id": "probe-supported-system-purpose",
      "query": "KTAS 是什麼類型的工具,主要用來做什麼?",
      "expected_behavior": "retrieve",
      "expected_rule_ids": ["ktas-public-system-purpose-001"],
      "expected_gap_id": null
    },
    {
      "probe_id": "probe-supported-workflow",
      "query": "公開介紹中的 KTAS 高階檢傷流程包含哪些階段?",
      "expected_behavior": "retrieve",
      "expected_rule_ids": ["ktas-public-workflow-001"],
      "expected_gap_id": null
    },
    {
      "probe_id": "probe-supported-level-one",
      "query": "哪一個 KTAS 級數代表需要立即處置且照護優先順序最高?",
      "expected_behavior": "retrieve",
      "expected_rule_ids": ["ktas-public-level-1-001"],
      "expected_gap_id": null
    },
    {
      "probe_id": "probe-supported-primary-considerations",
      "query": "公開研究列舉哪些主要考量類型?",
      "expected_behavior": "retrieve",
      "expected_rule_ids": ["ktas-public-primary-considerations-001"],
      "expected_gap_id": null
    },
    {
      "probe_id": "probe-supported-secondary-examples",
      "query": "血糖與脫水程度在公開描述中是哪一類考量的例子?",
      "expected_behavior": "retrieve",
      "expected_rule_ids": ["ktas-public-secondary-considerations-001"],
      "expected_gap_id": null
    },
    {
      "probe_id": "probe-gap-adult-complaints",
      "query": "請列出 KTAS 完整成人主訴目錄。",
      "expected_behavior": "disclose_gap",
      "expected_rule_ids": [],
      "expected_gap_id": "adult_chief_complaint_catalog"
    },
    {
      "probe_id": "probe-gap-thresholds",
      "query": "請提供每一個 KTAS 主訴的完整數值門檻。",
      "expected_behavior": "disclose_gap",
      "expected_rule_ids": [],
      "expected_gap_id": "complaint_specific_thresholds"
    },
    {
      "probe_id": "probe-gap-version-history",
      "query": "這份知識庫是否含有完整官方版本與變更紀錄?",
      "expected_behavior": "disclose_gap",
      "expected_rule_ids": [],
      "expected_gap_id": "official_version_history"
    }
  ],
  "outputs": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "hierarchical_children_path": "data/knowledge/ktas-public-v1/day-17/hierarchical-children.jsonl",
    "hierarchical_parents_path": "data/knowledge/ktas-public-v1/day-17/hierarchical-parents.jsonl",
    "fixed_window_diagnostic_path": "data/knowledge/ktas-public-v1/day-17/fixed-window-diagnostic.jsonl",
    "retrieval_probes_path": "data/knowledge/ktas-public-v1/day-17/retrieval-probes.jsonl",
    "public_summary_path": "results/public/day-17-chunking-and-metadata.json",
    "run_output_root": "results/runs/day-17"
  }
}

儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。

檔案 2:建立 src/triage_rag/knowledge_base/chunking.py

驗證來源版本與禁止欄位,建立平面候選、階層 child/parent、固定字元診斷及獨立檢索測試案例。

請在文字編輯器建立 src/triage_rag/knowledge_base/chunking.py,貼入以下完整內容並儲存:

"""Build deterministic Day 17 chunk stores and retrieval probes."""

from __future__ import annotations

from collections import Counter
from typing import Any, Dict, Iterable, List, Mapping, Sequence, Tuple

from triage_rag.reproducibility import canonical_json_bytes, sha256_bytes


JsonObject = Dict[str, Any]
EXPECTED_SCOPE = "knowledge_structure_engineering_not_retrieval_quality_result"
ALLOWED_PROBE_BEHAVIORS = {"retrieve", "disclose_gap"}


class ChunkingContractError(ValueError):
    """Raised when Day 17 structure or metadata breaks the contract."""


def _text(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ChunkingContractError(f"{field} 必須是非空字串")
    return value.strip()


def _unique(values: Iterable[str], field: str) -> None:
    values = list(values)
    duplicates = sorted(value for value, count in Counter(values).items() if count > 1)
    if duplicates:
        raise ChunkingContractError(f"{field} 必須唯一,重複值:{duplicates}")


def _record_sha256(record: Mapping[str, Any]) -> str:
    return sha256_bytes(canonical_json_bytes(record))


def _validate_inputs(
    units: Sequence[Mapping[str, Any]],
    coverage_report: Mapping[str, Any],
    config: Mapping[str, Any],
) -> Tuple[Dict[str, Mapping[str, Any]], set[str]]:
    if config.get("schema_version") != 1:
        raise ChunkingContractError("目前只支援 schema_version=1")
    if config.get("scope") != EXPECTED_SCOPE:
        raise ChunkingContractError(f"scope 必須是 {EXPECTED_SCOPE}")
    if not units:
        raise ChunkingContractError("Day 16 knowledge units 不可為空")

    source_config = config.get("source")
    if not isinstance(source_config, dict):
        raise ChunkingContractError("source 必須是物件")
    required_scope = source_config.get("required_kb_scope")
    if any(unit.get("scope") != required_scope for unit in units):
        raise ChunkingContractError("knowledge unit scope 與 Day 17 契約不同")
    if coverage_report.get("scope") != required_scope:
        raise ChunkingContractError("coverage report scope 與 Day 17 契約不同")
    if coverage_report.get("knowledge_records_sha256") != source_config.get(
        "knowledge_units_sha256"
    ):
        raise ChunkingContractError("coverage report 與指定 knowledge units 版本不同")

    rule_ids = [_text(unit.get("rule_id"), "rule_id") for unit in units]
    _unique(rule_ids, "rule_id")
    by_rule_id = {unit["rule_id"]: unit for unit in units}

    forbidden = set(config["metadata_policy"]["forbidden_fields"])
    for unit in units:
        present = sorted(forbidden.intersection(unit))
        if present:
            raise ChunkingContractError(
                f"{unit['rule_id']} 出現禁止欄位:{present}"
            )
        if unit.get("retrieval_eligible") is not True:
            raise ChunkingContractError(
                f"{unit['rule_id']} 不是 Day 17 可用的檢索候選"
            )

    coverage_items = coverage_report.get("coverage", {}).get("items", [])
    gap_ids = {
        item["coverage_id"] for item in coverage_items if item.get("status") == "gap"
    }
    if not gap_ids:
        raise ChunkingContractError("coverage report 必須至少保留一項 gap")
    return by_rule_id, gap_ids


def _metadata(unit: Mapping[str, Any], config: Mapping[str, Any]) -> JsonObject:
    policy = config["metadata_policy"]
    source = unit["source"]
    metadata = {
        "rule_id": unit["rule_id"],
        "topic": unit["topic"],
        "ktas_levels": unit["ktas_levels"],
        "source_id": source["source_id"],
        "source_url": source["url"],
        "source_locator": source["source_locator"],
        "claim_kind": unit["claim_kind"],
        "completeness": unit["completeness"],
        "age_group": policy["age_group_default"],
        "symptom_family": policy["symptom_family_default"],
        "source_content_sha256": unit["content_sha256"],
    }
    missing = sorted(set(policy["required_fields"]) - set(metadata))
    if missing:
        raise ChunkingContractError(f"{unit['rule_id']} metadata 缺少:{missing}")
    forbidden = sorted(set(policy["forbidden_fields"]).intersection(metadata))
    if forbidden:
        raise ChunkingContractError(
            f"{unit['rule_id']} metadata 出現禁止欄位:{forbidden}"
        )
    return metadata


def _search_text(unit: Mapping[str, Any]) -> str:
    return f"{unit['title']}\n{unit['content']}"


def _validate_parent_groups(
    groups: Any, by_rule_id: Mapping[str, Mapping[str, Any]]
) -> List[JsonObject]:
    if not isinstance(groups, list) or not groups:
        raise ChunkingContractError("parent_groups 必須是非空陣列")
    parent_ids = [_text(group.get("parent_id"), "parent_id") for group in groups]
    _unique(parent_ids, "parent_id")

    seen_rule_ids: List[str] = []
    normalized: List[JsonObject] = []
    for group in groups:
        parent_id = group["parent_id"]
        title = _text(group.get("title"), f"{parent_id}.title")
        rule_ids = group.get("rule_ids")
        if not isinstance(rule_ids, list) or not rule_ids:
            raise ChunkingContractError(f"{parent_id}.rule_ids 必須是非空陣列")
        _unique(rule_ids, f"{parent_id}.rule_ids")
        unknown = sorted(set(rule_ids) - set(by_rule_id))
        if unknown:
            raise ChunkingContractError(f"{parent_id} 引用未知 rule_id:{unknown}")
        seen_rule_ids.extend(rule_ids)
        normalized.append(
            {"parent_id": parent_id, "title": title, "rule_ids": list(rule_ids)}
        )

    duplicates = sorted(
        rule_id for rule_id, count in Counter(seen_rule_ids).items() if count > 1
    )
    if duplicates:
        raise ChunkingContractError(f"rule_id 不得屬於多個 parent:{duplicates}")
    missing = sorted(set(by_rule_id) - set(seen_rule_ids))
    if missing:
        raise ChunkingContractError(f"以下 rule_id 沒有 parent:{missing}")
    return normalized


def _build_flat_chunks(
    by_rule_id: Mapping[str, Mapping[str, Any]], config: Mapping[str, Any]
) -> List[JsonObject]:
    records = []
    for rule_id in sorted(by_rule_id):
        unit = by_rule_id[rule_id]
        search_text = _search_text(unit)
        metadata = _metadata(unit, config)
        records.append(
            {
                "schema_version": 1,
                "store_type": "flat",
                "retrieval_role": "search_candidate",
                "chunk_id": f"flat::{rule_id}",
                "parent_id": None,
                "search_text": search_text,
                "search_text_sha256": sha256_bytes(search_text.encode("utf-8")),
                "metadata": metadata,
            }
        )
    return records


def _build_hierarchy(
    by_rule_id: Mapping[str, Mapping[str, Any]],
    groups: Sequence[Mapping[str, Any]],
    config: Mapping[str, Any],
) -> Tuple[List[JsonObject], List[JsonObject]]:
    children = []
    parents = []
    for group in groups:
        parent_id = group["parent_id"]
        child_ids = []
        child_texts = []
        sources = set()
        topics = set()
        levels = set()
        for rule_id in group["rule_ids"]:
            unit = by_rule_id[rule_id]
            search_text = _search_text(unit)
            child_id = f"hier-child::{rule_id}"
            child_ids.append(child_id)
            child_texts.append(search_text)
            metadata = _metadata(unit, config)
            sources.add(metadata["source_id"])
            topics.add(metadata["topic"])
            levels.update(metadata["ktas_levels"])
            children.append(
                {
                    "schema_version": 1,
                    "store_type": "hierarchical",
                    "retrieval_role": "search_candidate",
                    "chunk_id": child_id,
                    "parent_id": parent_id,
                    "search_text": search_text,
                    "search_text_sha256": sha256_bytes(search_text.encode("utf-8")),
                    "metadata": metadata,
                }
            )
        context_text = f"{group['title']}\n\n" + "\n\n".join(child_texts)
        parent = {
            "schema_version": 1,
            "store_type": "hierarchical",
            "retrieval_role": "parent_context_only",
            "parent_id": parent_id,
            "title": group["title"],
            "child_ids": child_ids,
            "child_rule_ids": list(group["rule_ids"]),
            "child_search_texts": child_texts,
            "context_text": context_text,
            "context_text_sha256": sha256_bytes(context_text.encode("utf-8")),
            "metadata": {
                "source_ids": sorted(sources),
                "topics": sorted(topics),
                "ktas_levels": sorted(levels),
                "age_group": config["metadata_policy"]["age_group_default"],
                "symptom_family": config["metadata_policy"][
                    "symptom_family_default"
                ],
                "scope": "grouped_public_summary_context",
            },
        }
        parents.append(parent)
    children.sort(key=lambda item: item["metadata"]["rule_id"])
    parents.sort(key=lambda item: item["parent_id"])
    return children, parents


def _fixed_windows(
    parents: Sequence[Mapping[str, Any]], config: Mapping[str, Any]
) -> Tuple[List[JsonObject], JsonObject]:
    strategy = config["strategies"]["fixed_character_diagnostic"]
    window = int(strategy["window_characters"])
    overlap = int(strategy["overlap_characters"])
    if window <= 0 or overlap < 0 or overlap >= window:
        raise ChunkingContractError("固定字元視窗必須滿足 window > overlap >= 0")
    step = window - overlap

    records: List[JsonObject] = []
    fragmented_rule_ids = set()
    parent_stats = []
    for parent in parents:
        title_prefix = f"{parent['title']}\n\n"
        blocks = []
        cursor = len(title_prefix)
        rule_spans = {}
        for rule_id, block in zip(
            parent["child_rule_ids"], parent["child_search_texts"]
        ):
            start = cursor
            end = start + len(block)
            rule_spans[rule_id] = (start, end)
            blocks.append(block)
            cursor = end + 2
        text = title_prefix + "\n\n".join(blocks)

        starts = list(range(0, len(text), step))
        if starts and starts[-1] + overlap >= len(text):
            starts.pop()
        windows = [(start, min(start + window, len(text))) for start in starts]
        if not windows:
            windows = [(0, len(text))]

        for index, (start, end) in enumerate(windows, start=1):
            touched = sorted(
                rule_id
                for rule_id, (rule_start, rule_end) in rule_spans.items()
                if rule_start < end and rule_end > start
            )
            contained = sorted(
                rule_id
                for rule_id, (rule_start, rule_end) in rule_spans.items()
                if start <= rule_start and rule_end <= end
            )
            chunk_text = text[start:end]
            records.append(
                {
                    "schema_version": 1,
                    "store_type": "fixed_character_diagnostic",
                    "retrieval_role": "diagnostic_only",
                    "chunk_id": f"fixed::{parent['parent_id']}::{index:03d}",
                    "parent_id": parent["parent_id"],
                    "char_start": start,
                    "char_end": end,
                    "window_characters": window,
                    "overlap_characters": overlap,
                    "text": chunk_text,
                    "text_sha256": sha256_bytes(chunk_text.encode("utf-8")),
                    "touched_rule_ids": touched,
                    "fully_contained_rule_ids": contained,
                }
            )
        fragmented = sorted(
            rule_id
            for rule_id in rule_spans
            if not any(
                start <= rule_spans[rule_id][0] and rule_spans[rule_id][1] <= end
                for start, end in windows
            )
        )
        fragmented_rule_ids.update(fragmented)
        parent_stats.append(
            {
                "parent_id": parent["parent_id"],
                "character_count": len(text),
                "window_count": len(windows),
                "fragmented_rule_ids": fragmented,
            }
        )
    records.sort(key=lambda item: item["chunk_id"])
    return records, {
        "window_characters": window,
        "overlap_characters": overlap,
        "chunk_count": len(records),
        "fragmented_rule_count": len(fragmented_rule_ids),
        "fragmented_rule_ids": sorted(fragmented_rule_ids),
        "parents": parent_stats,
        "interpretation": "固定字元結果只作邊界診斷,不是檢索品質結論。",
    }


def _build_probes(
    config: Mapping[str, Any], rule_ids: set[str], gap_ids: set[str]
) -> List[JsonObject]:
    probes = config.get("retrieval_probes")
    if not isinstance(probes, list) or not probes:
        raise ChunkingContractError("retrieval_probes 必須是非空陣列")
    probe_ids = [_text(probe.get("probe_id"), "probe_id") for probe in probes]
    _unique(probe_ids, "probe_id")
    records = []
    for probe in probes:
        probe_id = probe["probe_id"]
        query = _text(probe.get("query"), f"{probe_id}.query")
        behavior = probe.get("expected_behavior")
        if behavior not in ALLOWED_PROBE_BEHAVIORS:
            raise ChunkingContractError(f"{probe_id}.expected_behavior 不支援")
        expected_rule_ids = probe.get("expected_rule_ids")
        if not isinstance(expected_rule_ids, list):
            raise ChunkingContractError(f"{probe_id}.expected_rule_ids 必須是陣列")
        unknown_rules = sorted(set(expected_rule_ids) - rule_ids)
        if unknown_rules:
            raise ChunkingContractError(f"{probe_id} 引用未知 rule_id:{unknown_rules}")
        expected_gap_id = probe.get("expected_gap_id")
        if behavior == "retrieve":
            if not expected_rule_ids or expected_gap_id is not None:
                raise ChunkingContractError(
                    f"{probe_id} 的 retrieve 必須有 rule_id 且不能有 gap"
                )
        else:
            if expected_rule_ids or expected_gap_id not in gap_ids:
                raise ChunkingContractError(
                    f"{probe_id} 的 disclose_gap 必須引用已知 gap 且沒有 rule_id"
                )
        record = {
            "schema_version": 1,
            "probe_id": probe_id,
            "query": query,
            "expected_behavior": behavior,
            "expected_rule_ids": list(expected_rule_ids),
            "expected_gap_id": expected_gap_id,
            "origin": "author_written_from_public_knowledge_not_patient_record",
            "uses_patient_record": False,
            "uses_reference_label": False,
        }
        record["probe_sha256"] = _record_sha256(record)
        records.append(record)
    return sorted(records, key=lambda item: item["probe_id"])


def build_day17_artifacts(
    units: Sequence[Mapping[str, Any]],
    coverage_report: Mapping[str, Any],
    config: Mapping[str, Any],
) -> Tuple[JsonObject, JsonObject]:
    """Build all deterministic Day 17 records and their public summary."""

    by_rule_id, gap_ids = _validate_inputs(units, coverage_report, config)
    groups = _validate_parent_groups(config.get("parent_groups"), by_rule_id)
    flat = _build_flat_chunks(by_rule_id, config)
    children, parents = _build_hierarchy(by_rule_id, groups, config)
    fixed, fixed_summary = _fixed_windows(parents, config)
    probes = _build_probes(config, set(by_rule_id), gap_ids)

    flat_by_rule = {item["metadata"]["rule_id"]: item for item in flat}
    child_by_rule = {item["metadata"]["rule_id"]: item for item in children}
    same_search_text = all(
        flat_by_rule[rule_id]["search_text"] == child_by_rule[rule_id]["search_text"]
        for rule_id in by_rule_id
    )
    same_search_sha = all(
        flat_by_rule[rule_id]["search_text_sha256"]
        == child_by_rule[rule_id]["search_text_sha256"]
        for rule_id in by_rule_id
    )
    same_metadata = all(
        flat_by_rule[rule_id]["metadata"] == child_by_rule[rule_id]["metadata"]
        for rule_id in by_rule_id
    )
    parent_rule_ids = [rule_id for parent in parents for rule_id in parent["child_rule_ids"]]
    unspecified_age = sum(
        item["metadata"]["age_group"] == "unspecified_in_public_summary"
        for item in flat
    )
    unspecified_symptom = sum(
        item["metadata"]["symptom_family"] == "unspecified_in_public_summary"
        for item in flat
    )
    supported_probe_count = sum(
        probe["expected_behavior"] == "retrieve" for probe in probes
    )
    gap_probe_count = len(probes) - supported_probe_count

    artifacts = {
        "flat_chunks": flat,
        "hierarchical_children": children,
        "hierarchical_parents": parents,
        "fixed_window_diagnostic": fixed,
        "retrieval_probes": probes,
    }
    output_hashes = {
        name: sha256_bytes(
            b"".join(canonical_json_bytes(record) + b"\n" for record in records)
        )
        for name, records in artifacts.items()
    }
    summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "source": {
            "knowledge_unit_count": len(units),
            "knowledge_units_sha256": config["source"]["knowledge_units_sha256"],
            "coverage_gap_count": len(gap_ids),
        },
        "primary_comparison": {
            "candidate_unit": "atomic_rule_unit",
            "flat_chunk_count": len(flat),
            "hierarchical_child_count": len(children),
            "hierarchical_parent_count": len(parents),
            "same_rule_ids": set(flat_by_rule) == set(child_by_rule),
            "same_search_text": same_search_text,
            "same_search_text_sha256": same_search_sha,
            "same_chunk_metadata": same_metadata,
            "only_intended_difference": "hierarchical child 命中後可依 parent_id 取得群組脈絡。",
        },
        "fixed_character_diagnostic": fixed_summary,
        "metadata": {
            "required_field_count": len(config["metadata_policy"]["required_fields"]),
            "flat_chunks_with_unspecified_age_group": unspecified_age,
            "flat_chunks_with_unspecified_symptom_family": unspecified_symptom,
            "patient_or_label_fields_present": False,
            "interpretation": "公開摘要沒有年齡與主訴家族時保留 unspecified,不自行推論。",
        },
        "hierarchy": {
            "all_rules_have_exactly_one_parent": (
                len(parent_rule_ids) == len(set(parent_rule_ids)) == len(by_rule_id)
            ),
            "orphan_rule_ids": sorted(set(by_rule_id) - set(parent_rule_ids)),
        },
        "retrieval_probes": {
            "probe_count": len(probes),
            "supported_probe_count": supported_probe_count,
            "gap_probe_count": gap_probe_count,
            "uses_patient_records": False,
            "uses_reference_labels": False,
            "status": "probe_contract_only_not_retrieval_score",
        },
        "output_sha256": output_hashes,
        "checks": {
            "flat_and_hierarchical_candidate_text_identical": (
                same_search_text and same_search_sha and same_metadata
            ),
            "every_rule_has_one_parent": len(parent_rule_ids)
            == len(set(parent_rule_ids))
            == len(by_rule_id),
            "missing_metadata_not_inferred": unspecified_age
            == unspecified_symptom
            == len(flat),
            "probes_independent_of_patient_labels": all(
                not probe["uses_patient_record"]
                and not probe["uses_reference_label"]
                for probe in probes
            ),
            "fixed_windows_are_diagnostic_only": all(
                item["retrieval_role"] == "diagnostic_only" for item in fixed
            ),
        },
        "warning": "本報告只驗證切塊與中繼資料契約,尚未執行向量化或檢索品質評估。",
    }
    return artifacts, summary

儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。

檔案 3:建立 scripts/build_day17_chunks.py

驗證 Day 16 輸入雜湊,呼叫切塊核心,寫出五份 JSONL、公開摘要與可追溯執行紀錄。

請在文字編輯器建立 scripts/build_day17_chunks.py,貼入以下完整內容並儲存:

#!/usr/bin/env python3
"""Build Day 17 flat, hierarchical, diagnostic, and probe artifacts."""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path

from triage_rag.knowledge_base.builder import read_jsonl, write_jsonl
from triage_rag.knowledge_base.chunking import build_day17_artifacts
from triage_rag.reproducibility import (
    file_record,
    git_state,
    load_json,
    resolve_project_path,
    sha256_file,
    write_json,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/knowledge/day-17-chunking-and-metadata.json"
ARTIFACT_OUTPUT_KEYS = {
    "flat_chunks": "flat_chunks_path",
    "hierarchical_children": "hierarchical_children_path",
    "hierarchical_parents": "hierarchical_parents_path",
    "fixed_window_diagnostic": "fixed_window_diagnostic_path",
    "retrieval_probes": "retrieval_probes_path",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="建立 Day 17 切塊、中繼資料、階層結構與檢索測試案例。"
    )
    parser.add_argument("--config", default=DEFAULT_CONFIG)
    return parser.parse_args()


def _verify_sha256(path: Path, expected: str, label: str) -> None:
    actual = sha256_file(path)
    if actual != expected:
        raise ValueError(
            f"{label} SHA-256 不同;預期 {expected},實際 {actual}。"
            "請先重建 Day 16,或更新 Day 17 契約版本。"
        )


def main() -> int:
    args = parse_args()
    config_path = resolve_project_path(PROJECT_ROOT, args.config)
    config = load_json(config_path)
    source = config["source"]
    units_path = resolve_project_path(PROJECT_ROOT, source["knowledge_units_path"])
    coverage_path = resolve_project_path(
        PROJECT_ROOT, source["coverage_report_path"]
    )
    _verify_sha256(
        units_path, source["knowledge_units_sha256"], "Day 16 knowledge units"
    )
    _verify_sha256(
        coverage_path, source["coverage_report_sha256"], "Day 16 coverage report"
    )

    units = read_jsonl(units_path)
    coverage_report = load_json(coverage_path)
    artifacts, summary = build_day17_artifacts(units, coverage_report, config)

    written_paths = {}
    for artifact_name, output_key in ARTIFACT_OUTPUT_KEYS.items():
        output_path = resolve_project_path(PROJECT_ROOT, config["outputs"][output_key])
        write_jsonl(output_path, artifacts[artifact_name])
        if read_jsonl(output_path) != artifacts[artifact_name]:
            raise RuntimeError(f"{artifact_name} 寫入後內容不同")
        expected_sha = summary["output_sha256"][artifact_name]
        if sha256_file(output_path) != expected_sha:
            raise RuntimeError(f"{artifact_name} 寫入後 SHA-256 不同")
        written_paths[artifact_name] = output_path

    summary_path = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["public_summary_path"]
    )
    write_json(summary_path, summary)

    started_at = datetime.now(timezone.utc)
    run_id = (
        f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-"
        f"{summary['output_sha256']['flat_chunks'][:8]}"
    )
    run_root = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["run_output_root"]
    )
    run_directory = run_root / run_id
    run_directory.mkdir(parents=True, exist_ok=False)
    output_records = [
        file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
        for path in [*written_paths.values(), summary_path]
    ]
    manifest = {
        "manifest_schema_version": 1,
        "run_id": run_id,
        "started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
        "command": [sys.executable, *sys.argv],
        "documented_command": [
            "poetry",
            "run",
            "python",
            "scripts/build_day17_chunks.py",
        ],
        "git": git_state(PROJECT_ROOT),
        "inputs": [
            file_record(PROJECT_ROOT, args.config),
            file_record(PROJECT_ROOT, source["knowledge_units_path"]),
            file_record(PROJECT_ROOT, source["coverage_report_path"]),
        ],
        "outputs": output_records,
        "parameters": {
            "candidate_unit": "atomic_rule_unit",
            "fixed_window_characters": config["strategies"][
                "fixed_character_diagnostic"
            ]["window_characters"],
            "fixed_overlap_characters": config["strategies"][
                "fixed_character_diagnostic"
            ]["overlap_characters"],
            "parent_count": len(config["parent_groups"]),
        },
        "privacy": "不讀取病患資料或參考標籤;retrieval probes 由公開知識與缺口人工撰寫。",
        "scope": config["scope"],
    }
    manifest_path = run_directory / "run-manifest.json"
    write_json(manifest_path, manifest)

    comparison = summary["primary_comparison"]
    diagnostic = summary["fixed_character_diagnostic"]
    probes = summary["retrieval_probes"]
    print("Day 17 切塊與中繼資料:通過")
    print(
        f"主要候選:flat {comparison['flat_chunk_count']} 筆;"
        f"hierarchical children {comparison['hierarchical_child_count']} 筆"
    )
    print(f"hierarchical parents:{comparison['hierarchical_parent_count']} 筆")
    print(
        f"固定字元診斷:{diagnostic['chunk_count']} chunks;"
        f"{diagnostic['fragmented_rule_count']} 筆規則沒有被任何單一視窗完整保留"
    )
    print(
        f"retrieval probes:{probes['probe_count']} 筆;"
        f"可回答 {probes['supported_probe_count']}、應揭露缺口 {probes['gap_probe_count']}"
    )
    print(f"公開摘要:{summary_path.relative_to(PROJECT_ROOT)}")
    print(f"執行紀錄:{manifest_path.relative_to(PROJECT_ROOT)}")
    print("注意:尚未向量化,也還沒有任何檢索分數。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。

檔案 4:建立 tests/test_chunking.py

驗證平面與階層候選公平一致、每筆只有一個父層、未知 metadata 不被推論,並阻擋病患標籤洩漏。

請在文字編輯器建立 tests/test_chunking.py,貼入以下完整內容並儲存:

from __future__ import annotations

import copy
import unittest
from pathlib import Path

from triage_rag.knowledge_base.builder import read_jsonl
from triage_rag.knowledge_base.chunking import (
    ChunkingContractError,
    build_day17_artifacts,
)
from triage_rag.reproducibility import load_json


ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = ROOT / "configs" / "knowledge" / "day-17-chunking-and-metadata.json"


class Day17ChunkingTests(unittest.TestCase):
    def setUp(self) -> None:
        self.config = load_json(CONFIG_PATH)
        self.units = read_jsonl(
            ROOT / self.config["source"]["knowledge_units_path"]
        )
        self.coverage = load_json(
            ROOT / self.config["source"]["coverage_report_path"]
        )

    def build(self, config: dict | None = None, units: list | None = None):
        return build_day17_artifacts(
            units or self.units,
            self.coverage,
            config or self.config,
        )

    def test_build_is_deterministic(self) -> None:
        first = self.build()
        second = self.build()
        self.assertEqual(first, second)
        artifacts, summary = first
        self.assertEqual(len(artifacts["flat_chunks"]), 9)
        self.assertEqual(len(artifacts["hierarchical_children"]), 9)
        self.assertEqual(len(artifacts["hierarchical_parents"]), 3)
        self.assertTrue(all(summary["checks"].values()))

    def test_flat_and_hierarchical_candidate_text_is_identical(self) -> None:
        artifacts, summary = self.build()
        flat = {
            item["metadata"]["rule_id"]: item
            for item in artifacts["flat_chunks"]
        }
        children = {
            item["metadata"]["rule_id"]: item
            for item in artifacts["hierarchical_children"]
        }
        self.assertEqual(set(flat), set(children))
        for rule_id in flat:
            self.assertEqual(flat[rule_id]["search_text"], children[rule_id]["search_text"])
            self.assertEqual(flat[rule_id]["metadata"], children[rule_id]["metadata"])
        self.assertTrue(
            summary["primary_comparison"]["same_search_text_sha256"]
        )

    def test_every_rule_has_exactly_one_parent(self) -> None:
        artifacts, summary = self.build()
        parent_rules = [
            rule_id
            for parent in artifacts["hierarchical_parents"]
            for rule_id in parent["child_rule_ids"]
        ]
        self.assertEqual(len(parent_rules), len(set(parent_rules)))
        self.assertEqual(set(parent_rules), {unit["rule_id"] for unit in self.units})
        self.assertTrue(summary["hierarchy"]["all_rules_have_exactly_one_parent"])

    def test_missing_age_and_symptom_metadata_are_not_inferred(self) -> None:
        artifacts, summary = self.build()
        for chunk in artifacts["flat_chunks"]:
            self.assertEqual(
                chunk["metadata"]["age_group"],
                "unspecified_in_public_summary",
            )
            self.assertEqual(
                chunk["metadata"]["symptom_family"],
                "unspecified_in_public_summary",
            )
        self.assertTrue(summary["checks"]["missing_metadata_not_inferred"])

    def test_fixed_windows_are_diagnostic_and_expose_fragmentation(self) -> None:
        artifacts, summary = self.build()
        fixed = artifacts["fixed_window_diagnostic"]
        self.assertTrue(fixed)
        self.assertTrue(all(item["retrieval_role"] == "diagnostic_only" for item in fixed))
        self.assertGreater(
            summary["fixed_character_diagnostic"]["fragmented_rule_count"], 0
        )

    def test_probes_use_neither_patient_records_nor_labels(self) -> None:
        artifacts, summary = self.build()
        probes = artifacts["retrieval_probes"]
        self.assertEqual(len(probes), 8)
        self.assertTrue(all(not item["uses_patient_record"] for item in probes))
        self.assertTrue(all(not item["uses_reference_label"] for item in probes))
        self.assertEqual(summary["retrieval_probes"]["supported_probe_count"], 5)
        self.assertEqual(summary["retrieval_probes"]["gap_probe_count"], 3)

    def test_rule_cannot_belong_to_two_parents(self) -> None:
        invalid = copy.deepcopy(self.config)
        invalid["parent_groups"][1]["rule_ids"].append(
            invalid["parent_groups"][0]["rule_ids"][0]
        )
        with self.assertRaisesRegex(ChunkingContractError, "多個 parent"):
            self.build(config=invalid)

    def test_patient_label_field_is_rejected(self) -> None:
        invalid_units = copy.deepcopy(self.units)
        invalid_units[0]["KTAS_expert"] = 1
        with self.assertRaisesRegex(ChunkingContractError, "禁止欄位"):
            self.build(units=invalid_units)


if __name__ == "__main__":
    unittest.main()

儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。

完成後產生必要檔案並做靜態檢查

以下命令會在需要時產生套件鎖定檔,接著檢查設定格式與 Python 語法;它們不會啟動模型,也不會把病患資料送到網路:

poetry run python -m json.tool configs/knowledge/day-17-chunking-and-metadata.json
poetry run python -m py_compile src/triage_rag/knowledge_base/chunking.py scripts/build_day17_chunks.py tests/test_chunking.py
poetry run python -m unittest tests.test_chunking

每個命令都應正常結束。若 JSON 顯示行號,先檢查貼上時是否遺漏逗號、引號或括號;若 py_compile 報錯,先依行號修正縮排或漏貼內容。靜態檢查通過後,再執行本文後面的正式步驟。

實際建立 Day 17 切塊產物

完成上方完整檔案與靜態檢查後,依序執行下面三個步驟。

步驟 1:再次執行八項單元測試

目的,是在寫出正式產物前,先確認公平比較、metadata 與資料隔離契約都有效。輸入是剛建立的 Day 17 設定、Day 16 兩個產物和測試程式;輸出是終端機中的測試結果,不會修改病患資料。

poetry run python -m unittest tests.test_chunking -v

成功時最後應看到:

Ran 8 tests

OK

若測試顯示 knowledge unit scopeSHA-256 不同,先重建 Day 16 並核對設定中的上游雜湊。若顯示禁止欄位,請檢查知識單元是否意外混入 KTAS_RNKTAS_expert、診斷或病患去向。

步驟 2:建立五份 JSONL 與公開摘要

目的,是把通過測試的結構真正寫成後續可以讀取的檔案。這支程式讀取 Day 17 設定、Day 16 知識單元和覆蓋報告;成功後會寫出五份 JSONL、一份公開摘要和一份本機 run manifest。

poetry run python scripts/build_day17_chunks.py

成功時終端機應出現類似以下內容;run_id 含執行時間,因此每次不同:

Day 17 切塊與中繼資料:通過
主要候選:flat 9 筆;hierarchical children 9 筆
hierarchical parents:3 筆
固定字元診斷:7 chunks;2 筆規則沒有被任何單一視窗完整保留
retrieval probes:8 筆;可回答 5、應揭露缺口 3
公開摘要:results/public/day-17-chunking-and-metadata.json
執行紀錄:results/runs/day-17/<run_id>/run-manifest.json
注意:尚未向量化,也還沒有任何檢索分數。

最後一句特別重要。程式完成的是可重現的知識結構,不是「階層檢索優於平面檢索」的實驗結果。

步驟 3:從公開摘要驗證關鍵條件

先用 Python 內建的 JSON 格式工具讀取摘要:

poetry run python -m json.tool results/public/day-17-chunking-and-metadata.json

接著用一段不依賴額外套件的命令,列出五份 JSONL 的筆數:

poetry run python -c 'from pathlib import Path; root=Path("data/knowledge/ktas-public-v1/day-17"); [(print(p.name, sum(1 for line in p.open(encoding="utf-8") if line.strip()))) for p in sorted(root.glob("*.jsonl"))]'

應看到:

fixed-window-diagnostic.jsonl 7
flat-chunks.jsonl 9
hierarchical-children.jsonl 9
hierarchical-parents.jsonl 3
retrieval-probes.jsonl 8

摘要中的 checks 五個值都應是 true。另外確認:

  • same_search_textsame_search_text_sha256same_chunk_metadata 都是 true
  • all_rules_have_exactly_one_parenttrueorphan_rule_ids 是空陣列。
  • 九筆 age_group 和九筆 symptom_family 都保留 unspecified_in_public_summary
  • patient_or_label_fields_presentfalse
  • retrieval_probes.statusprobe_contract_only_not_retrieval_score

如何解讀今天實際得到的數字?

本次執行把三個 parent 串接成 157、301 與 158 個字元,再用 140/30 固定視窗產生七個診斷 chunk。結果有兩筆規則沒有被任何單一視窗完整保留:

  • ktas-public-secondary-considerations-001
  • ktas-public-workflow-001

這只能支持以下觀察:在目前文字順序與 140/30 設定下,固定邊界確實可能讓一筆規則跨越兩個視窗。它不能支持「固定字元檢索一定較差」,因為今天還沒產生向量、沒有執行查詢,也沒有計算任何檢索指標。

主要比較則產生九筆 flat、九筆 hierarchical child 和三筆 parent。九對搜尋候選的內容、內容雜湊與 metadata 全部一致。這個結果證明的是公平比較的輸入已經準備好,不是階層結構已經帶來效能提升。

下圖彙整今天可驗證的數字。請先看三張大卡的範圍,再看下方五項工程檢查與最底部不能宣稱的結論。

固定字元診斷產生七個視窗與兩筆碎裂規則,metadata 有十一個必要欄位,八筆 probes 分為五筆可回答與三筆缺口

上圖刻意把「結構契約」與「檢索成績」分開。五項檢查通過,只代表產物符合今天寫下的規格;哪種切塊方法找得比較準,要等 Day 18 建立向量表示、後續執行檢索,再用固定 probes 計算結果。

常見錯誤與限制

把 overlap 當成完整語意保證

Overlap 只會重複一段文字。規則長於視窗,或切點落在更外側時,仍可能沒有任何單一視窗保留完整主張。應該直接檢查規則跨度,而不是看到有 overlap 就假設安全。

讓 parent 也進入主要搜尋候選

若 hierarchical 同時搜尋九個 child 與三個 parent,候選集合就和 flat 不同。這可能是未來值得測試的另一種策略,但不能混入今天只改變「命中後補回脈絡」的控制比較。

把未知 metadata 填成看似合理的分類

公開摘要沒有年齡群或主訴家族時,不能因為標題看起來像成人規則就填 adult。錯誤 metadata 會在未來過濾時直接排除正確內容,而且很難從


上一篇
Day 16|先管來源,再談檢索:建立可追溯的公開 KTAS 知識庫
系列文
30 天打造公開資料版急診檢傷系統:Side Project 與實驗計畫17
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言