从预测任务到数据契约

本章任务：让一个训练位置只回答一个明确问题

本章先不训练网络。你要把一组原始短文变成可追溯的数据清单、只从训练文本得到的字符词表，以及输入、目标、有效位置三张对齐的表。后面的模型无论有多少层，都必须服从这份协议：第一个位置根据开始标记预测首字符，之后只根据已有前缀预测下一字符，最后预测结束。任务若定义错了，一个成功优化的网络也可能只是在复制输入。

读者需要会 Python 列表、字典和循环，能进行分数乘除，知道矩阵的行列含义。概率、token、批次和条件范围将在首次使用时定义，不预设你已经学过语言模型。最终项目会实现自写的、全参数训练的 decoder-only Transformer；本章交付它的数据入口，不把查表代码或后续单个输出层的优化称为完整 Transformer 训练。

一、从联合概率推导下一字符任务

先暂时使用一个有限字母表，即允许出现的符号集合。一个 token 是编码器输出的离散单位；本章采用字符级编码，所以一个普通 token 对应一个 Unicode 字符，而不是一个完整中文词。设观察到的普通字符依次为 $x_1,\ldots,x_n$，在前面放固定开始标记 $x_0=\mathrm{BOS}$，后面放 $x_{n+1}=\mathrm{EOS}$。BOS 提供空前缀的表示，不是要求模型预测的随机事件；EOS 则是模型应预测的结束事件。

对两个事件 A、B，且 P(A)>0，条件概率定义为 P(B|A)=P(A∩B)/P(A)。把两边乘以 P(A)，得到 P(A∩B)=P(A)P(B|A)。现在让 A 表示“首字符是春”，B 表示“第二字符是风”。这一步没有假设两个字符独立。若事件 A 的概率为零，这个除法不能使用；实际观察到的路径在本例中都取正概率，零概率的边界留到下一章的损失讨论。

将相同定义再用于第三个事件 C：P(A∩B∩C)=P(A∩B)P(C|A∩B)。代入上一步，得到 P(A)P(B|A)P(C|A∩B)。每加一个字符，都把“之前所有事件同时发生”作为条件，因此可以继续到 EOS。这种连乘是概率链式分解，属于精确恒等式；它本身没有说下一个字符只依赖最近一个字符。

模型使用参数 θ 表示一整组待学习的数。把真实分布 p 换成可计算的 qθ，是选择模型族；这不是从链式法则推出模型一定正确。若模型只能读取最近 W 个 token，它计算的是截断条件 qθ(x下一项|最近 W 项)，相较完整前缀丢掉了信息。W 是资源与架构约定，也不是自然语言确实满足某种有限记忆定律的证明。后面会用独立文本检验这项近似的后果。

算例一：一句话的概率与一次预测的概率不同

构造一个仅用于手算的小分布，输出支持为“春、风、EOS”。表中的每一行给出指定前缀下的三个条件概率，均为非负且和为一。这里的数是教学设定，不是某个训练模型的测量。我们只计算“春风”恰好在风之后结束的事件，因此需要三个乘因子，而不只需要首字符或末字符的概率。

先取得“春”的概率零点六，再乘以在已有春时得到风的条件概率零点五，得到前两个字符为春风的概率零点三；再乘结束条件零点八，得到完整事件概率零点二四。若漏掉 EOS，得到零点三，含义变成“至少以春风开头”，还包含继续生成的事件。若把零点六、零点五、零点八相加，会得到一点九，超过一，也违背联合事件的计算规则。

独立检查有两项。第一项逐行相加，确认每行确实为一，避免把三个任意分数当成概率。第二项比较事件包含关系：恰好为春风并结束，是以春风开头的子事件，所以零点二四不能超过零点三。这个不等式检查不依赖重新照抄同一个连乘，能发现把最后条件概率取倒数等错误。

二、词表为什么只能从训练文本建立

数据集分成训练、验证、测试三部分。训练文本用来拟合参数及词表，验证文本用于选择学习率、训练轮数等决策，测试文本在这些选择结束后才计分。切分单位首先是完整原始文档，具有相同内容或同源改写的文档还应归为同组。随后再从各自集合内部生成窗口；先生成高度重叠窗口再随机切分，会让几乎相同的前缀进入不同集合。

我们使用原创短文“春风”“春雨”“雨”作训练，“秋雨”作验证，“雪风”作测试。与真实项目 CharTokenizer.fit 一致，先固定 UNK=0、BOS=1、EOS=2，再将仅在训练文本出现的字符去重并按 Unicode 编码排序，得到春为三、雨为四、风为五。编号只是地址，排序规则不提供“雨比春大”的语义。秋和雪不参与建表，它们进入已有的 UNK。这个排序与特殊项协议必须与项目文件 data.py 及检查点共同保持一致。

