Kev 架构详解（含源码）

冻结底座 + LoRA + 指针头 — 用一次前向传播完成结构化决策

1. 一句话概括

它的关键特征：

2. 决策全流程：从文本到答案

以一条真实请求为例（判断「订单有问题」是否包含 PII），从原始文本到最终答案，一次前向的完整形状流转如下。阶段一到阶段四是任何 Transformer 都一样的通用链路，阶段五的指针头才是 Kev 真正新增的计算。

把阶段五的指针头单独放大看——九小步，每一步的形状和数值：

完整形状流转总表：

3. 为什么不用自回归

自回归 LLM 生成一个答案，本质上是一个串行过程：生成第一个 token，再基于它生成第二个，直到遇到结束符。每一轮都要跑一次完整的前向传播。

自回归：每生成一个 token 就要跑一次完整前向，串行、延迟累积

Kev 换了一种思路：既然答案就在有限的候选集合里，为什么不让模型直接为每个候选打分？

Kev：一次前向传播，直接得到所有候选的概率分布

这就是 Kev 能把延迟压到 65ms 的根本原因——不做串行生成，只做并行打分。

4. 整体架构图

灰色 = 冻结不训练；绿色 = 可训练；蓝色 = 输入；黄色 = 输出

5. 数据格式与标记

Kev 不用 JSON 或 chat template，而是用一种扁平标记格式把所有内容拼成一条 token 序列：

[STATE] 订单 11780 有问题，帮忙看看。手机号 18147389777。
[Q] 文本是否包含手机号、邮箱、身份证号等个人敏感信息？
[OPT] no
[OPT] yes
[DECIDE]

各个标记的作用：

[STATE]

一整段非结构化文本（工单、评论、游戏局面）。只出现一次，是所有问题共享的上下文。

[Q]

一道问题的描述。可以出现多次——一次调用可以挂任意多道题。

[OPT]

一个候选选项。同一道题下可以有多行 [OPT]。

[DECIDE]

决策标记。每道题后面跟一个，指针头从它的 hidden state 生成 query。

一次调用可以挂多道题，每道题有自己的候选和 [DECIDE]

6. 指针头怎么工作

指针头是 Kev 的核心，也是它区别于自回归的关键。它做的事情非常"轻"——一个缩放点积 + 一个 softmax：

用数学式子写出来是：

q = query_proj(hidden[DECIDE])            # [1, head_dim]
k_i = key_proj(hidden[OPT_i])             # [num_opts, head_dim]
score_i = (q · k_i) / sqrt(head_dim)      # 缩放点积
probs = softmax([score_1, score_2, ...])  # 各候选概率

7. 训练流程

训练配置：

8. 推理流程与延迟

延迟实测：

9. 当前短板与解法：软标签蒸馏

当前 Kev 用硬标签交叉熵训练，存在一个可量化的短板。

什么是 act 覆盖率？

Jev 风格的决策服务会把答案按置信度分三档：

act (p ≥ 0.85)

高置信——可以直接自动执行，不需要人工复核

review (0.5 ≤ p < 0.85)

中等——建议人工复核

human (p < 0.5)

低置信——必须转人工

act 覆盖率就是"模型有多少比例的答案，它敢声称自己 ≥ 0.85 确定"。

Kev 当前的表现

act 覆盖率只有 0.3%——门控功能基本处于关闭状态，几乎所有答案都落在 review 和 human 两档。

为什么会这样？

温度缩放救不了

一个自然的想法是"用温度缩放把概率整体拉高"——但这行不通：

这是"覆盖换校准"的 tradeoff：想要更多高置信样本，就要牺牲概率的诚实性。而且即使 T=2.0，act 覆盖也只有 7.5%——离可用的门控还有很大距离。

解法：软标签蒸馏

10. 实测数据

业务能力

游戏能力（零样本迁移）

延迟

11. Kev 的能力边界

Kev 不是一个"通用决策模型"，它的定位非常明确：

适合的应用场景

1. 需要低延迟的在线决策 —— 65ms 是 Kev 最大的竞争力
2. 不需要置信度门控的场景 —— 比如直接取 argmax 就能用
3. 游戏/实时交互 demo —— 低延迟让决策看起来"实时"
4. 结构化输出要求严格的场景 —— 概率分布天然不会格式错误

