Laya 使用手册:下载、微调与推理部署

·  目录

01  Laya 是什么

noul 是项目自造词，本质是一个命名的伯努利分布。0.9 就是"90% 概率为真"。

输入：state（文本 / JSON）+ 类型化问题

↓

build_sequence() 构造序列：

[CLS] <问题类型> 指令 [SEP] [MASK] 选项0 [MASK] 选项1 ... [SEP] state [SEP]

↑ 每个选项前插一个 [MASK] token 作为锚点

↓

DecisionModel.forward()

├─ encoder：ModernBERT（双向自注意力，RoPE，交替局部/全局注意力）

├─ + type_emb(qtype)           # 按 choice/score/noul 注入类型嵌入

├─ head：2 层 TransformerEncoder（post-encoder 决策头）

├─ torch.gather 抽取 [MASK] 位置的隐状态

├─ scorer：LayerNorm→Linear→GELU→Linear(1)，每个选项打成 1 个标量 logit

└─ act_head：pooled [CLS] + 4 个分布特征 → 256 → [P(act), P(escalate)]

↓

softmax(logits / temperature) → 概率分布

confidence = 1 - H(p) / log(k)      # 归一化香农熵

r = log_score + w_sph · spherical_score − w_rps · RPS · 1[score 类型]

02  环境准备与模型下载

# 建议使用独立虚拟环境

python -m venv .venv

source .venv/bin/activate        # Windows: .venv\Scripts\activate

pip install -U laya

python -c "import laya; print(laya.__file__)"

pip install -q -U "laya>=0.1.6" "transformers>=4.48.0" "datasets>=3.0.0" \

safetensors huggingface_hub pyarrow pandas scipy accelerate tabulate

# 本手册的单卡 LoRA 额外需要：

pip install -q -U peft bitsandbytes

注意：如果你的显卡是 Blackwell / RTX 50 系列，装到的 PyTorch 可能不支持其 CUDA 架构，Laya 会自动回落到 CPU 并打印警告。此时按提示安装 nightly 版本：

import laya

agent = laya.load("convaiinnovations/laya")                              # 英语（仓库根目录）

agent_ml = laya.load("convaiinnovations/laya", subfolder="multilingual")   # 100+ 语言

agent_td = laya.load("convaiinnovations/laya", subfolder="typed-decisions")

# 只下英语 checkpoint（根目录文件）

huggingface-cli download convaiinnovations/laya \

--local-dir ./models/laya-en \

--exclude "multilingual/*" "typed-decisions/*"

# 只下多语言 checkpoint

huggingface-cli download convaiinnovations/laya \

--include "multilingual/*" \

--local-dir ./models/laya-ml

# 国内网络可用镜像

export HF_ENDPOINT=https://hf-mirror.com

huggingface-cli download convaiinnovations/laya --local-dir ./models/laya-en

from modelscope import snapshot_download

d = snapshot_download("convaiinnovations/laya")

agent = laya.load("/opt/models/laya-en", device="cuda")

# 加载多语言子目录时：

# agent = laya.load("/opt/models/laya", subfolder="multilingual")

模型目录/

├── rl_agent_config.json     # 必需。缺失则报 "Incompatible model"

├── model.safetensors        # 必需。缺失则报 "'model.safetensors' not found"

├── tokenizer/               # 可选。存在则从这里读；否则回退到 cfg["encoder"]

│   ├── tokenizer.json

│   ├── tokenizer_config.json

│   └── ...

└── encoder/                 # 可选。存在则用本地 config 构建空骨干再灌权重

└── config.json

输出的键名是 temperature 为 list、temperature_by_options 为 dict，两者都要写：Agent 先按 temp_bucket(qtype, k) 去 temperature_by_options 查，查不到才回落到 temperature[qt]。

03  推理使用

import laya

from laya import Router

# preload=True：所有 checkpoint 常驻内存，语言切换只花检测开销（<1 ms）

router = Router(preload=True)

