iT邦幫忙

2026 iThome 鐵人賽

DAY 23
0
佛心分享-SideProject30

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

Day 23|生命徵象硬門控(Hard Gate):能排除過輕候選,也可能刪掉正確路徑

  • 分享至 

  • xImage
  •  

Day 22 把九筆公開知識整理成三個父層,完成 Parent Top-2 再 Child Top-3 的階層式檢索增強生成(Retrieval-Augmented Generation, RAG)。實驗同時暴露了一個不可逆問題:父層沒有入選時,正確子規則就不會進入後續排序。

今天把同一個問題搬到生命徵象。假設候選分數把第四級排在最前面,但合成血氧飽和度已觸發警示;如果在檢索前只允許第一、二級候選,原本過輕的分數第一名(Top-1)就會消失。反過來,若生命徵象沒有觸發作者設定的門檻,Hard Gate 也不會主動補入高風險主訴證據;若門檻本身太積極,還可能直接刪掉原本正確的候選。

先把最重要的限制放在開頭:本篇沒有取得可重建完整韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)的公開數值門檻。因此,接下來六條門檻、六筆案例、候選分數與合成參考級數,全是作者為測試程式行為設計的合成工程示例,不是 KTAS 正式規則、真實病患資料或臨床標籤。

下圖用概念示意呈現警示型硬門控的兩面性:觸發時縮窄候選,未觸發時也可能看不見主訴風險。圖中沒有正式門檻或真實病患資訊。

五張急迫度候選卡經生命徵象硬閘門後只有紅橘兩張通過,另一條平穩監測路徑旁仍留有高風險主訴線索,右側由臨床人員審閱證據

上圖的重點不是「生命徵象可以決定檢傷級數」,而是「硬排除會改變後續還看得到什麼」。系統最後仍只是研究與教學用決策支援元件,臨床人員必須保有完整情境與最終判斷權。


本篇目標

完成本篇後,你會得到四個可驗證產物:

  1. 一份明確標示為非臨床的合成生命徵象門控契約。
  2. 一個實作 candidate_level <= most_severe_level 的可測試 Hard Gate。
  3. 六筆配對比較無門控與 Hard Gate 的合成壓力案例。
  4. 一份同時保留修正、無效與不可逆排除的公開追蹤紀錄(Trace)。

今天不會訓練五級分類器,也不會輸出真實病患的 KTAS 級數。公開結果只回答:「在鎖定候選分數與六筆合成案例後,Hard Gate 如何改變候選集合與合成 Top-1 方向?」

本篇會用到的名詞

中文名稱 英文全名/縮寫 本篇用途
檢索增強生成 Retrieval-Augmented Generation, RAG 先找外部知識,再讓後續元件使用取回證據的架構
硬門控 Hard Gate 直接排除不符合條件的候選;被排除後不再參與排序
生命徵象 Vital Signs 本篇以收縮壓、心率、呼吸速率與血氧飽和度作為合成警示輸入
候選 Candidate 可能進入後續排序的公開級數定義知識單元
合成參考級數 Synthetic Reference Level 作者為工程測試指定的預期級數,不是護理師或專家標籤
檢傷不足 Undertriage 在五級方向中,Top-1 數字大於合成參考級數;本篇只描述合成案例方向
檢傷過度 Overtriage 在五級方向中,Top-1 數字小於合成參考級數;本篇只描述合成案例方向
JavaScript 物件表示法 JavaScript Object Notation, JSON 保存可由程式驗證的設定、合成案例與結果
Poetry Poetry 管理 Python 版本、相依套件與專案內執行命令

候選(Candidate)不是最後答案。它只是仍有資格進入後續排序的知識單元。本篇為了單獨觀察門控,把每個級數候選的合成分數固定;這些分數不是機率、模型信心或臨床風險。

為什麼不能只看生命徵象

KTAS 官方公開介紹把 KTAS 定位為症狀導向工具。公開流程不是只讀取生命徵象,而是先看第一印象,再依病患主訴套用共同主要考量與主訴特定次要考量,最後判斷嚴重度與急迫度。KTAS 官方介紹與五級分類標準

公開研究對 KTAS 演算法的描述也列出多種主要考量,包括生命徵象、疼痛分數、出血相關狀況與受傷機轉;血糖與脫水程度則是次要考量例子。這個來源能支持「生命徵象是考量之一」,不能支持「生命徵象單獨就能還原完整 KTAS」。Moon 等人:Triage accuracy and causes of mistriage using the Korean Triage and Acuity Scale

因此,生命徵象 Hard Gate 比較適合扮演單向警示:觀察到作者事先鎖定的警示時,排除明顯較不急的候選;沒有觸發時,不能反過來宣稱第一、二級不可能。

先分清楚五種資料角色

本篇刻意把容易混用的資料拆開:

資料角色 本篇是否使用 說明
病患輸入資料 不讀取 Kaggle 的 1,267 筆病患列
公開規則文件 只讀取 Day 17 的五筆 KTAS 公開級數定義
護理師登錄級數 KTAS_RN 不讀取,也不放進門控輸入
專家重新判定級數 KTAS_expert 不讀取,也不拿來調整門檻
作者合成參考級數 只作六筆工程案例的工程預期值

synthetic_reference_level 之所以不命名成 KTAS_expert,是為了避免把作者假設偽裝成臨床標籤。即使合成方向計算使用「檢傷不足」與「檢傷過度」,分母仍只有六筆作者案例,不能報成病患檢傷不足率或檢傷過度率。

Hard Gate 到底做了什麼

本系列固定採用第一級最急、第五級最不急的數字方向。令候選級數為 d,所有級數集合為:

D = {1, 2, 3, 4, 5}

當一條或多條合成警示觸發時,每條警示會給出一個候選級數上限。若觸發上限集合是 A,本篇取數字最小、也就是最急迫的上限:

most_severe_level = min(A)

最後保留的候選集合是:

D_hard = {d ∈ D | d <= most_severe_level}

逐項解釋如下:

  • d 是某一筆公開級數定義的級數。
  • A 是本案例實際觸發的合成警示上限集合。
  • most_severe_level 是數字最小的上限。
  • d <= most_severe_level 代表只保留同樣或更急迫的級數。

例如,一條警示給出第二級上限時,most_severe_level = 2,保留集合是 {1, 2},第三至第五級直接消失。若另一條警示同時給出第一級上限,min({2, 1}) = 1,最後只保留 {1}

這裡最容易犯的錯是把不等號寫反。因為第一級最急迫,d >= 2 會保留第二至第五級,正好留下較不急候選;本篇測試會把這種方向漂移直接擋下。

六條門檻為什麼只能叫「合成工程示例」

Day 18 已把正式門檻政策鎖成 unavailable_in_current_public_knowledge_base,代表目前公開知識庫不足以還原逐主訴、逐年齡與完整數值門檻。Day 23 不會偷偷推翻這個缺口,而是另開一個 counterfactual_engineering_test_only 政策,專門測試程式邊界。

本篇鎖定的六條作者合成規則如下:

合成規則 候選級數上限 工程測試用途
Mental = 4 1 驗證單一第一級上限
Mental ∈ {2, 3} 2 驗證集合比對
SBP < 90 mmHg 2 驗證嚴格小於
HR >= 130 次/分 2 驗證包含等號的上界
RR >= 30 次/分 2 驗證另一個包含等號門檻
SpO2 < 92% 2 驗證血氧飽和度警示與缺失值

