Day 18 已經讓 qwen3-embedding:4b 把文字轉成向量，今天終於要把向量拿來搜尋。但「語意接近」不是唯一的搜尋線索：規則裡的專有名詞、縮寫與數字若恰好和問題相同，字面匹配也可能很有價值。

例如，讀者問「血糖與脫水程度是哪一類考量？」時，向量模型可以根據語意找相近段落；字詞搜尋則能直接抓到「血糖」與「脫水」這兩組詞。兩條路徑都可能找對，也都可能在知識庫沒有答案時，硬是排出一個看似最相關的候選。

先看下方的概念示意。左側同一個問題分成上下兩條路：上方比較整體語意，下方比對實際字詞；兩條路最後回到同一疊候選規則。它們不是各自回答一次，也不是讓兩個生成模型投票。

上圖就是今天的核心：比較的是「候選怎麼排序」，不是「誰直接決定檢傷級數」。本篇只使用九筆公開知識文字與八筆作者自寫問題，不會讀取 Kaggle 病患列、護理師登錄級數或專家重新判定級數。

本篇是離線檢索工程實驗，不是臨床效能評估。檢索第一名只代表在某一種排序方法下最接近問題，不代表內容足以回答問題，更不能取代醫師或護理師的判斷。

今天會完成什麼

完成本篇後，你會得到四項可觀察的產物：

1. 一個稠密檢索器，用 qwen3-embedding:4b 與餘弦相似度排列九筆知識。
2. 一個不需額外套件的字詞型稀疏檢索器，用相同九筆知識做排序。
3. 一個只根據兩份名次進行融合的混合檢索器。
4. 五筆可計分問題的前一、三、五筆召回率與平均倒數排名，以及三筆不能計分的知識缺口診斷。

這些產物只回答「三條檢索路徑能否依固定契約執行」與「這組小型問題出現什麼排名」。八筆問題太少，也太接近目前的公開知識文字，因此不能用來選出普遍最佳方法或調整參數。

本篇會用到的名詞

本篇新名詞超過三個，先用表格建立共同語言。正文第一次使用時仍會再解釋用途。

先把檢索在 RAG 裡的位置說清楚

檢索增強生成（Retrieval-Augmented Generation, RAG）是一種先從外部知識庫找資料，再把取回內容交給大型語言模型產生回答的方法。一條完整 RAG 管線至少包含：建立候選、檢索、可能的重排序、組裝證據、生成與安全檢查。

本系列整理的制度是韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS），數字越小代表急迫度越高。Kaggle 資料欄位 KTAS_RN 保存護理師登錄級數，KTAS_expert 保存專家重新判定級數；兩者都不是今天的檢索輸入。

今天只做「第一階段檢索」。程式收到一個問題後，會把九筆公開知識全部排出名次，再取前 k 筆（Top-k）。它不會：

• 產生自然語言答案。
• 預測 KTAS 第一級到第五級。
• 呼叫線上重排序服務。
• 套用生命徵象門控或拒答門檻。
• 讀取病患輸入、KTAS_RN 或 KTAS_expert。

這個邊界很重要。若今天把生成品質、檢傷標籤與第一階段檢索混在一起，排名錯誤時就無法知道問題出在搜尋、提示詞、生成模型或資料洩漏。

Dense：用整體語意找相近內容

稠密檢索（Dense Retrieval）先用向量嵌入模型，把問題與每筆文件都轉成固定長度的稠密向量。稠密向量的多數位置都有數值；電腦再用餘弦相似度（Cosine Similarity）比較兩個向量方向，分數較高的文件排在前面。

本篇延續 Day 18，固定使用 qwen3-embedding:4b，每段文字輸出 2,560 維向量。Qwen 官方說明建議查詢端加入任務指令，而文件端不加指令；因此八筆問題會先組成下列格式，九筆文件則直接使用 Day 17 的 search_text：Qwen3-Embedding 官方實作說明

模型透過本機模型執行工具 Ollama 提供的應用程式介面（Application Programming Interface, API）載入。Ollama 在本篇只把文字轉為向量，不負責產生回答；模型名稱、完整模型內容摘要（digest）、維度與禁止靜默截斷都會在執行前驗證。

Instruct: Given an emergency triage query in Traditional Chinese, retrieve relevant public KTAS knowledge passages. Do not infer unpublished thresholds.
Query: 血糖與脫水程度在公開描述中是哪一類考量的例子？

餘弦相似度的公式已在 Day 18 逐項計算過。這裡只重申解讀方式：同一個問題下，數值較大表示向量方向較接近，適合用來排列該問題的九筆文件；它不是機率，也不是醫療正確率。

Dense 的優勢是問題與文件不必逐字相同。例如「立即處置」與「照護優先順序最高」可以透過整體語意關係靠近。風險則是精確代碼、罕見詞、否定或數字可能被整體語意稀釋。

BM25：同時看詞頻、稀有程度與文件長度

稀疏檢索（Sparse Retrieval）依實際出現的詞彙建立權重。與每個位置都有浮點數的稠密向量不同，一筆文件只會命中整個詞彙表中的一部分位置，所以稱為「稀疏」。

本篇採用最佳匹配 25（Best Matching 25, BM25）。BM25 不是只計算某個詞出現幾次，它還會回答三個問題：

1. 查詢詞在這筆文件出現幾次？
2. 這個詞在整個知識庫是否少見？
3. 這筆文件是否因為特別長，而自然包含較多詞？

中文要先決定怎麼切詞

文字切分（Tokenization）是把輸入拆成 BM25 可以計數的搜尋單位。英文通常可以借助空白，但繁體中文句子不會在每個詞之間留空格；若切法不固定，同一份文字可能得到不同排名。

為了讓讀者只靠 Python 標準函式庫也能重跑，本篇採用簡單且確定的規則：

1. 先用 Unicode 正規化相容組合形式（Normalization Form Compatibility Composition, NFKC）正規化，統一全形與相容字元。
2. 英文字母轉為小寫；英文、數字與像 qwen3-embedding 的識別字保留為完整 token。
3. 連續繁體中文字同時拆成單字與相鄰雙字。
4. 同一個查詢中的重複 token 只計一次，固定為 binary unique terms。

例如：

輸入：KTAS 血糖與脫水
英文 token：ktas
中文單字：血、糖、與、脫、水
中文相鄰雙字：血糖、糖與、與脫、脫水

這不是宣稱「雙字切分」是最佳中文斷詞器，而是建立一個不用下載詞典、行為固定的基準。它會產生像「糖與」這類不自然雙字，因此日後若更換斷詞器，必須建立新實驗版本，不能讓新舊結果共用同一名稱。

把 BM25 公式逐項拆開

BM25 會使用反向文件頻率（Inverse Document Frequency, IDF），讓較少文件出現的查詢詞得到較高權重。文件 $D$ 對查詢 $Q$ 的 BM25 分數寫成：

$$
\operatorname{BM25}(D,Q)=
\sum_{q_i\in Q}
\operatorname{IDF}(q_i)
\frac{f(q_i,D)(k_1+1)}
{f(q_i,D)+k_1\left(1-b+b\frac{|D|}{\operatorname{avgdl}}\right)}
$$