# router = Router(preload=True, device="cuda")

state = {

"from": "user@acme.com",

"subject": "Duplicate charge on invoice #4411",

"body": "Hi, we were billed twice for March. Please refund the duplicate today or we will cancel our plan."

}

questions = {

"department": {

"type": "choice",

"instructions": "Which department should handle this request?",

"criteria": {

"billing": "invoices, payments, refunds",

"technical": "bugs, outages, system errors",

"sales": "pricing, new contracts",

"other": "everything else"

}

},

"urgency": {

"type": "score",

"instructions": "How urgent is this request?",

"criteria": ["not urgent", "soon", "critical deadline or blocking issue"]

},

"churn_risk": {

"type": "noul",

"instructions": "Does the user threaten to cancel or leave?"

},

"refund_requested": {

"type": "noul",

"instructions": "Does the user explicitly request a refund?"

}

}

res = router.predict(state, questions)

print("Department :", res["answers"]["department"]["choice"])   # -> billing

print("Urgency    :", res["answers"]["urgency"]["score"])        # -> 1.84

print("Churn Risk :", res["answers"]["churn_risk"]["noul"])      # -> 0.892

print("Routing    :", res["routing"]["model"])                   # -> english

res["routing"]

# {

#   'model': 'multilingual',

#   'repo': 'convaiinnovations/laya/multilingual',

#   'reason': 'non-Latin script (devanagari, 100% of letters); the English checkpoint cannot read it'

# }

# 只想知道会路由到哪，不实际推理：

router.route({"body": "Der Kunde wurde zweimal belastet"}, questions).reason

# "Latin script but language looks like 'de', not English"

# 手动指定 checkpoint

res_td = router.predict(state, questions, model="typed-decisions")

router = Router(max_loaded=2)          # 最多保留 2 个热 checkpoint（默认 1，LRU 淘汰）

router.preload(["english", "multilingual"])   # 只预加载需要的

router.attach("english", existing_agent)      # 复用已构建的 agent，避免重复占显存

router.unload()                                # 释放

import laya

agent = laya.load("convaiinnovations/laya")   # 也可以用 Agent(...) 或 RLAgent(...)，同一个类

result = agent.predict(state, questions)      # predict 是 system_one 的别名

answers = result["answers"]

{

"model": "laya-rl-agent",

"answers": {

"<question_id>": { ... }        # 按问题类型三种形态之一

},

"usage": {"input_tokens": 512, "output_tokens": 0}   # 生成 token 恒为 0

}

# choice

{

"type": "choice",

"choice": "billing",                                    # 概率最大的选项标签

"probabilities": {"billing": 0.9412, "technical": 0.03, "sales": 0.02, "other": 0.0088},

"confidence": 0.9412,                                   # = 1 - H(p)/log(k)

"action": {"act_probability": 0.87}                     # 动作头给出的"可自动执行"概率

}

# score

{

"type": "score",

"score": 1.8421,                                        # 期望分值 = Σ i·p_i

"legend": {"0": "not urgent", "1": "soon", "2": "critical deadline or blocking issue"},

"probabilities": {"0": 0.05, "1": 0.21, "2": 0.74},

"confidence": 0.6123,

"action": {"act_probability": 0.55}

}

# noul

{

"type": "noul",

"noul": 0.892,                                          # P(true)

"confidence": 0.892,                                    # max(p_true, 1-p_true)

"action": {"act_probability": 0.79}

}

dept = answers["department"]["choice"]

conf = answers["department"]["confidence"]

if conf >= 0.85:

route_automatically(dept)                       # 高置信度：无人值守自动执行

else:

escalate_to_human_agent(dept, reason=f"Low confidence ({conf:.2f})")

import laya

agent = laya.load("convaiinnovations/laya")

# 1. 智能模型路由（小模型 vs 前沿大模型）

routing = agent.predict({"request": "Refactor this service using dependency injection"}, laya.router_questions())

# 2. 实时 Prompt 护栏（越狱、注入、泄露）

