Day 19 讓稠密檢索與字詞檢索各自排出九筆公開知識，再用相互排序融合合成混合檢索（Hybrid Retrieval）排名。五筆容易問題在三條路徑都得到滿分，這不是完美結局，反而表示測試太容易，還無法看出第二階段排序是否有價值。

今天把問題改難。查詢會加入否定句、相近級數、相對順序、主要與次要考量對比，以及中英文混合表達。第一階段仍然負責「不要漏掉候選」，第二階段則重新逐一比較查詢與候選內容，決定哪一段應該排得更前面。

先看下方的概念示意。左側九張知識卡已經有初步順序；通過中間的語意檢視後，右側仍是完全相同的九張卡，只改變排列位置。

上圖表達兩個不可混淆的責任：第一階段檢索器建立候選集合；重排序器只在集合內調整名次。若第一階段完全漏掉必要規則，第二階段沒有辦法憑空把規則補回來。

本篇只把作者自寫查詢與九筆公開知識摘要送到線上服務，不會讀取或傳送病患列、護理師登錄級數、專家重新判定級數、診斷、處置或識別碼。結果是小型公開知識整合測試，不是病患分類或臨床安全評估。

今天會完成什麼

完成本篇後，你會得到五個可以實際檢查的產物：

1. 一份固定 Cohere 服務、模型、試用限制、隱私與失敗處理的模型鎖定檔。
2. 八筆可計分的困難公開知識問題，以及兩筆不能計分的知識缺口問題。
3. 一個不用額外 Cohere 套件、直接呼叫官方第二版端點的 Python 重排序用戶端。
4. 一份重排序前後的前 k 筆召回率、平均倒數排名、正規化折損累積增益與逐題名次。
5. 一個不保存金鑰、會節流、會驗證回應且可離線重讀的本機快取。

這些產物回答的是「線上重排序能否在固定九筆公開候選上可靠執行」與「這十題出現什麼排名變化」。它們不能回答模型對 1,267 筆病患的五級分類能力，也不能作為 Cohere 與其他供應商的普遍優劣比較。

本篇會用到的名詞

本篇的新名詞較多，先用表格建立共同語言。正文第一次使用時仍會補上用途與限制。

重排序器在 RAG 管線的哪一段

檢索增強生成（Retrieval-Augmented Generation, RAG）是一種先找外部證據，再讓大型語言模型根據證據產生回答的方法。今天仍沒有執行生成。

第一階段沿用三個已完成元件：稠密檢索（Dense Retrieval）用向量比較整體語意；最佳匹配 25（Best Matching 25, BM25）依字詞線索排序；相互排序融合（Reciprocal Rank Fusion, RRF）只融合兩條路徑的名次，建立 Hybrid 候選順序。今天的資料只走到：

使用者問題
  → Dense 與 BM25 建立完整排名
  → RRF 合成 Hybrid 排名
  → Cohere 重新排列相同九筆候選
  → 驗證候選集合與計算指標

本系列的公開知識主題是韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）。九筆候選只包含制度目的、流程、主要與次要考量，以及第一級到第五級的公開短摘要。部分知識庫沒有完整主訴目錄、各主訴數值門檻或逐版官方變更內容。

重排序器（Reranker）接收一個查詢與一組候選文件，對每個配對做較細緻的相關性判斷，再輸出新順序。它和生成模型有三個差別：

• 它不撰寫答案，只排列候選。
• 它不應新增或刪除文件，本篇要求回傳全部九筆。
• 它不能判斷知識庫是否完整；即使沒有答案，也會排出第一名。

因此本篇比較的不是「有重排序就比較聰明」，而是固定第一階段的九筆候選，只開關第二階段，觀察名次是否改變。這正是控制比較 D2：固定候選集合，唯一改變因素是重排序關閉或啟用。

為什麼選 Cohere rerank-v4.0-pro

Cohere 提供專門的 Rerank 線上服務。官方模型頁面目前把 rerank-v4.0-pro 定位為偏重品質、適合多語與較複雜使用情境的版本；rerank-v4.0-fast 則偏向較低延遲與較高吞吐量。本篇每題只有九筆候選，主要問題是繁體中文否定、相近概念與中英混合，不是大量即時請求，所以先鎖定品質導向的 rerank-v4.0-pro：Cohere Rerank 模型說明。

這個理由只是在執行前依任務需求選定一個方案，不是跑完分數後挑冠軍。本篇也不呼叫 rerank-v4.0-fast 或其他供應商，避免用同一組十題一邊比較、一邊選模型，再把最高結果誤稱為客觀測試。

程式呼叫第二版 Rerank 應用程式介面（Application Programming Interface, API）。這是一個採用表現狀態轉移（Representational State Transfer, REST）形式的網路介面，本機程式以固定網址與結構化資料送出請求：

POST https://api.cohere.com/v2/rerank

官方規格要求 model、query 與 documents，回應的 results 會依相關性排序，並提供原始文件索引 index 與相關分數 relevance_score：Cohere Rerank v2 API 參考。

線上服務不像本機 Ollama 模型能核對完整權重摘要。Cohere 沒有提供 rerank-v4.0-pro 權重的 digest，因此模型鎖定檔只能保存供應商、模型識別字、API 版本、查閱日期、每次請求指紋與安全雜湊演算法 256 位元（Secure Hash Algorithm 256-bit, SHA-256）。SHA-256 在這裡用來辨識固定輸入或回應內容，不是加密金鑰。即使模型名稱相同，仍不能保證服務端永遠完全不變；這是線上方案的重現性限制，不應藏起來。

試用金鑰的限制會改變程式設計

應用程式介面金鑰（Application Programming Interface Key, API Key）是授權請求的秘密字串。試用金鑰不是把字串貼進 Python 就結束，還要處理配額、資料政策與失敗行為。

Cohere 官方速率限制頁目前列出：試用金鑰的 Rerank 上限為每分鐘 10 次，所有端點合計每月 1,000 次 API 呼叫。因此本篇把相鄰網路請求的起始時間至少隔開 6.2 秒，並保存本機快取；正式數量與剩餘配額仍應以自己的 Cohere 控制台為準：Cohere Rate Limits。

