iT邦幫忙

2026 iThome 鐵人賽

DAY 27
0
AI Engineering

從 LLM 到 Harness: 打造隱私與可信任的繁中進階 OCR Agent系列 第 27 篇

Day 27 - 2-5% 抽樣人工稽核:統計設計與流程

  • 分享至 

  • xImage
  •  

昨天最後停在一張空表上:audits。

Day 26 的四張表裡,documents、messages、diff_events 都有程式自動寫,只有 audits 要靠人。欄位不多,auditor 是誰看的,decision 是四選一:model_wrong、model_right、gt_defect、unclear。Day 25 又規定只有人工判定過的事件能拿來調門檻。

那誰來填?什麼時候填?填幾筆?

全部都填不可能,那前面二十幾天的 Harness 就只是一台很貴的預處理機。一筆都不填也不行,整套驗證(Checksum、Logprob、Tesseract、XBRL)全是機器在檢查機器,總要有人偶爾進來確認它們沒有一起歪掉。

規劃書上我寫的是「2-5% 抽樣」。

老實說,這個數字是我當初憑感覺寫的。今天要做的事就是回頭把它算一遍,看它到底站不站得住。


先決定「一筆」是什麼

這比公式還重要。

抽樣單位可以是文件、頁、或 Day 18 切出來的區塊。直覺會選「頁」,因為人看頁最自然。但 Day 2 跟 Day 10 已經示範過這份文件有多不均質:純文字頁覆蓋率 99.4%,混排頁 76.2%,同一份 PDF 裡的頁面根本不是同一種東西。以頁為單位抽,混排的錯跟純文字的零錯被平均在一起,得到一個什麼都沒說明的數字。

以文件為單位更糟。一份 92 頁的台積電年報算「一筆」,稽核員根本不可能看完。

所以我選區塊。理由很實際:區塊是 Router 做決策的單位,也是 Verifier 下判定的單位。Day 23 的協定規定一個區塊從頭到尾共用一個 trace_id,所以在 Day 26 的表裡,「一個區塊」就是 messages 裡同一個 trace_id 的那一串訊息:任務、結果、判定,重試幾輪都在裡面。稽核的單位跟決策的單位對齊,稽核結果才能回頭說「是哪一條路、哪一個判定出了問題」。

代價是要給上下文。稽核員只看到一格孤零零的「45萬元」,根本不知道對不對。


樣本數怎麼算

我要回答的問題是:「這一批區塊裡,實際有錯的比例大概是多少?」這是估計比例,教科書公式是:

n0 = z² × p × (1 − p) / e²
  • z:信心水準對應的值,95% 取 1.96
  • p:預期的錯誤率
  • e:容許誤差,也就是你能接受估出來的比例上下差多少

母體不是無限大,要做有限母體校正:

n = n0 / (1 + (n0 − 1) / N)

p 取多少?我沒有 Harness 的錯誤率實測,這裡是設計選擇:假設 5%,容許誤差 ±2 個百分點。

n0 = 1.96² × 0.05 × 0.95 / 0.02²
   = 3.8416 × 0.0475 / 0.0004
   ≈ 456.2  → 457

然後我把不同批量代進去,算出來的東西讓我愣了一下:

一批的區塊數 N 需要的樣本 n 佔比
2,000 372 18.6%
5,000 419 8.4%
10,000 437 4.4%
12,000 440 3.7%
20,000 447 2.2%
40,000 452 1.1%

(p = 0.05、e = ±0.02、95% 信心,n 一律無條件進位)

樣本數幾乎不動。從兩千到四萬,n 只從 372 爬到 452。

變的是「佔比」。

這是我當初寫「2-5%」時完全沒想到的:抽樣要的是大致固定的絕對數量,百分比只在某個批量範圍裡剛好成立。我把 N 一路掃過去,這組參數下 2-5% 對應的批量大約是 9,000 到 22,000 個區塊。

批量小的時候,2-5% 會嚴重不夠。假設一批只有 2,000 個區塊,你照規劃抽 5% 也就是 100 筆:

e = 1.96 × sqrt(0.05 × 0.95 / 100 × (2000 − 100) / (2000 − 1))
  ≈ 1.96 × 0.02125
  ≈ 0.042

估出來 5%,實際可能在 0.8% 到 9.2% 之間。寬到沒辦法跟上個月比。

批量大時反過來,四萬個區塊抽 5% 是兩千筆,452 筆就夠了,多的人力拿去做風險抽樣划算得多。

Tip:所以我把規劃書那句話改寫成:「每批至少 440 筆隨機樣本;在一般批量下大約落在 2-5%」。先講絕對數,百分比只是換算。


抽到零個錯,不代表錯誤率是零

另一個很常見的誤讀。

假設某一批運氣很好,抽了 300 筆,人工看完一筆錯都沒有。報告上寫「錯誤率 0%」嗎?

不行。統計上有一條很好記的「三的法則」:抽 n 筆、零個錯,真實錯誤率的 95% 上界大約是 3/n。

抽樣筆數 零錯誤時的 95% 上界(3/n) 精確值
150 2.0% 1.98%
300 1.0% 0.99%
500 0.6% 0.60%

精確值是解 (1 − p)^n = 0.05 得來的,跟 3/n 幾乎一樣。

300 筆零錯誤,你只能說「錯誤率很可能低於 1%」。這比「0%」誠實,對要不要放寬驗證門檻,資訊量也更大。


錯很少的時候,別用課本那條區間

真的抽到錯了,要算信賴區間。課本上的常態近似(Wald 區間)是 p̂ ± z × sqrt(p̂(1−p̂)/n),但錯誤率很低、樣本又不大的時候,它會算得很難看。

舉個例子:300 筆裡抽到 6 筆錯,p̂ = 2%。

Wald:   0.02 ± 1.96 × sqrt(0.02 × 0.98 / 300)
      = 0.02 ± 0.0158
      → 0.42% ~ 3.58%

換成 Wilson 區間:

center = (p̂ + z²/2n) / (1 + z²/n)
half   = z × sqrt(p̂(1−p̂)/n + z²/4n²) / (1 + z²/n)
       → 0.92% ~ 4.29%

差最多的是上界:3.58% 對 4.29%。錯誤率不可能低於零,低比例時真實分布往右歪,對稱的 Wald 就會低估上界,偏偏上界是我最在意的那一邊。

我後面的腳本一律用 Wilson。


隨機抽樣跟風險抽樣要分開帳

到這裡都是均勻地隨機抽。能推估整體錯誤率,但很浪費:九成區塊是乾淨的純文字,人力大多花在確認「對,這段沒問題」。

我手上的錯誤分類給了一個很具體的線索。experiments/step4_zh_error_summary.md 裡 9 頁一共 87 個 diff 塊,其中 37 個落在 other,佔 42.5%。也就是我的規則式分類器認不出來的差異,將近一半的錯我連種類都說不出來。

這類東西最需要人看。

還有 Day 14 那個教訓:finish_reason 是 length 的輸出,幾乎等於整段崩掉(複讀、幻覺、700 多行空表格列)。就算後來重試成功,也要確認留下來的不是一段比較短的幻覺。這種區塊不抽,全部看。

三個條件在 Day 26 的表裡都查得到,不用另外加欄位:

層 條件(Day 26 的欄位) 怎麼處理
全檢層 這條 trace 有任何 messages.finish_reason = 'length' 全部看,不抽樣
旗標層 有 result 訊息的 messages.outcome 是 FLAG 或 FAIL 按比例隨機抽,再加抽 50%
分類不明層 有 diff_events.auto_label = 'other' 按比例隨機抽,再加抽 50%
一般層 其他 按比例隨機抽

重點在「再加抽」那一部分。加抽的樣本拿來找錯、拿來餵 Day 25 的飛輪,但不能拿來算整體錯誤率。因為它們是被刻意挑出來的高風險區塊,混進去算,錯誤率會被灌高,然後你會開始懷疑一套其實運作正常的系統。

