Agent

laya.Agent 加载一个 checkpoint，回答关于某个状态的类型化问题。laya.load 是 Agent(...) 的快捷方式，laya.RLAgent 是 Agent 的别名。ONNXAgent 在 CPU 上运行导出的 ONNX 模型；从 laya.onnx_agent 导入它。

名称、类型、默认值与代码保持英文；其余为译文（尚未翻译的条目暂显示英文原文）。

Agent

Agent(
    model_id_or_path: str = "convaiinnovations/laya",
    device: Optional[str] = None,
    token: Optional[str] = None,
    subfolder: Optional[str] = None,
    fast: bool = False,
    compile: bool = False,
    revision: Optional[str] = None,
    expected_sha256: Optional[Dict[str, str]] = None,
    lang_temperatures: Optional[Dict[str, Dict[str, 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,
    calibration: Optional[str] = None,
)

基类: HookRegistry

System 1 决策模型运行时：快速、非自回归、经过校准的决策。

dtype 是自动转换的目标类型，不是每次调用的精度。在 MPS 上，只有行数达到 mps_amp_min_rows 时调用才会自动转换，所以 dtype 可以是 float16，而某次调用实际以 float32 运行。dtype_for(rows) 返回行数为 rows 的调用所用的精度。

加载一个 Laya checkpoint。

revision 可选地把 Hub 下载固定到某个明确的 commit SHA/分支/tag；不给时，用 huggingface_hub 的常规默认值和已有的离线缓存。expected_sha256（{相对 checkpoint 目录的路径: 十六进制摘要}）在任何权重被解析或执行之前校验产物完整性；它是可选的，对本地目录同样生效。产物缺失会抛 FileNotFoundError，摘要不匹配会抛 ValueError；两者都会拒绝这次加载。

fast=True 把编码器/决策头的前向换成 TileLang 快速路径（仅 CUDA，需要 pip install laya[fast]）；见 Agent.accelerate。

compile=True 让模型跑在 torch.compile 下，并打开 ModernBERT 编码器的 reference_compile。torch.compile 会按输入形状特化，而 Laya 几乎每个请求都会遇到新形状，所以那些图通常是入不敷出；流量重复时才用它。fast=True 优先，因为 TileLang 路径替换掉的正是将要被编译的那个前向。

subfolder 从捆绑了多个 checkpoint 的仓库里挑一个，例如 Agent("convaiinnovations/laya", subfolder="multilingual")。只会下载那个子目录，所以捆绑不会让每个用户都付出整个家族的代价。

calibration 是可选的 JSON 路径，含 temperature 与 temperature_by_options。它在 checkpoint 配置之后应用，因此拟合出来的映射可以覆盖随包发布的标量，而不必重写 model.safetensors。

hooks / on_predict_start / on_predict_end 观察或塑造每一次预测；见 laya.hooks。hooks_raise=False 时钩子失败只警告并继续，hooks_concurrent=False 会把不适合并行运行的钩子串行化，hooks_timeout 以秒为单位限制每次钩子调用（None 表示不限制）。

参数

model_id_or_pathstr= "convaiinnovations/laya"
deviceOptional[str]= None
tokenOptional[str]= None
subfolderOptional[str]= None
fastbool= False
compilebool= False
revisionOptional[str]= None
expected_sha256Optional[Dict[str, str]]= None
lang_temperaturesOptional[Dict[str, Dict[str, Any]]]= None
hooks= None
on_predict_start= None
on_predict_end= None
hooks_raisebool= True
hooks_concurrentbool= True
hooks_timeoutOptional[float]= None
calibrationOptional[str]= None

accelerate

accelerate(use_graphs: bool = True, strict: bool = False)

把模型的前向换成 TileLang 快速路径（融合的 GEMM/GEGLU/LayerNorm/RoPE 内核、滑窗 flash attention、16 位常驻权重、按形状桶构建的 CUDA graph）。

快速路径按调用当时 agent 的自动转换 dtype 运行（bf16 或 fp16），所以它与被它替换掉的原始前向在舍入误差内一致（见 benchmarks/parity_fast.py）。改过 agent.dtype 之后，先调用 deaccelerate() 再调用 accelerate() 来重建它。启用成功返回 True。strict=False 时任何失败（没有 CUDA、缺 tilelang）都会保留原始路径。

参数

use_graphsbool= True
strictbool= False

warmup

warmup(shapes=None) -> float

立刻用每种形状的合成输入跑一遍前向，返回花掉的秒数。

compile=True 会在第一个需要 graph 的请求上做追踪与编译（在 GPU 上是几十秒），而 fast=True 会在首次使用时按形状桶构建它的内核与 CUDA graph。在加载之后、开始服务之前调用它，就能把这份代价从最初的几个请求里挪走。用原始前向时，这只是几次普通的前向传播。shapes 是 (rows, tokens, markers) 的列表；tokens 会被限制在 agent 的 max_len 以内。不会向任何调用方返回或记录任何东西，钩子也不会运行。

参数

shapes= None

deaccelerate

deaccelerate()

恢复原始前向。

dtype_for

dtype_for(rows: int) -> torch.dtype

行数为 rows 个问题行的一次前向所用的精度。

dtype 是自动转换的目标类型，在加载时设定一次。一次前向是否做自动转换是按调用决定的：在 MPS 上只有行数达到 mps_amp_min_rows 才会。行数为 rows 的前向会做自动转换时返回 dtype，否则返回 torch.float32。一次 predict 调用每个问题占一行。

参数

rowsint

predict_batch

predict_batch(
    states: List[Union[str, dict, list]],
    questions: Dict[str, Dict[str, Any]],
    batch_size: Optional[int] = None,
    lang: Optional[str] = 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,
    sort_by_length: bool = False,
    min_confidence: Optional[float] = None,
) -> List[Dict[str, Any]]

对多个状态评估同一组问题，把它们打包进共享的前向传播。

这是吞吐路径。system_one/predict 每次前向只处理一个状态；在 GPU 上那会让批次维度大部分空着。predict_batch 把多个状态的问题行归并进同一个张量，于是一次本来要 N 次串行前向的调用只需一次（或 ceil(len(states) / batch_size) 次），在 GPU 上每次决策快好几倍。

参数

statesList[Union[str, dict, list]]: 状态列表（每一项是文本字符串、JSON 字典或对话轮次列表）。同一组 questions 会对照每一个状态评估。
questionsDict[str, Dict[str, Any]]: 问题定义，与 system_one 接受的完全一致。
batch_sizeOptional[int]= None: 每次前向传播中状态数的可选上限。None 表示一次全发；批量处理很多或很长的状态时，设上它可以给峰值内存封顶。
langOptional[str]= None
hooksHookArg= None: 每次调用的钩子，追加在 Agent 上已安装的钩子之后。见 laya.hooks。
on_predict_startPredictHookArg= None: 每次调用的起始钩子。它可以改写状态/问题，或调用 ctx.skip(...) 短路推理。
on_predict_endPredictHookArg= None: 每次调用的结束钩子。它可以改写结果。
hooks_raiseOptional[bool]= None: 覆盖本次调用中 Agent 的 hooks_raise。
hooks_timeoutOptional[float]= None: 覆盖本次调用中 Agent 的 hooks_timeout。
max_lenOptional[int]= None: 覆盖本次调用中 agent 配置的 max_len。起始钩子也可以设置 ctx.max_len 来塑造 token 预算。
head_max_lenOptional[int]= None: 覆盖本次调用中 agent 配置的 head_max_len。起始钩子也可以设置 ctx.head_max_len。
sort_by_lengthbool= False: 把编码后长度相近的状态按「八个批次」为一组归到一起，以减少 padding。需要显式给出大于 1、且小于状态数的 batch_size；否则没有任何效果。结果保持输入顺序。这样会缓存最多八个批次的已分词状态，而不是一个。改变批次形状可能轻微改变浮点预测结果。
min_confidenceOptional[float]= None

返回值

每个状态一个结果字典的列表，形状与 system_one 的输出完全相同，并按下标与 states 对齐。

predict_long

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

对超出上下文窗口的状态评估问题：用互相重叠的窗口扫描它，并按问题聚合。

system_one/predict 会把超出 max_len 的状态截断到一个窗口（第一个；对话列表则取最后一个），其余部分被静默丢弃。predict_long 只把状态分词一次，切成互相重叠的 token 窗口，在共享的前向传播里给每个窗口打分（经由 predict_batch），再把各窗口的答案合并起来：

noul -> P(true) 取各窗口的最大值（只要有一个窗口支持，该陈述就成立）
choice-> 取单个最自信窗口的答案，这样局部信号不会被长文档里占绝大多数的中性窗口投票压过去（取平均会把它淹没 —— 中性多数说了算）
score -> 同样取最自信窗口的档位

返回的概率/置信度是作出决定的那个窗口的，不是针对整篇文档校准过的数字：没有任何信号时，noul 在多窗口上取最大值也会随窗口数往上漂；而当文档里没有任何决定性的内容时，choice 可能落在一个自信地中性的窗口上。因此每个答案都带着 answer["window"] —— 作出决定的窗口的 index、在分词后状态中的 token_start/token_end，以及窗口总数 count —— 这样调用方可以查看答案来自哪一段，而不是盲信那个原始数字。

已经装得进一个窗口的状态会直接交给 system_one（输出完全相同）。

钩子包住的是「为该状态作答的那次推理」——对一个需要多个窗口的文档来说，就是那次统一的、覆盖所有窗口的 predict_batch：on_predict_start 只触发一次，而 ctx.states 里按扫描顺序放着解码后的窗口文本 —— 不是调用方传入的 state，后者已被分词用以产生这些窗口。根据这条链最后留下什么，有三种结果：

ctx.skip([result]) 直接为文档作答：载荷回来时没有窗口归属信息，usage["windows"] 为 0，因为什么都没打分
保持本方法构建出来的扫描：每个窗口都被打分，每个答案都带 answer["window"]，usage["windows"] 就是窗口数
被改写过的扫描（ctx.states 以任何方式被替换）：答案按实际打过分的那些状态聚合，但没有任何答案带 answer["window"] —— 上面那些偏移描述的是本方法的窗口，不是模型读到的文本

参数

stateUnion[str, dict, list]
questionsDict[str, Dict[str, Any]]
windowOptional[int]= None: 每个窗口容纳的状态 token 数。默认取按问题计的状态预算（max_len - head_max_len - 8）—— 也就是每个问题都放得下的最大值。窗口更小能更好地隔离局部信号（一段很短的决定性文本在它所在窗口里占比更大，那个窗口就能清楚地把它分类出来），代价是窗口数更多；默认的大窗口偏向上下文与吞吐。noul 对此不敏感；而当决定性的一段只占一份长文档很小的一部分、其余都是中性内容时，choice/score 会受益于更小的窗口。
strideOptional[int]= None: 窗口之间的 token 步长。默认是 window // 2（50% 重叠），这样靠近边界的片段仍然能完整落进某个窗口。
aggregatestr= "auto": "auto"（即上面那套按类型的规则）是目前唯一的模式。
batch_sizeOptional[int]= None: 每次前向传播的窗口数上限，用来给超长状态的内存用量封顶。
langOptional[str]= None: 按语言选择温度，与 system_one 一致。
hooksHookArg= None: 每次调用的钩子，追加在 Agent 上已安装的钩子之后。见 laya.hooks。
on_predict_startPredictHookArg= None: 每次调用的起始钩子，与 system_one 一致。
on_predict_endPredictHookArg= None: 每次调用的结束钩子，与 system_one 一致。
hooks_raiseOptional[bool]= None: 覆盖本次调用中 Agent 的 hooks_raise。
hooks_timeoutOptional[float]= None: 覆盖本次调用中 Agent 的 hooks_timeout。

返回单个结果字典，形状与 system_one 相同，并加上 usage["windows"]。这个键始终存在，计的是模型为得出答案而打分的窗口数：装得进一个窗口的状态是 1；用 N 个重叠窗口扫描的文档是 N（起始钩子改写后就是它改写成的那个 N）；起始钩子直接为文档作答、或者在任何窗口被读取之前就没有留下要打分的状态时是 0 —— 两条路径都是如此，所以缓存下来的答案永远不会被读成「模型读过的窗口」。

跨多个窗口时，截断相关的键与其它 usage 字段一样合并：truncated、state_tokens 与 state_tokens_dropped 求和（因此 truncated 是被截断的窗口数，而 token 计数里包含重叠部分），truncated_questions 则是最后一个窗口的列表。两者可以不一致：只有靠前的某个窗口被截断时，truncated 大于 0 而 truncated_questions 是空的。当某个窗口大于该问题决策头留出的空间时，它就被截断 —— 这来自高于默认值的 window，或来自收窄了 max_len / head_max_len 的起始钩子。这里要判断的是 usage["truncated"] > 0，而不是 is True。

system_one

system_one(
    state: Union[str, dict, list],
    questions: Dict[str, Dict[str, Any]],
    lang: Optional[str] = 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]

在一次并行的前向传播里，对状态评估一组类型化问题。

参数

stateUnion[str, dict, list]

文本字符串、JSON 字典或对话轮次列表。

questionsDict[str, Dict[str, Any]]

把 question_id 映射到问题定义的字典。

choice: {"type": "choice", "instructions": "...", "criteria": {"optA": "...", ...}}
score: {"type": "score", "instructions": "...", "criteria": ["lvl0", "lvl1", ...]}
noul: {"type": "noul", "instructions": "...", "criteria": {"false": "...", "true": "..."}, "labels": {"false": "B", "true": "A"}}

Noul 的 criteria 与 labels 都是可选的。labels 只控制展示给模型的文字；它们的键仍然保持 false/true 的语义，返回的 noul 值永远表示 P(true)。为兼容起见，labels 默认为 false/true。

langOptional[str]= None

hooks= None

on_predict_start= None

on_predict_end= None

hooks_raiseOptional[bool]= None

hooks_timeoutOptional[float]= None

max_lenOptional[int]= None

head_max_lenOptional[int]= None

min_confidenceOptional[float]= None

返回值

含答案、概率、校准后的置信度与 token 用量的字典。问题为空时返回空答案与零 token 用量，不做分词也不做模型前向。

当决策头预算让两个选项落在同一段 token 上时，usage 会为碰巧如此的每个问题带一个 options 条目 —— total、distinct 与 tokens_per_option —— 因为在 58 段可区分的跨度里只挑得出 42 个时，这个天花板是预算造成的，不是模型造成的。选项全部存活的问题不会出现，所以什么都没被压掉的请求与原来完全一样。

usage 还会报告状态是否装得下：truncated、state_tokens、state_tokens_dropped，以及 truncated_questions（决策头留下的空间太少的那几个问题）。关心答案是否看到了整个状态的调用方，应该读 usage["truncated"]，而不是拿自己发出去的内容长度去估算。

要一次性给很多状态打分，见 predict_batch，它让这些状态共用前向传播。

decide

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 只能给一个；额外的关键字参数会透传给 predict / system_one。

参数

stateUnion[str, dict, list]
schemaAny= None
questionsOptional[Dict[str, Any]]= None
return_detailsbool= False
min_confidenceOptional[float]= None
predict_kwargs

decide_batch

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

一次性把多个状态对照同一个 schema（JSON schema 或 pydantic 模型）作答。

这是 :meth:decide 的吞吐版本：schema 只规划一次，它的问题通过 :meth:predict_batch 跑遍每一个状态（共享的前向传播，结果保持输入顺序），然后像 decide 那样把每个状态的答案投影出来。额外的关键字参数（batch_size=、lang=、hooks= 等）会透传给 predict_batch。见 laya.structured。

参数

statesList[Union[str, dict, list]]
schemaAny= None
questionsOptional[Dict[str, Any]]= None
return_detailsbool= False
min_confidenceOptional[float]= None
predict_kwargs

fit_temperatures

fit_temperatures(records, compute_ece: bool = False, seed: int = 0) -> Dict[str, Any]

用 CPU 上的记录拟合逐桶温度，并存到这个 agent 上。

records 是 (qtype, logits, target, k)。手上有带标签的前向记录时，用 laya.calibrate.records_from_labeled 构造它们；这个方法不下载权重，也不写 model.safetensors。seed 只在 compute_ece 为真时影响留出的 ECE 切分。checkpoint 的 cfg 保持加载时的样子。

参数

records
compute_ecebool= False
seedint= 0

save_calibration

save_calibration(path: str) -> None

写出温度，以及它们是为哪个 checkpoint 拟合的。不写权重。

参数

pathstr

load_calibration

load_calibration(path: str) -> None

把 save_calibration 写出的 JSON 映射读进这个 agent。

没有 version 的文件会被当作版本 1，仍然能加载。版本更新、但记录的 checkpoint 与这个 agent 对不上的文件会给出警告，也仍然加载。不是数字、或落在 [TEMP_MIN, TEMP_MAX] 之外的值，会像加载 checkpoint 时那样用 clamp_temperature 夹住。

参数

pathstr

load

load(
    model_id_or_path: str = "convaiinnovations/laya",
    device: Optional[str] = None,
    token: Optional[str] = None,
    subfolder: Optional[str] = None,
    fast: bool = False,
    compile: bool = False,
    revision: Optional[str] = None,
    expected_sha256: Optional[Dict[str, str]] = None,
    lang_temperatures: Optional[Dict[str, Dict[str, 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,
    calibration: Optional[str] = None,
) -> Agent

加载一个 Laya agent。

subfolder 从捆绑了多个 checkpoint 的仓库里挑一个：

laya.load("convaiinnovations/laya")                           # English (repo root)
laya.load("convaiinnovations/laya", subfolder="multilingual")
laya.load("convaiinnovations/laya", fast=True)                # TileLang GPU fast path
laya.load("convaiinnovations/laya", compile=True)              # torch.compile the model

revision/expected_sha256 会固定并校验下载下来的产物；见 Agent。hooks / on_predict_start / on_predict_end 观察或塑造每一次预测；见 laya.hooks。calibration 就是 Agent 接受的那个可选 JSON 路径。

参数

model_id_or_pathstr= "convaiinnovations/laya"
deviceOptional[str]= None
tokenOptional[str]= None
subfolderOptional[str]= None
fastbool= False
compilebool= False
revisionOptional[str]= None
expected_sha256Optional[Dict[str, str]]= None
lang_temperaturesOptional[Dict[str, Dict[str, Any]]]= None
hooks= None
on_predict_start= None
on_predict_end= None
hooks_raisebool= True
hooks_concurrentbool= True
hooks_timeoutOptional[float]= None
calibrationOptional[str]= None

ONNXAgent

ONNXAgent(
    model_id_or_path: str,
    onnx_path: str = "laya.onnx",
    token: Optional[str] = None,
    subfolder: Optional[str] = None,
    revision: Optional[str] = None,
    expected_sha256: Optional[Dict[str, str]] = None,
    hooks=None,
    on_predict_start=None,
    on_predict_end=None,
    hooks_raise: bool = True,
    hooks_concurrent: bool = True,
    hooks_timeout: Optional[float] = None,
    lang_temperatures: Optional[Dict[str, Dict[str, Any]]] = None,
    calibration: Optional[str] = None,
)

基类: HookRegistry

通过 ONNX 运行的 System 1 决策模型：针对 CPU 优化的快速决策。

加载一个由 ONNX Runtime 支撑的 Laya agent。

参数

model_id_or_pathstr: HuggingFace Hub ID，或原始 PyTorch checkpoint 的本地路径（用来加载分词器与配置）。
onnx_pathstr= "laya.onnx": 导出的 .onnx 文件的路径。
tokenOptional[str]= None: 私有或受限 checkpoint 用的可选 HuggingFace token；不给时退回 $HF_TOKEN，与 Agent 完全一致。只会去取分词器与配置 —— 计算图本身就是本地的 onnx_path。
subfolderOptional[str]= None: 从捆绑仓库下载时的可选子目录。
revisionOptional[str]= None: 可选的 Hub 版本（commit SHA/分支/tag）。不给时，用 huggingface_hub 的常规默认值和已有的离线缓存。
expected_sha256Optional[Dict[str, str]]= None: 可选的 {相对 checkpoint 目录的路径: 十六进制摘要}，在任何 checkpoint 文件被解析之前校验；它是可选的，对本地目录同样生效。产物缺失会抛 FileNotFoundError，摘要不匹配会抛 ValueError；两者都会拒绝这次加载。
hooksHookArg= None: 可选的预测钩子；见 laya.hooks。
on_predict_startPredictHookArg= None: 可选的起始钩子，在推理之前运行。
on_predict_endPredictHookArg= None: 可选的结束钩子，在推理之后运行。
hooks_raisebool= True: 为 False 时，钩子失败只警告，推理继续。
hooks_concurrentbool= True: 为 False 时，钩子用锁串行化。
hooks_timeoutOptional[float]= None: 以秒为单位限制每次钩子调用；None 表示不限制。
lang_temperaturesOptional[Dict[str, Dict[str, Any]]]= None: 可选的按语言温度覆盖，按语言代码索引，每项是 {"temperature": [3 floats], "temperature_by_options": {}}。当 system_one/predict 收到 lang= 时应用，与 PyTorch 的 Agent 对齐；否则跨后端切换会丢掉校准。
calibrationOptional[str]= None

load_calibration

load_calibration(path: str) -> None

把 save_calibration 写出的 JSON 映射读进这个 agent。

参数

pathstr

system_one

system_one(
    state: Union[str, dict, list],
    questions: Dict[str, Dict[str, Any]],
    lang: Optional[str] = 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]

在一次 ONNX Runtime 会话运行里，对状态评估一组类型化问题。

lang 选择按语言的温度覆盖（见 lang_temperatures），与 PyTorch Agent.system_one 的签名一致，因此两个后端可以互换使用。

它是基于 predict_batch 定义的，与 PyTorch 的 Agent.system_one 完全一样，这样单状态路径与批量路径就不会各自漂移。

参数

stateUnion[str, dict, list]: 文本字符串、JSON 字典或对话轮次列表。
questionsDict[str, Dict[str, Any]]: 问题定义，形状与 Agent.system_one 接受的一致。
langOptional[str]= None: 按语言的温度覆盖（见 lang_temperatures）。
hooksHookArg= None: 每次调用的钩子，追加在 agent 上已安装的钩子之后。
on_predict_startPredictHookArg= None: 每次调用的起始钩子。它可以改写状态/问题，或调用 ctx.skip(...) 短路推理。
on_predict_endPredictHookArg= None: 每次调用的结束钩子。它可以改写结果。
hooks_raiseOptional[bool]= None: 覆盖本次调用中 agent 的 hooks_raise。
hooks_timeoutOptional[float]= None: 覆盖本次调用中 agent 的 hooks_timeout。
max_lenOptional[int]= None: 覆盖本次调用中配置的 max_len。
head_max_lenOptional[int]= None: 覆盖本次调用中配置的 head_max_len。
min_confidenceOptional[float]= None: 可选的弃权阈值，作用在 answer_confidence 上（#361）；低于它的答案会带上 low_confidence: True 返回。

返回值

含答案、概率、校准后的置信度与 token 用量的字典。

要一次性给很多状态打分，见 predict_batch，它让这些状态共用会话运行。

predict_batch

predict_batch(
    states: List[Union[str, dict, list]],
    questions: Dict[str, Dict[str, Any]],
    batch_size: Optional[int] = None,
    lang: Optional[str] = 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,
    sort_by_length: bool = False,
    min_confidence: Optional[float] = None,
) -> List[Dict[str, Any]]

对多个状态评估同一组问题，共用 ONNX Runtime 的会话运行。

这是吞吐路径，与 laya.agent.Agent.predict_batch 对应：system_one 每次会话运行只归并一个状态的问题行，所以 N 个状态要 N 次运行。predict_batch 把多个状态的行归并进一次运行 —— 或 ceil(len(states) / batch_size) 次 —— 这正是 ONNX Runtime 自身的并行在 CPU 上发挥价值的地方。

参数

statesList[Union[str, dict, list]]: 状态列表（每一项是文本字符串、JSON 字典或对话轮次列表）。同一组 questions 会对照每一个状态评估。
questionsDict[str, Dict[str, Any]]: 问题定义，与 system_one 接受的完全一致。
batch_sizeOptional[int]= None: 每次会话运行中状态数的可选上限。None 表示一次全发；批量处理很多或很长的状态时，设上它可以给峰值内存封顶。
langOptional[str]= None: 应用到每一个状态的按语言温度覆盖；见 lang_temperatures。
hooksHookArg= None: 每次调用的钩子，追加在 agent 上已安装的钩子之后。
on_predict_startPredictHookArg= None: 每次调用的起始钩子。它可以改写状态/问题，或调用 ctx.skip(...) 短路推理。
on_predict_endPredictHookArg= None: 每次调用的结束钩子。它可以改写结果。
hooks_raiseOptional[bool]= None: 覆盖本次调用中 agent 的 hooks_raise。
hooks_timeoutOptional[float]= None: 覆盖本次调用中 agent 的 hooks_timeout。
max_lenOptional[int]= None: 覆盖本次调用中配置的 max_len。
head_max_lenOptional[int]= None: 覆盖本次调用中配置的 head_max_len。
sort_by_lengthbool= False: 把编码后长度相近的状态按「八个批次」为一组归到一起以减少 padding，与 Agent.predict_batch 完全一样。需要显式给出大于 1、且小于状态数的 batch_size；否则没有任何效果。结果保持输入顺序。改变批次形状可能让决策阈值附近的浮点预测结果轻微变化。
min_confidenceOptional[float]= None: 可选的弃权阈值，作用在 answer_confidence 上（#361）；低于它的答案会带上 low_confidence: True 返回。

返回值

每个状态一个结果字典的列表，形状与 system_one 的输出完全相同，并按下标与 states 对齐。

predict_long

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

对超出上下文窗口的状态评估问题：用互相重叠的窗口扫描它，并按问题聚合。

这是 laya.agent.Agent.predict_long 的 ONNX 版本，聚合规则相同：system_one 会把超出 max_len 的状态截断到一个窗口，其余部分被静默丢弃。predict_long 只把状态分词一次，切成互相重叠的 token 窗口，通过 predict_batch 给每个窗口打分 —— 于是这些窗口共用 ONNX Runtime 的会话运行，而不是各付一次 —— 再把各窗口的答案合并起来：

noul -> P(true) 取各窗口的最大值（只要有一个窗口支持，该陈述就成立）
choice-> 取单个最自信窗口的答案，这样局部信号不会被长文档里占绝大多数的中性窗口投票压过去
score -> 同样取最自信窗口的档位

返回的概率/置信度是作出决定的那个窗口的，不是针对整篇文档校准过的数字，理由与 PyTorch 那边的说明相同。每个答案都带着 answer["window"] —— 作出决定的窗口的 index、在分词后状态中的 token_start/token_end，以及窗口总数 count。

已经装得进一个窗口的状态会直接交给 system_one（输出完全相同）。

参数

stateUnion[str, dict, list]: 文本字符串、JSON 字典或对话轮次列表。
questionsDict[str, Dict[str, Any]]: 问题定义，与 system_one 接受的完全一致。
windowOptional[int]= None: 每个窗口容纳的状态 token 数。默认取按问题计的状态预算（max_len - head_max_len - 8）。与 Agent.predict_long 一样，窗口更小能更好地隔离局部信号，代价是窗口数更多。
strideOptional[int]= None: 窗口之间的 token 步长。默认是 window // 2（50% 重叠）。
aggregatestr= "auto": "auto"（即上面那套按类型的规则）是目前唯一的模式。
batch_sizeOptional[int]= None: 每次会话运行的窗口数上限，用来给超长状态的内存用量封顶。
langOptional[str]= None: 按语言选择温度，与 system_one 一致。
hooksHookArg= None: 每次调用的钩子，追加在 agent 上已安装的钩子之后。它们遵循 Agent.predict_long 的约定：包住为状态作答的那次推理；直接作答的起始钩子（ctx.skip(...)）会得到 usage["windows"] == 0 且没有窗口归属信息；被改写过的扫描在聚合时不带 answer["window"]。
on_predict_startPredictHookArg= None: 每次调用的起始钩子，与 system_one 一致。
on_predict_endPredictHookArg= None: 每次调用的结束钩子，与 system_one 一致。
hooks_raiseOptional[bool]= None: 覆盖本次调用中 agent 的 hooks_raise。
hooks_timeoutOptional[float]= None: 覆盖本次调用中 agent 的 hooks_timeout。

返回单个结果字典，形状与 system_one 相同，并加上 usage["windows"]。跨多个窗口时，截断相关的键与 Agent.predict_long 中的合并方式相同：truncated 是窗口数，truncated_questions 是最后一个窗口的列表，所以 truncated 大于 0 而列表为空是可能的。

decide

decide(
    state: Union[str, dict, list],
    schema: Any = None,
    questions: Optional[Dict[str, 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 只能给一个；额外的关键字参数会透传给 predict / system_one。

参数

stateUnion[str, dict, list]
schemaAny= None
questionsOptional[Dict[str, Dict[str, Any]]]= None
return_detailsbool= False
min_confidenceOptional[float]= None
predict_kwargs

decide_batch

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

通过 predict_batch 把多个状态对照同一个 schema 作答；见 laya.structured。

参数

statesList[Union[str, dict, list]]
schemaAny= None
questionsOptional[Dict[str, Dict[str, Any]]]= None
return_detailsbool= False
min_confidenceOptional[float]= None
predict_kwargs

量化导出

scripts/export_onnx.py --quantize 会在 fp32 导出的旁边写出一份 INT8 仅权重量化的副本（laya.onnx 也会生成 laya.int8.onnx）。动态的逐通道量化把 MatMul 权重转成 int8，激活值保持 fp32，所以不需要校准数据集，ONNXAgent 只要把 onnx_path 指向它就能加载结果。在英文 checkpoint 上，CPU（M 系列，20 个客服工单状态 x choice/noul/score）：模型文件 1.6 GB -> 581 MB，每状态 p50 延迟 ~340 ms -> ~250 ms（~1.35x），且与 fp32 相比决策零变化（单次概率最大漂移 0.09）。用逐张量 scale 代替逐通道，20 个状态里有 3 个翻转，漂移最高 0.29，这就是导出器用逐通道的原因。int8 图只能跑在 CPU 上：ONNX Runtime 在 CUDAExecutionProvider 上没有 INT8 MatMul 内核， GPU provider 会逐节点静默回退。

python scripts/export_onnx.py --model convaiinnovations/laya --output laya.onnx --quantize