快取（Cache）是把已驗證的線上回應存成本機 JSON。相同端點與 payload 會得到同一個請求指紋；下次執行先讀快取，只有缺少對應檔案才送網路請求。快取不保存 API Key 或 Authorization 標頭，也不提交 Git。

.env 只放本機，不能提交

本篇把金鑰放在專案根目錄的 .env：

COHERE_API_KEY=把自己的試用金鑰填在這裡

.env 是只存在自己電腦的環境變數檔；.env.example 則只保存變數名稱，可以公開。.gitignore 必須包含：

.env
.env.*
!.env.example

程式會先讀取作業系統已有的 COHERE_API_KEY，沒有時才讀根目錄 .env。任何錯誤訊息、快取、結果與文章都不會印出金鑰。若金鑰曾經出現在公開 commit、終端機截圖或文章內容，正確處理不是刪掉畫面而已，而是立即到供應商後台撤銷並換發。

試用環境不能傳病患資訊

Cohere 隱私政策指出，試用使用者的輸入與輸出可能用於研究與產品改進，且試用環境不適合處理個人資訊。因此即使 Kaggle 病患資料已公開，本篇仍完全不把病患列送到 Cohere：Cohere Privacy Policy。

下圖追蹤真正跨越網路邊界的欄位。送出前，程式先把本機候選縮小成公開 search_text 陣列；rule_id、人工評分、病患欄位與金鑰都留在本機。

上圖的回程只有候選索引與相關分數。本機程式再以送出前的索引位置還原 rule_id。這樣做同時達到兩個目的：遠端不需要知道內部識別字，回來後仍能核對同一組九筆候選是否完整。

十組問題怎麼建立

測試問題（Probe）是有預期行為的小型查詢。本篇十題全部由作者根據 Day 16 的公開知識與已標示缺口自行撰寫，不是從病患列改寫，也不使用 Kaggle 資料的護理師登錄欄位 KTAS_RN 或專家重新判定欄位 KTAS_expert。

八筆可計分問題故意比 Day 19 困難：

另外兩題問的是目前知識庫明確沒有的內容：所有 KTAS 主訴的完整數值門檻，以及完整官方版本編號、日期與逐版變更。它們只用來觀察重排序器在缺口上會排出什麼，不納入 Recall、MRR 或 nDCG 分母。

相關性不是只有零或一

Day 19 每題只有「相關」與「不相關」。今天加入相關等級（Relevance Grade）：1 代表有一些關聯，2 代表高度相關，3 代表直接回答檢索需求。實際八題的預期文件都標成 3；其他候選未列入時視為 0。

分級標記讓正規化折損累積增益能區分「直接答案排第一」與「只有部分相關內容排第一」。但標記仍由作者根據小型公開知識建立，不是多位獨立標註者的共識，因此結果要視為開發期整合證據。

三個指標各自回答什麼

Recall@k：必要文件有沒有進入前 k 筆

前 k 筆召回率（Recall at k, Recall@k）計算：

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

若一題有兩筆相關文件，第一名只容得下一筆，Recall@1 最高就是 $1/2=0.5$。這正是「同時找第一級與第二級」那題的合理結果；不能因為 0.5 就說檢索錯誤。

MRR：第一筆相關文件有多早出現

平均倒數排名（Mean Reciprocal Rank, MRR）先取每題第一筆相關文件名次的倒數，再對全部可計分問題平均：

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

第一名得到 1，第二名得到 0.5。MRR 很適合觀察第一筆證據，但不在意第二筆、第三筆相關文件，所以仍要搭配 Recall@k。

nDCG：較相關的文件是否排得更前面

正規化折損累積增益（Normalized Discounted Cumulative Gain, nDCG）先計算折損累積增益（Discounted Cumulative Gain, DCG）：

$$
\operatorname{DCG@k}=\sum_{i=1}^{k}\frac{2^{rel_i}-1}{\log_2(i+1)}
$$

• $i$ 是候選名次，第一名的 $i=1$。
• $rel_i$ 是第 $i$ 名文件的相關等級，範圍 0 到 3。
• $2^{rel_i}-1$ 讓高相關等級得到更大增益。
• $\log_2(i+1)$ 是名次折損；相關文件排得越後面，貢獻越小。

接著把實際 DCG 除以理想順序的 DCG：

$$
\operatorname{nDCG@k}=\frac{\operatorname{DCG@k}}{\operatorname{IDCG@k}}
$$

其中 IDCG 是 Ideal DCG，也就是先把相關等級由高到低排列後的最大可能分數。nDCG 越接近 1，代表實際順序越接近理想順序。

舉一個小型示意：兩筆相關文件的等級分別是 3 與 1。理想前三名把等級 3 放第一、等級 1 放第二：

$$
\operatorname{IDCG@3}=\frac{7}{\log_2 2}+\frac{1}{\log_2 3}
\approx7.6309
$$

若實際順序把等級 1 放第一、無關文件放第二、等級 3 放第三：

$$
\operatorname{DCG@3}=\frac{1}{\log_2 2}+\frac{7}{\log_2 4}=4.5
$$

所以 $\operatorname{nDCG@3}=4.5/7.6309\approx0.5897$。同樣兩筆文件都進入前三名，Recall@3 是 1.0；但 nDCG@3 會指出高相關文件排得不夠前面。

請求與回應必須通過哪些檢查

每題的 Hybrid 完整排名有九筆。程式依 Hybrid 名次建立 documents，但每筆只放公開 search_text。JavaScript 物件表示法（JavaScript Object Notation, JSON）是一種保存鍵值、陣列與數字的文字格式；送出的 JSON 形狀如下：

{
  "model": "rerank-v4.0-pro",
  "query": "作者自寫的公開知識問題",
  "documents": [
    "第一筆公開摘要",
    "第二筆公開摘要",
    "其餘七筆公開摘要"
  ],
  "top_n": 9,
  "max_tokens_per_doc": 512,
  "priority": 0
}

這裡的「其餘七筆」只是說明 JSON 形狀；正式程式會放入完整九個字串，不會真的送出省略文字。top_n=9 要求服務回傳全部候選，不讓線上服務代替本機刪除文件。

每次回應必須依序通過：

