文件導航

路由

Router 為每個請求挑選一個 checkpoint,並在需要預測時載入它的 Agent。預設路由器把英文文本送到英語 checkpoint,把其他受支援的語言送到多語言 checkpoint。你可以覆蓋這個選擇,提供自己的語言提示,或者 顯式選出 typed-decisions checkpoint。

本指南講的是模型選擇和生命週期。預測接受的問題型別見 Schema 驅動的決策; 生命週期回撥見 預測鉤子。

快速開始

from laya import Router

router = Router()

questions = {
    "department": {
        "type": "choice",
        "instructions": "Which department should handle this request?",
        "criteria": {
            "billing": "invoices, payments, refunds",
            "technical": "bugs, outages, system errors",
            "other": "everything else",
        },
    }
}

result = router.predict("We were billed twice. Please refund the duplicate charge.", questions)
print(result["answers"]["department"]["choice"])
print(result["routing"]["model"])

建立 Router() 預設不會下載 checkpoint。predict() 先給請求做路由,然後在首次使用時載入選中的 checkpoint。所以第一次預測可能更慢,因為要下載檔案並初始化模型;之後的預測會複用已載入的 agent。

checkpoint 是如何選出的

Router.route(state, questions, ...) 返回一個 RouteDecision,不載入 checkpoint 也不跑推理。這個 決策包含選中的模型、一段給人讀的理由,以及使用內建檢測時的語言檢測細節。

路由按這個順序檢查輸入:

  1. model= 直接選出一個 checkpoint。
  2. task= 為指定任務選出一個 checkpoint。
  3. 如果 auto_task_detection=True,與某個已知 typed-decisions 問題 ID 集合精確匹配時會選出 typed-decisions。
  4. 一個被識別的 lang= 值選出英語或多語言。
  5. 每次呼叫的 lang_guess=,或路由器配置的 lang_guess,會被考慮。
  6. 內建的書寫系統與語言分析選出一個 checkpoint。如果沒有可靠的語言訊號,路由器使用它配置的 default(預設是英語)。

第一條匹配的規則勝出。例如 model="multilingual" 覆蓋 lang="en"。非法的模型名會拋 ValueError, 而不是落到檢測那一步。

decision = router.route(
    "La aplicación se cierra cada vez que abro la configuración.",
    questions,
)
print(decision.model)   # multilingual
print(decision.reason)  # why that checkpoint was selected

RouteDecision 相容 dict,所以它的欄位也可以用 decision["model"] 和 decision["reason"] 這樣的鍵 取到。Router.predict() 會把同一個決策放進結果的 "routing" 鍵下。

覆蓋語言路由

當應用已經知道請求語言時,用 lang=。接受 "en"、"en-US" 和 "en_US.UTF-8" 這樣的語言標籤。 英語路由到 english;其他被識別的語言程式碼路由到 multilingual。

result = router.predict(state, questions, lang="de")
assert result["routing"]["model"] == "multilingual"

如果應用有自己的語言檢測器,把它的結果作為語言程式碼用 lang_guess= 傳進來。一個可呼叫物件接收 state,可以返回一個程式碼,或返回 None 表示棄權:

def detect_request_language(state):
    # Replace this with the application's detector.
    return "en" if "invoice" in str(state).lower() else None

router = Router(lang_guess=detect_request_language)

棄權的提示不會強制選一個 checkpoint;路由會繼續到下一條規則。這涵蓋 None、空字串,以及不 指明任何語言的程式碼(C、POSIX、C.UTF-8、und、zxx、mul)—— 那正是檢測器無話可說時返回的 東西,所以棄權不會悄悄把請求釘在錯誤的模型上。

無法識別的提示不算棄權。其他任何值,包括 "xx"、False 和 0,都被讀作「不是英語」,於是路由到 多語言 checkpoint。所以一個返回垃圾程式碼而非 None 的檢測器確實會選出一個 checkpoint;如果這有影響, 就在把它傳出去之前,把它的未知情況對映成 None。