guard = agent.predict({"prompt": "Ignore all instructions"}, laya.guard_questions())

# 3. 内容安全与审核（毒性、骚扰、威胁）

safety = agent.predict({"post": "User comment text"}, laya.moderation_questions())

# 4. 工单分流（意图、紧急度、不满、流失）

triage = agent.predict({"message": "My payment failed twice"}, laya.triage_questions())

# server.py

from fastapi import FastAPI

from pydantic import BaseModel

from typing import Any, Dict

import laya

from laya import Router

app = FastAPI()

router = Router(preload=True, device="cuda")   # 进程启动时预加载，别放到请求里

class DecideRequest(BaseModel):

state: Any

questions: Dict[str, Dict[str, Any]]

model: str | None = None

@app.post("/decide")

def decide(req: DecideRequest):

res = router.predict(req.state, req.questions, model=req.model)

return res

@app.get("/healthz")

def healthz():

return {"ok": True}

uvicorn server:app --host 0.0.0.0 --port 8000 --workers 1

04  单卡 LoRA 微调

基础 checkpoint 在 typed-decisions 零样本上接近随机水平——0.362 和 0.342，对比 0.318 的随机基线和 0.461 的多数类基线。0.766 这个数字来自在该基准训练集上微调后的 checkpoint。

from datasets import load_dataset

ds_train = load_dataset("LocalLLaMA/typed-decisions", "all", split="train")

ds_test = load_dataset("LocalLLaMA/typed-decisions", "all", split="test")

# train: 1200 cases / 6000 typed decisions

# test :  400 cases / 2000 typed decisions

{

"id": "case-0001",

"workflow": "customer_service",

"state": "{\"from\": \"user@acme.com\", \"subject\": \"Duplicate charge\", \"body\": \"We were billed twice...\"}",

"questions": "{\"department\": {\"type\": \"choice\", \"instructions\": \"Which department should handle this?\", \"criteria\": {\"billing\": \"invoices, payments, refunds\", \"technical\": \"bugs, outages\", \"sales\": \"pricing\", \"other\": \"everything else\"}}, \"churn_risk\": {\"type\": \"noul\", \"instructions\": \"Does the user threaten to cancel?\"}}",

"gold": "{\"department\": {\"label\": \"billing\", \"probabilities\": {\"billing\": 1.0, \"technical\": 0.0, \"sales\": 0.0, \"other\": 0.0}}, \"churn_risk\": {\"label\": true, \"probabilities\": {\"false\": 0.1, \"true\": 0.9}}}"

}

import json

import torch

from datasets import load_dataset

from transformers import AutoTokenizer

from huggingface_hub import snapshot_download

from laya.agent import _fix_tokenizer_config

from laya.common import build_sequence, render_options, QTYPES

MODEL_ID = "convaiinnovations/laya"

cfg = json.load(open(f"{snapshot_download(MODEL_ID)}/rl_agent_config.json"))

tok = AutoTokenizer.from_pretrained(cfg["encoder"])

_fix_tokenizer_config(snapshot_download(MODEL_ID))

MAX_LEN = cfg["max_len"]          # 512（英语）/ 1024

HEAD_MAX_LEN = cfg["head_max_len"]  # 192（英语）/ 256

def build_target(qtype, q, gold_q):

"""把 gold 转成与 render_options 顺序对齐的概率向量。"""

opts = render_options({"t": qtype, "ins": q["instructions"], "crit": q.get("criteria")})

k = len(opts)

probs = gold_q.get("probabilities")

if qtype == "choice":

keys = list(q["criteria"].keys()) if isinstance(q["criteria"], dict) else list(q["criteria"])

if probs:

return [float(probs.get(kk, 0.0)) for kk in keys]

return [1.0 if kk == gold_q["label"] else 0.0 for kk in keys]

if qtype == "score":

if probs:

return [float(probs.get(str(i), probs.get(i, 0.0))) for i in range(k)]

y = int(gold_q["label"] if "label" in gold_q else gold_q["score"])

