Evaluator tự động: code check, LLM-as-judge và tỉ lệ thành công "thật"

5.1Từ failure mode tới metric

Trực giác. Chương 3 giống bác sĩ khám và chẩn đoán: đọc từng trace, gọi tên các kiểu lỗi. Chương 4 giúp cả nhóm thống nhất chẩn đoán. Chương 5 là lúc lắp máy đo: với mỗi kiểu lỗi, ta muốn một phép đo tự động trả lời câu hỏi "trace này có mắc lỗi X không?" trên hàng nghìn trace mỗi ngày, không cần người đọc từng cái. Một metric tốt = một câu hỏi có/không, gắn với đúng một failure mode.

Ví dụ xuyên suốt của sách là một trợ lý CRM bất động sản: hiểu câu hỏi của môi giới, sinh SQL tìm nhà, soạn email cho khách, đọc lịch, tra cứu web. Error analysis đã cho ra một danh sách lỗi kiểu: SQL bỏ quên điều kiện lọc người dùng đã nêu, gộp nhóm sai theo ngày/tuần, bịa ra tool không tồn tại, email thiếu thông tin, giọng văn lệch persona của khách, gọi tool sai thứ tự, gọi tool không ai yêu cầu, hiểu nhầm địa danh trùng tên.

Sách đưa ra hai lý do để lọc lỗi Specification trước: (1) hiệu quả — evaluator tốn công xây, gán nhãn, căn chỉnh, bảo trì; đừng tiêu cho thứ sửa được bằng một câu trong prompt; (2) đo đúng thứ cần đo — nếu yêu cầu chưa được nói ra, điểm eval phản ánh độ may mắn đoán ý, không phản ánh năng lực của hệ thống.

5.2Reference-based vs reference-free

Trực giác. Reference-based giống chấm bài thi có đáp án: so output với một output "vàng" đã được người tuyển chọn. Reference-free giống chấm bài theo quy chế: không cần đáp án, chỉ cần kiểm các thuộc tính mà mọi output tốt đều phải có. Cả hai có thể đi kèm executability check — chạy thật output để xem nó chạy được, không chỉ trông đúng.

SELECT * FROM listings WHERE district='Q7' AND bedrooms >= 2

Code minh hoạ reference-free check ở bước 2 (tự viết, luật tách ràng buộc cố ý đơn giản):

import json, re

# --- (1) Reference-free: SQL có giữ đủ ràng buộc người dùng đã nêu? ---
def extract_constraints(request: str) -> dict:
    """Tách ràng buộc từ câu hỏi của môi giới (luật đơn giản, đủ cho demo)."""
    c = {}
    if m := re.search(r"dưới\s*(\d+)\s*(tỷ|triệu)", request):
        c["price"] = int(m[1]) * (10**9 if m[2] == "tỷ" else 10**6)
    if m := re.search(r"(\d+)\s*phòng ngủ", request):
        c["bedrooms"] = int(m[1])
    if re.search(r"nuôi (chó|mèo|thú)", request):
        c["pets_allowed"] = True
    return c

def sql_keeps_constraints(request: str, sql: str) -> dict:
    where = sql.lower().split("where", 1)[1] if "where" in sql.lower() else ""
    missing = []
    for col, val in extract_constraints(request).items():
        if col == "price" and not re.search(rf"price\s*(<=|<)\s*{val}\b", where):
            missing.append(col)
        elif col == "bedrooms" and not re.search(rf"bedrooms\s*(>=|=)\s*{val}\b", where):
            missing.append(col)
        elif col == "pets_allowed" and not re.search(r"pets_allowed\s*=\s*(true|1)", where):
            missing.append(col)
    return {"pass": not missing, "missing": missing}

if __name__ == "__main__":
    req = "Tìm căn 2 phòng ngủ dưới 5 tỷ ở Quận 7, nhà có nuôi chó"
    good = "SELECT * FROM listings WHERE district='Q7' AND bedrooms >= 2 AND price <= 5000000000 AND pets_allowed = true"
    bad  = "SELECT * FROM listings WHERE district='Q7' AND bedrooms >= 2"
    print(sql_keeps_constraints(req, good))   # {'pass': True, 'missing': []}
    print(sql_keeps_constraints(req, bad))    # {'pass': False, 'missing': ['price', 'pets_allowed']}

Và phiên bản reference-based theo kết quả thực thi (bước 1 + 3 gộp lại): chạy cả SQL sinh ra và SQL vàng trên một DB nhỏ trong bộ nhớ, so tập dòng trả về. Hai câu SQL viết khác nhau nhưng cùng nghĩa vẫn Pass — điều mà so chuỗi không làm được (tự viết, chỉ dùng thư viện chuẩn):

import sqlite3

ROWS = [  # (district, bedrooms, price_ty, pets_allowed)
    ("Q7", 2, 4.8, 1), ("Q7", 2, 5.6, 1), ("Q7", 3, 4.9, 0),
    ("Q7", 1, 3.2, 1), ("Q2", 2, 4.5, 1), ("Q7", 2, 4.1, 0),
]

def run(sql):
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE listings (district TEXT, bedrooms INT, price REAL, pets_allowed INT)")
    db.executemany("INSERT INTO listings VALUES (?,?,?,?)", ROWS)
    try:
        return set(db.execute(sql).fetchall()), None
    except sqlite3.Error as e:
        return None, str(e)

