Đánh giá hội thoại nhiều lượt: Session → Turn → Coherence

Bạn muốn biết bot đặt vé của mình có bị hiện tượng trên không. Lấy 50 yêu cầu đặt vé có đáp án chuẩn (chuyến bay đúng), mỗi yêu cầu gồm 4 mẩu thông tin: điểm đến, ngày, số khách, ràng buộc giờ bay.

1. Bản single-turn: gộp 4 mẩu vào một câu: "Đi Đà Nẵng 25/10, 3 người, không bay sau 20h". Chạy mỗi yêu cầu 10 lần.
2. Bản multi-turn (sharded): chia 4 mẩu thành 4 tin nhắn user, xen giữa là câu trả lời của bot. Cũng chạy 10 lần mỗi yêu cầu.
3. Đo: với mỗi yêu cầu có 10 điểm số. Tính trung bình, phân vị 90 (lúc bot làm tốt nhất — xấp xỉ "năng lực") và khoảng P90−P10 (độ dao động — xấp xỉ "thiếu ổn định"). Cách tách này là của paper gốc, không phải của sách.
4. Đọc kết quả: trung bình tụt từ ~88 xuống ~60, nhưng P90 chỉ tụt từ ~94 xuống ~88 trong khi độ dao động tăng từ ~12 lên ~60. Bot vẫn biết cách đặt đúng — nó chỉ "lúc được lúc không" tuỳ vào việc có chốt sớm hay không.
5. Hệ quả hành động: đổi sang model to hơn chưa chắc giải quyết; cần sửa hành vi chốt sớm (hỏi đủ thông tin trước khi đề xuất, hoặc rà lại ràng buộc trước mỗi đề xuất).

# Tách "năng lực" (aptitude) và "độ bất ổn" (unreliability) — số liệu minh hoạ tự sinh.
# Ý tưởng (theo paper gốc Laban et al.): chạy CÙNG một yêu cầu nhiều lần;
# aptitude ~ điểm ở phân vị 90 (lúc model làm tốt nhất),
# unreliability ~ khoảng cách P90 - P10 (độ dao động giữa lần tốt và lần tệ).
import random, statistics

random.seed(6)

def percentile(xs, p):
    xs = sorted(xs)
    k = (len(xs) - 1) * p / 100
    lo, hi = int(k), min(int(k) + 1, len(xs) - 1)
    return xs[lo] + (xs[hi] - xs[lo]) * (k - lo)

def simulate_runs(mode, n=10):
    """Điểm 0-100 của 10 lần chạy cho một yêu cầu (giả lập)."""
    runs = []
    for _ in range(n):
        if mode == "single":
            runs.append(min(100, random.gauss(88, 6)))
        else:  # multi-turn: đôi khi model "chốt sớm" và sai hẳn
            locked_early = random.random() < 0.4
            runs.append(random.gauss(30, 10) if locked_early else min(100, random.gauss(84, 7)))
    return runs

for mode in ("single", "multi"):
    A, U, M = [], [], []
    for task in range(50):                 # 50 yêu cầu khác nhau
        runs = simulate_runs(mode)
        A.append(percentile(runs, 90))
        U.append(percentile(runs, 90) - percentile(runs, 10))
        M.append(statistics.mean(runs))
    print(f"{mode:6s}  mean={statistics.mean(M):5.1f}  "
          f"aptitude(P90)={statistics.mean(A):5.1f}  unreliability(P90-P10)={statistics.mean(U):5.1f}")
# Kết quả: mean tụt mạnh, aptitude chỉ tụt nhẹ, unreliability tăng vọt
# -> vấn đề là "lúc được lúc không", không phải "model kém đi".

1. Cấp session: mục tiêu (giữ chỗ đúng chuyến, đúng số khách, không bay tối) có đạt? Có → PASS. Dừng ở đây nếu chỉ cần đo tỉ lệ thành công.
2. Nhưng rubric của bạn có thể có điều kiện "user không phải tự sửa lỗi của bot". Nếu có, session này thành FAIL — đây là lý do phải viết rubric trước khi chấm (xem 6.5).
3. Cấp turn (chỉ khi FAIL): đi lần lượt. B1 ✓. B2 ✗ — bỏ qua thông tin "thêm bé 4 tuổi" vừa nêu ở U2. Đây là lượt hỏng đầu tiên.
4. Cấp coherence: lỗi ở B2 thuộc loại "quên thông tin user vừa bổ sung". Thông tin nằm ngay lượt trước nên không phải truncation; B1 đã đưa đề xuất trước khi có đủ thông tin → nghiêng về anchoring (đề xuất lượt 1 được giữ nguyên số khách).
5. Hành động: ghi nhãn "anchoring — số khách" vào sổ lỗi (open coding kiểu Chương 3), chưa vội viết evaluator.

