Helpers

analyse(state: Union[str, bytes, Mapping, list, None]) -> Dict[str, object]

對一個狀態做完整檢測的結果。

返回 script、script_profile、language（盡力而為，可能是 None）、is_english、non_latin_fraction 與 mixed_segment（讓一個以英語為主的狀態被判成非英語的那一行或那個欄位，沒有就是 None）。

真正被讀取的是字串值。一個狀態裡有多個字串值時，只要有一個非英語就夠了：把所有值拼進同一段視窗，會讓一段很長的英語備註佔滿 4000 字元，或者把一條很短的德語訊息壓過去，於是那條訊息被送去了英語 checkpoint（#384）。段落掃描仍然在 4000 字元處停下，正是這一點讓超大欄位不至於變貴；沒掃到的值之後會單獨讀一遍。

detect_script(text: str) -> str

text 的主導文字系統：'latin'、'han'、'devanagari' …… 沒有字母時是 'unknown'。

is_english(state: Union[str, bytes, Mapping, list, None]) -> bool

當英語 checkpoint 有望讀懂這個狀態時為 True。

clean_email_body(body: str, max_chars: int = 3000) -> str

去掉郵件裡被引用的往來歷史、簽名與免責宣告，讓輸入保持聚焦。

max_chars 是結果被截到的長度，預設 3000 字元 —— 見 email_state，它接受同一個預算並透傳下來。

email_state(
    subject: str,
    body: str,
    sender: Optional[str] = None,
    clean: bool = True,
    max_chars: int = 3000,
    extra,
) -> Dict

為郵件分類構造一個乾淨的狀態字典。

max_chars 是 clean_email_body 截斷正文的預算，訊息較長時值得調大：預設值下正文在 3000 字元處就停了，於是寫在最後幾段裡的訴求永遠到不了模型 —— 走 predict_long 也一樣，它把狀態切成視窗掃描，本來就是為了能讀到超過一個視窗的內容。clean=False 時忽略此項，正文整段透傳。

其它任何關鍵字都會成為狀態的一個欄位，因此會被模型讀到；這裡拼錯字不是報錯，而是改了輸入。

triage_questions() -> Dict

客服工單分流用的預設問題。

email_questions(categories: Optional[Dict[str, str]] = None) -> Dict

來信分流與威脅過濾用的預設問題。

guard_questions() -> Dict

即時 LLM 輸入防護欄用的預設問題。

moderation_questions() -> Dict

內容安全與稽核用的預設問題。

router_questions() -> Dict

智慧模型路由用的預設問題。

shortlist_choice(
    state: Any,
    criteria: Any,
    embed_fn: Callable[[Sequence[str]], Any],
    k: int = DEFAULT_SHORTLIST_K,
    DEFAULT_SHORTLIST_K,
    instructions: Optional[str] = None,
    return_scores: bool = False,
) -> Any

返回 state 的前 k 個 choice 標籤。

embed_fn 把一個字串列表對映成形狀為 (len(texts), dim) 的陣列。它只被呼叫一次：先是查詢文本，然後按 criteria 順序每個選項一個字串。選項字串與 choice 問題的 render_options 一致。

當 k 不小於標籤數時，所有標籤按原順序返回，且不呼叫 embed_fn。

並列時保留靠前的標籤。排序用的是有符號餘弦，而不是相似度下限：得分為 0 的標籤 —— 完全沒有訊號，或者一個非有限向量被當作 0 處理 —— 確實會壓過更靠前但得分為負的標籤，而 k 會先丟掉負分標籤。

return_scores=True 時返回的是 (labels, scores) 對，其中 scores 按排名順序儲存每個保留標籤的有符號餘弦 —— 就是 predict_shortlist 在其 shortlist 後設資料裡報告的那些值。當沒有丟棄任何標籤時，scores 是 None，與該後設資料中完全一致。

predict_shortlist(
    agent: Any,
    state: Any,
    questions: Dict[str, Dict[str, Any]],
    embed_fn: Callable[[Sequence[str]], Any],
    k: int = DEFAULT_SHORTLIST_K,
    DEFAULT_SHORTLIST_K,
    predict_kwargs: Any,
) -> Dict[str, Any]

