Router

Router(
    models: Optional[Dict[str, str]] = None,
    device: Optional[str] = None,
    token: Optional[str] = None,
    revision: Optional[str] = None,
    revisions: Optional[Dict[str, Optional[str]]] = None,
    max_loaded: int = 2,
    default: str = "english",
    auto_task_detection: bool = False,
    standalone_repos: bool = False,
    preload: bool = False,
    lang_guess: Optional[Any] = None,
    hooks=None,
    on_predict_start=None,
    on_predict_end=None,
    hooks_raise: bool = True,
    hooks_concurrent: bool = True,
    hooks_timeout: Optional[float] = None,
    agent_kwargs: Optional[Dict[str, Any]] = None,
    sha256_digests: Optional[Dict[str, Optional[Dict[str, str]]]] = None,
)

按需載入 Laya 的 checkpoint，並把每個請求送到對的那個上。

from laya import Router

r = Router()
r.predict({"message": "Mein Konto wurde zweimal belastet"}, questions)   # -> multilingual
r.predict({"message": "I was charged twice"}, questions)                 # -> english
r.predict(state, questions, model="typed-decisions")                     # explicit

模型在第一次用到時才下載並構建。max_loaded 限制常駐的個數（超出時淘汰最久未用的），因為三個加起來約有 11.6 億參數。

預設值是 2，因為自動路由只會選 english 與 multilingual：上限設成 1 的話，每次語種切換都要重建剛剛淘汰掉的那個 checkpoint，而每次重建就是好幾秒 —— 正好落在 Router 存在的意義所在的那部分流量上。只用一種語言的流量永遠不會構建第二個 checkpoint，所以預設值對它沒有任何代價。記憶體吃緊的機器可以降到 1；當 auto_task_detection、顯式的 model= 或顯式的 task= 也可能用到 typed-decisions 時，把它提到 3（或者預載入）。

伺服器或演示場景應該改用預載入：冷載入要好幾秒，而語種檢測只要微秒，所以哪怕用預設值，某個語言第一次出現時仍然要付一次載入的代價。

r = Router(preload=True)                    # all three resident, routing is free
r = Router(preload=True, device="cuda")
r.preload(["english", "multilingual"])      # or just the two you serve

Hub 版本是可選的。revision 把同一個 commit 應用到所有模型；revisions={"english": "...", "multilingual": "..."} 則按模型覆蓋它，在獨立倉庫是在不同 commit 上審閱的時候很有用。兩者都不給時，用 huggingface_hub 的常規預設值和已有的離線快取。

revisions 裡為 None 或空白的條目表示「這個模型沒有覆蓋」，於是該模型繼承 revision —— 也就是 {"english": os.environ.get("EN_SHA")} 在變數未設定時寫出的形狀，這不能讓呼叫方失去自己確實要求過的那個固定。revisions 裡沒有任何東西能在 revision 固定其餘模型的同時把某一個解綁；想固定哪些模型，就把 revision 留空、直接點名。

空白的 revision 也以同樣的方式解讀，這是 Router 所傳遞內容的一處刻意改動：Router(revision=" ") 過去會到達 resolve_revision，那裡一個為真但空白的字串會抑制 LAYA_REVISION 回退並保留 huggingface_hub 的預設值；現在它在那之前就被丟棄，所以配置了空白的 Router 與什麼都沒配置的行為一致，$LAYA_REVISION 會生效。如果 revision 與 revisions 條目要以同樣的方式解讀，那這就是「這行配置從未填寫過」所必須有的含義。

這是較弱的來源最終壓過顯式參數的兩個地方之一。另一個是來自 LAYA_SHA256_DIGESTS 的按 checkpoint 摘要條目，它壓過通過 agent_kwargs 傳入的 expected_sha256；見類文件字串。

laya.Agent 接受的其它參數都可以通過 agent_kwargs 傳進來，它會併入 Router 構建的每一個 checkpoint：

Router(agent_kwargs={"lang_temperatures": {"de": {"temperature": [1.0, 1.4, 2.0]}}})
Router(agent_kwargs={"expected_sha256": {"model.safetensors": "a3f1..."}})
Router(agent_kwargs={"fast": True})

那裡的 expected_sha256 會在每個 checkpoint 上固定同樣的檔案，這正是單個常駐 checkpoint 或共享的 tokenizer.json 所需要的。它絕不會被某個 checkpoint 自帶的摘要整體丟棄：當某個 checkpoint 也有條目時（來自 sha256_digests 或來自按 checkpoint 的 LAYA_SHA256_DIGESTS），兩張對映會逐檔案合併，所以只有其中一方點名的檔案仍會被校驗。