def same_result(sql_gen, sql_gold):
    """Reference-based theo KẾT QUẢ THỰC THI: hai SQL viết khác nhau vẫn Pass nếu trả về cùng tập dòng."""
    got, err = run(sql_gen)
    if err:
        return {"pass": False, "reason": f"không chạy được: {err}"}
    want, _ = run(sql_gold)
    return {"pass": got == want, "extra": len(got - want), "missing": len(want - got)}

if __name__ == "__main__":
    gold = ("SELECT * FROM listings WHERE district='Q7' AND bedrooms >= 2 "
            "AND price <= 5 AND pets_allowed = 1")
    alt  = ("SELECT * FROM listings WHERE pets_allowed = 1 AND price BETWEEN 0 AND 5 "
            "AND bedrooms > 1 AND district = 'Q7'")                 # viết khác, cùng nghĩa
    bad  = "SELECT * FROM listings WHERE district='Q7' AND bedrooms >= 2"
    print(same_result(alt, gold))    # {'pass': True, 'extra': 0, 'missing': 0}
    print(same_result(bad, gold))    # {'pass': False, 'extra': 3, 'missing': 0}
    print(same_result("SELEC * FROM listings", gold))   # không chạy được

Lưu ý: DB thử phải có đủ "dòng bẫy" (căn quá giá, căn không cho nuôi thú...) thì mới phân biệt được SQL thiếu điều kiện. DB thử quá nghèo = check luôn Pass.

5.3Code check hay LLM-as-judge?

Trực giác. Nếu bạn viết được một unit test cho tiêu chí đó, đừng thuê giám khảo. Code check nhanh, rẻ, deterministic, dễ đọc: cùng input luôn cùng kết quả, sai thì đọc code là biết vì sao. LLM judge hiểu sắc thái nhưng đổi lại có bias, dao động giữa các lần chạy, tốn tiền inference và phải được kiểm định. Nguyên tắc: dùng code tới mức tối đa, chỉ gọi judge cho phần thật sự cần diễn giải.

Hai quy tắc của sách đáng nhớ: judge là một LLM riêng (khác lời gọi của app), và mỗi judge chỉ lo một failure mode. Nhiều khía cạnh chất lượng → nhiều judge; đó là điều bình thường. Judge làm tốt nhất với câu hỏi hẹp và nhị phân.

Mẫu hình hay dùng: cổng code trước, judge sau

Một bộ code check điển hình cho output dạng JSON (tự viết):

import json

# --- (2) Bộ check cấu trúc/luật cho output dạng JSON ---
TOOL_REGISTRY = {"search_listings", "get_calendar", "send_email"}

def check_output(raw: str, max_bullets=3, banned=("cam kết chắc chắn", "đảm bảo 100%")):
    res = {}
    try:
        obj = json.loads(raw); res["json_valid"] = True
    except json.JSONDecodeError:
        return {"json_valid": False}
    res["has_keys"] = {"summary", "tool_calls"} <= obj.keys()
    bullets = [l for l in obj.get("summary", "").splitlines() if l.strip().startswith("-")]
    res["bullets_ok"] = len(bullets) <= max_bullets
    res["no_banned"] = not any(b in obj.get("summary", "").lower() for b in banned)
    res["tools_known"] = all(t.get("name") in TOOL_REGISTRY for t in obj.get("tool_calls", []))
    return res

if __name__ == "__main__":
    out = '{"summary": "- Căn A\\n- Căn B", "tool_calls": [{"name": "reschedule_viewing"}]}'
    print(check_output(out))   # tools_known: False — reschedule_viewing không có trong registry
    print(check_output("xin lỗi, tôi không chắc"))   # {'json_valid': False}

5.4Bias của LLM judge & cách thăm dò

Sách dẫn nghiên cứu của Zheng et al. (2023, nhóm LMSYS / Chatbot Arena) về ba kiểu bias kinh điển khi dùng LLM làm giám khảo:

Bias không làm judge vô dụng, nhưng có nghĩa là điểm judge thô không tự đứng được — cần prompt tốt và quy trình kiểm định (mục 5.6–5.9). Tin tốt: bạn có thể tự thăm dò bias của judge trên dữ liệu của chính mình với vài chục ví dụ.

Công cụ thăm dò nhỏ (tự viết): chạy judge nhiều lần để đo độ ổn định, và so cặp hai chiều để trung hoà position bias. call_llm là hàm gọi model của bạn (ở đây thay bằng hàm giả để chạy thử).

import random
from collections import Counter
from judge_prompt import parse_verdict

def stability(call_llm, prompt, n=5):
    """Chạy judge n lần (temperature > 0). flip_rate > 0 = tiêu chí còn mơ hồ với ca này."""
    votes = Counter(parse_verdict(call_llm(prompt)) for _ in range(n))
    top, cnt = votes.most_common(1)[0]
    return {"votes": dict(votes), "majority": top, "flip_rate": 1 - cnt / n}

def pairwise_swapped(call_llm, request, ans_a, ans_b):
    """Judge so cặp, chấm cả hai thứ tự để trung hoà position bias."""
    tpl = "Yêu cầu: {r}\nTrả lời 1: {x}\nTrả lời 2: {y}\nCâu nào tốt hơn? Chỉ trả về '1', '2' hoặc 'hoà'."
    first = call_llm(tpl.format(r=request, x=ans_a, y=ans_b)).strip()   # A đứng trước
    second = call_llm(tpl.format(r=request, x=ans_b, y=ans_a)).strip()  # B đứng trước
    win1 = {"1": "A", "2": "B"}.get(first, "hoà")
    win2 = {"1": "B", "2": "A"}.get(second, "hoà")
    return win1 if win1 == win2 else "hoà (không nhất quán)"

