Day 25 已經把系統輸出分成回答、證據不足拒答與技術失敗,也確認引用必須落在本次取回的證據集合內。不過,那十筆資料是作者刻意設計的合成契約案例,不能回答一個更根本的問題:面對同一批病患紀錄,複雜方法是否真的比簡單方法好?
假設一個系統用了向量檢索、階層路由、生命徵象門控與文字生成,最後得到 54% 準確率。只看這個數字,我們不知道 54% 是好是壞。如果「永遠猜資料中最多的級數」已經有 52%,複雜系統增加的價值就很有限;如果簡單模型只有 30%,解讀又完全不同。
因此,本篇不急著替完整架構宣布勝負,而是先建立能重跑、能稽核、資料範圍清楚的比較地板。下圖是概念示意,請先觀察左側三個已能在相同資料契約下執行的工具,以及右側仍以半透明方式保留的複雜方法。

上圖的平衡秤不是要把不同工具假裝成同一種方法,而是提醒我們:只有在相同資料、相同切分與相同指標下得到的結果,才可以並排。已在公開知識題或合成案例通過工程測試的方法,還不能直接加入病患分類排行榜。
本篇實際完成三件事:
B0。B1。B2。三者共用 Day 15 已完成的 1,267 筆資料、五次重複五折分派與專家重新判定級數。本篇同時保留 6,335 筆折外預測供後續配對分析,但公開結果只包含聚合數字。
本篇的新名詞較多,先用表格建立快速索引;後續第一次進入方法時仍會用白話重新說明。
| 中文名稱 | 英文全名/縮寫 | 本篇用途 |
|---|---|---|
| 檢索增強生成 | Retrieval-Augmented Generation, RAG | 先檢索外部知識,再讓生成模型依證據回答的複雜方法家族 |
| 大型語言模型 | Large Language Model, LLM | B4 之後規劃使用的文字生成模型;本篇沒有執行病患折外版本 |
| 基準方法 | Baseline | 提供最低比較地板,避免只看複雜方法自己的結果 |
| 折外預測 | Out-of-Fold prediction, OOF prediction | 每筆資料只由沒有看過它的訓練折模型產生預測 |
| 平衡準確率 | Balanced Accuracy | 先計算每個級數召回率,再對五級取平均,降低大類別支配結果的問題 |
| 巨集平均 F1 分數 | Macro-averaged F1 score, Macro F1 | 先計算每個級數的精確率與召回率調和平均,再對五級平均 |
| 平均絕對誤差 | Mean Absolute Error, MAE | 衡量預測級數平均離參考級數幾級 |
| 檢傷不足 | Undertriage | 預測數字大於參考級數,也就是把案例判得較不緊急 |
| 檢傷過度 | Overtriage | 預測數字小於參考級數,也就是把案例判得較緊急 |
| Ridge 迴歸 | Ridge Regression | 以平方誤差加上係數懲罰的線性模型;本篇把連續輸出四捨五入為五級 |
韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)用 1 到 5 表示急迫程度,數字越小越緊急。本系列使用的公開研究資料把現場護理師登錄與專家重新判定分開保存,研究也明確描述原始與專家重新檢傷的比較用途。PLOS ONE 原始研究
本篇的三種級數不可混用:
| 欄位/產物 | 角色 | 本篇如何使用 |
|---|---|---|
KTAS_expert |
專家重新判定級數 | B0、B2 的訓練目標與評估參考 |
KTAS_RN |
現場護理師登錄級數 | 只做一次描述性人類現場參考,不當模型輸入 |
B0_predicted_level、B2_predicted_level |
模型預測級數 | 只由對應訓練折擬合後,對驗證折產生 |
KTAS_expert 是本次工程比較的參考標籤,不代表沒有測量誤差的絕對真相;KTAS_RN 也不是可以和演算法互換的「另一個模型」。本篇不使用診斷、檢查、處置、住院結果或急診停留時間等檢傷後資訊,避免把未來才知道的結果洩漏進輸入。
檢索增強生成(Retrieval-Augmented Generation, RAG)會先從外部知識庫找資料,再讓大型語言模型(Large Language Model, LLM)依取回內容產生回答。這種架構能提供來源脈絡,但元件多、失敗路徑也多;原始 RAG 研究同樣把「參數內知識」與「可取回的非參數知識」分開討論。Lewis 等人的 RAG 論文
基準方法(Baseline)不是為了假裝簡單模型足以臨床使用,而是回答三個不同問題:
B0:完全不看特徵,只利用訓練標籤分布,最低地板有多高?B1:只靠目前可公開重建的規則,能安全回答多少案例?B2:不使用主訴文字或公開規則,只看檢傷當下結構化欄位,線性方法能做到哪裡?常數預測器是正式的比較工具,而不是玩笑。常見機器學習工具也提供依訓練資料最高頻類別預測的虛設分類器(Dummy Classifier),目的正是建立可解讀的簡單基準。scikit-learn DummyClassifier 文件
Day 08 的比較矩陣規劃了 B0 到 P0。本篇沒有把「規劃」寫成「完成」,而是先依目前證據標記就緒狀態。
| ID | 方法 | 本篇狀態 | 能否加入 Day 26 病患 OOF 比較 |
|---|---|---|---|
B0 |
訓練折多數類別 | 已執行 | 可以 |
B1 |
部分公開規則,缺口即拒答 | 已執行 | 可以,但覆蓋率為 0,分類指標不計算 |
B2 |
結構化序位 Ridge 教學基準 | 已執行 | 可以 |
B3 |
主訴文字嵌入加結構化特徵 | 尚未執行 | 不可以 |
B4 |
相同本機 LLM、不使用檢索 | 尚未執行 | 不可以 |
B5 |
平面稠密 RAG | 只有公開知識問答工程結果 | 不可以 |
H0 |
階層 RAG、不門控 | 只有公開知識問答工程結果 | 不可以 |
H1 |
階層 RAG 加生命徵象 Hard Gate | 只有合成門控結果 | 不可以 |
P0 |
完整 Safety Union RAG | 只有元件契約結果 | 不可以 |
Day 21 到 Day 25 已證明部分元件可以依固定合約運作,但那些資料是公開知識問題或作者合成案例,不是這 1,267 筆病患紀錄的折外分類。若把「六題通過五題」和「1,267 筆準確率」放在同一欄排序,分母、任務與標籤全部不同,數字沒有可比較意義。
這也是本篇沒有硬湊 B4 到 P0 結果的原因。透明報告應清楚說明研究資料、預測目標、分析方法與模型評估,而不是用空缺數字製造完成感;TRIPOD+AI 提供的就是預測模型研究透明報告框架,但它不是模型品質認證或臨床部署許可。TRIPOD+AI 說明與清單
公平比較不只等於「使用同一份逗號分隔值檔案(Comma-Separated Values, CSV)」。至少要固定以下四個條件:
record_index 分派。B2 刻意不使用 Chief_complain,讓本篇先建立純結構化地板。KTAS_expert 計算相同指標;拒答的 B1 另外報覆蓋率與拒答理由。折外預測(Out-of-Fold prediction, OOF prediction)表示每筆驗證資料都由「訓練時沒有看過該筆資料」的模型預測。下圖要看的重點,是補值中位數、平均值、標準差、類別清單與 Ridge 係數都只能在訓練折擬合;驗證折只能套用已完成的狀態。