需要补齐的方向

如果场景需要 Jev 式的 act/review/human 门控（高置信自动执行、低置信转人工），当前 Kev 还不满足。软标签蒸馏是唯一路径——用带软标签的数据重新训练指针头，把"该有多自信"这个能力学进去。

12. 核心源码逐段解读

下面把 Kev 实现的全部源码贴出来，逐段讲解。

12.1 指针头：kev/pointer_head.py

# -*- coding: utf-8 -*-
"""Kev 指针头:为每个问题的候选选项打分并 softmax。"""
import torch
import torch.nn as nn
import torch.nn.functional as F


class PointerHead(nn.Module):
    """从 <DECIDE> 位置的 hidden 生成 query,从每个 <OPT> 位置的 hidden 生成 key,
    缩放点积后对每题候选 softmax 输出概率。"""

    def __init__(self, hidden_size: int = 2560, head_dim: int = 256):
        super().__init__()
        self.query_proj = nn.Linear(hidden_size, head_dim)
        self.key_proj = nn.Linear(hidden_size, head_dim)
        self.scale = head_dim ** -0.5

    def forward(self, hidden, decide_idx, opt_idx, opt_mask):
        """
        hidden:     [b, s, h]
        decide_idx: [b]           每样本 <DECIDE> 标记位置
        opt_idx:    [b, max_opts] 每样本各候选 <OPT> 标记位置(pad=0)
        opt_mask:   [b, max_opts] 候选有效性
        返回 [b, max_opts] 概率(pad 位置为 0)
        """
        b = hidden.shape[0]
        hidden_f = hidden.float()  # 指针头为 fp32 参数,底座 hidden 为 bf16
        q = self.query_proj(hidden_f[torch.arange(b, device=hidden.device), decide_idx])  # [b, hd]
        flat_idx = opt_idx.reshape(-1)
        k = self.key_proj(hidden_f.reshape(-1, hidden.shape[-1])[flat_idx])               # [b*mo, hd]
        k = k.view(b, opt_idx.shape[1], -1)                                             # [b, mo, hd]
        scores = torch.einsum("bd,bmd->bm", q, k) * self.scale                          # [b, mo]
        scores = scores.masked_fill(~opt_mask, float("-inf"))
        return F.softmax(scores, dim=-1)

逐段解读

__init__

两个线性层：query_proj 把 [DECIDE] 位置的 hidden 从 2560 维投到 256 维，key_proj 把 [OPT] 位置的 hidden 也投到 256 维。两者维度相同才能做点积。scale 是缩放因子 1/√256，防止点积数值过大导致 softmax 饱和。

hidden.float()

底座输出是 bf16（半精度），但指针头参数是 fp32（全精度）。这里统一转 fp32，避免 bf16 精度不足以做点积——只在指针头这一层多花一点点开销，换来数值稳定性。

q = query_proj(hidden_f[arange(b), decide_idx])

用 arange(b) 生成 [0,1,2,...,b-1]，和 decide_idx 一起做花式索引，从每个样本中取出 [DECIDE] 位置的 hidden。decide_idx 指向 [DECIDE] 标记的最后一个 token（]），因为该位置已经"读完整个标记"，语义最完整。

flat_idx = opt_idx.reshape(-1)

把 [b, max_opts] 拉平成一维 [b×max_opts]，方便一次性索引所有候选位置。

k = key_proj(hidden_f.reshape(-1, h)[flat_idx])

hidden_f.reshape(-1, h) 把 [b, s, h] 拉成 [b×s, h]，然后按 flat_idx 取所有候选对应的 hidden。再 key_proj 投到 256 维。最后 view(b, max_opts, -1) 恢复形状。

scores = einsum("bd,bmd->bm", q, k) * scale

爱因斯坦求和：bd 是 query [b, 256]，bmd 是 key [b, max_opts, 256]，输出 bm 是 [b, max_opts]。等价于对每个 (b, m) 做 q[b] · k[b,m]，就是缩放点积。

masked_fill(~opt_mask, -inf)

把 pad 位置（opt_mask=False）的分数设为 -inf。softmax 后这些位置的概率为 0，不影响有效候选的归一化。