return [1.0 if i == y else 0.0 for i in range(k)]

# noul: [P(false), P(true)]

p_true = float(probs.get("true", 0.0)) if probs else (1.0 if gold_q["label"] else 0.0)

return [1.0 - p_true, p_true]

def make_item(row):

state = json.loads(row["state"])

questions = json.loads(row["questions"])

gold = json.loads(row["gold"])

items = []

for qid, qdef in questions.items():

t = qdef["type"]

crit = qdef.get("criteria")

if t == "choice" and isinstance(crit, list):

crit = {c: None for c in crit}

ins = qdef["instructions"]

if not isinstance(ins, str):

ins = json.dumps(ins)

q_internal = {"t": t, "ins": ins, "crit": crit}

seq, markers = build_sequence(tok, state, q_internal, MAX_LEN, HEAD_MAX_LEN)

if len(markers) != len(render_options(q_internal)):

continue          # 选项超出 head_max_len 预算，跳过

items.append({

"ids": seq,

"markers": markers,

"qtype": QTYPES[t],

"target": build_target(t, qdef, gold[qid]),

"label": int(gold[qid].get("label", -1)) if isinstance(gold[qid].get("label"), (int, bool)) else -1,

})

return items

all_items = []

for row in ds_train:

all_items.extend(make_item(row))

torch.save(all_items, "train_items.pt")

print(f"built {len(all_items)} training items")

from laya.common import build_model

m = build_model(cfg)

names = sorted({n.split(".")[-1] for n, _ in m.encoder.named_modules()})

print([n for n in names if n in ("Wqkv", "Wo", "Wi", "Wd")])

# ModernBERT: Wqkv / Wo（注意力）, Wi / Wd（GeGLU 前馈）

from peft import LoraConfig, get_peft_model

lora_cfg = LoraConfig(

r=16,                     # 12–16 GB 显存用 16；24 GB 可用 32

lora_alpha=32,            # 惯例 = 2 × r

lora_dropout=0.05,

bias="none",

target_modules=["Wqkv", "Wo", "Wi", "Wd"],

task_type="FEATURE_EXTRACTION",    # 我们要的是 last_hidden_state，不是分类头

)

model = build_model(cfg)

model = get_peft_model(model, lora_cfg)

# 决策头必须全参训练：它是随机初始化的，且只占几百 K 到 ~2M 参数

for name, p in model.named_parameters():

if "encoder" not in name:

p.requires_grad = True

model.print_trainable_parameters()

model.encoder.config.reference_compile = False

model.enable_input_require_grads()          # LoRA + gradient checkpointing 必需

model.gradient_checkpointing_enable()

EPOCHS = 4

MICRO_BATCH = 8        # 8 sequences per forward pass per GPU

GRAD_ACCUM = 4         # Effective batch across 2 GPUs = 64 sequences (8 * 2 * 4)

GROUP_SIZE = 4         # GRPO baseline samples

LR_ENCODER = 2.5e-5    # Encoder adaptation rate

LR_HEAD = 1.0e-4       # Head adaptation rate

SIGMA_START = 0.4

SIGMA_END = 0.1

# cfg:

cfg["gradient_checkpointing"] = True

cfg["max_tokens_per_batch"] = 4096

cfg["max_len"] = 1024

cfg["head_max_len"] = 256

# 1. 采样 G 组带噪 logit，噪声做零和投影

eps = torch.randn((GROUP_SIZE,) + logits.shape, device=device) * sigma * mask

eps = (eps - eps.sum(-1, keepdim=True) / k) * mask

z = logits.detach().unsqueeze(0) + eps

q = torch.softmax(z.masked_fill(~mask, -1e4), -1)

# 2. 适当评分规则奖励 + GRPO 组基线优势

with torch.no_grad():

r = proper_reward(q, target.unsqueeze(0), batch["qtype"].to(device), mask,

w_sph=0.75, w_rps=1.0)