先對每個 choice 問題做候選篩選，再呼叫一次 predict 或 system_one。

非 choice 的問題原樣透傳。標籤數 <= k 的 choice 問題也原樣透傳，且不呼叫 embed_fn。呼叫方傳入的 questions 字典不會被修改。

返回的字典是模型結果加一個 shortlist 條目。經過篩選的 choice 問題，其機率只在保留下來的標籤上歸一。shortlist[qid] 裡是 labels、scores、k、n 與 passthrough。labels 是篩選產生的排序順序；當 passthrough 置位、沒有做排序時，就是 criteria 本身的順序。scores 是按該順序每個保留標籤的有符號餘弦 —— 包含負值，絕不夾到 0 —— 或者當沒有丟棄任何標籤時為 None。

額外的關鍵字參數會透傳給 predict / system_one（例如 Router 上的 model=）。

embed_fn_from_agent(
    agent: Any,
    max_length: int = 512,
    batch_size: int = 32,
) -> Callable[[Sequence[str]], np.ndarray]

對 agent 上已經載入好的 checkpoint 編碼器做平均池化。

返回的可呼叫物件用 agent.tok 與 agent.model.encoder 嵌入一個字串列表。它不跑決策頭，也不下載權重。傳入一個專門的 bi-encoder 作為 embed_fn 通常能篩得更準；這個輔助函式是給手上只有 Laya checkpoint 的呼叫方用的。

平均時排除 padding 位置。編碼器的 train/eval 標誌保持呼叫方設定的樣子（載入好的 Agent 本來就在 eval）。每次呼叫都用當前的 agent.device，回退到 CPU 之後也一樣。

cached_embed_fn(
    embed_fn: Callable[[Sequence[str]], Any],
    maxsize: int = 4096,
) -> Callable[[Sequence[str]], np.ndarray]

按輸入字串快取 embed_fn 的輸出，用一個 LRU 上界限制。

predict_shortlist 每次呼叫都要嵌入查詢文本以及每個選項的文本。當每個請求篩選的都是同一組選項 —— 比如固定的意圖或標籤列表，README 裡的 BANKING77 例子就是這樣 —— 選項那些行在呼叫之間並沒有變化，卻每次都被重新嵌入一遍。把嵌入器包一層：

embed_fn = cached_embed_fn(embed_fn_from_agent(agent))

第一次呼叫不變，之後每次重複呼叫都只剩下嵌入新的查詢文本。

查表按字串精確匹配。快取裡沒有的文本會先去重，再放在一次 embed_fn 呼叫裡嵌入，所以冷快取與不包一層時花費的批次呼叫次數相同。行以 float32 儲存；快取最多保留 maxsize 個字串，超出後淘汰最久未用的那一項，記憶體上界約為 maxsize * dim * 4 位元組。embed_fn 拋錯或返回的形狀不對時，什麼都不快取。

這個包裝器可以安全地在多個執行緒間共用：鎖只覆蓋快取的讀寫，從不覆蓋嵌入呼叫本身。返回的可呼叫物件帶有 cache_info() —— 一個含 size、maxsize、hits 與 misses 的字典 —— 以及 cache_clear()。如果 embed_fn 背後的模型或權重換了，請清空快取。

check_min_confidence(v: Any)

校驗可選的棄權閾值 min_confidence（#361、#394）。

要麼是 [0.0, 1.0] 區間內的實數（所有答案共用一個閾值；布林值會被拒絕，儘管 isinstance(True, int) 成立），要麼是按桶對映（見 :func:check_min_confidence_map），讓閾值可以隨選項個數而不同。返回的是校驗後的值 —— 標量情形是 float，對映情形是 dict[str, float] —— 下面兩個門控函式都接受這種形式。

check_min_confidence_map(m: Dict[Any, Any]) -> Dict[str, float]

校驗一張按桶的棄權閾值對映（#394）。

鍵是 common.temp_bucket 拼法下的選項數桶字串 —— "choice:2"、"choice:3-5"、"score:6-10"、"noul:2" 等等 —— 外加一個可選的 "default"，用於對映沒有點名的任何桶。值是 [0.0, 1.0] 內的浮點數。一個置信度閾值無法跨選項個數遷移（#394）；這讓呼叫方可以按每個桶的校準實際能達到的水平來門控它。用 :func:laya.calibrate.fit_abstention_thresholds 擬合一張。