公式中的符號分別代表：

• $Q$ 是查詢拆出的不重複 token 集合。
• $D$ 是目前計分的文件。
• $q_i$ 是查詢中的一個 token。
• $f(q_i,D)$ 是 token $q_i$ 在文件 $D$ 出現的次數。
• $|D|$ 是文件切分後的 token 數量。
• $\operatorname{avgdl}$ 是全部九筆文件的平均 token 數量。
• $k_1$ 控制同一個詞重複出現時，分數增加多快達到飽和；本篇固定為 1.2。
• $b$ 控制文件長度正規化的強度；本篇固定為 0.75。
• $\operatorname{IDF}$ 是剛才定義的反向文件頻率，詞越少見，權重通常越大。

本篇的反向文件頻率採用正值版本：

$$
\operatorname{IDF}(q_i)=
\ln\left(1+\frac{N-n(q_i)+0.5}{n(q_i)+0.5}\right)
$$

其中 $N$ 是文件總數，本篇為 9；$n(q_i)$ 是含有 token $q_i$ 的文件數。若一個詞只出現在 1 筆文件，示意 IDF 是 $\ln(1+8.5/1.5)\approx1.8971$；若它出現在 8 筆文件，則是 $\ln(1+1.5/8.5)\approx0.1625$。前者較稀有，因此同一次命中的影響較大。

假設稀有詞在某文件出現一次，而且該文件長度剛好等於平均長度，詞頻分數中的分子與分母都會化成 2.2，所以該詞的 BM25 貢獻就是約 1.8971。這只是說明公式的數學例子，不是本次實驗某筆文件的實際分數。

BM25 分數只能在「同一個問題、同一份索引」內排序。它與 Dense 的餘弦相似度沒有共同刻度，不能寫成 0.6 × cosine + 0.4 × BM25，除非先建立另外一套有依據的正規化與驗證契約。

BM25 的經典整理可參考 Robertson 與 Zaragoza 對機率相關性框架、詞頻飽和及文件長度正規化的說明：The Probabilistic Relevance Framework: BM25 and Beyond。

Hybrid：只融合名次，不混合異質分數

混合檢索（Hybrid Retrieval）希望同時利用 Dense 的語意訊號與 BM25 的字面訊號。本篇採用相互排序融合（Reciprocal Rank Fusion, RRF），不直接相加兩種分數，而是只看每筆候選在各清單中的名次。

一筆文件 $d$ 的 RRF 分數是：

$$
\operatorname{RRF}(d)=
\sum_{r\in R}\frac{1}{k+r(d)}
$$

• $R$ 是要融合的排名清單集合，本篇只有 Dense 與 BM25 兩份。
• $r(d)$ 是文件 $d$ 在某份清單中的名次，第一名就是 1。
• $k$ 是降低極前面名次影響幅度的常數，本篇依原始 RRF 實驗固定為 60；它不是 Top-k，也不是 KTAS 級數。

若候選 A 在 Dense 排第 1、BM25 排第 3，RRF 分數是：

$$
\frac{1}{60+1}+\frac{1}{60+3}
=\frac{1}{61}+\frac{1}{63}
\approx0.03227
$$

若候選 B 在兩邊都排第 2，分數是 $1/62+1/62\approx0.03226$，所以 A 會以很小差距排在 B 前面。重點不是 0.03227 本身，而是「在任一清單很前面」與「在兩份清單都不差」都會得到貢獻。

RRF 輸入必須使用完全相同的候選集合。本篇每一個問題都要求 Dense、BM25 與 Hybrid 各自包含同樣九個 rule_id，不能讓其中一種方法先偷刪難題。若集合不同，程式直接報錯。RRF 的原始研究與 $k=60$ 設定可參考：Reciprocal Rank Fusion outperforms Condorcet and individual Rank Learning Methods。

下圖把固定契約畫成同一條資料流。請注意 Dense 與 BM25 都先產生九筆完整名次，RRF 才接手；最右側的前 k 筆仍只是候選，今天沒有重排序器。

上圖也說明為什麼三種原始分數不能橫向比較。餘弦相似度、BM25 分數與 RRF 分數各自只負責方法內排序；公開摘要會保存三種分數供追溯，但不會把它們畫在同一刻度上。

如何知道檢索有沒有把相關文件排前面

只有排名清單還不夠，我們需要先寫好「哪一筆文件應該被找到」，再計算指標。今天使用兩個互補但仍很初步的檢索指標。

Recall@k：相關文件有沒有進入前 k 筆

前 k 筆召回率（Recall at k, Recall@k）的公式是：

$$
\operatorname{Recall@k}=
\frac{\text{前 k 筆中的相關文件數}}
{\text{這個問題的全部相關文件數}}
$$

例如，一個問題有 2 筆相關文件，前 3 名只找到其中 1 筆，Recall@3 就是 $1/2=0.5$。數值越大表示漏掉的相關文件越少，但不代表前 k 筆沒有無關文件，也不代表生成答案一定安全。

本篇五個可計分問題各自只有 1 筆預期相關文件，因此每題 Recall@1 只有 0 或 1。最後再把五題平均，得到各方法的 mean Recall@1、Recall@3 與 Recall@5。

MRR：第一筆相關文件排得多前面

平均倒數排名（Mean Reciprocal Rank, MRR）先找每題第一筆相關文件的名次，再取倒數與平均：

$$
\operatorname{MRR}=\frac{1}{|Q|}\sum_{i=1}^{|Q|}\frac{1}{\operatorname{rank}_i}
$$

• $|Q|$ 是可計分問題數。
• $\operatorname{rank}_i$ 是第 $i$ 題第一筆相關文件的名次。
• 第一名的倒數是 1，第二名是 0.5，第五名是 0.2；越接近 1 代表相關文件通常越早出現。

若三題的第一筆相關文件分別排第 1、第 2 與第 5，MRR 是 $(1+0.5+0.2)/3\approx0.5667$。MRR 只看第一筆相關文件；若每題有多個相關文件，它不會告訴我們後面的相關文件是否都被找回，所以仍要和 Recall@k 一起看。

八筆 probe 為什麼要分成兩種

測試問題（Probe）是作者依 Day 16 可追溯公開知識與已明示缺口撰寫的小型查詢。它不是病患紀錄，也沒有 KTAS_RN 或 KTAS_expert。八筆 probe 在執行前就固定，不會看完排名才改預期答案。

三個知識缺口是完整成人主訴目錄、各主訴完整數值門檻，以及完整官方版本與變更紀錄。為什麼不能把它們算成「沒有相關文件，所以 Recall@k 應為 1」？因為本篇的第一階段檢索器沒有拒答門檻，無論知識庫是否支援，它一定會把九筆文件排出第一名。若此時把任一 top-1 當成正確，就是把「系統有回傳」誤寫成「知識庫有答案」。

因此三筆缺口只做診斷，scored=false，完全不進入五筆支援問題的分母。Day 20 會先建立更困難的測試與重排序契約；知識缺口與拒答機制則要等後續安全篇章另外驗證。

必須先修正 Day 18 留下的比較期待