Ba trace của bot đặt vé, cả ba đều có triệu chứng "đề xuất vi phạm điều user đã nói". Metadata log: lượt nêu ràng buộc, lượt vi phạm, các lượt thực sự có trong prompt, lượt bot đề xuất lần đầu.

1. Trace A: ràng buộc ở lượt 2, vi phạm ở lượt 11; app dùng cửa sổ trượt chỉ giữ 6 lượt gần nhất → prompt ở lượt 11 chứa lượt 6–11. Ràng buộc không có trong prompt → truncation. Sửa prompt vô ích — model không thấy thì không thể tuân theo.
2. Trace B: bot đề xuất ngay lượt 1 (khi mới biết điểm đến), ràng buộc được nêu ở lượt 4, lượt 5 bot vẫn đề xuất y như cũ → anchoring.
3. Trace C: hội thoại 15 lượt, prompt đầy đủ, nhưng từ lượt 14 bot không còn trả giá theo định dạng "tổng tiền + phí" mà system prompt yêu cầu → instruction drift.
4. Kiểm chứng: với A, thử tăng cửa sổ lên 20 lượt và chạy lại prefix → lỗi biến mất thì chẩn đoán đúng. Với B, thêm chỉ dẫn "rà lại ràng buộc" → đo lại. Với C, chèn nhắc định dạng mỗi 5 lượt.

Có thể tự động hoá bước phân loại sơ bộ bằng vài quy tắc trên metadata — chỉ để xếp hàng cho người đọc, không thay việc đọc trace:

# Phân loại sơ bộ nguyên nhân gốc của một lỗi coherence (heuristic, tự viết).
# Input: metadata của trace mà hầu hết hệ thống log sẵn được.
from dataclasses import dataclass

@dataclass
class CoherenceFailure:
    constraint_turn: int        # lượt user nêu ràng buộc (vd. "không bay đêm")
    fail_turn: int              # lượt agent vi phạm
    turns_in_context: list      # các lượt THỰC SỰ nằm trong prompt ở lượt fail
    first_answer_turn: int      # lượt agent lần đầu đưa ra đáp án/đề xuất
    answer_changed_after_new_info: bool  # đáp án có đổi khi user bổ sung?
    system_rules_broken: bool   # vi phạm quy tắc trong system prompt?

def diagnose(f: CoherenceFailure) -> str:
    # 1) Model có NHÌN THẤY ràng buộc không? Không thấy -> truncation.
    if f.constraint_turn not in f.turns_in_context:
        return "context truncation -> sửa kiến trúc (tóm tắt / memory / pin ràng buộc)"
    # 2) Đoán sớm TRƯỚC khi user nói đủ, rồi không đổi -> anchoring.
    if f.first_answer_turn < f.constraint_turn and not f.answer_changed_after_new_info:
        return "anchoring -> prompt: rà lại mọi ràng buộc trước khi trả lời"
    # 3) Luật hệ thống vẫn trong context nhưng bị lơi ở lượt muộn -> drift.
    if f.system_rules_broken and f.fail_turn >= 8:
        return "instruction drift -> nhắc lại chỉ dẫn then chốt định kỳ"
    return "chưa rõ -> đọc trace thủ công, có thể là lỗi single-turn trá hình"

cases = {
    "A: cửa sổ trượt 6 lượt": CoherenceFailure(2, 11, list(range(6, 12)), 3, False, False),
    "B: đề xuất ở lượt 1":    CoherenceFailure(4, 5, list(range(0, 6)), 1, False, False),
    "C: lượt 14 quên format": CoherenceFailure(0, 14, list(range(0, 15)), 2, True, True),
}
for name, f in cases.items():
    print(f"{name:26s} => {diagnose(f)}")

Đội bạn cài memory store đơn giản: mỗi lần user nói fact mới thì thêm một dòng. Chạy 4 test theo 4 khía cạnh:

1. Storage: sau session 1, store có "ăn chay" → PASS.
2. Retrieval: khi hỏi về chuyến bay, chỉ lấy fact chủ đề flight (ghế cửa sổ), không kéo fact ăn uống → PASS.
3. Update: session 3 user nói "ăn cả cá"; khi truy xuất chủ đề ăn uống, store trả về cả hai "ăn chay" và "ăn cả cá" → FAIL. Agent nhận hai fact mâu thuẫn và có thể chọn cái cũ.
4. Staleness: 14 tháng sau, fact ghế ngồi quá hạn 365 ngày không còn được áp dụng → PASS.
5. Kết luận: bug nằm ở tầng lưu trữ (append), không phải ở prompt. Nếu chỉ chấm cuộc hội thoại cuối, bạn sẽ thấy "bot gợi ý sai món" mà không biết vì sao — test theo khía cạnh chỉ thẳng vào tầng lỗi.

# Bộ test memory xuyên session: Storage / Retrieval / Update / Staleness.
# MemoryStore là bản "có bug" cố ý: update bằng cách APPEND thay vì ghi đè.
import datetime as dt

class MemoryStore:
    def __init__(self):
        self.facts = []                      # (key, value, topic, saved_at)
    def save(self, key, value, topic, when):
        self.facts.append((key, value, topic, when))   # BUG: không xoá bản cũ
    def retrieve(self, topic, now, max_age_days=365):
        return [(k, v) for k, v, t, w in self.facts
                if t == topic and (now - w).days <= max_age_days]

def run_suite(store):
    d0 = dt.date(2026, 1, 5)
    results = {}
    # Session 1: user nói sự thật mới
    store.save("diet", "ăn chay", "food", d0)
    store.save("seat", "ghế cửa sổ", "flight", d0)
    results["storage"] = any(v == "ăn chay" for _, v, *_ in store.facts)
    # Session 2 (hỏi về chuyến bay): chỉ lấy fact liên quan
    got = store.retrieve("flight", d0 + dt.timedelta(days=3))
    results["retrieval"] = got == [("seat", "ghế cửa sổ")]
    # Session 3: user phủ định fact cũ
    store.save("diet", "ăn cả cá", "food", d0 + dt.timedelta(days=40))
    diet = [v for k, v in store.retrieve("food", d0 + dt.timedelta(days=41)) if k == "diet"]
    results["update"] = diet == ["ăn cả cá"]            # mong đợi đúng 1 giá trị mới
    # Session 4 (14 tháng sau): fact ghế ngồi đã quá hạn, không được áp dụng
    old = store.retrieve("flight", d0 + dt.timedelta(days=430))
    results["staleness"] = old == []
    return results

for aspect, ok in run_suite(MemoryStore()).items():
    print(f"{aspect:10s} {'PASS' if ok else 'FAIL'}")
# -> update FAIL: store trả về cả "ăn chay" lẫn "ăn cả cá" -> agent có thể gợi ý sai.

1. Người: 6 thành viên (PM, 2 dev, CS lead, designer, QA). Mỗi người 12 tác vụ → 72 session trong ~3 giờ.
2. Ma trận ý định × persona (mỗi người bốc thăm): ý định = {tra cứu hành lý, vé 1 chiều, khứ hồi + khách sạn, đổi ngày bay, hoàn vé, gia đình có trẻ nhỏ}; persona = {cộc lốc, dài dòng, hay đổi ý, bực bội, không rành công nghệ, gõ không dấu}.
3. Luật chơi: ít nhất 1/3 tác vụ phải kéo dài ≥ 10 lượt (bổ sung thông tin nhỏ giọt, đổi ý giữa chừng) — vì lỗi hay nằm ở hội thoại dài.
4. Ghi chú ngay sau mỗi session: mục tiêu là gì, có đạt không, cảm giác chỗ nào "sai sai". Ghi chú này là nguyên liệu open coding (Chương 3).
5. Buổi đọc chung ngày hôm sau: mỗi người đọc trace của người khác, gắn nhãn Pass/Fail cấp session trước khi thảo luận — để lộ ra chỗ rubric còn mơ hồ.

Rubric v1: "Session PASS nếu user đặt được vé mong muốn."

