iT邦幫忙

2026 iThome 鐵人賽

DAY 29
0
佛心分享-SideProject30

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

# Day 29|模型為什麼錯?用 Evidence Trace 寫出可信的結果與限制

  • 分享至 

  • xImage
  •  

Day 28 已把 Day 26 的折外預測(Out-of-Fold Prediction, OOF Prediction)拆成序位距離、錯誤方向與高急迫級安全指標。線性序位 Ridge 迴歸(Ridge Regression)基準 B2 在第 1 次重複(repeat 1)的 1,267 筆紀錄中完全相符 685 筆、跨兩級以上錯誤 73 筆;參考第一、二級共 246 筆,其中 191 筆被預測為第三至第五級。這些數字告訴我們「錯了多少」,卻還沒有回答「錯誤集中在哪裡」。

錯誤分析最容易出現兩種看似合理的捷徑。第一種是只挑模型修好的漂亮案例,忽略模型把原本正確答案改錯的案例。第二種是看到最終級數錯誤,就直接寫成「檢索失敗」或「大型語言模型(Large Language Model, LLM)亂答」,即使實驗根本沒有保存檢索候選、提示詞與生成輸出。

下圖是概念示意。匿名紀錄依序經過輸入、證據與最終決策檢查站,中間的斷裂連線提醒我們:只有保存每一站的資料,臨床覆核者才有機會定位問題。圖中沒有真實病患、正式醫療規則或本篇實驗數字。

匿名檢傷紀錄依序通過輸入、證據與決策檢查站,中間斷裂連線由臨床人員覆核的概念示意

上圖也刻意把臨床人員放在流程末端。這套系統仍是研究中的決策支援候選;錯誤追蹤不是為了替模型自動辯解,而是讓人知道哪一段需要修正、哪些結論目前不能成立。

本篇把從輸入到輸出的逐站紀錄稱為證據追蹤紀錄(Evidence Trace)。若系統是先找外部知識、再交給生成模型回答的檢索增強生成(Retrieval-Augmented Generation, RAG),Evidence Trace 才能讓我們分開檢查檢索與生成。B0 是每折只學訓練資料多數級數的基準,B2 是只用檢傷當下結構化欄位的序位 Ridge 基準;P0 則是本系列規劃中的完整 RAG 方法識別碼,Day 29 尚未完成它的病患 OOF。

本篇會完成五件事:

  1. 固定使用 repeat 1,對高急迫級、跨兩級錯誤、生命徵象缺失、疼痛、意識反應、外傷與急診場域建立可重跑切片。
  2. 同時呈現 B2 修正 B0 與弄壞 B0 的案例,避免成功案例選例偏誤。
  3. 依四種事先寫入設定的優先序,各抽五筆本機私人案例供人工稽核。
  4. 用七筆作者合成工程案例驗證 Evidence Trace 錯誤分類器,清楚分開檢索、生成與拒答錯誤。
  5. 把缺少的病患 RAG 中間產物保留為 null,並建立未來完整 P0 必須保存的 trace 欄位。

本篇仍是事後探索性分析。切片規則是在 Day 26 結果已經觀察後建立,不能把切片差異寫成原因、因果效應或臨床安全證明。


本篇會用到的名詞

中文名稱 英文全名/縮寫 本篇用途
韓國急診檢傷與急迫度分級量表 Korean Triage and Acuity Scale, KTAS 本資料的五級參考制度;第一級最急迫、第五級最不急迫
折外預測 Out-of-Fold Prediction, OOF Prediction 每筆紀錄由沒有使用該筆資料擬合的模型產生預測
錯誤切片 Error Slice 依事先定義的欄位或錯誤型態分組,逐組保留分母與錯誤率
證據追蹤紀錄 Evidence Trace 保存從輸入、檢索、門控到生成與引用的逐步紀錄,讓錯誤能定位到不同元件
檢索查詢 Query 送入檢索器的問題表示;它決定候選知識如何被搜尋
生成脈絡 Context 檢索與門控後真正交給生成器的證據;它不是全部候選
提示詞 Prompt 告訴生成器任務、限制與輸出格式的指令;版本漂移會改變結果
規則識別碼 rule_id 連結候選、context、引用與來源的穩定代碼,用來判斷必要證據是否齊全
匿名紀錄索引 record_index 對齊同一資料列的輸入、參考級數與預測,不代表病患識別碼
檢索增強生成 Retrieval-Augmented Generation, RAG 先從外部知識庫檢索資料,再讓生成模型依取回內容形成回答的方法
檢索 Retrieval 從候選知識中選出與當前問題相關的規則或文件
生成 Generation 根據提示詞與最後 context 產生結構化輸出的步驟
拒答 Abstention 證據、輸入或系統狀態不符合回答條件時,停止提出級數並轉交人工處理
錯誤分類體系 Error Taxonomy 用一致規則把錯誤分成可區分、可計數的類別
疼痛數字評定量表 Numeric Rating Scale, NRS 以 0 到 10 記錄疼痛程度;無疼痛時不適用,有疼痛卻缺值時代表未記錄
事後分析 Post-hoc Analysis 看過結果後才建立的分析;可用來探索,不能偽裝成事前確認性計畫
外部效度 External Validity 結果能否合理延伸到其他醫院、時間、族群或制度的程度
安全雜湊演算法 256 位元 Secure Hash Algorithm 256-bit, SHA-256 核對設定、資料與輸出是否仍是同一個位元版本
JavaScript 物件表示法 JavaScript Object Notation, JSON 保存可由人與程式共同讀取的契約、trace 與結果
測試夾具 Test Fixture 為測試固定準備的合成輸入與鎖定預期,不是病患樣本

B0 在 repeat 1 永遠輸出第三級。B2 不是 RAG,也沒有 LLM 生成步驟;這個差異會決定本篇能做與不能做的歸因。

先鎖住三層證據,不讓不同任務混在一起

錯誤分析至少有三層,每一層的輸入、輸出與允許結論都不同。

第一層:真實 repeat 1 OOF 錯誤切片

這一層使用 Day 26 已完成的 1,267 筆 repeat 1 OOF,透過 record_index 對齊檢傷當下輸入與稽核分層。它能回答:

  • 哪些參考級數較常完全相符、檢傷不足或跨兩級以上錯誤?
  • 生命徵象缺失較多的紀錄,描述性錯誤率是多少?
  • B2 修正或弄壞 B0 各有多少筆?

它不能回答「RAG 為什麼錯」,因為 B2 沒有檢索或生成元件。

第二層:本機私人固定案例稽核

這一層從真實 repeat 1 紀錄依固定優先序抽出 20 筆,保留主訴、生命徵象、參考級數與兩個基準預測,供研究者在本機人工覆核。逐筆內容寫入 results/runs/day-29/<run_id>/day-29-private-error-audit.jsonresults/runs/ 是保存每次執行細節且被 Git 忽略的資料夾,不會被提交成公開文章資產。

私人稽核可以產生後續假設,例如「某類缺失組合可能需要不同表示方式」。它仍不能直接證明機制,除非下一輪實驗事先鎖定假設並留下所需 trace。

第三層:合成 Evidence Trace 分類器

這一層使用七筆作者撰寫的合成工程案例,不含病患列、人工標籤或模型輸出。它只驗證:當必要 rule_id、取回 rule_id、是否拒答、是否該拒答與合成參考答案都存在時,程式能否一致分出檢索錯誤、生成錯誤、碰巧答對與拒答錯誤。

RAG 的原始設計本來就包含檢索器與生成器兩種不同角色;因此要定位元件問題,必須保存兩者之間的中間產物,而不是只看最後答案。Lewis 等人:Retrieval-Augmented Generation