有兩個例外，都是刻意的，也都有測試，因為「逐檔案合併」並非全部真相，而這個差別是一項供應鏈控制：

• 扁平的 LAYA_SHA256_DIGESTS —— {artifact: digest} 而不是 {model: {...}} —— 在這裡根本不算一層。verify_digests 會自己應用它，但只在其它東西都沒固定時（if expected is None），所以任何到達 Agent 的 expected_sha256，無論來自這裡還是來自某個 checkpoint 條目，都意味著那次載入不會去查這個扁平變數。在真實檔案上驗證過：一個固定 model.safetensors 的扁平變數加一張固定 tokenizer.json 的 agent_kwargs 對映，會載入一個被篡改的 model.safetensors。如果兩者都需要，請用按 checkpoint 的形狀，或者把所有在意的檔案都寫進同一張對映。這與 main 相比沒有變化。
• 顯式的 {} 或 None 條目會遮蔽本應生效的東西 —— 那正是「這一個不校驗地載入」所必須有的含義，test_an_explicit_none_entry_masks_a_flat_environment_map 固定了這一點。

而且優先順序是「按 checkpoint 壓過共享」，無論各自來自哪裡，所以從 LAYA_SHA256_DIGESTS 合成的按 checkpoint 條目，會壓過這裡在程式碼中傳入的 expected_sha256。環境變數壓過顯式參數，在一項這樣的控制上值得直說；test_an_environment_pin_overrides_the_shared_one_per_checkpoint 固定了這一點。

對於兩邊都點名的檔案，按 checkpoint 的條目勝出。兩者並不一樣具體：agent_kwargs 對映會到達 Router 構建的每一個 checkpoint，而 model.safetensors 是每個 checkpoint 都會用的、卻指向不同檔案的同一個名字，所以針對它的共享條目不可能同時是對所有 checkpoint 的正確說法。不會因為這個重疊而報錯 —— 拒絕它會連帶拒絕「共享固定 + 按 checkpoint 覆蓋」這一常規形狀，而這種形狀在這一切存在之前就能正確載入。如果你想知道某個 checkpoint 是針對哪個摘要校驗的，把它讀回來：交給每個 Agent 的對映就是上面描述的合併結果。

agent_kwargs 與 sha256_digests 都是公開且可變的，而某個 checkpoint 的條目是在載入時、而不是在構造時讀取的，所以之後才賦上的固定 —— 或者就地加進已有條目 —— 也算數。

Router 自己要用到的名字 —— model_id_or_path、device、token、subfolder、revision 以及那些 hook 參數 —— 在這裡會被拒絕，而不是被悄悄覆蓋；其餘的名字則在構造時對照 Agent.__init__ 檢查，所以寫錯的選項會在 Router(...) 那一行就失敗，而不是等到第一個請求。

產物摘要是可選的，而且始終按模型給：sha256_digests={"english": {...}} 把那個 {path relative to the checkpoint dir: hexdigest} 對映傳給負責載入它的那個 Agent，於是被篡改或調包過的權重檔案在解析之前就會被拒絕。revision 沒有 Router 級的等價物，因為摘要和 commit SHA 不同，是沒法共用的：捆綁倉庫為 english、multilingual 與 typed-decisions 各自帶了一個 model.safetensors，所以一張扁平的對映表最多隻能對上其中一個。把某個模型寫成 None 或 {} 不會為它自己加入任何檔案，因此會不校驗地載入它，除非 agent_kwargs["expected_sha256"] 把它固定住；它仍會遮蔽扁平的 LAYA_SHA256_DIGESTS —— 而這樣列出它正是為了這個。

同樣的拆分也提供給只用環境變數配置的程序：當 LAYA_SHA256_DIGESTS 裡是按鍵到模型組織的對映（{"english": {...}, "multilingual": {...}}）時，會按 checkpoint 分別灌進去，於是一個常駐多個 checkpoint 的伺服器可以給每個都釘上自己的摘要，而不是載入到第二個就拒絕啟動。扁平的 LAYA_SHA256_DIGESTS 保持原來的含義，由 laya.revisions 應用到該程序載入的每一個 checkpoint 上 —— 這對只加載一個 checkpoint 的程序是正確的。參數裡給了條目時，對該模型以參數為準，覆蓋環境變數。巢狀變數如果只點名了部分 checkpoint、沒點名其它，那它對其它 checkpoint 什麼也沒說：它們保留 agent_kwargs["expected_sha256"] 為它們固定的一切，因為從環境變數固定一個 checkpoint，並不是要求停止校驗其餘。