1. Hai người chấm cùng 40 session; bất đồng ở 9 session (đồng thuận 78%).
2. Đọc 9 ca bất đồng: (a) user tự bỏ đi vì hết chuyến phù hợp và bot nói rõ điều đó — người A chấm Fail, người B chấm Pass; (b) bot đặt đúng nhưng user phải sửa bot 3 lần; (c) bot đặt đúng nhưng báo giá sai ở giữa rồi tự sửa.
3. Rubric v2 (bổ sung): PASS nếu tất cả: ① kết thúc bằng giữ chỗ khớp mọi ràng buộc cứng user nêu (ngày, số khách, ngân sách, giờ bay), hoặc bot kết luận đúng rằng không có lựa chọn thỏa mãn và đề xuất phương án thay thế; ② không có thông tin sai về giá/chính sách chưa được đính chính trước khi user xác nhận; ③ user không phải nhắc lại cùng một thông tin quá 1 lần.
4. Chấm lại 40 session với v2: bất đồng còn 2 → đồng thuận 95%. Hai ca còn lại được ghi vào phần "ví dụ biên" của rubric.

1. Gộp thành 1 prompt: "Vé phổ thông tiết kiệm chặng nội địa được ký gửi bao nhiêu kg?" (đưa kèm đúng tài liệu mà hệ thống retrieval lấy về).
2. Chạy 5 lần: 5/5 lần vẫn trả lời 30kg → lỗi không phụ thuộc hội thoại.
3. Đào tiếp: tài liệu được retrieve là bảng chính sách năm cũ → lỗi retrieval/grounding, sửa bằng kỹ thuật một lượt (cập nhật kho tài liệu, eval retrieval ở Chương 7).
4. Tiết kiệm được gì: không cần xây dataset hội thoại, không cần sửa cơ chế context — một test single-turn là đủ bắt hồi quy.

1. Gộp thành 1 prompt: "2 người, có bé nhỏ, không bay đêm, ngày 14, lấy chuyến rẻ nhất" → 10/10 lần bot chọn chuyến 14:20 hợp lệ. Lỗi không tái hiện → multi-turn thật.
2. Cắt prefix: lấy U1…U8 (bỏ B8) — tức "N-1 lượt" đầy đủ trước lượt hỏng — làm một test case.
3. Sample nhiều lần: gọi model 10 lần trên prefix đó; đếm bao nhiêu lần đề xuất tôn trọng ràng buộc. Một lần chạy không nói lên gì — thứ ta đo là tỉ lệ giữ ràng buộc.
4. Nhân rộng: gom 31 prefix tương tự từ log (hoặc sinh tổng hợp bằng cách chèn ràng buộc ở lượt đầu và kéo dài hội thoại) → một bộ test trí nhớ ngắn hạn có quy mô, dùng để so sánh các cách sửa.

Code dưới thay hai LLM bằng hàm mock để thấy rõ khung: 3 persona, mục tiêu cần 4 thông tin (điểm đến, ngày, số khách, ngân sách). Agent mock có cửa sổ ngữ cảnh 6 lượt và hỏi từng thông tin một.

# Vòng lặp user simulator tối giản. simulator_turn() và agent_turn() là mock;
# thực tế mỗi hàm là một lời gọi LLM (simulator có system prompt persona + goal).
import random

PERSONAS = {
    "kiem_loi":  {"style": "cộc lốc", "changes_mind": 0.0, "patience": 6},
    "doi_y":     {"style": "hay đổi ý", "changes_mind": 0.5, "patience": 10},
    "buc_boi":   {"style": "gắt, ít kiên nhẫn", "changes_mind": 0.1, "patience": 4},
}
GOAL = {"slots": ["điểm đến", "ngày đi", "số khách", "ngân sách"]}

def simulator_turn(persona, state, rng):
    missing = [s for s in GOAL["slots"] if s not in state["given"]]
    if state["booked"]:
        return "[DONE]"
    if state["turn"] >= persona["patience"]:
        return "[GIVE_UP]"                       # user thật cũng bỏ đi như vậy
    if state["given"] and rng.random() < persona["changes_mind"]:
        slot = rng.choice(sorted(state["given"]))
        state["given"][slot] = "GIÁ TRỊ MỚI"
        return f"À đổi {slot} nhé"
    slot = missing[0] if missing else None
    if slot:
        state["given"][slot] = "ok"
        return f"({persona['style']}) {slot}: ..."
    return "Đặt luôn đi"

