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 的常规默认值和已有的离线缓存。

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})

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 或 {} 表示不校验直接加载。

同样的拆分也提供给只用环境变量配置的进程：当 LAYA_SHA256_DIGESTS 里是按键到模型组织的映射（{"english": {...}, "multilingual": {...}}）时，会按 checkpoint 分别灌进去，于是一个常驻多个 checkpoint 的服务器可以给每个都钉上自己的摘要，而不是加载到第二个就拒绝启动。扁平的 LAYA_SHA256_DIGESTS 保持原来的含义，由 laya.revisions 应用到该进程加载的每一个 checkpoint 上 —— 这对只加载一个 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,
) -> 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,
) -> 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"),
}