Hook 是可選的，執行在 Router 這一層：on_route 看到路由決定，on_load / on_evict 看到模型的生命週期，on_predict_start / on_predict_end 包住整個「路由 + 推理」呼叫。見 laya.hooks。

load(name: str)

返回 name 對應的 Agent，第一次用到時下載並構建它。

併發呼叫方共享同一個 Agent，不會重複構建。

attach(name: str, agent: Any)

把一個已經構建好的 Agent 註冊到 name 下，而不是再載入一份。

當程序出於其它原因已經載入了某個 checkpoint 時很有用：一個已經構建過 convaiinnovations/laya 的演示，可以直接把它交給 router，而不必再為一個 421M 參數的副本付出代價 —— 以及記憶體。

preload(names: Optional[List[str]] = None)

提前下載並構建 checkpoint，讓任何請求都不必再付模型載入的代價。

冷載入要好幾秒，而語種檢測只要微秒。所有 checkpoint 都常駐之後，路由就幾乎是免費的 —— 這正是伺服器或演示場景想要的。max_loaded 會被提高到同時容納所請求的 checkpoint 和所有已經常駐的 agent，所以增量預載入不會把任何一方擠出去。

unload(name: Optional[str] = None)

釋放一個模型，或全部釋放。

loaded_revisions: Dict[str, Optional[str]]

每個常駐 agent 是從哪個 commit SHA 載入的（本地路徑為 None）。

route(
    state: Union[str, dict, list, None],
    questions: Optional[Dict[str, Any]] = None,
    model: Optional[str] = None,
    task: Optional[str] = None,
    lang: Optional[str] = None,
    lang_guess: Optional[Any] = None,
    hooks=None,
    hooks_raise: Optional[bool] = None,
    hooks_timeout: Optional[float] = None,
) -> RouteDecision

決定用哪個 checkpoint，然後讓 on_route 鉤子觀察或替換這個決定。

ctx.decision 就是那個 RouteDecision；鉤子可以替換它（例如釘住某個 checkpoint），替換後的那個才會被返回並使用。hooks 是每次呼叫的鉤子，追加在 Router 上已安裝的鉤子之後。

predict(
    state: Union[str, dict, list],
    questions: Dict[str, Any],
    model: Optional[str] = None,
    task: Optional[str] = None,
    lang: Optional[str] = None,
    lang_guess: Optional[Any] = None,
    hooks=None,
    on_predict_start=None,
    on_predict_end=None,
    hooks_raise: Optional[bool] = None,
    hooks_timeout: Optional[float] = None,
    max_len: Optional[int] = None,
    head_max_len: Optional[int] = None,
    min_confidence: Optional[float] = None,
) -> Dict[str, Any]

先路由，再用選中的 checkpoint 一次前向傳播回答所有問題。

結果就是通常的 system_one 載荷，外加一個記錄本次決定的 routing 鍵。Router 級的 on_predict_start / on_predict_end 鉤子包住整個「路由 + 推理」呼叫，並能看到 ctx.decision；見 laya.hooks。max_len / head_max_len 覆蓋本次呼叫中 agent 的 token 預算（起始鉤子也可以設定 ctx.max_len / ctx.head_max_len）。

predict_long(
    state: Union[str, dict, list],
    questions: Dict[str, Any],
    model: Optional[str] = None,
    task: Optional[str] = None,
    lang: Optional[str] = None,
    lang_guess: Optional[Any] = None,
    window: Optional[int] = None,
    stride: Optional[int] = None,
    aggregate: str = "auto",
    batch_size: Optional[int] = None,
    hooks=None,
    on_predict_start=None,
    on_predict_end=None,
    hooks_raise: Optional[bool] = None,
    hooks_timeout: Optional[float] = None,
) -> Dict[str, Any]

先路由，然後掃描狀態的每一個視窗，而不只是第一個。

predict 只用一個視窗給狀態打分：超出 max_len 的部分會被截掉（對話列表取最後一個視窗，其餘取第一個），永遠到不了模型。這個方法的路由行為與 predict 完全一致 —— 同樣的 model/task/lang 提示、同樣的 Router 級鉤子、同樣的 routing 鍵與 usage —— 只是用所路由到的那個 agent 的 predict_long 給狀態打分，後者把狀態切成互相重疊的視窗並按問題聚合。聚合規則就是 laya.agent.Agent.predict_long 的那套：noul 取最強的視窗，choice/score 取最自信的那個。