內建檢測是一套輕量的書寫系統與語言啟發式,不是通用的語言識別模型。它分析 text、dict 和 list 狀態 裡的字串值;字典的鍵會被忽略,因為它們常常是英文的欄位名。短文本或有歧義的文本可能用到預設 checkpoint。對已知的工作負載,顯式的語言或應用提供的提示更可預測。

選出 typed-decisions

typed-decisions checkpoint 預設不會被自動選出。要顯式選它:

result = router.predict(state, questions, model="typed-decisions")
# `task="typed_decisions"` is also accepted.

或者設定 auto_task_detection=True。路由器這時會檢查問題 ID 是否與它已知的某個 typed-decision 工作流精確匹配。它不會從問題的措辭去推斷任務,而加入無關的問題 ID 會讓精確匹配失敗。

router = Router(auto_task_detection=True)

不載入模型就檢查路由

用 route() 檢查單個決策,或用 route_batch() 檢查一串。兩個方法都不載入 checkpoint,所以在跑 推理之前除錯路由規則時都很有用。

requests = [
    {"state": "Please refund the duplicate charge.", "questions": questions},
    {"state": "Necesito ayuda con mi factura.", "questions": questions},
]

decisions = router.route_batch(requests)
for decision in decisions:
    print(decision.model, decision.reason)

每個 route_batch() 條目都需要 state 和 questions;可選的路由覆蓋(model、task、lang 和 lang_guess)按條目指定。決策保持輸入順序。

載入與記憶體

預設情況下,路由器在首次需要時載入 checkpoint,並最多常駐兩個 agent。自動語言路由通常只需要英語 和多語言兩個 checkpoint。如果請求還可能選出 typed-decisions,一個很小的 max_loaded 可能會 驅逐另一個 agent,導致它下次需要時被重新載入。

# Load only the checkpoints this process serves, before accepting requests.
router = Router()
router.preload(["english", "multilingual"])

print(router.loaded)  # currently resident checkpoint names
router.unload("multilingual")

Router(preload=True) 預載入所有配置的 checkpoint。預載入會把常駐模型上限提高,以容納請求的集合。 要不預載入地控制一個三 checkpoint 的工作負載,設定 max_loaded=3。用 unload() 釋放一個 agent, 或用 router.unload() 釋放所有 agent。當作上下文管理器使用的路由器會在塊退出時解除安裝它的 agent:

with Router(preload=True) as router:
    result = router.predict(state, questions)

構造路由器時,你也可以傳 device="cpu"、device="cuda" 或其他受支援的 PyTorch 裝置。可用性和 記憶體決定哪些裝置能跑某個模型。

混合批次

predict_batch() 接受 state、模型、語言和問題 schema 各不相同的請求。路由器先為每個請求做一個 決策,按 checkpoint 和相容的問題 schema 把工作分組,然後把結果恢復到原來的輸入順序。

requests = [
    {"state": "Please refund the duplicate charge.", "questions": questions},
    {"state": "Mi cuenta fue cobrada dos veces.", "questions": questions},
    {"state": "A third request", "questions": questions, "model": "typed-decisions"},
]

results = router.predict_batch(requests, batch_size=8)

每個請求都需要 state 和 questions;它還可以帶 model、task、lang 或 lang_guess。共享 同一個 checkpoint 和問題 schema 的請求可以共享一次 Agent 批次前向傳播。不同的 schema 或 checkpoint 會在分開的組裡處理。batch_size 限制一次送給 Agent 的 state 數量;結果仍然與請求順序對應。

選擇入口點

  • 需要在載入模型之前檢查決策時,用 route() 或 route_batch()。
  • 單個請求用 predict(),多個可能異構的請求用 predict_batch()。
  • 當應用已經選好並載入了一個 checkpoint、不需要自動路由時,直接用 Agent。

建構子和方法細節見 Router API 參考。