if __name__ == "__main__":
    flaky = lambda _: random.choice(['{"answer": "Pass"}', '{"answer": "Fail"}'])
    print(stability(flaky, "prompt bất kỳ", n=5))          # flip_rate thường 0.2–0.4
    always_first = lambda prompt: "1"                       # judge luôn chọn câu đứng trước
    print(pairwise_swapped(always_first, "tóm tắt hợp đồng", "bản ngắn", "bản dài"))

5.5Viết prompt LLM-as-judge

Trực giác. Prompt judge là "đề bài" cho một người chấm thuê ngoài chưa từng thấy sản phẩm của bạn. Người chấm đó chỉ giỏi khi đề bài hẹp, định nghĩa rõ, có bài mẫu đã chấm, và phải nộp bài theo mẫu cố định. Bốn thành phần sách nêu ra đúng là bốn thứ đó:

1. Một task, một tiêu chíKhông hỏi "email có tốt không?", mà hỏi một câu hẹp như "giọng văn có hợp với khách hạng sang không?".
2. Định nghĩa Pass/Fail chính xácPass = lỗi vắng mặt, Fail = lỗi xuất hiện — lấy thẳng từ mô tả failure mode ở error analysis, dùng đúng từ ngữ mà người gán nhãn đã dùng.
3. Few-shot examplesVài ca Pass rõ và Fail rõ, lấy từ trace đã gán nhãn — chỉ từ tập train. Dùng thang nhiều mức thì phải có ví dụ cho mọi mức.
4. Output có cấu trúcJSON gồm reasoning (1–2 câu) + answer (Pass/Fail): vừa giúp judge "nghĩ trước khi kết luận", vừa dễ đọc lại khi debug.

Hãy đánh giá email sau có tốt không, cho điểm 1-10.
Email: {{email}}

Bạn là người chấm chất lượng cho trợ lý CRM bất động sản.
Chỉ chấm MỘT tiêu chí: email gửi khách có nhắc lại ĐỦ các tiêu chí tìm nhà mà khách đã nêu không.

FAIL nếu email bỏ sót hoặc nói sai ít nhất một tiêu chí khách đã nêu
(khu vực, ngân sách, số phòng ngủ, thú cưng, hướng nhà...).
PASS nếu mọi tiêu chí khách đã nêu đều được nhắc đúng. Giọng văn, độ dài, lỗi chính tả
KHÔNG thuộc tiêu chí này — đừng để chúng ảnh hưởng tới kết luận.

[2–4 ví dụ từ tập train: tiêu chí của khách · email · JSON kết quả]

Trả về DUY NHẤT một JSON: {"reasoning": "<1-2 câu>", "answer": "Pass" hoặc "Fail"}

Tiêu chí của khách: {criteria}
Email: {email}

Code dựng prompt từ tập train và bóc kết quả an toàn (tự viết):

import json, re

JUDGE_PROMPT = """Bạn là người chấm chất lượng cho trợ lý CRM bất động sản.
Chỉ chấm MỘT tiêu chí: email gửi khách có nhắc lại ĐỦ các tiêu chí tìm nhà mà khách đã nêu không.

FAIL nếu email bỏ sót hoặc nói sai ít nhất một tiêu chí khách đã nêu
(khu vực, ngân sách, số phòng ngủ, thú cưng, hướng nhà...).
PASS nếu mọi tiêu chí khách đã nêu đều được nhắc đúng. Giọng văn, độ dài, lỗi chính tả
KHÔNG thuộc tiêu chí này — đừng để chúng ảnh hưởng tới kết luận.

{examples}
Trả về DUY NHẤT một JSON: {{"reasoning": "<1-2 câu>", "answer": "Pass" hoặc "Fail"}}

Tiêu chí của khách: {criteria}
Email: {email}"""

def build_prompt(train_examples, criteria, email):
    shots = "\n".join(
        f"Ví dụ {i+1}\nTiêu chí của khách: {ex['criteria']}\nEmail: {ex['email']}\n"
        f"Kết quả: {json.dumps({'reasoning': ex['why'], 'answer': ex['label']}, ensure_ascii=False)}\n"
        for i, ex in enumerate(train_examples))           # CHỈ lấy từ tập train
    return JUDGE_PROMPT.format(examples=shots, criteria=criteria, email=email)

def parse_verdict(text):
    """Bóc JSON đầu tiên trong output; lỗi định dạng -> None (đếm riêng, đừng lặng lẽ coi là Pass)."""
    m = re.search(r"\{.*\}", text, re.S)
    try:
        obj = json.loads(m[0]) if m else {}
    except json.JSONDecodeError:
        return None
    ans = str(obj.get("answer", "")).strip().capitalize()
    return ans if ans in ("Pass", "Fail") else None

if __name__ == "__main__":
    train = [{"criteria": "Q7, dưới 5 tỷ, 2PN", "email": "Gửi anh 3 căn 2PN ở Q7, giá 4,2–4,9 tỷ...",
              "why": "Nhắc đủ khu vực, ngân sách, số phòng.", "label": "Pass"}]
    print(build_prompt(train, "Thủ Đức, nuôi mèo, dưới 3 tỷ", "Em gửi chị vài căn ở Thủ Đức dưới 3 tỷ ạ."))
    print(parse_verdict('Đây là kết quả: {"reasoning": "thiếu tiêu chí mèo", "answer": "fail"}'))  # Fail
    print(parse_verdict("Tôi nghĩ là Pass"))                                                   # None