adv = r - r.mean(0, keepdim=True)

adv = adv / (adv.std() + 1e-6)

# 3. 策略梯度损失 + 软交叉熵引导

logp = -(((z - logits.unsqueeze(0)) ** 2) * mask).sum(-1) / (2 * sigma ** 2)

loss_rl = -(adv * logp).mean()

loss_ce = -(target * torch.log_softmax(logits.masked_fill(~mask, -1e4), -1)).sum(-1).mean()

loss = (loss_rl + 1.0 * loss_ce) / GRAD_ACCUM

progress = epoch / max(1, EPOCHS - 1)

sigma = SIGMA_START + (SIGMA_END - SIGMA_START) * progress   # 0.4 → 0.1

w_sph=0.75 是官方 notebook 的值（源码里 proper_reward 的默认值是 0.5）。0.75 表示更偏向软目标匹配。

def fit_one_temp(sel):

"""sel: [(logits_list, target_list), ...]，返回拟合温度，clamp 到 [0.1, 10.0]"""

if len(sel) < 10:

return 1.0

kmax = max(len(z) for z, _ in sel)

Z = torch.full((len(sel), kmax), -1e4)

T = torch.zeros((len(sel), kmax))

for i, (z, t) in enumerate(sel):

Z[i, :len(z)] = torch.tensor(z)

T[i, :len(t)] = torch.tensor(t, dtype=torch.float32)

log_t = torch.zeros(1, requires_grad=True)

opt = torch.optim.LBFGS([log_t], lr=0.1, max_iter=100)

def closure():

opt.zero_grad()

loss = -(T * torch.log_softmax(Z / log_t.exp(), -1)).sum(-1).mean()

loss.backward()

return loss

opt.step(closure)

return float(torch.clamp(log_t.exp(), 0.1, 10.0).item())

fitted_temps = [1.2, 1.2, 1.2]      # 拟合失败时的兜底值

try:

for qt in range(3):             # 0=choice, 1=score, 2=noul

sel = [(z, t) for q_type, z, t in calib_preds if q_type == qt]

if sel:

fitted_temps[qt] = fit_one_temp(sel)

except Exception as e:

print("Temperature fitting fallback:", e)

cfg["temperature"] = fitted_temps

from laya.common import temp_bucket          # 例如 "choice:6-10", "noul:2"

cfg["temperature_by_options"] = {"choice:6-10": 1.43, "noul:2": 1.15}

import json, os

import torch

OUTPUT_DIR = "laya-ft-my-domain"

os.makedirs(OUTPUT_DIR, exist_ok=True)

# 1. 合并 LoRA 权重（如果想让产物是纯 safetensors，便于统一加载）

from peft import PeftModel

merged = model.merge_and_unload()          # 合并后仍是 DecisionModel

# 2. 保存权重

from safetensors.torch import save_file

save_file({k: v.contiguous().cpu() for k, v in merged.state_dict().items()},

os.path.join(OUTPUT_DIR, "model.safetensors"))

# 3. 保存 tokenizer（Agent 会优先从 tokenizer/ 读）

tok.save_pretrained(os.path.join(OUTPUT_DIR, "tokenizer"))

# 4. 保存 encoder 配置（Agent 检测到 encoder/ 就用本地 config 构建）

from transformers import AutoConfig

AutoConfig.from_pretrained(cfg["encoder"]).save_pretrained(os.path.join(OUTPUT_DIR, "encoder"))

# 5. 保存配置（含拟合好的温度）

cfg["temperature"] = fitted_temps

json.dump(cfg, open(os.path.join(OUTPUT_DIR, "rl_agent_config.json"), "w"), indent=2)

# 6. 加载验证

import laya

agent = laya.Agent(OUTPUT_DIR, device="cuda")

print(agent.predict(state, questions)["answers"])

from huggingface_hub import HfApi

api = HfApi()

api.create_repo("your-org/laya-ft-my-domain", repo_type="model", exist_ok=True)

