路由

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 参考。