下圖把三層證據放在同一條路徑上。請注意最下方紅色邊界:目前病患 P0 的檢索與生成歸因不是零,而是沒有必要資料,所以必須寫成 null

真實 OOF 切片、私人固定案例稽核與合成 Evidence Trace 三層邊界流程

上圖的重點是「不跨層借證據」。真實 OOF 可以報病患紀錄上的錯誤分布;合成 trace 可以驗證分類規則;兩者不能拼成「病患 RAG 已完成元件錯誤分析」。

真實資料如何對齊,又有哪些欄位不能使用

來源研究回溯檢視兩所急診的成人就醫紀錄,最終納入 1,267 筆匿名醫療紀錄;同一資料列的護理師登錄 KTAS、專家重新判定 KTAS 與輸入欄位角色,已在 Day 11–15 鎖定。PLOS ONE:Triage accuracy and causes of mistriage using the Korean Triage and Acuity Scale

Day 29 以三份資料對齊:

輸入 本篇用途 公開方式
Day 26 完整 OOF JSON 取得 repeat 1 的參考級數、B0B2 預測 只公開聚合,逐列預測不提交
data/interim/ktas-v1/model-input-candidates.csv 取得檢傷當下可得的年齡、主訴、意識、疼痛與生命徵象 主訴與逐筆數值只留本機私人稽核
data/interim/ktas-v1/audit-strata.csv 取得急診場域與每小時到院人數供稽核 公開檔只報預先定義的場域聚合

模型輸入表若出現 KTAS_RNKTAS_expertError_groupmistriage、診斷、處置、急診停留時間或檢傷完成時間,程式會直接失敗。這些欄位不是當下可用的安全輸入:有些是答案或答案衍生物,有些在檢傷之後才形成。

公開資料沒有病患識別碼,所以 1,267 筆只能稱為檢傷紀錄,不能寫成 1,267 位彼此獨立的病患。個人預後或診斷之多變量預測模型透明報告加人工智慧(Transparent Reporting of a multivariable prediction model for Individual Prognosis Or Diagnosis plus Artificial Intelligence, TRIPOD+AI)要求完整、透明地報告資料使用、模型評估、重要子群效能與研究限制;子群表不會自動解決代表性或相依性問題。TRIPOD+AI statement

錯誤定義必須先於看案例

KTAS 第一級最急迫、第五級最不急迫。本篇對每一筆紀錄定義:

[
Error_i = \hat{y}_i-y_i
]

其中 (y_i) 是專家參考級數,(\hat{y}_i) 是模型預測級數。

  • (Error_i=0):完全相符。
  • (Error_i>0):預測數字較大,屬於檢傷不足方向。
  • (Error_i<0):預測數字較小,屬於檢傷過度方向。
  • (|Error_i|\ge 2):跨兩級以上錯誤。
  • (y_i\in{1,2}) 且 (\hat{y}_i\in{3,4,5}):高急迫級嚴重檢傷不足。

例如參考第二級、預測第四級時,Error=+2,同時是檢傷不足、跨兩級以上錯誤與高急迫級嚴重檢傷不足。參考第四級、預測第二級時,Error=-2,是檢傷過度與跨兩級以上錯誤,但不是高急迫級嚴重不足。

這些定義全部寫在設定檔,而不是研究者打開個案後臨時決定。

Repeat 1 整體結果:B2 有修正,也有新錯誤

repeat 1 的 B0B2 結果如下:

指標 B0 B2
完全相符 487/1,267(38.44%) 685/1,267(54.06%)
檢傷不足 246/1,267(19.42%) 297/1,267(23.44%)
檢傷過度 534/1,267(42.15%) 285/1,267(22.49%)
跨兩級以上錯誤 101/1,267(7.97%) 73/1,267(5.76%)
高急迫級嚴重不足 246/246(100.00%) 191/246(77.64%)

B2 完全相符率較高、跨兩級以上錯誤較少,也修正了一部分 B0 對高急迫級永遠預測第三級的問題。但整體檢傷不足率由 19.42% 增加到 23.44%。所以本篇不能把「錯誤較少」縮寫成「所有安全方向都改善」。

若只把每個模型各自的準確率並排,還看不到同一筆紀錄如何變化。配對轉換矩陣補上這一層:

B0/B2 完全相符狀態 筆數 比例
兩者都相符 380 30.0%
B2 修正 B0 305 24.1%
B2 弄壞 B0 107 8.4%
兩者都錯 475 37.5%

305 筆修正案例與 107 筆退步案例必須同時保留。只寫前者會把模型改進故事變成選例偏誤。

預先定義切片:先報分母,再報比率

本篇切片只使用下列檢傷當下欄位或稽核欄位:參考急迫度、六項生命徵象缺失數、疼痛/NRS 語意、意識反應、急診場域與是否外傷。這些切片的目的,是找出值得下一輪驗證的區域,不是替每個差異發明因果故事。

下圖選出參考急迫度、生命徵象缺失帶與急診場域三組切片。每一列先顯示 n,再同時畫出完全相符率、檢傷不足率與跨兩級以上錯誤率。

八個錯誤切片的樣本數、B2 完全相符率、檢傷不足率與跨兩級以上錯誤率比較

上圖有三個值得保留、但不能過度解讀的觀察:

  1. 高急迫級 246 筆中,B2 完全相符率只有 17.89%,檢傷不足率為 81.71%,跨兩級以上錯誤率為 11.38%。其中 191 筆是第三至第五級的嚴重不足;另外十筆雖然仍在第一、二級之間錯開,會被一般檢傷不足計入,卻不符合嚴重不足定義。
  2. 六項生命徵象缺兩項以上只有 29 筆,完全相符率為 44.83%、跨兩級以上錯誤率為 24.14%。這個切片分母很小,而且缺失可能與病情、量測流程或記錄習慣共同相關,不能直接寫成「缺失造成錯誤」。
  3. 地方急診 688 筆與區域急診 579 筆的完全相符率分別為 56.98% 與 50.60%。兩個場域的參考級數分布、流程與病患組成未必相同,原始差異不是場域因果效應。

其他預先定義切片提供補充方向:

切片 n B2 完全相符 B2 檢傷不足 B2 跨兩級以上
有疼痛且 NRS 已記錄 711 55.27% 22.36% 2.81%
清醒 1,187 54.68% 22.75% 5.22%
對聲音反應 39 35.90% 35.90% 12.82%
對疼痛反應 28 42.86% 35.71% 21.43%
無反應 13 76.92% 23.08% 0.00%
非外傷 1,023 51.91% 24.44% 5.87%
外傷 244 63.11% 19.26% 5.33%

「有疼痛但 NRS 未記錄」的切片少於五筆。設定檔把公開最小儲存格設為五,程式會輸出 record_count=nullrecord_count_display="<5",並把該格所有指標設為 null。為避免讀者用 1,267 筆總數減去其他兩格回推小格,程式還會把剩餘類別中筆數較小的一格做互補抑制,顯示為「已抑制」。它不會把缺失改寫成 NRS=0,也不為了湊表格公開精確數字。

模型卡(Model Card)概念鼓勵模型發布者提供分解後的評估數字、適用情境與限制,讓平均分數之外的行為更透明;但切片選擇與小樣本解讀仍需要明確規則。Mitchell 等人:Model Cards for Model Reporting

固定抽樣規則:不要用故事反過來挑案例

私人案例依下列順序抽樣,每一類按 record_index 由小到大取五筆,跨優先序不重複:

  1. B2 高急迫級嚴重檢傷不足,母群 191 筆。
  2. 排除前一類後的其餘跨兩級以上錯誤,母群 45 筆。
  3. B0 原本完全相符、B2 改錯,母群 107 筆。
  4. B0 原本錯、B2 修正為完全相符,母群 305 筆。