resolve_min_confidence(
    answer: Dict[str, Any],
    thresholds: Dict[str, float],
    default: float = 0.0,
) -> float

在按桶對映下，這個答案的選項數桶所被門控的閾值。

對於對映沒有點名的桶，依次回退到對映的 "default" 條目，再回退到 default（0.0 —— 不門控任何東西），所以未配置的桶永遠不會出其不意地棄權。

flag_low_confidence(results: List[Dict[str, Any]], min_confidence: float) -> None

可選的棄權標記（#361）：把置信度低於 min_confidence 的答案標出來。

讀取 answer_confidence（即 max(p)，校準指標所描述、且不隨選項個數漂移的那個量），answer_confidence 缺失時退回 confidence。原始答案與置信度保持不變：當答案低於閾值時加上 low_confidence: True，而先前被標記過的答案如果重新達標，就把它去掉（例如結果字典被複用時，或者用不同閾值重新評估時）。

min_confidence 要麼是浮點數（所有答案共用一個閾值），要麼是按桶對映（#394）；後者下，每個答案按它自己選項數所在桶的閾值，經由 :func:resolve_min_confidence 做門控。

apply_confidence_gate(
    results: List[Dict[str, Any]],
    min_confidence: Optional[float] = None,
) -> None

在真正施加過門控的那些答案上，報告置信度門控的狀態。

門控是一種策略，而一個連施加與否都無法觀察的策略根本算不上策略。設定 min_confidence 後，這會把 abstention（:data:GATE_STATES 之一）寫到每個答案上，同時寫上 abstention_threshold，於是呼叫方能回答三個原本回答不了的問題：

• 有多大比例的決策棄權了，而不是靠 low_confidence 是否恰好被設定去反推；
• 有多少答案門控無法決定，這是布林值根本無法表達的；
• 是什麼閾值產生了這些結果 —— flag_low_confidence 會把閾值消費掉，所以沒有它，一次帶逐類閾值的批次執行就無法重新拆分。

GATE_UNEVALUATED 是布林值無法表達的情形：門控跑了，而答案沒有帶可用的置信度，於是門控無法決定。把它報告成通過，和報告成一個標記是同一個謊言。

未設定 min_confidence 時，這什麼都不寫。 沒有 abstention，沒有 abstention_threshold，也沒有標記。這就是全部的約定：一次未門控的呼叫返回的載荷與之前完全相同，而呼叫方是靠欄位是否出現（而不是從欄位裡讀出的第四個值）判斷門控跑過沒有。要在每次呼叫時無條件地呼叫它，取代 if min_confidence is not None: 這種守衛：正是那種守衛留下了一條什麼都不報告的通路，而這正是本函式要區分出來的狀態。

標記本身仍歸 :func:flag_low_confidence 所有 —— 這裡是委託，而不是重新實現那條規則，所以布林值與所報告的狀態不會漂移開。

恰好為 0.0 的 min_confidence 確實被設定了，所以狀態會被報告，而 :func:flag_low_confidence 把 0.0 當作空操作，因為沒有東西能掉到它下面。於是每個帶可用置信度的答案都讀作 passed，而閾值回顯正是把它與在真實閾值下一次真正的通過區分開的東西。

GATE_STATES = (GATE_PASSED, GATE_ABSTAINED, GATE_UNEVALUATED)

answer_confidence(p: np.ndarray, k: int) -> float

落在所報告答案上的機率質量：max(p)。

溫度縮放擬合的就是這個量，本倉庫裡每一個校準數字也都是在這個量上算的 —— 兩個基準測試框架在呼叫 ece_score 之前都取 conf = max(probs)。README 的門控一節依賴的是隨之而來的性質：在置信度為 c 時返回的答案裡，大約有 c 的比例是對的。這個性質是有條件的，而預設情況下條件並不成立 —— 只有在針對這個 checkpoint、這個選項個數，把溫度擬合出來並在留出資料上驗證過之後才成立。已釋出的 checkpoint 是過度自信的：choice:11+ 是一個約 10 倍的銳化器，會返回落在 1.0 上的點質量，所以對它施加閾值會讓選中率低於模型準確率（issue #394）。