def agent_turn(user_msg, state, window=6):
    # mock thô của cửa sổ trượt: quá `window` lượt thì các slot nói từ đầu rơi khỏi prompt
    state["seen"] = state["given"].copy() if state["turn"] < window else {}
    if user_msg == "Đặt luôn đi":
        if len(state["seen"]) < len(GOAL["slots"]):
            return "Bạn muốn đi đâu ạ?"          # hỏi lại thứ đã được nói -> lỗi coherence
        state["booked"] = True
        return "Đã giữ chỗ!"
    return "Ghi nhận. Còn gì nữa không ạ?"

def simulate(persona_name, seed):
    rng, persona = random.Random(seed), PERSONAS[persona_name]
    state, transcript = {"given": {}, "booked": False, "turn": 0}, []
    while True:
        u = simulator_turn(persona, state, rng)
        if u in ("[DONE]", "[GIVE_UP]"):
            return transcript, ("PASS" if u == "[DONE]" else "FAIL")
        transcript += [("user", u), ("agent", agent_turn(u, state))]
        state["turn"] += 1

for name in PERSONAS:
    labels = [simulate(name, s)[1] for s in range(20)]
    print(f"{name:9s} pass rate = {labels.count('PASS')}/20")

1. Kết quả: "kiệm lời" 20/20, "hay đổi ý" 3/20, "bực bội" 0/20.
2. Đọc "bực bội" 0/20: persona chỉ chịu 4 lượt, còn bot cần 5 lượt vì hỏi từng thông tin một → lỗi thiết kế luồng hỏi (nên hỏi gộp), không phải lỗi trí nhớ.
3. Đọc "hay đổi ý" 3/20: đổi ý kéo dài hội thoại quá 6 lượt → rơi khỏi cửa sổ → bot hỏi lại điểm đến → truncation.
4. Bài học: một con số pass rate tổng (23/60 = 38%) che mất hai nguyên nhân hoàn toàn khác nhau. Luôn tách pass rate theo persona và goal.

Release B đổi giao diện: thêm nút gợi ý nhanh nên user hỏi ngắn hơn. Pass rate tổng tăng từ 68% lên 82% — team định ăn mừng.

1. Tách bucket: bucket 2–4 giữ ~92%, bucket 5–9 giữ ~66–67%. Bucket 10+ từ 34% còn 20%.
2. Vì sao tổng tăng? Tỉ trọng session ngắn (dễ) tăng từ 38% lên 71% tổng số session. Tổng tăng là do trộn dữ liệu thay đổi, không phải do bot giỏi lên — một dạng nghịch lý Simpson.
3. Nhưng cẩn thận: bucket 10+ của B chỉ có 15 session; khoảng tin cậy 7%–45% chồng lên 24%–47% của A. Chưa kết luận được "B tệ hơn" ở bucket dài — chỉ biết "không có bằng chứng B tốt hơn", và cần thêm mẫu (dùng simulator với persona dài dòng/hay đổi ý để bù).
4. Báo cáo đúng: luôn in bảng theo bucket kèm n và CI; con số tổng chỉ là dòng cuối.

# Báo cáo pass rate theo bucket độ dài + minh hoạ "bẫy con số tổng" (số liệu minh hoạ).
import math

def wilson(k, n, z=1.96):
    p = k / n
    c = (p + z*z/(2*n)) / (1 + z*z/n)
    h = z * math.sqrt(p*(1-p)/n + z*z/(4*n*n)) / (1 + z*z/n)
    return c - h, c + h

def bucket(turns):
    return "2-4" if turns <= 4 else "5-9" if turns <= 9 else "10+"

def report(name, sessions):
    """sessions: list[(so_luot, pass_bool)]"""
    by = {}
    for turns, ok in sessions:
        by.setdefault(bucket(turns), []).append(ok)
    total = sum(ok for _, ok in sessions)
    print(f"{name}: overall {total}/{len(sessions)} = {total/len(sessions):.0%}")
    for b in ("2-4", "5-9", "10+"):
        xs = by.get(b, [])
        if xs:
            lo, hi = wilson(sum(xs), len(xs))
            print(f"   {b:4s} {sum(xs):3d}/{len(xs):<3d} = {sum(xs)/len(xs):4.0%}  CI [{lo:.0%}, {hi:.0%}]")