Day 18 結尾原本預告要在 Day 19 正式比較原始、語意與雙重三種數值表示。但目前不能公平完成這個比較：Day 18 只有一筆純合成生命徵象查詢，而現有部分公開 KTAS 知識沒有足以為那筆數值案例標註相關規則的完整主訴門檻。

若為了湊出 Recall 或 MRR，先用直覺指定某條規則是答案，就會把未公開門檻偽裝成標準答案。因此 Day 19 改用 Day 17 已在看排名前固定的八筆「公開知識問題」，先比較 Dense、BM25 與 Hybrid 三種檢索方法。三種數值表示的正式比較會延後，直到取得可追溯且適用的規則，或建立不依賴虛構臨床標籤的有效 probe。

這不是換掉失敗結果，而是保護實驗問題與資料能回答的範圍。Day 18 的單一合成案例仍然保留為工程冒煙測試，不會被重新包裝成檢索效能證據。

實作前先看檔案如何分工

讀者不需要前往任何程式碼網站。接下來會提供每個必要檔案的完整內容；先用白話理解資料夾與檔案的分工：

configs/retrieval/
└── day-19-retrieval-contract.json       # 固定來源雜湊、三種方法、指標與輸出政策

src/triage_rag/retrieval/
├── __init__.py                           # Python 子套件入口
└── core.py                               # 切詞、BM25、RRF、輸入驗證與指標

scripts/
└── run_day19_retrieval_benchmark.py      # 串接文件、問題、Ollama、排名與輸出

tests/
└── test_retrieval.py                     # 十二項不需啟動模型的單元測試

results/public/
└── day-19-retrieval-benchmark.json       # 可公開的排名、指標與限制摘要

results/runs/day-19/<run-id>/
├── document-embeddings.jsonl             # 九筆文件完整向量，本機保存
├── query-embeddings.jsonl                # 八筆問題完整向量，本機保存
├── retrieval-results.json                # 每題三種方法的完整九筆排名
└── run-manifest.json                      # 輸入、輸出、版本、參數與雜湊

configs/ 保存人與程式都能閱讀的實驗契約；src/ 保存可被其他程式匯入的核心邏輯；scripts/ 放讀者直接執行的自動化入口；tests/ 用合法與故意破壞的輸入檢查安全條件；results/public/ 保存不含完整向量的穩定摘要；results/runs/ 則讓每次本機執行各自保留完整產物。每次執行還會建立執行清單（run manifest），用來記錄輸入、輸出、命令、參數、模型版本與檔案雜湊。

入口會沿用 Day 17 的 flat-chunks.jsonl 與 retrieval-probes.jsonl，以及 Day 18 已完成的 Ollama 用戶端。JavaScript 物件表示法（JavaScript Object Notation, JSON）是保存結構化設定與結果的文字格式；逐行 JSON（JSON Lines, JSONL）則讓每一行各自保存一個完整 JSON 物件。設定中的安全雜湊演算法 256 位元（Secure Hash Algorithm 256-bit, SHA-256）會確認上游檔案沒有在不知情時被替換。

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

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

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 17 已建立的九筆 flat chunks 與八筆 retrieval probes、Day 18 的 Ollama 用戶端與已下載的 qwen3-embedding:4b。本篇不讀取 Kaggle 病患列或參考標籤；以下是 Day 19 新增的完整實驗契約、檢索模組、執行入口與測試。完整向量、逐題排名、公開摘要與 run manifest 都由程式自動產生，不需要手動建立。

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

mkdir -p configs/retrieval src/triage_rag/retrieval scripts tests results/public results/runs/day-19

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

檔案 1：建立 configs/retrieval/day-19-retrieval-contract.json

鎖定上游檔案雜湊、Dense／BM25／Hybrid 的固定條件、五筆可計分與三筆缺口政策，以及本機與公開輸出位置。

請在文字編輯器建立 configs/retrieval/day-19-retrieval-contract.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "experiment_id": "day-19-dense-bm25-hybrid-retrieval",
  "scope": "author_written_public_knowledge_probe_benchmark_not_clinical_evaluation",
  "sources": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "retrieval_probes_path": "data/knowledge/ktas-public-v1/day-17/retrieval-probes.jsonl",
    "retrieval_probes_sha256": "98335e9265741e9c8428654c9e4dd92b07249600f0b0f66f5e8d2ff6a0e7e18f",
    "embedding_config_path": "configs/representation/day-18-numeric-semantics.json",
    "embedding_config_sha256": "3d24a3688dbff4e617270f6fa9e953c055c835c4b0b10718eb6fa5a5ef3ada8b",
    "model_lock_path": "configs/models/local-ollama-model-lock.json",
    "model_lock_sha256": "0cd15cb4d7e735a19cbe439fe9a9344f717029694a4441764f67e396033cc7f4"
  },
  "input_contract": {
    "expected_document_count": 9,
    "expected_probe_count": 8,
    "expected_supported_probe_count": 5,
    "expected_gap_probe_count": 3,
    "supported_behavior": "retrieve",
    "gap_behavior": "disclose_gap",
    "forbidden_fields": [
      "record_index",
      "KTAS_RN",
      "KTAS_expert",
      "Group",
      "diagnosis",
      "disposition",
      "length_of_stay",
      "Error_group",
      "mistriage"
    ]
  },
  "methods": {
    "dense": {
      "name": "qwen3_embedding_cosine",
      "query_instruction_source": "embedding_config",
      "document_instruction": null,
      "reranker_enabled": false
    },
    "bm25": {
      "name": "bm25_deterministic_zh_char_bigram",
      "tokenizer": "unicode_nfkc_lowercase_latin_number_cjk_unigram_bigram_v1",
      "k1": 1.2,
      "b": 0.75,
      "query_term_frequency": "binary_unique_terms",
      "reranker_enabled": false
    },
    "hybrid": {
      "name": "reciprocal_rank_fusion_dense_bm25",
      "rrf_k": 60,
      "inputs": ["dense", "bm25"],
      "same_candidate_set_required": true,
      "reranker_enabled": false
    }
  },
  "evaluation": {
    "cutoffs": [1, 3, 5],
    "supported_probe_metrics": ["mean_recall_at_k", "mean_reciprocal_rank"],
    "gap_probe_policy": "diagnostic_only_not_scored_without_abstention_threshold",
    "primary_comparison_methods": ["dense", "bm25", "hybrid"],
    "selection_policy": "do_not_select_or_tune_method_from_eight_probe_results",
    "online_reranker": "disabled_until_day_20_provider_contract"
  },
  "outputs": {
    "public_summary_path": "results/public/day-19-retrieval-benchmark.json",
    "run_output_root": "results/runs/day-19"
  }
}

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

檔案 2：建立 src/triage_rag/retrieval/__init__.py

建立 retrieval 子套件，公開 BM25、RRF、輸入驗證與檢索評估函式。

請在文字編輯器建立 src/triage_rag/retrieval/__init__.py，貼入以下完整內容並儲存：

"""Deterministic retrieval helpers introduced in Day 19."""

from .core import (
    BM25Index,
    RetrievalContractError,
    evaluate_rankings,
    reciprocal_rank_fusion,
    tokenize_for_bm25,
    validate_retrieval_inputs,
)