每次呼叫的鉤子（hooks、on_predict_start、on_predict_end、hooks_raise、hooks_timeout）包住整個「路由 + 掃描」過程，與它們包住 predict 的方式完全相同：掃描跑在最後，所以直接作答（ctx.skip(...)）或改寫狀態的起始鉤子會生效。這裡不接受 max_len / head_max_len —— 視窗大小由 window 或 checkpoint 的預算決定，而覆蓋單視窗截斷這件事正是 predict_long 本身的用途。

decide(
    state: Union[str, dict, list],
    schema: Any = None,
    questions: Optional[Dict[str, Any]] = None,
    return_details: bool = False,
    min_confidence: Optional[float] = None,
    predict_kwargs,
) -> Any

把 state 對照一個 schema（JSON schema 或 pydantic 模型）作答，返回型別化的值。

見 laya.structured。schema 與 questions 只能給一個；額外的關鍵字參數（例如 model=、task=、hooks=）會透傳給 predict。

decide_batch(
    states: Sequence[Any],
    schema: Any = None,
    questions: Optional[Dict[str, Any]] = None,
    return_details: bool = False,
    min_confidence: Optional[float] = None,
    predict_kwargs,
) -> List[Any]

一次性把多個 state 對照同一個 schema（JSON schema 或 pydantic 模型）作答。

這是 :meth:decide 的吞吐版本：schema 只規劃一次，它的問題通過 :meth:predict_batch 跑遍每一個 state（分組的前向傳播，結果保持輸入順序），然後像 decide 那樣把每個 state 的答案投影出來。額外的關鍵字參數（batch_size=、model=、hooks= 等）會透傳給 predict_batch。見 laya.structured。

route_batch(
    requests: Sequence[Dict[str, Any]],
    hooks_timeout: Optional[float] = None,
    hooks=None,
    hooks_raise: Optional[bool] = None,
) -> List[RouteDecision]

對一批異構請求做路由，不載入任何 checkpoint。

每個請求是一個對映，含 state 與 questions，外加 :meth:route 接受的那幾個可選路由覆蓋項：model、task、lang 與 lang_guess。返回的決定保持輸入順序。

這一步刻意與推理分開，好讓呼叫方在付出模型載入代價之前先檢查或聚合路由決定。

predict_batch(
    requests: Sequence[Dict[str, Any]],
    batch_size: Optional[int] = None,
    hooks_timeout: Optional[float] = None,
    min_confidence: Optional[float] = None,
    sort_by_length: bool = False,
    hooks=None,
    on_predict_start=None,
    on_predict_end=None,
    hooks_raise: Optional[bool] = None,
) -> List[Dict[str, Any]]

以儘量少的模型切換，對一批異構請求做路由並執行。

請求先被路由、按 checkpoint 分組。在每個 checkpoint 內，問題 schema 相同的請求會交給 Agent.predict_batch，讓它們的狀態共用前向傳播。最後把結果恢復到請求原本的順序。

各請求可以各自指定 model、task、lang、lang_guess、max_len 或 head_max_len，也可以使用不同的問題 schema。max_len / head_max_len 是 token 預算覆蓋項的按請求形式 —— predict 是把它們當呼叫參數傳的：它們為那一個請求設定該 checkpoint 的狀態預算與問題頭預算，於是可以提出一個很寬的問題，而不必把這一批裡其它請求也壓到同一個視窗。要求不同預算的請求會被拆進不同的前向傳播，因為一次 Agent.predict_batch 呼叫對它所有的狀態只帶一個預算。起始鉤子仍然可以在 ctx 上替換這兩個值。

Router 級預測鉤子按請求逐個執行，與 predict 一致：每個請求有自己的 PredictContext，所以 on_predict_start 可以替換該請求的狀態、問題或 token 預算，或者用 ctx.skip(...) 跳過它，而 on_predict_end 能看到並替換它的結果。請求是在各自的起始鉤子跑完之後才分組做前向傳播的，而同一個 checkpoint 組內請求的結束順序與它們開始的順序相反。如果某個 checkpoint 組失敗，該組裡所有起始鉤子已經跑過的請求都會隨異常一起失敗 —— 命中快取的也包括在內：每個請求都會先收到 on_error、再收到 on_predict_end，然後異常才繼續往上拋。

RouteDecision()

路由的結果：選了哪個模型、為什麼、以及探測到了什麼。

表現得像一個 dict，所以可以直接序列化進 API 響應。

DEFAULT_MODELS = {
    "english": (BUNDLE_REPO, None),
    "multilingual": (BUNDLE_REPO, "multilingual"),
    "typed-decisions": (BUNDLE_REPO, "typed-decisions"),
}