def make(counts):  # counts: {bucket_turns: (n_pass, n_total)}
    out = []
    for turns, (k, n) in counts.items():
        out += [(turns, True)] * k + [(turns, False)] * (n - k)
    return out

# Release A: nhiều hội thoại dài.  Release B: đổi UI -> user hỏi ngắn hơn.
release_a = make({3: (88, 96), 7: (61, 92), 12: (22, 64)})
release_b = make({3: (170, 185), 7: (40, 60), 12: (3, 15)})
report("Release A", release_a)
report("Release B", release_b)
# Overall B > A, nhưng bucket 10+ của B tệ hơn -> con số tổng che mất sự xuống cấp.

1. Danh sách perturbation cố định (6 loại): gõ không dấu, đổi mục tiêu, mâu thuẫn thông tin trước (đổi số khách), thêm mơ hồ, phản bác agent, chêm chủ đề lạc.
2. Với mỗi trace đang Pass (40 trace): xáo thứ tự perturbation theo seed, áp cộng dồn, sau mỗi bước chạy lại agent + judge; dừng khi Fail.
3. v1: median = 3, không trace nào sống sót hết; "cú cuối" hay gặp nhất: mâu thuẫn thông tin (14 lần), đổi mục tiêu (9 lần) → đúng chỗ yếu anchoring.
4. v2 (thêm bước rà ràng buộc + trip sheet): median = 5, 15/40 trace sống sót hết. Chỗ yếu còn lại chuyển sang "thêm mơ hồ" → việc tiếp theo là dạy bot hỏi lại khi thiếu thông tin.

# Brittleness score: với mỗi trace đang PASS, áp perturbation lần lượt (thứ tự ngẫu nhiên)
# cho tới khi agent FAIL; đếm số bước. Median thấp = mong manh. (Mock, số liệu minh hoạ.)
import random, statistics

PERTURBATIONS = ["typo_khong_dau", "doi_muc_tieu", "mau_thuan_thong_tin",
                 "them_mo_ho", "phan_bac_agent", "chem_chu_de_lac"]
# Xác suất agent CHỊU ĐƯỢC từng perturbation (giả định cho 2 phiên bản agent).
ROBUST = {
    "v1": {"typo_khong_dau": .95, "doi_muc_tieu": .45, "mau_thuan_thong_tin": .40,
           "them_mo_ho": .70, "phan_bac_agent": .60, "chem_chu_de_lac": .85},
    "v2": {"typo_khong_dau": .97, "doi_muc_tieu": .85, "mau_thuan_thong_tin": .80,
           "them_mo_ho": .80, "phan_bac_agent": .75, "chem_chu_de_lac": .90},
}
CAP = len(PERTURBATIONS) + 1   # sống sót hết 6 perturbation -> ghi 7 (bị "censored")

def still_passes(agent, applied, rng):
    # mock: mỗi perturbation mới là một phép thử độc lập
    return rng.random() < ROBUST[agent][applied[-1]]

def brittleness(agent, rng):
    order = PERTURBATIONS[:]
    rng.shuffle(order)
    applied = []
    for step, p in enumerate(order, start=1):
        applied.append(p)
        if not still_passes(agent, applied, rng):
            return step, p
    return CAP, None

for agent in ("v1", "v2"):
    rng = random.Random(7)
    scores, killers = [], {}
    for trace in range(40):               # 40 trace đang PASS
        s, killer = brittleness(agent, rng)
        scores.append(s)
        if killer:
            killers[killer] = killers.get(killer, 0) + 1
    top = sorted(killers.items(), key=lambda kv: -kv[1])[:2]
    print(f"{agent}: median brittleness = {statistics.median(scores)}, "
          f"sống sót hết = {scores.count(CAP)}/40, hay gãy vì: {top}")

Mock giả định mỗi perturbation là phép thử độc lập — thực tế chúng tương tác (typo + mơ hồ khó hơn tổng hai cái), đó là lý do phải chạy lại agent thật sau mỗi bước.

Failure mode lớn nhất trong sổ lỗi là "đề xuất vi phạm ràng buộc đã nêu". Đây là việc có thể kiểm bằng code phần lớn: trích ràng buộc (ngân sách, giờ bay, số khách) từ các lượt user — lượt sau ghi đè lượt trước — rồi so với đề xuất cuối.