排序不是隨機抽樣,也不代表母群分布;它的目的只是讓每次重跑都取得相同 20 筆,方便多人覆核與追蹤修正。若研究問題需要估計案例類型比例,應另設機率抽樣與抽樣權重,不能使用這 20 筆固定案例。

下圖左側保留四種正確性轉換,右側才列抽樣優先序。這個版面提醒我們:修正案例只是四格中的一格,不是完整錯誤分析。

B0 到 B2 的四種正確性轉換矩陣與每類五筆的固定案例抽樣優先序

上圖的逐筆資料不會出現在文章或公開 JSON。公開檔只記錄四個母群數、每類選取五筆與總共 20 筆;實際 record_index、主訴與生命徵象留在 gitignored 的本機 run 目錄。

統一案例分析表要記錄什麼

逐筆人工稽核不應只留一段自由文字。建議使用固定欄位,讓兩位覆核者能看到相同證據並區分觀察與推論:

欄位 要記錄的內容 為什麼需要
audit_case_id 本機稽核代碼,不作為公開識別碼 讓覆核意見可以回連同一案例
selection_priority 此案例因哪個固定規則入選 揭露選例機制,避免看完故事才挑案例
參考與預測 reference_levelB0B2 先保存可觀察結果
錯誤方向與距離 完全相符/不足/過度、絕對差幾級 避免把不同錯誤混成一類
當下可用輸入 主訴、意識、疼痛、生命徵象與缺失狀態 只檢查決策當下資訊,不倒灌診斷與處置
Evidence Trace 狀態 完整、部分缺失或不存在 決定能否做 retrieval/generation 歸因
覆核者觀察 逐項寫看到的事實 與機制假設分開
機制假設 可能的表示、檢索、門控或生成問題 必須標示尚待驗證
下一個可驗證動作 新增測試、補 trace、鎖定切片或重跑實驗 把案例故事轉成工程任務

例如一筆完全合成的教學案例可以寫成:參考第二級、B2 第四級,生命徵象有兩項未記錄;觀察是「跨兩級檢傷不足且缺兩項生命徵象」,不能直接寫成「缺失值導致錯誤」。若沒有 query、候選與 prompt,retrieval 與 generation 欄位都應填 not_available,不是依主觀猜測補上錯誤類別。

Evidence Trace:先判斷拒答,再拆檢索與生成

完整 Evidence Trace 至少要能回答三個問題:

  1. 這筆案例依契約是否應該拒答?
  2. 若可以回答,必要的規則是否出現在最後 context?
  3. 若證據齊全,生成結果是否仍錯誤或不受引用支持?

順序不能顛倒。應拒答卻回答的案例,即使最後「碰巧」符合合成參考答案,仍違反拒答政策;檢索缺少必要證據卻答對,也不能算端到端有支持。

下圖使用七筆作者合成工程案例說明判斷順序。每個類別只有一筆,目的只是確定分類器能走到所有分支,不是估計錯誤比例。

先判斷拒答邊界,再檢查必要 rule ID 與生成輸出的 Evidence Trace 分類流程

上圖對應七個錯誤類別:

類別 判斷條件 修正方向
端到端有支持 必要證據齊全、答案符合合成參考、也不應拒答 保留為正向對照
檢索正確但生成錯誤 必要 rule_id 已取回,答案仍錯 檢查 prompt、結構化輸出、引用與生成驗證
檢索錯誤但碰巧答對 必要 rule_id 缺失,答案卻符合合成參考 不能當作受證據支持;修正檢索並加入反事實測試
檢索錯誤 必要 rule_id 缺失且答案錯 檢查 query、候選集合、排序與門控
合理拒答 契約要求拒答,系統也拒答 確認人工接手與原因碼完整
不必要拒答 證據足夠且不應拒答,系統卻拒答 檢查門檻、缺失判斷與衝突規則
漏掉拒答 契約要求拒答,系統仍回答 優先修正失敗即關閉(fail-closed)路徑

七筆 fixture 的 rule_id 都以 SYN-RULE-* 命名,synthetic_only=true,並固定 expected_error_class。程式若分類結果與預期不同就失敗,不會只輸出一張看似合理的表。

為什麼目前不能對病患 B2 做 retrieval/generation 歸因

Day 26 的逐筆 OOF 欄位只有:

record_index
repeat_id
fold_id
reference_level
B0_predicted_level
B1_decision_status
B1_abstention_reason_code
B2_continuous_score
B2_predicted_level

這些欄位沒有 query、候選規則、分數、來源、門控、最後 context、prompt、模型原始輸出或引用。更根本地說,B2 是序位 Ridge,不是 RAG。從 B2_predicted_level=4 不能倒推出「檢索選錯規則」或「生成器忽略證據」。

因此公開結果保留:

{
  "patient_retrieval_error_count": null,
  "patient_generation_error_count": null,
  "patient_abstention_error_count": null,
  "normal_vital_signs_plus_high_risk_complaint": null,
  "negation_error": null,
  "rare_symptom_error": null
}

「生命徵象正常但主訴高風險」需要事先鎖定主訴風險本體與對應規則;「否定錯誤」與「罕見症狀」需要事先鎖定文字標註規範與雙人覆核。現在直接以關鍵字掃描英文主訴再宣布錯誤類別,會把研究者臨時選的詞表偽裝成標準答案,所以本篇不做。

未來 P0 必須保存的逐筆 Evidence Trace

完整 P0 若要進入病患 OOF 比較,每筆至少保存:

  1. case_trace_id:本機可追蹤、公開不可回推病患的案例代碼。
  2. input_snapshot_sha256:當次輸入快照雜湊,證明覆核的是同一版資料。
  3. query_text_or_hash:實際檢索 query 或受控雜湊。
  4. candidate_rule_ids_with_scores_and_origins:所有候選 rule_id、分數與來源路徑。
  5. gate_decisions:生命徵象、主訴、階層與拒答門控的逐步決策。
  6. final_context_rule_ids:真正交給生成器的證據集合。
  7. prompt_version_and_hash:提示詞版本與內容雜湊。
  8. raw_model_output:生成器原始輸出,不先覆寫失敗。
  9. validated_structured_output:通過動態 schema 後的結果。
  10. abstention_reason_code:拒答原因與人工接手狀態。
  11. reference_levelprediction_level:參考與最終提案級數。
  12. human_audit_notes:覆核者觀察、機制假設與後續動作分欄保存。

候選分數與最後 context 都要保留,因為「必要規則曾進入候選但在重排序後被丟掉」和「一開始根本沒有取回」是不同修正路徑。原始輸出與驗證後輸出也要分開,否則 schema 修復會把生成失敗從紀錄中抹掉。

如何把工程紀錄寫成結果、討論與限制

一段可信的報告應分成三層,而不是把觀察、解釋與結論混在同一句。

結果只寫可重現觀察

可直接寫:

在 repeat 1 的 1,267 筆 OOF 中,B2 完全相符 685 筆;73 筆為跨兩級以上錯誤。參考第一、二級共 246 筆,其中 191 筆被預測為第三至第五級。生命徵象缺兩項以上切片只有 29 筆,其跨兩級以上錯誤率為 24.14%。

這段話保留資料版本、repeat、分子、分母與事件定義,沒有使用「導致」「因為」或「更安全」。

討論提出機制時要標示證據狀態

目前可以寫:

生命徵象缺兩項以上切片出現較高的跨兩級錯誤率,是值得下一輪預先鎖定驗證的訊號。由於該切片只有 29 筆,且 B2 沒有 RAG trace,本分析無法區分資料缺失、病例組成、場域流程或模型表示造成的差異。

不能寫:

模型因為缺失值而檢索錯誤。

後一句同時聲稱因果與檢索機制,但目前兩者都沒有足夠證據。

限制要對應結論邊界