__all__ = [
    "BM25Index",
    "RetrievalContractError",
    "evaluate_rankings",
    "reciprocal_rank_fusion",
    "tokenize_for_bm25",
    "validate_retrieval_inputs",
]

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

檔案 3：建立 src/triage_rag/retrieval/core.py

實作繁中文字詞切分、BM25 排名、相互排序融合、候選集合契約、Recall@k、MRR 與缺口排除政策。

請在文字編輯器建立 src/triage_rag/retrieval/core.py，貼入以下完整內容並儲存：

"""BM25, reciprocal-rank fusion, input validation, and retrieval metrics."""

from __future__ import annotations

import math
import re
import unicodedata
from collections import Counter
from typing import Any, Dict, Iterable, List, Mapping, Sequence


JsonObject = Dict[str, Any]
TOKEN_PATTERN = re.compile(
    r"[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*|\d+(?:\.\d+)?|[\u3400-\u9fff]+"
)
SUPPORTED_BEHAVIORS = {"retrieve", "disclose_gap"}


class RetrievalContractError(ValueError):
    """Raised when a retrieval input or ranking violates the Day 19 contract."""


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


def _duplicates(values: Iterable[str]) -> List[str]:
    counts = Counter(values)
    return sorted(value for value, count in counts.items() if count > 1)


def _all_keys(value: Any) -> Iterable[str]:
    if isinstance(value, Mapping):
        for key, child in value.items():
            yield str(key)
            yield from _all_keys(child)
    elif isinstance(value, list):
        for child in value:
            yield from _all_keys(child)


def tokenize_for_bm25(text: str) -> List[str]:
    """Create deterministic Latin/number tokens plus CJK unigrams and bigrams."""

    normalized = unicodedata.normalize("NFKC", _nonempty_text(text, "text")).lower()
    tokens: List[str] = []
    for piece in TOKEN_PATTERN.findall(normalized):
        if all("\u3400" <= character <= "\u9fff" for character in piece):
            characters = list(piece)
            tokens.extend(characters)
            tokens.extend(
                left + right for left, right in zip(characters, characters[1:])
            )
        else:
            tokens.append(piece)
    if not tokens:
        raise RetrievalContractError("BM25 文字正規化後沒有可用 token")
    return tokens


class BM25Index:
    """Small deterministic BM25 index with fixed Robertson-style positive IDF."""

    def __init__(
        self,
        documents: Sequence[Mapping[str, Any]],
        *,
        k1: float = 1.2,
        b: float = 0.75,
    ) -> None:
        if not isinstance(documents, list) or not documents:
            raise RetrievalContractError("BM25 documents 必須是非空陣列")
        if not math.isfinite(k1) or k1 <= 0:
            raise RetrievalContractError("BM25 k1 必須是大於 0 的有限數字")
        if not math.isfinite(b) or not 0 <= b <= 1:
            raise RetrievalContractError("BM25 b 必須介於 0 到 1")

        normalized_documents: List[JsonObject] = []
        for item in documents:
            rule_id = _nonempty_text(item.get("rule_id"), "rule_id")
            text = _nonempty_text(item.get("text"), f"{rule_id}.text")
            normalized_documents.append({**dict(item), "rule_id": rule_id, "text": text})
        duplicates = _duplicates(item["rule_id"] for item in normalized_documents)
        if duplicates:
            raise RetrievalContractError(f"BM25 rule_id 不可重複：{duplicates}")

        self.documents = normalized_documents
        self.k1 = float(k1)
        self.b = float(b)
        self.term_frequencies = [
            Counter(tokenize_for_bm25(item["text"])) for item in self.documents
        ]
        self.document_lengths = [sum(counter.values()) for counter in self.term_frequencies]
        self.average_document_length = sum(self.document_lengths) / len(
            self.document_lengths
        )
        self.document_frequencies: Counter[str] = Counter()
        for counter in self.term_frequencies:
            self.document_frequencies.update(counter.keys())

    def _idf(self, term: str) -> float:
        document_count = len(self.documents)
        frequency = self.document_frequencies.get(term, 0)
        return math.log(
            1.0 + (document_count - frequency + 0.5) / (frequency + 0.5)
        )

    def score(self, query: str, document_index: int) -> float:
        if document_index not in range(len(self.documents)):
            raise RetrievalContractError("BM25 document_index 超出範圍")
        query_terms = sorted(set(tokenize_for_bm25(query)))
        frequencies = self.term_frequencies[document_index]
        document_length = self.document_lengths[document_index]
        score = 0.0
        for term in query_terms:
            term_frequency = frequencies.get(term, 0)
            if term_frequency == 0:
                continue
            denominator = term_frequency + self.k1 * (
                1.0
                - self.b
                + self.b * document_length / self.average_document_length
            )
            score += self._idf(term) * (
                term_frequency * (self.k1 + 1.0) / denominator
            )
        return score

    def rank(self, query: str) -> List[JsonObject]:
        ranked = []
        for index, document in enumerate(self.documents):
            ranked.append(
                {
                    "rule_id": document["rule_id"],
                    "chunk_id": document.get("chunk_id"),
                    "topic": document.get("topic"),
                    "bm25_score": self.score(query, index),
                }
            )
        ranked.sort(key=lambda item: (-item["bm25_score"], item["rule_id"]))
        for rank, item in enumerate(ranked, start=1):
            item["rank"] = rank
        return ranked

    def summary(self) -> JsonObject:
        return {
            "document_count": len(self.documents),
            "unique_token_count": len(self.document_frequencies),
            "average_document_length": self.average_document_length,
            "minimum_document_length": min(self.document_lengths),
            "maximum_document_length": max(self.document_lengths),
            "k1": self.k1,
            "b": self.b,
        }


def _validated_ranking(
    method: str, ranking: Sequence[Mapping[str, Any]]
) -> Dict[str, int]:
    if not isinstance(ranking, list) or not ranking:
        raise RetrievalContractError(f"{method} ranking 必須是非空陣列")
    rule_ids = [_nonempty_text(item.get("rule_id"), "rule_id") for item in ranking]
    duplicates = _duplicates(rule_ids)
    if duplicates:
        raise RetrievalContractError(f"{method} ranking 有重複候選：{duplicates}")
    ranks = [item.get("rank") for item in ranking]
    if any(isinstance(rank, bool) or not isinstance(rank, int) for rank in ranks):
        raise RetrievalContractError(f"{method} rank 必須是整數")
    if sorted(ranks) != list(range(1, len(ranking) + 1)):
        raise RetrievalContractError(f"{method} rank 必須從 1 連續排列")
    return {rule_id: rank for rule_id, rank in zip(rule_ids, ranks)}