验证文本“秋雨”编码为 [1,0,4,2]。未知比例只在原始普通字符上计数：一个未知除以两个字符，等于二分之一；不把 BOS、EOS 混入分母。UNK 使不同未见字符合并，信息已不可逆，所以不能把含 UNK 的解码结果说成原文完整还原。字符级方案的这个限制后续可以通过字节级编码改进，但比较损失时必须同步记录预测单位。

切分规则与词表规则防的是两种不同泄漏。整篇重复进入训练和测试会让测试更像记忆；从验证集建立字符表，则把验证分布的信息用于模型设计。后者不一定直接给出每题答案，却不符合这里事先固定的 train-only 协议。我们需要准确陈述是哪一种边界被突破，而不是把所有问题都笼统叫“作弊”。

算例二：把文档拆成输入、目标与有效位置

“春风”的完整编号为 [1,3,5,2]。取除最后项之外的序列作输入 X，取除第一项之外的序列作目标 Y，于是 X=[1,3,5]，Y=[3,5,2]。第零位置读 BOS 预测春；第一位置可读 BOS 和春，预测风；第二位置可读 BOS、春、风，预测 EOS。训练时整个 Y 已经存储，不代表模型允许读取它。模型的可读前缀要在注意力处继续被因果遮罩限制。

再取短文“雨”，完整编号 [1,4,2]，未填充输入 [1,4]、目标 [4,2]。项目 collate 仅右填充，补齐到三个位置后，短文输入为 [1,4,2]，目标为 [4,2,−100]，有效标记由目标不等于 −100 得到 [1,1,0]；长文有效标记为 [1,1,1]。两篇合计五次有效预测。输入右侧的二只是填充值，不新增 PAD 类别；目标二仍是必须计分的真实 EOS。不能根据 input==EOS 判断哪些位置有效。

若错误地令 Y=X，第一位置目标变成 BOS，第二位置目标变成春，网络可以学到复制当前位置即可降低损失。检测它的方法不是看数组长度，因为错误数组仍长三；应逐位置核对“前缀之后的真实下一项”。另一个失败是把两篇直接接成“春风雨”，让风之后预测雨，抹掉原文档结束。跨文档连接可以成为另一种训练约定，但不能一边删除边界一边声称在学习逐文档结束。

三、矩阵记账让每个维度都有语义

令 B 表示批次中的文档数，T 表示补齐后的预测位置数，V 表示词表项数，D 表示将来嵌入向量的宽度。X、Y、有效标记的形状均为 B×T；X 元素是合法 token 编号，Y 的有效元素是目标编号，填充元素为词表之外的 −100，有效标记为布尔值。嵌入表 E 为 V×D，按 X 查表得到 B×T×D。不能拿 Y 中的 −100 查表。未来输出层把最后一维映射为 V，未归一化预测分数 logits 的形状为 B×T×V。

本批 B=2、T=3、V=6，若暂定 D=4，则 E 有二十四个数，查表后的激活也有二十四个数，输出分数有三十六个数。这两个二十四数值相等只是巧合，参数和激活仍是不同对象。同一嵌入行可在多个位置被读取，参数只存一份。词表增加一个字符会改变 V 和输出类别数，不会让原有文档的预测位置自动变多；−100 也不应被算作第七类。

参数是跨样本共享、以后会被训练修改的数；本批输入、目标和有效标记则是样本数据，不应被优化器更新。编号置换提供一个语义检查：交换春与风的编号，并同步交换词表、嵌入行和输出类别，表示的任务不变。若程序直接把编号当成连续大小参与回归，这种等价性就会被破坏。下一章会把这个检查具体落实到嵌入查表。

还有一个概率边界必须说清：每个条件行归一化，并不单独保证无限生成最终一定遇到 EOS。一个始终把概率一分给春的过程会永不结束。这里推导的是指定有限路径的概率，尚未证明所有有限文档的概率加起来为一。实际生成要设置结束与长度上限；若到上限强制截断，应记录为截断，不能伪称模型主动预测了结束。

长文以后需要切成长度有限的窗口，但窗口结束不一定是文档结束。若原文还有下一个字符，就不应在每个窗口尾部伪造 EOS；BOS 也只表示真正文档开头，不能在任意截断前缀上冒充空历史。窗口内目标仍由原序列紧邻下一项给出，模型读取范围则受上下文长度限制。本章短文无需窗口截断，先把边界语义固定，后续扩展加载器时才能检查它有没有改变任务。

四、独立程序：建立并检查本章契约

下面只使用标准库，复刻项目 train-only 排序词表和 collate 的右填充语义，不导入 PyTorch。内容指纹与同组检查是额外的前置数据审计：SHA-256 用来核对字节内容是否变化，不证明来源许可或语句真实；示例重复副本属于同一训练组，去重后保留一份。真实 split_documents 检查文档 ID 划分，本章额外检查组与完全重复内容不跨集合；不要把语义近重复也当成 exact hash 已经解决。

