Day 17 把九筆可追溯的公開知識整理成平面與階層候選,但候選文件準備好,不代表病患查詢也準備好了。向量嵌入(Vector Embedding,以下簡稱 Embedding)會把文字轉成固定長度的數字;若把資料表裡的呼吸率數字、意識代碼與疼痛缺失狀態原封不動交給模型,模型是否理解欄位名稱、單位與代碼,仍然需要驗證。
今天要處理的就是這段轉換。我們會把同一筆純合成案例寫成原始表示、語意表示與雙重表示,再用本機向量模型產生三個 Embedding。這不是要在今天選出冠軍,而是先建立公平、可重跑且不偷加醫療規則的比較起點。
先看下方的概念示意。左側只有一份輸入,中間的三條路徑只是用不同文字表達同一批觀察值,右側才是三個向量;沒有任何一條路徑可以偷偷改動原始數字。

上圖建立了今天最重要的公平原則:表示方法可以改變文字,不可以改變病患輸入。若語意版把 91 改成別的數字,或雙重版比原始版多用了參考標籤,後面的相似度就已經失去比較意義。
本篇所有案例都是人工建立的合成資料,不是真實病患紀錄,也不代表任何人的診斷或建議。系統仍只定位為離線決策支援實驗,不能取代醫師或護理師的判斷。
完成本篇後,你會得到四項可觀察的產物:
這四項產物會回答「管線能不能安全地執行」,不會回答「哪種表示最好」。後者需要先為數值案例建立有可追溯相關規則的固定檢索測試集;目前部分公開知識不足以提供這種標註,因此不能只看今天的一筆案例,也不能在下一篇自行指定答案。
本篇新名詞較多,先用下表建立共同語言。正文第一次使用時仍會再解釋一次。
| 中文名稱 | 英文全名/縮寫 | 本篇用途 |
|---|---|---|
| 韓國急診檢傷與急迫度分級量表 | Korean Triage and Acuity Scale, KTAS | 本系列研究的五級檢傷制度;數字越小代表急迫度越高 |
| 檢索增強生成 | Retrieval-Augmented Generation, RAG | 先找公開知識,再把證據交給生成模型的整體方法 |
| 向量嵌入 | Vector Embedding, Embedding | 把一段文字轉成固定長度數字,供電腦比較語意接近程度 |
| 稠密向量 | Dense Vector | 多數位置都有數值的向量;今天每筆有 2,560 個位置 |
| 維度 | Dimension | 一個向量包含多少個數字 |
| 餘弦相似度 | Cosine Similarity | 比較兩個向量方向是否接近的分數 |
| L2 範數 | L2 norm/Euclidean norm | 衡量向量長度;今天用它檢查模型輸出是否已正規化 |
| 應用程式介面 | Application Programming Interface, API | 程式向本機模型執行工具 Ollama 傳送文字並接收向量的入口 |
| JavaScript 物件表示法 | JavaScript Object Notation, JSON | 保存可由人與程式共同閱讀的設定與摘要 |
| 逐行 JSON | JSON Lines, JSONL | 一行保存一個 JSON 物件,供程式逐筆讀取 |
| 安全雜湊演算法 256 位元 | Secure Hash Algorithm 256-bit, SHA-256 | 為檔案或文字建立固定摘要,檢查輸入是否變更 |
| 超文字傳輸通訊協定 | Hypertext Transfer Protocol, HTTP | 本機程式與 Ollama API 溝通使用的通訊規則 |
| 統一資源定位符 | Uniform Resource Locator, URL | 指向 API 或事實來源的位置 |
| 原始表示 | Raw Representation | 保留資料表欄位名稱與原始值 |
| 語意表示 | Semantic Representation | 補上中文名稱、單位、代碼意義與缺失狀態 |
| 雙重表示 | Dual Representation | 同時保留原始表示與語意表示 |
向量嵌入(Vector Embedding,以下簡稱 Embedding)會把文字轉成固定長度的稠密向量(Dense Vector)。「稠密」表示向量中的多數位置都有數字;模型不是把每個字直接換成單一代碼,而是用整組數字表示它學到的文字關係。
問題在於,能處理數字字元不等於理解數值的臨床語意。這種辨識數字大小、單位與上下文關係的能力,常被稱為數值理解能力(numeracy)。以下三段輸入對人類指向同一筆觀察,但對模型不是完全相同的字串:
RR=28
呼吸率 28 次/分鐘
RR=28;呼吸率 28 次/分鐘
第一段最接近來源資料,卻要求模型知道 RR 是什麼。第二段對讀者較友善,但如果轉換程式寫錯,就可能遺失欄位代碼或原值。第三段最長,資訊也最重複。哪一種在目前知識庫上更適合,應由固定測試回答,不能先靠直覺指定。
今天用一筆「合成案例:呼吸困難」示範。合成輸入包含意識代碼、疼痛狀態、收縮壓、舒張壓、心率、呼吸率、體溫與血氧飽和度。這些數字只為測試欄位與程式流程而設計,不是 KTAS 判定範例。
資料欄位首次出現時,也要把名稱說完整:
Chief_complain,保存病患到院時描述的主要問題。Mental;本篇只依 Day 11 的欄位政策,把代碼 1 到 4 分別還原成清醒、對聲音有反應、對疼痛有反應與無反應。Pain;0 代表記錄為無疼痛,1 代表記錄為有疼痛。SBP,單位是毫米汞柱(millimetres of mercury, mmHg)。DBP,單位同樣是 mmHg。HR,單位是次/分鐘。RR,單位是次/分鐘。BT,單位是攝氏度(°C)。Saturation,單位是百分比(%)。NRS_pain,有記錄時介於 0 到 10。設定檔中的 null 表示沒有值;原始文字再把它顯示成 <MISSING>,讓「欄位存在但沒有記錄」不會被空字串吞掉。
下圖把程式真正產生的三種文字並排。閱讀時請觀察三件事:數字是否保留、單位是否只在語意版補上,以及任何一欄是否擅自加入「正常」「危險」或 KTAS 級數。