上圖中的限制同樣重要:來源資料沒有病患識別碼,因此只能確認 record_index 層級每次恰好驗證一次,不能宣稱同一病患從未跨折。常見的資料洩漏包括先對全體資料補值或縮放,再做交叉驗證;官方機器學習文件也建議先切分,且只在訓練資料上呼叫會學習狀態的前處理。scikit-learn 常見陷阱
B0 完全不讀病患特徵。每一折只計算訓練折中 KTAS_expert 最常出現的級數,接著把該級數填給整個驗證折;若兩級同票,固定選數字較小、較緊急的級數。
可以把規則寫成:
[
\hat{y}{B0}=\arg\max{k\in{1,2,3,4,5}}N_{train,k}
]
其中:
例如訓練折五級筆數依序為 20、170、390、360、70,第三級的 390 筆最多,所以這一折不管輸入內容都預測第三級。這個方法可以得到不低的整體準確率,卻可能完全漏掉少數的第一、二或第五級;因此不能只看準確率。
B1 的目標不是測試正式 KTAS 規則,而是檢查「目前專案公開且可重建的知識」能否把單筆病患走完決策路徑。Day 16 的覆蓋盤點仍有五個直接影響個案分類的缺口:
在這些缺口補齊前,程式無法證明任一個案經過完整且版本正確的規則,所以 B1 對所有驗證紀錄輸出 abstained,理由固定為 incomplete_public_rule_coverage。
零覆蓋是一個有效結果,不是程式壞掉,也不代表正式 KTAS 制度不能分類。它只表示:目前公開知識不足以讓本專案誠實重建病患級 rule-only baseline。把五級公開定義直接當完整個案演算法,才是危險的假完成。
B2 使用 Ridge 迴歸(Ridge Regression)建立一個線性地板。Ridge 會在平方誤差之外懲罰過大的係數,降低特徵彼此相關時係數不穩定的問題。本篇使用 NumPy 的閉式解,不新增模型套件。
輸入只包含檢傷當下可得的結構化欄位:
| 類型 | 欄位 | 訓練折內處理 |
|---|---|---|
| 數值 | Age、NRS_pain、SBP、DBP、HR、RR、BT、Saturation |
中位數補值、缺失指示欄、平均與母體標準差縮放 |
| 類別 | Sex、Arrival mode、Injury、Mental、Pain |
只依訓練折觀察值建立獨熱編碼(one-hot encoding)欄位;未知類別全為 0 |
| 刻意排除 | Chief_complain |
留給 B3 的文字加結構化比較 |
線性模型的目標可寫成:
[
\hat{\beta}=\arg\min_{\beta}\sum_i(y_i-X_i\beta)^2+\alpha\sum_{j=1}^{p}\beta_j^2
]
其中:
KTAS_expert 級數。模型先輸出連續分數,再計算 floor(score + 0.5) 並限制在 1 到 5。例如連續分數 2.62 會變成第三級,0.84 會先四捨五入再被限制為第一級。
這個做法使用級數的先後順序,但也隱含「第一級到第二級」和「第四級到第五級」距離相同的近似。它不是序位邏輯斯迴歸(Ordinal Logistic Regression),也不是正式臨床模型;名稱刻意寫成「序位 Ridge 教學基準」,避免把數字編碼誤稱為更完整的序位機率模型。
平衡準確率(Balanced Accuracy)會先計算每一級召回率,再平均:
[
Balanced\ Accuracy=\frac{Recall_1+Recall_2+Recall_3+Recall_4+Recall_5}{5}
]
若某模型五級召回率是 0%、20%、80%、50%、0%,平衡準確率就是 ((0+0.2+0.8+0.5+0)/5=0.30),也就是 30%。這個指標能揭露模型是否只照顧大類別;官方定義同樣把它描述為各類別召回率的平均。scikit-learn balanced accuracy 文件
平均絕對誤差(Mean Absolute Error, MAE)則保留級數距離:
[
MAE=\frac{1}{n}\sum_{i=1}^{n}|\hat{y}_i-y_i|
]
例如三筆參考級數為 [1, 3, 5],預測為 [3, 3, 4],絕對誤差為 [2, 0, 1],所以 MAE 是 ((2+0+1)/3=1) 級。MAE 越小越好,但它仍沒有反映錯誤方向。
本篇因此同時報告:
prediction > reference,數字越大代表判得越不緊急。prediction < reference,數字越小代表判得更緊急。B1 沒有回答任何紀錄,所以不計算分類指標。若把拒答硬轉成第五級再算錯誤率,會混淆「不知道」與「判為不緊急」兩種完全不同的輸出。
以下檔案共同完成契約、核心運算、執行入口、驗證與圖片重建。設定與結果使用 JavaScript 物件表示法(JavaScript Object Notation, JSON),它是一種可由人閱讀、也能讓程式驗證的結構化文字格式:
| 路徑 | 類型與資料夾用途 | 輸入 | 輸出/影響 |
|---|---|---|---|
configs/baselines/day-26-fair-baselines.json |
configs/baselines/ 保存基準契約的 JSON 設定 |
Day 08–09、Day 14–16 與 Day 21–25 產物雜湊、欄位與方法政策 | 鎖定三個執行基準與九方法就緒狀態 |
src/triage_rag/baselines/__init__.py |
src/ 是可重用 Python 套件;此檔公開 baseline 介面 |
核心函式名稱 | 讓 runner 與測試使用穩定 import |
src/triage_rag/baselines/simple.py |
基準核心 Python 程式 | 三份 Day 15 資料、契約 | 驗證切分、折內前處理、預測與指標 |
scripts/run_day26_baselines.py |
scripts/ 保存可直接執行的自動化入口 |
設定、CSV、上游公開 JSON | 公開摘要、完整 OOF 結果與 manifest |
tests/test_simple_baselines.py |
tests/ 保存不變量測試 |
真實本機前置產物與小型合成表格 | 驗證洩漏防線、指標、拒答與重跑一致 |
scripts/figures/day-26/generate_day26_figures.py |
可重跑圖表程式 | Day 26 設定與公開結果 | 兩張 1920×1080 繁中技術圖 |
Day 15 已完整建立三個本機前置檔案:model-input-candidates.csv 保存檢傷當下候選輸入,reference-labels.csv 分開保存 KTAS_RN 與 KTAS_expert,day-15-split-assignments.csv 保存五次五折分派。本篇不把資料列貼進文章,也不提交它們到版本控制;若缺少這三個產物,應先完成 Day 15「實際執行 Day 15」步驟,而不是自行捏造病患資料。
接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾,依下列順序建立檔案;每個程式碼區塊都是該檔案的完整內容,不含省略號。
本篇沿用 Day 08–09 的比較與研究契約、Day 15 已完成且留在本機的模型輸入、參考標籤與五次五折分派,以及 Day 16 的公開知識缺口摘要;Day 21–25 的公開結果只用來鎖定既有 RAG 元件仍非病患 OOF 的比較邊界。以下是 Day 26 新增的完整設定、baseline 套件、runner、測試與技術圖片生成程式;資料 CSV、公開摘要、完整折外預測與 run manifest 都由前置步驟或執行入口產生,不需要人工建立或貼入病患資料。
先從專案根目錄建立需要的資料夾:
mkdir -p configs/baselines src/triage_rag/baselines scripts scripts/figures/day-26 tests results/public results/runs/day-26
如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案,把對應區塊完整貼入後儲存。
configs/baselines/day-26-fair-baselines.json鎖定三個執行基準、九方法就緒登錄、來源雜湊、允許/禁止欄位、五次五折、指標、輸出與非臨床解讀邊界。
請在文字編輯器建立 configs/baselines/day-26-fair-baselines.json,貼入以下完整內容並儲存:
{
"schema_version": 1,
"experiment_id": "day-26-locked-oof-simple-baselines",
"scope": "locked_record_level_oof_engineering_baselines_not_confirmatory_or_clinical_evaluation",
"sources": {
"model_inputs_path": "data/interim/ktas-v1/model-input-candidates.csv",
"model_inputs_sha256": "42c86a5ca22f06b51b8227799c77dcb91879e88fcef1a18e30a9cf543ba8b2a4",
"reference_labels_path": "data/interim/ktas-v1/reference-labels.csv",
"reference_labels_sha256": "69fa2b40eef7bac4d61c88fd1a91b639b334e157021440dd5317b94f77a830ff",
"split_assignments_path": "data/processed/ktas-v1/day-15-split-assignments.csv",
"split_assignments_sha256": "4e254967c612063c57aca7e3c1f8960828135b1c762401438b4b52f279a44c51",
"day14_contract_path": "configs/data/day-14-ktas-data-contract.json",
"day14_contract_sha256": "2a7d0fc31782ae0b9e90c1a98eadd722e4236c264bc0ee6ee7f4f6b0f98c4e86",
"day15_contract_path": "configs/data/day-15-quality-and-split-contract.json",
"day15_contract_sha256": "afef6b89c9ae9a048326271c2ad74649ae0c029f24bbd839581aa46129498cf5",
"day15_public_result_path": "results/public/day-15-data-quality-and-splits.json",
"day15_public_result_sha256": "4fa89330a10b4278c9c249ec3360a600ad904fc0c5137c32757d4bfaf706d4ad",
"day16_coverage_path": "results/public/day-16-knowledge-coverage.json",
"day16_coverage_sha256": "1e1a50c01b18a4cac4c7e2c87d37cb49dc2fac1ed2475259d914d1c933967588",
"day08_comparison_contract_path": "configs/experiments/day-08-controlled-comparisons.json",
"day08_comparison_contract_sha256": "dd2f9486712b9abcd22e4b2b71130f6b8bb06d147caa14f3807ff89c25dd2cd0",
"day09_research_protocol_path": "configs/experiments/day-09-research-protocol.json",
"day09_research_protocol_sha256": "be9954891978cf654cf0031adaea6fa37d4e940d067b3e02293b3199a9cf0e19",
"day21_public_result_path": "results/public/day-21-flat-basic-rag.json",
"day21_public_result_sha256": "7d5ca9f3594da7d13da3c323244303ae2d52f7d66e219fa6608e1d18705ec3de",
"day22_public_result_path": "results/public/day-22-hierarchical-rag.json",
"day22_public_result_sha256": "fa3150c5ce32e6fd16bc9b30306e0f09526b46965179939a4f81afd0d090e82e",
"day23_public_result_path": "results/public/day-23-vital-hard-gate.json",
"day23_public_result_sha256": "e2533938753e362fedffd6964b1209eed9aa6334ccb945c296541a603a85aa93",
"day24_public_result_path": "results/public/day-24-safety-union-gate.json",
"day24_public_result_sha256": "561a599b562b31b03faabafc39c074b0209708af3a53c5b6325a18db431674fb",
"day25_public_result_path": "results/public/day-25-grounded-output-and-abstention.json",
"day25_public_result_sha256": "af3b0245dabded6891bee989d292349c943f60b3e24ed00b2d57089b4a11fa65"
},
"data_contract": {
"row_link_key": "record_index",
"reference_label_field": "KTAS_expert",
"human_reference_field": "KTAS_RN",
"valid_levels": [1, 2, 3, 4, 5],
"label_direction": "smaller_number_is_more_urgent",
"record_count": 1267,
"repeat_count": 5,
"fold_count": 5,
"assignment_row_count": 6335,
"each_record_is_validation_once_per_repeat": true,
"patient_identifier_available": false,
"patient_level_split_claim_allowed": false,
"forbidden_model_input_fields": [
"record_index",
"KTAS_RN",
"KTAS_expert",
"Group",
"diagnosis",
"disposition",
"length_of_stay",
"Error_group",
"mistriage"
]
},
"executed_baselines": {
"B0": {
"name": "training-fold majority class",
"role": "ignores patient features and predicts the most frequent KTAS_expert level in the current training fold",
"patient_feature_fields": [],
"fit_inputs": ["training_fold_KTAS_expert"],
"tie_break": "smaller_level_number",
"coverage_policy": "answer_every_validation_record"
},
"B1": {
"name": "partial public-rule baseline with fail-closed abstention",
"role": "checks whether the current public knowledge can map an individual encounter through a complete KTAS decision path",
"patient_feature_fields": [],
"fit_inputs": [],
"required_gap_ids": [
"adult_chief_complaint_catalog",
"pediatric_chief_complaint_catalog",
"complaint_specific_thresholds",
"complete_decision_paths",
"official_version_history"
],
"decision": "abstain_all_until_complete_versioned_patient_level_rules_are_available",
"abstention_reason_code": "incomplete_public_rule_coverage",
"coverage_policy": "zero_coverage_is_a_valid_baseline_result_not_a_system_failure"
},
"B2": {
"name": "structured ordinal ridge teaching baseline",
"role": "uses only triage-time structured fields and the numeric order of levels; it is not ordinal logistic regression",
"numeric_fields": [
"Age",
"NRS_pain",
"SBP",
"DBP",
"HR",
"RR",
"BT",
"Saturation"
],
"categorical_fields": [
"Sex",
"Arrival mode",
"Injury",
"Mental",
"Pain"
],
"excluded_available_field": "Chief_complain",
"preprocessing": {
"numeric_imputation": "training_fold_median",
"numeric_missing_indicators": true,
"numeric_scaling": "training_fold_population_mean_and_standard_deviation",
"categorical_encoding": "training_fold_one_hot_unknown_all_zero",
"fit_scope": "training_fold_only"
},
"model": {
"loss": "squared_error_on_ordered_level_numbers",
"ridge_alpha": 1.0,
"penalize_intercept": false,
"prediction_rounding": "floor_continuous_score_plus_0_5",
"prediction_clipping": [1, 5],
"hyperparameter_search": false
},
"coverage_policy": "answer_every_validation_record"
}
},
"comparison_registry": [
{
"id": "B0",
"name": "訓練折多數類別",
"status": "executed_day26",
"uses_structured_fields": false,
"uses_chief_complaint_text": false,
"uses_public_rule_retrieval": false,
"uses_generation": false
},
{
"id": "B1",
"name": "部分公開規則;不足即拒答",
"status": "executed_day26",
"uses_structured_fields": false,
"uses_chief_complaint_text": false,
"uses_public_rule_retrieval": true,
"uses_generation": false
},
{
"id": "B2",
"name": "結構化序位 Ridge 教學基準",
"status": "executed_day26",
"uses_structured_fields": true,
"uses_chief_complaint_text": false,
"uses_public_rule_retrieval": false,
"uses_generation": false
},
{
"id": "B3",
"name": "主訴文字嵌入加結構化特徵",
"status": "planned_not_executed",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": false,
"uses_generation": false
},
{
"id": "B4",
"name": "相同本機 LLM、不使用檢索",
"status": "planned_not_executed",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": false,
"uses_generation": true
},
{
"id": "B5",
"name": "平面稠密 RAG",
"status": "public_knowledge_qa_only_not_patient_oof_ready",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": true,
"uses_generation": true
},
{
"id": "H0",
"name": "階層 RAG、不門控",
"status": "public_knowledge_qa_only_not_patient_oof_ready",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": true,
"uses_generation": true
},
{
"id": "H1",
"name": "階層 RAG 加生命徵象 Hard Gate",
"status": "synthetic_gate_only_not_patient_oof_ready",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": true,
"uses_generation": true
},
{
"id": "P0",
"name": "完整 Safety Union RAG",
"status": "component_contracts_only_not_patient_oof_ready",
"uses_structured_fields": true,
"uses_chief_complaint_text": true,
"uses_public_rule_retrieval": true,
"uses_generation": true
}
],
"evaluation": {
"classification_metrics": [
"accuracy",
"balanced_accuracy",
"macro_f1",
"mean_absolute_error",
"undertriage_rate",
"overtriage_rate",
"severe_undertriage_rate",
"two_or_more_level_error_rate",
"per_class_recall"
],
"abstention_metrics": ["coverage", "abstention_rate", "reason_counts"],
"severe_undertriage_definition": "reference_level_in_1_or_2_and_prediction_level_in_3_to_5",
"two_or_more_level_error_definition": "absolute_prediction_minus_reference_is_at_least_2",
"repeat_summary": "arithmetic_mean_and_range_over_five_complete_oof_repeats",
"uncertainty_status": "not_computed_until_day28_locked_record_level_paired_bootstrap",
"confirmatory_status": "not_eligible_because_day09_multiplicity_and_safety_margins_remain_unlocked"
},
"human_reference_context": {
"id": "KTAS_RN",
"role": "observed_triage_nurse_reference_for_context_not_a_trained_algorithm_or_ranked_model",
"compared_with": "KTAS_expert",
"model_input_use_forbidden": true,
"fold_training_not_applicable": true
},
"outputs": {
"public_summary_path": "results/public/day-26-simple-baselines.json",
"run_output_root": "results/runs/day-26",
"full_result_filename": "day-26-oof-baseline-results.json"
},
"limitations": [
"B0、B1 與 B2 共用 Day 15 的紀錄層級五次重複五折分派;公開資料沒有病患識別碼,因此不能宣稱病患層級獨立切分。",
"B1 的零覆蓋是目前公開規則缺少完整主訴目錄、門檻、決策路徑與版本歷史所導致的 fail-closed 結果,不代表正式 KTAS 規則本身無法檢傷。",
"B2 是不新增套件的序位 Ridge 教學基準,以級數數字做平方誤差並隱含等距近似;它不是正式 ordinal logistic regression,也沒有使用主訴文字。",
"B2 的 alpha=1.0、欄位與前處理在執行前固定,沒有用折外結果調參;本篇不做方法選拔。",
"KTAS_RN 只提供來源資料中的人類現場參考,不是模型、不是可交換的 ground truth,也不參與 B0 或 B2 輸入。",
"B3、B4、B5、H0、H1 與 P0 尚未在相同病患 OOF 契約下執行;既有公開知識題或合成門控結果不能和 Day 26 的病患分類指標混排。",
"本篇不計算信賴區間,不做確認性推論,也不能據此宣稱臨床安全或部署資格。"
]
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/baselines/__init__.py建立 baselines 子套件入口,公開 Day 26 契約驗證、前處理、指標與折外執行函式。
請在文字編輯器建立 src/triage_rag/baselines/__init__.py,貼入以下完整內容並儲存:
"""Leakage-aware baseline models and evaluation helpers."""
from .simple import (
BaselineContractError,
classification_metrics,
fit_structured_preprocessor,
run_day26_oof,
transform_structured_features,
validate_day26_inputs,
)
__all__ = [
"BaselineContractError",
"classification_metrics",
"fit_structured_preprocessor",
"run_day26_oof",
"transform_structured_features",
"validate_day26_inputs",
]
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/baselines/simple.py驗證資料與切分,實作訓練折多數類別、部分規則全拒答、折內前處理、序位 Ridge、折外預測與安全方向指標。
請在文字編輯器建立 src/triage_rag/baselines/simple.py,貼入以下完整內容並儲存:
"""Day 26 simple baselines with fold-local preprocessing and ordinal metrics."""
from __future__ import annotations
import math
from collections import Counter
from typing import Any, Mapping
import numpy as np
import pandas as pd
JsonObject = dict[str, Any]
class BaselineContractError(ValueError):
"""Raised when data or evaluation inputs violate the locked contract."""
def _require_columns(frame: pd.DataFrame, fields: list[str], name: str) -> None:
missing = [field for field in fields if field not in frame.columns]
if missing:
raise BaselineContractError(f"{name} 缺少必要欄位:{missing}")
def _safe_rate(numerator: int, denominator: int) -> float | None:
return round(numerator / denominator, 6) if denominator else None
def validate_day26_inputs(
model_inputs: pd.DataFrame,
labels: pd.DataFrame,
assignments: pd.DataFrame,
contract: Mapping[str, Any],
day16_coverage: Mapping[str, Any],
) -> JsonObject:
"""Validate row alignment, fold completeness, leakage fields, and B1 gaps."""
data = contract["data_contract"]
row_key = str(data["row_link_key"])
reference = str(data["reference_label_field"])
human = str(data["human_reference_field"])
levels = [int(value) for value in data["valid_levels"]]
b2 = contract["executed_baselines"]["B2"]
feature_fields = [*b2["numeric_fields"], *b2["categorical_fields"]]
_require_columns(model_inputs, [row_key, *feature_fields], "model_inputs")
_require_columns(labels, [row_key, reference, human], "reference_labels")
_require_columns(
assignments, [row_key, "repeat_id", "fold_id", "seed"], "assignments"
)
for name, frame in {"model_inputs": model_inputs, "reference_labels": labels}.items():
if frame[row_key].isna().any() or frame[row_key].duplicated().any():
raise BaselineContractError(f"{name} 的 {row_key} 必須完整且唯一")
if model_inputs[row_key].tolist() != labels[row_key].tolist():
raise BaselineContractError("model_inputs 與 reference_labels 的列鍵或順序不同")
if len(model_inputs) != int(data["record_count"]):
raise BaselineContractError("資料筆數與 Day 26 契約不一致")
for field in [reference, human]:
observed = set(pd.to_numeric(labels[field], errors="coerce").dropna().astype(int))
if labels[field].isna().any() or not observed.issubset(levels):
raise BaselineContractError(f"{field} 出現缺失或契約外級數")
forbidden = set(data["forbidden_model_input_fields"])
overlap = sorted(forbidden.intersection(feature_fields))
if overlap:
raise BaselineContractError(f"B2 特徵包含禁止欄位:{overlap}")
if b2["excluded_available_field"] in feature_fields:
raise BaselineContractError("Chief_complain 必須排除於 Day 26 B2")
expected_repeats = set(range(1, int(data["repeat_count"]) + 1))
expected_folds = set(range(1, int(data["fold_count"]) + 1))
if set(assignments["repeat_id"].astype(int)) != expected_repeats:
raise BaselineContractError("repeat_id 集合不完整")
if len(assignments) != int(data["assignment_row_count"]):
raise BaselineContractError("assignment 筆數不符合契約")
expected_ids = set(model_inputs[row_key].tolist())
for repeat_id, repeat_rows in assignments.groupby("repeat_id"):
if set(repeat_rows["fold_id"].astype(int)) != expected_folds:
raise BaselineContractError(f"repeat {repeat_id} 的 fold 不完整")
if repeat_rows[row_key].duplicated().any() or set(repeat_rows[row_key]) != expected_ids:
raise BaselineContractError(f"repeat {repeat_id} 未讓每筆紀錄恰好驗證一次")
gaps = {
str(item["coverage_id"]): str(item["status"])
for item in day16_coverage["coverage"]["items"]
}
required_gaps = list(contract["executed_baselines"]["B1"]["required_gap_ids"])
unresolved = [gap for gap in required_gaps if gaps.get(gap) != "gap"]
if unresolved:
raise BaselineContractError(f"B1 必要知識缺口未被 Day 16 證實:{unresolved}")
return {
"record_count": len(model_inputs),
"assignment_row_count": len(assignments),
"repeat_count": len(expected_repeats),
"fold_count": len(expected_folds),
"aligned_record_indices": True,
"each_record_validated_once_per_repeat": True,
"forbidden_fields_absent_from_b2": True,
"b1_required_gaps_confirmed": required_gaps,
"patient_level_independence_claim_allowed": False,
}
def fit_structured_preprocessor(
frame: pd.DataFrame,
numeric_fields: list[str],
categorical_fields: list[str],
) -> JsonObject:
"""Fit medians, population scaling, and category levels on one training fold."""
_require_columns(frame, [*numeric_fields, *categorical_fields], "training fold")
numeric: JsonObject = {}
feature_names: list[str] = []
for field in numeric_fields:
values = pd.to_numeric(frame[field], errors="coerce")
if values.notna().sum() == 0:
raise BaselineContractError(f"訓練折的 {field} 全部缺失,無法補值")
median = float(values.median())
imputed = values.fillna(median).to_numpy(dtype=float)
mean = float(imputed.mean())
std = float(imputed.std(ddof=0))
if not math.isfinite(std) or std == 0:
std = 1.0
numeric[field] = {"median": median, "mean": mean, "std": std}
feature_names.extend([field, f"{field}__missing"])
categories: JsonObject = {}
for field in categorical_fields:
observed = sorted(
{
str(value)
for value in frame[field].dropna().tolist()
}
)
categories[field] = observed
feature_names.extend(f"{field}=={value}" for value in observed)
return {
"numeric": numeric,
"categories": categories,
"feature_names": feature_names,
"fit_row_count": len(frame),
}
def transform_structured_features(frame: pd.DataFrame, state: Mapping[str, Any]) -> np.ndarray:
"""Apply a fitted fold-local state without learning from validation rows."""
columns: list[np.ndarray] = []
for field, policy in state["numeric"].items():
values = pd.to_numeric(frame[field], errors="coerce")
missing = values.isna().to_numpy(dtype=float)
imputed = values.fillna(float(policy["median"])).to_numpy(dtype=float)
scaled = (imputed - float(policy["mean"])) / float(policy["std"])
columns.extend([scaled, missing])
for field, categories in state["categories"].items():
values = frame[field].map(lambda value: str(value) if pd.notna(value) else None)
columns.extend(values.eq(category).to_numpy(dtype=float) for category in categories)
matrix = np.column_stack(columns) if columns else np.empty((len(frame), 0))
if matrix.shape[1] != len(state["feature_names"]):
raise BaselineContractError("特徵矩陣寬度與前處理狀態不一致")
return matrix
def _fit_ridge(x: np.ndarray, y: np.ndarray, alpha: float) -> np.ndarray:
design = np.column_stack([np.ones(len(x)), x])
penalty = np.eye(design.shape[1]) * alpha
penalty[0, 0] = 0.0
return np.linalg.solve(design.T @ design + penalty, design.T @ y)
def _predict_ridge(x: np.ndarray, coefficients: np.ndarray, levels: list[int]) -> tuple[np.ndarray, np.ndarray]:
continuous = np.column_stack([np.ones(len(x)), x]) @ coefficients
predicted = np.floor(continuous + 0.5).astype(int)
predicted = np.clip(predicted, min(levels), max(levels))
return continuous, predicted
def _majority_level(values: np.ndarray, levels: list[int]) -> int:
counts = Counter(int(value) for value in values)
return min(levels, key=lambda level: (-counts[level], level))
def classification_metrics(
reference: np.ndarray | list[int], predicted: np.ndarray | list[int], levels: list[int]
) -> JsonObject:
"""Return ordinal safety and classification metrics for answered records."""
y_true = np.asarray(reference, dtype=int)
y_pred = np.asarray(predicted, dtype=int)
if len(y_true) != len(y_pred) or len(y_true) == 0:
raise BaselineContractError("分類指標需要等長且非空的 reference 與 predicted")
if not set(y_true).issubset(levels) or not set(y_pred).issubset(levels):
raise BaselineContractError("指標輸入包含契約外級數")
confusion = np.zeros((len(levels), len(levels)), dtype=int)
level_to_index = {level: index for index, level in enumerate(levels)}
for actual, guess in zip(y_true, y_pred, strict=True):
confusion[level_to_index[int(actual)], level_to_index[int(guess)]] += 1
recalls: dict[str, float | None] = {}
f1_values: list[float] = []
for level in levels:
index = level_to_index[level]
tp = int(confusion[index, index])
actual_n = int(confusion[index, :].sum())
predicted_n = int(confusion[:, index].sum())
recalls[str(level)] = _safe_rate(tp, actual_n)
denominator = 2 * tp + (predicted_n - tp) + (actual_n - tp)
f1_values.append((2 * tp / denominator) if denominator else 0.0)
under = int((y_pred > y_true).sum())
over = int((y_pred < y_true).sum())
severe_mask = np.isin(y_true, [1, 2])
severe_under = int((severe_mask & (y_pred >= 3)).sum())
absolute_error = np.abs(y_pred - y_true)
return {
"answered_count": len(y_true),
"accuracy": round(float((y_true == y_pred).mean()), 6),
"balanced_accuracy": round(
sum(value or 0.0 for value in recalls.values()) / len(levels), 6
),
"macro_f1": round(sum(f1_values) / len(f1_values), 6),
"mean_absolute_error": round(float(absolute_error.mean()), 6),
"undertriage_rate": _safe_rate(under, len(y_true)),
"overtriage_rate": _safe_rate(over, len(y_true)),
"severe_undertriage_rate": _safe_rate(severe_under, int(severe_mask.sum())),
"two_or_more_level_error_rate": _safe_rate(int((absolute_error >= 2).sum()), len(y_true)),
"per_class_recall": recalls,
"confusion_matrix_rows_reference_columns_prediction": confusion.tolist(),
}
def _summarize_repeat_metrics(metrics: list[JsonObject], levels: list[int]) -> JsonObject:
fields = [
"accuracy",
"balanced_accuracy",
"macro_f1",
"mean_absolute_error",
"undertriage_rate",
"overtriage_rate",
"severe_undertriage_rate",
"two_or_more_level_error_rate",
]
summary: JsonObject = {}
for field in fields:
values = [float(item[field]) for item in metrics]
summary[field] = {
"mean": round(float(np.mean(values)), 6),
"minimum": round(min(values), 6),
"maximum": round(max(values), 6),
}
summary["per_class_recall"] = {}
for level in levels:
values = [float(item["per_class_recall"][str(level)] or 0.0) for item in metrics]
summary["per_class_recall"][str(level)] = {
"mean": round(float(np.mean(values)), 6),
"minimum": round(min(values), 6),
"maximum": round(max(values), 6),
}
combined = np.sum(
[np.asarray(item["confusion_matrix_rows_reference_columns_prediction"]) for item in metrics],
axis=0,
)
summary["combined_confusion_matrix_rows_reference_columns_prediction"] = combined.tolist()
return summary
def run_day26_oof(
model_inputs: pd.DataFrame,
labels: pd.DataFrame,
assignments: pd.DataFrame,
contract: Mapping[str, Any],
) -> tuple[JsonObject, JsonObject]:
"""Execute B0, fail-closed B1, and fold-local B2 across every locked fold."""
data = contract["data_contract"]
row_key = str(data["row_link_key"])
reference_field = str(data["reference_label_field"])
human_field = str(data["human_reference_field"])
levels = [int(value) for value in data["valid_levels"]]
b2 = contract["executed_baselines"]["B2"]
numeric = list(b2["numeric_fields"])
categorical = list(b2["categorical_fields"])
alpha = float(b2["model"]["ridge_alpha"])
inputs = model_inputs.set_index(row_key, drop=False)
target = labels.set_index(row_key)[reference_field].astype(int)
oof_rows: list[JsonObject] = []
fold_audit: list[JsonObject] = []
repeat_results: list[JsonObject] = []
for repeat_id in sorted(assignments["repeat_id"].unique()):
repeat_assignment = assignments.loc[assignments["repeat_id"].eq(repeat_id)]
repeat_oof: list[JsonObject] = []
for fold_id in sorted(repeat_assignment["fold_id"].unique()):
validation_ids = repeat_assignment.loc[
repeat_assignment["fold_id"].eq(fold_id), row_key
].tolist()
training_ids = repeat_assignment.loc[
~repeat_assignment["fold_id"].eq(fold_id), row_key
].tolist()
train_x = inputs.loc[training_ids]
valid_x = inputs.loc[validation_ids]
train_y = target.loc[training_ids].to_numpy(dtype=int)
valid_y = target.loc[validation_ids].to_numpy(dtype=int)
majority = _majority_level(train_y, levels)
b0_prediction = np.full(len(validation_ids), majority, dtype=int)
state = fit_structured_preprocessor(train_x, numeric, categorical)
x_train = transform_structured_features(train_x, state)
x_valid = transform_structured_features(valid_x, state)
coefficients = _fit_ridge(x_train, train_y.astype(float), alpha)
continuous, b2_prediction = _predict_ridge(x_valid, coefficients, levels)
fold_audit.append(
{
"repeat_id": int(repeat_id),
"fold_id": int(fold_id),
"training_count": len(training_ids),
"validation_count": len(validation_ids),
"b0_training_majority_level": majority,
"b2_feature_count": x_train.shape[1],
"b2_preprocessor_fit_scope": "training_fold_only",
"b2_coefficient_count_including_intercept": len(coefficients),
}
)
for position, record_index in enumerate(validation_ids):
row = {
row_key: int(record_index),
"repeat_id": int(repeat_id),
"fold_id": int(fold_id),
"reference_level": int(valid_y[position]),
"B0_predicted_level": int(b0_prediction[position]),
"B1_decision_status": "abstained",
"B1_abstention_reason_code": contract["executed_baselines"]["B1"]["abstention_reason_code"],
"B2_continuous_score": round(float(continuous[position]), 8),
"B2_predicted_level": int(b2_prediction[position]),
}
repeat_oof.append(row)
oof_rows.append(row)
repeat_oof.sort(key=lambda row: row[row_key])
y_true = [row["reference_level"] for row in repeat_oof]
repeat_results.append(
{
"repeat_id": int(repeat_id),
"record_count": len(repeat_oof),
"B0": classification_metrics(
y_true, [row["B0_predicted_level"] for row in repeat_oof], levels
),
"B1": {
"answered_count": 0,
"abstained_count": len(repeat_oof),
"coverage": 0.0,
"abstention_rate": 1.0,
"classification_metrics": None,
"reason_counts": {
contract["executed_baselines"]["B1"]["abstention_reason_code"]: len(repeat_oof)
},
},
"B2": classification_metrics(
y_true, [row["B2_predicted_level"] for row in repeat_oof], levels
),
}
)
human_metrics = classification_metrics(
labels[reference_field].astype(int).tolist(),
labels[human_field].astype(int).tolist(),
levels,
)
public = {
"schema_version": 1,
"experiment_id": contract["experiment_id"],
"scope": contract["scope"],
"data_and_split_audit": {
"record_count": len(model_inputs),
"repeat_count": int(data["repeat_count"]),
"fold_count": int(data["fold_count"]),
"out_of_fold_prediction_count_per_answering_baseline": len(oof_rows),
"split_unit": "record_index",
"patient_identifier_available": False,
"patient_level_independence_claim_allowed": False,
},
"executed_baselines": {
"B0": {
"name": contract["executed_baselines"]["B0"]["name"],
"coverage": 1.0,
"repeat_summary": _summarize_repeat_metrics(
[item["B0"] for item in repeat_results], levels
),
},
"B1": {
"name": contract["executed_baselines"]["B1"]["name"],
"coverage": 0.0,
"abstention_rate": 1.0,
"classification_metrics": None,
"abstention_reason_code": contract["executed_baselines"]["B1"]["abstention_reason_code"],
"reason_counts_over_all_oof_prediction_slots": {
contract["executed_baselines"]["B1"]["abstention_reason_code"]: len(oof_rows)
},
"required_gap_ids": contract["executed_baselines"]["B1"]["required_gap_ids"],
},
"B2": {
"name": contract["executed_baselines"]["B2"]["name"],
"coverage": 1.0,
"repeat_summary": _summarize_repeat_metrics(
[item["B2"] for item in repeat_results], levels
),
},
},
"per_repeat_metrics": repeat_results,
"human_reference_context": {
"id": human_field,
"role": contract["human_reference_context"]["role"],
"single_observed_comparison_not_cross_validated": True,
"metrics_against_KTAS_expert": human_metrics,
},
"comparison_registry": contract["comparison_registry"],
"interpretation_boundaries": {
"descriptive_oof_metrics_available": True,
"confidence_intervals_computed": False,
"confirmatory_inference_eligible": False,
"model_selection_performed": False,
"clinical_safety_or_deployment_claim_allowed": False,
},
"limitations": contract["limitations"],
}
private = {
"schema_version": 1,
"experiment_id": contract["experiment_id"],
"fold_audit": fold_audit,
"oof_predictions": sorted(oof_rows, key=lambda row: (row["repeat_id"], row[row_key])),
}
return public, private
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
scripts/run_day26_baselines.py核對所有上游雜湊,執行 25 個 folds,寫出公開聚合、gitignored 逐筆 OOF 結果與 run manifest。
請在文字編輯器建立 scripts/run_day26_baselines.py,貼入以下完整內容並儲存:
#!/usr/bin/env python3
"""Execute the locked Day 26 out-of-fold simple baselines."""
from __future__ import annotations
import argparse
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path
import pandas as pd
from triage_rag.baselines.simple import run_day26_oof, validate_day26_inputs
from triage_rag.reproducibility import (
canonical_json_bytes,
file_record,
git_state,
installed_versions,
load_json,
sha256_bytes,
sha256_file,
write_json,
)
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = "configs/baselines/day-26-fair-baselines.json"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="執行 Day 26 B0、B1 與 B2 的固定折外基準。"
)
parser.add_argument("--contract", default=DEFAULT_CONTRACT)
return parser.parse_args()
def _validate_source_hashes(contract: dict) -> list[str]:
source_paths: list[str] = []
for key, relative_path in contract["sources"].items():
if not key.endswith("_path"):
continue
hash_key = f"{key[:-5]}_sha256"
if hash_key not in contract["sources"]:
raise ValueError(f"來源 {key} 缺少配對的 SHA-256")
path = PROJECT_ROOT / relative_path
observed = sha256_file(path)
expected = contract["sources"][hash_key]
if observed != expected:
raise ValueError(f"{relative_path} SHA-256 不符:{observed}")
source_paths.append(relative_path)
return source_paths
def main() -> int:
args = parse_args()
contract_path = PROJECT_ROOT / args.contract
contract = load_json(contract_path)
if contract.get("schema_version") != 1:
raise ValueError("目前只支援 Day 26 schema_version=1")
source_paths = _validate_source_hashes(contract)
sources = contract["sources"]
model_inputs = pd.read_csv(PROJECT_ROOT / sources["model_inputs_path"])
labels = pd.read_csv(PROJECT_ROOT / sources["reference_labels_path"])
assignments = pd.read_csv(PROJECT_ROOT / sources["split_assignments_path"])
day16 = load_json(PROJECT_ROOT / sources["day16_coverage_path"])
validation = validate_day26_inputs(
model_inputs, labels, assignments, contract, day16
)
public, private = run_day26_oof(model_inputs, labels, assignments, contract)
public["contract_validation"] = validation
public_sha256 = sha256_bytes(canonical_json_bytes(public))
public_path = PROJECT_ROOT / contract["outputs"]["public_summary_path"]
write_json(public_path, public)
started_at = datetime.now(timezone.utc)
run_id = f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-{public_sha256[:8]}"
run_directory = PROJECT_ROOT / contract["outputs"]["run_output_root"] / run_id
run_directory.mkdir(parents=True, exist_ok=False)
private_path = run_directory / contract["outputs"]["full_result_filename"]
write_json(private_path, private)
tracked_inputs = [args.contract, *source_paths]
manifest = {
"manifest_schema_version": 1,
"run_id": run_id,
"experiment_id": contract["experiment_id"],
"started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
"command": [sys.executable, *sys.argv],
"documented_command": [
"poetry",
"run",
"python",
"scripts/run_day26_baselines.py",
],
"git": git_state(PROJECT_ROOT),
"runtime": {
"python": platform.python_version(),
"packages": installed_versions(["numpy", "pandas"]),
},
"inputs": [file_record(PROJECT_ROOT, path) for path in tracked_inputs],
"parameters": {
"executed_baseline_ids": ["B0", "B1", "B2"],
"ridge_alpha": contract["executed_baselines"]["B2"]["model"]["ridge_alpha"],
"repeat_count": contract["data_contract"]["repeat_count"],
"fold_count": contract["data_contract"]["fold_count"],
},
"outputs": [
file_record(PROJECT_ROOT, str(public_path.relative_to(PROJECT_ROOT))),
file_record(PROJECT_ROOT, str(private_path.relative_to(PROJECT_ROOT))),
],
"privacy": "公開檔只含聚合與每次重複指標;6335 筆折外預測留在 gitignored results/runs/day-26/。",
"scope": contract["scope"],
}
manifest_path = run_directory / "run-manifest.json"
write_json(manifest_path, manifest)
b0 = public["executed_baselines"]["B0"]["repeat_summary"]
b2 = public["executed_baselines"]["B2"]["repeat_summary"]
print("Day 26 公平基準:完成")
print(f"資料與切分:{len(model_inputs):,} 筆 × 5 次重複五折")
print(f"B0 平均 accuracy:{b0['accuracy']['mean']:.6f}")
print(f"B1 coverage:{public['executed_baselines']['B1']['coverage']:.1f}")
print(f"B2 平均 accuracy:{b2['accuracy']['mean']:.6f}")
print(f"B2 平均 MAE:{b2['mean_absolute_error']['mean']:.6f}")
print(f"公開摘要:{public_path.relative_to(PROJECT_ROOT)}")
print(f"本機完整結果:{private_path.relative_to(PROJECT_ROOT)}")
print(f"穩定公開摘要:{public_sha256}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/test_simple_baselines.py以十項測試驗證折內補值、未知類別、序位方向、資料對齊、知識缺口、6,335 筆完整性與重跑一致。
請在文字編輯器建立 tests/test_simple_baselines.py,貼入以下完整內容並儲存:
"""Tests for Day 26 leakage-aware simple baselines."""
from __future__ import annotations
import copy
import json
import unittest
from pathlib import Path
import numpy as np
import pandas as pd
from triage_rag.baselines.simple import (
BaselineContractError,
classification_metrics,
fit_structured_preprocessor,
run_day26_oof,
transform_structured_features,
validate_day26_inputs,
)
ROOT = Path(__file__).resolve().parents[1]
def load_json(path: str) -> dict:
with (ROOT / path).open(encoding="utf-8") as handle:
return json.load(handle)
class Day26SimpleBaselineTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.contract = load_json("configs/baselines/day-26-fair-baselines.json")
sources = cls.contract["sources"]
cls.model_inputs = pd.read_csv(ROOT / sources["model_inputs_path"])
cls.labels = pd.read_csv(ROOT / sources["reference_labels_path"])
cls.assignments = pd.read_csv(ROOT / sources["split_assignments_path"])
cls.day16 = load_json(sources["day16_coverage_path"])
def test_preprocessor_uses_training_median_and_missing_indicator(self) -> None:
train = pd.DataFrame({"Age": [10.0, np.nan, 30.0], "Sex": [1, 2, 1]})
state = fit_structured_preprocessor(train, ["Age"], ["Sex"])
self.assertEqual(state["numeric"]["Age"]["median"], 20.0)
transformed = transform_structured_features(
pd.DataFrame({"Age": [np.nan], "Sex": [3]}), state
)
self.assertEqual(transformed.shape, (1, 4))
self.assertAlmostEqual(transformed[0, 0], 0.0)
self.assertAlmostEqual(transformed[0, 1], 1.0)
self.assertEqual(transformed[0, 2:].tolist(), [0.0, 0.0])
def test_preprocessor_rejects_all_missing_numeric_training_field(self) -> None:
frame = pd.DataFrame({"Age": [np.nan], "Sex": [1]})
with self.assertRaisesRegex(BaselineContractError, "全部缺失"):
fit_structured_preprocessor(frame, ["Age"], ["Sex"])
def test_classification_metrics_follow_ordinal_direction(self) -> None:
metrics = classification_metrics(
[1, 2, 3, 4, 5], [3, 2, 2, 5, 5], [1, 2, 3, 4, 5]
)
self.assertEqual(metrics["accuracy"], 0.4)
self.assertEqual(metrics["undertriage_rate"], 0.4)
self.assertEqual(metrics["overtriage_rate"], 0.2)
self.assertEqual(metrics["severe_undertriage_rate"], 0.5)
self.assertEqual(metrics["two_or_more_level_error_rate"], 0.2)
def test_classification_metrics_reject_empty_input(self) -> None:
with self.assertRaisesRegex(BaselineContractError, "非空"):
classification_metrics([], [], [1, 2, 3, 4, 5])
def test_real_inputs_pass_locked_contract(self) -> None:
validation = validate_day26_inputs(
self.model_inputs,
self.labels,
self.assignments,
self.contract,
self.day16,
)
self.assertEqual(validation["record_count"], 1267)
self.assertEqual(validation["assignment_row_count"], 6335)
self.assertTrue(validation["forbidden_fields_absent_from_b2"])
def test_duplicate_record_index_is_rejected(self) -> None:
broken = self.model_inputs.copy()
broken.loc[1, "record_index"] = broken.loc[0, "record_index"]
with self.assertRaisesRegex(BaselineContractError, "完整且唯一"):
validate_day26_inputs(
broken, self.labels, self.assignments, self.contract, self.day16
)
def test_missing_assignment_is_rejected(self) -> None:
broken = self.assignments.iloc[:-1].copy()
with self.assertRaisesRegex(BaselineContractError, "assignment 筆數"):
validate_day26_inputs(
self.model_inputs, self.labels, broken, self.contract, self.day16
)
def test_unconfirmed_rule_gap_is_rejected(self) -> None:
broken = copy.deepcopy(self.day16)
target = next(
item
for item in broken["coverage"]["items"]
if item["coverage_id"] == "adult_chief_complaint_catalog"
)
target["status"] = "covered"
with self.assertRaisesRegex(BaselineContractError, "必要知識缺口"):
validate_day26_inputs(
self.model_inputs,
self.labels,
self.assignments,
self.contract,
broken,
)
def test_real_oof_run_is_complete_and_deterministic(self) -> None:
public_a, private_a = run_day26_oof(
self.model_inputs, self.labels, self.assignments, self.contract
)
public_b, private_b = run_day26_oof(
self.model_inputs, self.labels, self.assignments, self.contract
)
self.assertEqual(public_a, public_b)
self.assertEqual(private_a, private_b)
self.assertEqual(len(private_a["oof_predictions"]), 6335)
self.assertEqual(len(private_a["fold_audit"]), 25)
self.assertEqual(public_a["executed_baselines"]["B1"]["coverage"], 0.0)
self.assertEqual(
public_a["executed_baselines"]["B1"][
"reason_counts_over_all_oof_prediction_slots"
]["incomplete_public_rule_coverage"],
6335,
)
self.assertAlmostEqual(
public_a["executed_baselines"]["B0"]["repeat_summary"]["accuracy"]["mean"],
0.384373,
)
self.assertAlmostEqual(
public_a["executed_baselines"]["B2"]["repeat_summary"]["accuracy"]["mean"],
0.537648,
)
def test_human_reference_is_context_not_oof_input(self) -> None:
public, _ = run_day26_oof(
self.model_inputs, self.labels, self.assignments, self.contract
)
human = public["human_reference_context"]
self.assertTrue(human["single_observed_comparison_not_cross_validated"])
self.assertAlmostEqual(
human["metrics_against_KTAS_expert"]["accuracy"], 0.853197
)
self.assertNotIn(
"KTAS_RN", self.contract["executed_baselines"]["B2"]["numeric_fields"]
)
self.assertNotIn(
"KTAS_RN",
self.contract["executed_baselines"]["B2"]["categorical_fields"],
)
if __name__ == "__main__":
unittest.main()
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
scripts/figures/day-26/generate_day26_figures.py使用 Pillow 直接讀取 Day 26 設定與公開結果,重建兩張 1920×1080 繁中技術圖,不手動抄寫結果數字。
請在文字編輯器建立 scripts/figures/day-26/generate_day26_figures.py,貼入以下完整內容並儲存:
#!/usr/bin/env python3
"""Rebuild Day 26 fold-local pipeline and exact-result figures with Pillow."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from PIL import Image, ImageDraw, ImageFont
ROOT = Path(__file__).resolve().parents[3]
CONFIG_PATH = ROOT / "configs/baselines/day-26-fair-baselines.json"
RESULT_PATH = ROOT / "results/public/day-26-simple-baselines.json"
OUTPUT_DIRECTORY = ROOT / "articles/assets/day-26"
FONT_PATH = Path("/System/Library/Fonts/STHeiti Medium.ttc")
WIDTH, HEIGHT = 1920, 1080
NAVY = "#102A43"
BLUE = "#2563A7"
TEAL = "#168C8C"
ORANGE = "#E78A32"
RED = "#C95056"
PALE_BLUE = "#EAF2FA"
PALE_TEAL = "#E7F5F3"
PALE_ORANGE = "#FFF1DF"
PALE_RED = "#FBEAEC"
WHITE = "#FFFFFF"
INK = "#25384A"
MUTED = "#5F7182"
GRID = "#CDD9E5"
def load_json(path: Path) -> dict[str, Any]:
with path.open(encoding="utf-8") as handle:
return json.load(handle)
def font(size: int) -> ImageFont.FreeTypeFont:
return ImageFont.truetype(str(FONT_PATH), size=size)
def canvas() -> tuple[Image.Image, ImageDraw.ImageDraw]:
image = Image.new("RGB", (WIDTH, HEIGHT), "#F7FAFC")
return image, ImageDraw.Draw(image)
def title(draw: ImageDraw.ImageDraw, heading: str, subtitle: str) -> None:
draw.text((90, 64), heading, fill=NAVY, font=font(54))
draw.text((92, 137), subtitle, fill=MUTED, font=font(28))
draw.rounded_rectangle((90, 190, 1830, 196), radius=3, fill=TEAL)
def centered(draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], text: str, *, size: int, fill: str = INK) -> None:
bounds = draw.textbbox((0, 0), text, font=font(size))
x = box[0] + (box[2] - box[0] - (bounds[2] - bounds[0])) / 2
y = box[1] + (box[3] - box[1] - (bounds[3] - bounds[1])) / 2
draw.text((x, y), text, fill=fill, font=font(size))
def arrow(draw: ImageDraw.ImageDraw, start: tuple[int, int], end: tuple[int, int], color: str = BLUE) -> None:
draw.line([start, end], fill=color, width=10)
x, y = end
draw.polygon([(x, y), (x - 22, y - 17), (x - 22, y + 17)], fill=color)
def rounded_box(draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], *, fill: str, outline: str) -> None:
draw.rounded_rectangle(box, radius=28, fill=fill, outline=outline, width=4)
def draw_pipeline(config: dict[str, Any]) -> Image.Image:
image, draw = canvas()
repeats = config["data_contract"]["repeat_count"]
folds = config["data_contract"]["fold_count"]
title(
draw,
"相同切分,前