Bạn chấm câu trả lời của bot CSKH. Chỉ chấm MỘT tiêu chí:
bot có hứa hẹn quyền lợi (hoàn tiền, đổi trả, bồi thường, thời hạn) VƯỢT QUÁ đoạn chính sách được cung cấp không.
FAIL: có ít nhất một lời hứa không có căn cứ trong đoạn chính sách (kể cả hứa gián tiếp như "chắc chắn sẽ được").
PASS: mọi quyền lợi được nhắc đều có trong chính sách, hoặc bot không hứa gì.
Không chấm: giọng văn, độ dài, có giải quyết được vấn đề hay không.
[3 ví dụ từ train: 1 Pass, 1 Fail rõ, 1 Fail tinh tế "em nghĩ chắc được hoàn tiền ạ"]
Trả về JSON {"reasoning": "<trích lời hứa nếu có>", "answer": "Pass"|"Fail"}
Chính sách: {policy}
Câu hỏi khách: {question}
Câu trả lời của bot: {answer}

5.6Chia dữ liệu: train / dev / test

Trực giác. Xây judge ≈ train một classifier, chỉ khác là "huấn luyện" bằng cách viết prompt và chọn ví dụ thay vì gradient descent. Vì vậy nó cũng overfit được: nếu bạn sửa prompt cho tới khi judge đúng hết trên đúng những ca bạn đang nhìn, con số đó không nói gì về ca mới. Hình dung: train = sách giáo khoa (được chép vào prompt), dev = đề ôn tập (làm đi làm lại), test = đề thi thật (niêm phong, mở đúng một lần).

Code chia dữ liệu có kiểm tra rò rỉ + (tiện thể) ước lượng Pass@k dùng ở mục 5.11 (tự viết):

import random
from math import comb

def split_labeled(traces, n_train_per_class=6, seed=42):
    """traces: list[dict] có 'id' và 'label' ('Pass'/'Fail').
    Train: vài ca rõ ràng mỗi lớp. Phần còn lại chia đôi dev/test, giữ cân bằng lớp."""
    rng = random.Random(seed)
    out = {"train": [], "dev": [], "test": []}
    for label in ("Pass", "Fail"):
        group = [t for t in traces if t["label"] == label]
        rng.shuffle(group)
        clear = [t for t in group if t.get("clear", True)]      # ưu tiên ca rõ cho few-shot
        train = clear[:n_train_per_class]
        rest = [t for t in group if t not in train]
        half = len(rest) // 2
        out["train"] += train
        out["dev"] += rest[:half]
        out["test"] += rest[half:]
    ids = [set(t["id"] for t in v) for v in out.values()]
    assert not (ids[0] & ids[1] or ids[0] & ids[2] or ids[1] & ids[2]), "rò rỉ giữa các tập!"
    return out

def pass_at_k(n, c, k):
    """Ước lượng không chệch Pass@k khi sinh n mẫu, c mẫu đúng (n >= k)."""
    if n - c < k:
        return 1.0
    return 1 - comb(n - c, k) / comb(n, k)

if __name__ == "__main__":
    data = [{"id": i, "label": "Pass" if i % 2 else "Fail"} for i in range(96)]
    s = split_labeled(data)
    print({k: len(v) for k, v in s.items()})             # {'train': 12, 'dev': 42, 'test': 42}
    print(round(pass_at_k(n=10, c=3, k=1), 3), round(pass_at_k(10, 3, 5), 3))   # 0.3 0.917

5.7Vòng lặp tinh chỉnh judge & TPR/TNR

Trực giác. Bạn không viết được prompt judge hoàn hảo ngay lần đầu, cũng như không viết được code không bug ngay lần đầu. Quy trình là một vòng lặp debug: chạy judge trên dev, đọc các ca judge nói khác người, sửa định nghĩa/ví dụ, chạy lại.

TPR và TNR đọc từ confusion matrix

Quy ước của sách: positive = Pass. Vậy TPR = p/P: trong các trace người gán Pass, judge cũng nói Pass bao nhiêu phần. TNR = f/F: trong các trace người gán Fail, judge bắt được bao nhiêu phần. Ví dụ tự dựng: test set có 50 trace Pass và 50 trace Fail theo nhãn người.

Code in TPR/TNR trên dev kèm danh sách ca sai để đọc (tự viết):

def review_dev(rows):
    """rows: list[dict] có 'id', 'human' ('Pass'/'Fail'), 'judge', 'reasoning'.
    In TPR/TNR trên dev + liệt kê ca sai để đọc tay (bước 4 của vòng lặp)."""
    P = [r for r in rows if r["human"] == "Pass"]
    F = [r for r in rows if r["human"] == "Fail"]
    tpr = sum(r["judge"] == "Pass" for r in P) / len(P)
    tnr = sum(r["judge"] == "Fail" for r in F) / len(F)
    print(f"dev: TPR={tpr:.2f} ({len(P)} Pass)  TNR={tnr:.2f} ({len(F)} Fail)")
    for kind, group, wrong in (("FALSE PASS (lọt lỗi)", F, "Pass"),
                               ("FALSE FAIL (đánh trượt oan)", P, "Fail")):
        bad = [r for r in group if r["judge"] == wrong]
        print(f"\n== {kind}: {len(bad)} ca")
        for r in bad:
            print(f"  #{r['id']}: {r['reasoning'][:90]}")
    return tpr, tnr