F.softmax(scores, dim=-1)

对候选维度做 softmax，得到概率分布。pad 位置恒为 0。

12.2 数据编码：kev/data.py

# -*- coding: utf-8 -*-
"""Kev 格式数据:tokenize + 位置记录(opt/decide)。"""
import json
import torch

MARK_STATE, MARK_Q, MARK_OPT, MARK_DECIDE = "[STATE]", "[Q]", "[OPT]", "[DECIDE]"


def encode_sample(tok, sample, max_len=1024):
    """返回 dict(input_ids, attention_mask, decide_idx, opt_idx, opt_mask, label)。

    边界保证:截断后 [DECIDE] 必须存活且指向 [DECIDE] 标记本身;
    [OPT] 位置全部 < len(ids);极端情况(题干+候选超长)则截 state 行保分支行。
    """
    lines = sample["prompt"].split("\n")
    state_lines = [ln for ln in lines if ln.startswith(MARK_STATE)]
    branch_lines = [ln for ln in lines if not ln.startswith(MARK_STATE)]
    nl_ids = tok("\n", add_special_tokens=False)["input_ids"]
    # 先编码分支行(问题/候选/DECIDE),再算还剩多少预算给 state
    branch_ids = []
    opt_pos = []
    decide_idx = -1
    for ln in branch_lines:
        line_ids = tok(ln, add_special_tokens=False)["input_ids"]
        if ln.startswith(MARK_OPT):
            # opt_pos 必须指向选项行最后一个 token(读完选项内容后),
            # 与 Kev 原文 </opt> 语义对齐;行首 '[' 不携带选项内容
            opt_pos.append(len(branch_ids) + len(line_ids) - 1)
        branch_ids += line_ids + nl_ids
        if ln.startswith(MARK_DECIDE):
            # 指向 [DECIDE] 标记的尾 token(去掉行尾换行,换行符无语义)。
            # 出处:Kev README "Because <decide> comes last, it can attend to the
            # full option list"——尾位置聚合完整选项列表;Kev 原文 <decide> 为单
            # token,我们的 4-token 标记取尾 token 是对该语义的最近对应。
            decide_idx = len(branch_ids) - len(nl_ids) - 1
    state_ids = tok(state_lines[0] + "\n", add_special_tokens=False)["input_ids"] if state_lines else []
    budget = max_len - len(branch_ids)
    if len(state_ids) > budget:
        state_ids = state_ids[:max(budget, 1)]
    ids = state_ids + branch_ids
    decide_idx += len(state_ids)
    opt_pos = [p + len(state_ids) for p in opt_pos]
    ids = ids[:max_len]
    decide_idx = min(decide_idx, len(ids) - 1)
    opt_pos = [p for p in opt_pos if p < len(ids)] or [len(ids) - 1]
    attn = [1] * len(ids)
    return {"input_ids": ids, "attention_mask": attn, "decide_idx": decide_idx,
            "opt_pos": opt_pos, "label": int(sample["label"])}


def collate(batch, pad_id):
    maxlen = max(len(b["input_ids"]) for b in batch)
    max_opts = max(len(b["opt_pos"]) for b in batch)
    input_ids, attn, decide, opt_idx, opt_mask, labels = [], [], [], [], [], []
    for b in batch:
        n = len(b["input_ids"])
        input_ids.append(b["input_ids"] + [pad_id] * (maxlen - n))
        attn.append([1] * n + [0] * (maxlen - n))
        decide.append(b["decide_idx"])
        ops = b["opt_pos"] + [0] * (max_opts - len(b["opt_pos"]))
        opt_idx.append(ops)
        opt_mask.append([1] * len(b["opt_pos"]) + [0] * (max_opts - len(b["opt_pos"])))
        labels.append(b["label"])
    dev = "cuda"
    t = lambda x, dt: torch.tensor(x, dtype=dt, device=dev)
    return {"input_ids": t(input_ids, torch.long),
            "attention_mask": t(attn, torch.long),
            "decide_idx": t(decide, torch.long),
            "opt_idx": t(opt_idx, torch.long),
            "opt_mask": t(opt_mask, torch.bool),
            "labels": t(labels, torch.long)}