# Evaluator cấp turn có chủ đích: trích ràng buộc user nêu trong CẢ hội thoại,
# rồi kiểm tra đề xuất cuối có tôn trọng không. Phần trích xuất ở đây dùng regex
# cho gọn; thực tế thường là một LLM call trả JSON, còn phần KIỂM TRA vẫn là code.
import re

def extract_constraints(user_turns):
    c = {}
    for i, t in enumerate(user_turns):
        t = t.lower()
        if m := re.search(r"dưới (\d+) ?(tr|triệu)", t):
            c["max_price"] = (int(m.group(1)) * 1_000_000, i)      # lưu cả lượt nêu
        if "không bay đêm" in t or "không bay tối" in t:
            c["no_red_eye"] = (True, i)
        if m := re.search(r"(\d+) (người|khách)", t):
            c["pax"] = (int(m.group(1)), i)                        # lượt sau ghi đè lượt trước
    return c

def check_offer(constraints, offer):
    violations = []
    if "max_price" in constraints and offer["total"] > constraints["max_price"][0]:
        violations.append(f"vượt ngân sách nêu ở lượt {constraints['max_price'][1] + 1}")
    if "no_red_eye" in constraints and not (6 <= offer["depart_hour"] < 21):
        violations.append(f"bay đêm dù user dặn ở lượt {constraints['no_red_eye'][1] + 1}")
    if "pax" in constraints and offer["pax"] != constraints["pax"][0]:
        violations.append(f"sai số khách (lượt {constraints['pax'][1] + 1} nói {constraints['pax'][0]})")
    return ("PASS", []) if not violations else ("FAIL", violations)

user_turns = [
    "Mình muốn đi Đà Nẵng cuối tháng 10, 3 người",
    "Ngân sách dưới 9 triệu cả nhà nhé, và không bay đêm vì có bé nhỏ",
    "À bà ngoại đi cùng, thành 4 người",
    "Ok chốt chuyến rẻ nhất giúp mình",
]
offer = {"total": 8_400_000, "depart_hour": 22, "pax": 3}
print(check_offer(extract_constraints(user_turns), offer))

# Judge cấp session (khung prompt tự viết — nhớ hiệu chỉnh với nhãn người, xem Ch.5):
SESSION_JUDGE = """Bạn chấm một phiên chat của trợ lý đặt vé.
Mục tiêu user: {goal}
Tiêu chí PASS (tất cả phải đúng): {rubric}
Hội thoại đầy đủ:
{transcript}
Trả về JSON: {{"reasoning": "...", "first_failing_turn": <số hoặc null>, "verdict": "PASS|FAIL"}}"""

1. Kết quả: FAIL với 2 vi phạm — bay 22h dù user dặn ở lượt 2; sai số khách (lượt 3 đã đổi thành 4). Evaluator trả luôn lượt nêu ràng buộc → drill-down gần như miễn phí.
2. Vì sao lai (hybrid): trích xuất ràng buộc từ ngôn ngữ tự nhiên thì LLM làm tốt hơn regex; còn so sánh số và giờ thì code chính xác và rẻ hơn LLM.
3. Judge session (khung prompt ở cuối code) yêu cầu trả first_failing_turn — lợi ích kép: chấm session và gợi ý chỗ drill-down.

Như Chương 5: judge chỉ đáng tin sau khi đo độ khớp với nhãn người trên tập dev/test tách biệt. 100 session có nhãn người (40 Fail, 60 Pass):

1. TPR (bắt được Fail thật): 36/40 = 0.90. TNR (nhận đúng Pass): 50/60 ≈ 0.83.
2. Đọc 10 ca false-Fail: 7 ca judge phạt vì "bot hỏi lại một lần" — rubric cho phép nhắc lại 1 lần. Sửa prompt judge: đưa nguyên văn điều kiện ③ của rubric và 2 ví dụ biên.
3. Đọc 4 ca false-Pass: đều là hội thoại > 20 lượt, vi phạm ràng buộc nằm ở giữa → judge cũng "quên"! Giải pháp: cho judge nhận thêm danh sách ràng buộc đã trích (từ check 6.10a) thay vì tự dò trong transcript dài.
4. Ước lượng pass rate thật trên production từ tỉ lệ judge + TPR/TNR (công thức hiệu chỉnh ở Chương 5).