if __name__ == "__main__":
    rows = [
        {"id": 1, "human": "Pass", "judge": "Pass", "reasoning": "Nhắc đủ Q7, 5 tỷ, 2PN."},
        {"id": 2, "human": "Fail", "judge": "Pass", "reasoning": "Email nói 'ngân sách phù hợp' nên coi như đủ."},
        {"id": 3, "human": "Fail", "judge": "Fail", "reasoning": "Không nhắc 'nuôi mèo'."},
        {"id": 4, "human": "Pass", "judge": "Fail", "reasoning": "Không ghi hướng nhà (nhưng khách không hề nêu hướng)."},
    ]
    review_dev(rows)

5.8Hiệu chỉnh tỉ lệ thành công: công thức Rogan–Gladen

Đã chốt judge, đã có TPR/TNR từ test. Giờ chạy judge trên m trace production chưa gán nhãn, được k cái "Pass" → p_obs = k/m. Nhưng p_obs không phải tỉ lệ thật θ, vì judge vừa đánh trượt oan vừa cho lọt lỗi.

Trực giác. Giống cân nhà bếp luôn lệch: nếu biết cân "ăn bớt" bao nhiêu thì đọc số rồi cộng bù là ra cân nặng thật. TPR/TNR chính là "độ lệch của cân", đo một lần trên test set bằng các quả cân chuẩn (nhãn người).

Muốn tự tin rằng hiệu chỉnh "không chệch"? Đừng tin lời — mô phỏng. Đoạn code dưới lặp lại toàn bộ quy trình 5.000 lần với một judge khó tính (TPR 0,70, TNR 0,95) trên app có θ thật = 0,80:

import numpy as np

def simulate(theta=0.80, tpr=0.70, tnr=0.95, n=50, m=2000, runs=5000, seed=7):
    """Lặp lại cả quy trình nhiều lần: gán test mới, chạy judge trên production mới.
    So trung bình của số thô và số đã hiệu chỉnh với θ thật."""
    rng = np.random.default_rng(seed)
    p_true = tpr * theta + (1 - tnr) * (1 - theta)        # xác suất judge nói Pass
    raw, corr = [], []
    for _ in range(runs):
        tpr_hat = rng.binomial(n, tpr) / n                  # đo trên test n Pass
        tnr_hat = rng.binomial(n, tnr) / n                  # và n Fail
        p_obs = rng.binomial(m, p_true) / m
        raw.append(p_obs)
        d = tpr_hat + tnr_hat - 1
        if d >= 0.05:
            corr.append(np.clip((p_obs + tnr_hat - 1) / d, 0, 1))
    return np.mean(raw), np.mean(corr), np.percentile(corr, [2.5, 97.5])

if __name__ == "__main__":
    raw, corr, (lo, hi) = simulate()
    print(f"θ thật = 0.800 | trung bình số thô = {raw:.3f} | trung bình θ̂ = {corr:.3f}")
    print(f"95% giữa của θ̂ qua các lần lặp: [{lo:.3f}, {hi:.3f}]")

Kết quả khi chạy: số thô trung bình ≈ 0,570 (lệch 23 điểm), θ̂ trung bình ≈ 0,806. Phần lệch nhỏ còn lại (+0,006) đến từ việc kẹp ở 1 và việc chia cho TPR ước lượng — gần như không đáng kể so với độ rộng dao động [0,66; 1,00]. Bài học giống thí nghiệm tư duy của sách: hiệu chỉnh xử lý độ lệch, còn lỗi judge hiện ra dưới dạng độ bất định.

Hàm hiệu chỉnh dùng ở các mục sau:

import numpy as np

def tpr_tnr(y, yhat):
    """y, yhat: 1 = Pass, 0 = Fail. Trả về (TPR, TNR)."""
    y, yhat = np.asarray(y), np.asarray(yhat)
    return yhat[y == 1].mean(), 1 - yhat[y == 0].mean()

def rogan_gladen(p_obs, tpr, tnr, min_denom=0.05):
    d = tpr + tnr - 1
    if d < min_denom:
        return np.nan                      # judge ~ tung đồng xu: không hiệu chỉnh
    return float(np.clip((p_obs + tnr - 1) / d, 0, 1))

Ghi chú: bảng tương ứng trong bản Early Release mô tả hướng dịch chuyển của ước lượng bằng lời theo chiều ngược với bảng trên (có vẻ nhầm chiều positive); hướng ở đây được suy trực tiếp từ công thức và khớp với ví dụ 5.8b.

5.9Bootstrap CI & quyết định ship

Trực giác. TPR/TNR được đo trên một test set cụ thể. Nếu tình cờ rút được test set khác, TPR/TNR sẽ hơi khác, và θ̂ cũng khác theo. Bootstrap hỏi: "nếu test set của tôi chỉ là một lần rút ngẫu nhiên, thì θ̂ có thể dao động bao nhiêu?" — và trả lời bằng cách giả lập việc rút lại từ chính dữ liệu đang có.

Code đầy đủ (tự viết; ý tưởng giống thư viện nhỏ judgy mà tác giả mở mã nguồn, chỉ phụ thuộc numpy). Tham số prod_noise là phần mở rộng của người viết: rút lại cả phía production để tính thêm độ bất định do m hữu hạn.

import numpy as np