本篇限制至少包括:事後切片、沒有病患識別碼、兩所急診的代表性、KTAS 不可直接改名成台灣急診檢傷與急迫度分級量表(Taiwan Triage and Acuity Scale, TTAS)、小樣本切片、缺少病患 P0 trace,以及尚未完成外部驗證。

外部效度尤其不能由內部切片推定。來源研究也提醒不同檢傷量表與醫療環境需謹慎套用;本篇的 KTAS 結果不能直接宣稱適用台灣急診,更不能直接改成 TTAS 成果。PLOS ONE:KTAS 研究背景與限制

條件式優勢比「模型全面更好」更接近結果

B2 相對 B0 的優勢是有條件的:它修正 305 筆原本錯誤,也弄壞 107 筆原本正確;在第 3 級完全相符率為 78.03%,在高急迫第一、二級只有 17.89%;不同缺失帶與場域的描述性結果也不同。

因此較精確的敘述是:

在這份資料的 repeat 1 中,B2 相較 B0 提高整體完全相符率並減少跨兩級錯誤,但仍保留大量高急迫級嚴重不足,且在部分小型切片中錯誤率較高。這些觀察只用來產生下一輪預先鎖定的驗證問題。

它不是:

B2 已證明比 B0 安全,也能推廣到其他醫院。

後一句需要預先鎖定的安全界值、完整 P0 對照、病患層級相依處理與外部資料;Day 29 都尚未提供。

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

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

本篇沿用 Day 11、14、15 的欄位與缺失契約、Day 26 已完成且留在本機的 6,335 列 OOF、Day 28 的 repeat 1 聚合,以及 Day 15 留在本機的模型輸入與稽核分層 CSV。以下是 Day 29 新增或修改後的完整錯誤稽核契約、七筆作者合成 evidence trace、evaluation 子套件、runner、十二項測試與四張技術圖生成程式。公開聚合、20 筆私人固定案例與 run manifest 由執行入口產生;私人案例只留在 gitignored run 目錄,Image 2 開場圖則是文章資產,不是錯誤分析程式的輸入。

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

mkdir -p configs/evaluation tests/fixtures src/triage_rag/evaluation scripts scripts/figures/day-29 tests results/public results/runs/day-29 articles/assets/day-29

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

檔案 1:建立 configs/evaluation/day-29-error-audit.json

鎖定 repeat 1、五級錯誤方向、六組切片、小於五筆抑制、四種私人抽樣優先序、七類合成 Evidence Trace、病患 P0 null 邊界與輸出位置。

請在文字編輯器建立 configs/evaluation/day-29-error-audit.json,貼入以下完整內容並儲存:

{
  "schema_version": 1,
  "experiment_id": "day-29-repeat1-error-slices-and-evidence-trace",
  "scope": "exploratory_error_analysis_of_already_observed_day26_repeat1_oof_predictions_plus_synthetic_component_trace_teaching_cases",
  "sources": {
    "day11_field_policy_path": "configs/data/day-11-ktas-field-policy.json",
    "day11_field_policy_sha256": "32a9f0cf054a094f076bc79935175db85d99723b0ff5ddffc26d5cb6ab89237f",
    "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",
    "day26_baseline_contract_path": "configs/baselines/day-26-fair-baselines.json",
    "day26_baseline_contract_sha256": "ff6c36653da05e83276bd1f5ee671069afa5283b19a94e7103228de2a2f88dd6",
    "day28_public_path": "results/public/day-28-safety-statistics.json",
    "day28_public_sha256": "d58ee14eb7b36b56007768c023c91182ab631d0eb0d6c5979ff1945baed4adde",
    "synthetic_trace_fixture_path": "tests/fixtures/day-29-synthetic-evidence-traces.json",
    "synthetic_trace_fixture_sha256": "615135a27640b61cbd0e362d4f68ca4534f2d9ef2022d84cc240d8afe8a2345e",
    "day26_private_result_glob": "results/runs/day-26/*/day-26-oof-baseline-results.json",
    "model_inputs_path": "data/interim/ktas-v1/model-input-candidates.csv",
    "audit_strata_path": "data/interim/ktas-v1/audit-strata.csv"
  },
  "row_contract": {
    "analysis_repeat_id": 1,
    "row_key": "record_index",
    "repeat_key": "repeat_id",
    "reference_field": "reference_level",
    "comparator_prediction_field": "B0_predicted_level",
    "candidate_prediction_field": "B2_predicted_level",
    "valid_levels": [1, 2, 3, 4, 5],
    "expected_record_count": 1267,
    "required_model_input_fields": [
      "Sex",
      "Age",
      "Arrival mode",
      "Injury",
      "Chief_complain",
      "Mental",
      "Pain",
      "NRS_pain",
      "SBP",
      "DBP",
      "HR",
      "RR",
      "BT",
      "Saturation"
    ],
    "required_audit_fields": ["Group", "Patients number per hour"],
    "forbidden_model_input_fields": [
      "KTAS_RN",
      "KTAS_expert",
      "Error_group",
      "mistriage",
      "Diagnosis in ED",
      "Disposition",
      "Length of stay_min",
      "KTAS duration_min"
    ],
    "larger_prediction_number_means": "less_urgent",
    "undertriage_definition": "predicted_level_greater_than_reference_level",
    "severe_undertriage_definition": "reference_in_1_2_and_prediction_in_3_4_5",
    "two_or_more_level_error_definition": "absolute_prediction_minus_reference_at_least_2",
    "patient_identifier_available": false,
    "patient_level_independence_claim_allowed": false
  },
  "slice_policy": {
    "public_minimum_cell_size": 5,
    "secondary_suppression_required": true,
    "secondary_suppression_rule": "when_any_primary_cell_is_below_k_also_suppress_the_smallest_remaining_cell_in_that_exhaustive_table_to_prevent_subtraction",
    "vital_fields": ["SBP", "DBP", "HR", "RR", "BT", "Saturation"],
    "slice_columns": [
      "reference_acuity_group",
      "vital_missing_band",
      "pain_state",
      "mental_response",
      "site_group",
      "injury_status"
    ],
    "category_labels": {
      "reference_acuity_group": {
        "high_1_2": "高急迫級(1–2 級)",
        "middle_3": "第 3 級",
        "lower_4_5": "較低急迫級(4–5 級)"
      },
      "vital_missing_band": {
        "none": "六項生命徵象皆有值",
        "one": "缺一項生命徵象",
        "two_or_more": "缺兩項以上生命徵象"
      },
      "pain_state": {
        "no_pain_nrs_not_applicable": "無疼痛,NRS 不適用",
        "pain_nrs_observed": "有疼痛且 NRS 已記錄",
        "pain_nrs_unknown": "有疼痛但 NRS 未記錄"
      },
      "mental_response": {
        "alert": "清醒",
        "verbal": "對聲音反應",
        "pain": "對疼痛反應",
        "unresponsive": "無反應"
      },
      "site_group": {
        "local_ed": "地方急診",
        "regional_ed": "區域急診"
      },
      "injury_status": {
        "non_injury": "非外傷",
        "injury": "外傷"
      }
    }
  },
  "private_case_sampling": {
    "selection_is_deterministic": true,
    "sort_key": "record_index",
    "cases_per_priority": 5,
    "without_replacement_across_priorities": true,
    "priorities": [
      "severe_undertriage",
      "two_or_more_level_error_excluding_severe_undertriage",
      "b2_broke_b0_exact",
      "b2_fixed_b0_exact"
    ],
    "private_fields": [
      "record_index",
      "reference_level",
      "B0_predicted_level",
      "B2_predicted_level",
      "B0_direction",
      "B2_direction",
      "B2_absolute_error",
      "transition",
      "Sex",
      "Age",
      "Arrival mode",
      "Injury",
      "Chief_complain",
      "Mental",
      "Pain",
      "NRS_pain",
      "SBP",
      "DBP",
      "HR",
      "RR",
      "BT",
      "Saturation",
      "Group",
      "Patients number per hour",
      "reference_acuity_group",
      "vital_missing_band",
      "pain_state",
      "mental_response",
      "site_group",
      "injury_status"
    ],
    "public_patient_rows_allowed": false,
    "public_chief_complaint_allowed": false,
    "run_output_is_gitignored": true
  },
  "synthetic_trace_contract": {
    "required_fields": [
      "case_id",
      "synthetic_only",
      "required_evidence_rule_ids",
      "retrieved_rule_ids",
      "answer_matches_synthetic_reference",
      "abstained",
      "abstention_expected",
      "expected_error_class"
    ],
    "allowed_error_classes": [
      "end_to_end_supported",
      "retrieval_correct_generation_error",
      "retrieval_error_coincidental_correct",
      "retrieval_error",
      "justified_abstention",
      "unnecessary_abstention",
      "missed_abstention"
    ],
    "expected_case_count": 7,
    "patient_records_used": false,
    "model_outputs_used": false,
    "statistics_eligible": false
  },
  "patient_evidence_trace_availability": {
    "status": "not_available_for_day26_b2",
    "reason": "B2 是傳統序位 Ridge 基準;Day 26 OOF 沒有 query、候選、rule_id、分數、prompt、引用或生成輸出。",
    "must_not_infer_retrieval_or_generation_failure_from_final_level": true,
    "future_p0_required_fields": [
      "case_trace_id",
      "input_snapshot_sha256",
      "query_text_or_hash",
      "candidate_rule_ids_with_scores_and_origins",
      "gate_decisions",
      "final_context_rule_ids",
      "prompt_version_and_hash",
      "raw_model_output",
      "validated_structured_output",
      "abstention_reason_code",
      "reference_level",
      "prediction_level",
      "human_audit_notes"
    ]
  },
  "claim_policy": {
    "analysis_is_post_hoc_exploratory": true,
    "day26_results_were_observed_before_slice_analysis": true,
    "slice_differences_are_descriptive_not_causal": true,
    "small_cells_are_suppressed_not_interpreted": true,
    "normal_vital_signs_plus_high_risk_complaint_status": "not_computable_without_a_prelocked_complaint_risk_ontology_and_complete_patient_rag_trace",
    "negation_and_rare_symptom_status": "not_computable_without_a_prelocked_text_annotation_protocol",
    "clinical_safety_or_deployment_claim_allowed": false,
    "decision_support_only": true
  },
  "outputs": {
    "public_summary_path": "results/public/day-29-error-audit.json",
    "run_output_root": "results/runs/day-29",
    "private_audit_filename": "day-29-private-error-audit.json"
  }
}

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