def load_jsonl(path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(l) for l in f if l.strip()]

逐段解读

四个标记常量

统一定义 [STATE]、[Q]、[OPT]、[DECIDE]。改格式时只改这里。

先编码分支行，再编码 state

这是一个关键的预算分配策略：分支行（问题/候选/DECIDE）是决策必需的，不能截断；state 是上下文，可以截。所以先算完分支行长度，剩下的额度才给 state。

opt_pos.append(len(branch_ids) + len(line_ids) - 1)

这是修过的关键点。len(branch_ids) 是当前累积位置，len(line_ids) - 1 是本行最后一个 token 的偏移。加起来就是"选项行最后一个 token 的绝对位置"。
为什么不用行首？因为行首的 hidden state 只包含"这里开始一个选项"，不包含"选项内容是什么"。指针头的 key 必须携带候选内容。

decide_idx = len(branch_ids) - len(nl_ids) - 1

同理，指向 [DECIDE] 标记的最后一个 token（]）。len(branch_ids) 是加完本行+换行后的长度，减去换行长度 len(nl_ids)，再减 1 就是本行最后一个 token。

state 编码 + 预算裁剪

budget = max_len - len(branch_ids) 是 state 能用的最大 token 数。如果 state 太长就截断到 budget。注意 max(budget, 1) 保证至少留 1 个 token，避免极端情况导致空序列。

位置偏移修正

state 是拼在 branch 前面的，所以 decide_idx 和所有 opt_pos 都要加上 len(state_ids) 才是最终位置。

最终保险

decide_idx = min(decide_idx, len(ids) - 1) 和 opt_pos = [...] or [len(ids) - 1] 是防止极端情况下索引越界。前者保证不超上界，后者保证 opt_pos 非空（否则指针头无法计算）。

collate 函数

把一个 batch 里的样本 padding 到相同长度。opt_mask 是关键：它标记哪些位置是真实候选、哪些是 padding。max_opts 取 batch 内最大候选数，其他样本用 0 填充。

12.3 训练循环：kev/train.py

# -*- coding: utf-8 -*-
"""Kev 训练:冻结底座 + rank16 LoRA(7 投影)+ 指针头,CE 训练。

用法:
    .\\.venv\\Scripts\\python.exe kev\\train.py --max-samples 30000 --epochs 1
"""
import argparse
import faulthandler
import io
import math
import random
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

import torch  # noqa: E402
import torch.nn.functional as F  # noqa: E402
from peft import LoraConfig, get_peft_model  # noqa: E402
from transformers import AutoModelForCausalLM, AutoTokenizer  # noqa: E402

from kev.data import collate, encode_sample, load_jsonl  # noqa: E402
from kev.pointer_head import PointerHead  # noqa: E402