def reciprocal_rank_fusion(
    rankings: Mapping[str, Sequence[Mapping[str, Any]]],
    *,
    rrf_k: int = 60,
) -> List[JsonObject]:
    """Fuse complete rankings while requiring the same candidate set."""

    if not isinstance(rankings, Mapping) or len(rankings) < 2:
        raise RetrievalContractError("RRF 至少需要兩份 ranking")
    if isinstance(rrf_k, bool) or not isinstance(rrf_k, int) or rrf_k <= 0:
        raise RetrievalContractError("RRF k 必須是正整數")
    rank_maps = {
        _nonempty_text(method, "method"): _validated_ranking(method, list(ranking))
        for method, ranking in rankings.items()
    }
    candidate_sets = [set(rank_map) for rank_map in rank_maps.values()]
    if any(candidate_set != candidate_sets[0] for candidate_set in candidate_sets[1:]):
        raise RetrievalContractError("RRF 輸入必須使用完全相同的候選集合")

    fused = []
    for rule_id in sorted(candidate_sets[0]):
        component_ranks = {
            method: rank_map[rule_id] for method, rank_map in rank_maps.items()
        }
        fused.append(
            {
                "rule_id": rule_id,
                "rrf_score": sum(
                    1.0 / (rrf_k + rank) for rank in component_ranks.values()
                ),
                "component_ranks": component_ranks,
            }
        )
    fused.sort(key=lambda item: (-item["rrf_score"], item["rule_id"]))
    for rank, item in enumerate(fused, start=1):
        item["rank"] = rank
    return fused


def validate_retrieval_inputs(
    documents: Sequence[Mapping[str, Any]],
    probes: Sequence[Mapping[str, Any]],
    config: Mapping[str, Any],
) -> JsonObject:
    contract = config["input_contract"]
    if len(documents) != int(contract["expected_document_count"]):
        raise RetrievalContractError("文件數量與 Day 19 契約不同")
    if len(probes) != int(contract["expected_probe_count"]):
        raise RetrievalContractError("probe 數量與 Day 19 契約不同")

    forbidden = set(contract["forbidden_fields"])
    present_forbidden = sorted(
        forbidden.intersection(
            key for item in [*documents, *probes] for key in _all_keys(item)
        )
    )
    if present_forbidden:
        raise RetrievalContractError(f"檢索輸入出現禁止欄位：{present_forbidden}")

    rule_ids = [
        _nonempty_text(item.get("metadata", {}).get("rule_id"), "rule_id")
        for item in documents
    ]
    duplicates = _duplicates(rule_ids)
    if duplicates:
        raise RetrievalContractError(f"文件 rule_id 不可重複：{duplicates}")
    if any(item.get("retrieval_role") != "search_candidate" for item in documents):
        raise RetrievalContractError("每筆文件都必須是 search_candidate")
    for rule_id, item in zip(rule_ids, documents):
        _nonempty_text(item.get("search_text"), f"{rule_id}.search_text")

    probe_ids = [_nonempty_text(item.get("probe_id"), "probe_id") for item in probes]
    duplicates = _duplicates(probe_ids)
    if duplicates:
        raise RetrievalContractError(f"probe_id 不可重複：{duplicates}")
    supported_count = 0
    gap_count = 0
    document_rule_ids = set(rule_ids)
    for probe in probes:
        probe_id = probe["probe_id"]
        _nonempty_text(probe.get("query"), f"{probe_id}.query")
        behavior = probe.get("expected_behavior")
        if behavior not in SUPPORTED_BEHAVIORS:
            raise RetrievalContractError(f"{probe_id}.expected_behavior 不支援")
        if probe.get("uses_patient_record") is not False:
            raise RetrievalContractError(f"{probe_id} 不得使用病患紀錄")
        if probe.get("uses_reference_label") is not False:
            raise RetrievalContractError(f"{probe_id} 不得使用參考標籤")
        expected = probe.get("expected_rule_ids")
        if not isinstance(expected, list) or len(expected) != len(set(expected)):
            raise RetrievalContractError(f"{probe_id}.expected_rule_ids 格式錯誤")
        unknown = sorted(set(expected) - document_rule_ids)
        if unknown:
            raise RetrievalContractError(f"{probe_id} 引用未知規則：{unknown}")
        if behavior == contract["supported_behavior"]:
            supported_count += 1
            if not expected or probe.get("expected_gap_id") is not None:
                raise RetrievalContractError(f"{probe_id} 支援案例標記不一致")
        else:
            gap_count += 1
            if expected:
                raise RetrievalContractError(f"{probe_id} 缺口案例標記不一致")
            _nonempty_text(probe.get("expected_gap_id"), f"{probe_id}.expected_gap_id")

    if supported_count != int(contract["expected_supported_probe_count"]):
        raise RetrievalContractError("支援 probe 數量與契約不同")
    if gap_count != int(contract["expected_gap_probe_count"]):
        raise RetrievalContractError("缺口 probe 數量與契約不同")
    return {
        "document_count": len(documents),
        "probe_count": len(probes),
        "supported_probe_count": supported_count,
        "gap_probe_count": gap_count,
        "uses_patient_records": False,
        "uses_reference_labels": False,
    }


def evaluate_rankings(
    probes: Sequence[Mapping[str, Any]],
    rankings_by_probe: Mapping[str, Mapping[str, Sequence[Mapping[str, Any]]]],
    *,
    methods: Sequence[str],
    cutoffs: Sequence[int],
) -> JsonObject:
    if not methods or len(methods) != len(set(methods)):
        raise RetrievalContractError("methods 必須是非空且不可重複")
    if (
        not cutoffs
        or list(cutoffs) != sorted(set(cutoffs))
        or any(isinstance(value, bool) or not isinstance(value, int) or value <= 0 for value in cutoffs)
    ):
        raise RetrievalContractError("cutoffs 必須是遞增的正整數")

    positive_results = []
    gap_results = []
    for probe in probes:
        probe_id = probe["probe_id"]
        method_rankings = rankings_by_probe.get(probe_id)
        if not isinstance(method_rankings, Mapping) or set(method_rankings) != set(methods):
            raise RetrievalContractError(f"{probe_id} 缺少完整方法 ranking")
        validated = {
            method: _validated_ranking(method, list(method_rankings[method]))
            for method in methods
        }
        if probe["expected_behavior"] == "retrieve":
            expected = list(probe["expected_rule_ids"])
            method_metrics = {}
            for method in methods:
                rank_map = validated[method]
                expected_ranks = [rank_map[rule_id] for rule_id in expected]
                first_rank = min(expected_ranks)
                method_metrics[method] = {
                    "first_relevant_rank": first_rank,
                    "reciprocal_rank": 1.0 / first_rank,
                    "recall_at_k": {
                        str(cutoff): sum(rank <= cutoff for rank in expected_ranks)
                        / len(expected_ranks)
                        for cutoff in cutoffs
                    },
                }
            positive_results.append(
                {
                    "probe_id": probe_id,
                    "query": probe["query"],
                    "expected_rule_ids": expected,
                    "methods": method_metrics,
                }
            )
        else:
            gap_results.append(
                {
                    "probe_id": probe_id,
                    "query": probe["query"],
                    "expected_gap_id": probe["expected_gap_id"],
                    "scored": False,
                    "reason": "尚未鎖定拒答門檻；第一階段檢索器一定會回傳候選。",
                    "top_rule_ids": {
                        method: min(validated[method], key=validated[method].get)
                        for method in methods
                    },
                }
            )

    if not positive_results:
        raise RetrievalContractError("至少需要一筆可計分的 retrieve probe")
    aggregate = {}
    for method in methods:
        aggregate[method] = {
            "supported_probe_count": len(positive_results),
            "mean_reciprocal_rank": sum(
                item["methods"][method]["reciprocal_rank"]
                for item in positive_results
            )
            / len(positive_results),
            "mean_recall_at_k": {
                str(cutoff): sum(
                    item["methods"][method]["recall_at_k"][str(cutoff)]
                    for item in positive_results
                )
                / len(positive_results)
                for cutoff in cutoffs
            },
        }
    return {
        "aggregate_supported_metrics": aggregate,
        "positive_probe_results": positive_results,
        "gap_probe_diagnostics": gap_results,
    }

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