收縮壓(Systolic Blood Pressure, SBP)、心率(Heart Rate, HR)、呼吸速率(Respiratory Rate, RR)與周邊血氧飽和度(Peripheral Oxygen Saturation, SpO2)都來自 Day 14 允許在初次檢傷決策前使用的欄位。表中的數字只用來驗證 <>=、多警示合併與候選排除,不能拿來判斷任何人的 KTAS 級數。

Mental 是資料集的四級意識反應代碼:1 代表清醒、2 代表對聲音有反應、3 代表對疼痛有反應、4 代表無反應。它不是格拉斯哥昏迷指數(Glasgow Coma Scale, GCS)。本篇設定明確寫入 gcs_policy,程式也拒絕把不存在的 GCS 欄位偷換成 Mental

下圖由左到右呈現輸入、合成警示、數字最小的級數上限與候選篩選。請特別觀察:無警示不是「正常證明」,缺失也不是零或正常值。

Mental、SBP、HR、RR 與 SpO2 輸入先比對六條合成規則,多條警示取數字最小的級數上限,再只保留級數小於等於上限的公開定義候選;無警示或缺失時不縮窄候選

上圖把「警示計算」和「候選排序」分開。Hard Gate 只改變誰還能參賽;保留下來的候選仍使用同一組合成分數排序。這樣比較時唯一改變的是門控,不會同時混入不同向量嵌入(Embedding)、重排序器或生成模型。

缺失值不是正常值

Day 15 的缺失政策要求:生命徵象缺失時要保留「未記錄」狀態,不得補成零,也不得改寫成正常。本篇把這項政策延伸到 Hard Gate:

  1. 某欄位缺失時,不評估依賴該欄位的門檻。
  2. 若其他已觀察欄位仍觸發警示,照樣套用那條警示,同時留下缺失欄位追蹤紀錄。
  3. 若沒有任何已觀察欄位觸發、但至少一欄缺失,狀態是 indeterminate_missing_values
  4. 不確定狀態不縮窄五級候選,避免把缺失誤當成排除依據。

這不是說「全保留就安全」。它只表示硬門控不能從不存在的數值推論風險不存在。Day 24 還會加入主訴與高風險保護候選,處理單靠生命徵象無法補救的路徑。

配對實驗如何只改一個變因

六筆案例都包含完全相同的五筆公開級數定義候選。每筆案例另外鎖定五個不同的合成候選分數;無門控與 Hard Gate 共用同一組分數。

兩條路徑只有一個差別:

比較項目 無門控 Hard Gate
公開候選來源 五筆級數定義 同一組五筆級數定義
合成候選分數 固定 完全相同
生命徵象警示 不套用 套用六條合成規則
候選排除 有警示時保留 level <= ceiling
Top-1 方向 對合成參考級數比較 對同一合成參考級數比較
真實病患/臨床標籤 不使用 不使用

這個微型實驗不執行大型語言模型(Large Language Model, LLM)生成。原因是今天要單獨測試候選集合;若同時加入自由文字生成,輸出差異就可能來自生成隨機性,而不是 Hard Gate。

六筆合成壓力案例

六筆案例不是為了模擬真實盛行率,而是刻意覆蓋六種程式邊界:

  1. SpO2 警示把過輕的第四級候選排除。
  2. Mental = 4 只保留第一級。
  3. 五條警示同時觸發,驗證數字最小上限優先。
  4. 高風險主訴案例未觸發生命徵象門檻,驗證錯誤不會自動被救回。
  5. SpO2 缺失,驗證缺失不會被當成正常。
  6. HR 剛好等於合成門檻,驗證 >= 的等號與正確候選遭硬排除。

其中第四筆雖然在案例敘述中寫「高風險主訴」,Hard Gate 本身不讀取主訴。這不是資料遺漏,而是刻意建立 Day 24 要解決的反例:只用生命徵象的門控看不到主訴保護訊號。

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

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

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 14 的輸入欄位契約、Day 15 的缺失值政策、Day 17 的五筆公開級數定義,以及 Day 18 已確認的正式數值門檻知識缺口。以下六筆案例、候選分數與 synthetic_reference_level 都是作者為工程壓力測試撰寫的合成資料,不讀取 Kaggle 病患列、護理師標籤或專家標籤。以下是 Day 23 新增或修改後的完整設定、合成案例、RAG 套件入口、Hard Gate 核心、執行入口與測試;公開摘要、完整追蹤紀錄與執行清單(Run Manifest)都由執行入口自動產生,不需要手動建立。

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

mkdir -p configs/rag tests/fixtures src/triage_rag/rag scripts tests results/public results/runs/day-23

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

檔案 1:建立 configs/rag/day-23-vital-hard-gate.json

鎖定五級方向、六條作者合成門檻、缺失值與多警示政策、上游雜湊、六筆案例數、輸出與禁止臨床使用的邊界。

請在文字編輯器建立 configs/rag/day-23-vital-hard-gate.json,貼入以下完整內容並儲存:

{
  "schema_version": 1,
  "experiment_id": "day-23-synthetic-vital-hard-gate",
  "scope": "author_written_synthetic_hard_gate_microbenchmark_not_patient_triage_or_clinical_evaluation",
  "sources": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "day14_data_contract_path": "configs/data/day-14-ktas-data-contract.json",
    "day14_data_contract_sha256": "2a7d0fc31782ae0b9e90c1a98eadd722e4236c264bc0ee6ee7f4f6b0f98c4e86",
    "day15_quality_contract_path": "configs/data/day-15-quality-and-split-contract.json",
    "day15_quality_contract_sha256": "afef6b89c9ae9a048326271c2ad74649ae0c029f24bbd839581aa46129498cf5",
    "day18_numeric_contract_path": "configs/representation/day-18-numeric-semantics.json",
    "day18_numeric_contract_sha256": "3d24a3688dbff4e617270f6fa9e953c055c835c4b0b10718eb6fa5a5ef3ada8b",
    "synthetic_cases_path": "tests/fixtures/day-23-synthetic-gate-cases.json",
    "synthetic_cases_sha256": "f40c9e3dd9a1679c58e6664e94b90bc215adc882537d39d3a086746a3e77179e"
  },
  "level_contract": {
    "valid_levels": [1, 2, 3, 4, 5],
    "direction": "smaller_number_is_more_urgent",
    "hard_filter_expression": "candidate_level <= most_severe_level",
    "candidate_source": "the_five_public_level_definition_chunks_only",
    "shared_score_policy": "the_same_author_written_synthetic_candidate_scores_are_used_before_and_after_gating"
  },
  "synthetic_threshold_policy": {
    "policy_role": "counterfactual_engineering_test_only",
    "authority": "author_written_not_an_official_ktas_threshold_table",
    "clinical_use": "prohibited",
    "age_scope": "not_age_adjusted_and_therefore_not_clinically_interpretable",
    "missing_value_policy": "never_impute_and_never_treat_missing_as_normal",
    "no_alarm_policy": "pass_through_all_five_level_candidates",
    "multiple_alarm_policy": "minimum_candidate_ceiling_wins",
    "gcs_policy": "not_used_because_the_project_input_contract_has_Mental_not_GCS",
    "rules": [
      {
        "rule_id": "synthetic-mental-unresponsive",
        "field": "Mental",
        "operator": "in",
        "values": [4],
        "most_severe_level": 1,
        "display": "Mental = 4"
      },
      {
        "rule_id": "synthetic-mental-responsive-to-stimulus",
        "field": "Mental",
        "operator": "in",
        "values": [2, 3],
        "most_severe_level": 2,
        "display": "Mental ∈ {2, 3}"
      },
      {
        "rule_id": "synthetic-sbp-lt-90",
        "field": "SBP",
        "operator": "lt",
        "threshold": 90,
        "most_severe_level": 2,
        "display": "SBP < 90 mmHg"
      },
      {
        "rule_id": "synthetic-hr-gte-130",
        "field": "HR",
        "operator": "gte",
        "threshold": 130,
        "most_severe_level": 2,
        "display": "HR ≥ 130 次/分"
      },
      {
        "rule_id": "synthetic-rr-gte-30",
        "field": "RR",
        "operator": "gte",
        "threshold": 30,
        "most_severe_level": 2,
        "display": "RR ≥ 30 次/分"
      },
      {
        "rule_id": "synthetic-saturation-lt-92",
        "field": "Saturation",
        "operator": "lt",
        "threshold": 92,
        "most_severe_level": 2,
        "display": "SpO2 < 92%"
      }
    ]
  },
  "evaluation": {
    "expected_case_count": 6,
    "expected_alarm_case_count": 4,
    "expected_no_alarm_case_count": 1,
    "expected_indeterminate_case_count": 1,
    "top_k_trace": 3,
    "direction_definition": {
      "exact": "candidate_top1_equals_synthetic_reference_level",
      "undertriage": "candidate_top1_is_numerically_greater_than_synthetic_reference_level",
      "overtriage": "candidate_top1_is_numerically_smaller_than_synthetic_reference_level"
    },
    "claim_boundary": "direction_counts_describe_only_six_synthetic_contract_cases_and_are_not_accuracy_or_safety_estimates"
  },
  "outputs": {
    "public_summary_path": "results/public/day-23-vital-hard-gate.json",
    "run_output_root": "results/runs/day-23",
    "result_filename": "vital-hard-gate-results.json"
  },
  "limitations": [
    "六筆案例、候選分數與 synthetic_reference_level 全由作者為工程測試撰寫,不是真實病患、護理師標籤或專家標籤。",
    "數值門檻只用來驗證程式邊界、缺失值與不可逆排除,不是 KTAS 官方規則,也不能用於臨床判斷。",
    "公開知識庫只有五級定義與部分流程摘要,沒有完整主訴目錄、年齡分層、正式數值門檻或套用順序。",
    "候選分數是固定的合成排序訊號,不是稠密檢索、混合檢索、重排序器、生成模型或臨床風險分數。",
    "本篇只檢查候選集合與合成 Top-1 方向,不宣稱分類準確率、召回率、檢傷不足率或臨床安全性。",
    "系統只能作為研究與教學用決策支援元件,不能取代護理師、醫師或正式檢傷流程。"
  ]
}

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

檔案 2:建立 tests/fixtures/day-23-synthetic-gate-cases.json

保存六筆作者合成壓力案例、五級共同候選分數、合成參考級數與逐筆鎖定預期;不含真實病患或臨床標籤。

請在文字編輯器建立 tests/fixtures/day-23-synthetic-gate-cases.json,貼入以下完整內容並儲存:

{
  "schema_version": 1,
  "scope": "author_written_synthetic_gate_stress_cases_not_patient_records_or_clinical_labels",
  "score_semantics": "每級分數是為了固定共同候選排序的合成檢索分數;不是模型信心、機率或臨床風險。",
  "reference_semantics": "synthetic_reference_level 是作者為工程測試指定的預期級數,只用來檢查方向與候選是否遭排除;不是護理師標籤、專家標籤或 KTAS 判定。",
  "cases": [
    {
      "case_id": "alarm-low-saturation-rescue",
      "scenario": "合成低血氧警示,未門控時過輕候選分數最高。",
      "inputs": {
        "Mental": 1,
        "SBP": 104,
        "HR": 112,
        "RR": 28,
        "Saturation": 89
      },
      "candidate_scores": {
        "1": 0.82,
        "2": 0.91,
        "3": 0.71,
        "4": 0.96,
        "5": 0.55
      },
      "synthetic_reference_level": 2,
      "expected": {
        "route_status": "hard_gate_applied",
        "triggered_rule_ids": [
          "synthetic-saturation-lt-92"
        ],
        "most_severe_level": 2,
        "retained_levels": [1, 2],
        "baseline_top1": 4,
        "gated_top1": 2,
        "reference_retained": true,
        "outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "alarm-unresponsive-rescue",
      "scenario": "合成無反應警示要求只保留第一級候選。",
      "inputs": {
        "Mental": 4,
        "SBP": 118,
        "HR": 96,
        "RR": 20,
        "Saturation": 97
      },
      "candidate_scores": {
        "1": 0.9,
        "2": 0.76,
        "3": 0.95,
        "4": 0.62,
        "5": 0.51
      },
      "synthetic_reference_level": 1,
      "expected": {
        "route_status": "hard_gate_applied",
        "triggered_rule_ids": [
          "synthetic-mental-unresponsive"
        ],
        "most_severe_level": 1,
        "retained_levels": [1],
        "baseline_top1": 3,
        "gated_top1": 1,
        "reference_retained": true,
        "outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "multiple-alarms-most-severe-wins",
      "scenario": "兩條合成警示同時觸發,應採用數字最小的第一級上限。",
      "inputs": {
        "Mental": 4,
        "SBP": 86,
        "HR": 136,
        "RR": 32,
        "Saturation": 90
      },
      "candidate_scores": {
        "1": 0.92,
        "2": 0.97,
        "3": 0.75,
        "4": 0.64,
        "5": 0.5
      },
      "synthetic_reference_level": 1,
      "expected": {
        "route_status": "hard_gate_applied",
        "triggered_rule_ids": [
          "synthetic-mental-unresponsive",
          "synthetic-sbp-lt-90",
          "synthetic-hr-gte-130",
          "synthetic-rr-gte-30",
          "synthetic-saturation-lt-92"
        ],
        "most_severe_level": 1,
        "retained_levels": [1],
        "baseline_top1": 2,
        "gated_top1": 1,
        "reference_retained": true,
        "outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "no-alarm-high-risk-complaint",
      "scenario": "合成高風險主訴,但五個輸入值都未觸發合成生命徵象規則。",
      "inputs": {
        "Mental": 1,
        "SBP": 122,
        "HR": 88,
        "RR": 18,
        "Saturation": 98
      },
      "candidate_scores": {
        "1": 0.7,
        "2": 0.91,
        "3": 0.96,
        "4": 0.66,
        "5": 0.52
      },
      "synthetic_reference_level": 2,
      "expected": {
        "route_status": "pass_through_no_alarm",
        "triggered_rule_ids": [],
        "most_severe_level": null,
        "retained_levels": [1, 2, 3, 4, 5],
        "baseline_top1": 3,
        "gated_top1": 3,
        "reference_retained": true,
        "outcome": "persistent_undertriage"
      }
    },
    {
      "case_id": "missing-saturation-is-indeterminate",
      "scenario": "血氧飽和度缺失,不能把缺失值解讀成正常,也不能據此縮窄候選。",
      "inputs": {
        "Mental": 1,
        "SBP": 116,
        "HR": 92,
        "RR": 20,
        "Saturation": null
      },
      "candidate_scores": {
        "1": 0.68,
        "2": 0.89,
        "3": 0.76,
        "4": 0.95,
        "5": 0.57
      },
      "synthetic_reference_level": 2,
      "expected": {
        "route_status": "indeterminate_missing_values",
        "triggered_rule_ids": [],
        "most_severe_level": null,
        "retained_levels": [1, 2, 3, 4, 5],
        "baseline_top1": 4,
        "gated_top1": 4,
        "reference_retained": true,
        "outcome": "persistent_undertriage"
      }
    },
    {
      "case_id": "boundary-alarm-removes-reference",
      "scenario": "合成心率剛好觸發作者門檻,但合成參考級數為第三級,用來暴露硬排除的不可逆風險。",
      "inputs": {
        "Mental": 1,
        "SBP": 110,
        "HR": 130,
        "RR": 22,
        "Saturation": 96
      },
      "candidate_scores": {
        "1": 0.72,
        "2": 0.9,
        "3": 0.96,
        "4": 0.67,
        "5": 0.53
      },
      "synthetic_reference_level": 3,
      "expected": {
        "route_status": "hard_gate_applied",
        "triggered_rule_ids": [
          "synthetic-hr-gte-130"
        ],
        "most_severe_level": 2,
        "retained_levels": [1, 2],
        "baseline_top1": 3,
        "gated_top1": 2,
        "reference_retained": false,
        "outcome": "reference_removed_overtriage"
      }
    }
  ]
}

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

檔案 3:建立 src/triage_rag/rag/__init__.py

更新 RAG 子套件入口,保留 Day 21、22 公開函式並加入 Day 23 Hard Gate 契約、警示、篩選、逐筆 trace 與聚合函式。

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

"""Retrieval-augmented generation pipelines."""

from triage_rag.rag.flat import (
    FlatRagContractError,
    build_context,
    build_user_message,
    evaluate_answer,
    validate_flat_contract,
    validate_public_cases,
)
from triage_rag.rag.hierarchical import (
    HierarchicalRagContractError,
    build_hierarchical_context,
    build_hierarchical_user_message,
    expand_selected_children,
    expected_parent_ids,
    group_context_for_generation,
    rank_expanded_children,
    rank_parents,
    validate_hierarchical_contract,
)
from triage_rag.rag.vital_gate import (
    VitalHardGateContractError,
    direction_against_synthetic_reference,
    evaluate_vital_gate,
    rank_level_candidates,
    run_vital_hard_gate_case,
    select_level_candidates,
    summarize_vital_hard_gate_results,
    validate_vital_hard_gate_contract,
)

__all__ = [
    "FlatRagContractError",
    "build_context",
    "build_user_message",
    "evaluate_answer",
    "validate_flat_contract",
    "validate_public_cases",
    "HierarchicalRagContractError",
    "build_hierarchical_context",
    "build_hierarchical_user_message",
    "expand_selected_children",
    "expected_parent_ids",
    "group_context_for_generation",
    "rank_expanded_children",
    "rank_parents",
    "validate_hierarchical_contract",
    "VitalHardGateContractError",
    "direction_against_synthetic_reference",
    "evaluate_vital_gate",
    "rank_level_candidates",
    "run_vital_hard_gate_case",
    "select_level_candidates",
    "summarize_vital_hard_gate_results",
    "validate_vital_hard_gate_contract",
]

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

檔案 4:建立 src/triage_rag/rag/vital_gate.py

驗證合成範圍、五級公開候選與禁止欄位,實作缺失感知警示、最嚴重級數合併、候選硬排除、方向比較與描述性聚合。

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

"""Deterministic vital-sign hard gate for the synthetic Day 23 microbenchmark."""

from __future__ import annotations

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


JsonObject = Dict[str, Any]
VALID_LEVELS = (1, 2, 3, 4, 5)
VALID_OPERATORS = {"lt", "lte", "gt", "gte", "eq", "in"}
FORBIDDEN_CASE_FIELDS = {
    "KTAS_RN",
    "KTAS_expert",
    "error_group",
    "Length of stay_min",
    "Disposition",
    "Diagnosis in ED",
    "result",
}
EXPECTED_SCOPE = (
    "author_written_synthetic_hard_gate_microbenchmark_"
    "not_patient_triage_or_clinical_evaluation"
)


class VitalHardGateContractError(ValueError):
    """Raised when the Day 23 safety or comparison contract drifts."""


def _is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _assert_no_forbidden_fields(value: Any, *, location: str = "case") -> None:
    if isinstance(value, Mapping):
        overlap = sorted(FORBIDDEN_CASE_FIELDS.intersection(value))
        if overlap:
            raise VitalHardGateContractError(
                f"{location} 含禁止欄位:{overlap}"
            )
        for key, nested in value.items():
            _assert_no_forbidden_fields(nested, location=f"{location}.{key}")
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            _assert_no_forbidden_fields(nested, location=f"{location}[{index}]")


def select_level_candidates(documents: Sequence[Mapping[str, Any]]) -> List[JsonObject]:
    """Select exactly one public level-definition chunk for each level 1–5."""

    candidates: List[JsonObject] = []
    for document in documents:
        metadata = document.get("metadata")
        if not isinstance(metadata, Mapping):
            continue
        if metadata.get("topic") != "level_definition":
            continue
        levels = metadata.get("ktas_levels")
        if not isinstance(levels, list) or len(levels) != 1:
            raise VitalHardGateContractError(
                "level_definition chunk 必須只對應一個級數"
            )
        level = levels[0]
        if level not in VALID_LEVELS:
            raise VitalHardGateContractError(f"非法級數:{level}")
        candidates.append(
            {
                "level": int(level),
                "rule_id": metadata.get("rule_id"),
                "search_text": document.get("search_text"),
                "source_url": metadata.get("source_url"),
            }
        )

    counts = Counter(item["level"] for item in candidates)
    if counts != Counter({level: 1 for level in VALID_LEVELS}):
        raise VitalHardGateContractError(
            f"五級公開定義候選必須各一筆,實際為 {dict(sorted(counts.items()))}"
        )
    for candidate in candidates:
        if not isinstance(candidate["rule_id"], str) or not candidate["rule_id"]:
            raise VitalHardGateContractError("level candidate 缺少 rule_id")
        if not isinstance(candidate["search_text"], str) or not candidate["search_text"].strip():
            raise VitalHardGateContractError("level candidate 缺少 search_text")
        if not isinstance(candidate["source_url"], str) or not candidate["source_url"]:
            raise VitalHardGateContractError("level candidate 缺少 source_url")
    return sorted(candidates, key=lambda item: item["level"])


def validate_vital_hard_gate_contract(
    config: Mapping[str, Any],
    fixture: Mapping[str, Any],
    candidates: Sequence[Mapping[str, Any]],
    day14_contract: Mapping[str, Any],
    day15_contract: Mapping[str, Any],
) -> JsonObject:
    """Validate the synthetic-only gate, upstream fields, cases, and candidates."""

    if config.get("scope") != EXPECTED_SCOPE:
        raise VitalHardGateContractError("Day 23 scope 不允許改成病患或臨床評估")
    level_contract = config.get("level_contract")
    if not isinstance(level_contract, Mapping):
        raise VitalHardGateContractError("缺少 level_contract")
    required_level_values = {
        "valid_levels": list(VALID_LEVELS),
        "direction": "smaller_number_is_more_urgent",
        "hard_filter_expression": "candidate_level <= most_severe_level",
        "candidate_source": "the_five_public_level_definition_chunks_only",
        "shared_score_policy": (
            "the_same_author_written_synthetic_candidate_scores_are_used_"
            "before_and_after_gating"
        ),
    }
    drift = {
        key: {"actual": level_contract.get(key), "expected": expected}
        for key, expected in required_level_values.items()
        if level_contract.get(key) != expected
    }
    if drift:
        raise VitalHardGateContractError(f"五級方向或公平比較契約漂移:{drift}")

    policy = config.get("synthetic_threshold_policy")
    if not isinstance(policy, Mapping):
        raise VitalHardGateContractError("缺少 synthetic_threshold_policy")
    required_policy = {
        "policy_role": "counterfactual_engineering_test_only",
        "authority": "author_written_not_an_official_ktas_threshold_table",
        "clinical_use": "prohibited",
        "age_scope": "not_age_adjusted_and_therefore_not_clinically_interpretable",
        "missing_value_policy": "never_impute_and_never_treat_missing_as_normal",
        "no_alarm_policy": "pass_through_all_five_level_candidates",
        "multiple_alarm_policy": "minimum_candidate_ceiling_wins",
        "gcs_policy": "not_used_because_the_project_input_contract_has_Mental_not_GCS",
    }
    policy_drift = {
        key: {"actual": policy.get(key), "expected": expected}
        for key, expected in required_policy.items()
        if policy.get(key) != expected
    }
    if policy_drift:
        raise VitalHardGateContractError(f"合成門檻安全政策漂移:{policy_drift}")

    numeric_fields = set(day14_contract["input_constraints"]["numeric_fields"])
    vital_fields = set(day15_contract["missingness_policy"]["vital_sign_fields"])
    allowed_gate_fields = vital_fields | {"Mental"}
    if day15_contract["missingness_policy"].get("rag_representation") != (
        "保留缺失狀態,明確寫成未記錄或不適用,不得轉寫為正常。"
    ):
        raise VitalHardGateContractError("Day 15 缺失值政策已漂移")

    rules = policy.get("rules")
    if not isinstance(rules, list) or not rules:
        raise VitalHardGateContractError("合成門檻至少需要一條規則")
    rule_ids: set[str] = set()
    rule_fields: set[str] = set()
    for rule in rules:
        if not isinstance(rule, Mapping):
            raise VitalHardGateContractError("每條門檻規則都必須是 JSON 物件")
        rule_id = rule.get("rule_id")
        field = rule.get("field")
        operator = rule.get("operator")
        level = rule.get("most_severe_level")
        if not isinstance(rule_id, str) or not rule_id:
            raise VitalHardGateContractError("門檻規則缺少 rule_id")
        if rule_id in rule_ids:
            raise VitalHardGateContractError(f"門檻 rule_id 重複:{rule_id}")
        rule_ids.add(rule_id)
        if field not in numeric_fields or field not in allowed_gate_fields:
            raise VitalHardGateContractError(f"門檻使用未允許欄位:{field}")
        if field == "GCS":
            raise VitalHardGateContractError("資料契約沒有 GCS,不得以 Mental 冒充")
        rule_fields.add(str(field))
        if operator not in VALID_OPERATORS:
            raise VitalHardGateContractError(f"未知 operator:{operator}")
        if level not in VALID_LEVELS:
            raise VitalHardGateContractError(f"非法 most_severe_level:{level}")
        if operator == "in":
            values = rule.get("values")
            if not isinstance(values, list) or not values or not all(
                _is_number(value) for value in values
            ):
                raise VitalHardGateContractError(f"{rule_id} 缺少數值 values")
        elif not _is_number(rule.get("threshold")):
            raise VitalHardGateContractError(f"{rule_id} 缺少數值 threshold")

    selected_levels = [int(candidate["level"]) for candidate in candidates]
    if selected_levels != list(VALID_LEVELS):
        raise VitalHardGateContractError("候選必須是排序後的第一至第五級")

    cases = fixture.get("cases")
    expected_count = int(config["evaluation"]["expected_case_count"])
    if not isinstance(cases, list) or len(cases) != expected_count:
        raise VitalHardGateContractError(
            f"合成案例應為 {expected_count} 筆,實際為 {len(cases) if isinstance(cases, list) else '非陣列'}"
        )
    _assert_no_forbidden_fields(cases)
    case_ids: set[str] = set()
    for case in cases:
        if not isinstance(case, Mapping):
            raise VitalHardGateContractError("每筆案例都必須是 JSON 物件")
        case_id = case.get("case_id")
        if not isinstance(case_id, str) or not case_id:
            raise VitalHardGateContractError("案例缺少 case_id")
        if case_id in case_ids:
            raise VitalHardGateContractError(f"case_id 重複:{case_id}")
        case_ids.add(case_id)
        inputs = case.get("inputs")
        if not isinstance(inputs, Mapping):
            raise VitalHardGateContractError(f"{case_id} 缺少 inputs")
        if set(inputs) != rule_fields:
            raise VitalHardGateContractError(
                f"{case_id} inputs 應為 {sorted(rule_fields)},實際為 {sorted(inputs)}"
            )
        for field, value in inputs.items():
            if value is not None and not _is_number(value):
                raise VitalHardGateContractError(f"{case_id}.{field} 必須是數值或 null")
        mental = inputs.get("Mental")
        if mental is not None and mental not in day14_contract["input_constraints"]["allowed_values"]["Mental"]:
            raise VitalHardGateContractError(f"{case_id}.Mental 不在資料契約值域")
        saturation = inputs.get("Saturation")
        if saturation is not None and not 0 <= float(saturation) <= 100:
            raise VitalHardGateContractError(f"{case_id}.Saturation 超出結構值域")
        scores = case.get("candidate_scores")
        if not isinstance(scores, Mapping) or set(scores) != {
            str(level) for level in VALID_LEVELS
        }:
            raise VitalHardGateContractError(f"{case_id} 必須提供五級共同分數")
        numeric_scores = [scores[str(level)] for level in VALID_LEVELS]
        if not all(_is_number(score) and 0 <= float(score) <= 1 for score in numeric_scores):
            raise VitalHardGateContractError(f"{case_id} 分數必須介於 0 與 1")
        if len(set(float(score) for score in numeric_scores)) != len(VALID_LEVELS):
            raise VitalHardGateContractError(f"{case_id} 分數不可同分")
        if case.get("synthetic_reference_level") not in VALID_LEVELS:
            raise VitalHardGateContractError(f"{case_id} 合成參考級數非法")

    return {
        "scope_is_synthetic_only": True,
        "candidate_count": len(candidates),
        "candidate_levels": selected_levels,
        "case_count": len(cases),
        "threshold_rule_count": len(rules),
        "threshold_fields": sorted(rule_fields),
        "forbidden_patient_or_label_fields_absent": True,
        "missing_values_are_not_imputed_or_treated_as_normal": True,
        "gcs_is_not_used": True,
    }


def _rule_matches(value: float | int, rule: Mapping[str, Any]) -> bool:
    operator = str(rule["operator"])
    if operator == "in":
        return value in rule["values"]
    threshold = rule["threshold"]
    if operator == "lt":
        return value < threshold
    if operator == "lte":
        return value <= threshold
    if operator == "gt":
        return value > threshold
    if operator == "gte":
        return value >= threshold
    if operator == "eq":
        return value == threshold
    raise VitalHardGateContractError(f"未知 operator:{operator}")


def evaluate_vital_gate(
    inputs: Mapping[str, Any],
    rules: Sequence[Mapping[str, Any]],
) -> JsonObject:
    """Evaluate all observable fields; missing fields never become normal values."""

    fields = sorted({str(rule["field"]) for rule in rules})
    missing_fields = [field for field in fields if inputs.get(field) is None]
    triggered = []
    for rule in rules:
        field = str(rule["field"])
        value = inputs.get(field)
        if value is None:
            continue
        if _rule_matches(value, rule):
            triggered.append(
                {
                    "rule_id": rule["rule_id"],
                    "field": field,
                    "observed_value": value,
                    "display": rule["display"],
                    "most_severe_level": int(rule["most_severe_level"]),
                }
            )
    if triggered:
        most_severe_level = min(item["most_severe_level"] for item in triggered)
        route_status = "hard_gate_applied"
    elif missing_fields:
        most_severe_level = None
        route_status = "indeterminate_missing_values"
    else:
        most_severe_level = None
        route_status = "pass_through_no_alarm"
    return {
        "route_status": route_status,
        "missing_fields": missing_fields,
        "triggered_rules": triggered,
        "triggered_rule_ids": [item["rule_id"] for item in triggered],
        "most_severe_level": most_severe_level,
    }


def rank_level_candidates(
    candidates: Sequence[Mapping[str, Any]],
    scores: Mapping[str, Any],
    *,
    most_severe_level: int | None,
) -> List[JsonObject]:
    """Apply the hard filter, then rank the retained public level chunks."""

    if most_severe_level is not None and most_severe_level not in VALID_LEVELS:
        raise VitalHardGateContractError("most_severe_level 必須介於 1 到 5")
    retained = [
        candidate
        for candidate in candidates
        if most_severe_level is None
        or int(candidate["level"]) <= most_severe_level
    ]
    if not retained:
        raise VitalHardGateContractError("Hard Gate 不得產生空候選集合")
    ranked = [
        {
            "level": int(candidate["level"]),
            "rule_id": candidate["rule_id"],
            "synthetic_score": round(float(scores[str(candidate["level"])]), 6),
            "source_url": candidate["source_url"],
        }
        for candidate in retained
    ]
    ranked.sort(key=lambda item: (-item["synthetic_score"], item["level"]))
    for rank, item in enumerate(ranked, start=1):
        item["rank"] = rank
    return ranked


def direction_against_synthetic_reference(top1: int, reference: int) -> str:
    if top1 == reference:
        return "exact"
    if top1 > reference:
        return "undertriage"
    return "overtriage"


def _derive_outcome(
    baseline_direction: str,
    gated_direction: str,
    *,
    reference_retained: bool,
) -> str:
    if baseline_direction == "undertriage" and gated_direction == "exact":
        return "rescued_undertriage"
    if baseline_direction == "undertriage" and gated_direction == "undertriage":
        return "persistent_undertriage"
    if not reference_retained and gated_direction == "overtriage":
        return "reference_removed_overtriage"
    if baseline_direction == gated_direction:
        return "unchanged"
    return f"{baseline_direction}_to_{gated_direction}"


def run_vital_hard_gate_case(
    case: Mapping[str, Any],
    candidates: Sequence[Mapping[str, Any]],
    rules: Sequence[Mapping[str, Any]],
    *,
    top_k: int,
) -> JsonObject:
    """Run one paired no-gate versus hard-gate synthetic case with a trace."""

    reference = int(case["synthetic_reference_level"])
    gate = evaluate_vital_gate(case["inputs"], rules)
    baseline_ranking = rank_level_candidates(
        candidates,
        case["candidate_scores"],
        most_severe_level=None,
    )
    gated_ranking = rank_level_candidates(
        candidates,
        case["candidate_scores"],
        most_severe_level=gate["most_severe_level"],
    )
    baseline_top1 = int(baseline_ranking[0]["level"])
    gated_top1 = int(gated_ranking[0]["level"])
    baseline_direction = direction_against_synthetic_reference(
        baseline_top1, reference
    )
    gated_direction = direction_against_synthetic_reference(gated_top1, reference)
    retained_levels = sorted(int(item["level"]) for item in gated_ranking)
    reference_retained = reference in retained_levels
    outcome = _derive_outcome(
        baseline_direction,
        gated_direction,
        reference_retained=reference_retained,
    )
    actual_expected = {
        "route_status": gate["route_status"],
        "triggered_rule_ids": gate["triggered_rule_ids"],
        "most_severe_level": gate["most_severe_level"],
        "retained_levels": retained_levels,
        "baseline_top1": baseline_top1,
        "gated_top1": gated_top1,
        "reference_retained": reference_retained,
        "outcome": outcome,
    }
    if actual_expected != case["expected"]:
        raise VitalHardGateContractError(
            f"{case['case_id']} 與鎖定預期不同:actual={actual_expected},expected={case['expected']}"
        )
    return {
        "case_id": case["case_id"],
        "scenario": case["scenario"],
        "inputs": dict(case["inputs"]),
        "synthetic_reference": {
            "level": reference,
            "role": "author_written_engineering_reference_not_a_clinical_label",
        },
        "gate": gate,
        "baseline": {
            "candidate_count": len(baseline_ranking),
            "top1_level": baseline_top1,
            "direction": baseline_direction,
            "top_k": baseline_ranking[:top_k],
        },
        "hard_gate": {
            "candidate_count": len(gated_ranking),
            "retained_levels": retained_levels,
            "reference_retained": reference_retained,
            "top1_level": gated_top1,
            "direction": gated_direction,
            "top_k": gated_ranking[:top_k],
        },
        "outcome": outcome,
        "locked_expectation_passed": True,
    }


def summarize_vital_hard_gate_results(
    cases: Sequence[Mapping[str, Any]],
) -> JsonObject:
    """Aggregate descriptive counts without presenting them as clinical rates."""

    baseline_directions = Counter(str(case["baseline"]["direction"]) for case in cases)
    gated_directions = Counter(str(case["hard_gate"]["direction"]) for case in cases)
    route_statuses = Counter(str(case["gate"]["route_status"]) for case in cases)
    outcomes = Counter(str(case["outcome"]) for case in cases)
    return {
        "case_count": len(cases),
        "route_status_counts": dict(sorted(route_statuses.items())),
        "baseline_direction_counts": {
            key: baseline_directions.get(key, 0)
            for key in ("exact", "undertriage", "overtriage")
        },
        "hard_gate_direction_counts": {
            key: gated_directions.get(key, 0)
            for key in ("exact", "undertriage", "overtriage")
        },
        "outcome_counts": dict(sorted(outcomes.items())),
        "reference_retained_count": sum(
            bool(case["hard_gate"]["reference_retained"]) for case in cases
        ),
        "reference_removed_count": sum(
            not bool(case["hard_gate"]["reference_retained"]) for case in cases
        ),
        "all_locked_expectations_passed": all(
            bool(case["locked_expectation_passed"]) for case in cases
        ),
        "claim_boundary": (
            "這些是六筆作者合成契約案例的描述性計數,不是病患分類率、"
            "檢傷不足率、檢傷過度率或臨床安全估計。"
        ),
    }

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

檔案 5:建立 scripts/run_day23_vital_hard_gate.py

核對 Day 14、15、17、18 上游雜湊,配對執行無門控與 Hard Gate,最後寫出公開摘要、完整 trace 與 run manifest。

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

#!/usr/bin/env python3
"""Run the Day 23 synthetic no-gate versus vital hard-gate microbenchmark."""

from __future__ import annotations

import argparse
import platform
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
from triage_rag.rag import (
    run_vital_hard_gate_case,
    select_level_candidates,
    summarize_vital_hard_gate_results,
    validate_vital_hard_gate_contract,
)
from triage_rag.reproducibility import (
    file_record,
    git_state,
    load_json,
    resolve_project_path,
    sha256_file,
    write_json,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/rag/day-23-vital-hard-gate.json"
EXPECTED_SCOPE = (
    "author_written_synthetic_hard_gate_microbenchmark_"
    "not_patient_triage_or_clinical_evaluation"
)
IMPLEMENTATION_PATHS = (
    "src/triage_rag/rag/__init__.py",
    "src/triage_rag/rag/vital_gate.py",
    "scripts/run_day23_vital_hard_gate.py",
    "tests/test_vital_hard_gate.py",
)
JsonObject = Dict[str, Any]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "以六筆作者合成案例配對比較無門控與生命徵象 Hard Gate;"
            "不讀取病患資料或臨床標籤。"
        )
    )
    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 23 契約。"
        )


def _load_and_verify_sources(
    config: Mapping[str, Any],
) -> tuple[List[Path], JsonObject]:
    sources = config["sources"]
    source_pairs = (
        ("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
        (
            "day14_data_contract_path",
            "day14_data_contract_sha256",
            "Day 14 data contract",
        ),
        (
            "day15_quality_contract_path",
            "day15_quality_contract_sha256",
            "Day 15 quality contract",
        ),
        (
            "day18_numeric_contract_path",
            "day18_numeric_contract_sha256",
            "Day 18 numeric contract",
        ),
        (
            "synthetic_cases_path",
            "synthetic_cases_sha256",
            "Day 23 synthetic cases",
        ),
    )
    paths: List[Path] = []
    for path_key, hash_key, label in source_pairs:
        path = resolve_project_path(PROJECT_ROOT, str(sources[path_key]))
        _verify_sha256(path, str(sources[hash_key]), label)
        paths.append(path)
    loaded = {
        "fixture": load_json(paths[4]),
        "day14": load_json(paths[1]),
        "day15": load_json(paths[2]),
        "day18": load_json(paths[3]),
    }
    return paths, loaded


def _verify_cross_day_boundaries(
    config: Mapping[str, Any],
    loaded: Mapping[str, Any],
) -> None:
    if config.get("scope") != EXPECTED_SCOPE:
        raise ValueError("Day 23 scope 不允許病患或臨床評估")
    day18_policy = loaded["day18"]["threshold_policy"]
    if day18_policy.get("status") != "unavailable_in_current_public_knowledge_base":
        raise ValueError("Day 18 公開門檻缺口狀態已改變,應先建立新版契約")
    if day18_policy.get("rules") != [] or not day18_policy.get("fail_closed"):
        raise ValueError("Day 18 不得偷偷加入未鎖定的正式門檻")
    fixture = loaded["fixture"]
    if fixture.get("scope") != (
        "author_written_synthetic_gate_stress_cases_"
        "not_patient_records_or_clinical_labels"
    ):
        raise ValueError("Day 23 fixture 必須明示為作者合成案例")


def _validate_aggregate(
    aggregate: Mapping[str, Any],
    evaluation: Mapping[str, Any],
) -> None:
    route_counts = aggregate["route_status_counts"]
    checks = {
        "case_count": (
            aggregate["case_count"],
            evaluation["expected_case_count"],
        ),
        "alarm_case_count": (
            route_counts.get("hard_gate_applied", 0),
            evaluation["expected_alarm_case_count"],
        ),
        "no_alarm_case_count": (
            route_counts.get("pass_through_no_alarm", 0),
            evaluation["expected_no_alarm_case_count"],
        ),
        "indeterminate_case_count": (
            route_counts.get("indeterminate_missing_values", 0),
            evaluation["expected_indeterminate_case_count"],
        ),
    }
    mismatches = {
        key: {"actual": actual, "expected": expected}
        for key, (actual, expected) in checks.items()
        if actual != expected
    }
    if mismatches:
        raise ValueError(f"Day 23 聚合結果與鎖定案例數不同:{mismatches}")
    if not aggregate["all_locked_expectations_passed"]:
        raise ValueError("至少一筆 Day 23 案例未通過鎖定預期")


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)

    print("步驟 1/4:核對上游雜湊、合成範圍與缺失值政策。", flush=True)
    source_paths, loaded = _load_and_verify_sources(config)
    _verify_cross_day_boundaries(config, loaded)
    documents = read_jsonl(source_paths[0])
    candidates = select_level_candidates(documents)
    input_contract = validate_vital_hard_gate_contract(
        config,
        loaded["fixture"],
        candidates,
        loaded["day14"],
        loaded["day15"],
    )

    print("步驟 2/4:配對執行六筆無門控與 Hard Gate 合成案例。", flush=True)
    rules = config["synthetic_threshold_policy"]["rules"]
    top_k = int(config["evaluation"]["top_k_trace"])
    case_results = []
    for index, case in enumerate(loaded["fixture"]["cases"], start=1):
        trace = run_vital_hard_gate_case(
            case,
            candidates,
            rules,
            top_k=top_k,
        )
        case_results.append(trace)
        print(
            f"  {index:02d}/06 {trace['case_id']}:{trace['outcome']}",
            flush=True,
        )

    print("步驟 3/4:彙整方向、候選保留與不可逆排除。", flush=True)
    aggregate = summarize_vital_hard_gate_results(case_results)
    _validate_aggregate(aggregate, config["evaluation"])
    public_summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "level_contract": config["level_contract"],
        "synthetic_threshold_policy": config["synthetic_threshold_policy"],
        "evaluation_contract": config["evaluation"],
        "input_contract": input_contract,
        "candidate_catalog": [
            {
                "level": candidate["level"],
                "rule_id": candidate["rule_id"],
                "source_url": candidate["source_url"],
            }
            for candidate in candidates
        ],
        "aggregate": aggregate,
        "case_results": case_results,
        "checks": {
            "source_hashes_match": True,
            "day18_official_numeric_threshold_gap_remains_fail_closed": True,
            "thresholds_are_explicitly_synthetic_and_nonclinical": True,
            "same_candidate_scores_used_for_paired_comparison": True,
            "missing_values_never_imputed_or_treated_as_normal": True,
            "only_public_level_definition_chunks_used": True,
            "patient_rows_not_read": True,
            "nurse_and_expert_labels_not_read": True,
            "statistics_eligible": False,
        },
        "limitations": config["limitations"],
        "interpretation": (
            "六筆作者合成案例顯示:警示型硬門控能移除過輕候選,"
            "但未觸發或缺失時不會補入高風險主訴證據,且硬排除可能刪掉"
            "合成參考候選。這是程式行為壓力測試,不是臨床效能或安全結論。"
        ),
    }

    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)
    full_result_path = run_directory / config["outputs"]["result_filename"]
    write_json(full_result_path, public_summary)
    public_path = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["public_summary_path"]
    )
    write_json(public_path, public_summary)

    manifest = {
        "manifest_schema_version": 1,
        "run_id": run_id,
        "experiment_id": config["experiment_id"],
        "started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
        "command": [
            sys.executable,
            str(Path(__file__).relative_to(PROJECT_ROOT)),
            "--config",
            args.config,
        ],
        "git": git_state(PROJECT_ROOT),
        "runtime": {"python": platform.python_version()},
        "parameters": {
            "candidate_levels": list(range(1, 6)),
            "top_k_trace": top_k,
            "threshold_policy_role": config["synthetic_threshold_policy"][
                "policy_role"
            ],
            "clinical_use": config["synthetic_threshold_policy"]["clinical_use"],
        },
        "inputs": [
            file_record(PROJECT_ROOT, args.config),
            *[
                file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
                for path in source_paths
            ],
            *[file_record(PROJECT_ROOT, path) for path in IMPLEMENTATION_PATHS],
        ],
        "outputs": [
            file_record(
                PROJECT_ROOT, str(full_result_path.relative_to(PROJECT_ROOT))
            ),
            file_record(PROJECT_ROOT, str(public_path.relative_to(PROJECT_ROOT))),
        ],
        "scope": config["scope"],
    }
    manifest_path = run_directory / "run-manifest.json"
    write_json(manifest_path, manifest)

    print("步驟 4/4:寫出公開摘要、完整結果與 run manifest。", flush=True)
    print(f"公開摘要:{public_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"完整結果:{full_result_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"執行紀錄:{manifest_path.relative_to(PROJECT_ROOT)}", flush=True)
    return 0


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

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

檔案 6:建立 tests/test_vital_hard_gate.py

以十五項離線測試驗證級數方向、GCS 邊界、禁止標籤、缺失值、多警示、門檻等號、候選保留、錯誤修正與不可逆排除。

請在文字編輯器建立 tests/test_vital_hard_gate.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.rag import (
    VitalHardGateContractError,
    evaluate_vital_gate,
    rank_level_candidates,
    run_vital_hard_gate_case,
    select_level_candidates,
    summarize_vital_hard_gate_results,
    validate_vital_hard_gate_contract,
)
from triage_rag.reproducibility import load_json


ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = ROOT / "configs" / "rag" / "day-23-vital-hard-gate.json"


class Day23VitalHardGateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.config = load_json(CONFIG_PATH)
        sources = self.config["sources"]
        self.fixture = load_json(ROOT / sources["synthetic_cases_path"])
        self.day14 = load_json(ROOT / sources["day14_data_contract_path"])
        self.day15 = load_json(ROOT / sources["day15_quality_contract_path"])
        documents = read_jsonl(ROOT / sources["flat_chunks_path"])
        self.candidates = select_level_candidates(documents)
        self.rules = self.config["synthetic_threshold_policy"]["rules"]

    def test_full_synthetic_contract_passes(self) -> None:
        summary = validate_vital_hard_gate_contract(
            self.config,
            self.fixture,
            self.candidates,
            self.day14,
            self.day15,
        )
        self.assertTrue(summary["scope_is_synthetic_only"])
        self.assertEqual(summary["candidate_levels"], [1, 2, 3, 4, 5])
        self.assertEqual(summary["case_count"], 6)
        self.assertEqual(summary["threshold_rule_count"], 6)
        self.assertTrue(summary["gcs_is_not_used"])

    def test_public_candidates_are_exactly_five_level_definitions(self) -> None:
        self.assertEqual([item["level"] for item in self.candidates], [1, 2, 3, 4, 5])
        self.assertTrue(all(item["rule_id"].startswith("ktas-public-level-") for item in self.candidates))

    def test_gate_direction_drift_is_rejected(self) -> None:
        invalid = copy.deepcopy(self.config)
        invalid["level_contract"]["hard_filter_expression"] = (
            "candidate_level >= most_severe_level"
        )
        with self.assertRaisesRegex(VitalHardGateContractError, "方向"):
            validate_vital_hard_gate_contract(
                invalid, self.fixture, self.candidates, self.day14, self.day15
            )

    def test_gcs_cannot_be_smuggled_in_as_mental(self) -> None:
        invalid = copy.deepcopy(self.config)
        invalid["synthetic_threshold_policy"]["rules"][0]["field"] = "GCS"
        with self.assertRaisesRegex(VitalHardGateContractError, "未允許欄位|GCS"):
            validate_vital_hard_gate_contract(
                invalid, self.fixture, self.candidates, self.day14, self.day15
            )

    def test_patient_label_field_in_fixture_is_rejected(self) -> None:
        invalid = copy.deepcopy(self.fixture)
        invalid["cases"][0]["inputs"]["KTAS_expert"] = 2
        with self.assertRaisesRegex(VitalHardGateContractError, "禁止欄位"):
            validate_vital_hard_gate_contract(
                self.config, invalid, self.candidates, self.day14, self.day15
            )

    def test_tied_synthetic_scores_are_rejected(self) -> None:
        invalid = copy.deepcopy(self.fixture)
        invalid["cases"][0]["candidate_scores"]["1"] = invalid["cases"][0][
            "candidate_scores"
        ]["2"]
        with self.assertRaisesRegex(VitalHardGateContractError, "不可同分"):
            validate_vital_hard_gate_contract(
                self.config, invalid, self.candidates, self.day14, self.day15
            )

    def test_missing_value_is_indeterminate_not_normal(self) -> None:
        case = next(
            item
            for item in self.fixture["cases"]
            if item["case_id"] == "missing-saturation-is-indeterminate"
        )
        gate = evaluate_vital_gate(case["inputs"], self.rules)
        self.assertEqual(gate["route_status"], "indeterminate_missing_values")
        self.assertEqual(gate["missing_fields"], ["Saturation"])
        self.assertIsNone(gate["most_severe_level"])

    def test_observed_alarm_still_applies_when_another_field_is_missing(self) -> None:
        inputs = {
            "Mental": 4,
            "SBP": 120,
            "HR": 80,
            "RR": 18,
            "Saturation": None,
        }
        gate = evaluate_vital_gate(inputs, self.rules)
        self.assertEqual(gate["route_status"], "hard_gate_applied")
        self.assertEqual(gate["most_severe_level"], 1)
        self.assertEqual(gate["missing_fields"], ["Saturation"])

    def test_multiple_alarms_use_smallest_level_number(self) -> None:
        case = next(
            item
            for item in self.fixture["cases"]
            if item["case_id"] == "multiple-alarms-most-severe-wins"
        )
        gate = evaluate_vital_gate(case["inputs"], self.rules)
        self.assertEqual(gate["most_severe_level"], 1)
        self.assertEqual(len(gate["triggered_rules"]), 5)

    def test_gte_boundary_is_inclusive(self) -> None:
        case = next(
            item
            for item in self.fixture["cases"]
            if item["case_id"] == "boundary-alarm-removes-reference"
        )
        gate = evaluate_vital_gate(case["inputs"], self.rules)
        self.assertIn("synthetic-hr-gte-130", gate["triggered_rule_ids"])

    def test_level_filter_uses_less_than_or_equal_direction(self) -> None:
        scores = {str(level): 1.0 - level / 10 for level in range(1, 6)}
        ranking = rank_level_candidates(
            self.candidates,
            scores,
            most_severe_level=2,
        )
        self.assertEqual(sorted(item["level"] for item in ranking), [1, 2])

    def test_no_alarm_preserves_all_candidates_and_ranking(self) -> None:
        case = next(
            item
            for item in self.fixture["cases"]
            if item["case_id"] == "no-alarm-high-risk-complaint"
        )
        trace = run_vital_hard_gate_case(
            case,
     

上一篇
Day 22|建立階層式檢索增強生成:先找父層,真的比較好嗎?
系列文
30 天打造公開資料版急診檢傷系統:Side Project 與實驗計畫23
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言