昨天把「決定去哪」跟「負責送過去」拆開了。今天處理前者。
我原本以為這篇會是最技術的一篇,寫下去才發現,它最難的部分根本不是程式。程式碼一百多行就寫完了,真正花時間的是每一個 if 背後那句話:這種情況,我願意承擔哪一種失敗?
這兩個詞是從安全領域借來的。
fail-open:系統某個環節壞了,讓東西照樣通過。門禁系統斷電時門是開的。
fail-closed:壞了就全部擋下。斷電時門是鎖的。
哪個比較好?看門後面是什麼。火災逃生門一定要 fail-open,機房門一定要 fail-closed。同一棟大樓裡,兩種門都會有。
放到 Router 上,「壞了」有好幾種:
每一種都要個別決定停還是走,而且答案會隨著資料的敏感等級而變。這就是為什麼 fail-open 跟 fail-closed 不能當成一個全域開關。在設定檔裡寫一行 fail_mode: closed 看起來很負責,其實只是把問題藏起來。
我最後的決定是這樣:
| 壞掉的環節 | 公開資料(L0) | 內部資料(L1) | 機密(L2) | 個資(L3) |
|---|---|---|---|---|
| 分級器出錯 | 一律當 L3 處理 | 同左 | 同左 | 同左 |
| 主模型掛了 | 換下一條已核准的路 | 換本地的路 | 停,進人工佇列 | 停,進人工佇列 |
| Verifier 掛了 | 放行,標記「未驗證」 | 放行,標記「未驗證」 | 停,進人工佇列 | 停,進人工佇列 |
| 紀錄寫不進去 | 停 | 停 | 停 | 停 |
最後一列是整張表裡唯一不分等級的。沒有紀錄的路由決策,等於 Day 27 的稽核表又多了一格填不出來的欄位。寧可整條線停下來,也不要留下一段沒有紀錄的空窗。這一條我沒有猶豫過。
寫到 Verifier 那一列的時候,我停了一下。
Day 11 講過,gpt-oss:20b 這個 Verifier 到今天還沒有真正跑起來。本機顯卡 4GB 不夠,GB10 上正跑著 OCR 主模型,而且我沒有權限在上面另開服務。介面定好了,模型沒接上。
也就是說,照上面這張表,「Verifier 掛了」不是偶發事件,是現在的常態。
結果就是:此時此刻,任何 L2、L3 等級的區塊,全部都會進人工佇列。
我一開始覺得這很荒謬,寫了一大套自動化,結果機密資料一筆都自動不了。後來想想,這其實就是 fail-closed 的真實代價,而且它把代價攤在我眼前,而不是藏起來。如果我因為 Verifier 還沒好就把 L2 改成 fail-open,那這張表就只是一張等 Verifier 上線才生效的裝飾品,而在那之前的每一筆機密區塊,都是在沒有第二雙眼睛的情況下被放行的。
所以我選擇讓它卡著。卡著會痛,痛才會有人去把 Verifier 接起來。
這也連回 Day 27:人工佇列的容量是有限的。L2 以上全部進人工,佇列一定會長。稽核員的時間要先分給全檢層跟這個佇列,剩下的才輪到隨機抽樣。這是一筆實際的人力帳,不是設計圖上畫個箭頭就解決的事。
分四級,定義用「外洩時會發生什麼事」來講,比用形容詞清楚:
| 等級 | 例子 | 外洩的後果 |
|---|---|---|
| L0 公開 | MOPS 上已公告的年報、重大訊息 | 沒什麼後果,本來就公開 |
| L1 內部 | 內部整理的工作底稿、比對報表 | 尷尬,但不違法 |
| L2 機密 | 尚未公告的契約草稿、董事會資料 | 可能涉及內線交易或違約 |
| L3 個資 | 含身分證字號、銀行帳號的文件 | 違反個資法 |
有件事要先講:這系列從 Day 14 開始用的台積電年報,是公開文件,照這張表屬於 L0。那為什麼要搞這麼複雜?
因為系統不會永遠只處理公開年報。同一條 Pipeline 明天可能被拿去處理一份還沒公告的契約草稿,而那份草稿的 PDF 看起來跟年報沒什麼兩樣。分級不能等出事了再加。
判斷依據我用兩層。
第一層是來源標籤。文件進系統時由上傳的人或上游流程標注:mops_public、internal、contract_draft。沒有標籤、或標籤不認得的,直接當 L3。這是整個分級邏輯裡最重要的一行:不認得就當最嚴重的。
第二層是內容偵測,只能往上調,不能往下調。一份標成 internal 的文件,如果內容裡掃到身分證字號,就升到 L3。反過來,一份標成 contract_draft 的文件,不管內容看起來多無害,程式都不會自己把它降成 L0。降級只能由人簽核。
身分證字號偵測有個小坑。台灣身分證的格式是一個英文字母加九個數字,第一個數字是 1 或 2。單用正則去抓,會誤判一堆東西,像是某些料號、單據編號剛好長得一樣。好在身分證有檢查碼,把字母換成對應的兩位數,配上權重加總,能被 10 整除才是合法號碼。加上檢查碼驗證之後,誤判會少很多(我沒有量過少多少,這是依照檢查碼的原理推的)。
誤判往哪邊偏比較可以接受?往嚴格那邊。把一個料號當成身分證,最壞的結果是那個區塊多進一次人工佇列;把一個真的身分證字號漏掉,就是個資外洩。
每一次路由決策寫一筆紀錄。我要的欄位:
decision_id:這筆決策的唯一編號region_id:對應 Day 26 記錄層的區塊rule_version:這次用的是哪一版路由規則level 與 evidence:判了哪一級、依據是什麼tried:依序試了哪些路、每條路的結果(被等級擋掉、端點不健康、選中)chosen、outcome:最後去了哪、結果是完成、未驗證完成、進人工,還是拒絕input_digest:輸入內容的摘要值,用來事後比對「是不是同一段內容」evidence 那一欄我踩了一個設計上的坑。直覺會寫成 "命中身分證字號:A123456789",方便除錯。但這樣紀錄檔本身就變成一份個資清單了。為了保護個資而建的紀錄,自己成了最大的外洩源。所以 evidence 只記「哪條規則命中、命中幾次」,絕不記命中的值。
input_digest 是另一個坑,比上一個更隱蔽。
我原本寫的是 sha256(內容)。心想雜湊是單向的,存雜湊不存原文,很安全。
後來想到一個問題:如果內容很短、可能的值又很有限呢?一段只有「負責人 A123456789」的文字,身分證字號的組合數量是有限的,攻擊者可以把所有可能的號碼都算一遍 sha256,拿去跟紀錄比對,就能反推回原文。對短文字來說,單純雜湊不等於匿名。
改成 HMAC:雜湊時加一把只有系統知道的金鑰。金鑰不跟紀錄放在一起,拿到紀錄檔的人沒有金鑰,就沒辦法做這種窮舉比對。代價是金鑰要管理,而且金鑰一換,舊紀錄的摘要就對不上新的了,換金鑰時要記下版本。
把上面講的全部寫成一支程式。每一級各有一條「降級梯子」,由上往下試,試到能用的就停:
| 等級 | 梯子 |
|---|---|
| L0 | 本地主模型 → 已核准的外部端點(目前沒有) → 本地 Tesseract → 人工 |
| L1 | 本地主模型 → 本地 Tesseract → 人工 |
| L2 | 本地主模型 → 人工 |
| L3 | 本地主模型 → 人工 |
L0 那格的「已核准外部端點」是我刻意留的範例。這系列的原則是全程不出機房,所以它現在不存在,健康檢查永遠回傳 false。留著它是為了示範一件事:就算哪天真的核准了一個外部端點,它在路由表裡的 max_level 也只能是 L0,梯子寫錯了,程式還會再擋一次。
L2、L3 沒有 Tesseract 那一階。原因是 Tesseract 的輸出品質撐不起機密文件的判讀,而且在 Verifier 還沒上線的情況下,它的結果沒人能交叉確認。與其拿一個品質不夠的結果放行,不如直接交給人。
from __future__ import annotations
import hashlib
import hmac
import json
import os
import re
import time
import uuid
from dataclasses import dataclass, field, asdict
from enum import IntEnum
from typing import Callable
RULE_VERSION = "router-policy/2026-09-25.1"
TRACE_KEY = os.environ.get("ROUTER_TRACE_KEY", "dev-only-key").encode("utf-8")
TRACE_KEY_VERSION = os.environ.get("ROUTER_TRACE_KEY_VERSION", "dev")
class Sensitivity(IntEnum):
L0_PUBLIC = 0
L1_INTERNAL = 1
L2_CONFIDENTIAL = 2
L3_RESTRICTED = 3
# ---------- 敏感分級 ----------
# 身分證字母對應的兩位數(注意 I、O、W 等字母不照字母順序)
_ID_LETTER = dict(zip("ABCDEFGHJKLMNPQRSTUVXYWZIO", range(10, 36)))
ID_RE = re.compile(r"(?<![A-Z0-9])([A-Z])([12]\d{8})(?!\d)")
ACCOUNT_RE = re.compile(r"帳號[::\s]*\d[\d-]{9,}")
SOURCE_LEVEL = {
"mops_public": Sensitivity.L0_PUBLIC,
"internal": Sensitivity.L1_INTERNAL,
"contract_draft": Sensitivity.L2_CONFIDENTIAL,
}
def valid_tw_id(letter: str, digits: str) -> bool:
n = _ID_LETTER[letter]
total = n // 10 + (n % 10) * 9
for d, w in zip(digits[:8], (8, 7, 6, 5, 4, 3, 2, 1)):
total += int(d) * w
total += int(digits[8])
return total % 10 == 0
@dataclass
class Classification:
level: Sensitivity
evidence: list[str] # 只記規則名與次數,不記命中的值
def classify(text: str, source_tag: str | None) -> Classification:
level = SOURCE_LEVEL.get(source_tag or "", Sensitivity.L3_RESTRICTED)
evidence = [f"source_tag={source_tag or 'missing'}"]
ids = [m for m in ID_RE.finditer(text) if valid_tw_id(m.group(1), m.group(2))]
if ids:
level = Sensitivity.L3_RESTRICTED
evidence.append(f"tw_national_id x{len(ids)}")
if ACCOUNT_RE.search(text):
level = Sensitivity.L3_RESTRICTED
evidence.append("bank_account_pattern")
return Classification(level, evidence) # 只會往上調,沒有往下調的路徑
# ---------- 路徑與降級梯子 ----------
@dataclass(frozen=True)
class Route:
name: str
max_level: Sensitivity # 這條路最高能收哪一級
local: bool
ROUTES = {
"gemma4_26b_local": Route("gemma4_26b_local", Sensitivity.L3_RESTRICTED, True),
"tesseract_local": Route("tesseract_local", Sensitivity.L1_INTERNAL, True),
"human_queue": Route("human_queue", Sensitivity.L3_RESTRICTED, True),
"external_approved": Route("external_approved", Sensitivity.L0_PUBLIC, False),
}
LADDER = {
Sensitivity.L0_PUBLIC: ["gemma4_26b_local", "external_approved", "tesseract_local", "human_queue"],
Sensitivity.L1_INTERNAL: ["gemma4_26b_local", "tesseract_local", "human_queue"],
Sensitivity.L2_CONFIDENTIAL: ["gemma4_26b_local", "human_queue"],
Sensitivity.L3_RESTRICTED: ["gemma4_26b_local", "human_queue"],
}
FAIL_OPEN_WHEN_VERIFIER_DOWN = {
Sensitivity.L0_PUBLIC: True,
Sensitivity.L1_INTERNAL: True,
Sensitivity.L2_CONFIDENTIAL: False,
Sensitivity.L3_RESTRICTED: False,
}
@dataclass
class Decision:
decision_id: str
region_id: str
input_digest: str
digest_key_version: str
rule_version: str
level: str
evidence: list[str]
tried: list[dict] = field(default_factory=list)
chosen: str | None = None
verified: bool | None = None
outcome: str = "pending" # done / done_unverified / queued_human / rejected
ts: float = field(default_factory=time.time)
def digest(text: str) -> str:
return hmac.new(TRACE_KEY, text.encode("utf-8"), hashlib.sha256).hexdigest()
def route_region(
region_id: str,
text: str,
source_tag: str | None,
healthy: Callable[[str], bool],
verifier_up: bool,
) -> Decision:
try:
c = classify(text, source_tag)
except Exception as exc: # 分級器自己壞了:fail-closed
c = Classification(Sensitivity.L3_RESTRICTED, [f"classifier_error:{type(exc).__name__}"])
d = Decision(
decision_id=str(uuid.uuid4()),
region_id=region_id,
input_digest=digest(text),
digest_key_version=TRACE_KEY_VERSION,
rule_version=RULE_VERSION,
level=c.level.name,
evidence=c.evidence,
)
for name in LADDER[c.level]:
route = ROUTES[name]
if c.level > route.max_level: # 梯子寫錯也擋得住
d.tried.append({"route": name, "result": "blocked_by_level"})
continue
if not healthy(name):
d.tried.append({"route": name, "result": "unhealthy"})
continue
d.tried.append({"route": name, "result": "selected"})
d.chosen = name
break
if d.chosen is None:
d.outcome = "rejected"
elif d.chosen == "human_queue":
d.outcome = "queued_human"
elif verifier_up:
d.verified, d.outcome = True, "done"
elif FAIL_OPEN_WHEN_VERIFIER_DOWN[c.level]:
d.verified, d.outcome = False, "done_unverified"
else:
d.tried.append({"route": "verifier", "result": "down_fail_closed"})
d.chosen, d.outcome = "human_queue", "queued_human"
return d
def write_trace(d: Decision, path: str = "router_trace.jsonl") -> None:
"""寫不進去就讓例外往上拋:紀錄失敗時整條線停下來,四個等級都一樣。"""
line = json.dumps(asdict(d), ensure_ascii=False)
with open(path, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
os.fsync(f.fileno())
有幾個地方我想特別說明。
Tesseract 的 max_level 我設成 L1。這樣就算有人手滑把它加進 L2 的梯子,程式也會記一筆 blocked_by_level 然後跳過。梯子是政策,max_level 是護欄,兩個各寫一次,看起來重複,但我寧可重複。
write_trace 沒有包 try/except。這是故意的。紀錄寫失敗要讓整個流程炸掉,呼叫端不該有機會「先吞掉例外、之後再補寫」。fsync 讓每一筆都真的落到磁碟,效能會差一點,對一個每秒處理不了幾個區塊的 OCR 系統來說,這點成本不重要。
TRACE_KEY 從環境變數讀,預設值是 dev-only-key,只是方便本機跑。正式環境沒設金鑰就啟動,應該直接拒絕,這個檢查我放在啟動流程裡,沒寫在這支。
情境一:GB10 的主模型掛了,Verifier 正常。
p61-left L0_PUBLIC -> tesseract_local done ['source_tag=mops_public']
draft-07 L2_CONFIDENTIAL -> human_queue queued_human ['source_tag=contract_draft']
scan-13 L3_RESTRICTED -> human_queue queued_human ['source_tag=internal', 'tw_national_id x1']
unknown-2 L3_RESTRICTED -> human_queue queued_human ['source_tag=missing']
p61-left 是 Day 15 那段竹環字號的裁罰文字,公開資料,主模型掛了就退到 Tesseract。scan-13 標的是 internal,但內容裡有一組通過檢查碼的身分證字號(範例用的是常被拿來當教材的 A123456789),被升到 L3。unknown-2 沒有來源標籤,直接當 L3。
情境二:主模型正常,Verifier 掛了,也就是我現在的真實狀態。
p61-left L0_PUBLIC gemma4_26b_local done_unverified
memo-3 L1_INTERNAL gemma4_26b_local done_unverified
draft-07 L2_CONFIDENTIAL human_queue queued_human
draft-07 的 tried 欄位會留下兩筆:先是選中了主模型,然後是 verifier: down_fail_closed。事後看紀錄,可以清楚知道它不是因為主模型壞掉才進人工,而是因為沒人驗證。這個差別在稽核時很重要,前者要修機器,後者要去催 Verifier 上線。
這兩段輸出是我在本機用範例資料跑出來的,只證明邏輯照設計走,不代表任何真實文件的處理結果。
分級的第一層依賴來源標籤,而標籤是人標的。有人把契約草稿標成 mops_public,內容偵測又沒掃到個資,這份草稿就會被當 L0 處理。程式擋不住這種錯。我能做的是把「誰標的、什麼時候標的」一起記進紀錄,出事時至少追得到。
內容偵測目前只有身分證字號跟銀行帳號兩條。統一編號、電話、地址都還沒做,公司名稱更難,因為年報裡本來就滿滿都是公司名稱,你沒辦法把它們都當敏感。
還有 Verifier 的問題。只要它一天沒接上,L2 以上就一天全靠人工。這件事沒有程式上的捷徑。
三十天裡,我寫的最後一段程式碼,是一段讓系統在不確定時停下來的程式碼。