Day 13 建立了可以重跑的專案環境:套件版本已有鎖定檔,設定、程式、測試與執行結果也各自有固定位置。現在環境可以重現了,下一個問題是:程式究竟允許哪一筆紀錄、哪一個欄位進入模型?
想像一位病患剛抵達急診,當下已知主訴、到院方式、意識反應與生命徵象。幾個小時後,同一份資料庫又多了急診診斷、處置、住院去向與停留時間。這些欄位在資料表中並排出現,電腦不會自動知道哪些資訊來自未來。
如果我們直接把整列資料丟給模型,模型可能利用「住進加護病房」或「急診停留很久」猜測病患很嚴重。離線分數看起來會很好,真正的初次檢傷現場卻不可能先知道這些結果。
下圖是概念示意。閱讀時請看中央的檢查關卡:左側是初次檢傷期間可能取得的資訊,右側則是必須留在決策時間點之後的診斷、處置、去向與時間結果。

上圖要表達的不是「越早產生的欄位都能使用」。答案本身、人工比較基準與容易形成場域捷徑的欄位,即使在時間上很接近檢傷決策,也可能不該進入模型。今天要建立的就是同時考慮時間與資料角色的雙重邊界。
本篇會沿用韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)公開資料,但不會訓練模型,也不需要啟動本機大型語言模型執行工具 Ollama 或下載任何模型。今天的成果是一份可由程式執行的資料契約,以及實際通過 1,267 筆公開資料的驗證結果。
讀完並跟著操作後,你會完成以下四件事:
本篇的新名詞較多,先建立一份快速索引。詞彙表只提供方向,後面的正文仍會用實際資料解釋每個名詞。
| 中文名稱 | 英文全名/縮寫 | 本篇用途 |
|---|---|---|
| 研究群體 | Cohort | 說明哪些候選紀錄經過哪些條件後,成為本次分析資料 |
| 急診就醫事件 | Emergency Department Encounter, ED Encounter | 一次急診到院及其初次檢傷紀錄;不保證等於一位唯一病患 |
| 資料結構契約 | Data Schema,本文後續簡稱 Schema | 固定欄位名稱、角色、型別、可否缺失與允許值 |
| 索引時間 | Index Time | 模型理論上要做出預測的時間點 |
| 時間截點 | Temporal Cutoff | 規定只可使用索引時間以前已知的資訊 |
| 預測變數 | Predictor | 模型真正可以讀取的輸入欄位,也常被稱為特徵 |
| 參考標籤 | Reference Label | 用來比較模型預測的目標答案,本系列使用專家重新判定級數 |
| 資料洩漏 | Data Leakage | 模型取得真實使用情境中不該知道的答案、未來資訊或評估資料 |
| 資料契約 | Data Contract | 把 cohort、時間點、Schema、欄位角色與驗證規則寫成可審查且可執行的約定 |
研究群體(Cohort)是符合一組明確納入與排除條件的紀錄集合。建立 cohort 時不能只說「我們有 1,267 筆資料」,還要回答資料從哪裡來、排除了什麼,以及最後每一列代表什麼。
本系列把研究單位定義為「一筆符合條件的急診初次檢傷紀錄」,也就是一次急診就醫事件(Emergency Department Encounter, ED Encounter)的公開紀錄。這個定義刻意不寫成「一位病患」,因為 Kaggle 公開資料沒有可用的病患識別碼。我們無法確認同一個人是否在不同日期重複就醫。
這個限制會影響後續資料切分。如果同一位病患真的出現多次,而這些紀錄被分到訓練集與測試集,模型可能記住同一人的模式。目前資料無法直接排除這個風險,所以 Day 15 只能建立紀錄層級切分,不能宣稱已完成病患層級切分。
公開來源記載,資料來自兩所都市教學醫院,在 2016 年 10 月至 2017 年 9 月之間隨機選取 20 天,蒐集年齡大於 15 歲的急診紀錄。最初共有 1,540 筆候選紀錄,之後排除三類資料:檢傷作業時間大於或等於 30 分鐘 63 筆、檢傷前取消掛號 119 筆,以及資料不足以供專家重新判定 91 筆,最後留下 1,267 筆。Moon 等人:KTAS 檢傷準確性與錯誤原因
Day 12 取得的 data.csv 是 Kaggle 提供的逗號分隔值(Comma-Separated Values, CSV)資料檔。雖然名稱是 CSV,這份來源實際使用分號分隔欄位,讀取方式已由 Day 12 的資料卡與管線鎖定。
下圖依照公開來源提供的排除數量逐步核算。每個方框下方的剩餘筆數都由 configs/data/day-14-ktas-data-contract.json 自動讀取,不是繪圖時手動輸入。