import hashlib
import json

records = [
    {"id":"t1", "group":"g1", "split":"train", "text":"春风"},
    {"id":"t2", "group":"g2", "split":"train", "text":"春雨"},
    {"id":"t3", "group":"g3", "split":"train", "text":"雨"},
    {"id":"v1", "group":"g4", "split":"validation", "text":"秋雨"},
    {"id":"s1", "group":"g5", "split":"test", "text":"雪风"},
    {"id":"copy", "group":"g1", "split":"train", "text":"春风"},
]
for row in records:
    row["source"] = "课程原创短句"
    row["text"] = row["text"].replace("\r\n", "\n")
    row["sha256"] = hashlib.sha256(row["text"].encode("utf-8")).hexdigest()

group_split = {}; digest_split = {}; unique = []; seen = set()
for row in records:
    for key, registry in ((row["group"], group_split), (row["sha256"], digest_split)):
        if key in registry and registry[key] != row["split"]:
            raise ValueError("同组或相同文本跨越数据切分")
        registry[key] = row["split"]
    if row["sha256"] not in seen:
        unique.append(row); seen.add(row["sha256"])

UNK, BOS, EOS, IGNORE = 0, 1, 2, -100
characters = sorted({ch for row in unique if row["split"] == "train" for ch in row["text"]})
vocab = ["<unk>", "<bos>", "<eos>"] + characters
stoi = {ch:i for i,ch in enumerate(vocab)}
def encode(text):
    return [BOS] + [stoi.get(ch, UNK) for ch in text] + [EOS]
def make_batch(texts):
    if not texts: raise ValueError("不能拼接空批次")
    sequences = [encode(text) for text in texts]
    width = max(len(seq)-1 for seq in sequences)
    x=[]; y=[]
    for seq in sequences:
        n = len(seq)-1
        x.append(seq[:-1]+[EOS]*(width-n))
        y.append(seq[1:]+[IGNORE]*(width-n))
    valid = [[int(target != IGNORE) for target in row] for row in y]
    return x,y,valid

x,y,valid = make_batch(["春风", "雨"])
assert vocab == ["<unk>","<bos>","<eos>","春","雨","风"]
assert x == [[1,3,5],[1,4,2]] and y == [[3,5,2],[4,2,-100]]
assert valid == [[1,1,1],[1,1,0]] and sum(map(sum,valid)) == 5
assert encode("秋雨") == [1,0,4,2]
assert sum(target == EOS for row in y for target in row) == 2
for text, xr, yr, mr in zip(["春风","雨"],x,y,valid):
    seq=encode(text)
    for t,flag in enumerate(mr):
        assert flag == int(yr[t] != IGNORE)
        if flag: assert xr[t] == seq[t] and yr[t] == seq[t+1]
empty_x,empty_y,empty_valid=make_batch(["雨", ""])
assert empty_x == [[1,4],[1,2]] and empty_y == [[4,2],[2,-100]]
assert sum(map(sum,empty_valid)) == 3

rows = [[0.6,0.3,0.1],[0.2,0.5,0.3],[0.1,0.1,0.8]]
assert all(abs(sum(row)-1) < 1e-12 for row in rows)
prefix_probability = rows[0][0] * rows[1][1]
complete_probability = prefix_probability * rows[2][2]
assert abs(complete_probability-0.24) < 1e-12
assert complete_probability <= prefix_probability
vocab_hash=hashlib.sha256(json.dumps(vocab,ensure_ascii=False).encode()).hexdigest()
print("joint_probability", round(complete_probability, 6))
print("vocab", vocab)
print("X", x, "Y", y, "valid", valid)
print("valid_targets", sum(map(sum, valid)))
print("validation_unknown_rate", sum(ch not in stoi for ch in "秋雨") / len("秋雨"))
print("vocab_sha256", vocab_hash)
print("all_contract_checks_passed")

程序通过说明本章给定例子的切分、词表和移位契约一致，并不说明未来模型已经因果，也不说明去除了所有近重复。要验证检查器确实敏感，可把 copy 的 split 改成 test，应在建词表之前报错；把 Y 改成 X，应在逐位置检查失败。这些已知错误是阳性对照，避免一个总是返回通过的检查器给出虚假安全感。

五、递进练习与逐步解答

本章项目增量与下一章接口

现在保存的是带来源、组与内容指纹的文档清单，冻结的 token→编号表，以及 X、Y、有效标记的语义。还没有神经网络参数更新，也没有训练质量指标。下一章将接收 B×T 的编号和目标，定义如何把每个位置的 V 个分数变成概率，再推导目标似然、交叉熵与分数梯度；可学习嵌入将替换本章仅记账的 E。后面注意力仍需另外保证“只能读取前缀”，标签正确不能代替因果性。