api.upload_folder(folder_path=OUTPUT_DIR, repo_id="your-org/laya-ft-my-domain")

import numpy as np

from laya.common import ece_score

all_confs, all_corrects = [], []

for row in ds_test:

state, questions, gold = (json.loads(row[k]) for k in ("state", "questions", "gold"))

res = agent.predict(state, questions)["answers"]

for qid, a in res.items():

all_confs.append(a["confidence"])

if a["type"] == "choice":

all_corrects.append(float(a["choice"] == gold[qid]["label"]))

elif a["type"] == "noul":

all_corrects.append(float((a["noul"] >= 0.5) == bool(gold[qid]["label"])))

else:

all_corrects.append(float(round(a["score"]) == int(gold[qid]["label"])))

print("Accuracy:", np.mean(all_corrects))

print("ECE     :", ece_score(np.array(all_confs), np.array(all_corrects)))

05  能力边界与选型建议

# 方案 1：调大预算（注意显存和延迟会涨）

agent.cfg["head_max_len"] = 512

agent.cfg["max_len"] = 1024          # 可到 2048 / 4096 / 8192

顺带一提，laya-multilingual 的 mmBERT-base 编码器本身支持 RoPE 到 8192 上下文，所以上下文不是硬瓶颈，token 预算分配才是。

06  速查表与资源

# 加载

import laya

agent = laya.load("convaiinnovations/laya")                              # 英语

agent = laya.load("convaiinnovations/laya", subfolder="multilingual")     # 多语言

agent = laya.Agent("./laya-ft-my-domain", device="cuda")                  # 本地微调产物

agent = laya.RLAgent("./laya-ft-my-domain")                               # 同 Agent 的别名

# 推理

res = agent.predict(state, questions)          # predict 是 system_one 的别名

res["answers"][qid]["choice"]                  # choice 结果

res["answers"][qid]["score"]                   # score 期望分值

res["answers"][qid]["noul"]                    # P(true)

res["answers"][qid]["confidence"]              # 1 - H(p)/log(k)

res["answers"][qid]["action"]["act_probability"]

# 路由

from laya import Router

router = Router(preload=True, device="cuda", max_loaded=2)

router.predict(state, questions, model="typed-decisions")

router.route(state, questions).reason

router.preload(["english", "multilingual"])

router.attach("english", agent); router.unload()

# 配置（改 cfg 后需重建 Agent 才生效）

agent.cfg["max_len"] = 1024

agent.cfg["head_max_len"] = 512

# 内置预设

laya.router_questions(); laya.guard_questions()

laya.moderation_questions(); laya.triage_questions()

from laya.common import (

QTYPES,                # {"choice": 0, "score": 1, "noul": 2}

QTYPE_NAMES,           # 反向映射

serialize_state,       # state → str (JSON, ensure_ascii=False)

render_criterion,      # 单个 criterion → 文本

render_options,        # 问题 → 选项文本列表（noul 恒为 [false, true]）

build_sequence,        # 构造 [CLS]...[MASK]opt...[SEP]state[SEP]

DecisionModel,         # encoder + head + type_emb + scorer + act_head

build_model,           # cfg → DecisionModel

proper_reward,         # 适当评分规则奖励

td_lambda_targets,     # 多轮轨迹的 TD(λ) 目标

ece_score,             # 期望校准误差

confidence_from_probs, # 归一化熵置信度

temp_bucket,           # (qtype, k) → "choice:6-10" 之类的桶名

collate_items,         # batch → 张量字典

)

from laya.agent import _fix_tokenizer_config

Laya 不是"更小的 LLM"，而是把决策从生成里剥离出来的另一类模型。 它不可替代 LLM，但它能让你的路由、分类、审核、护栏这些高频反射式决策，从 1500 ms 降到 33 ms，成本从按 token 计费降到 0，并且给出第一个在数学上站得住的置信度——足以支撑真正的自动执行与人工升级门控。

前提是：你必须微调它。 零样本的 Laya 约等于抛硬币。