檔案 4：建立 scripts/run_day19_retrieval_benchmark.py

驗證上游雜湊與本機模型 digest，產生 Dense、BM25、Hybrid 完整排名，並寫出逐題結果、公開摘要與執行紀錄。

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

#!/usr/bin/env python3
"""Run the Day 19 dense, BM25, and RRF hybrid retrieval benchmark."""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Mapping

from triage_rag.knowledge_base.builder import read_jsonl, write_jsonl
from triage_rag.representation.embedding import (
    embed_texts,
    get_runtime_metadata,
    rank_documents,
    vector_sha256,
)
from triage_rag.reproducibility import (
    file_record,
    git_state,
    load_json,
    resolve_project_path,
    sha256_bytes,
    sha256_file,
    write_json,
)
from triage_rag.retrieval.core import (
    BM25Index,
    evaluate_rankings,
    reciprocal_rank_fusion,
    validate_retrieval_inputs,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/retrieval/day-19-retrieval-contract.json"
METHODS = ("dense", "bm25", "hybrid")
JsonObject = Dict[str, Any]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="以相同 KTAS 公開知識候選比較 Dense、BM25 與 RRF Hybrid。"
    )
    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 19 契約。"
        )


def _verify_sources(config: Mapping[str, Any]) -> List[Path]:
    sources = config["sources"]
    pairs = [
        ("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
        (
            "retrieval_probes_path",
            "retrieval_probes_sha256",
            "Day 17 retrieval probes",
        ),
        (
            "embedding_config_path",
            "embedding_config_sha256",
            "Day 18 embedding 設定",
        ),
        ("model_lock_path", "model_lock_sha256", "本機模型鎖定"),
    ]
    paths = []
    for path_key, hash_key, label in pairs:
        path = resolve_project_path(PROJECT_ROOT, sources[path_key])
        _verify_sha256(path, sources[hash_key], label)
        paths.append(path)
    return paths


def _verify_runtime_model(
    runtime: Mapping[str, Any],
    embedding: Mapping[str, Any],
    model_lock: Mapping[str, Any],
) -> None:
    dense_lock = model_lock["models"]["dense_embedding"]
    checks = {
        "lock_name": (dense_lock.get("name"), embedding["model"]),
        "lock_id": (
            dense_lock.get("model_id"),
            embedding["expected_model_id_prefix"],
        ),
        "runtime_name": (runtime.get("model"), embedding["model"]),
        "runtime_digest": (
            runtime.get("digest"),
            embedding["expected_model_digest"],
        ),
        "parameter_size": (
            runtime.get("parameter_size"),
            embedding["expected_parameter_size"],
        ),
        "quantization_level": (
            runtime.get("quantization_level"),
            embedding["expected_quantization_level"],
        ),
    }
    mismatches = {
        name: {"actual": actual, "expected": expected}
        for name, (actual, expected) in checks.items()
        if actual != expected
    }
    if mismatches:
        raise ValueError(f"本機 embedding 模型與 Day 19 契約不同：{mismatches}")


def _round_floats(value: Any) -> Any:
    if isinstance(value, float):
        return round(value, 8)
    if isinstance(value, dict):
        return {key: _round_floats(child) for key, child in value.items()}
    if isinstance(value, list):
        return [_round_floats(child) for child in value]
    return value


def _public_ranking(
    ranking: List[JsonObject],
    *,
    top_k: int,
    document_by_rule: Mapping[str, Mapping[str, Any]],
) -> List[JsonObject]:
    public = []
    for item in ranking[:top_k]:
        document = document_by_rule[item["rule_id"]]
        record = {
            "rank": item["rank"],
            "rule_id": item["rule_id"],
            "chunk_id": document["chunk_id"],
            "topic": document["metadata"]["topic"],
        }
        for field in ("cosine_similarity", "bm25_score", "rrf_score"):
            if field in item:
                record[field] = round(float(item[field]), 8)
        if "component_ranks" in item:
            record["component_ranks"] = item["component_ranks"]
        public.append(record)
    return public


def main() -> int:
    args = parse_args()
    started_at = datetime.now(timezone.utc)
    config_path = resolve_project_path(PROJECT_ROOT, args.config)
    config = load_json(config_path)
    if config.get("schema_version") != 1:
        raise ValueError("目前只支援 schema_version=1")
    if config.get("scope") != (
        "author_written_public_knowledge_probe_benchmark_not_clinical_evaluation"
    ):
        raise ValueError("Day 19 scope 不正確")
    source_paths = _verify_sources(config)

    sources = config["sources"]
    documents = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["flat_chunks_path"])
    )
    probes = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["retrieval_probes_path"])
    )
    input_summary = validate_retrieval_inputs(documents, probes, config)

    embedding_config = load_json(
        resolve_project_path(PROJECT_ROOT, sources["embedding_config_path"])
    )
    embedding = embedding_config["embedding"]
    model_lock = load_json(
        resolve_project_path(PROJECT_ROOT, sources["model_lock_path"])
    )
    runtime = get_runtime_metadata(
        embedding["endpoint"],
        embedding["model"],
        timeout_seconds=int(embedding["timeout_seconds"]),
    )
    _verify_runtime_model(runtime, embedding, model_lock)

    document_texts = [item["search_text"] for item in documents]
    document_vectors, document_embedding_metrics = embed_texts(
        document_texts, embedding
    )
    query_inputs = [
        f"Instruct: {embedding['query_instruction']}\nQuery: {probe['query']}"
        for probe in probes
    ]
    query_vectors, query_embedding_metrics = embed_texts(query_inputs, embedding)

    bm25_config = config["methods"]["bm25"]
    bm25_documents = [
        {
            "rule_id": item["metadata"]["rule_id"],
            "chunk_id": item["chunk_id"],
            "topic": item["metadata"]["topic"],
            "text": item["search_text"],
        }
        for item in documents
    ]
    bm25_index = BM25Index(
        bm25_documents,
        k1=float(bm25_config["k1"]),
        b=float(bm25_config["b"]),
    )
    rrf_k = int(config["methods"]["hybrid"]["rrf_k"])
    rankings_by_probe: Dict[str, Dict[str, List[JsonObject]]] = {}
    for probe, query_vector in zip(probes, query_vectors):
        dense = rank_documents(query_vector, documents, document_vectors)
        bm25 = bm25_index.rank(probe["query"])
        hybrid = reciprocal_rank_fusion(
            {"dense": dense, "bm25": bm25}, rrf_k=rrf_k
        )
        rankings_by_probe[probe["probe_id"]] = {
            "dense": dense,
            "bm25": bm25,
            "hybrid": hybrid,
        }

    methods = list(config["evaluation"]["primary_comparison_methods"])
    if methods != list(METHODS):
        raise ValueError(f"Day 19 方法順序必須是 {METHODS}")
    cutoffs = list(config["evaluation"]["cutoffs"])
    evaluation = evaluate_rankings(
        probes,
        rankings_by_probe,
        methods=methods,
        cutoffs=cutoffs,
    )

    run_id = (
        f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-"
        f"{sha256_file(config_path)[: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)

    document_embedding_records = []
    for document, vector in zip(documents, document_vectors):
        document_embedding_records.append(
            {
                "schema_version": 1,
                "chunk_id": document["chunk_id"],
                "rule_id": document["metadata"]["rule_id"],
                "search_text_sha256": document["search_text_sha256"],
                "model": embedding["model"],
                "model_digest": runtime["digest"],
                "dimensions": embedding["dimensions"],
                "vector_sha256": vector_sha256(vector),
                "vector": vector,
            }
        )
    query_embedding_records = []
    for probe, query_input, vector in zip(probes, query_inputs, query_vectors):
        query_embedding_records.append(
            {
                "schema_version": 1,
                "probe_id": probe["probe_id"],
                "origin": probe["origin"],
                "uses_patient_record": False,
                "uses_reference_label": False,
                "query_input": query_input,
                "query_input_sha256": sha256_bytes(query_input.encode("utf-8")),
                "model": embedding["model"],
                "model_digest": runtime["digest"],
                "dimensions": embedding["dimensions"],
                "vector_sha256": vector_sha256(vector),
                "vector": vector,
            }
        )
    document_vectors_path = run_directory / "document-embeddings.jsonl"
    query_vectors_path = run_directory / "query-embeddings.jsonl"
    full_results_path = run_directory / "retrieval-results.json"
    write_jsonl(document_vectors_path, document_embedding_records)
    write_jsonl(query_vectors_path, query_embedding_records)
    write_json(
        full_results_path,
        {
            "schema_version": 1,
            "experiment_id": config["experiment_id"],
            "evaluation": evaluation,
            "rankings_by_probe": rankings_by_probe,
        },
    )

    document_by_rule = {
        item["metadata"]["rule_id"]: item for item in documents
    }
    public_rankings = {
        probe["probe_id"]: {
            method: _public_ranking(
                rankings_by_probe[probe["probe_id"]][method],
                top_k=max(cutoffs),
                document_by_rule=document_by_rule,
            )
            for method in methods
        }
        for probe in probes
    }
    public_summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "runtime": runtime,
        "input_contract": input_summary,
        "method_contracts": {
            "dense": {
                **config["methods"]["dense"],
                "model": embedding["model"],
                "model_digest": runtime["digest"],
                "dimensions": embedding["dimensions"],
                "query_instruction": embedding["query_instruction"],
                "truncate": embedding["truncate"],
            },
            "bm25": {
                **config["methods"]["bm25"],
                "index_summary": _round_floats(bm25_index.summary()),
            },
            "hybrid": config["methods"]["hybrid"],
        },
        "evaluation_contract": config["evaluation"],
        "evaluation": _round_floats(evaluation),
        "top_rankings": public_rankings,
        "query_vector_sha256": {
            probe["probe_id"]: vector_sha256(vector)
            for probe, vector in zip(probes, query_vectors)
        },
        "checks": {
            "source_hashes_match": True,
            "model_lock_matches_runtime": True,
            "same_nine_candidates_for_all_methods": all(
                len(rankings_by_probe[probe["probe_id"]][method])
                == input_summary["document_count"]
                for probe in probes
                for method in methods
            ),
            "hybrid_only_fuses_dense_and_bm25_ranks": True,
            "gap_probes_excluded_from_supported_metrics": all(
                item["scored"] is False
                for item in evaluation["gap_probe_diagnostics"]
            ),
            "patient_and_reference_label_data_absent": True,
            "generation_gate_and_reranker_disabled": True,
        },
        "score_interpretation": (
            "cosine、BM25 與 RRF 分數只在各自方法內排序，不可跨方法比較絕對大小。"
        ),
        "warning": (
            "這是五筆可計分與三筆缺口的作者自寫小型 probe set；"
            "不得用來宣稱某檢索器普遍較好、調整參數、評估五級分類或推論臨床安全。"
        ),
    }
    public_summary_path = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["public_summary_path"]
    )
    write_json(public_summary_path, public_summary)

    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/run_day19_retrieval_benchmark.py",
        ],
        "git": git_state(PROJECT_ROOT),
        "runtime": runtime,
        "embedding_usage": {
            "documents": document_embedding_metrics,
            "queries": query_embedding_metrics,
        },
        "parameters": {
            "bm25_k1": bm25_index.k1,
            "bm25_b": bm25_index.b,
            "rrf_k": rrf_k,
            "cutoffs": cutoffs,
        },
        "inputs": [file_record(PROJECT_ROOT, args.config)]
        + [
            file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
            for path in source_paths
        ],
        "outputs": [
            file_record(
                PROJECT_ROOT, str(document_vectors_path.relative_to(PROJECT_ROOT))
            ),
            file_record(
                PROJECT_ROOT, str(query_vectors_path.relative_to(PROJECT_ROOT))
            ),
            file_record(
                PROJECT_ROOT, str(full_results_path.relative_to(PROJECT_ROOT))
            ),
            file_record(PROJECT_ROOT, config["outputs"]["public_summary_path"]),
        ],
        "privacy": (
            "只使用九筆公開 KTAS 摘要與八筆作者自寫 probe；"
            "不讀取 Kaggle 病患列、KTAS_RN 或 KTAS_expert。"
        ),
        "scope": config["scope"],
    }
    write_json(run_directory / "run-manifest.json", manifest)

    print("Day 19 Dense／BM25／Hybrid 檢索 benchmark：通過")
    print(
        f"候選：{input_summary['document_count']} 筆；probe："
        f"{input_summary['supported_probe_count']} 筆可計分＋"
        f"{input_summary['gap_probe_count']} 筆缺口診斷"
    )
    for method in methods:
        metrics = evaluation["aggregate_supported_metrics"][method]
        recall = metrics["mean_recall_at_k"]
        recall_text = ", ".join(
            f"R@{cutoff}={recall[str(cutoff)]:.4f}" for cutoff in cutoffs
        )
        print(
            f"{method}: {recall_text}, "
            f"MRR={metrics['mean_reciprocal_rank']:.4f}"
        )
    print("缺口 probe：只列候選，不納入 Recall 或 MRR 分母")
    print(f"公開摘要：{public_summary_path.relative_to(PROJECT_ROOT)}")
    print(f"完整向量與排名：{run_directory.relative_to(PROJECT_ROOT)}")
    print("注意：八筆作者自寫 probe 不能用來選出普遍最佳檢索器。")
    return 0


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

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