Session có 3 lượt khó (lượt 3, 6, 10). So sánh per_turn (khó → mạnh, dễ → rẻ), sticky (chốt model theo lượt đầu), sticky_escalate (bắt đầu rẻ, gặp lượt khó thì lên model mạnh và ở luôn):

# Chi phí 1 session dưới 3 chính sách routing, có tính prompt cache.
# Giá/token là SỐ MINH HOẠ, không phải bảng giá thật của nhà cung cấp nào.
PRICE = {  # USD / 1M token: input chưa cache, input đọc từ cache, output
    "strong": {"in": 3.00, "cached": 0.30, "out": 15.0},
    "cheap":  {"in": 0.25, "cached": 0.025, "out": 1.25},
}
TURN_TOKENS, OUT_TOKENS, SYSTEM = 700, 250, 2000
# độ khó từng lượt của MỘT session (0 = dễ, 1 = khó), vd. từ classifier của router
DIFFICULTY = [0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 0, 0]

def route(policy, t, prev):
    hard = DIFFICULTY[t] == 1
    if policy == "per_turn":
        return "strong" if hard else "cheap"
    if policy == "sticky":
        return prev or ("strong" if hard else "cheap")      # chốt model ở lượt đầu
    if policy == "sticky_escalate":
        return "strong" if (hard or prev == "strong") else "cheap"  # lên rồi không xuống

def route_seq(policy):
    seq, prev = [], None
    for t in range(len(DIFFICULTY)):
        prev = route(policy, t, prev)
        seq.append(prev)
    return seq

def session_cost(policy):
    seq = route_seq(policy)
    cost, cached_tokens, total_in = 0.0, 0, 0
    warm = {"strong": 0, "cheap": 0}   # số token prefix đã nằm trong cache của từng model
    for t, m in enumerate(seq):
        ctx = SYSTEM + t * (TURN_TOKENS + OUT_TOKENS) + TURN_TOKENS
        hit = min(warm[m], ctx)        # đổi model -> cache model mới lạnh hoặc chỉ có prefix cũ
        p = PRICE[m]
        cost += (hit * p["cached"] + (ctx - hit) * p["in"] + OUT_TOKENS * p["out"]) / 1e6
        warm[m] = ctx + OUT_TOKENS     # (bỏ qua TTL của cache cho gọn)
        cached_tokens, total_in = cached_tokens + hit, total_in + ctx
    switches = sum(a != b for a, b in zip(seq, seq[1:]))
    hard_on_cheap = sum(d == 1 and m == "cheap" for d, m in zip(DIFFICULTY, seq))
    return cost, switches, cached_tokens / total_in, hard_on_cheap

for pol in ("per_turn", "sticky", "sticky_escalate"):
    c, sw, hit, hc = session_cost(pol)
    print(f"{pol:16s} cost=${c:.4f}  switches={sw}  cache-hit={hit:.0%}  hard-turns-on-cheap={hc}")

1. sticky rẻ nhất (~$0.009) và cache tốt nhất, nhưng cả 3 lượt khó rơi vào model rẻ → rủi ro chất lượng cao nhất.
2. sticky_escalate không đưa lượt khó nào cho model rẻ và chỉ đổi 1 lần, nhưng đắt nhất (~$0.095) vì lên model mạnh từ lượt 3 và ở đó 10 lượt.
3. per_turn ở giữa (~$0.054), không lượt khó nào vào model rẻ, nhưng đổi model 6 lần → 6 ranh giới có rủi ro handoff inconsistency, cache hit thấp nhất (77%).
4. Kết luận: không chính sách nào thắng tuyệt đối; chi phí đã rõ, còn chất lượng phải đo bằng judge session trên nhiều session. Chỉ số so sánh hợp lý: chi phí trên mỗi session thành công = tổng chi phí / số session Pass.

Giả sử chạy 200 session ảo (simulator 6.7) mỗi chính sách, judge session cho pass rate: sticky 71%, per_turn 83%, sticky_escalate 86% (minh hoạ). Chi phí/session thành công: sticky ≈ 0.0085/0.71 ≈ $0.012; per_turn ≈ 0.0544/0.83 ≈ $0.066; sticky_escalate ≈ 0.0951/0.86 ≈ $0.111. Nếu yêu cầu chất lượng tối thiểu là 80%, per_turn thắng; và bước tiếp theo là thử biến thể "per_turn nhưng cấm đổi model trong 2 lượt sau lượt khó" để giảm số lần đổi.