下面的 confidence_from_probs 報告的是另一個量、另一個尺度，沒有這個保證，所以兩者不能拿同一個閾值去比。

confidence_from_probs(p: np.ndarray, k: int) -> float

歸一化夏農熵置信度：1 - H(p) / log(k)。

它刻畫的是整個分佈有多集中。有用，但沒有校準：溫度縮放擬合的不是它，報告的 ECE 量的也不是它。見 answer_confidence。

ece_score(conf: np.ndarray, correct: np.ndarray, bins: int = 15) -> float

按置信度分箱計算的期望校準誤差（ECE）。

fit_temperatures = fit_temperature_map

fit_one_temperature(pairs: Sequence, min_n: Optional[int] = None) -> float

用 NLL + LBFGS 在 log T 上擬合單個標量 T。

結果是優化出的尺度經過 clamp_temperature 之後的值，因此落在 [TEMP_MIN, TEMP_MAX] 內（值不是數時取中性的 1.0）。給出的樣本對少於 min_n 時返回 1.0。min_n 預設是 MIN_BUCKET_N（每個桶的下限）。型別級擬合傳的是更低的 MIN_TYPE_N，因此一個填不滿任何桶的資料集仍然能得到一個標量，而不是停在 1.0。

fit_temperature_map(
    records: Iterable,
    compute_ece: bool = False,
    seed: int = 0,
) -> Dict[str, Any]

擬合型別級標量與逐桶溫度。

MIN_BUCKET_N 是每個桶的下限：更小的桶不進入 temperature_by_options，由型別級標量覆蓋它們。MIN_TYPE_N 是隻作用於那個標量的、單獨且更低的下限。

compute_ece=False（預設值，也是 Agent.fit_temperatures 存下來的那條路徑）在全部記錄上擬合，且不返回 report 鍵。這條路徑上 seed 被忽略。

compute_ece=True 會按 temp_bucket 分層，從每個桶裡留出 ECE_HOLDOUT_FRAC 的比例，用 seed 保證同一批記錄總是以同樣的方式切分。溫度只在剩下的部分上擬合，ECE 也只在留出的記錄上評分。report["n"] 是傳入的記錄數；report["n_eval"] 是 ECE 所依據的留出條數。某個桶如果留出之後會低於 MIN_BUCKET_N，就在它的全部記錄上擬合、不進入評估集，並寫進 report["buckets_excluded_from_eval"]，而不是被丟掉。n_by_bucket 始終按完整輸入計數，包括擬合本身只用了子集的情況。

fit_abstention_thresholds(
    records: Iterable,
    temperature: Sequence[float],
    temperature_by_options: Dict[str, float],
    binning_map: Optional[Dict[str, Dict[str, Any]]] = None,
    target_error: float = 0.10,
    min_bucket_n: int = MIN_ABSTAIN_BUCKET_N,
    MIN_ABSTAIN_BUCKET_N,
    conservative: bool = True,
) -> Dict[str, float]

擬合一個逐 temp_bucket 的棄權閾值，讓門控在每個桶裡都守住目標誤差。

單個 min_confidence 無法跨選項個數遷移（#394）：2 選項與 12 選項答案的校準置信度處在不同的尺度上，所以同一刀切會隨問題不同而過度棄權或棄權不足。這個函式改為每個桶擬合一刀，鍵與 temperature_by_options 完全一致（common.temp_bucket，例如 "choice:3-5"），結果是一張 min_confidence 對映，:func:laya.confidence.check_min_confidence / :func:laya.confidence.apply_confidence_gate 直接接受它。

records 與 fit_temperature_map 消費的是同一批 (qtype, logits, target[, k]) 元組（由 records_from_labeled 構造）。置信度是校準後的 max(p) —— logits 先被擬合出的 temperature / temperature_by_options 縮放，所以閾值與執行時報告的數字處在同一尺度上。target_error 是已接受答案中可容忍的誤差；min_bucket_n 會略過太小的桶不擬合，conservative 則加上一樣本量的餘量。這些閾值是校準集上的經驗切點，不是形式化的覆蓋率保證 —— 生產門控要在留出資料上驗證（fit_temperature_map(..., compute_ece=True) 會給出一個留出切分）。