檔案 5：建立 tests/test_retrieval.py

驗證 BM25 切詞與參數、RRF 公式與相同候選集合、禁止欄位、缺口標記，以及 Recall／MRR 分母政策。

請在文字編輯器建立 tests/test_retrieval.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.retrieval.core import (
    BM25Index,
    RetrievalContractError,
    evaluate_rankings,
    reciprocal_rank_fusion,
    tokenize_for_bm25,
    validate_retrieval_inputs,
)
from triage_rag.reproducibility import load_json


ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = ROOT / "configs" / "retrieval" / "day-19-retrieval-contract.json"


class Day19RetrievalTests(unittest.TestCase):
    def setUp(self) -> None:
        self.config = load_json(CONFIG_PATH)
        sources = self.config["sources"]
        self.documents = read_jsonl(ROOT / sources["flat_chunks_path"])
        self.probes = read_jsonl(ROOT / sources["retrieval_probes_path"])

    def test_tokenizer_preserves_latin_number_and_cjk_bigrams(self) -> None:
        tokens = tokenize_for_bm25("KTAS 第一級 1 與 SpO2=91.0")
        self.assertIn("ktas", tokens)
        self.assertIn("1", tokens)
        self.assertIn("spo2", tokens)
        self.assertIn("91.0", tokens)
        self.assertIn("第一", tokens)
        self.assertIn("一級", tokens)

    def test_bm25_ranks_rare_exact_term_first(self) -> None:
        index = BM25Index(
            [
                {"rule_id": "a", "text": "一般流程與照護順序"},
                {"rule_id": "b", "text": "血糖與脫水程度"},
            ]
        )
        ranking = index.rank("脫水")
        self.assertEqual(ranking[0]["rule_id"], "b")
        self.assertGreater(ranking[0]["bm25_score"], ranking[1]["bm25_score"])

    def test_bm25_rejects_invalid_parameters(self) -> None:
        documents = [{"rule_id": "a", "text": "文字"}]
        with self.assertRaisesRegex(RetrievalContractError, "k1"):
            BM25Index(documents, k1=0)
        with self.assertRaisesRegex(RetrievalContractError, "b"):
            BM25Index(documents, b=1.1)

    def test_rrf_formula_and_tie_break_are_deterministic(self) -> None:
        fused = reciprocal_rank_fusion(
            {
                "dense": [
                    {"rule_id": "a", "rank": 1},
                    {"rule_id": "b", "rank": 2},
                ],
                "bm25": [
                    {"rule_id": "b", "rank": 1},
                    {"rule_id": "a", "rank": 2},
                ],
            },
            rrf_k=60,
        )
        self.assertEqual([item["rule_id"] for item in fused], ["a", "b"])
        self.assertAlmostEqual(
            fused[0]["rrf_score"], 1 / 61 + 1 / 62
        )

    def test_rrf_rejects_candidate_set_changes(self) -> None:
        with self.assertRaisesRegex(RetrievalContractError, "候選集合"):
            reciprocal_rank_fusion(
                {
                    "dense": [{"rule_id": "a", "rank": 1}],
                    "bm25": [{"rule_id": "b", "rank": 1}],
                }
            )

    def test_day17_inputs_pass_contract(self) -> None:
        summary = validate_retrieval_inputs(
            self.documents, self.probes, self.config
        )
        self.assertEqual(summary["document_count"], 9)
        self.assertEqual(summary["supported_probe_count"], 5)
        self.assertEqual(summary["gap_probe_count"], 3)

    def test_patient_or_label_fields_are_rejected(self) -> None:
        invalid = copy.deepcopy(self.probes)
        invalid[0]["KTAS_expert"] = 1
        with self.assertRaisesRegex(RetrievalContractError, "禁止欄位"):
            validate_retrieval_inputs(self.documents, invalid, self.config)

    def test_blank_document_search_text_is_rejected(self) -> None:
        invalid = copy.deepcopy(self.documents)
        invalid[0]["search_text"] = "   "
        with self.assertRaisesRegex(RetrievalContractError, "search_text"):
            validate_retrieval_inputs(invalid, self.probes, self.config)

    def test_gap_probe_cannot_claim_a_relevant_rule(self) -> None:
        invalid = copy.deepcopy(self.probes)
        gap = next(item for item in invalid if item["expected_behavior"] == "disclose_gap")
        gap["expected_rule_ids"] = [self.documents[0]["metadata"]["rule_id"]]
        with self.assertRaisesRegex(RetrievalContractError, "缺口案例"):
            validate_retrieval_inputs(self.documents, invalid, self.config)

    def test_gap_probe_requires_a_nonempty_gap_id(self) -> None:
        invalid = copy.deepcopy(self.probes)
        gap = next(item for item in invalid if item["expected_behavior"] == "disclose_gap")
        gap["expected_gap_id"] = ""
        with self.assertRaisesRegex(RetrievalContractError, "expected_gap_id"):
            validate_retrieval_inputs(self.documents, invalid, self.config)

    def test_metrics_keep_gap_probes_out_of_denominator(self) -> None:
        probes = [
            {
                "probe_id": "positive",
                "query": "正向",
                "expected_behavior": "retrieve",
                "expected_rule_ids": ["a"],
            },
            {
                "probe_id": "gap",
                "query": "缺口",
                "expected_behavior": "disclose_gap",
                "expected_rule_ids": [],
                "expected_gap_id": "gap-id",
            },
        ]
        rankings = {
            probe_id: {
                "dense": [
                    {"rule_id": "b", "rank": 1},
                    {"rule_id": "a", "rank": 2},
                ],
                "bm25": [
                    {"rule_id": "a", "rank": 1},
                    {"rule_id": "b", "rank": 2},
                ],
                "hybrid": [
                    {"rule_id": "a", "rank": 1},
                    {"rule_id": "b", "rank": 2},
                ],
            }
            for probe_id in ("positive", "gap")
        }
        result = evaluate_rankings(
            probes,
            rankings,
            methods=["dense", "bm25", "hybrid"],
            cutoffs=[1, 2],
        )
        self.assertEqual(
            result["aggregate_supported_metrics"]["dense"]["supported_probe_count"],
            1,
        )
        self.assertEqual(
            result["aggregate_supported_metrics"]["dense"]["mean_reciprocal_rank"],
            0.5,
        )
        self.assertFalse(result["gap_probe_diagnostics"][0]["scored"])

    def test_metrics_reject_incomplete_method_set(self) -> None:
        probe = {
            "probe_id": "p",
            "query": "查詢",
            "expected_behavior": "retrieve",
            "expected_rule_ids": ["a"],
        }
        with self.assertRaisesRegex(RetrievalContractError, "完整方法"):
            evaluate_rankings(
                [probe],
                {"p": {"dense": [{"rule_id": "a", "rank": 1}]}},
                methods=["dense", "bm25"],
                cutoffs=[1],
            )


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

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

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

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

poetry run python -m json.tool configs/retrieval/day-19-retrieval-contract.json
poetry run python -m py_compile src/triage_rag/retrieval/__init__.py src/triage_rag/retrieval/core.py scripts/run_day19_retrieval_benchmark.py tests/test_retrieval.py
poetry run python -m unittest tests.test_retrieval

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