檔案 2:建立 tests/fixtures/day-29-synthetic-evidence-traces.json

保存七筆作者合成工程 trace,逐類覆蓋端到端支持、檢索、生成、碰巧答對、合理拒答、不必要拒答與漏掉拒答;不含病患列或模型輸出。

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

[
  {
    "case_id": "TRACE-01",
    "synthetic_only": true,
    "required_evidence_rule_ids": ["SYN-RULE-A"],
    "retrieved_rule_ids": ["SYN-RULE-A", "SYN-RULE-B"],
    "answer_matches_synthetic_reference": true,
    "abstained": false,
    "abstention_expected": false,
    "expected_error_class": "end_to_end_supported"
  },
  {
    "case_id": "TRACE-02",
    "synthetic_only": true,
    "required_evidence_rule_ids": ["SYN-RULE-A"],
    "retrieved_rule_ids": ["SYN-RULE-A"],
    "answer_matches_synthetic_reference": false,
    "abstained": false,
    "abstention_expected": false,
    "expected_error_class": "retrieval_correct_generation_error"
  },
  {
    "case_id": "TRACE-03",
    "synthetic_only": true,
    "required_evidence_rule_ids": ["SYN-RULE-C"],
    "retrieved_rule_ids": ["SYN-RULE-D"],
    "answer_matches_synthetic_reference": true,
    "abstained": false,
    "abstention_expected": false,
    "expected_error_class": "retrieval_error_coincidental_correct"
  },
  {
    "case_id": "TRACE-04",
    "synthetic_only": true,
    "required_evidence_rule_ids": ["SYN-RULE-C"],
    "retrieved_rule_ids": ["SYN-RULE-D"],
    "answer_matches_synthetic_reference": false,
    "abstained": false,
    "abstention_expected": false,
    "expected_error_class": "retrieval_error"
  },
  {
    "case_id": "TRACE-05",
    "synthetic_only": true,
    "required_evidence_rule_ids": [],
    "retrieved_rule_ids": [],
    "answer_matches_synthetic_reference": null,
    "abstained": true,
    "abstention_expected": true,
    "expected_error_class": "justified_abstention"
  },
  {
    "case_id": "TRACE-06",
    "synthetic_only": true,
    "required_evidence_rule_ids": ["SYN-RULE-E"],
    "retrieved_rule_ids": ["SYN-RULE-E"],
    "answer_matches_synthetic_reference": null,
    "abstained": true,
    "abstention_expected": false,
    "expected_error_class": "unnecessary_abstention"
  },
  {
    "case_id": "TRACE-07",
    "synthetic_only": true,
    "required_evidence_rule_ids": [],
    "retrieved_rule_ids": [],
    "answer_matches_synthetic_reference": false,
    "abstained": false,
    "abstention_expected": true,
    "expected_error_class": "missed_abstention"
  }
]

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

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

更新 evaluation 子套件入口,保留 Day 28 統計函式並公開 Day 29 對齊、切片、固定抽樣與合成 trace 分類函式。

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

"""Evaluation utilities for ordinal triage safety analyses."""

from .error_audit import (
    ErrorAuditContractError,
    build_repeat_error_frame,
    classify_synthetic_trace,
    evaluate_synthetic_traces,
    select_private_audit_cases,
    summarize_error_frame,
    summarize_slices,
)

from .safety_statistics import (
    EvaluationContractError,
    classification_metrics,
    exact_mcnemar,
    holm_adjust,
    paired_record_bootstrap,
    risk_coverage_curve,
    validate_day28_inputs,
)

__all__ = [
    "ErrorAuditContractError",
    "build_repeat_error_frame",
    "classify_synthetic_trace",
    "EvaluationContractError",
    "evaluate_synthetic_traces",
    "classification_metrics",
    "exact_mcnemar",
    "holm_adjust",
    "paired_record_bootstrap",
    "risk_coverage_curve",
    "select_private_audit_cases",
    "summarize_error_frame",
    "summarize_slices",
    "validate_day28_inputs",
]

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

檔案 4:建立 src/triage_rag/evaluation/error_audit.py

驗證 OOF 與兩份 CSV 的一對一對齊,實作五級方向、缺失與疼痛語意、聚合、小格抑制、私人固定抽樣及七類 Evidence Trace 分類。

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

"""Repeat-1 error slicing and synthetic evidence-trace taxonomy for Day 29."""

from __future__ import annotations

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

import numpy as np
import pandas as pd


JsonObject = dict[str, Any]


class ErrorAuditContractError(ValueError):
    """Raised when Day 29 inputs or synthetic traces violate the contract."""


def _rate(numerator: int, denominator: int) -> float | None:
    return round(numerator / denominator, 6) if denominator else None


def _direction(reference: pd.Series, predicted: pd.Series) -> pd.Series:
    result = pd.Series("exact", index=reference.index, dtype="object")
    result.loc[predicted > reference] = "undertriage"
    result.loc[predicted < reference] = "overtriage"
    return result