當將要提供這些閾值的 agent 裝了一張 binning_map 時，要把它傳進來 —— 無論這張對映是 Agent.fit_binning 裝的，還是由帶 binning_map 的校準載荷裝的 —— 因為執行時會在任何東西讀取 answer_confidence 之前，先經由那張對映重新校準它，所以沒有它擬合出的切點，落在的是一個門控根本看不到的尺度上。這樣閾值就落在分箱後的尺度上，兩者擬合的先後順序也不再重要。在 1,200 條合成的 12 選項記錄上、target_error=0.10 時的實測結果：沒有對映擬合出的切點，在未分箱置信度上以 50% 覆蓋率守住 9.8% 的誤差；而一旦把同一個數字與分箱後的置信度相比，它在 25.6% 的誤差下就放行了 94.5% 的答案。

fit_binning_map(
    records: Iterable,
    temperature: Sequence[float],
    temperature_by_options: Dict[str, float],
    bins: int = 15,
    min_bucket_n: int = MIN_BINNING_BUCKET_N,
    MIN_BINNING_BUCKET_N,
) -> Dict[str, Dict[str, Any]]

為 answer_confidence 擬合一張逐 temp_bucket 的直方圖分箱重校準對映。

溫度縮放對每個桶只施加一個標量；它修不好一條可靠性曲線不是簡單銳化/鈍化的桶（隨英文包釋出的 choice:11+ 就是這樣一個病態桶）。直方圖分箱是非參數替代方案：把一個桶裡校準後的置信度分到 [0, 1] 上 bins 個等寬箱裡，再把落進某個箱的每個置信度對映到那個箱的經驗準確率。它不需要單調性假設，也不需要額外的依賴（只用 NumPy；等滲迴歸會引入 scikit-learn）。

records 與 fit_temperature_map 消費的是同一批 (qtype, logits, target[, k]) 元組；置信度是校準後的 max(p)（logits 先被擬合出的 temperature / temperature_by_options 縮放），所以分箱對映是疊加在溫度對映之上，而不是取代它。返回 {bucket: {"bins": N, "values": [recalibrated confidence per bin]}}；低於 min_bucket_n 的桶會被略去。用 :func:apply_binning_map 施加它。空箱（校準集從未產生過的置信度區間）對映到它自己的中點，也就是讓那個區域保持不變，所以沒見過的值永遠不會被重校準成一個憑空捏造的 0。

apply_binning_map(
    confidence: float,
    bucket: str,
    binning_map: Dict[str, Dict[str, Any]],
) -> float

按選項數 bucket（common.temp_bucket）重校準一個 answer_confidence。

當對映裡沒有該桶的條目時，返回原樣的置信度，所以對映沒有為它擬合的桶會直接通過，而不是被強行賦一個錯誤的值。

fit_binning(
    records,
    min_bucket_n: int = MIN_BINNING_BUCKET_N,
    MIN_BINNING_BUCKET_N,
) -> Dict[str, Any]

在這個 agent 已擬合的溫度之上擬合一張直方圖分箱對映，並存下來。

records 與 fit_temperatures 消費的是同一批 (qtype, logits, target[, k]) 元組。這張對映的鍵與 temperature_by_options 完全一致，疊加在當前溫度之上，由 save_calibration 寫成 binning_map。

render_options(q: Dict) -> List[str]

按標籤下標順序渲染選項文本。Noul 的語義順序永遠是 [false, true]。

proper_reward(
    q: torch.Tensor,
    target: torch.Tensor,
    qtype: torch.Tensor,
    mask: torch.Tensor,
    w_sph: float = 0.5,
    w_rps: float = 1.0,
    log_floor: float = -9.21,
) -> torch.Tensor

嚴格 proper 評分規則的獎勵：log score + spherical score + ranked probability score。

q: [..., N, K] 報告的分佈 target: [N, K]（one-hot 或軟目標分佈）

td_lambda_targets(p_true: torch.Tensor, batch: Dict, lam: float = 1.0) -> torch.Tensor

多輪對話軌跡的 TD(lambda) 目標。

QTYPES = {"choice": 0, "score": 1, "noul": 2}

QTYPE_NAMES = {v: k for k, v in QTYPES.items()}