1. results 恰好有九筆。
2. 每個 index 都是 0 到 8 的整數，而且不可重複。
3. 九個索引全部出現，沒有新增或刪除候選。
4. relevance_score 是有限數字，且依回應順序遞減。
5. meta.api_version.version 必須是 2。
6. 本機還原後的九個 rule_id 集合與 Hybrid 完全相同。

相關分數（Relevance Score）只負責同一次 Cohere 回應內的排序。它不是機率、不是分類信心，也不能和 Hybrid 的 RRF 分數相減。今天比較的是名次與以名次計算的指標，不是兩種異質分數的絕對大小。

逾時、重試與失敗不能默默跳過

線上服務可能遇到網路中斷、逾時、速率限制或暫時無法使用。超文字傳輸通訊協定（Hypertext Transfer Protocol, HTTP）狀態碼用三位數表示請求結果，例如 429 代表請求過多，503 代表服務暫時無法使用，504 代表閘道逾時。本篇固定：

• 單次逾時 90 秒。
• 最多嘗試 2 次。
• 只對 HTTP 429、503、504 重試。
• Retry-After 最多接受 60 秒。
• 相鄰網路請求至少間隔 6.2 秒。
• 最終仍失敗時停止整次實驗，不把 Hybrid 原排序假裝成 Cohere 結果。

離線快取模式 --offline-cache-only 更嚴格：只允許讀既有快取，缺少任一請求就停止，完全不呼叫網路。這適合配額用完後重算指標或核對公開結果。

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

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

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 17 的九筆公開 flat chunks、Day 18 的 Ollama embedding 用戶端、Day 19 的 BM25／RRF，以及已下載的 qwen3-embedding:4b。請先自行申請 Cohere 試用 API Key；以下完整檔案會確保 .env 被 Git 忽略，正式請求只傳作者自寫問題與公開 search_text，不讀取或傳送 Kaggle 病患列與標籤。Cohere cache、完整排名與 run manifest 由程式自動產生，不需要手動建立。

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

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

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

檔案 1：建立 .gitignore

排除本機 .env、逐次執行產物、原始病患資料與其他不可提交檔案；保留可公開的 .env.example。

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

# 本地參考文件與受限制資料
*.pdf
data/raw/
data/interim/
data/processed/

# 本地執行產物
results/runs/
.venv/
poetry.toml
__pycache__/
*.pyc
*.egg-info/

# 本機 API 金鑰與環境變數
.env
.env.*
!.env.example

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

檔案 2：建立 .env.example

提供 API Key 變數名稱範本，不保存真實 Cohere 金鑰。

請在文字編輯器建立 .env.example，貼入以下完整內容並儲存：

# 請只在本機的 .env 填入真實值；不要把 API Key 提交到 Git。
COHERE_API_KEY=

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

檔案 3：建立 configs/models/day-20-cohere-reranker-lock.json

鎖定 Cohere Rerank API v2、rerank-v4.0-pro、試用限制、欄位最小化、快取、重試、隱私與版本限制。

請在文字編輯器建立 configs/models/day-20-cohere-reranker-lock.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "lock_id": "day-20-cohere-reranker-2026-08-10",
  "provider": "Cohere",
  "service": "Rerank API v2",
  "endpoint": "https://api.cohere.com/v2/rerank",
  "model": "rerank-v4.0-pro",
  "model_role": "online_multilingual_reranker_for_public_knowledge_candidates",
  "selection_reason": "Cohere 官方將 rerank-v4.0-pro 定位為適合多語與較複雜使用情境的品質導向版本；本系列候選少，不以高吞吐為主要目標。",
  "authentication": {
    "environment_variable": "COHERE_API_KEY",
    "local_env_file": ".env",
    "secret_value_committed": false
  },
  "request_contract": {
    "top_n": 9,
    "max_tokens_per_doc": 512,
    "priority": 0,
    "candidate_count": 9,
    "candidate_payload": "search_text_only_in_hybrid_rank_order",
    "candidate_ids_sent": false,
    "patient_records_sent": false,
    "reference_labels_sent": false
  },
  "trial_key_limits": {
    "rerank_requests_per_minute": 10,
    "all_api_calls_per_month": 1000,
    "usage_cost": "trial_usage_free_but_limited",
    "minimum_start_interval_seconds": 6.2
  },
  "failure_policy": {
    "timeout_seconds": 90,
    "maximum_attempts": 2,
    "retryable_status_codes": [429, 503, 504],
    "maximum_retry_after_seconds": 60,
    "cache_mode": "read_through_local_cache",
    "offline_cache_only_supported": true,
    "on_final_failure": "abort_without_claiming_reranked_result"
  },
  "privacy_policy": {
    "account_kind": "trial_key",
    "allowed_payload": "author_written_queries_and_public_knowledge_summaries_only",
    "forbidden_payload": "personal_information_patient_rows_clinical_labels_or_identifiers",
    "reason": "Cohere 的試用環境不適合處理個人資訊，且試用輸入與輸出可能用於研究與產品改進。"
  },
  "versioning": {
    "provider_model_digest_available": false,
    "lock_basis": "provider_model_id_api_version_accessed_date_and_response_hashes",
    "limitation": "供應商沒有提供可核對的模型權重 digest；相同 model ID 未必保證服務端永遠完全不變。"
  },
  "verification": {
    "smoke_tested_at": "2026-08-10",
    "http_status": 200,
    "response_api_version": "2",
    "observed_search_units": 1,
    "contained_patient_data": false
  },
  "official_sources": {
    "model_page": "https://docs.cohere.com/v2/docs/rerank",
    "api_reference": "https://docs.cohere.com/v2/reference/rerank",
    "rate_limits": "https://docs.cohere.com/v2/docs/rate-limits",
    "privacy_policy": "https://cohere.com/privacy",
    "accessed_at": "2026-08-10"
  }
}

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

檔案 4：建立 configs/retrieval/day-20-rerank-contract.json

鎖定九筆候選、八筆可計分難題、兩筆缺口、第一階段 Hybrid、相關等級、評估指標與輸出路徑。

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