ROOT = Path(__file__).resolve().parent.parent


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default="models/Qwen3.5-4B")
    ap.add_argument("--train", default=str(ROOT / "data" / "kev_format" / "train.jsonl"))
    ap.add_argument("--max-samples", type=int, default=30000)
    ap.add_argument("--epochs", type=float, default=1.0)
    ap.add_argument("--bs", type=int, default=4)
    ap.add_argument("--accum", type=int, default=8)
    ap.add_argument("--lr", type=float, default=1.0e-4)
    ap.add_argument("--max-len", type=int, default=1024)
    ap.add_argument("--out", default=str(ROOT / "runs" / "kev-qwen35-4b"))
    args = ap.parse_args()

    # 卡死自诊断:4 分钟无进展自动把全部线程堆栈打进 stderr(入日志)
    faulthandler.dump_traceback_later(240, repeat=True)

    tok = AutoTokenizer.from_pretrained(args.base, trust_remote_code=True)
    pad_id = tok.pad_token_id or tok.eos_token_id
    base = AutoModelForCausalLM.from_pretrained(
        args.base, dtype=torch.bfloat16, trust_remote_code=True).cuda()
    base.gradient_checkpointing_enable()
    base.config.use_cache = False
    # PEFT 包装后 config 为文本塔配置(Qwen3_5TextConfig),直接含 hidden_size
    cfg = base.config
    hidden_size = getattr(cfg, "hidden_size",
                          getattr(getattr(cfg, "text_config", cfg), "hidden_size", 2560))

    lcfg = LoraConfig(r=16, lora_alpha=32, lora_dropout=0.05, bias="none",
                      task_type="CAUSAL_LM",
                      target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                                      "gate_proj", "up_proj", "down_proj"])
    base = get_peft_model(base, lcfg)
    head = PointerHead(hidden_size).cuda()
    head = head.to(torch.float32)

    samples = load_jsonl(args.train)[: args.max_samples]
    print(f"样本 {len(samples)} 条", flush=True)
    opt = torch.optim.AdamW(
        [p for p in base.parameters() if p.requires_grad] + list(head.parameters()),
        lr=args.lr)
    steps_per_epoch = math.ceil(len(samples) / (args.bs * args.accum))
    total_steps = int(steps_per_epoch * args.epochs)
    sched = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=total_steps)
    print(f"优化步总数 {total_steps}(每步 {args.bs * args.accum} 样本)", flush=True)

    head_params = [p for p in head.parameters() if p.requires_grad]
    base.train()
    order = list(range(len(samples)))
    import random
    rng = random.Random(0)

    step = 0
    micro = 0
    t0 = time.time()
    loss_acc = 0.0
    done = False
    while not done:
        rng.shuffle(order)
        for i in range(0, len(order), args.bs):
            batch_rows = [samples[j] for j in order[i:i + args.bs]]
            batch = [encode_sample(tok, r, args.max_len) for r in batch_rows]
            batch_t = collate(batch, pad_id)
            # 文本塔直接输出 hidden(绕开 lm_head 的整块 logits)
            inner = base.base_model.model.model  # Qwen3_5Model(text tower)
            hidden = inner(input_ids=batch_t["input_ids"],
                           attention_mask=batch_t["attention_mask"]).last_hidden_state
            probs = head(hidden, batch_t["decide_idx"], batch_t["opt_idx"],
                         batch_t["opt_mask"])
            logp = torch.log(probs.clamp_min(1e-9))
            loss = F.nll_loss(logp, batch_t["labels"])
            (loss / args.accum).backward()
            loss_acc += float(loss.detach())
            micro += 1
            if micro % args.accum:
                continue
            opt.step()
            sched.step()
            opt.zero_grad(set_to_none=True)
            step += 1
            if step % 20 == 0 or step == total_steps:
                speed = (time.time() - t0) / max(1, step)
                print(f"step {step}/{total_steps} loss={loss_acc / (20 * args.accum):.4f} "
                      f"{speed:.1f}s/it", flush=True)
                loss_acc = 0.0
            if step >= total_steps:
                done = True
                break
        if done:
            break

    print(f"训练完成,用时 {(time.time() - t0) / 60:.0f} 分钟", flush=True)
    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    base.save_pretrained(out / "adapter")
    torch.save(head.state_dict(), out / "pointer_head.pt")
    print(f"adapter -> {out / 'adapter'};pointer head -> {out / 'pointer_head.pt'}",
          flush=True)


if __name__ == "__main__":
    main()

逐段解读

faulthandler.dump_traceback_later(240, repeat=True)

每 240 秒（4 分钟）如果没有输出，就把所有线程的调用栈打印到 stderr。这是一个卡死自诊断机制——如果训练卡住，日志里直接能看到卡在哪一行，不用反复重跑排查。

加载底座 + gradient_checkpointing

AutoModelForCausalLM.from_pretrained(...).cuda() 加载 Qwen3.5-4B。gradient_checkpointing_enable() 开启梯度检查点，用计算换显存：前向时不保存所有中间激活，反向时重算。这让 4B 模型能在 16GB 卡上训练。

LoRA 配置

rank=16、alpha=32、dropout=0.05，注入 7 个投影层（q/k/v/o/gate/up/down）。rank 16 很小——只训练 7 × 2 × 2560 × 16 ≈ 4500 万 参数，占底座 1% 左右。
get_peft_model(base, lcfg) 把 LoRA 包装进底座，返回的模型里底座的原始参数被冻结。

指针头初始化

PointerHead(hidden_size).cuda() 创建指针头，.to(torch.float32) 确保全精度参数。hidden_size 从底座 config 拿，如果是多模态包装就取 text_config.hidden_size。