兩種樣本要用一個欄位分開:random、risk、census。只有 random 進估計。


抽樣腳本

我最早的草稿是 random.seed(seed) 配 random.sample(rows, n),寫完才想到一個洞:random.sample 對同一個 seed 可以重現,前提是輸入的順序一樣。而我的 SQL 沒寫 ORDER BY,SQLite 並不保證回傳順序,哪天補了一個索引、或資料庫做過 VACUUM,同一個 seed 可能就抽出另一組區塊。這種 bug 最討厭,平常完全不會發作,偏偏在稽核結果被質疑、需要重抽對照的那一天才出現。

改成用 hash 決定順序:每個 trace 算 sha256(seed + trace_id),依這個值排序後取前 n 個。同一個 seed、同一批區塊,不管資料庫怎麼回傳,抽出來永遠是同一組。稽核結果被質疑時,別人可以拿同一個 seed 自己重抽一次對照。

為什麼要替 Day 26 的 schema 加表

我本來想直接寫進 audits,結果塞不進去。

audits.event_id 是 NOT NULL,指向 diff_events。也就是說,audits 只能判「某一個 diff 事件」,而 diff 事件只會出現在某個對照組跟模型不一致的地方。一個區塊如果模型、Tesseract、文字層三方都一致,它一筆 diff 都沒有,audits 就沒有地方寫它。

偏偏這種區塊最需要被隨機抽到。Day 2 那 26 條錯誤全是安靜的,三方一起錯的情況一定存在。樣本如果只能落在有 diff 的地方,估出來的就只是「有 diff 的區塊裡的錯誤率」。

另一個缺口是批次:Day 26 沒有「這批抽了誰、用哪個 seed」的概念。

所以我加三張表,當成 Day 26 schema 的 v3 擴充。判定欄位直接沿用 audits 的 auditor 跟 decision,選項一模一樣,不另外發明一套詞:

-- Day 26 schema 的擴充(v3):抽樣批次、樣本、樣本層級的人工判定
CREATE TABLE IF NOT EXISTS audit_batches (
    batch_id      TEXT PRIMARY KEY,
    seed          TEXT NOT NULL,
    rule_version  TEXT NOT NULL,
    window_start  TEXT NOT NULL,
    window_end    TEXT NOT NULL,
    population    INTEGER NOT NULL,       -- 不含全檢層
    target_n      INTEGER NOT NULL,
    params_json   TEXT NOT NULL,
    created_at    TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS audit_samples (
    batch_id     TEXT NOT NULL REFERENCES audit_batches(batch_id),
    trace_id     TEXT NOT NULL,           -- 對應 messages.trace_id,一個區塊一條 trace
    stratum      TEXT NOT NULL,
    sample_kind  TEXT NOT NULL CHECK (sample_kind IN ('random', 'risk', 'census')),
    weight       REAL NOT NULL,           -- N_h / n_h;risk、census 不進估計
    PRIMARY KEY (batch_id, trace_id)
);

CREATE TABLE IF NOT EXISTS audit_reviews (
    batch_id    TEXT NOT NULL,
    trace_id    TEXT NOT NULL,
    auditor     TEXT NOT NULL,            -- 沿用 audits.auditor
    decision    TEXT NOT NULL CHECK (decision IN
                    ('model_wrong', 'model_right', 'gt_defect', 'unclear')),  -- 沿用 audits.decision
    audited_at  TEXT NOT NULL,
    PRIMARY KEY (batch_id, trace_id, auditor),
    FOREIGN KEY (batch_id, trace_id) REFERENCES audit_samples(batch_id, trace_id)
);

audit_reviews 判整個區塊,拿來算錯誤率;看的過程中對個別 diff 事件下的判斷,照舊寫進 audits 餵 calibrate_min_prob。一個判區塊,一個判事件,不重疊。

下面這支是抽樣腳本,直接查 Day 26 的 messages 和 diff_events:

# audit_sampler.py:從 Day 26 的記錄層挑出本批要給人看的區塊(trace)
from __future__ import annotations

import argparse
import hashlib
import json
import math
import sqlite3
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

RULE_VERSION = "audit-sampler/0.4"
Z95 = 1.959964
EXT_DDL = Path(__file__).with_name("audit_ext.sql")   # 上面那段擴充 SQL

# 一個 trace 一列:歷史裡有沒有被截斷過、有沒有被判 FLAG/FAIL、有沒有 other 類的 diff
POPULATION_SQL = """
SELECT m.trace_id,
       MAX(m.body_type = 'result' AND m.finish_reason = 'length')    AS truncated,
       MAX(m.body_type = 'result' AND m.outcome IN ('FLAG', 'FAIL')) AS flagged,
       COALESCE(MAX(d.auto_label = 'other'), 0)                   AS has_other
FROM messages m
LEFT JOIN diff_events d ON d.msg_id = m.msg_id
GROUP BY m.trace_id
HAVING MIN(m.created_at) >= ? AND MIN(m.created_at) < ?
"""


def required_n(N: int, p: float = 0.05, e: float = 0.02, z: float = Z95) -> int:
    """估計比例用的樣本數,含有限母體校正。"""
    if N <= 0:
        return 0
    n0 = z * z * p * (1 - p) / (e * e)
    return min(N, math.ceil(n0 / (1 + (n0 - 1) / N)))


def stable_u(seed: str, trace_id: str) -> float:
    """同一個 seed + trace_id 永遠得到同一個 [0,1) 值,不受 SQL 回傳順序影響。"""
    h = hashlib.sha256(f"{seed}:{trace_id}".encode("utf-8")).digest()
    return int.from_bytes(h[:8], "big") / 2**64


def stratum_of(row: sqlite3.Row) -> str:
    if row["truncated"]:
        return "census_truncated"          # 歷史裡出現過 length:全看,不抽
    if row["flagged"]:
        return "risk_flagged"
    if row["has_other"]:
        return "risk_other"
    return "normal"


@dataclass
class Pick:
    trace_id: str
    stratum: str
    sample_kind: str   # "random" 可推估全體;"risk" 只拿來找錯;"census" 全檢
    weight: float      # 這一筆代表母體裡幾筆(N_h / n_h)


def draw(rows, seed: str, p: float, e: float, risk_extra: float = 0.5):
    by_stratum: dict[str, list] = defaultdict(list)
    for r in rows:
        by_stratum[stratum_of(r)].append(r)

    sampled_pop = sum(len(v) for k, v in by_stratum.items() if k != "census_truncated")
    n_total = required_n(sampled_pop, p, e)
    picks: list[Pick] = []

    for stratum, members in sorted(by_stratum.items()):
        members.sort(key=lambda r: stable_u(seed, r["trace_id"]))
        N_h = len(members)
        if stratum == "census_truncated":
            picks += [Pick(r["trace_id"], stratum, "census", 1.0) for r in members]
            continue
        # 比例配置:每層至少 1 筆,才不會有空層
        n_h = min(N_h, max(1, round(n_total * N_h / sampled_pop)))
        picks += [Pick(r["trace_id"], stratum, "random", N_h / n_h) for r in members[:n_h]]
        if stratum.startswith("risk_"):
            extra = min(N_h - n_h, math.ceil(n_h * risk_extra))
            picks += [Pick(r["trace_id"], stratum, "risk", 0.0)
                      for r in members[n_h:n_h + extra]]
    return picks, n_total, sampled_pop


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--db", required=True)
    ap.add_argument("--batch", required=True)
    ap.add_argument("--since", required=True)   # trace 第一則訊息的時間,含
    ap.add_argument("--until", required=True)   # 不含
    ap.add_argument("--seed", required=True)
    ap.add_argument("--p", type=float, default=0.05)   # 預期錯誤率
    ap.add_argument("--e", type=float, default=0.02)   # 容許誤差
    args = ap.parse_args()

    con = sqlite3.connect(args.db)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys = ON")
    con.executescript(EXT_DDL.read_text(encoding="utf-8"))
    rows = con.execute(POPULATION_SQL, (args.since, args.until)).fetchall()

    picks, n_total, pop = draw(rows, args.seed, args.p, args.e)
    with con:
        con.execute(
            "INSERT OR IGNORE INTO audit_batches VALUES (?,?,?,?,?,?,?,?,?)",
            (args.batch, args.seed, RULE_VERSION, args.since, args.until, pop, n_total,
             json.dumps({"p": args.p, "e": args.e, "z": Z95}),
             datetime.now(timezone.utc).isoformat()),
        )
        con.executemany(
            "INSERT OR IGNORE INTO audit_samples VALUES (?,?,?,?,?)",
            [(args.batch, k.trace_id, k.stratum, k.sample_kind, k.weight) for k in picks],
        )

    summary = defaultdict(int)
    for k in picks:
        summary[(k.stratum, k.sample_kind)] += 1
    print(f"母體(不含全檢層){pop} 筆,目標隨機樣本 {n_total} 筆({n_total / pop:.2%})")
    for (s, kind), c in sorted(summary.items()):
        print(f"  {s:<18} {kind:<7} {c}")


if __name__ == "__main__":
    main()

audit_batches 把 seed、規則版本、時間範圍、參數一起存。三個月後有人問「9 月那批為什麼沒抽到某份文件」,答案要查得到,不能靠我記得。

抽樣跟判定分兩張表,抽樣又是 INSERT OR IGNORE,同一批重跑抽到同一組,填好的判定一筆都不動。稽核員看了一下午,被一次手滑重跑洗掉,想到就頭痛。

我照 Day 26 的 SQL 建了空資料庫,塞進合成的 12,000 條 trace 跑過,也重跑確認抽到同一組、判定還在。層別比例是我隨手設的,底下數字只代表程式行為,不是實測:

母體(不含全檢層)11942 筆,目標隨機樣本 440 筆(3.68%)
  census_truncated   census  58
  normal             random  362
  risk_flagged       random  46
  risk_flagged       risk    23
  risk_other         random  33
  risk_other         risk    17

隨機樣本 362 + 46 + 33 = 441,比 440 多一筆,是比例配置用 round 的誤差。每層至少一筆,比總數精準更重要,我接受。


人看完之後,錯誤率怎麼算回來

分層抽樣不能直接把錯的筆數除以總筆數。每一層被抽中的機率不一樣,要用層權重加回來:

p̂ = Σ (N_h / N) × p̂_h
Var(p̂) = Σ (N_h / N)² × (1 − n_h / N_h) × p̂_h(1 − p̂_h) / (n_h − 1)

(1 − n_h / N_h) 是每一層的有限母體校正。N_h 不用另外存,權重乘上這層抽出的筆數就是它。程式寫起來很短:

import math
import sqlite3
from collections import defaultdict

Z95 = 1.959964

# 取第一位稽核員的判定;第二位只拿來算一致率
SAMPLE_SQL = """
SELECT s.stratum, s.weight,
       (SELECT r.decision FROM audit_reviews r
         WHERE r.batch_id = s.batch_id AND r.trace_id = s.trace_id
         ORDER BY r.audited_at LIMIT 1) AS decision
FROM audit_samples s
WHERE s.batch_id = ? AND s.sample_kind = 'random'
"""


def estimate(con: sqlite3.Connection, batch_id: str) -> tuple[float, float, float]:
    acc = defaultdict(lambda: {"drawn": 0, "w": 0.0, "judged": 0, "wrong": 0})
    for stratum, w, decision in con.execute(SAMPLE_SQL, (batch_id,)):
        a = acc[stratum]
        a["drawn"] += 1
        a["w"] = w                                   # 同一層的權重都一樣
        if decision in ("model_wrong", "model_right", "gt_defect"):
            a["judged"] += 1
            a["wrong"] += decision == "model_wrong"  # gt_defect 代表模型沒錯

    strata = [(a["w"] * a["drawn"], a["judged"], a["wrong"]) for a in acc.values()]
    N = sum(N_h for N_h, _, _ in strata)
    rate, var = 0.0, 0.0
    for N_h, n_h, x_h in strata:
        if n_h == 0:
            raise ValueError("有一層還沒有任何有效判定,先別算")
        p_h = x_h / n_h
        W_h = N_h / N
        rate += W_h * p_h
        if n_h > 1:
            var += W_h ** 2 * (1 - n_h / N_h) * p_h * (1 - p_h) / (n_h - 1)

    half = Z95 * math.sqrt(var)
    return rate, max(0.0, rate - half), rate + half

unclear 不進分子也不進分母,等於假設看不出來的那些,錯誤率跟看得出來的差不多。這假設不一定成立,所以 unclear 太多時(門檻下面講),這個數字我不拿去報告。gt_defect 是文字層錯、模型對,算進分母但不算錯。

這裡用的是分層的常態近似,不是 Wilson。分層之後要做 Wilson 比較麻煩,我的折衷是:整體估計用這支,另外每一層各自再算一次 Wilson 區間當參考。如果某一層抽到的錯少於 5 筆,那一層的數字我只看 Wilson 上界,不看點估計。


稽核流程本身

統計只解決「抽哪些」。人怎麼看,影響一樣大。

我目前的流程寫成這樣,還沒有真的讓別人照著跑過,是設計:

稽核員看到的畫面只有三樣東西:區塊截圖、區塊在整頁中的位置框、模型輸出。Verifier 的判定刻意藏起來。我怕的是錨定效應,稽核員看到 Verifier 說 PASS,心裡就先鬆了一半,看得比較快。把機器的答案給人看,人就不獨立了。這跟 Day 11 說 Verifier 不能同源是同一件事。

判定就用 audits.decision 那四個選項,只是對象從一個 diff 事件換成整個區塊:model_right、model_wrong、unclear,外加 gt_defect(模型對、文字層錯)。unclear 會單獨統計;如果某一批 unclear 超過一成,代表稽核指引寫得不夠清楚,要先修指引。

每批另外隨機挑 10% 給第二個人重看,一樣寫進 audit_reviews,換一個 auditor。估計只取第一位的判定,不一致的先記下來。一致率太低的話,錯誤率算得再正確也沒什麼好信的。

model_wrong 的區塊要寫一句話說明錯在哪,然後進 Day 25 的飛輪:它會變成回歸測試的一筆,下次改 prompt 或換模型之前,先確認這些曾經錯過的區塊沒有再錯一次。

還有一條我最在意:稽核員發現的錯,不能改掉記錄層裡的模型輸出。判定寫進 audit_reviews,正確文字寫 audits.corrected_text,原文永遠保留。這是 Day 7「不覆寫原文」的延伸,改掉了,你就再也說不清楚系統當初錯成什麼樣子。


一個沒解決的問題

想像一下稽核員點開一筆 model_wrong。

messages.model 寫著 gemma-4-26b,strategy 寫著 tight_crop,dpi 是 400,attempt 是 2,finish_reason 是 stop,原始輸出的路徑跟雜湊都在。每個欄位都填得好好的。

她想問的是:這種區塊,一開始就該走這條路嗎?

整張 schema 沒有一個欄位答得出來。記錄層知道最後去了哪,不知道是誰決定的、依據哪條規則、那條規則是哪一版、當時有沒有別的路被考慮過又被跳過。

路是 Router 選的。那些號稱會自動替你挑模型的現成 LLM Router,選路的時候到底有沒有留下理由?


上一篇
Day 26 - JSONL/SQLite 記錄設計:資料怎麼變成校準燃料
系列文
從 LLM 到 Harness: 打造隱私與可信任的繁中進階 OCR Agent 共 27 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言