File size: 5,421 Bytes
12eb3a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | #!/usr/bin/env python3
"""決定性檢查：同一組條件重複執行，逐位元組比對每一個 token 的 logits。
為什麼需要這個工具
------------------
`sdq/experiments.py` 會用「同一個提示詞跑幾次，第一個生成 token 是否相同」當
正確性指紋。但在 S7/S8 那兩輪，這個指紋偶爾會跳出「⚠️輸出不一致」，
看起來像是 prefill 的累加順序或執行緒交錯造成的非確定性。
實際上那個警告是 **harness 自己的解析問題**：`re.search(r"assistant:\\s*(.*)")`
的 `\\s*` 會跨過換行，assistant 第一行是空的時候就抓到下一行的內容，
而下一行在哪裡換行取決於 stdout 有沒有剛好切成兩個 chunk。
這個工具直接比對 logits（`--dump-logits --logits-out`），把「是不是真的非確定」
用數據回答，而不是靠字串。量測結果（2026-10-04，本機，預設條件）：
4 次相同執行 → 48 個 token × 248320 個 vocab 的 logits **全部逐位元組相同**
也就是說分頁路徑在同一台機器、同一組條件下是**決定性**的；
設定 `--pager-off` 之外的任何變數才會讓輸出改變。
用法
----
python tools/determinism_check.py --model <gguf> --bin <epq-chat> -n 4
python tools/determinism_check.py --model <gguf> --bin <epq-chat> \\
--runs 3 --env SDQ_FORCE_UNIFORM_SLOT=1 # 比較不同設定是否也一致
"""
from __future__ import annotations
import argparse
import os
import struct
import subprocess
import sys
import tempfile
import time
from pathlib import Path
import numpy as np
REPO = Path(__file__).resolve().parent.parent
DEFAULT_PROMPT = "請用三句話說明什麼是 SSD、RAM、CPU 之間的資料分層。"
def load_logits(path: str) -> np.ndarray:
"""讀 epq-chat --logits-out 寫的檔案：(n_token, vocab) 的 float32。"""
b = Path(path).read_bytes()
n = struct.unpack_from("<i", b, 0)[0]
a = np.frombuffer(b, dtype=np.float32, offset=4)
k = a.size // n
return a[: k * n].reshape(k, n)
def run_once(binary: str, model: str, prompt: str, out: str, n_predict: int,
env_extra: dict, threads: int, ctx: int) -> float:
env = dict(os.environ)
env.update(env_extra)
cmd = [binary, "-m", model, "-p", prompt, "-n", str(n_predict), "-c", str(ctx),
"-t", str(threads), "--temp", "0", "--no-chat", "--no-interactive",
"--dump-logits", "--logits-out", out, "--ram-budget-mb", "8192"]
t0 = time.time()
p = subprocess.run(cmd, capture_output=True, text=True, env=env)
if p.returncode != 0:
print(f"[det] 執行失敗 rc={p.returncode}\n{p.stderr[-1500:]}")
sys.exit(2)
return time.time() - t0
def main() -> int:
ap = argparse.ArgumentParser(description="SDQ 決定性檢查（逐位元組比對 logits）")
ap.add_argument("--model", required=True)
ap.add_argument("--bin", default=str(REPO.parent / "llama.cpp" / "build" / "bin" / "epq-chat"),
help="epq-chat 路徑（預設 <repo>/../llama.cpp/build/bin/epq-chat）")
ap.add_argument("--runs", type=int, default=4, help="重複執行幾次")
ap.add_argument("-n", "--n-predict", type=int, default=16,
help="生成幾個 token（每個都會存一份 logits）")
ap.add_argument("--threads", type=int, default=8)
ap.add_argument("--ctx", type=int, default=4096)
ap.add_argument("--prompt", default=DEFAULT_PROMPT)
ap.add_argument("--env", action="append", default=[],
help="額外的環境變數（可重複），用來比較不同設定")
args = ap.parse_args()
env_extra = dict(kv.split("=", 1) for kv in args.env)
print(f"[det] 條件：runs={args.runs} n_predict={args.n_predict} "
f"threads={args.threads} ctx={args.ctx} env={env_extra or '{}'}")
with tempfile.TemporaryDirectory() as td:
mats = []
for r in range(args.runs):
out = str(Path(td) / f"logits_{r}.bin")
dt = run_once(args.bin, args.model, args.prompt, out, args.n_predict,
env_extra, args.threads, args.ctx)
m = load_logits(out)
mats.append(m)
print(f" run {r}: {dt:5.1f}s logits shape={m.shape}")
ref = mats[0]
all_same = True
for r, m in enumerate(mats[1:], start=1):
if m.shape != ref.shape:
print(f" ❌ run{r} 的 shape 與 run0 不同：{m.shape} vs {ref.shape}")
all_same = False
continue
diff_rows = np.where(np.abs(ref - m).max(axis=1) > 0)[0]
if diff_rows.size == 0:
print(f" ✅ run{r} 與 run0 **逐位元組完全相同**"
f"（{ref.shape[0]} token × {ref.shape[1]} vocab）")
else:
all_same = False
i = int(diff_rows[0])
d = float(np.abs(ref[i] - m[i]).max())
print(f" ❌ run{r} 從第 {i} 個 token 開始有差異（max abs={d:.3e}，"
f"argmax run0={ref[i].argmax()} run{r}={m[i].argmax()}）")
print()
print("[det] 結論：", "決定性 ✅（同條件重複執行逐位元組一致）" if all_same
else "非決定性 ❌（同條件重複執行會有差異）")
return 0 if all_same else 1
if __name__ == "__main__":
sys.exit(main())
|