def _require_columns(frame: pd.DataFrame, fields: Sequence[str], name: str) -> None:
    missing = [field for field in fields if field not in frame.columns]
    if missing:
        raise ErrorAuditContractError(f"{name} 缺少欄位:{missing}")


def _validate_unique_keys(
    frame: pd.DataFrame, key: str, expected_count: int, name: str
) -> None:
    if len(frame) != expected_count:
        raise ErrorAuditContractError(
            f"{name} 應有 {expected_count} 列,實際為 {len(frame)} 列"
        )
    if frame[key].isna().any() or frame[key].duplicated().any():
        raise ErrorAuditContractError(f"{name} 的 {key} 必須非空且唯一")


def build_repeat_error_frame(
    oof_rows: Sequence[Mapping[str, Any]],
    model_inputs: pd.DataFrame,
    audit_strata: pd.DataFrame,
    contract: Mapping[str, Any],
) -> tuple[pd.DataFrame, JsonObject]:
    """Validate, align, and derive repeat-specific error and audit-slice fields."""

    row_contract = contract["row_contract"]
    row_key = str(row_contract["row_key"])
    repeat_key = str(row_contract["repeat_key"])
    reference_field = str(row_contract["reference_field"])
    comparator_field = str(row_contract["comparator_prediction_field"])
    candidate_field = str(row_contract["candidate_prediction_field"])
    repeat_id = int(row_contract["analysis_repeat_id"])
    expected_count = int(row_contract["expected_record_count"])
    required_oof = [
        row_key,
        repeat_key,
        reference_field,
        comparator_field,
        candidate_field,
    ]
    oof = pd.DataFrame(oof_rows)
    _require_columns(oof, required_oof, "Day 26 OOF")
    selected = oof.loc[oof[repeat_key].astype(int) == repeat_id, required_oof].copy()
    _validate_unique_keys(selected, row_key, expected_count, "repeat 1 OOF")

    levels = set(int(value) for value in row_contract["valid_levels"])
    for field in [reference_field, comparator_field, candidate_field]:
        numeric = pd.to_numeric(selected[field], errors="coerce")
        if numeric.isna().any() or not set(numeric.astype(int)).issubset(levels):
            raise ErrorAuditContractError(f"{field} 含空值或五級契約外值")
        selected[field] = numeric.astype(int)

    required_inputs = [
        row_key, *[str(value) for value in row_contract["required_model_input_fields"]]
    ]
    required_audit = [
        row_key, *[str(value) for value in row_contract["required_audit_fields"]]
    ]
    _require_columns(model_inputs, required_inputs, "模型輸入表")
    _require_columns(audit_strata, required_audit, "稽核分層表")
    forbidden = set(str(value) for value in row_contract["forbidden_model_input_fields"])
    present_forbidden = sorted(forbidden.intersection(model_inputs.columns))
    if present_forbidden:
        raise ErrorAuditContractError(f"模型輸入表出現禁止欄位:{present_forbidden}")
    _validate_unique_keys(
        model_inputs.loc[:, required_inputs], row_key, expected_count, "模型輸入表"
    )
    _validate_unique_keys(
        audit_strata.loc[:, required_audit], row_key, expected_count, "稽核分層表"
    )

    frame = selected.merge(
        model_inputs.loc[:, required_inputs], on=row_key, how="inner", validate="one_to_one"
    ).merge(
        audit_strata.loc[:, required_audit],
        on=row_key,
        how="inner",
        validate="one_to_one",
    )
    if len(frame) != expected_count:
        raise ErrorAuditContractError("OOF、模型輸入與稽核分層的 record_index 無法完整對齊")
    frame = frame.sort_values(row_key).reset_index(drop=True)

    reference = frame[reference_field]
    comparator = frame[comparator_field]
    candidate = frame[candidate_field]
    frame["B0_direction"] = _direction(reference, comparator)
    frame["B2_direction"] = _direction(reference, candidate)
    frame["B0_absolute_error"] = (comparator - reference).abs().astype(int)
    frame["B2_absolute_error"] = (candidate - reference).abs().astype(int)
    frame["B0_severe_undertriage"] = reference.isin([1, 2]) & comparator.isin([3, 4, 5])
    frame["B2_severe_undertriage"] = reference.isin([1, 2]) & candidate.isin([3, 4, 5])

    b0_correct = frame["B0_direction"].eq("exact")
    b2_correct = frame["B2_direction"].eq("exact")
    frame["transition"] = np.select(
        [b0_correct & b2_correct, ~b0_correct & b2_correct, b0_correct & ~b2_correct],
        ["both_correct", "b2_fixed_b0", "b2_broke_b0"],
        default="both_wrong",
    )

    frame["reference_acuity_group"] = np.select(
        [reference.isin([1, 2]), reference.eq(3)],
        ["high_1_2", "middle_3"],
        default="lower_4_5",
    )
    vital_fields = [str(value) for value in contract["slice_policy"]["vital_fields"]]
    frame["vital_missing_count"] = frame[vital_fields].isna().sum(axis=1).astype(int)
    frame["vital_missing_band"] = np.select(
        [frame["vital_missing_count"].eq(0), frame["vital_missing_count"].eq(1)],
        ["none", "one"],
        default="two_or_more",
    )

    pain = pd.to_numeric(frame["Pain"], errors="coerce")
    if pain.isna().any() or not set(pain.astype(int)).issubset({0, 1}):
        raise ErrorAuditContractError("Pain 含空值或契約外代碼")
    frame["pain_state"] = np.select(
        [pain.eq(0), pain.eq(1) & frame["NRS_pain"].notna()],
        ["no_pain_nrs_not_applicable", "pain_nrs_observed"],
        default="pain_nrs_unknown",
    )
    code_maps = {
        "Mental": {1: "alert", 2: "verbal", 3: "pain", 4: "unresponsive"},
        "Group": {1: "local_ed", 2: "regional_ed"},
        "Injury": {1: "non_injury", 2: "injury"},
    }
    output_fields = {
        "Mental": "mental_response",
        "Group": "site_group",
        "Injury": "injury_status",
    }
    for source, mapping in code_maps.items():
        numeric = pd.to_numeric(frame[source], errors="coerce")
        if numeric.isna().any() or not set(numeric.astype(int)).issubset(mapping):
            raise ErrorAuditContractError(f"{source} 含空值或契約外代碼")
        frame[output_fields[source]] = numeric.astype(int).map(mapping)

    audit = {
        "analysis_repeat_id": repeat_id,
        "record_count": len(frame),
        "row_key": row_key,
        "one_oof_row_per_record": True,
        "model_inputs_aligned_one_to_one": True,
        "audit_strata_aligned_one_to_one": True,
        "forbidden_model_input_fields_absent": True,
        "patient_identifier_available": False,
        "patient_level_independence_claim_allowed": False,
    }
    return frame, audit


def _method_metrics(frame: pd.DataFrame, method: str) -> JsonObject:
    direction = frame[f"{method}_direction"]
    severe_reference = frame["reference_level"].isin([1, 2])
    severe_under = frame[f"{method}_severe_undertriage"]
    two_plus = frame[f"{method}_absolute_error"].ge(2)
    count = len(frame)
    exact_count = int(direction.eq("exact").sum())
    under_count = int(direction.eq("undertriage").sum())
    over_count = int(direction.eq("overtriage").sum())
    severe_count = int(severe_reference.sum())
    severe_under_count = int(severe_under.sum())
    two_plus_count = int(two_plus.sum())
    return {
        "record_count": count,
        "exact_count": exact_count,
        "exact_rate": _rate(exact_count, count),
        "undertriage_count": under_count,
        "undertriage_rate": _rate(under_count, count),
        "overtriage_count": over_count,
        "overtriage_rate": _rate(over_count, count),
        "two_or_more_level_error_count": two_plus_count,
        "two_or_more_level_error_rate": _rate(two_plus_count, count),
        "severe_reference_count": severe_count,
        "severe_undertriage_count": severe_under_count,
        "severe_undertriage_rate": _rate(severe_under_count, severe_count),
    }


