路由
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 也不跑推理。這個
決策包含選中的模型、一段給人讀的理由,以及使用內建檢測時的語言檢測細節。
路由按這個順序檢查輸入:
model=直接選出一個 checkpoint。task=為指定任務選出一個 checkpoint。- 如果
auto_task_detection=True,與某個已知 typed-decisions 問題 ID 集合精確匹配時會選出typed-decisions。 - 一個被識別的
lang=值選出英語或多語言。 - 每次呼叫的
lang_guess=,或路由器配置的lang_guess,會被考慮。 - 內建的書寫系統與語言分析選出一個 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 參考。