上圖也標出了本系列能驗證的邊界。Kaggle 公開的 data.csv 已經是最終 1,267 筆資料,不包含被排除的 273 筆原始紀錄。因此,我們可以核對公開來源的算術,也能驗證 CSV 的 1,267 筆與 24 個欄位,卻不能假裝能逐列重新執行原始排除流程。
這裡還有一個容易忽略的細節:「年齡大於 15 歲」表示實際納入值從 16 歲開始,不是大於或等於 15 歲。本篇的契約因此把 Age 設為必須大於 15,而不是大於或等於 15。
索引時間(Index Time)是模型應該做出預測的時刻。本系列把索引時間定義為「初次檢傷資料蒐集完成、級數即將被決定的時刻」。模型只能模擬這一刻能做的事,不能借用後續急診流程才產生的資訊。
時間截點(Temporal Cutoff)則是資料選擇規則:候選輸入必須在索引時間以前已經可得。以同一筆呼吸困難案例來看,主訴、呼吸率每分鐘 28 次與血氧飽和度 91% 可以是候選輸入;急診診斷、住院去向與完整停留時間則尚未發生。
然而,「時間上已經存在」仍然不是充分條件。急診護理師登錄的 KTAS_RN 在初次檢傷完成時就會存在,但它是我們要比較的人類基準。把它放進模型,模型只要複製護理師級數就能得到漂亮結果,因此仍屬資料洩漏。
同理,record_index 雖然存在於處理後檔案,也不能成為預測變數。它只是把候選輸入、稽核資料與參考標籤的同一列重新對齊,沒有臨床意義。
資料洩漏(Data Leakage)是模型在訓練、調整或測試時,看到真實使用情境中不該取得的資訊。資料洩漏不只包含答案欄位,也包含檢傷後結果、由答案計算出的欄位,以及可能讓模型走捷徑的稽核資訊。
本篇沿用 Day 11 已核對的 24 個來源欄位,並加上明確時間階段與輸入權限。六種角色如下:
候選輸入中的生命徵象包括收縮壓(Systolic Blood Pressure, SBP)、舒張壓(Diastolic Blood Pressure, DBP)、心率(Heart Rate, HR)、呼吸率(Respiratory Rate, RR)、體溫(Body Temperature, BT)與來源欄位 Saturation 所表示的血氧飽和度。疼痛欄位則把是否疼痛的 Pain,與疼痛數字評定量表(Numeric Rating Scale, NRS)的 NRS_pain 分開保存。
KTAS_RN 中的 RN 源自註冊護理師(Registered Nurse, RN)。本系列把這一欄保留為人類比較基準;KTAS_expert 是專家根據初始紀錄重新判定的主要參考標籤。兩欄都不是病患輸入,也不能互稱為模型答案。
| 資料角色 | 欄位數 | 實際欄位 | 是否進入預設模型輸入 |
|---|---|---|---|
| 候選模型輸入 | 14 | Sex、Age、Arrival mode、Injury、Chief_complain、Mental、Pain、NRS_pain、SBP、DBP、HR、RR、BT、Saturation |
可以,但仍要通過型別、缺失與值域檢查 |
| 稽核或分層專用 | 2 | Group、Patients number per hour |
不進預設模型,只用於場域與壅塞分層 |
| 人類比較基準 | 1 | KTAS_RN |
不可以 |
| 主要參考標籤 | 1 | KTAS_expert |
不可以 |
| 檢傷後結果 | 4 | Diagnosis in ED、Disposition、Length of stay_min、KTAS duration_min |
不可以 |
| 比較後衍生欄位 | 2 | Error_group、mistriage |
不可以 |
下圖將六種角色放回時間線。閱讀時先找到中央紅色的初次檢傷決策線,再看每一類欄位為什麼停在不同位置。

