Text-to-SQL Agent Eval 有一種高風險失敗:SQL 沒有 syntax error,查詢順利跑完、答案也像真的,業務數字卻算錯。若測試資料仍持續更新,拿未綁定 frozen snapshot 的舊 golden rows 驗收新資料也會失真;live refresh 除了保存當下 rows,還要保存「如何在指定資料狀態下重建答案」的完整配方。
這篇會從零搭出一條可重跑路徑:讓 Agent SQL 與人工審過的 reference query 讀同一份 snapshot,依明確的結果契約處理排序、重複列、NULL、時區與近似數值,再保存 prompt、trace、SQL、schema/資料版本與 row diff。若你還不熟整體評測概念,可先搭配 AI Evals 新手教學;本文專攻資料會變動時,Text-to-SQL 的結果層驗收。
Text-to-SQL Agent Eval 要驗的是答案,不只是可執行
把資料庫想成一家還在營業的餐廳。你問「今天已付款訂單是多少」,Agent 交出一張能刷過收銀機的單據,只能證明格式可用;它可能把 pending 訂單也加進去。若你隔天再拿昨天的總額驗收,新的訂單又會讓正確 SQL 看似答錯。
因此先把三個概念拆開:
| 檢查層 | 它能證明什麼 | 它不能證明什麼 |
|---|---|---|
| Executability | SQL 能解析、能執行、沒有被 timeout | 查到的列與業務語意正確 |
| Result correctness | 候選結果符合當下 reference 結果與比較契約 | reference query 本身一定沒有寫錯 |
| Business invariants | 總額非負、租戶不越界、欄位範圍合理 | 所有語意錯誤都能被規則涵蓋 |
本文說的「真值」不是永遠正確的神諭,而是同一份 snapshot 上,由人工審查過的 reference query 與 invariants 在評測當下重算出的預期結果。它仍是程式碼,也要 code review、測試與版本管理。研究 benchmark 也早已從只看一個資料庫實例,走向跨資料變體或動態互動;例如官方 Test Suite SQL Evaluation 會在多個資料庫變體執行,正是在防止錯 SQL 因資料巧合而蒙混過關。