上圖中的三種表示可以這樣理解。
原始表示(Raw Representation)保留欄位識別字與原始值:
Chief_complain=合成案例:呼吸困難 | Mental=1 | Pain=0 | NRS_pain=<MISSING> | SBP=92 | DBP=58 | HR=118 | RR=28 | BT=37.2 | Saturation=91
它的優點是轉換最少,日後容易回頭核對資料表。缺點是 Mental=1、Pain=0 與 <MISSING> 都需要額外的欄位知識才能正確理解。
語意表示(Semantic Representation)把相同輸入改寫成完整句子:
主訴:合成案例:呼吸困難。
意識反應:清醒。
疼痛:記錄為無疼痛;疼痛分數不適用。
生命徵象:收縮壓 92 mmHg;舒張壓 58 mmHg;心率 118 次/分鐘;呼吸率 28 次/分鐘;體溫 37.2 °C;血氧飽和度 91 %。
這一版不是把數值「翻譯成病情」,而是把資料契約已經確定的名稱、單位與代碼意義寫清楚。RR=28 仍然是 28,沒有被改成「偏快」;Saturation=91 也沒有被改成「低血氧」。
雙重表示(Dual Representation)依序串接完整原始表示與完整語意表示。它不是讓兩個模型投票,也不是把數值平均,而是讓同一個 Embedding 模型同時看到來源欄位與白話說明。
雙重表示可能降低欄位代碼不清楚的問題,也可能因文字變長與重複而引入雜訊。因此「資訊較多」仍不等於「檢索一定較好」。
缺失值最容易在語意化時被寫錯。今天至少區分三種情況:
| 原始條件 | 正確語意 | 不能寫成 |
|---|---|---|
Pain=0 且 NRS_pain=null |
記錄為無疼痛;疼痛分數不適用 | 疼痛分數未知或疼痛 0 分 |
Pain=1 且 NRS_pain=null |
記錄為有疼痛;疼痛分數未記錄 | 無疼痛 |
Saturation=null |
血氧飽和度未記錄,不能視為正常 | 血氧正常 |
第一列的「不適用」與第二列的「未知」尤其不能合併。前者有 Pain=0 作為上下文;後者只知道有疼痛,卻不知道分數。若把兩者都補成 0,查詢就會出現資料中不存在的觀察。
韓國急診檢傷與急迫度分級量表(Korean Triage and Acuity Scale, KTAS)是五級制度,但 KTAS 並不是只看一張通用生命徵象表就能完成判定。Day 16 建立的部分公開知識庫只保存可追溯的制度、流程與級數摘要,沒有足以支援所有主訴、年齡與情境的完整數值門檻。
因此,今天不會從一般醫療網站找一組血壓或血氧範圍,再把它改名為 KTAS 規則。那樣會混淆三件不同的事:
程式採用「失敗關閉」(fail closed)政策:正式門檻設定的 rules 保持空陣列;只要沒有來源網址、版本與唯一命中的規則,輸出就只能是「沒有來源規則,因此未解讀」。它不能產生正常、偏高、偏低或 KTAS 級數。
因為未來若取得合法、完整而且可引用的規則,程式仍需要正確處理大於、小於、是否包含邊界與規則重疊。今天用純合成規則測試這個機制,例如:
合成下區間:小於 10
合成上區間:大於等於 10
數值 9.9 應命中下區間,數值 10 應命中上區間;若兩條規則都包含 10,程式必須報錯。這裡的 10 沒有任何臨床意義,只是在測試 < 與 >= 的程式邊界。
本篇鎖定 qwen3-embedding:4b 作為稠密向量模型。選擇理由不是它取得醫療認證,而是我們已經下載並鎖定這個版本,它支援多語文字、查詢任務指令與可調整維度,適合建立可重跑的繁體中文檢索實驗。Qwen 官方模型卡指出 4B 版本最高可輸出 2,560 維向量;本篇固定使用完整的 2,560 維,不在今天比較降維效果。Qwen3-Embedding-4B 官方模型卡
4B 表示模型約有四十億個參數。本機版本使用 Q4_K_M 量化設定;量化(quantization)用較少位元保存權重,以降低記憶體與儲存需求。今天記錄這個識別字是為了重現環境,不是在比較量化方法。
「2,560 維」表示每段文字最後會得到 2,560 個數字。人類不需要替每一維命名;我們要確認的是模型版本、輸入契約、維度、數值有效性與比較方式都固定。
Qwen 官方建議在查詢端加入任務說明,而文件端不必加入。今天的固定查詢說明是:
Given an emergency triage query in Traditional Chinese, retrieve relevant public KTAS knowledge passages. Do not infer unpublished thresholds.
完整查詢格式是 Instruct: 任務說明 加上 Query: 三種表示之一。Day 17 每筆候選中真正送進檢索器的文字欄位叫做 search_text;九筆文件直接沿用這個欄位,不加查詢說明。任務說明使用英文是依官方多語範例固定的實驗契約,不代表病患主訴要翻譯成英文。Qwen3-Embedding 官方實作說明
我們同時固定 truncate=false,也就是禁止模型在輸入過長時靜默截斷;keep_alive=5m 讓模型在最後一次請求後保留載入五分鐘,減少連續兩次呼叫重載模型的時間;單次請求逾時則固定為 180 秒。模型雜湊摘要(digest)是模型內容的長識別字,本篇會核對完整 digest,而不是只核對容易重複或更新的模型名稱。
下圖呈現實際資料流。查詢端三筆與文件端九筆都交給同一個模型,輸出維度也相同;只有查詢端多了固定任務說明。