def tpr_tnr(y, yhat):
    """y, yhat: 1 = Pass, 0 = Fail. Trả về (TPR, TNR)."""
    y, yhat = np.asarray(y), np.asarray(yhat)
    return yhat[y == 1].mean(), 1 - yhat[y == 0].mean()

def rogan_gladen(p_obs, tpr, tnr, min_denom=0.05):
    d = tpr + tnr - 1
    if d < min_denom:
        return np.nan                      # judge ~ tung đồng xu: không hiệu chỉnh
    return float(np.clip((p_obs + tnr - 1) / d, 0, 1))

def corrected_rate(y, yhat, k, m, B=10_000, seed=0, prod_noise=False):
    """Ước lượng tỉ lệ Pass thật + CI 95% bootstrap.
    y, yhat: nhãn người & judge trên TEST. k/m: judge nói Pass k trên m trace production.
    prod_noise=True: resample thêm phía production (mở rộng, không có trong sách)."""
    rng = np.random.default_rng(seed)
    y, yhat = np.asarray(y), np.asarray(yhat)
    p_obs = k / m
    theta_hat = rogan_gladen(p_obs, *tpr_tnr(y, yhat))
    thetas = []
    for _ in range(B):
        idx = rng.integers(0, len(y), len(y))          # resample CẶP (nhãn, dự đoán)
        yy, pp = y[idx], yhat[idx]
        if yy.all() or not yy.any():
            continue                                   # mẫu thiếu hẳn một lớp
        po = rng.binomial(m, p_obs) / m if prod_noise else p_obs
        t = rogan_gladen(po, *tpr_tnr(yy, pp))
        if not np.isnan(t):
            thetas.append(t)
    lo, hi = np.percentile(thetas, [2.5, 97.5])
    return theta_hat, lo, hi

def make_labels(P, tp, F, tn):
    """Dựng lại vector nhãn từ confusion matrix (tiện cho ví dụ)."""
    y = [1] * P + [0] * F
    yhat = [1] * tp + [0] * (P - tp) + [0] * tn + [1] * (F - tn)
    return y, yhat

if __name__ == "__main__":
    y, yhat = make_labels(P=50, tp=45, F=50, tn=40)
    print("TPR=%.2f TNR=%.2f" % tpr_tnr(y, yhat))                 # TPR=0.90 TNR=0.80
    print("θ̂=%.3f  CI95=[%.3f, %.3f]" % corrected_rate(y, yhat, k=750, m=1000))
    # θ̂=0.786  CI95≈[0.69, 0.90]   (số chính xác tuỳ seed)

Đọc CI để quyết định ship

5.10Case study đầu-cuối: judge "hứa quyền lợi vượt chính sách"

[v1]
Bạn chấm câu trả lời của bot CSKH. Tiêu chí duy nhất: bot có cam kết quyền lợi
không có trong chính sách dưới đây không?
FAIL: có cam kết ngoài chính sách. PASS: không có.
[2 ví dụ từ train]
Trả về JSON {"reasoning": "...", "answer": "Pass"|"Fail"}
Chính sách: {policy}   Hội thoại: {conversation}

[v2 — thêm vào v1]
"Lời hứa" gồm cả khẳng định mềm hoặc gián tiếp khiến khách tin rằng mình SẼ nhận
được quyền lợi (ví dụ: "chắc chắn được", "em sẽ xin cho anh", "yên tâm là được hoàn").
[+2 ví dụ Fail lời hứa mềm, +1 ví dụ Pass từ chối lịch sự]

[v3 — thêm vào v2]
Nhắc lại đúng một điều kiện có trong chính sách (kèm điều kiện) là PASS.
Cam kết về THỜI GIAN hoặc SỐ TIỀN khác với chính sách cũng là lời hứa vượt chính sách.
Trong reasoning, trích nguyên văn lời hứa và dòng chính sách liên quan (nếu có).
[+1 ví dụ Pass trích điều kiện, +1 ví dụ Fail hứa thời gian]

5.11(Tuỳ chọn) Metric cho nhóm output

Khi agent sinh nhiều ứng viên cho một input (nhiều completion, nhiều tài liệu truy xuất, nhiều phương án), sách gợi ý một nhóm metric riêng:

5.12Vận hành: judge cũng cần bảo trì

Judge không phải "xây một lần dùng mãi". Dữ liệu người dùng trôi, model của app được nâng cấp, chính model judge được nhà cung cấp cập nhật, và tiêu chí của sản phẩm thay đổi. Sách khuyên căn chỉnh lại định kỳ (ví dụ hàng tuần): gán thêm vài trace, tính lại TPR/TNR và CI; lỗi mới phát hiện thì bổ sung vào cả ba tập. Judge còn là một hệ ngẫu nhiên: chạy 3–5 lần trên cùng trace — lý lẽ đổi mà nhãn giữ thì ổn; nhãn lật là dấu hiệu tiêu chí còn mơ hồ ở loại ca đó.

Hàm đo độ ổn định (trích từ judge_probes.py ở mục 5.4) — chạy nó hàng tuần trên ~20 ca, theo dõi flip_rate trung bình:

def stability(call_llm, prompt, n=5):
    """Chạy judge n lần (temperature > 0). flip_rate > 0 = tiêu chí còn mơ hồ với ca này."""
    votes = Counter(parse_verdict(call_llm(prompt)) for _ in range(n))
    top, cnt = votes.most_common(1)[0]
    return {"votes": dict(votes), "majority": top, "flip_rate": 1 - cnt / n}

5.13Cạm bẫy thường gặp