先定義一題:Reference Query、輸出契約與 Invariants
先用一題 SQLite 小案例貫穿全文。資料庫有 customers 與 orders;問題是:「截至 as_of,列出每位客戶已付款總額與最後付款時間。」正確規格至少包含四部分:
- 輸入:自然語言問題、可用 schema,以及固定的
as_of。 - Reference Query:由熟悉資料模型的人撰寫、審查並版本化的 SQL。
- Output Contract:需要哪些欄位、是否看順序、重複列如何算、數值與時間如何比。
- Invariants:例如不得跨 tenant、金額不可為負、結果客戶必須存在。
-- reference.sql:只計入已付款,並注入固定 as_of
SELECT
c.name AS customer,
SUM(o.amount_usd) AS total_usd,
strftime('%Y-%m-%dT%H:%M:%SZ', MAX(unixepoch(o.paid_at)), 'unixepoch')
AS last_paid_at
FROM customers AS c
LEFT JOIN orders AS o
ON o.customer_id = c.id
AND o.status = 'paid'
AND unixepoch(o.paid_at) <= unixepoch(:as_of)
GROUP BY c.id, c.name;
一條「查得到但答錯」的候選 SQL 可能只把條件寫成 o.status <> 'cancelled'。它沒有報錯,結果也有客戶、金額與時間;但 pending 訂單被算進去,Alice 的 30.30 美元就膨脹成 130.30 美元。這不是 execution error,而是 result mismatch。
這個 fixture 使用的 unixepoch() 是 SQLite 3.38.0 加入的 convenience function;本次驗證環境是 SQLite 3.51.0。若 CI 還停在更舊 runtime,先升級或依 SQLite 3.38 release notes 改用相容的 strftime('%s', ...) 寫法。
把題目規格放進版本控制,做法很像 Specification-first 開發:先定義可驗收的介面,再讓 Agent 產 SQL。不要等模型輸出後,才臨時決定什麼答案算對。
第一步:把兩條 SQL 鎖在同一份資料快照
若先在 live database 跑 candidate、隔幾秒再跑 reference,中間剛好有訂單寫入,兩條 SQL 就算語意完全相同也可能得到不同結果。PostgreSQL 預設的 Read Committed 正是每個 statement 取得新 snapshot;官方 transaction isolation 文件 明確說明,同一 transaction 內連續兩個 SELECT 仍可能看到不同提交。
SQLite:先做獨立 snapshot 檔
本機與 CI 的一個簡單做法,是用 Python 標準庫的 Connection.backup() 產生獨立資料庫。SQLite 官方說明,完成的 Online Backup 會得到一致 snapshot;若來源採 WAL,別只複製主 .db 檔,因為 WAL 也是資料庫持久狀態的一部分。
from pathlib import Path
import sqlite3
source = sqlite3.connect("file:live.db?mode=ro", uri=True)
target = Path("eval-v2.sqlite")
if target.exists():
raise FileExistsError(target)
snapshot = sqlite3.connect(target)
source.backup(snapshot)
snapshot.close()
source.close()
# 評測時只打開 snapshot,不再讀 live.db
uri = Path("eval-v2.sqlite").resolve().as_uri()
db = sqlite3.connect(uri + "?mode=ro&immutable=1", uri=True)
db.execute("PRAGMA query_only = ON")
mode=ro 讓這個 connection 不能寫入;immutable=1 則是 harness 對 SQLite 的保證:這個 snapshot artifact 不會再改。官方 immutable URI 文件 提醒,若檔案其實仍會變動,SQLite 可能回傳錯誤結果,甚至報 SQLITE_CORRUPT,所以它不能拿來當權限邊界。上面的簡化例子先拒絕覆寫既有目標;正式 pipeline 還應寫到全新暫存檔,驗證 hash 與可讀性後再原子發布,避免 backup 失敗留下半成品。SQLite 也明說 PRAGMA query_only 不等於真正的安全 sandbox。若 SQL 來自不受信任的 Agent,仍要加 authorizer、timeout、輸出上限與程序隔離。
PostgreSQL:同一 transaction 共用 snapshot
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY;
SET LOCAL TIME ZONE 'UTC';
SET LOCAL DateStyle TO 'ISO, YMD';
-- execute reference SQL
-- 再執行 candidate SQL;避免不受信任查詢先改變 session/暫存狀態
-- execute candidate SQL
ROLLBACK;
若兩條 SQL 必須分散到不同 worker,可用 PostgreSQL 的 pg_export_snapshot() 匯出暫時 snapshot,再由其他 session 在第一個查詢前匯入。不過 exporter 必須保持 transaction 開啟;要做數天後仍可重播的 artifact,應改存不可變資料庫映像或用 pg_dump --snapshot 產出持久備份。
第二步:在評測當下重算 Current Ground Truth
綁定 frozen snapshot 的 golden rows 很適合歷史 regression replay;問題出在拿舊 rows 直接驗已變動的 live data。Current ground truth 會另外保存「重建答案的配方」:每次評測先決定 snapshot 與 as_of,再在那份 snapshot 上執行 reference query。資料從 v2 前進到 v3 時,不必人工改每一列預期值;同一條 reference 會生成 v3 對應的預期結果。
snapshot_id + data_version + as_of。時間也要當輸入,不要讓 SQL 自己偷讀牆上時鐘。PostgreSQL 的 CURRENT_TIMESTAMP 固定在 transaction 起點,clock_timestamp() 卻會在執行中變動;官方 date/time functions 對兩者有清楚區分。亂數、sequence、外部 UDF 或遠端 API 也一樣:能注入就固定,不能固定就標成不適合 deterministic comparison。
Reference query 不是唯一防線。再加一組獨立 invariants,例如「結果不得含其他 tenant」、「付款總額不得小於退款後下限」、「總列數不可超過可見客戶數」。它們能抓 oracle 與 candidate 同時犯的某些錯,但不能取代 reference;兩者是交叉驗證,不是二選一。
第三步:先寫比較契約,再做 Row Diff
資料列不是轉成 JSON 後做 == 就結束。比較規則必須由題目語意決定,並跟測試案例一起版本化:
| 維度 | 預設安全做法 | 何時改規則 |
|---|---|---|
| 列順序 | 題目沒要求順序時,以 multiset 比較 | 排行、Top N 或逐列序列屬答案時,先明訂 ties policy |
| 重複列 | 保留 multiplicity,用 Counter,不要用 set | 只有規格明說去重時才忽略 |
| NULL | 使用獨立 typed sentinel | 不可自動變成 0、空字串或文字「NULL」 |
| 數值 | decimal/numeric 精確比;浮點依欄位定 tolerance | 金額可依業務幣別量化,不能全站硬四捨五入兩位 |
| 時間 | 解析帶時區的 typed value,再轉 UTC | 若題目要比較當地日期,時區就是契約的一部分 |
| 欄位 | 先驗名稱、型別與必要欄位,再比 rows | 別讓欄位錯置因值剛好相同而通過 |
PostgreSQL 不保證沒有最外層 ORDER BY 的輸出順序;但這不代表所有題目都必須看順序。題目只問「有哪些客戶」時,multiset 足夠;題目問「營收前十名」時,還要先定義邊界同分怎麼辦。若產品一定要 exactly 10 rows,需用唯一 total order 打破平手;若業務要保留第十名的全部 ties,則應採 tie-preserving 契約,例如 PostgreSQL FETCH ... WITH TIES。另一方面,SELECT ALL 是預設並保留重複列,官方 select-list 文件 也說明只有 DISTINCT 才去重,所以用 Python set(rows) 會把一類真錯誤吃掉。
NULL 與數值也不能靠直覺。一般 SQL 比較碰到 NULL 會得到 unknown;需要 null-safe equality 時,可參考 PostgreSQL 的 IS NOT DISTINCT FROM。numeric 是精確值,real/double precision 則是近似值;官方 numeric type 說明 並沒有替所有業務提供一個萬用 tolerance,門檻必須按欄位定義。