上圖還有一個容易忽略的設定:truncate=false。Ollama 的 Embedding 應用程式介面(Application Programming Interface, API)預設可以截斷過長輸入;本篇選擇關閉,讓超過模型上限的輸入直接失敗,避免內容在沒有警告時消失。請求送到本機 http://127.0.0.1:11434/api/embed;這個接收請求的位置稱為 API 端點(endpoint),不會把文字交給線上模型服務。Ollama Embed API 官方文件
餘弦相似度(Cosine Similarity)比較兩個向量的方向。公式如下:
$$
\operatorname{cosine}(\mathbf{q},\mathbf{d})=
\frac{\mathbf{q}\cdot\mathbf{d}}
{\lVert\mathbf{q}\rVert_2\lVert\mathbf{d}\rVert_2}
$$
公式中的每個符號都有明確意義:
用二維小例子計算,若 $\mathbf{q}=[1,0]$、$\mathbf{d}=[0.8,0.6]$,內積是 $1\times0.8+0\times0.6=0.8$;兩個向量長度都為 1,因此餘弦相似度是 $0.8/(1\times1)=0.8$。分數越接近 1,代表方向越相似,但不等於內容必然正確,也不等於臨床上更安全。
本篇還檢查每個模型輸出的 L2 範數是否接近 1。這稱為 L2 正規化;容許誤差固定為 0.0001。若維度錯誤、出現非數字、無限值、零向量或長度超出容許誤差,程式會中止,而不是帶著壞向量繼續排名。
讀者不需要知道任何程式碼儲存庫的存在。接下來會提供每個必要檔案的完整內容;請先理解它們的輸入與輸出:
configs/representation/
└── day-18-numeric-semantics.json # 政策、上游雜湊、模型與輸出契約
tests/fixtures/
└── day-18-synthetic-cases.json # 四筆純合成案例
src/triage_rag/representation/
├── __init__.py # Python 子套件入口
├── numeric_semantics.py # 三表示、缺失狀態與門檻安全檢查
└── embedding.py # 本機 Ollama API、向量驗證與相似度
scripts/
└── run_day18_numeric_embeddings.py # 串接所有輸入並寫出產物
tests/
└── test_numeric_representation.py # 十四項不需啟動模型的單元測試
configs/ 是保存可讀設定的資料夾;tests/fixtures/ 放不含真實病患的固定測試輸入;src/ 放可被其他程式匯入的核心邏輯;scripts/ 放讀者直接執行的入口;tests/ 則負責故意餵入合法與錯誤資料,確認安全檢查真的會生效。Poetry 是本系列用來管理 Python 版本、套件與虛擬環境的工具,開始建立檔案前要先完成 Day 13 的環境設定。
Day 18 入口會讀取 Day 11 欄位政策、Day 15 缺失政策、模型鎖定檔、Day 17 九筆平面候選(flat chunks),以及今天的四筆合成案例。Kaggle 資料欄位 KTAS_RN 是護理師登錄級數,KTAS_expert 是專家重新判定級數;兩者都是比較或評估資訊,今天的查詢不得讀取。
入口會把完整 2,560 維向量寫進 results/runs/day-18/,這個資料夾保留在本機;可公開的聚合摘要則寫到 results/public/day-18-numeric-embedding-smoke.json。每次執行也會留下執行清單(run manifest),記錄輸入、輸出、命令、版本與檔案摘要。今天的排名診斷保留前 k 筆(top-k)中的三筆;k 只是要保留的最高排名文件數,不是 KTAS 級數。
接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾,依下列順序建立檔案;每個程式碼區塊都是該檔案的完整內容,不含省略號。
本篇沿用 Day 11 的欄位政策、Day 13 的 Poetry 與重現性工具、Day 15 的缺失政策、Day 17 已建立的九筆 flat chunks,以及先前已下載的 qwen3-embedding:4b。本篇不讀取 Kaggle 病患列或參考標籤;以下是 Day 18 新增的完整設定、四筆合成案例、表示模組、本機 Ollama 用戶端、執行入口與測試。完整向量、公開摘要與 run manifest 都由程式自動產生,不需要手動建立。
先從專案根目錄建立需要的資料夾:
mkdir -p configs/representation tests/fixtures src/triage_rag/representation scripts tests results/public results/runs/day-18
如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案,把對應區塊完整貼入後儲存。
configs/representation/day-18-numeric-semantics.json鎖定上游檔案雜湊、三種數值表示、沒有公開門檻時的關閉政策、本機 embedding 模型契約與輸出路徑。
請在文字編輯器建立 configs/representation/day-18-numeric-semantics.json,貼入以下完整內容並儲存:
{
"schema_version": 1,
"experiment_id": "day-18-numeric-semantic-representation",
"scope": "synthetic_numeric_representation_and_embedding_smoke_not_clinical_evaluation",
"sources": {
"field_policy_path": "configs/data/day-11-ktas-field-policy.json",
"field_policy_sha256": "32a9f0cf054a094f076bc79935175db85d99723b0ff5ddffc26d5cb6ab89237f",
"missingness_policy_path": "configs/data/day-15-quality-and-split-contract.json",
"missingness_policy_sha256": "afef6b89c9ae9a048326271c2ad74649ae0c029f24bbd839581aa46129498cf5",
"model_lock_path": "configs/models/local-ollama-model-lock.json",
"model_lock_sha256": "0cd15cb4d7e735a19cbe439fe9a9344f717029694a4441764f67e396033cc7f4",
"flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
"flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa"
},
"synthetic_cases_path": "tests/fixtures/day-18-synthetic-cases.json",
"synthetic_cases_sha256": "760e9e4f53d070d53368f7ae335662adf81c9d7967acf5d7fb60797f1a68a672",
"embedding_case_id": "day18-synthetic-respiratory",
"representation": {
"required_fields": [
"case_id",
"synthetic_only",
"Chief_complain",
"Mental",
"Pain",
"NRS_pain",
"SBP",
"DBP",
"HR",
"RR",
"BT",
"Saturation"
],
"forbidden_fields": [
"record_index",
"KTAS_RN",
"KTAS_expert",
"Group",
"diagnosis",
"disposition",
"length_of_stay",
"Error_group",
"mistriage"
],
"field_order": [
"SBP",
"DBP",
"HR",
"RR",
"BT",
"Saturation"
],
"field_specs": {
"SBP": {"name_zh": "收縮壓", "unit": "mmHg"},
"DBP": {"name_zh": "舒張壓", "unit": "mmHg"},
"HR": {"name_zh": "心率", "unit": "次/分鐘"},
"RR": {"name_zh": "呼吸率", "unit": "次/分鐘"},
"BT": {"name_zh": "體溫", "unit": "°C"},
"Saturation": {"name_zh": "血氧飽和度", "unit": "%"}
},
"mental_mapping": {
"1": "清醒",
"2": "對聲音有反應",
"3": "對疼痛有反應",
"4": "無反應"
},
"raw_missing_token": "<MISSING>",
"unknown_phrase": "未記錄,不能視為正常",
"pain_not_applicable_phrase": "記錄為無疼痛;疼痛分數不適用",
"pain_unknown_phrase": "記錄為有疼痛;疼痛分數未記錄"
},
"threshold_policy": {
"profile_id": "public-ktas-thresholds-unavailable-v1",
"profile_version": "2026-08-10.v1",
"status": "unavailable_in_current_public_knowledge_base",
"source_url": null,
"rules": [],
"fail_closed": true,
"observed_status": "not_interpreted_no_source_rule",
"missing_status": "missing_not_recorded",
"explanation": "目前部分公開 KTAS 知識庫沒有逐主訴、年齡與完整數值門檻,因此只保留原值、單位與缺失狀態,不標示正常、偏高、偏低或 KTAS 級數。"
},
"embedding": {
"runtime": "Ollama",
"endpoint": "http://127.0.0.1:11434",
"model": "qwen3-embedding:4b",
"expected_model_id_prefix": "df5bd2e3c74c",
"expected_model_digest": "df5bd2e3c74cd8d069d21dc038f1b359fcdc9458fce1c99bd43c9eb1518ff907",
"expected_parameter_size": "4.0B",
"expected_quantization_level": "Q4_K_M",
"dimensions": 2560,
"truncate": false,
"keep_alive": "5m",
"timeout_seconds": 180,
"query_instruction": "Given an emergency triage query in Traditional Chinese, retrieve relevant public KTAS knowledge passages. Do not infer unpublished thresholds.",
"document_instruction": null,
"normalization_check_tolerance": 0.0001
},
"retrieval_smoke": {
"expected_rule_id": "ktas-public-primary-considerations-001",
"top_k": 3,
"status": "single_synthetic_case_diagnostic_not_retrieval_benchmark"
},
"outputs": {
"public_summary_path": "results/public/day-18-numeric-embedding-smoke.json",
"run_output_root": "results/runs/day-18"
}
}
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/fixtures/day-18-synthetic-cases.json提供四筆沒有真實病歷、病患索引或 KTAS 標籤的純合成案例,用來測試數值、缺失與代碼轉譯。
請在文字編輯器建立 tests/fixtures/day-18-synthetic-cases.json,貼入以下完整內容並儲存:
[
{
"case_id": "day18-synthetic-respiratory",
"synthetic_only": true,
"Chief_complain": "合成案例:呼吸困難",
"Mental": 1,
"Pain": 0,
"NRS_pain": null,
"SBP": 92,
"DBP": 58,
"HR": 118,
"RR": 28,
"BT": 37.2,
"Saturation": 91
},
{
"case_id": "day18-synthetic-pain-score-unknown",
"synthetic_only": true,
"Chief_complain": "合成案例:胸部疼痛",
"Mental": 2,
"Pain": 1,
"NRS_pain": null,
"SBP": 118,
"DBP": 74,
"HR": 102,
"RR": 22,
"BT": 36.9,
"Saturation": 97
},
{
"case_id": "day18-synthetic-missing-saturation",
"synthetic_only": true,
"Chief_complain": "合成案例:暈眩",
"Mental": 1,
"Pain": 0,
"NRS_pain": null,
"SBP": 126,
"DBP": 80,
"HR": 86,
"RR": 18,
"BT": 36.7,
"Saturation": null
},
{
"case_id": "day18-synthetic-fever",
"synthetic_only": true,
"Chief_complain": "合成案例:發燒與倦怠",
"Mental": 1,
"Pain": 1,
"NRS_pain": 3,
"SBP": 118,
"DBP": 74,
"HR": 102,
"RR": 22,
"BT": 39.1,
"Saturation": null
}
]
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/representation/__init__.py建立 Day 18 起使用的 representation 子套件。
請在文字編輯器建立 src/triage_rag/representation/__init__.py,貼入以下完整內容並儲存:
"""Representation helpers introduced from Day 18 onward."""
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/representation/numeric_semantics.py驗證合成案例、建立 raw/semantic/dual 三種文字,並在缺少可追溯門檻時拒絕產生正常、異常或級數標籤。
請在文字編輯器建立 src/triage_rag/representation/numeric_semantics.py,貼入以下完整內容並儲存:
"""Build raw, semantic, and dual text without inventing KTAS thresholds."""
from __future__ import annotations
import math
from collections import Counter
from numbers import Real
from typing import Any, Dict, Iterable, List, Mapping, Sequence
from triage_rag.reproducibility import canonical_json_bytes, sha256_bytes
JsonObject = Dict[str, Any]
AVAILABLE_THRESHOLD_STATUSES = {
"available_for_research",
"available_for_synthetic_test_only",
}
class RepresentationContractError(ValueError):
"""Raised when a case or threshold profile breaks the Day 18 contract."""
def _nonempty_text(value: Any, field: str) -> str:
if not isinstance(value, str) or not value.strip():
raise RepresentationContractError(f"{field} 必須是非空字串")
return value.strip()
def _number_or_none(value: Any, field: str) -> float | int | None:
if value is None:
return None
if isinstance(value, bool) or not isinstance(value, Real):
raise RepresentationContractError(f"{field} 必須是數字或 null")
if not math.isfinite(float(value)):
raise RepresentationContractError(f"{field} 必須是有限數字")
return value
def _format_number(value: float | int) -> str:
numeric = float(value)
if numeric.is_integer():
return str(int(numeric))
return f"{numeric:.6f}".rstrip("0").rstrip(".")
def _validate_case(case: Mapping[str, Any], config: Mapping[str, Any]) -> None:
representation = config["representation"]
required = set(representation["required_fields"])
missing = sorted(required - set(case))
if missing:
raise RepresentationContractError(f"合成案例缺少必要欄位:{missing}")
forbidden = sorted(set(representation["forbidden_fields"]).intersection(case))
if forbidden:
raise RepresentationContractError(f"合成案例出現禁止欄位:{forbidden}")
unexpected = sorted(set(case) - required)
if unexpected:
raise RepresentationContractError(f"合成案例出現未定義欄位:{unexpected}")
_nonempty_text(case.get("case_id"), "case_id")
complaint = _nonempty_text(case.get("Chief_complain"), "Chief_complain")
if not complaint.startswith("合成案例:"):
raise RepresentationContractError("Chief_complain 必須以『合成案例:』開頭")
if case.get("synthetic_only") is not True:
raise RepresentationContractError("Day 18 只接受 synthetic_only=true 的案例")
mental = case.get("Mental")
if (
isinstance(mental, bool)
or not isinstance(mental, int)
or str(mental) not in representation["mental_mapping"]
):
raise RepresentationContractError(f"Mental 代碼不支援:{mental}")
pain = case.get("Pain")
if isinstance(pain, bool) or pain not in {0, 1}:
raise RepresentationContractError("Pain 只能是 0 或 1")
nrs = _number_or_none(case.get("NRS_pain"), "NRS_pain")
if nrs is not None and not 0 <= float(nrs) <= 10:
raise RepresentationContractError("NRS_pain 必須介於 0 到 10")
if case.get("Pain") == 0 and nrs is not None:
raise RepresentationContractError("Pain=0 時 NRS_pain 必須是 null,表示不適用")
for field in representation["field_order"]:
value = _number_or_none(case.get(field), field)
if value is not None and float(value) <= 0:
raise RepresentationContractError(f"{field} 已記錄時必須大於 0")
saturation = case.get("Saturation")
if saturation is not None and float(saturation) > 100:
raise RepresentationContractError("Saturation 不可大於 100")
def _rule_matches(value: float, rule: Mapping[str, Any]) -> bool:
lower = rule.get("minimum")
upper = rule.get("maximum")
if lower is not None:
lower = float(lower)
if rule.get("minimum_inclusive", True):
if value < lower:
return False
elif value <= lower:
return False
if upper is not None:
upper = float(upper)
if rule.get("maximum_inclusive", True):
if value > upper:
return False
elif value >= upper:
return False
return True
def interpret_numeric_value(
field: str,
value: float | int | None,
threshold_policy: Mapping[str, Any],
) -> JsonObject:
"""Apply a sourced threshold profile, or fail closed when none is available."""
if value is None:
return {
"field": field,
"status": threshold_policy["missing_status"],
"rule_id": None,
"label": None,
}
status = threshold_policy.get("status")
if status not in AVAILABLE_THRESHOLD_STATUSES:
if threshold_policy.get("fail_closed") is not True:
raise RepresentationContractError("沒有來源門檻時必須 fail_closed=true")
if threshold_policy.get("rules"):
raise RepresentationContractError("門檻狀態不可用時 rules 必須是空陣列")
return {
"field": field,
"status": threshold_policy["observed_status"],
"rule_id": None,
"label": None,
}
source_url = threshold_policy.get("source_url")
profile_version = threshold_policy.get("profile_version")
if not isinstance(source_url, str) or not source_url.strip():
raise RepresentationContractError("可用門檻必須提供 source_url")
if not isinstance(profile_version, str) or not profile_version.strip():
raise RepresentationContractError("可用門檻必須提供 profile_version")
field_rules = [
rule for rule in threshold_policy.get("rules", []) if rule.get("field") == field
]
if not field_rules:
raise RepresentationContractError(f"{field} 沒有可用門檻規則")
matches = [rule for rule in field_rules if _rule_matches(float(value), rule)]
if len(matches) != 1:
raise RepresentationContractError(
f"{field}={value} 必須恰好命中一條規則,目前命中 {len(matches)} 條"
)
rule = matches[0]
return {
"field": field,
"status": "interpreted_with_sourced_rule",
"rule_id": _nonempty_text(rule.get("rule_id"), "rule_id"),
"label": _nonempty_text(rule.get("label"), "label"),
"source_url": source_url,
"profile_version": profile_version,
}
def _raw_text(case: Mapping[str, Any], config: Mapping[str, Any]) -> str:
representation = config["representation"]
missing = representation["raw_missing_token"]
fields = ["Chief_complain", "Mental", "Pain", "NRS_pain", *representation["field_order"]]
values = []
for field in fields:
value = case[field]
rendered = missing if value is None else str(value)
values.append(f"{field}={rendered}")
return " | ".join(values)
def _pain_text(case: Mapping[str, Any], config: Mapping[str, Any]) -> str:
representation = config["representation"]
if case["Pain"] == 0:
return representation["pain_not_applicable_phrase"]
nrs = case["NRS_pain"]
if nrs is None:
return representation["pain_unknown_phrase"]
return f"記錄為有疼痛;疼痛數字評定量表分數為 {_format_number(nrs)}/10"
def _semantic_text(case: Mapping[str, Any], config: Mapping[str, Any]) -> str:
representation = config["representation"]
vital_parts = []
for field in representation["field_order"]:
spec = representation["field_specs"][field]
value = case[field]
if value is None:
vital_parts.append(f"{spec['name_zh']}{representation['unknown_phrase']}")
else:
vital_parts.append(
f"{spec['name_zh']} {_format_number(value)} {spec['unit']}"
)
return "\n".join(
[
f"主訴:{case['Chief_complain']}。",
f"意識反應:{representation['mental_mapping'][str(case['Mental'])]}。",
f"疼痛:{_pain_text(case, config)}。",
"生命徵象:" + ";".join(vital_parts) + "。",
]
)
def build_case_representations(
case: Mapping[str, Any], config: Mapping[str, Any]
) -> JsonObject:
"""Build three views that preserve the same observed inputs."""
_validate_case(case, config)
raw = _raw_text(case, config)
semantic = _semantic_text(case, config)
dual = f"原始欄位:\n{raw}\n\n語意說明:\n{semantic}"
threshold_status = {
field: interpret_numeric_value(field, case[field], config["threshold_policy"])
for field in config["representation"]["field_order"]
}
values = {
field: case[field]
for field in [
"Mental",
"Pain",
"NRS_pain",
*config["representation"]["field_order"],
]
}
texts = {"raw": raw, "semantic": semantic, "dual": dual}
return {
"schema_version": 1,
"case_id": case["case_id"],
"synthetic_only": True,
"source_case_sha256": sha256_bytes(canonical_json_bytes(case)),
"observed_values": values,
"threshold_status": threshold_status,
"representations": {
name: {
"text": text,
"text_sha256": sha256_bytes(text.encode("utf-8")),
}
for name, text in texts.items()
},
}
def build_all_case_representations(
cases: Sequence[Mapping[str, Any]], config: Mapping[str, Any]
) -> List[JsonObject]:
if not isinstance(cases, list) or not cases:
raise RepresentationContractError("合成案例必須是非空陣列")
case_ids = [str(case.get("case_id")) for case in cases]
duplicates = sorted(
case_id for case_id, count in Counter(case_ids).items() if count > 1
)
if duplicates:
raise RepresentationContractError(f"case_id 不可重複:{duplicates}")
return [build_case_representations(case, config) for case in cases]
def representation_texts(record: Mapping[str, Any]) -> Dict[str, str]:
return {
name: record["representations"][name]["text"]
for name in ("raw", "semantic", "dual")
}
def observed_fields_with_status(
record: Mapping[str, Any], status: str
) -> Iterable[str]:
return (
field
for field, item in record["threshold_status"].items()
if item["status"] == status
)
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
src/triage_rag/representation/embedding.py以 Python 標準函式庫呼叫本機 Ollama,驗證模型、向量數量、2,560 維、有限數值與 L2 正規化,再計算 cosine similarity。
請在文字編輯器建立 src/triage_rag/representation/embedding.py,貼入以下完整內容並儲存:
"""Minimal local Ollama embedding client with strict response validation."""
from __future__ import annotations
import json
import math
from typing import Any, Dict, List, Mapping, Sequence, Tuple
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen
from triage_rag.reproducibility import canonical_json_bytes, sha256_bytes
JsonObject = Dict[str, Any]
class OllamaEmbeddingError(RuntimeError):
"""Raised when the local Ollama service violates the embedding contract."""
def _local_endpoint(endpoint: str) -> str:
parsed = urlparse(endpoint)
if parsed.scheme != "http" or parsed.hostname not in {
"127.0.0.1",
"localhost",
"::1",
}:
raise OllamaEmbeddingError("Day 18 只允許本機 HTTP Ollama endpoint")
return endpoint.rstrip("/")
def _request_json(
endpoint: str,
path: str,
*,
payload: Mapping[str, Any] | None = None,
timeout_seconds: int = 30,
) -> JsonObject:
base = _local_endpoint(endpoint)
data = None if payload is None else canonical_json_bytes(payload)
request = Request(
base + path,
data=data,
method="GET" if data is None else "POST",
headers={"Content-Type": "application/json"},
)
try:
with urlopen(request, timeout=timeout_seconds) as response:
body = response.read().decode("utf-8")
except HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise OllamaEmbeddingError(
f"Ollama 回傳 HTTP {exc.code}:{detail[:300]}"
) from exc
except URLError as exc:
raise OllamaEmbeddingError(
"無法連線本機 Ollama;請先啟動 Ollama 並確認 11434 連接埠。"
) from exc
try:
value = json.loads(body)
except json.JSONDecodeError as exc:
raise OllamaEmbeddingError("Ollama 回應不是合法 JSON") from exc
if not isinstance(value, dict):
raise OllamaEmbeddingError("Ollama 回應必須是 JSON 物件")
if value.get("error"):
raise OllamaEmbeddingError(f"Ollama 錯誤:{value['error']}")
return value
def get_runtime_metadata(
endpoint: str, model: str, *, timeout_seconds: int = 30
) -> JsonObject:
version = _request_json(
endpoint, "/api/version", timeout_seconds=timeout_seconds
).get("version")
if not isinstance(version, str) or not version.strip():
raise OllamaEmbeddingError("/api/version 缺少有效的 version")
tags = _request_json(endpoint, "/api/tags", timeout_seconds=timeout_seconds)
models = tags.get("models")
if not isinstance(models, list):
raise OllamaEmbeddingError("/api/tags 缺少 models 陣列")
matches = [
item
for item in models
if item.get("name") == model or item.get("model") == model
]
if len(matches) != 1:
raise OllamaEmbeddingError(
f"本機必須恰好存在一個 {model},目前找到 {len(matches)} 個"
)
item = matches[0]
details = item.get("details") or {}
return {
"ollama_version": version,
"model": item.get("name") or item.get("model"),
"digest": item.get("digest"),
"size_bytes": item.get("size"),
"parameter_size": details.get("parameter_size"),
"quantization_level": details.get("quantization_level"),
"format": details.get("format"),
"family": details.get("family"),
}
def validate_embedding_response(
response: Mapping[str, Any],
*,
expected_model: str,
expected_count: int,
expected_dimensions: int,
normalization_tolerance: float,
) -> Tuple[List[List[float]], JsonObject]:
if response.get("model") != expected_model:
raise OllamaEmbeddingError(
f"Ollama 回應模型應為 {expected_model},實際為 {response.get('model')}"
)
embeddings = response.get("embeddings")
if not isinstance(embeddings, list) or len(embeddings) != expected_count:
raise OllamaEmbeddingError(
f"向量數量應為 {expected_count},實際為 "
f"{len(embeddings) if isinstance(embeddings, list) else '非陣列'}"
)
normalized: List[List[float]] = []
norms = []
for index, vector in enumerate(embeddings):
if not isinstance(vector, list) or len(vector) != expected_dimensions:
raise OllamaEmbeddingError(
f"第 {index + 1} 筆向量維度應為 {expected_dimensions}"
)
numbers = []
for value in vector:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise OllamaEmbeddingError("向量只能包含數字")
number = float(value)
if not math.isfinite(number):
raise OllamaEmbeddingError("向量不可包含 NaN 或無限值")
numbers.append(number)
norm = math.sqrt(sum(value * value for value in numbers))
if norm == 0:
raise OllamaEmbeddingError("向量長度不可為 0")
if abs(norm - 1.0) > normalization_tolerance:
raise OllamaEmbeddingError(
f"第 {index + 1} 筆向量 L2 norm={norm},超出正規化容許值"
)
normalized.append(numbers)
norms.append(norm)
metrics = {
"model": response.get("model"),
"total_duration_ns": response.get("total_duration"),
"load_duration_ns": response.get("load_duration"),
"prompt_eval_count": response.get("prompt_eval_count"),
"vector_count": len(normalized),
"vector_dimensions": expected_dimensions,
"l2_norm_min": min(norms),
"l2_norm_max": max(norms),
}
return normalized, metrics
def embed_texts(
texts: Sequence[str],
config: Mapping[str, Any],
) -> Tuple[List[List[float]], JsonObject]:
if not isinstance(texts, list) or not texts:
raise OllamaEmbeddingError("embedding 輸入必須是非空文字陣列")
if any(not isinstance(text, str) or not text.strip() for text in texts):
raise OllamaEmbeddingError("每筆 embedding 輸入都必須是非空字串")
payload = {
"model": config["model"],
"input": list(texts),
"truncate": config["truncate"],
"dimensions": config["dimensions"],
"keep_alive": config["keep_alive"],
}
response = _request_json(
config["endpoint"],
"/api/embed",
payload=payload,
timeout_seconds=int(config["timeout_seconds"]),
)
return validate_embedding_response(
response,
expected_model=str(config["model"]),
expected_count=len(texts),
expected_dimensions=int(config["dimensions"]),
normalization_tolerance=float(config["normalization_check_tolerance"]),
)
def cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float:
if len(left) != len(right) or not left:
raise OllamaEmbeddingError("cosine similarity 需要相同且非零的向量維度")
left_norm = math.sqrt(sum(float(value) ** 2 for value in left))
right_norm = math.sqrt(sum(float(value) ** 2 for value in right))
if left_norm == 0 or right_norm == 0:
raise OllamaEmbeddingError("cosine similarity 不接受零向量")
dot = sum(float(a) * float(b) for a, b in zip(left, right))
return dot / (left_norm * right_norm)
def vector_sha256(vector: Sequence[float]) -> str:
return sha256_bytes(canonical_json_bytes(list(vector)))
def rank_documents(
query_vector: Sequence[float],
documents: Sequence[Mapping[str, Any]],
document_vectors: Sequence[Sequence[float]],
) -> List[JsonObject]:
if len(documents) != len(document_vectors):
raise OllamaEmbeddingError("文件與文件向量數量不同")
ranked = []
for document, vector in zip(documents, document_vectors):
ranked.append(
{
"rule_id": document["metadata"]["rule_id"],
"chunk_id": document["chunk_id"],
"topic": document["metadata"]["topic"],
"cosine_similarity": cosine_similarity(query_vector, vector),
}
)
ranked.sort(key=lambda item: (-item["cosine_similarity"], item["rule_id"]))
for rank, item in enumerate(ranked, start=1):
item["rank"] = rank
return ranked
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
scripts/run_day18_numeric_embeddings.py驗證上游雜湊與本機模型 digest,向量化九筆公開文件及同一合成案例的三種查詢,寫出完整向量、公開摘要與執行紀錄。
請在文字編輯器建立 scripts/run_day18_numeric_embeddings.py,貼入以下完整內容並儲存:
#!/usr/bin/env python3
"""Run Day 18 synthetic numeric representations and local embedding smoke test."""
from __future__ import annotations
import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Mapping
from triage_rag.knowledge_base.builder import read_jsonl, write_jsonl
from triage_rag.representation.embedding import (
cosine_similarity,
embed_texts,
get_runtime_metadata,
rank_documents,
vector_sha256,
)
from triage_rag.representation.numeric_semantics import (
build_all_case_representations,
representation_texts,
)
from triage_rag.reproducibility import (
file_record,
git_state,
load_json,
resolve_project_path,
sha256_bytes,
sha256_file,
write_json,
)
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/representation/day-18-numeric-semantics.json"
REPRESENTATION_NAMES = ("raw", "semantic", "dual")
JsonObject = Dict[str, Any]
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="建立 Day 18 數值三表示,並以本機 Ollama 執行向量冒煙測試。"
)
parser.add_argument("--config", default=DEFAULT_CONFIG)
return parser.parse_args()
def _verify_sha256(path: Path, expected: str, label: str) -> None:
actual = sha256_file(path)
if actual != expected:
raise ValueError(
f"{label} SHA-256 不同;預期 {expected},實際 {actual}。"
"請先重建上游產物,或建立新版 Day 18 契約。"
)
def _verify_sources(config: Mapping[str, Any]) -> List[Path]:
sources = config["sources"]
pairs = [
("field_policy_path", "field_policy_sha256", "Day 11 欄位政策"),
(
"missingness_policy_path",
"missingness_policy_sha256",
"Day 15 缺失政策",
),
("model_lock_path", "model_lock_sha256", "本機模型鎖定"),
("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
]
paths = []
for path_key, sha_key, label in pairs:
path = resolve_project_path(PROJECT_ROOT, sources[path_key])
_verify_sha256(path, sources[sha_key], label)
paths.append(path)
return paths
def _verify_model_lock(config: Mapping[str, Any]) -> None:
lock = load_json(
resolve_project_path(PROJECT_ROOT, config["sources"]["model_lock_path"])
)
dense = lock["models"]["dense_embedding"]
embedding = config["embedding"]
if dense["name"] != embedding["model"]:
raise ValueError("Day 18 model 與本機模型鎖定名稱不同")
if dense["model_id"] != embedding["expected_model_id_prefix"]:
raise ValueError("Day 18 model ID 與本機模型鎖定不同")
def _verify_runtime_model(
runtime: Mapping[str, Any], embedding: Mapping[str, Any]
) -> None:
checks = {
"model": (runtime.get("model"), embedding["model"]),
"digest": (runtime.get("digest"), embedding["expected_model_digest"]),
"parameter_size": (
runtime.get("parameter_size"),
embedding["expected_parameter_size"],
),
"quantization_level": (
runtime.get("quantization_level"),
embedding["expected_quantization_level"],
),
}
mismatches = {
key: {"actual": actual, "expected": expected}
for key, (actual, expected) in checks.items()
if actual != expected
}
if mismatches:
raise ValueError(f"本機 embedding 模型與 Day 18 鎖定不同:{mismatches}")
def _select_case(
records: List[JsonObject], case_id: str
) -> JsonObject:
matches = [record for record in records if record["case_id"] == case_id]
if len(matches) != 1:
raise ValueError(f"embedding_case_id 必須恰好命中一筆,目前 {len(matches)} 筆")
return matches[0]
def _rounded_ranked(items: List[JsonObject], top_k: int) -> List[JsonObject]:
return [
{
**item,
"cosine_similarity": round(item["cosine_similarity"], 8),
}
for item in items[:top_k]
]
def main() -> int:
args = parse_args()
started_at = datetime.now(timezone.utc)
config_path = resolve_project_path(PROJECT_ROOT, args.config)
config = load_json(config_path)
if config.get("schema_version") != 1:
raise ValueError("目前只支援 schema_version=1")
if config.get("scope") != (
"synthetic_numeric_representation_and_embedding_smoke_not_clinical_evaluation"
):
raise ValueError("Day 18 scope 不正確")
source_paths = _verify_sources(config)
_verify_model_lock(config)
cases_path = resolve_project_path(PROJECT_ROOT, config["synthetic_cases_path"])
_verify_sha256(
cases_path,
config["synthetic_cases_sha256"],
"Day 18 合成案例 fixture",
)
cases = load_json(cases_path)
representation_records = build_all_case_representations(cases, config)
selected = _select_case(representation_records, config["embedding_case_id"])
selected_texts = representation_texts(selected)
flat_path = resolve_project_path(
PROJECT_ROOT, config["sources"]["flat_chunks_path"]
)
documents = read_jsonl(flat_path)
if not documents or any(
item.get("retrieval_role") != "search_candidate" for item in documents
):
raise ValueError("Day 17 flat chunks 必須是非空搜尋候選")
embedding = config["embedding"]
runtime = get_runtime_metadata(
embedding["endpoint"],
embedding["model"],
timeout_seconds=int(embedding["timeout_seconds"]),
)
_verify_runtime_model(runtime, embedding)
document_texts = [item["search_text"] for item in documents]
document_vectors, document_metrics = embed_texts(document_texts, embedding)
query_inputs = [
f"Instruct: {embedding['query_instruction']}\nQuery: {selected_texts[name]}"
for name in REPRESENTATION_NAMES
]
query_vectors, query_metrics = embed_texts(query_inputs, embedding)
retrieval = {}
expected_rule = config["retrieval_smoke"]["expected_rule_id"]
top_k = int(config["retrieval_smoke"]["top_k"])
for name, query_vector in zip(REPRESENTATION_NAMES, query_vectors):
ranked = rank_documents(query_vector, documents, document_vectors)
expected_item = next(
item for item in ranked if item["rule_id"] == expected_rule
)
retrieval[name] = {
"expected_rule_id": expected_rule,
"expected_rule_rank": expected_item["rank"],
"expected_rule_cosine_similarity": round(
expected_item["cosine_similarity"], 8
),
"expected_rule_within_top_k": expected_item["rank"] <= top_k,
"top_k": _rounded_ranked(ranked, top_k),
}
pairwise = {}
for left_index, left_name in enumerate(REPRESENTATION_NAMES):
for right_index in range(left_index + 1, len(REPRESENTATION_NAMES)):
right_name = REPRESENTATION_NAMES[right_index]
pairwise[f"{left_name}_vs_{right_name}"] = round(
cosine_similarity(query_vectors[left_index], query_vectors[right_index]),
8,
)
run_id = (
f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-"
f"{runtime['digest'][:8]}"
)
run_root = resolve_project_path(
PROJECT_ROOT, config["outputs"]["run_output_root"]
)
run_directory = run_root / run_id
run_directory.mkdir(parents=True, exist_ok=False)
document_embedding_records = []
for document, vector in zip(documents, document_vectors):
document_embedding_records.append(
{
"schema_version": 1,
"chunk_id": document["chunk_id"],
"rule_id": document["metadata"]["rule_id"],
"search_text_sha256": document["search_text_sha256"],
"model": embedding["model"],
"model_digest": runtime["digest"],
"dimensions": embedding["dimensions"],
"vector_sha256": vector_sha256(vector),
"vector": vector,
}
)
query_embedding_records = []
for name, query_input, vector in zip(
REPRESENTATION_NAMES, query_inputs, query_vectors
):
query_embedding_records.append(
{
"schema_version": 1,
"case_id": selected["case_id"],
"representation": name,
"synthetic_only": True,
"query_input": query_input,
"query_input_sha256": sha256_bytes(query_input.encode("utf-8")),
"model": embedding["model"],
"model_digest": runtime["digest"],
"dimensions": embedding["dimensions"],
"vector_sha256": vector_sha256(vector),
"vector": vector,
}
)
document_vectors_path = run_directory / "document-embeddings.jsonl"
query_vectors_path = run_directory / "query-embeddings.jsonl"
write_jsonl(document_vectors_path, document_embedding_records)
write_jsonl(query_vectors_path, query_embedding_records)
no_interpreted_thresholds = all(
item["status"]
in {
config["threshold_policy"]["observed_status"],
config["threshold_policy"]["missing_status"],
}
for record in representation_records
for item in record["threshold_status"].values()
)
summary = {
"schema_version": 1,
"experiment_id": config["experiment_id"],
"scope": config["scope"],
"runtime": runtime,
"embedding_contract": {
"model": embedding["model"],
"dimensions": embedding["dimensions"],
"truncate": embedding["truncate"],
"query_instruction": embedding["query_instruction"],
"documents_have_instruction": False,
"query_vector_count": len(query_vectors),
"document_vector_count": len(document_vectors),
"query_vector_sha256": {
name: vector_sha256(vector)
for name, vector in zip(REPRESENTATION_NAMES, query_vectors)
},
},
"synthetic_cases": {
"case_count": len(representation_records),
"embedding_case_id": selected["case_id"],
"uses_patient_records": False,
"uses_reference_labels": False,
},
"threshold_policy": {
"profile_id": config["threshold_policy"]["profile_id"],
"status": config["threshold_policy"]["status"],
"rule_count": len(config["threshold_policy"]["rules"]),
"fail_closed": config["threshold_policy"]["fail_closed"],
"interpretation": config["threshold_policy"]["explanation"],
},
"representation_example": selected,
"query_pairwise_cosine": pairwise,
"retrieval_smoke": {
"status": config["retrieval_smoke"]["status"],
"results": retrieval,
},
"checks": {
"source_hashes_match": True,
"model_lock_matches_runtime": True,
"vector_dimensions_match": all(
len(vector) == embedding["dimensions"]
for vector in [*document_vectors, *query_vectors]
),
"vectors_l2_normalized": (
abs(document_metrics["l2_norm_min"] - 1.0)
<= embedding["normalization_check_tolerance"]
and abs(document_metrics["l2_norm_max"] - 1.0)
<= embedding["normalization_check_tolerance"]
and abs(query_metrics["l2_norm_min"] - 1.0)
<= embedding["normalization_check_tolerance"]
and abs(query_metrics["l2_norm_max"] - 1.0)
<= embedding["normalization_check_tolerance"]
),
"truncate_disabled": embedding["truncate"] is False,
"no_unsourced_threshold_interpretation": no_interpreted_thresholds,
"patient_and_label_data_absent": True,
},
"warning": (
"這是四筆純合成案例中的單一案例向量冒煙測試;排名只用來確認管線可執行,"
"不是數值表示優劣、正式檢索品質、五級分類或臨床安全結論。"
),
}
public_summary_path = resolve_project_path(
PROJECT_ROOT, config["outputs"]["public_summary_path"]
)
write_json(public_summary_path, summary)
manifest = {
"manifest_schema_version": 1,
"run_id": run_id,
"started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
"command": [sys.executable, *sys.argv],
"documented_command": [
"poetry",
"run",
"python",
"scripts/run_day18_numeric_embeddings.py",
],
"git": git_state(PROJECT_ROOT),
"runtime": runtime,
"embedding_usage": {
"documents": document_metrics,
"queries": query_metrics,
},
"inputs": [
file_record(PROJECT_ROOT, args.config),
file_record(PROJECT_ROOT, config["synthetic_cases_path"]),
*[
file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
for path in source_paths
],
],
"outputs": [
file_record(
PROJECT_ROOT, str(document_vectors_path.relative_to(PROJECT_ROOT))
),
file_record(PROJECT_ROOT, str(query_vectors_path.relative_to(PROJECT_ROOT))),
file_record(PROJECT_ROOT, config["outputs"]["public_summary_path"]),
],
"privacy": (
"只使用四筆 synthetic_only=true 合成案例與公開 KTAS 摘要;"
"不讀取 Kaggle 病患列、KTAS_RN 或 KTAS_expert。"
),
"scope": config["scope"],
}
manifest_path = run_directory / "run-manifest.json"
write_json(manifest_path, manifest)
ranks = ", ".join(
f"{name}={retrieval[name]['expected_rule_rank']}"
for name in REPRESENTATION_NAMES
)
print("Day 18 數值表示與本機 embedding:通過")
print(f"合成案例:{len(representation_records)} 筆;實際向量化 1 筆三種表示")
print(
f"模型:{runtime['model']};digest {runtime['digest'][:12]};"
f"{embedding['dimensions']} 維;{runtime['quantization_level']}"
)
print(f"文件向量:{len(document_vectors)} 筆;查詢向量:{len(query_vectors)} 筆")
print(f"預期規則排名:{ranks}")
print(f"公開摘要:{public_summary_path.relative_to(PROJECT_ROOT)}")
print(f"完整向量與執行紀錄:{run_directory.relative_to(PROJECT_ROOT)}")
print("注意:單一合成案例不能用來選出最佳數值表示。")
return 0
if __name__ == "__main__":
raise SystemExit(main())
儲存後先確認檔名與相對路徑完全一致,再繼續建立下一個檔案。
tests/test_numeric_representation.py以十四項單元測試驗證數值保存、疼痛與缺失語意、合成邊界、禁止標籤、向量維度及正規化契約。
請在文字編輯器建立 tests/test_numeric_representation.py,貼入以下完整內容並儲存:
from __future__ import annotations
import copy
import unittest
from pathlib import Path
from triage_rag.representation.embedding import (
OllamaEmbeddingError,
cosine_similarity,
validate_embedding_response,
)
from triage_rag.representation.numeric_semantics import (
RepresentationContractError,
build_all_case_representations,
build_case_representations,
interpret_numeric_value,
)
from triage_rag.reproducibility import load_json
ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = (
ROOT / "configs" / "representation" / "day-18-numeric-semantics.json"
)
class Day18NumericRepresentationTests(unittest.TestCase):
def setUp(self) -> None:
self.config = load_json(CONFIG_PATH)
self.cases = load_json(ROOT / self.config["synthetic_cases_path"])
def build(self, case: dict | None = None):
return build_case_representations(case or self.cases[0], self.config)
def test_all_synthetic_cases_build_with_three_representations(self) -> None:
records = build_all_case_representations(self.cases, self.config)
self.assertEqual(len(records), 4)
for record in records:
self.assertTrue(record["synthetic_only"])
self.assertEqual(
set(record["representations"]), {"raw", "semantic", "dual"}
)
def test_raw_semantic_and_dual_preserve_observed_values(self) -> None:
record = self.build()
raw = record["representations"]["raw"]["text"]
semantic = record["representations"]["semantic"]["text"]
dual = record["representations"]["dual"]["text"]
self.assertIn("RR=28", raw)
self.assertIn("Saturation=91", raw)
self.assertIn("呼吸率 28 次/分鐘", semantic)
self.assertIn("血氧飽和度 91 %", semantic)
self.assertIn(raw, dual)
self.assertIn(semantic, dual)
def test_no_pain_makes_missing_nrs_not_applicable(self) -> None:
semantic = self.build()["representations"]["semantic"]["text"]
self.assertIn("記錄為無疼痛;疼痛分數不適用", semantic)
self.assertNotIn("疼痛分數未記錄", semantic)
def test_pain_with_missing_nrs_stays_unknown(self) -> None:
record = self.build(self.cases[1])
semantic = record["representations"]["semantic"]["text"]
self.assertIn("記錄為有疼痛;疼痛分數未記錄", semantic)
def test_missing_saturation_is_not_called_normal(self) -> None:
record = self.build(self.cases[2])
semantic = record["representations"]["semantic"]["text"]
self.assertIn("血氧飽和度未記錄,不能視為正常", semantic)
self.assertEqual(
record["threshold_status"]["Saturation"]["status"],
"missing_not_recorded",
)
def test_mental_code_is_translated_from_field_policy(self) -> None:
record = self.build(self.cases[1])
semantic = record["representations"]["semantic"]["text"]
self.assertIn("意識反應:對聲音有反應", semantic)
def test_unsourced_thresholds_fail_closed(self) -> None:
result = interpret_numeric_value("RR", 28, self.config["threshold_policy"])
self.assertEqual(result["status"], "not_interpreted_no_source_rule")
self.assertIsNone(result["label"])
def te