• Prompt judge không có ví dụ. Lỗi phổ biến nhất: judge thiếu neo về thế nào là Fail trong bối cảnh của bạn → chấm lỏng lẻo.
• Một judge ôm nhiều tiêu chí. Gộp giọng văn + nội dung + bước tiếp theo vào một Pass/Fail → mơ hồ, khó chẩn đoán. Tách thành nhiều judge hẹp.
• Bỏ qua căn chỉnh. Judge "chạy luôn" với tiêu chí phổ thông (cảm xúc tích cực/tiêu cực), nhưng tiêu chí mang "vibe" riêng của sản phẩm phải được dạy.
• Rò rỉ dữ liệu. Trace dùng làm few-shot lại nằm trong tập đo, hoặc sửa prompt theo kết quả test → TPR/TNR đẹp giả.
• Báo accuracy thay vì TPR + TNR. Che mất judge đang lệch về phía nào.
• Báo số thô p_obs như tỉ lệ thành công. Số thô bị méo theo lỗi của judge; hãy báo θ̂ đã hiệu chỉnh kèm CI.
• So hai hệ thống bằng điểm ước lượng. 0,86 vs 0,88 với CI rộng ±0,08 là "chưa phân biệt được", không phải "B tốt hơn".
• Căn chỉnh một lần rồi thôi. Dữ liệu drift, model update, tiêu chí đổi. Gán thêm vài trace và tính lại TPR/TNR, CI định kỳ; lỗi mới → bổ sung vào cả ba tập.
• Quên judge cũng ngẫu nhiên. Chạy 3–5 lần cùng trace: lý lẽ đổi mà nhãn giữ → ổn; nhãn lật → tiêu chí còn mơ hồ.

5.14Áp dụng cho dự án model-routing

Bối cảnh: bạn xây một router gửi mỗi request LLM tới model phù hợp — model nhỏ/rẻ cho request dễ, model lớn/đắt cho request khó — để cân bằng chất lượng, chi phí và latency. Câu hỏi eval thật sự không phải "model X giỏi không?" mà là: "chính sách routing P có giữ chất lượng chấp nhận được trong khi rẻ/nhanh hơn không — và tôi chắc chắn tới mức nào?" Mọi công cụ của chương này đều dùng được, với vài điều chỉnh riêng.

A · Đo cái gì, bằng gì

B · Judge "chấp nhận được" — tuyệt đối hay tham chiếu?

Prompt tự viết cho judge có tham chiếu (tiêu chí hẹp, chặn verbosity và "phong cách khác = sai"):

Bạn đánh giá câu trả lời ỨNG VIÊN cho một yêu cầu của người dùng.
Tiêu chí DUY NHẤT: ứng viên có CHẤP NHẬN ĐƯỢC để gửi cho người dùng không.
PASS: giải quyết đúng yêu cầu chính; không có lỗi sự thật/logic làm hỏng kết quả;
      nếu yêu cầu có ràng buộc (định dạng, độ dài, ngôn ngữ) thì tuân thủ.
FAIL: sai hoặc thiếu phần chính của yêu cầu; có lỗi làm người dùng hành động sai;
      từ chối một yêu cầu hợp lệ.
Câu trả lời THAM CHIẾU chỉ để giúp bạn hiểu yêu cầu — ứng viên KHÔNG cần giống tham chiếu.
Ngắn hơn, khác cách diễn đạt, khác cách giải nhưng đúng → vẫn PASS.
Nếu tham chiếu sai mà ứng viên đúng → PASS.
[3–4 ví dụ từ train: 1 Pass ngắn gọn, 1 Fail dài-mà-sai, 1 Fail từ chối oan, 1 Pass khác cách giải]
Trả về JSON {"reasoning": "<1-2 câu, chỉ ra lỗi cụ thể nếu Fail>", "answer": "Pass"|"Fail"}
Yêu cầu: {request}
Tham chiếu: {reference}
Ứng viên: {candidate}

C · Hiệu chỉnh theo từng model (stratified correction)

Chương gốc giả định một cặp TPR/TNR cho toàn bộ dữ liệu. Với routing, output của model nhỏ và model lớn khác nhau về phong cách: model lớn thường dài, trau chuốt → judge dễ bị verbosity bias → TNR trên output model lớn có thể thấp hơn. Vì vậy: test set riêng cho từng model (mỗi model ~50 Pass + 50 Fail), hiệu chỉnh θ̂ riêng cho từng model, rồi gộp theo tỉ lệ traffic: θ̂_policy = Σ w_model · θ̂_model.

Code so sánh hai policy (tự viết, dùng lại judge_stats.py ở mục 5.9):

import numpy as np
from judge_stats import rogan_gladen, make_labels

def rates(y, yhat):
    return yhat[y == 1].mean(), 1 - yhat[y == 0].mean()

def policy_theta(policy, judge_rates, p_obs_override=None):
    """policy: {model: (k, m)}. Trọng số = tỉ lệ traffic thật của từng model."""
    total = sum(m for _, m in policy.values())
    theta = 0.0
    for model, (k, m) in policy.items():
        p_obs = (p_obs_override or {}).get(model, k / m)
        t = rogan_gladen(p_obs, *judge_rates[model])
        if np.isnan(t):
            return np.nan
        theta += (m / total) * t
    return theta