def summarize_error_frame(frame: pd.DataFrame) -> JsonObject:
    """Return overall B0/B2 metrics and exact-correctness transitions."""

    transitions = Counter(str(value) for value in frame["transition"])
    return {
        "record_count": len(frame),
        "B0": _method_metrics(frame, "B0"),
        "B2": _method_metrics(frame, "B2"),
        "exact_correctness_transitions": {
            key: {
                "count": int(transitions[key]),
                "rate": _rate(int(transitions[key]), len(frame)),
            }
            for key in ["both_correct", "b2_fixed_b0", "b2_broke_b0", "both_wrong"]
        },
    }


def summarize_slices(frame: pd.DataFrame, contract: Mapping[str, Any]) -> JsonObject:
    """Aggregate declared slices and suppress every public cell smaller than k."""

    policy = contract["slice_policy"]
    minimum = int(policy["public_minimum_cell_size"])
    if minimum < 2:
        raise ErrorAuditContractError("public_minimum_cell_size 必須至少為 2")
    output: JsonObject = {}
    for column in policy["slice_columns"]:
        labels = policy["category_labels"][column]
        observed = set(str(value) for value in frame[column].dropna().unique())
        unknown = sorted(observed.difference(labels))
        if unknown:
            raise ErrorAuditContractError(f"{column} 出現未登錄類別:{unknown}")
        counts = {
            str(category): int(frame[column].eq(category).sum())
            for category in labels
        }
        primary_suppressed = {
            category for category, count in counts.items() if count < minimum
        }
        secondary_suppressed: set[str] = set()
        if primary_suppressed and policy.get("secondary_suppression_required") is True:
            remaining = [
                category for category in labels if category not in primary_suppressed
            ]
            if remaining:
                secondary_suppressed.add(
                    min(remaining, key=lambda category: (counts[category], category))
                )
        rows: list[JsonObject] = []
        for category, label in labels.items():
            selected = frame.loc[frame[column].eq(category)]
            count = len(selected)
            suppression_reason = None
            if category in primary_suppressed:
                suppression_reason = "primary_below_minimum_cell_size"
            elif category in secondary_suppressed:
                suppression_reason = "secondary_to_prevent_subtraction"
            if suppression_reason is not None:
                rows.append(
                    {
                        "category": category,
                        "label": label,
                        "suppressed": True,
                        "suppression_reason": suppression_reason,
                        "record_count": None,
                        "record_count_display": (
                            f"<{minimum}"
                            if suppression_reason == "primary_below_minimum_cell_size"
                            else "已抑制"
                        ),
                        "B0": None,
                        "B2": None,
                    }
                )
            else:
                rows.append(
                    {
                        "category": category,
                        "label": label,
                        "suppressed": False,
                        "suppression_reason": None,
                        "record_count": count,
                        "record_count_display": str(count),
                        "B0": _method_metrics(selected, "B0"),
                        "B2": _method_metrics(selected, "B2"),
                    }
                )
        output[str(column)] = rows
    return {
        "minimum_public_cell_size": minimum,
        "suppression_rule": (
            "cells_below_k_and_one_complementary_cell_are_not_published; "
            "all_suppressed_cell_metrics_are_null"
        ),
        "secondary_suppression_required": bool(
            policy.get("secondary_suppression_required")
        ),
        "tables": output,
    }


def _json_value(value: Any) -> Any:
    if value is None or pd.isna(value):
        return None
    if isinstance(value, np.generic):
        return value.item()
    return value


def select_private_audit_cases(
    frame: pd.DataFrame, contract: Mapping[str, Any]
) -> tuple[list[JsonObject], JsonObject]:
    """Select deterministic non-overlapping private cases using locked priorities."""

    policy = contract["private_case_sampling"]
    row_key = str(contract["row_contract"]["row_key"])
    per_priority = int(policy["cases_per_priority"])
    if per_priority <= 0:
        raise ErrorAuditContractError("cases_per_priority 必須大於零")
    masks = {
        "severe_undertriage": frame["B2_severe_undertriage"],
        "two_or_more_level_error_excluding_severe_undertriage": frame[
            "B2_absolute_error"
        ].ge(2)
        & ~frame["B2_severe_undertriage"],
        "b2_broke_b0_exact": frame["transition"].eq("b2_broke_b0"),
        "b2_fixed_b0_exact": frame["transition"].eq("b2_fixed_b0"),
    }
    unknown = [priority for priority in policy["priorities"] if priority not in masks]
    if unknown:
        raise ErrorAuditContractError(f"未知的私人抽樣優先序:{unknown}")

    selected_keys: set[int] = set()
    private_rows: list[JsonObject] = []
    counts: JsonObject = {}
    fields = [str(value) for value in policy["private_fields"]]
    _require_columns(frame, fields, "私人稽核表")
    for priority in policy["priorities"]:
        eligible = frame.loc[masks[priority]].sort_values(row_key)
        population_count = len(eligible)
        eligible = eligible.loc[~eligible[row_key].astype(int).isin(selected_keys)]
        chosen = eligible.head(per_priority)
        counts[str(priority)] = {
            "population_count_before_cross_priority_deduplication": population_count,
            "available_after_cross_priority_deduplication": len(eligible),
            "selected_count": len(chosen),
        }
        for _, row in chosen.iterrows():
            record_key = int(row[row_key])
            selected_keys.add(record_key)
            record = {
                field: _json_value(row[field])
                for field in fields
            }
            record["audit_case_id"] = f"PRIVATE-{len(private_rows) + 1:03d}"
            record["selection_priority"] = str(priority)
            private_rows.append(record)
    return private_rows, {
        "selection_is_deterministic": True,
        "sort_key": row_key,
        "cases_per_priority": per_priority,
        "without_replacement_across_priorities": True,
        "selected_case_count": len(private_rows),
        "priority_counts": counts,
        "public_patient_rows_allowed": False,
        "public_chief_complaint_allowed": False,
    }


def classify_synthetic_trace(trace: Mapping[str, Any]) -> tuple[str, JsonObject]:
    """Classify one synthetic trace without guessing missing patient RAG evidence."""

    required = trace.get("required_evidence_rule_ids")
    retrieved = trace.get("retrieved_rule_ids")
    if not isinstance(required, list) or not isinstance(retrieved, list):
        raise ErrorAuditContractError("合成 trace 的 required/retrieved rule IDs 必須是陣列")
    if any(not isinstance(value, str) or not value for value in [*required, *retrieved]):
        raise ErrorAuditContractError("合成 trace 的 rule_id 必須是非空字串")
    if len(required) != len(set(required)) or len(retrieved) != len(set(retrieved)):
        raise ErrorAuditContractError("合成 trace 的 rule_id 不得重複")
    for field in ["abstained", "abstention_expected"]:
        if not isinstance(trace.get(field), bool):
            raise ErrorAuditContractError(f"合成 trace 的 {field} 必須是布林值")

    abstained = bool(trace["abstained"])
    abstention_expected = bool(trace["abstention_expected"])
    answer_match = trace.get("answer_matches_synthetic_reference")
    if abstained:
        if answer_match is not None:
            raise ErrorAuditContractError("拒答案例不能同時宣稱答案正確或錯誤")
        category = "justified_abstention" if abstention_expected else "unnecessary_abstention"
    elif abstention_expected:
        if not isinstance(answer_match, bool):
            raise ErrorAuditContractError("未拒答案例必須提供合成參考答案比對")
        category = "missed_abstention"
    else:
        if not isinstance(answer_match, bool):
            raise ErrorAuditContractError("已回答案例必須提供合成參考答案比對")
        retrieval_hit = set(required).issubset(set(retrieved))
        if retrieval_hit and answer_match:
            category = "end_to_end_supported"
        elif retrieval_hit:
            category = "retrieval_correct_generation_error"
        elif answer_match:
            category = "retrieval_error_coincidental_correct"
        else:
            category = "retrieval_error"
    detail = {
        "required_evidence_count": len(required),
        "retrieved_evidence_count": len(retrieved),
        "all_required_evidence_retrieved": set(required).issubset(set(retrieved)),
        "abstained": abstained,
        "abstention_expected": abstention_expected,
    }
    return category, detail