第四步:組一個可重跑的 Text-to-SQL Agent Eval Harness
下面是 Python 標準庫版本的簡化核心骨架。它假設前一步已產生 eval-v2.sqlite,而且 candidate 也使用契約中的三個欄位 alias,再把 reference 與 candidate 送進同一個唯讀 runner。為了讓流程容易閱讀,片段沒有包含 fixture DDL、資料 mutation、authorizer、manifest 寫檔與 replay CLI;下一節的可重跑結果來自包含這些元件的完整測試 harness,不要把這段片段誤當成可直接部署的完整程式。完整系統可接到你現有的 AI Agent Harness;關鍵不是框架名稱,而是資料、比較器與 trace 都能被同一個 run_id 找回。
from collections import Counter
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_EVEN
from pathlib import Path
import math, sqlite3, time
CONTRACT = {
"columns": ("customer", "total_usd", "last_paid_at"),
"row_order": "ignore",
"duplicates": "preserve",
"money_decimals": 2,
"timezone": "UTC",
}
def open_snapshot(path):
uri = Path(path).resolve().as_uri() + "?mode=ro&immutable=1"
db = sqlite3.connect(uri, uri=True)
db.execute("PRAGMA query_only = ON")
return db
def run_select(path, sql, params, timeout_s=2):
db = open_snapshot(path)
trace, deadline = [], time.monotonic() + timeout_s
try:
db.set_progress_handler(
lambda: 1 if time.monotonic() > deadline else 0, 1000
)
db.set_trace_callback(trace.append)
cur = db.execute(sql, params)
return [d[0] for d in cur.description], cur.fetchall(), trace
finally:
db.close()
def normalize(columns, rows):
if len(columns) != len(set(columns)):
raise ValueError(f"duplicate column names: {columns}")
if (CONTRACT["row_order"], CONTRACT["duplicates"],
CONTRACT["timezone"]) != ("ignore", "preserve", "UTC"):
raise NotImplementedError("this compact example supports one policy")
pos = {name: i for i, name in enumerate(columns)}
if set(pos) != set(CONTRACT["columns"]):
raise ValueError(f"column mismatch: {columns}")
bag = Counter()
for row in rows:
customer = row[pos["customer"]]
amount = row[pos["total_usd"]]
stamp = row[pos["last_paid_at"]]
if not isinstance(customer, str):
raise TypeError("customer must be text")
if stamp is not None:
if not isinstance(stamp, str):
raise TypeError("timestamp must be ISO-8601 text")
parsed = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("naive timestamp rejected")
stamp = parsed.astimezone(timezone.utc).isoformat()
if amount is not None:
if isinstance(amount, bool) or not isinstance(
amount, (int, float, Decimal)
):
raise TypeError("total_usd must be numeric")
if isinstance(amount, float) and not math.isfinite(amount):
raise ValueError("non-finite float rejected")
quantum = Decimal(1).scaleb(-CONTRACT["money_decimals"])
decimal_amount = Decimal(str(amount))
if not decimal_amount.is_finite():
raise ValueError("non-finite decimal rejected")
normalized_amount = decimal_amount.quantize(
quantum, rounding=ROUND_HALF_EVEN
)
if normalized_amount == 0:
normalized_amount = abs(normalized_amount)
item = (
("text", customer),
("null", None) if amount is None else (
"decimal",
str(normalized_amount),
),
("null", None) if stamp is None else ("timestamp_utc", stamp),
)
bag[item] += 1
return bag
def evaluate(path, reference_sql, candidate_sql, params):
ref = run_select(path, reference_sql, params)
got = run_select(path, candidate_sql, params)
expected = normalize(ref[0], ref[1])
actual = normalize(got[0], got[1])
return {
"equal": expected == actual,
"missing": list((expected - actual).items()),
"unexpected": list((actual - expected).items()),
"trace": {"reference": ref[2], "candidate": got[2]},
}
這段程式故意把 NULL、decimal、timestamp 標成不同型別,並以 Counter 保留相同 row 出現幾次。簡化版要求欄位 alias 完全相同;完整實測版則顯式設定 alias mapping,並用 authorizer 允許 fixture 讀取所需的 SELECT、READ、FUNCTION 與 RECURSIVE action classes。這個 action-level guard 是測試 fixture 的防線,不是 production function allowlist;正式 runner 應再按 function name 放行安全函式,並加入 row/byte 上限。單靠 progress handler 只能限制部分計算時間,不是完整隔離。
把失敗保存成 Replay Artifact
不要只留一行 score=0。至少把下列欄位寫進 JSON manifest,snapshot 則放在受控 artifact storage:
case_id、原始 prompt、schema prompt 與 Agent trace。- candidate SQL、reference SQL、bindings,以及 reference 的人工審查版本。
snapshot_sha256、schema hash、應用層 data version、as_of。- 資料庫、driver、比較器與 normalization contract 的版本。
- expected/actual row count、missing/unexpected multiset 與 diff hash。
- timeout、錯誤類型、執行時間,以及是否真的呼叫模型。
PRAGMA data_version 不能直接當跨機器 artifact ID;SQLite 官方說它只適合比較同一 connection 上不同時間的回傳。較穩妥的方式,是在資料更新 transaction 裡同步寫入應用層 logical_data_version,再搭配 snapshot 與 schema SHA-256。trace 的結構化做法可延伸閱讀 AI Agent Observability。
AlphaLab 本機可重跑示範:沒有模型,也能先驗 Harness
為了把「能重播」做成可驗證敘述,我們在 2026 年 8 月 20 日用 Python 3.9.6、SQLite 3.51.0 與標準庫建立一個固定 fixture。它沒有呼叫 LLM 或任何 API;candidate 是刻意漏掉 status='paid' 的靜態錯誤 SQL,所以這次只測 evaluator mechanics,不代表任何模型的準確率。
| 檢查 | 實際結果 |
|---|---|
| 自動化測試 | 7/7 通過,涵蓋欄位/列順序、時區、typed NULL、half-even、重複列、寫入拒絕與 replay |
| 資料漂移 | snapshot 保持 logical v2;snapshot 完成後 live database 前進到 v3 |
| 第一次評測 | equal=false,1 筆 missing、1 筆 unexpected |
| Row diff | Alice 預期 30.30/01:30Z,錯誤候選得到 130.30/03:00Z |
| 重播 | 五項 snapshot/schema/版本/時間/runtime integrity check 全通過,diff SHA-256 相同 |
# CLI 結果的人工排版摘要
snapshot logical version: 2
live version after snapshot: 3
model_invoked: false
equal: false
missing multiset entries: 1
unexpected multiset entries: 1
diff SHA-256:
6090913611b262464052c2fbeefb734405b0357016a5476de48db736449d14c8
reproduced: true
expected_failure_reproduced: true
all five integrity checks: true
這個結果證明的是:snapshot 之後活資料即使新增一筆回溯時間的訂單,重播仍能在 v2 重現同一個失敗。它沒有測 concurrent writer 恰好插入 backup 中途,也沒有證明 reference SQL 對其他 schema 都正確;這兩點要由整合測試與 oracle review 另行補上。
接到 LangSmith 等 Eval Runner
Harness core 不必綁定平台。LangSmith 目前的 evaluation workflow 接受 dataset、target function 與自訂 evaluator;可把 snapshot hash、資料版本與 comparator version 放進 experiment metadata,再讓 evaluator 回傳 sql_result_equal 與 row diff 摘要。平台負責 experiment 與比較介面;若 target/client 已依文件加上 tracing instrumentation,也能保留 trace。你的 database adapter 才負責 snapshot 與 deterministic result comparison。
results = client.evaluate(
target, # prompt -> {"sql": candidate_sql}
data=examples,
evaluators=[execution_correct],
experiment_prefix="text2sql-snapshot-v2",
max_concurrency=1,
metadata={
"db_snapshot_sha256": snapshot_sha256,
"data_version": 2,
"comparator_version": "bag-v1",
},
)
max_concurrency=1 與上述 metadata 是本文的評測設計,不是 LangSmith 強制要求。若要比較不同模型或供應商,先固定 snapshot、prompt、tool schema 與 comparator,再重跑同一套案例;這也能與 AI Agent 搜尋 API 評測 的「固定任務、分層驗收」方法互相參照。
失敗如何進入 Regression Suite
每次 production incident 或人工抽查發現錯答,不要只修 prompt。先保存當時 snapshot 與 manifest,再把案例分成兩種:
- Frozen replay:永遠對同一份資料 artifact 重播,確認舊錯不復發;適合 CI。
- Live refresh:在新資料版本上重新執行 reference,確認規則面對資料漂移仍成立;適合排程 smoke test。
兩者不能互相取代。Frozen replay 可重現,卻抓不到新 schema、新值域與資料分布;live refresh 能看到漂移,卻不能單獨證明歷史事故已被修復。一個實用的 suite 是兩層並行:每次 commit 跑小型 frozen cases,每日或每週對隔離副本跑 current-ground-truth cases。
上正式資料前:唯讀不是唯一安全邊界
Agent 產生的 SQL 要視為不受信任輸入。依 threat model 調整之外,建議至少包含:
- 使用專用、最小權限的唯讀 DB role 或隔離 replica,不共用應用程式寫入帳號。
- 只暴露必要 view/schema,保留 tenant 與 row-level policy;不要把 prompt 當授權層。
- 設定 statement timeout、transaction timeout、row/byte cap 與並發上限。
- 拒絕多 statement、寫入語句、危險 function 與未允許 extension。
- 將 query runner 放進隔離程序或容器,限制 CPU、memory、網路與檔案存取。
- log 寫入前先遮蔽 PII;snapshot 與 manifest 依資料敏感度加密、限權與到期。
READ ONLY、SQLite authorizer 或 SQL parser 都只是防線之一,不是完整安全證明。Production runner 還要處理成本與資源耗盡,例如笛卡兒積、超大輸出或昂貴 function;真正的邊界在資料庫權限與執行環境,不在「請勿修改資料」這句 prompt。
7 個常見錯法
- 只驗 SQL 沒報錯:這只完成 executability,沒有驗 result correctness。
- Candidate 與 reference 各讀一次 live DB:資料漂移會製造假 FAIL 或假 PASS。
- 把 rows 轉成 set:重複列數量消失;預設應用 multiset。
- 所有浮點一律 round(2):這是某些金額欄位的契約,不是通用真理。
- 只靠 invariants:「金額非負」通過,不代表漏掉付款狀態篩選。
- Manifest 沒有 snapshot:只有 hash 卻拿不回資料,歷史失敗仍無法重播。
- 把 reference 當永不會錯:oracle 的 schema、SQL 與 reviewer 也要版本化與測試。
Text-to-SQL Agent Eval FAQ
1. Golden SQL 和 Golden Rows 要留哪一個?
兩者用途不同。Golden SQL/reference query 是可在新 snapshot 重算的配方;frozen rows 是歷史 regression 的證據。Live refresh 以前者為主,事故重播則兩者都留。
2. 沒有 ORDER BY 就一定判錯嗎?
不一定。如果題目只要求成員集合,忽略順序但保留重複數即可;若題目要求排行、Top N 或逐列序列,才把順序列入契約。Exactly N 的 deterministic output 需要能打破平手的 total order;若規格保留邊界同分,則按 ties-aware 契約比較。
3. 可以讓另一個 LLM 判兩份答案是否相同嗎?
可當補充,不應取代 deterministic comparator。欄位、型別、multiset 與 tolerance 都能由程式精確驗證;LLM judge 適合處理自然語言說明,不適合把明確 row diff 變成另一個機率問題。
4. Snapshot 會不會讓測試看不到真實世界?
單一 snapshot 只代表一個狀態。所以同時維護 frozen replay 與多個 live-refresh fixtures,覆蓋 schema 版本、空資料、重複值、NULL、時區切換與極端分布。
5. Reference Query 太複雜怎麼辦?
把 oracle 拆小。用經審查的 view、簡單 aggregation、獨立 invariants 與手工 micro-fixture 交叉驗證。若 reference 與 candidate 共享同一段錯誤商業邏輯,結果相同也只是共同犯錯。
6. 先從多少案例開始?
先挑一小組高風險問題。刻意放入付款狀態、日期邊界、NULL、重複列、同分排序與 tenant 權限;每次真實失敗再新增一個 frozen case,讓 suite 逐步覆蓋真正發生過的錯誤模式。
給新手的 7 個重點
- SQL 沒報錯只代表可執行,不代表答案正確。
- Candidate 與 reference 必須讀同一 snapshot、同一
as_of。 - Live refresh 要保存可重建配方;歷史 regression 仍保留綁定 snapshot 的 frozen rows。
- 先寫 output contract;排序、NULL、重複列、時間與數值沒有萬用規則。
- 用 multiset row diff 保存 missing 與 unexpected,不只留一個分數。
- Prompt、trace、SQL、snapshot、schema、版本與比較器要能綁回同一個 run。
- Frozen replay 防舊錯復發,live refresh 抓新漂移;可靠 suite 需要兩者。
接著閱讀
左右滑動查看更多推薦
結語:先讓一個錯答案能被原地重現
回到開頭的記憶把手:可靠 SQL Eval =同一份資料快照 × 當下 Reference Query × 明確比較契約 × 可重播證據。這四項缺一,Text-to-SQL Agent 都可能在「查得到」的表象下悄悄答錯,或在資料變動後留下無法重現的紅燈。
今天先選一個最常被問、又牽涉狀態或日期條件的 SQL 問題:寫好 reference 與 output contract,對資料庫做 snapshot,放入一條故意漏條件的 candidate,再確認 harness 能產生 row diff、保存 manifest、離線重播同一失敗。當這條最小路徑穩定後,再接上真正的 Agent 與排程;你建立的就不只是一次 demo,而是一套會隨資料與事故累積而變強的 regression system。