上圖左上方的 14 欄是預設模型唯一能讀取的臨床候選值。下方兩個稽核欄位雖然可能在當時存在,仍先隔離以免模型記住醫院場域或不清楚的壅塞計算方式。右側的標籤、事後結果與比較後衍生欄位則一律禁止進入模型。
這種設計也符合臨床預測模型透明報告指引加人工智慧擴充版(Transparent Reporting of a Multivariable Prediction Model for Individual Prognosis or Diagnosis Plus Artificial Intelligence, TRIPOD+AI)的核心精神:研究必須清楚說明預測變數、結果標籤、資料來源與模型預計使用的情境,讓讀者能判斷評估是否對應真實使用條件。Collins 等人:TRIPOD+AI statement
資料結構契約(Data Schema,本文後續簡稱 Schema)規定一張表應該有哪些欄位、欄位順序、型別、允許值與缺失條件。資料契約(Data Contract)比 Schema 再多一層:它也記錄 cohort、索引時間、欄位角色、輸出範圍與限制。
本篇接下來會建立 configs/data/day-14-ktas-data-contract.json。configs/ 是保存可審查規則與參數的資料夾,data/ 子目錄則集中資料來源、欄位政策與資料契約。.json 表示檔案使用 JavaScript 物件表示法(JavaScript Object Notation, JSON)。
這份契約會接收或引用以下內容:
以下是檔案中的核心節錄。這段不是另一份可以單獨執行的 JSON;為了聚焦時間邊界,省略了 cohort 說明、完整欄位清單與值域細節。正文前面的表格已完整列出六種角色與 24 個欄位。
{
"prediction_task": {
"index_time_name": "初次檢傷決策時間點",
"allowed_feature_window": "只允許在初次檢傷決策前已知,而且不是答案或答案衍生物的欄位",
"primary_reference_label": "KTAS_expert",
"human_reference": "KTAS_RN"
},
"role_contract": {
"model_input_candidate": {
"temporal_phase": "at_or_before_initial_triage_decision",
"allowed_in_model_input": true
},
"post_triage_outcome": {
"temporal_phase": "after_initial_triage_decision",
"allowed_in_model_input": false
}
}
}
這段設定把「時間」與「角色」分開表示。temporal_phase 說明欄位何時形成,allowed_in_model_input 則是最終權限。兩個條件都通過,欄位才有資格進入預設模型輸入。
可重複使用的驗證邏輯位於 src/triage_rag/data/contracts.py。src/ 是存放核心 Python 套件的資料夾,triage_rag/data/ 子目錄專門處理資料契約。這支程式會讀取契約、欄位政策與 Day 12 的處理後資料,回傳只含筆數、欄位名稱、缺失數與檢查結果的聚合報告。
它依序檢查:
record_index 是否在三份表中依序對齊。excluded-fields.json 是否完整登錄禁止輸入欄位。下面是實際程式中負責最後一道洩漏檢查的片段。輸入是候選模型欄位與所有非模型欄位名稱;輸出是兩者的交集。只要交集不是空集合,程式就會停止,不會繼續產生通過報告。
actual_model_fields = [
name for name in model_inputs.columns if name != row_link_key
]
leaked = sorted(set(actual_model_fields).intersection(protected_fields))
if leaked:
raise ContractError(f"模型輸入出現受保護欄位:{leaked}")
這段檢查之前還有「欄位名稱與順序必須完全相同」的規則。也就是說,即使有人加入一個尚未登錄的新欄位,程式也會先以多餘欄位拒絕它,避免只靠永遠追不上新名稱的黑名單。
測試夾具(Test Fixture,下文簡稱 fixture)是自動化測試預先準備的固定輸入。tests/fixtures/day-14-prepared/ 是保存本篇 fixture 的資料夾,裡面有三筆純合成急診案例,並模擬 Day 12 產生的四個檔案:候選輸入、稽核欄位、參考標籤及禁止欄位清單。這些資料只驗證程式行為,不是 KTAS 規則範例,也不計入模型成績。
自動化測試位於 tests/test_data_contract.py。它執行五項檢查:
KTAS_expert 偷放進模型輸入時必須失敗。RR 欄位時必須失敗。Disposition 時必須失敗。測試不只確認「正確資料能通過」,也故意建立錯誤資料,確認程式真的會阻擋。如果測試只走成功路徑,我們無法知道黑名單是否只是看起來存在。
接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾,依下列順序建立檔案;每個程式碼區塊都是該檔案的完整內容,不含省略號。
本篇沿用 Day 11 欄位政策、Day 12 三份處理後資料與 Day 13 的重現性工具;以下內容是 Day 14 新增的完整檔案。
先從專案根目錄建立需要的資料夾:
mkdir -p configs/data scripts src/triage_rag/data tests/fixtures/day-14-prepared
如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案,把對應區塊完整貼入後儲存。
configs/data/day-14-ktas-data-contract.json固定 cohort、索引時間、欄位角色、資料表結構、值域與輸出範圍。
請在文字編輯器建立 configs/data/day-14-ktas-data-contract.json,貼入以下完整內容並儲存:
{
"schema_version": 1,
"contract_id": "day-14-ktas-initial-triage-v1",
"status": "implemented-and-source-validated",
"source_policy_path": "configs/data/day-11-ktas-field-policy.json",
"dataset_card_path": "configs/data/day-12-kaggle-ktas-dataset-card.json",
"source_reference": "https://doi.org/10.1371/journal.pone.0216972",
"cohort": {
"unit_of_analysis": "一筆符合條件的急診初次檢傷紀錄,不等同一位唯一病患",
"candidate_record_count": 1540,
"candidate_definition": "兩所急診在隨機選取的 20 天內,年齡大於 15 歲的就醫紀錄",
"exclusions": [
{
"reason": "檢傷作業時間大於或等於 30 分鐘",
"removed": 63,
"remaining": 1477
},
{
"reason": "檢傷前取消掛號",
"removed": 119,
"remaining": 1358
},
{
"reason": "資料不足以供專家重新判定",
"removed": 91,
"remaining": 1267
}
],
"final_eligible_record_count": 1267,
"patient_identity_available": false,
"limitation": "公開資料沒有可用的病患識別碼,因此不能宣稱每列來自不同病患,也不能完成病患層級去重或切分。"
},
"prediction_task": {
"decision": "根據初次檢傷當下資訊,預測專家重新判定的 KTAS 五級序位標籤",
"index_time_name": "初次檢傷決策時間點",
"index_time_definition": "初次檢傷資料蒐集完成、級數即將被決定的時刻",
"allowed_feature_window": "只允許在初次檢傷決策前已知,而且不是答案或答案衍生物的欄位",
"primary_reference_label": "KTAS_expert",
"human_reference": "KTAS_RN",
"label_boundary_note": "KTAS_RN 即使在檢傷完成時已存在,仍是人類比較基準;KTAS_expert 是主要參考標籤,兩者都不得成為模型輸入。"
},
"role_contract": {
"model_input_candidate": {
"temporal_phase": "at_or_before_initial_triage_decision",
"allowed_in_model_input": true
},
"audit_or_stratification_only": {
"temporal_phase": "context_or_uncertain_window",
"allowed_in_model_input": false
},
"human_reference_only": {
"temporal_phase": "at_initial_triage_decision",
"allowed_in_model_input": false
},
"reference_label": {
"temporal_phase": "expert_review_after_source_record",
"allowed_in_model_input": false
},
"post_triage_outcome": {
"temporal_phase": "after_initial_triage_decision",
"allowed_in_model_input": false
},
"derived_evaluation_only": {
"temporal_phase": "after_label_comparison",
"allowed_in_model_input": false
}
},
"prepared_tables": {
"model_inputs": {
"filename": "model-input-candidates.csv",
"roles": [
"model_input_candidate"
]
},
"audit_strata": {
"filename": "audit-strata.csv",
"roles": [
"audit_or_stratification_only"
]
},
"reference_labels": {
"filename": "reference-labels.csv",
"roles": [
"human_reference_only",
"reference_label",
"derived_evaluation_only"
]
}
},
"excluded_fields_filename": "excluded-fields.json",
"row_link_key": "record_index",
"input_constraints": {
"required_non_null": [
"Sex",
"Age",
"Arrival mode",
"Injury",
"Chief_complain",
"Mental",
"Pain"
],
"numeric_fields": [
"Sex",
"Age",
"Arrival mode",
"Injury",
"Mental",
"Pain",
"NRS_pain",
"SBP",
"DBP",
"HR",
"RR",
"BT",
"Saturation"
],
"non_empty_text_fields": [
"Chief_complain"
],
"allowed_values": {
"Sex": [
1,
2
],
"Arrival mode": [
1,
2,
3,
4,
5,
6,
7
],
"Injury": [
1,
2
],
"Mental": [
1,
2,
3,
4
],
"Pain": [
0,
1
]
},
"numeric_bounds": {
"Age": {
"exclusive_minimum": 15
},
"NRS_pain": {
"minimum": 0,
"maximum": 10
}
}
},
"label_constraints": {
"required_non_null": [
"KTAS_RN",
"KTAS_expert",
"mistriage"
],
"allowed_values": {
"KTAS_RN": [
1,
2,
3,
4,
5
],
"KTAS_expert": [
1,
2,
3,
4,
5
],
"mistriage": [
0,
1,
2
]
}
},
"output_policy": {
"aggregate_only": true,
"scope": "data_contract_validation_not_model_evaluation",
"output_root": "results/runs/day-14"
}
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/data/__init__.py建立資料驗證子套件並公開契約驗證函式。
請在文字編輯器建立 src/triage_rag/data/__init__.py,貼入以下完整內容並儲存:
"""Data-contract helpers for the KTAS side project."""
from .contracts import ContractError, validate_prepared_contract
__all__ = ["ContractError", "validate_prepared_contract"]
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/data/contracts.py實作 cohort、欄位、型別、值域、對齊鍵與資料洩漏檢查。
請在文字編輯器建立 src/triage_rag/data/contracts.py,貼入以下完整內容並儲存:
"""Validate Day 14 cohort, schema, and leakage boundaries."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Mapping, Sequence
import pandas as pd
JsonObject = dict[str, Any]
class ContractError(ValueError):
"""Raised when prepared data violates the declared contract."""
def load_json(path: Path) -> Any:
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)
def fields_for_roles(policy: Mapping[str, Any], roles: Sequence[str]) -> list[str]:
selected = set(roles)
return [
str(field["source_name"])
for field in policy["fields"]
if field["role"] in selected
]
def validate_cohort_flow(contract: Mapping[str, Any]) -> JsonObject:
cohort = contract["cohort"]
candidate_count = int(cohort["candidate_record_count"])
remaining = candidate_count
total_removed = 0
for step in cohort["exclusions"]:
removed = int(step["removed"])
remaining -= removed
total_removed += removed
if remaining != int(step["remaining"]):
raise ContractError(
f"cohort 流程算術不一致:{step['reason']} 後應剩 {remaining} 筆,"
f"契約卻寫 {step['remaining']} 筆"
)
final_count = int(cohort["final_eligible_record_count"])
if remaining != final_count:
raise ContractError(
f"cohort 最終筆數不一致:依排除數量應為 {remaining},契約為 {final_count}"
)
return {
"candidate_record_count": candidate_count,
"total_excluded": total_removed,
"final_eligible_record_count": final_count,
"arithmetic_valid": True,
}
def validate_role_contract(
contract: Mapping[str, Any], policy: Mapping[str, Any]
) -> tuple[list[str], list[str], dict[str, list[str]]]:
role_contract = contract["role_contract"]
policy_roles = {str(field["role"]) for field in policy["fields"]}
missing_roles = sorted(policy_roles.difference(role_contract))
if missing_roles:
raise ContractError(f"role_contract 缺少欄位角色:{missing_roles}")
allowed_roles = {
role
for role, rule in role_contract.items()
if rule["allowed_in_model_input"] is True
}
if allowed_roles != {"model_input_candidate"}:
raise ContractError(
"模型輸入角色只能是 model_input_candidate,"
f"目前設定為 {sorted(allowed_roles)}"
)
grouped = {
role: fields_for_roles(policy, [role]) for role in sorted(policy_roles)
}
allowed_fields = grouped["model_input_candidate"]
protected_fields = [
name
for role, names in grouped.items()
if role != "model_input_candidate"
for name in names
]
return allowed_fields, protected_fields, grouped
def expected_columns(
contract: Mapping[str, Any],
policy: Mapping[str, Any],
table_name: str,
) -> list[str]:
table = contract["prepared_tables"][table_name]
return [str(contract["row_link_key"]), *fields_for_roles(policy, table["roles"])]
def validate_exact_columns(
frame: pd.DataFrame, expected: Sequence[str], table_name: str
) -> None:
actual = frame.columns.tolist()
if actual != list(expected):
missing = [name for name in expected if name not in actual]
extra = [name for name in actual if name not in expected]
raise ContractError(
f"{table_name} 欄位契約不符;缺少={missing},多出={extra},"
f"預期順序={list(expected)}"
)
def validate_row_links(
frames: Mapping[str, pd.DataFrame], row_link_key: str, expected_count: int
) -> None:
expected = list(range(1, expected_count + 1))
for name, frame in frames.items():
if len(frame) != expected_count:
raise ContractError(
f"{name} 預期 {expected_count} 筆,實際為 {len(frame)} 筆"
)
observed = frame[row_link_key].tolist()
if observed != expected:
raise ContractError(
f"{name} 的 {row_link_key} 必須依序為 1 到 {expected_count}"
)
def validate_constraint_block(
frame: pd.DataFrame,
constraints: Mapping[str, Any],
*,
block_name: str,
) -> None:
for name in constraints.get("required_non_null", []):
if frame[name].isna().any():
raise ContractError(f"{block_name} 的必要欄位 {name} 不可為空")
for name in constraints.get("numeric_fields", []):
original_missing = frame[name].isna()
numeric = pd.to_numeric(frame[name], errors="coerce")
invalid = numeric.isna() & ~original_missing
if invalid.any():
raise ContractError(
f"{block_name} 的 {name} 有 {int(invalid.sum())} 筆不是數值"
)
for name in constraints.get("non_empty_text_fields", []):
empty = frame[name].fillna("").astype(str).str.strip().eq("")
if empty.any():
raise ContractError(
f"{block_name} 的 {name} 有 {int(empty.sum())} 筆空字串"
)
for name, allowed in constraints.get("allowed_values", {}).items():
observed = set(frame[name].dropna().tolist())
unexpected = sorted(observed.difference(set(allowed)))
if unexpected:
raise ContractError(f"{block_name} 的 {name} 出現未允許值:{unexpected}")
for name, bounds in constraints.get("numeric_bounds", {}).items():
values = pd.to_numeric(frame[name], errors="coerce").dropna()
if "minimum" in bounds and (values < bounds["minimum"]).any():
raise ContractError(f"{block_name} 的 {name} 小於最小值 {bounds['minimum']}")
if "exclusive_minimum" in bounds and (
values <= bounds["exclusive_minimum"]
).any():
raise ContractError(
f"{block_name} 的 {name} 必須大於 {bounds['exclusive_minimum']}"
)
if "maximum" in bounds and (values > bounds["maximum"]).any():
raise ContractError(f"{block_name} 的 {name} 大於最大值 {bounds['maximum']}")
def validate_excluded_fields(
excluded: Mapping[str, Any], grouped_fields: Mapping[str, list[str]]
) -> None:
checked_roles = [
"post_triage_outcome",
"human_reference_only",
"reference_label",
"derived_evaluation_only",
]
for role in checked_roles:
expected = grouped_fields[role]
actual = excluded.get(role)
if actual != expected:
raise ContractError(
f"excluded-fields.json 的 {role} 不符;預期={expected},實際={actual}"
)
def validate_prepared_contract(
contract: Mapping[str, Any],
policy: Mapping[str, Any],
prepared_dir: Path,
) -> JsonObject:
"""Validate prepared KTAS tables and return an aggregate-only report."""
if contract.get("schema_version") != 1:
raise ContractError("目前只支援 schema_version=1 的 Day 14 契約")
cohort_report = validate_cohort_flow(contract)
allowed_fields, protected_fields, grouped_fields = validate_role_contract(
contract, policy
)
frames: dict[str, pd.DataFrame] = {}
for table_name, table in contract["prepared_tables"].items():
path = prepared_dir / table["filename"]
if not path.is_file():
raise FileNotFoundError(f"找不到 Day 12 產物:{path}")
frame = pd.read_csv(path)
validate_exact_columns(
frame,
expected_columns(contract, policy, table_name),
table["filename"],
)
frames[table_name] = frame
row_link_key = str(contract["row_link_key"])
expected_count = int(contract["cohort"]["final_eligible_record_count"])
validate_row_links(frames, row_link_key, expected_count)
model_inputs = frames["model_inputs"]
reference_labels = frames["reference_labels"]
actual_model_fields = [name for name in model_inputs.columns if name != row_link_key]
leaked = sorted(set(actual_model_fields).intersection(protected_fields))
if leaked:
raise ContractError(f"模型輸入出現受保護欄位:{leaked}")
if actual_model_fields != allowed_fields:
raise ContractError(
f"模型輸入欄位與政策不一致;預期={allowed_fields},實際={actual_model_fields}"
)
validate_constraint_block(
model_inputs,
contract["input_constraints"],
block_name="model-input-candidates.csv",
)
validate_constraint_block(
reference_labels,
contract["label_constraints"],
block_name="reference-labels.csv",
)
excluded_path = prepared_dir / str(contract["excluded_fields_filename"])
if not excluded_path.is_file():
raise FileNotFoundError(f"找不到 Day 12 產物:{excluded_path}")
validate_excluded_fields(load_json(excluded_path), grouped_fields)
return {
"schema_version": 1,
"contract_id": contract["contract_id"],
"status": "passed",
"scope": contract["output_policy"]["scope"],
"cohort": cohort_report,
"prepared_data": {
"record_count": expected_count,
"model_input_field_count": len(allowed_fields),
"audit_field_count": len(grouped_fields["audit_or_stratification_only"]),
"protected_field_count": len(protected_fields),
"model_input_missing_counts": {
name: int(model_inputs[name].isna().sum()) for name in allowed_fields
},
},
"leakage_boundary": {
"allowed_model_input_fields": allowed_fields,
"protected_fields": protected_fields,
"protected_fields_in_model_input": leaked,
},
"checks": {
"cohort_arithmetic_valid": True,
"prepared_row_counts_match": True,
"record_indexes_aligned": True,
"table_schemas_exact": True,
"input_constraints_valid": True,
"label_constraints_valid": True,
"protected_fields_absent_from_model_input": True,
"excluded_field_registry_matches_policy": True,
},
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
scripts/validate_day14_contract.py讀取三份前置設定與 Day 12 產物,呼叫核心驗證並留下聚合報告。
請在文字編輯器建立 scripts/validate_day14_contract.py,貼入以下完整內容並儲存:
#!/usr/bin/env python3
"""Validate the Day 14 cohort, schema, and leakage contract."""
from __future__ import annotations
import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path
from triage_rag.data.contracts import load_json, validate_prepared_contract
from triage_rag.reproducibility import (
canonical_json_bytes,
file_record,
git_state,
sha256_bytes,
sha256_file,
write_json,
)
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = "configs/data/day-14-ktas-data-contract.json"
DEFAULT_POLICY = "configs/data/day-11-ktas-field-policy.json"
DEFAULT_DATASET_CARD = "configs/data/day-12-kaggle-ktas-dataset-card.json"
DEFAULT_PREPARED_DIR = "data/interim/ktas-v1"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="驗證 Day 14 cohort、欄位 schema 與資料洩漏邊界。"
)
parser.add_argument("--contract", default=DEFAULT_CONTRACT)
parser.add_argument("--policy", default=DEFAULT_POLICY)
parser.add_argument("--dataset-card", default=DEFAULT_DATASET_CARD)
parser.add_argument("--prepared-dir", default=DEFAULT_PREPARED_DIR)
return parser.parse_args()
def main() -> int:
args = parse_args()
contract_path = PROJECT_ROOT / args.contract
policy_path = PROJECT_ROOT / args.policy
dataset_card_path = PROJECT_ROOT / args.dataset_card
prepared_dir = PROJECT_ROOT / args.prepared_dir
contract = load_json(contract_path)
policy = load_json(policy_path)
dataset_card = load_json(dataset_card_path)
source_counts = {
int(contract["cohort"]["final_eligible_record_count"]),
int(policy["observed_structure"]["record_count"]),
int(dataset_card["version_lock"]["record_count"]),
}
if len(source_counts) != 1:
raise ValueError(
"Day 11 欄位政策、Day 12 資料卡與 Day 14 cohort 的最終筆數不一致"
)
report = validate_prepared_contract(contract, policy, prepared_dir)
report_sha256 = sha256_bytes(canonical_json_bytes(report))
started_at = datetime.now(timezone.utc)
run_id = f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-{report_sha256[:8]}"
output_root = PROJECT_ROOT / contract["output_policy"]["output_root"]
run_directory = output_root / run_id
run_directory.mkdir(parents=True, exist_ok=False)
report_path = run_directory / "contract-validation.json"
write_json(report_path, report)
prepared_inputs = [
str((Path(args.prepared_dir) / table["filename"]).as_posix())
for table in contract["prepared_tables"].values()
]
prepared_inputs.append(
str((Path(args.prepared_dir) / contract["excluded_fields_filename"]).as_posix())
)
manifest = {
"manifest_schema_version": 1,
"run_id": run_id,
"started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
"command": ["poetry", "run", "python", *sys.argv],
"git": git_state(PROJECT_ROOT),
"inputs": [
file_record(PROJECT_ROOT, path)
for path in [
args.contract,
args.policy,
args.dataset_card,
*prepared_inputs,
]
],
"outputs": [
{
"path": str(report_path.relative_to(PROJECT_ROOT)),
"sha256": sha256_file(report_path),
"deterministic_content_sha256": report_sha256,
"size_bytes": report_path.stat().st_size,
}
],
"scope": contract["output_policy"]["scope"],
"privacy": "輸出只含欄位名稱、筆數、缺失數與雜湊,不含逐筆病歷值。",
}
manifest_path = run_directory / "run-manifest.json"
write_json(manifest_path, manifest)
print("Day 14 資料契約驗證:通過")
print(f"資料筆數:{report['prepared_data']['record_count']:,}")
print(f"模型候選輸入:{report['prepared_data']['model_input_field_count']} 欄")
print(f"受保護欄位:{report['prepared_data']['protected_field_count']} 欄")
print("模型輸入中的受保護欄位:0 欄")
print(f"執行目錄:{run_directory.relative_to(PROJECT_ROOT)}")
print(f"穩定報告摘要:{report_sha256}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/test_data_contract.py以合法與故意破壞的合成輸入驗證契約真的會阻擋錯誤。
請在文字編輯器建立 tests/test_data_contract.py,貼入以下完整內容並儲存:
from __future__ import annotations
import copy
import json
import shutil
import tempfile
import unittest
from pathlib import Path
import pandas as pd
from triage_rag.data.contracts import (
ContractError,
load_json,
validate_prepared_contract,
)
ROOT = Path(__file__).resolve().parents[1]
CONTRACT_PATH = ROOT / "configs" / "data" / "day-14-ktas-data-contract.json"
POLICY_PATH = ROOT / "configs" / "data" / "day-11-ktas-field-policy.json"
FIXTURE_DIR = ROOT / "tests" / "fixtures" / "day-14-prepared"
class Day14DataContractTests(unittest.TestCase):
def setUp(self) -> None:
self.contract = load_json(CONTRACT_PATH)
self.policy = load_json(POLICY_PATH)
def synthetic_contract(self) -> dict:
contract = copy.deepcopy(self.contract)
contract["cohort"]["candidate_record_count"] = 3
contract["cohort"]["candidate_definition"] = "三筆純合成測試紀錄"
contract["cohort"]["exclusions"] = []
contract["cohort"]["final_eligible_record_count"] = 3
return contract
def copied_fixture(self, directory: str) -> Path:
destination = Path(directory) / "prepared"
shutil.copytree(FIXTURE_DIR, destination)
return destination
def test_synthetic_prepared_data_passes_contract(self) -> None:
report = validate_prepared_contract(
self.synthetic_contract(), self.policy, FIXTURE_DIR
)
self.assertEqual(report["status"], "passed")
self.assertEqual(report["prepared_data"]["record_count"], 3)
self.assertEqual(report["prepared_data"]["model_input_field_count"], 14)
self.assertEqual(
report["leakage_boundary"]["protected_fields_in_model_input"], []
)
def test_reference_label_in_model_input_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
prepared = self.copied_fixture(directory)
input_path = prepared / "model-input-candidates.csv"
frame = pd.read_csv(input_path)
frame["KTAS_expert"] = [2, 3, 3]
frame.to_csv(input_path, index=False)
with self.assertRaisesRegex(ContractError, "KTAS_expert"):
validate_prepared_contract(
self.synthetic_contract(), self.policy, prepared
)
def test_missing_input_column_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
prepared = self.copied_fixture(directory)
input_path = prepared / "model-input-candidates.csv"
frame = pd.read_csv(input_path).drop(columns=["RR"])
frame.to_csv(input_path, index=False)
with self.assertRaisesRegex(ContractError, "RR"):
validate_prepared_contract(
self.synthetic_contract(), self.policy, prepared
)
def test_invalid_cohort_arithmetic_is_rejected(self) -> None:
contract = self.synthetic_contract()
contract["cohort"]["exclusions"] = [
{"reason": "合成排除", "removed": 1, "remaining": 3}
]
with self.assertRaisesRegex(ContractError, "cohort 流程算術不一致"):
validate_prepared_contract(contract, self.policy, FIXTURE_DIR)
def test_incomplete_excluded_field_registry_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
prepared = self.copied_fixture(directory)
registry_path = prepared / "excluded-fields.json"
registry = json.loads(registry_path.read_text(encoding="utf-8"))
registry["post_triage_outcome"].remove("Disposition")
registry_path.write_text(
json.dumps(registry, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
with self.assertRaisesRegex(ContractError, "post_triage_outcome"):
validate_prepared_contract(
self.synthetic_contract(), self.policy, prepared
)
if __name__ == "__main__":
unittest.main()
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/fixtures/day-14-prepared/model-input-candidates.csv三筆合成候選輸入。
請在文字編輯器建立 tests/fixtures/day-14-prepared/model-input-candidates.csv,貼入以下完整內容並儲存:
record_index,Sex,Age,Arrival mode,Injury,Chief_complain,Mental,Pain,NRS_pain,SBP,DBP,HR,RR,BT,Saturation
1,1,68,2,1,合成案例:呼吸困難,1,0,,92,58,118,28,37.2,91
2,2,45,3,2,合成案例:腳踝外傷,1,1,6,132,82,88,18,36.8,98
3,1,31,1,1,合成案例:發燒與倦怠,1,1,3,118,74,102,22,39.1,
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/fixtures/day-14-prepared/audit-strata.csv三筆合成稽核分層。
請在文字編輯器建立 tests/fixtures/day-14-prepared/audit-strata.csv,貼入以下完整內容並儲存:
record_index,Group,Patients number per hour
1,1,8
2,2,6
3,1,9
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/fixtures/day-14-prepared/reference-labels.csv三筆合成參考標籤。
請在文字編輯器建立 tests/fixtures/day-14-prepared/reference-labels.csv,貼入以下完整內容並儲存:
record_index,KTAS_RN,KTAS_expert,Error_group,mistriage
1,2,2,,0
2,4,3,4,2
3,3,3,,0
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/fixtures/day-14-prepared/excluded-fields.json合成測試使用的禁止欄位登錄表。
請在文字編輯器建立 tests/fixtures/day-14-prepared/excluded-fields.json,貼入以下完整內容並儲存:
{
"note": "以下欄位不會寫入模型輸入檔。",
"post_triage_outcome": [
"Diagnosis in ED",
"Disposition",
"Length of stay_min",
"KTAS duration_min"
],
"human_reference_only": [
"KTAS_RN"
],
"reference_label": [
"KTAS_expert"
],
"derived_evaluation_only": [
"Error_group",
"mistriage"
]
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
以下命令會在需要時產生套件鎖定檔,接著檢查設定格式與 Python 語法;它們不會啟動模型,也不會把病患資料送到網路:
poetry run python -m json.tool configs/data/day-14-ktas-data-contract.json
poetry run python -m py_compile src/triage_rag/data/contracts.py scripts/validate_day14_contract.py tests/test_data_contract.py
每個命令都應正常結束。若 JSON 顯示行號,先檢查貼上時是否遺漏逗號、引號或括號;若 py_compile 報錯,先依行號修正縮排或漏貼內容。靜態檢查通過後,再執行本文後面的正式步驟。
執行入口是 scripts/validate_day14_contract.py。scripts/ 用來保存讀者可以從終端機啟動的自動化程式;這支 .py 檔負責串接設定、核心驗證模組、Day 12 產物與 Day 13 建立的執行產物清單(Run Manifest,下文簡稱 run manifest),不重複實作核心規則。
本篇沿用 Python 相依套件管理與封裝工具(Python dependency management and packaging tool, Poetry)執行所有命令。Poetry 會依 poetry.lock 使用 Day 13 鎖定的 Python 與套件環境,但它不會下載 KTAS 資料或 Ollama 模型。
本機的 data/ 資料夾保存不提交 Git 的逐筆資料,interim/ 子目錄則保存已正規化並按角色拆分的中介產物。因此 data/interim/ktas-v1/ 是 Day 12 輸出、Day 14 讀取的交接位置。
它讀取以下輸入:
configs/data/day-14-ktas-data-contract.json:本篇新增的資料契約。configs/data/day-11-ktas-field-policy.json:24 個來源欄位的角色政策。configs/data/day-12-kaggle-ktas-dataset-card.json:公開來源與版本紀錄。data/interim/ktas-v1/:Day 12 產生的候選輸入、稽核欄位、參考標籤與排除清單。results/ 是保存本機執行產物的資料夾,runs/ 子目錄再按日期與驗證工作分開每次執行。成功後,程式會在 results/runs/day-14/ 建立獨立執行目錄,寫入 contract-validation.json 與 run-manifest.json。前者保存穩定的聚合檢查結果,後者保存本次時間、Git 狀態、輸入檔雜湊與輸出摘要。兩個檔案都不包含逐筆病歷值。
poetry sync,而且目前位於自己的專案根目錄。data/interim/ktas-v1/ 尚不存在,執行下列命令。程式會從公開端點下載鎖定版本;若檔案已存在,則先核對內容摘要再重用本機快取。poetry run python scripts/run_ktas_pipeline.py
record_index 與 14 個候選輸入。data/interim/ktas-v1/ 中存在 model-input-candidates.csv、audit-strata.csv、reference-labels.csv 與 excluded-fields.json。流程會以安全雜湊演算法 256 位元版本(Secure Hash Algorithm 256-bit, SHA-256)核對來源內容;若摘要不同,流程必須停止,不要直接修改預期值。poetry run python -m unittest discover -s tests -v
ok,最後出現 Ran 8 tests 與 OK。FAIL 或 ERROR 都代表工程契約尚未通過,不應繼續做資料切分或模型實驗。本次實際執行結果為:
Ran 8 tests in 0.042s
OK
poetry run python scripts/validate_day14_contract.py
contract-validation.json 與 run-manifest.json,而且終端機印出 64 字元的穩定報告摘要。本次實際得到的穩定摘要為:
6b32f9057046912680f5d978790ffe2236bea9205bcd9e73a771b35abdcab6dd
這個摘要只代表契約報告內容一致,不是 Kaggle ZIP 的雜湊,也不是模型版本或效能指標。
results/runs/day-14/ 最近一次執行的 contract-validation.json。poetry run python - <<'PY'
import json
from pathlib import Path
runs = sorted(path for path in Path("results/runs/day-14").iterdir() if path.is_dir())
if not runs:
raise SystemExit("尚未找到 Day 14 成功執行結果")
report = json.loads(
(runs[-1] / "contract-validation.json").read_text(encoding="utf-8")
)
print("狀態:", report["status"])
print("資料筆數:", report["prepared_data"]["record_count"])
print("候選輸入欄位:", report["prepared_data"]["model_input_field_count"])
print(
"模型輸入中的受保護欄位:",
report["leakage_boundary"]["protected_fields_in_model_input"],
)
PY
passed、資料筆數為 1267、候選輸入欄位為 14,最後一行是空陣列 []。下圖把這次實際執行的聚合結果整理在同一個畫面。上排是資料與欄位數量,下排八項檢查則對應核心程式逐項驗證的契約。

上圖證明的是「目前這份處理後資料符合我們寫下的工程契約」。它沒有測量任何模型準確率,也不能證明 14 個候選輸入都具有因果意義、資料品質良好或足以支援臨床部署。
黑名單只能擋住已知名稱。假設未來有人新增 final_diagnosis_code,但忘記把它加入黑名單,單純的交集檢查可能看不出問題。
因此本篇採用兩層防線:
白名單負責限制「只能有哪些」,角色政策則說明「為什麼只能是它們」。未來若真的需要新增欄位,必須先更新來源說明、時間階段、資料角色、測試與契約版本,而不是直接讓模型讀取。
本篇處理的是欄位層級與時間角色的第一道防線,仍有其他洩漏風險:
scikit-learn 官方文件也提醒,前處理、特徵選擇與參數學習不能讓測試資料影響訓練流程,否則評估會過度樂觀。scikit-learn:Common pitfalls and recommended practices
這些問題需要資料切分順序與訓練管線共同處理,會留到 Day 15。今天先確保最明顯的答案欄位與未來結果根本無法進入模型輸入檔。
公開資料沒有可用病患識別碼,所以只能說 1,267 筆符合條件的紀錄。後續結果必須保留這項限制。
公開來源提供 1,540 筆候選紀錄及三項排除數量;Kaggle CSV 只提供最終 1,267 筆。我們能核算數量,不能檢查被排除紀錄的每一列是否符合原因。
NRS_pain、生命徵象與 Saturation 都可能缺失。Schema 允許來源中已知的缺失,不會擅自把缺失補成正常值。缺失機制與處理方式是 Day 15 的主題。
契約能證明欄位邊界、型別與已知值域一致,不能證明公開資料涵蓋完整 KTAS 規則,也不能讓模型取代護理師或醫師。系統仍定位為離線研究與決策支援實驗。
Day 14 把「一筆資料怎麼進系統」變成可以執行的規則:
record_index 只負責跨表對齊,不是模型特徵。現在我們已經知道哪些欄位「有資格被考慮」,但還沒有決定缺失值如何處理,也還沒有建立公平的訓練與測試切分。
Day 15 會接著處理三個真實資料一定會遇到的問題:缺失值、五級標籤不平衡與資料切分。我們會先判斷缺失代表沒有測量、無法測量還是資料未記錄,再建立不讓測試資料參與前處理的切分順序。
Day 14 產生的 14 欄白名單、record_index 對齊規則與 KTAS_expert 標籤隔離,會直接成為 Day 15 建立訓練、驗證與測試資料的輸入契約。
unittest