def compare_policies(test, pol_a, pol_b, B=5000, seed=1):
    """test: {model: (y, yhat)} — judge được hiệu chỉnh RIÊNG cho output của từng model.
    Trả về θ_A, θ_B, Δ = θ_B − θ_A và CI 95% của Δ (paired bootstrap)."""
    rng = np.random.default_rng(seed)
    test = {mdl: (np.asarray(y), np.asarray(p)) for mdl, (y, p) in test.items()}
    base = {mdl: rates(y, p) for mdl, (y, p) in test.items()}
    th_a, th_b = policy_theta(pol_a, base), policy_theta(pol_b, base)
    deltas = []
    for _ in range(B):
        jr = {}
        for mdl, (y, p) in test.items():               # resample test MỘT lần, dùng chung cho A và B
            idx = rng.integers(0, len(y), len(y))
            if y[idx].all() or not y[idx].any():
                break
            jr[mdl] = rates(y[idx], p[idx])
        else:
            noise = lambda pol: {mdl: rng.binomial(m, k / m) / m for mdl, (k, m) in pol.items()}
            a = policy_theta(pol_a, jr, noise(pol_a))
            b = policy_theta(pol_b, jr, noise(pol_b))
            if not (np.isnan(a) or np.isnan(b)):
                deltas.append(b - a)
    lo, hi = np.percentile(deltas, [2.5, 97.5])
    return th_a, th_b, th_b - th_a, (lo, hi)

if __name__ == "__main__":
    test = {"small": make_labels(50, 46, 50, 44),     # TPR .92, TNR .88
            "large": make_labels(50, 48, 50, 39)}     # TPR .96, TNR .78 (verbosity!)
    policy_a = {"small": (3520, 4000), "large": (5449, 6000)}   # 40% / 60% traffic
    policy_b = {"small": (5824, 7000), "large": (2702, 3000)}   # 70% / 30% traffic
    th_a, th_b, d, (lo, hi) = compare_policies(test, policy_a, policy_b)
    print(f"θ_A={th_a:.3f}  θ_B={th_b:.3f}  Δ={d:+.3f}  CI95=[{lo:+.3f}, {hi:+.3f}]")
    margin = -0.03                                     # chấp nhận tụt tối đa 3 điểm
    print("non-inferior" if lo > margin else "CHƯA chứng minh được không-tệ-hơn")

D · Bẫy riêng của bài toán routing

• Selection bias của chính router. Model S chỉ thấy request "dễ" mà router gửi tới. θ̂_S dưới policy A (0,95) không cho biết S làm tốt thế nào nếu nhận thêm request (0,89 dưới B). Muốn ước lượng một policy mới trước khi triển khai: giữ một lát ngẫu nhiên (ví dụ 5% request route ngẫu nhiên), hoặc shadow — gửi song song một mẫu request tới cả hai model và chấm cả hai.
• Judge nằm trong pool. Nếu judge cùng họ với L, self-enhancement làm θ̂_L đẹp giả. Chọn judge ngoài pool, hoặc đo TPR/TNR riêng cho từng model để phát hiện chênh lệch.
• Một TPR/TNR chung cho mọi model. Bấm nút "Giả sử judge như nhau" trong LAB 5 để thấy θ̂ thay đổi thế nào khi bỏ qua việc judge dễ dãi hơn với L.
• Thêm model mới vào pool mà không đo lại judge. Output của model mới có thể rơi vào vùng judge chưa được căn chỉnh — cần test set cho model đó.
• Tối ưu một chiều. Báo cáo luôn đi thành bộ ba: θ̂ + CI (chất lượng), chi phí/1k, latency p95. Một policy "rẻ hơn 45%" chỉ có ý nghĩa khi đặt cạnh CI của mức tụt chất lượng.

E · Chính router đúng sai thế nào: route thừa & route thiếu

Ngoài "policy tốt tới đâu", bạn còn muốn biết router sai ở đâu để sửa bộ phân loại độ khó. Dùng dữ liệu shadow (cả S và L cùng trả lời một mẫu request, judge chấm cả hai), mỗi request rơi vào một trong bốn ô — giống confusion matrix, nhưng lần này người bị chấm là router:

5.15Quy trình tổng kết & nối với các chương khác

1. Mỗi failure mode → một metric tự độngChỉ cho lỗi Generalization còn sót sau khi đã sửa prompt (taxonomy từ Chương 3).
2. Tiêu chí chính xác → codeSo AST của SQL, regex trường bắt buộc, chạy thử tool call. Nhớ kiểm cả code check trên vài chục nhãn.
3. Tiêu chí cần phán đoán → LLM-as-judgeGiọng văn, độ đầy đủ, mức liên quan — nhị phân, một tiêu chí mỗi judge, 4 thành phần prompt.
4. Kiểm định judgeNhãn gold (Chương 4) → train/dev/test; lặp trên dev; TPR/TNR trên test đúng một lần.
5. Báo cáo θ̂ + CIRogan–Gladen để hiệu chỉnh, bootstrap để lượng hoá độ bất định; quyết định bằng cận dưới.
6. Bảo trì liên tụcCăn chỉnh lại định kỳ; đưa evaluator vào CI/CD (Chương 9) và giao diện review (Chương 10).

Bảng tra nhanh

Thuật ngữ Anh–Việt của chương

Tự kiểm tra

Tóm tắt diễn giải bằng tiếng Việt cho mục đích học tập cá nhân, không phải bản dịch. Mọi ví dụ số, case study, khung prompt, đoạn code và lab là tự dựng để minh hoạ. Nguồn: Evals for AI Engineers — Shreya Shankar & Hamel Husain, O'Reilly (Early Release).