{
  "schema_version": 1,
  "experiment_id": "day-20-cohere-rerank-public-knowledge-probes",
  "scope": "author_written_public_knowledge_reranking_integration_not_clinical_evaluation",
  "sources": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "embedding_config_path": "configs/representation/day-18-numeric-semantics.json",
    "embedding_config_sha256": "3d24a3688dbff4e617270f6fa9e953c055c835c4b0b10718eb6fa5a5ef3ada8b",
    "local_model_lock_path": "configs/models/local-ollama-model-lock.json",
    "local_model_lock_sha256": "0cd15cb4d7e735a19cbe439fe9a9344f717029694a4441764f67e396033cc7f4",
    "reranker_lock_path": "configs/models/day-20-cohere-reranker-lock.json",
    "reranker_lock_sha256": "e9a320c2f10aa2483df431173765b938e906df8ff3ab60a2cb63f28658c26ff8"
  },
  "input_contract": {
    "expected_document_count": 9,
    "expected_probe_count": 10,
    "expected_supported_probe_count": 8,
    "expected_gap_probe_count": 2,
    "supported_behavior": "retrieve",
    "gap_behavior": "disclose_gap",
    "relevance_grade_minimum": 1,
    "relevance_grade_maximum": 3,
    "forbidden_fields": [
      "record_index",
      "KTAS_RN",
      "KTAS_expert",
      "Group",
      "diagnosis",
      "disposition",
      "length_of_stay",
      "Error_group",
      "mistriage"
    ]
  },
  "probes": [
    {
      "probe_id": "hard-level-2-negation",
      "query": "不要找最高優先的立即處置，也不要找較低優先級；我要的是僅次於第一級、可能威脅生命或身體功能且需要快速治療的公開定義。",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-level-2-001": 3
      },
      "critical_expected_rule_ids": ["ktas-public-level-2-001"],
      "difficulty_tags": ["negation", "relative_priority", "near_level_distractors"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-level-4-with-level-5-distractor",
      "query": "不是慢性低風險的最低級，也不是立即處置；哪個公開級數會依年齡、疼痛、惡化或併發症可能性，安排一至兩小時內處置或重新評估？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-level-4-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["negation", "time_expression", "near_level_distractors"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-level-3-relative-order",
      "query": "哪個公開級數排在第一、二級之後，第四、五級之前，並描述病況可能進展到需要治療？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-level-3-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["relative_priority", "multiple_level_mentions"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-primary-not-secondary",
      "query": "不要回答血糖或脫水這類次要例子；公開研究列出的共通主要考量類型有哪些？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-primary-considerations-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["negation", "primary_secondary_contrast"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-secondary-not-primary",
      "query": "不是生命徵象、疼痛、出血或受傷機轉等主要類型；哪一段把血糖與脫水明確列為次要考量例子？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-secondary-considerations-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["negation", "primary_secondary_contrast"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-workflow-sequence",
      "query": "我不是在問單一級數。哪一段描述從第一印象與感染詢問開始，經主訴的主要、次要考量，最後判斷嚴重度與急迫度的流程？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-workflow-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["sequence", "exclude_level_definition"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-system-purpose-code-switch",
      "query": "Which passage defines KTAS as 症狀導向的 five-level tool for urgency and care priority，而不是某一級的個別定義？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-system-purpose-001": 3
      },
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["code_switching", "exclude_level_definition"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "hard-two-high-priority-definitions",
      "query": "若要同時找出最高優先、需要立即處置的定義，以及僅次於它、需要快速治療的定義，應取回哪兩段？",
      "expected_behavior": "retrieve",
      "expected_rule_grades": {
        "ktas-public-level-1-001": 3,
        "ktas-public-level-2-001": 3
      },
      "critical_expected_rule_ids": [
        "ktas-public-level-1-001",
        "ktas-public-level-2-001"
      ],
      "difficulty_tags": ["multiple_relevant_documents", "near_level_distractors"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "gap-complete-thresholds",
      "query": "請提供每一個 KTAS 主訴可直接套用的完整數值門檻。",
      "expected_behavior": "disclose_gap",
      "expected_gap_id": "complaint_specific_thresholds",
      "expected_rule_grades": {},
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["knowledge_gap"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "probe_id": "gap-official-version-history",
      "query": "請根據目前知識庫列出完整官方版本編號、發布日期與逐版變更內容。",
      "expected_behavior": "disclose_gap",
      "expected_gap_id": "official_version_history",
      "expected_rule_grades": {},
      "critical_expected_rule_ids": [],
      "difficulty_tags": ["knowledge_gap"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    }
  ],
  "first_stage": {
    "dense_model": "qwen3-embedding:4b",
    "bm25_k1": 1.2,
    "bm25_b": 0.75,
    "rrf_k": 60,
    "candidate_method": "hybrid_complete_ranking",
    "candidate_count": 9
  },
  "reranker": {
    "lock_path": "configs/models/day-20-cohere-reranker-lock.json",
    "provider": "Cohere",
    "model": "rerank-v4.0-pro",
    "endpoint": "https://api.cohere.com/v2/rerank",
    "api_key_environment_variable": "COHERE_API_KEY",
    "local_env_file": ".env",
    "top_n": 9,
    "max_tokens_per_doc": 512,
    "priority": 0,
    "same_candidate_set_required": true,
    "candidate_order_source": "hybrid",
    "remote_document_fields": ["search_text"],
    "timeout_seconds": 90,
    "maximum_attempts": 2,
    "retryable_status_codes": [429, 503, 504],
    "maximum_retry_after_seconds": 60,
    "minimum_start_interval_seconds": 6.2,
    "cache_root": "results/runs/day-20/_cohere-cache"
  },
  "evaluation": {
    "methods": ["hybrid_before", "cohere_after"],
    "cutoffs": [1, 3, 5],
    "supported_probe_metrics": [
      "mean_recall_at_k",
      "mean_reciprocal_rank",
      "mean_ndcg_at_k"
    ],
    "critical_rule_ids": [
      "ktas-public-level-1-001",
      "ktas-public-level-2-001"
    ],
    "critical_miss_cutoff": 3,
    "gap_probe_policy": "diagnostic_only_not_scored_without_abstention_threshold",
    "selection_policy": "integration_evidence_only_do_not_tune_or_compare_other_providers"
  },
  "outputs": {
    "public_summary_path": "results/public/day-20-cohere-rerank-benchmark.json",
    "run_output_root": "results/runs/day-20"
  }
}

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

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

公開 Day 19 檢索與 Day 20 Cohere 重排序、輸入驗證、快取及 nDCG 評估函式。

請在文字編輯器建立 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,
)
from .reranking import (
    CohereRerankClient,
    CohereRerankError,
    RerankContractError,
    build_cohere_payload,
    evaluate_reranking,
    load_api_key,
    ndcg_at_k,
    validate_cohere_response,
    validate_day20_inputs,
)


__all__ = [
    "BM25Index",
    "RetrievalContractError",
    "evaluate_rankings",
    "reciprocal_rank_fusion",
    "tokenize_for_bm25",
    "validate_retrieval_inputs",
    "CohereRerankClient",
    "CohereRerankError",
    "RerankContractError",
    "build_cohere_payload",
    "evaluate_reranking",
    "load_api_key",
    "ndcg_at_k",
    "validate_cohere_response",
    "validate_day20_inputs",
]

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

檔案 6：建立 src/triage_rag/retrieval/reranking.py

實作不洩漏 API Key 的 Cohere REST 用戶端、試用節流、可驗證快取、候選集合不變量、Recall／MRR／nDCG 與缺口排除。

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

"""Cohere Rerank client, cache, Day 20 contracts, and graded metrics."""

from __future__ import annotations

import json
import math
import os
import time
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Mapping, Sequence, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from triage_rag.reproducibility import canonical_json_bytes, load_json, write_json


JsonObject = Dict[str, Any]
Transport = Callable[[str, Mapping[str, str], Mapping[str, Any], int], Tuple[JsonObject, float]]
SUPPORTED_BEHAVIORS = {"retrieve", "disclose_gap"}


class RerankContractError(ValueError):
    """Raised when Day 20 input, provider output, or evaluation is invalid."""


class CohereRerankError(RuntimeError):
    """Raised for a sanitized Cohere network or HTTP failure."""

    def __init__(
        self,
        message: str,
        *,
        status_code: int | None = None,
        retry_after_seconds: float | None = None,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.retry_after_seconds = retry_after_seconds


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


def _duplicates(values: Iterable[str]) -> List[str]:
    seen: set[str] = set()
    duplicates: set[str] = set()
    for value in values:
        if value in seen:
            duplicates.add(value)
        seen.add(value)
    return sorted(duplicates)


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 load_api_key(
    environment_variable: str,
    env_file: Path,
) -> str:
    """Load one secret without logging it or overriding an existing environment value."""

    name = _nonempty_text(environment_variable, "environment_variable")
    existing = os.environ.get(name)
    if existing and existing.strip():
        return existing.strip()
    if not env_file.is_file():
        raise RerankContractError(
            f"找不到 {name}；請建立本機 {env_file.name}，不要把 API Key 寫進程式或 Git。"
        )
    for raw_line in env_file.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export ") :].lstrip()
        if "=" not in line:
            continue
        key, value = line.split("=", maxsplit=1)
        if key.strip() != name:
            continue
        secret = value.strip()
        if len(secret) >= 2 and secret[0] == secret[-1] and secret[0] in {"'", '"'}:
            secret = secret[1:-1]
        if secret:
            return secret
    raise RerankContractError(
        f"{env_file.name} 中的 {name} 未設定；請只在本機填入，不要提交秘密。"
    )


def validate_day20_inputs(
    documents: Sequence[Mapping[str, Any]],
    probes: Sequence[Mapping[str, Any]],
    config: Mapping[str, Any],
) -> JsonObject:
    contract = config["input_contract"]
    if not isinstance(documents, list) or len(documents) != int(
        contract["expected_document_count"]
    ):
        raise RerankContractError("文件數量與 Day 20 契約不同")
    if not isinstance(probes, list) or len(probes) != int(
        contract["expected_probe_count"]
    ):
        raise RerankContractError("probe 數量與 Day 20 契約不同")

    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 RerankContractError(f"Day 20 輸入出現禁止欄位：{present_forbidden}")

    rule_ids: List[str] = []
    for index, document in enumerate(documents):
        metadata = document.get("metadata")
        if not isinstance(metadata, Mapping):
            raise RerankContractError(f"documents[{index}].metadata 格式錯誤")
        rule_id = _nonempty_text(metadata.get("rule_id"), f"documents[{index}].rule_id")
        _nonempty_text(document.get("search_text"), f"{rule_id}.search_text")
        if document.get("retrieval_role") != "search_candidate":
            raise RerankContractError(f"{rule_id} 必須是 search_candidate")
        rule_ids.append(rule_id)
    duplicates = _duplicates(rule_ids)
    if duplicates:
        raise RerankContractError(f"文件 rule_id 不可重複：{duplicates}")

    supported_count = 0
    gap_count = 0
    probe_ids: List[str] = []
    known_rules = set(rule_ids)
    grade_minimum = int(contract["relevance_grade_minimum"])
    grade_maximum = int(contract["relevance_grade_maximum"])
    for probe in probes:
        probe_id = _nonempty_text(probe.get("probe_id"), "probe_id")
        probe_ids.append(probe_id)
        _nonempty_text(probe.get("query"), f"{probe_id}.query")
        if probe.get("origin") != "author_written_from_public_knowledge_not_patient_record":
            raise RerankContractError(f"{probe_id}.origin 不符合公開知識契約")
        if probe.get("uses_patient_record") is not False:
            raise RerankContractError(f"{probe_id} 不得使用病患紀錄")
        if probe.get("uses_reference_label") is not False:
            raise RerankContractError(f"{probe_id} 不得使用參考標籤")
        behavior = probe.get("expected_behavior")
        if behavior not in SUPPORTED_BEHAVIORS:
            raise RerankContractError(f"{probe_id}.expected_behavior 不支援")
        grades = probe.get("expected_rule_grades")
        if not isinstance(grades, Mapping):
            raise RerankContractError(f"{probe_id}.expected_rule_grades 格式錯誤")
        normalized_grades: Dict[str, int] = {}
        for raw_rule_id, raw_grade in grades.items():
            rule_id = _nonempty_text(raw_rule_id, f"{probe_id}.expected_rule_id")
            if (
                isinstance(raw_grade, bool)
                or not isinstance(raw_grade, int)
                or not grade_minimum <= raw_grade <= grade_maximum
            ):
                raise RerankContractError(f"{probe_id}.{rule_id} relevance grade 超出範圍")
            normalized_grades[rule_id] = raw_grade
        unknown = sorted(set(normalized_grades) - known_rules)
        if unknown:
            raise RerankContractError(f"{probe_id} 引用未知規則：{unknown}")

        critical = probe.get("critical_expected_rule_ids")
        if not isinstance(critical, list):
            raise RerankContractError(f"{probe_id}.critical_expected_rule_ids 格式錯誤")
        critical_ids = [
            _nonempty_text(item, f"{probe_id}.critical_expected_rule_id")
            for item in critical
        ]
        if _duplicates(critical_ids) or not set(critical_ids).issubset(normalized_grades):
            raise RerankContractError(f"{probe_id} 關鍵規則必須是不可重複的相關規則子集")

        if behavior == contract["supported_behavior"]:
            supported_count += 1
            if not normalized_grades or probe.get("expected_gap_id") is not None:
                raise RerankContractError(f"{probe_id} 支援案例標記不一致")
        else:
            gap_count += 1
            if normalized_grades or critical_ids:
                raise RerankContractError(f"{probe_id} 缺口案例不可標記相關規則")
            _nonempty_text(probe.get("expected_gap_id"), f"{probe_id}.expected_gap_id")

    duplicates = _duplicates(probe_ids)
    if duplicates:
        raise RerankContractError(f"probe_id 不可重複：{duplicates}")
    if supported_count != int(contract["expected_supported_probe_count"]):
        raise RerankContractError("支援 probe 數量與 Day 20 契約不同")
    if gap_count != int(contract["expected_gap_probe_count"]):
        raise RerankContractError("缺口 probe 數量與 Day 20 契約不同")
    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,
        "remote_payload_is_public_text_only": True,
    }


def build_cohere_payload(
    *,
    model: str,
    query: str,
    documents: Sequence[str],
    top_n: int,
    max_tokens_per_doc: int,
    priority: int,
) -> JsonObject:
    model = _nonempty_text(model, "model")
    query = _nonempty_text(query, "query")
    if not isinstance(documents, list) or not documents:
        raise RerankContractError("documents 必須是非空陣列")
    normalized_documents = [
        _nonempty_text(text, f"documents[{index}]")
        for index, text in enumerate(documents)
    ]
    if isinstance(top_n, bool) or not isinstance(top_n, int) or top_n != len(documents):
        raise RerankContractError("top_n 必須等於候選數，避免線上服務刪除候選")
    if (
        isinstance(max_tokens_per_doc, bool)
        or not isinstance(max_tokens_per_doc, int)
        or max_tokens_per_doc <= 0
    ):
        raise RerankContractError("max_tokens_per_doc 必須是正整數")
    if isinstance(priority, bool) or not isinstance(priority, int) or not 0 <= priority <= 999:
        raise RerankContractError("priority 必須介於 0 到 999")
    return {
        "model": model,
        "query": query,
        "documents": normalized_documents,
        "top_n": top_n,
        "max_tokens_per_doc": max_tokens_per_doc,
        "priority": priority,
    }


def _request_fingerprint(endpoint: str, payload: Mapping[str, Any]) -> str:
    request_contract = {
        "endpoint": _nonempty_text(endpoint, "endpoint"),
        "payload": dict(payload),
    }
    return sha256(canonical_json_bytes(request_contract)).hexdigest()


def _response_sha256(response: Mapping[str, Any]) -> str:
    return sha256(canonical_json_bytes(dict(response))).hexdigest()


def _safe_provider_message(body: bytes, fallback: str) -> str:
    try:
        parsed = json.loads(body.decode("utf-8"))
        message = parsed.get("message") if isinstance(parsed, Mapping) else None
        if isinstance(message, str) and message.strip():
            return message.strip()[:300]
    except (UnicodeDecodeError, json.JSONDecodeError):
        pass
    return fallback


def _default_transport(
    endpoint: str,
    headers: Mapping[str, str],
    payload: Mapping[str, Any],
    timeout_seconds: int,
) -> Tuple[JsonObject, float]:
    request = Request(
        endpoint,
        data=canonical_json_bytes(dict(payload)),
        headers=dict(headers),
        method="POST",
    )
    started = time.perf_counter()
    try:
        with urlopen(request, timeout=timeout_seconds) as response:
            body = response.read()
    except HTTPError as error:
        body = error.read()
        retry_after: float | None = None
        raw_retry_after = error.headers.get("Retry-After") if error.headers else None
        if raw_retry_after:
            try:
                retry_after = float(raw_retry_after)
            except ValueError:
                retry_after = None
        message = _safe_provider_message(body, f"Cohere HTTP {error.code}")
        raise CohereRerankError(
            f"Cohere Rerank 請求失敗：HTTP {error.code}，{message}",
            status_code=error.code,
            retry_after_seconds=retry_after,
        ) from error
    except URLError as error:
        raise CohereRerankError("無法連線 Cohere Rerank API；請檢查網路與 endpoint。") from error
    latency_ms = (time.perf_counter() - started) * 1000.0
    try:
        parsed = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise CohereRerankError("Cohere Rerank 回應不是合法 UTF-8 JSON。") from error
    if not isinstance(parsed, dict):
        raise CohereRerankError("Cohere Rerank 回應必須是 JSON 物件。")
    return parsed, latency_ms


def validate_cohere_response(
    response: Mapping[str, Any],
    candidate_rule_ids: Sequence[str],
) -> Tuple[List[JsonObject], JsonObject]:
    if not isinstance(response, Mapping):
        raise RerankContractError("Cohere response 必須是物件")
    candidate_ids = [
        _nonempty_text(rule_id, f"candidate_rule_ids[{index}]")
        for index, rule_id in enumerate(candidate_rule_ids)
    ]
    if not candidate_ids or _duplicates(candidate_ids):
        raise RerankContractError("candidate_rule_ids 必須非空且不可重複")
    results = response.get("results")
    if not isinstance(results, list) or len(results) != len(candidate_ids):
        raise RerankContractError("Cohere 必須回傳全部候選，不能新增或刪除")

    seen_indices: set[int] = set()
    ranking: List[JsonObject] = []
    previous_score = math.inf
    for rank, item in enumerate(results, start=1):
        if not isinstance(item, Mapping):
            raise RerankContractError("Cohere results 每一筆都必須是物件")
        index = item.get("index")
        score = item.get("relevance_score")
        if isinstance(index, bool) or not isinstance(index, int) or index not in range(len(candidate_ids)):
            raise RerankContractError("Cohere result index 超出候選範圍")
        if index in seen_indices:
            raise RerankContractError("Cohere result index 不可重複")
        if isinstance(score, bool) or not isinstance(score, (int, float)) or not math.isfinite(score):
            raise RerankContractError("Cohere relevance_score 必須是有限數字")
        score = float(score)
        if score > previous_score + 1e-12:
            raise RerankContractError("Cohere results 必須依 relevance_score 遞減")
        previous_score = score
        seen_indices.add(index)
        ranking.append(
            {
                "rule_id": candidate_ids[index],
                "rank": rank,
                "input_rank": index + 1,
                "input_index": index,
                "relevance_score": score,
            }
        )
    if seen_indices != set(range(len(candidate_ids))):
        raise RerankContractError("Cohere 回應沒有涵蓋完整候選索引")
    if {item["rule_id"] for item in ranking} != set(candidate_ids):
        raise RerankContractError("Cohere 重排序前後的候選集合不同")

    meta = response.get("meta")
    if not isinstance(meta, Mapping):
        raise RerankContractError("Cohere response 缺少 meta")
    api_version = meta.get("api_version")
    if not isinstance(api_version, Mapping) or str(api_version.get("version")) != "2":
        raise RerankContractError("Cohere API version 必須是 2")
    billed_units = meta.get("billed_units")
    search_units = None
    if isinstance(billed_units, Mapping):
        raw_search_units = billed_units.get("search_units")
        if isinstance(raw_search_units, int) and not isinstance(raw_search_units, bool):
            search_units = raw_search_units
    metadata = {
        "response_id": response.get("id"),
        "api_version": "2",
        "search_units": search_units,
    }
    return ranking, metadata


@dataclass(frozen=True)
class RerankOutcome:
    ranking: List[JsonObject]
    request_fingerprint: str
    response_sha256: str
    latency_ms: float
    fetched_at_utc: str
    cache_hit: bool
    response_metadata: JsonObject


class CohereRerankClient:
    """Minimal REST client that never persists the API key or authorization header."""

    def __init__(
        self,
        *,
        endpoint: str,
        model: str,
        api_key: str,
        top_n: int,
        max_tokens_per_doc: int,
        priority: int,
        timeout_seconds: int,
        maximum_attempts: int,
        retryable_status_codes: Sequence[int],
        maximum_retry_after_seconds: float,
        minimum_start_interval_seconds: float,
        cache_root: Path,
        transport: Transport | None = None,
    ) -> None:
        self.endpoint = _nonempty_text(endpoint, "endpoint")
        self.model = _nonempty_text(model, "model")
        self._api_key = _nonempty_text(api_key, "api_key")
        self.top_n = top_n
        self.max_tokens_per_doc = max_tokens_per_doc
        self.priority = priority
        if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, int) or timeout_seconds <= 0:
            raise RerankContractError("timeout_seconds 必須是正整數")
        if isinstance(maximum_attempts, bool) or not isinstance(maximum_attempts, int) or maximum_attempts <= 0:
            raise RerankContractError("maximum_attempts 必須是正整數")
        if maximum_retry_after_seconds < 0 or minimum_start_interval_seconds < 0:
            raise RerankContractError("重試與節流秒數不可小於 0")
        self.timeout_seconds = timeout_seconds
        self.maximum_attempts = maximum_attempts
        self.retryable_status_codes = set(retryable_status_codes)
        self.maximum_retry_after_seconds = float(maximum_retry_after_seconds)
        self.minimum_start_interval_seconds = float(minimum_start_interval_seconds)
        self.cache_root = cache_root
        self.transport = transport or _default_transport
        self._last_network_start: float | None = None
        self.network_calls = 0
        self.cache_hits = 0

    def _throttle(self) -> None:
        if self._last_network_start is None:
            return
        elapsed = time.monotonic() - self._last_network_start
        remaining = self.minimum_start_interval_seconds - elapsed
        if remaining > 0:
            time.sleep(remaining)

    def rerank(
        self,
        *,
        query: str,
        documents: Sequence[str],
        candidate_rule_ids: Sequence[str],
        offline_cache_only: bool = False,
    ) -> RerankOutcome:
        if len(documents) != len(candidate_rule_ids):
            raise RerankContractError("documents 與 candidate_rule_ids 數量不同")
        payload = build_cohere_payload(
            model=self.model,
            query=query,
            documents=list(documents),
            top_n=self.top_n,
            max_tokens_per_doc=self.max_tokens_per_doc,
            priority=self.priority,
        )
        fingerprint = _request_fingerprint(self.endpoint, payload)
        cache_path = self.cache_root / f"{fingerprint}.json"
        if cache_path.is_file():
            cached = load_json(cache_path)
            if cached.get("request_fingerprint") != fingerprint:
                raise RerankContractError("Cohere cache fingerprint 不一致")
            response = cached.get("response")
            ranking, metadata = validate_cohere_response(response, candidate_rule_ids)
            response_hash = _response_sha256(response)
            if response_hash != cached.get("response_sha256"):
                raise RerankContractError("Cohere cache response SHA-256 不一致")
            self.cache_hits += 1
            return RerankOutcome(
                ranking=ranking,
                request_fingerprint=fingerprint,
                response_sha256=response_hash,
                latency_ms=float(cached["latency_ms"]),
                fetched_at_utc=_nonempty_text(cached.get("fetched_at_utc"), "fetched_at_utc"),
                cache_hit=True,
                response_metadata=metadata,
            )
        if offline_cache_only:
            raise CohereRerankError(
                "offline-cache-only 模式找不到對應 Cohere cache；沒有送出網路請求。"
            )

        last_error: CohereRerankError | None = None
        for attempt in range(1, self.maximum_attempts + 1):
            self._throttle()
            self._last_network_start = time.monotonic()
            self.network_calls += 1
            try:
                response, latency_ms = self.transport(
                    self.endpoint,
                    {
                        "Authorization": f"Bearer {self._api_key}",
                        "Content-Type": "application/json",
                        "X-Client-Name": "ithome-2026-sideproject30-day20",
                    },
                    payload,
                    self.timeout_seconds,
                )
                ranking, metadata = validate_cohere_response(response, candidate_rule_ids)
                response_hash = _response_sha256(response)
                fetched_at_utc = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                write_json(
                    cache_path,
                    {
                        "cache_schema_version": 1,
                        "endpoint": self.endpoint,
                        "model": self.model,
                        "request_fingerprint": fingerprint,
                        "response_sha256": response_hash,
                        "latency_ms": round(float(latency_ms), 8),
                        "fetched_at_utc": fetched_at_utc,
                        "response": response,
                    },
                )
                return RerankOutcome(
                    ranking=ranking,
                    request_fingerprint=fingerprint,
                    response_sha256=response_hash,
                    latency_ms=float(latency_ms),
                    fetched_at_utc=fetched_at_utc,
                    cache_hit=False,
                    response_metadata=metadata,
                )
            except CohereRerankError as error:
                last_error = error
                retryable = error.status_code in self.retryable_status_codes
                if not retryable or attempt == self.maximum_attempts:
                    break
                retry_after = error.retry_after_seconds
                if retry_after is None:
                    retry_after = min(2.0**attempt, self.maximum_retry_after_seconds)
                retry_after = min(retry_after, self.maximum_retry_after_seconds)
                if retry_after > 0:
                    time.sleep(retry_after)
        assert last_error is not None
        raise CohereRerankError(
            f"Cohere Rerank 在 {self.maximum_attempts} 次內未成功；本次不產生重排序結果。",
            status_code=last_error.status_code,
        ) from last_error


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