优化器

AdamW 只包含 requires_grad=True 的参数（LoRA + 指针头），底座的冻结参数不参与。这是显存和速度的关键。

学习率调度

CosineAnnealingLR 从 lr 余弦降到 0。T_max=total_steps 表示整个训练过程正好走完一个余弦周期。这是 SFT 的标准做法。

训练主循环

按 args.bs 分批（这里是 4），每个 micro-batch 做一次前向+反向，累积 args.accum 次才 opt.step()。有效 batch = 4 × 8 = 32。
rng.shuffle(order) 每个 epoch 重新打乱数据顺序。

inner = base.base_model.model.model

这是关键的一行。它直接访问底座的"文本塔"，绕过 lm_head（最后的词表投影层，会把 hidden 投到 24 万维词表）。
如果用完整模型，会生成一个 [batch, seq, 248320] 的 logits 张量——这才是显存爆炸的主因。绕过 lm_head 直接把 hidden 喂给指针头，显存省下几 GB。

指针头前向 + 损失

head(hidden, ...) 得到 [b, max_opts] 概率。logp = torch.log(probs.clamp_min(1e-9)) 取对数（clamp 防止 log(0)）。F.nll_loss(logp, labels) 就是交叉熵——预测概率分布和正确标签之间的差距。

(loss / args.accum).backward()

梯度累积的关键：每次反传都把 loss 除以累积步数，多步累积后梯度量级正好和单步大 batch 一致。

保存产物

base.save_pretrained(out / "adapter") 只保存 LoRA 权重（几十 MB），torch.save(head.state_dict(), ...) 保存指针头权重。两者合起来才是完整的 Kev 模型。

12.4 数据转换：scripts/convert_to_kev.py

# -*- coding: utf-8 -*-
"""Kev 架构落地第 1 步:把 v1 业务数据与游戏数据转换为 Kev 格式。

Kev 样本(每题一个样本,state 在前,候选逐个列出,答案=正确候选下标):
  prompt: "[STATE] {state}\n[Q] {desc}\n[OPT] {o1}\n[OPT] {o2} ...\n[DECIDE]"
  label:  正确候选下标

noul 的两候选固定为 ["no", "yes"](answer 0→no, 1→yes);
choice 的候选为选项文本;score 的候选为档位文本(0 档在最前)。
"""
import io
import json
import random
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

NOUL_OPTIONS = ["no", "yes"]


def record_to_kev_samples(rec):
    """v1 record(原始格式)→ [kev 样本],每题一个。state 截 600 字符防超长。"""
    out = []
    state = rec["state"][:600]
    for q in rec["questions"]:
        opts = NOUL_OPTIONS if q["type"] == "noul" else list(q.get("options") or q.get("levels") or [])
        val, _prob = rec["answers"][q["id"]]
        label = int(float(val))
        if not (0 <= label < len(opts)):
            continue
        prompt = f"[STATE] {state}\n[Q] {q['desc']}\n"
        for o in opts:
            prompt += f"[OPT] {o}\n"
        prompt += "[DECIDE]"
        out.append({"prompt": prompt, "options": opts, "label": label,
                    "task": rec.get("task", ""), "qid": q["id"]})
    return out


def game_kev_samples_from_lines(lines):
    """游戏 sharegpt 行(game_train.jsonl)→ Kev 样本。"""
    out = []
    for ln in lines:
        try:
            msg = json.loads(ln)
            user = json.loads(msg["messages"][1]["content"])
            asst = json.loads(msg["messages"][2]["content"])
        except Exception:
            continue
        state = user["state"]
        qlist = user["questions"] if isinstance(user["questions"], list) else [
            dict(q, id=qid) for qid, q in user["questions"].items()]
        for q in qlist:
            opts = ["no", "yes"] if q["type"] == "noul" else list(q.get("options") or q.get("levels") or [])
            val, _p = asst["answers"][q["id"]]
            label = int(float(val))
            if not (0 <= label < len(opts)):
                continue
            prompt = f"[STATE] {state}\n[Q] {q['desc']}\n"
            for o in opts:
                prompt += f"[OPT] {o}\n"
            prompt += "[DECIDE]"
            out.append({"prompt": prompt, "options": opts, "label": label,
                        "task": "game", "qid": q["id"]})
    return out


