Day 22 把九筆公開知識整理成三個父層,完成 Parent Top-2 再 Child Top-3 的階層式檢索增強生成(Retrieval-Augmented Generation, RAG)。實驗同時暴露了一個不可逆問題:父層沒有入選時,正確子規則就不會進入後續排序。
今天把同一個問題搬到生命徵象。假設候選分數把第四級排在最前面,但合成血氧飽和度已觸發警示;如果在檢索前只允許第一、二級候選,原本過輕的分數第一名(Top-1)就會消失。反過來,若生命徵象沒有觸發作者設定的門檻,Hard Gate 也不會主動補入高風險主訴證據;若門檻本身太積極,還可能直接刪掉原本正確的候選。
先把最重要的限制放在開頭:本篇沒有取得可重建完整韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)的公開數值門檻。因此,接下來六條門檻、六筆案例、候選分數與合成參考級數,全是作者為測試程式行為設計的合成工程示例,不是 KTAS 正式規則、真實病患資料或臨床標籤。
下圖用概念示意呈現警示型硬門控的兩面性:觸發時縮窄候選,未觸發時也可能看不見主訴風險。圖中沒有正式門檻或真實病患資訊。

上圖的重點不是「生命徵象可以決定檢傷級數」,而是「硬排除會改變後續還看得到什麼」。系統最後仍只是研究與教學用決策支援元件,臨床人員必須保有完整情境與最終判斷權。
完成本篇後,你會得到四個可驗證產物:
candidate_level <= most_severe_level 的可測試 Hard Gate。今天不會訓練五級分類器,也不會輸出真實病患的 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,是為了避免把作者假設偽裝成臨床標籤。即使合成方向計算使用「檢傷不足」與「檢傷過度」,分母仍只有六筆作者案例,不能報成病患檢傷不足率或檢傷過度率。
本系列固定採用第一級最急、第五級最不急的數字方向。令候選級數為 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。
下圖由左到右呈現輸入、合成警示、數字最小的級數上限與候選篩選。請特別觀察:無警示不是「正常證明」,缺失也不是零或正常值。

上圖把「警示計算」和「候選排序」分開。Hard Gate 只改變誰還能參賽;保留下來的候選仍使用同一組合成分數排序。這樣比較時唯一改變的是門控,不會同時混入不同向量嵌入(Embedding)、重排序器或生成模型。
Day 15 的缺失政策要求:生命徵象缺失時要保留「未記錄」狀態,不得補成零,也不得改寫成正常。本篇把這項政策延伸到 Hard Gate:
indeterminate_missing_values。這不是說「全保留就安全」。它只表示硬門控不能從不存在的數值推論風險不存在。Day 24 還會加入主訴與高風險保護候選,處理單靠生命徵象無法補救的路徑。
六筆案例都包含完全相同的五筆公開級數定義候選。每筆案例另外鎖定五個不同的合成候選分數;無門控與 Hard Gate 共用同一組分數。
兩條路徑只有一個差別:
| 比較項目 | 無門控 | Hard Gate |
|---|---|---|
| 公開候選來源 | 五筆級數定義 | 同一組五筆級數定義 |
| 合成候選分數 | 固定 | 完全相同 |
| 生命徵象警示 | 不套用 | 套用六條合成規則 |
| 候選排除 | 無 | 有警示時保留 level <= ceiling |
| Top-1 方向 | 對合成參考級數比較 | 對同一合成參考級數比較 |
| 真實病患/臨床標籤 | 不使用 | 不使用 |
這個微型實驗不執行大型語言模型(Large Language Model, LLM)生成。原因是今天要單獨測試候選集合;若同時加入自由文字生成,輸出差異就可能來自生成隨機性,而不是 Hard Gate。
六筆案例不是為了模擬真實盛行率,而是刻意覆蓋六種程式邊界:
Mental = 4 只保留第一級。>= 的等號與正確候選遭硬排除。其中第四筆雖然在案例敘述中寫「高風險主訴」,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 "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案,把對應區塊完整貼入後儲存。
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 方向,不宣稱分類準確率、召回率、檢傷不足率或臨床安全性。",
"系統只能作為研究與教學用決策支援元件,不能取代護理師、醫師或正式檢傷流程。"
]
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
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"
}
}
]
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
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",
]
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
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": (
"這些是六筆作者合成契約案例的描述性計數,不是病患分類率、"
"檢傷不足率、檢傷過度率或臨床安全估計。"
),
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
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())
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
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,