def ndcg_at_k(
    ranking_rule_ids: Sequence[str],
    relevance_grades: Mapping[str, int],
    cutoff: int,
) -> float:
    if isinstance(cutoff, bool) or not isinstance(cutoff, int) or cutoff <= 0:
        raise RerankContractError("nDCG cutoff 必須是正整數")

    def dcg(grades: Sequence[int]) -> float:
        return sum(
            (2.0**grade - 1.0) / math.log2(rank + 1.0)
            for rank, grade in enumerate(grades, start=1)
        )

    observed = [int(relevance_grades.get(rule_id, 0)) for rule_id in ranking_rule_ids[:cutoff]]
    ideal = sorted((int(value) for value in relevance_grades.values()), reverse=True)[:cutoff]
    ideal_score = dcg(ideal)
    if ideal_score == 0:
        raise RerankContractError("沒有正相關文件時不可計算 nDCG")
    return dcg(observed) / ideal_score


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

    positive_results: List[JsonObject] = []
    gap_results: List[JsonObject] = []
    critical_pairs: List[Tuple[str, str]] = []
    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 RerankContractError(f"{probe_id} 缺少完整方法 ranking")
        rank_maps = {
            method: _rank_map(list(method_rankings[method]), method)
            for method in methods
        }
        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 RerankContractError(f"{probe_id} 重排序前後候選集合不同")

        if probe["expected_behavior"] == "retrieve":
            grades = {str(key): int(value) for key, value in probe["expected_rule_grades"].items()}
            relevant_ids = set(grades)
            for rule_id in probe["critical_expected_rule_ids"]:
                if rule_id not in critical_set:
                    raise RerankContractError(f"{probe_id} 使用未登錄的關鍵規則")
                critical_pairs.append((probe_id, rule_id))
            per_method = {}
            for method in methods:
                rank_map = rank_maps[method]
                if not relevant_ids.issubset(rank_map):
                    raise RerankContractError(f"{probe_id}.{method} 缺少相關規則")
                first_rank = min(rank_map[rule_id] for rule_id in relevant_ids)
                ordered_rule_ids = [
                    rule_id for rule_id, _ in sorted(rank_map.items(), key=lambda item: item[1])
                ]
                per_method[method] = {
                    "first_relevant_rank": first_rank,
                    "reciprocal_rank": 1.0 / first_rank,
                    "recall_at_k": {
                        str(cutoff): sum(rank_map[rule_id] <= cutoff for rule_id in relevant_ids)
                        / len(relevant_ids)
                        for cutoff in cutoffs
                    },
                    "ndcg_at_k": {
                        str(cutoff): ndcg_at_k(ordered_rule_ids, grades, cutoff)
                        for cutoff in cutoffs
                    },
                    "relevant_ranks": {
                        rule_id: rank_map[rule_id] for rule_id in sorted(relevant_ids)
                    },
                }
            positive_results.append(
                {
                    "probe_id": probe_id,
                    "query": probe["query"],
                    "difficulty_tags": probe["difficulty_tags"],
                    "expected_rule_grades": grades,
                    "critical_expected_rule_ids": list(probe["critical_expected_rule_ids"]),
                    "methods": per_method,
                }
            )
        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(rank_maps[method], key=rank_maps[method].get)
                        for method in methods
                    },
                }
            )

    if not positive_results:
        raise RerankContractError("至少需要一筆可計分 probe")
    aggregate: Dict[str, JsonObject] = {}
    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
            },
            "mean_ndcg_at_k": {
                str(cutoff): sum(
                    item["methods"][method]["ndcg_at_k"][str(cutoff)]
                    for item in positive_results
                )
                / len(positive_results)
                for cutoff in cutoffs
            },
        }

    critical_summary = {}
    for method in methods:
        missed = 0
        for probe_id, rule_id in critical_pairs:
            ranking = rankings_by_probe[probe_id][method]
            rank_map = _rank_map(list(ranking), method)
            if rank_map[rule_id] > critical_miss_cutoff:
                missed += 1
        critical_summary[method] = {
            "cutoff": critical_miss_cutoff,
            "expected_probe_rule_pair_count": len(critical_pairs),
            "missed_probe_rule_pair_count": missed,
            "miss_rate": missed / len(critical_pairs) if critical_pairs else None,
            "interpretation": "只檢查第一、二級公開定義是否進入候選，不是病患安全指標。",
        }
    return {
        "aggregate_supported_metrics": aggregate,
        "critical_public_definition_miss": critical_summary,
        "positive_probe_results": positive_results,
        "gap_probe_diagnostics": gap_results,
    }