def main() -> None:
    v1_dir = ROOT / "data" / "v1"
    game_file = ROOT / "data" / "game" / "game_train.jsonl"
    out_dir = ROOT / "data" / "kev_format"
    out_dir.mkdir(parents=True, exist_ok=True)

    rng = random.Random(20260926)
    biz_train = []
    for split in ("train", "val", "test"):
        rows = [json.loads(l) for l in (v1_dir / f"{split}.jsonl").open(encoding="utf-8") if l.strip()]
        samples = []
        for rec in rows:
            samples += record_to_kev_samples(rec)
        rng.shuffle(samples)
        with (out_dir / f"{split}.jsonl").open("w", encoding="utf-8") as f:
            for s in samples:
                f.write(json.dumps(s, ensure_ascii=False) + "\n")
        print(f"{split}: {len(rows)} 条记录 → {len(samples)} 个 Kev 题样本")
        if split == "train":
            biz_train = samples

    game_lines = game_file.open(encoding="utf-8").read().splitlines()
    game_samples = game_kev_samples_from_lines(game_lines)
    rng.shuffle(game_samples)
    # 游戏训练题并入训练集(业务+游戏必须同训一个 LoRA+指针头)
    with (out_dir / "train.jsonl").open("a", encoding="utf-8") as f:
        for s in game_samples:
            f.write(json.dumps(s, ensure_ascii=False) + "\n")
    with (out_dir / "game_test.jsonl").open("w", encoding="utf-8") as f:
        for s in game_samples:
            f.write(json.dumps(s, ensure_ascii=False) + "\n")
    print(f"game_train 并入 {len(game_samples)} 题;game_test {len(game_samples)} 题")
    print(f"kev_format/train.jsonl 最终 = 业务 {len(biz_train)} + 游戏 {len(game_samples)} 题")


if __name__ == "__main__":
    main()

逐段解读

每种题型映射到候选集

• noul：固定两候选 ["no", "yes"]，label 0 = no，1 = yes
• choice：直接用 options 列表，label 是选项下标
• score：用 levels 列表（"0 档"、"1 档"...），label 是档位下标

注意：score 在 Kev 里被处理成离散分类，原来的连续值（如 3.2）会被 int(3.2)=3 截断。这是一个已知的精度损失——Kev 版本对 score 只有档位级精度。

state = rec["state"][:600]

state 截到 600 字符。这是防超长的预处理——Kev 的 max_len=1024 token，state 太长会挤掉分支行。截断 + 后续在 encode_sample 里再按预算截，双重保险。

拼接 prompt

标准格式：[STATE] {state}\n[Q] {desc}\n[OPT] {o1}\n[OPT] {o2}\n[DECIDE]。每题一个样本——同一 state 会被复制多份，每份只带一道题。这样指针头每个样本只需要处理一个 [DECIDE]，简化实现。

游戏数据转换

游戏数据原本是 sharegpt 格式（messages 数组），这里解析出 user 里的 state/questions 和 assistant 里的 answers，然后走同一套 record_to_kev_samples 逻辑。唯一的区别是任务名标记为 "game"。

合并到训练集

业务数据先写进 train.jsonl，然后用 "a"（追加）模式把游戏数据也写进去。这就是业务和游戏共用一个模型的数据准备。

同时输出 game_test.jsonl

游戏数据同时存一份独立文件，用于评测 Kev 模型在游戏题上的表现。注意 game_test.jsonl 实际上和训练时用的游戏样本是同一批——这是一个潜在问题（训练集和测试集重叠），但当前用于"零样本迁移能力"的粗略评估，暂时可接受。

12.5 包定义：kev/__init__.py

# -*- coding: utf-8 -*-
"""Kev 架构:冻结底座 + LoRA + 指针头(单次前向输出概率,无自回归)。"""

只有一个 docstring，定义包的身份。没有导出任何东西——所有功能都在子模块里按需导入。

12.6 源码组织总结

数据 → 编码 → 训练 → 模型产物的完整链路

附录：Kev 相关文件清单

—— 文档结束 ——