def evaluate_synthetic_traces(
    traces: Sequence[Mapping[str, Any]], contract: Mapping[str, Any]
) -> JsonObject:
    """Validate the locked synthetic fixture and report taxonomy coverage."""

    policy = contract["synthetic_trace_contract"]
    if len(traces) != int(policy["expected_case_count"]):
        raise ErrorAuditContractError("合成 evidence trace 筆數與契約不一致")
    required_fields = [str(value) for value in policy["required_fields"]]
    allowed = set(str(value) for value in policy["allowed_error_classes"])
    seen: set[str] = set()
    results: list[JsonObject] = []
    counts: Counter[str] = Counter()
    for index, trace in enumerate(traces):
        missing = [field for field in required_fields if field not in trace]
        if missing:
            raise ErrorAuditContractError(f"合成 trace 第 {index} 筆缺少欄位:{missing}")
        case_id = str(trace["case_id"])
        if not case_id or case_id in seen:
            raise ErrorAuditContractError("合成 trace 的 case_id 必須非空且唯一")
        seen.add(case_id)
        if trace.get("synthetic_only") is not True:
            raise ErrorAuditContractError("每筆 evidence trace 都必須標示 synthetic_only=true")
        expected = str(trace["expected_error_class"])
        if expected not in allowed:
            raise ErrorAuditContractError(f"{case_id} 的 expected_error_class 未登錄")
        observed, detail = classify_synthetic_trace(trace)
        passed = observed == expected
        if not passed:
            raise ErrorAuditContractError(
                f"{case_id} 分類不符:預期 {expected},實際 {observed}"
            )
        counts[observed] += 1
        results.append(
            {
                "case_id": case_id,
                "synthetic_only": True,
                "observed_error_class": observed,
                "locked_expectation_passed": True,
                **detail,
            }
        )
    return {
        "case_count": len(results),
        "all_locked_expectations_passed": True,
        "patient_records_used": False,
        "model_outputs_used": False,
        "statistics_eligible": False,
        "class_counts": {name: int(counts[name]) for name in policy["allowed_error_classes"]},
        "case_results": results,
    }

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

檔案 5:建立 scripts/run_day29_error_audit.py

核對 Day 11、14、15、26、28 與 fixture 雜湊,讀取最新相容 OOF,寫出公開摘要、gitignored 私人案例與 run manifest。

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

#!/usr/bin/env python3
"""Run the Day 29 repeat-1 error slices and synthetic evidence-trace audit."""

from __future__ import annotations

import argparse
import glob
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pandas as pd

from triage_rag.evaluation.error_audit import (
    build_repeat_error_frame,
    evaluate_synthetic_traces,
    select_private_audit_cases,
    summarize_error_frame,
    summarize_slices,
)
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/evaluation/day-29-error-audit.json"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="執行 Day 29 repeat 1 錯誤切片、私人固定抽樣與合成 evidence trace 分類。"
    )
    parser.add_argument("--contract", default=DEFAULT_CONTRACT)
    parser.add_argument(
        "--oof",
        default=None,
        help="指定 Day 26 完整 OOF JSON;未指定時使用排序後最後一份相容產物。",
    )
    return parser.parse_args()


def _validate_source_hashes(contract: dict[str, Any]) -> list[str]:
    paths: list[str] = []
    sources = contract["sources"]
    for key, relative_path in sources.items():
        if not key.endswith("_path"):
            continue
        hash_key = f"{key[:-5]}_sha256"
        if hash_key not in sources:
            continue
        path = PROJECT_ROOT / relative_path
        observed = sha256_file(path)
        expected = str(sources[hash_key])
        if observed != expected:
            raise ValueError(f"{relative_path} SHA-256 不符:{observed}")
        paths.append(str(relative_path))
    return paths


def _resolve_oof_path(contract: dict[str, Any], requested: str | None) -> Path:
    if requested:
        candidate = (PROJECT_ROOT / requested).resolve()
        if PROJECT_ROOT.resolve() not in candidate.parents:
            raise ValueError("--oof 必須位於專案資料夾內")
        if not candidate.is_file():
            raise FileNotFoundError(f"找不到 OOF:{candidate}")
        return candidate
    pattern = str(PROJECT_ROOT / contract["sources"]["day26_private_result_glob"])
    candidates = [Path(path) for path in sorted(glob.glob(pattern))]
    if not candidates:
        raise FileNotFoundError(
            "找不到 Day 26 完整 OOF;請先執行 poetry run python scripts/run_day26_baselines.py"
        )
    for candidate in reversed(candidates):
        payload = load_json(candidate)
        if (
            payload.get("schema_version") == 1
            and payload.get("experiment_id") == "day-26-locked-oof-simple-baselines"
        ):
            return candidate
    raise ValueError("找到 Day 26 檔案,但沒有相容的完整 OOF 產物")


def _validate_private_output_ignore(contract: dict[str, Any]) -> None:
    run_root = str(contract["outputs"]["run_output_root"]).strip("/")
    if not run_root.startswith("results/runs/day-29"):
        raise ValueError("Day 29 私人輸出必須位於 results/runs/day-29")
    gitignore_path = PROJECT_ROOT / ".gitignore"
    if not gitignore_path.is_file():
        raise FileNotFoundError("缺少 .gitignore,拒絕寫出逐筆私人稽核")
    patterns = {
        line.strip().rstrip("/")
        for line in gitignore_path.read_text(encoding="utf-8").splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    }
    if "results/runs" not in patterns:
        raise ValueError(".gitignore 必須包含 results/runs/,拒絕寫出逐筆私人稽核")


def _validate_day28_repeat1(overall: dict[str, Any], day28: dict[str, Any]) -> None:
    repeat = next(
        item for item in day28["per_repeat_metrics"] if int(item["repeat_id"]) == 1
    )
    fields = [
        "exact_count",
        "undertriage_count",
        "overtriage_count",
        "two_or_more_level_error_count",
        "severe_undertriage_count",
    ]
    for method in ["B0", "B2"]:
        for field in fields:
            if int(overall[method][field]) != int(repeat[method][field]):
                raise ValueError(f"Day 29 {method}.{field} 與 Day 28 repeat 1 不一致")


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 29 schema_version=1")
    _validate_private_output_ignore(contract)
    source_paths = _validate_source_hashes(contract)
    sources = contract["sources"]
    oof_path = _resolve_oof_path(contract, args.oof)
    oof_payload = load_json(oof_path)
    model_inputs_path = PROJECT_ROOT / sources["model_inputs_path"]
    audit_strata_path = PROJECT_ROOT / sources["audit_strata_path"]
    model_inputs = pd.read_csv(model_inputs_path)
    audit_strata = pd.read_csv(audit_strata_path)
    frame, input_audit = build_repeat

上一篇
Day 28|Accuracy 之外:序位指標、安全終點與配對統計
下一篇
Day 30|完成可重現 Side Project:打包成果、限制與下一站
系列文
30 天打造公開資料版急診檢傷系統:Side Project 與實驗計畫30
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言