开始前,会这些就够了
- 知道概率之和为一
- 能区分条件概率与整个流程的发生概率
- 能运行只使用标准库的Python脚本
一个便宜的草稿模型先提出候选词,目标模型决定收不收。如果拒绝,重新按目标模型抽一次,看起来相当稳妥;但接受分支已经留下了一部分结果,再把完整目标分布加进来,会让某些词拿到超过应有的份额。错误藏在两个分支合起来的账里。
本篇固定同一段上下文,只研究下一个词的单步概率。我们不加载模型、不测推理速度,也不随机抽取大量样本;程序用精确分数枚举每条可能分支,保存输入表、汇总JSON和分支CSV。这个小实验用来说明推测采样中的拒绝修正,完整多词解码还有其他步骤。
把例子带到自己的电脑上
下载完整练习包
需要 Python 3.10 或更新版本;制作时实测 3.14.0。使用标准库,不需要安装第三方包。下载 ZIP 后选择“全部提取”,进入能看到 README.md 的文件夹。
下载 ZIP · 5 个文件包里有什么
- speculative_ledger.py:完整示例
- README.md:环境、范围与独立练习
- observed.stdout.txt:制作时的观察记录
- verification.json:对应源码的验证范围
- SHA256SUMS.txt:文件内容摘要
在这个文件夹打开 PowerShell,逐行运行
python --version
python -B speculative_ledger.py解压后在含示例文件的文件夹打开 PowerShell,按命令运行。先预测结果,再与当次输出对照。 程序会在当前目录写入教学报告,建议使用独立练习文件夹。 用精确分数枚举单步、三项分布的概率路径;不执行真实语言模型,也不测量推理速度或多步缓存。
找不到命令或文件时,先按 README 检查环境和所在目录。文中标出的阅读时间只估算文章篇幅,安装环境和动手练习可按自己的节奏进行。
先约定两个模型各想抽到什么
词表只有A、B、C。目标概率p为[0.5,0.3,0.2],草稿概率q为[0.2,0.6,0.2]。两者都针对同一段已经确定的上下文,词的顺序和含义一致。p是我们最终希望保留的分布,q是实际先用来提出候选的分布;它们不是两次随机抽样得到的频率。
可以把一次输出的全部机会想成总量为一的预算。最终A应分到0.5,B应分到0.3,C应分到0.2。草稿给B的机会偏多、给A的机会偏少,所以不能直接照收全部草稿。也不能用“两个模型最喜欢的词一样”代替整个概率表的比较。
本文始终用p表示目标、q表示草稿。第二篇参考论文使用相反字母,阅读时应按目标和草稿的角色对应,不能直接照搬符号。这里的概率已是实际采样要用的分布;若另有温度或截断处理,应先明确处理后的概率是什么。
接受概率要乘上提出这个词的概率
草稿提出词i后,以min(1,pᵢ/qᵢ)的概率接受。对A,目标想要0.5、草稿只给0.2,提出的A全部接受;对B,只接受其中一半;对C,也全部接受。因此三个条件接受概率依次为1、0.5、1。条件在这里很具体:已经知道草稿提出的是哪一个词。
把它乘回草稿提出该词的概率,才得到整个流程中“提出并接受它”的份额:A是0.2,B是0.6×0.5=0.3,C是0.2。三个份额之和为0.7,剩余0.3走拒绝分支。B的条件接受率0.5和全流程接受B的概率0.3不是同一个数字。
一般地,接受份额就是min(pᵢ,qᵢ)。它不会超过目标想要的份额,因此留下的是尚未补足的部分,而不是一份需要从头重做的完整目标预算。我们接下来只应把缺的部分补上。
拒绝后按完整目标分布重抽,会重复分配
错误方案在0.3的拒绝分支内,重新按p抽取。这个分支给A、B、C新增的全流程份额分别是0.3×0.5=0.15、0.3×0.3=0.09、0.3×0.2=0.06。与已经接受的份额相加,最终得到[0.35,0.39,0.26]。它仍然是合法概率分布,却已经不是目标[0.5,0.3,0.2]。
原因是B和C的目标份额早已由接受分支凑齐,重新按完整p分配又给了它们一部分。A原本缺0.3,却只补到0.15。检查概率是否非负、总和是否为一,都发现不了这个错误;还要逐项检查最终分布是否等于事先指定的目标。
这不表示错误方案在每一组输入上都会出错。例如两个分布完全一样时根本不会拒绝;它们的非零位置完全分开时,所有草稿都被拒绝,直接从目标重抽恰好正确。程序保留这些边界例,避免把一个明确反例写成过强的全称结论。这里讨论的是保留前文min(1,pᵢ/qᵢ)接受规则后,只更改回退分布的方案;不能把结论推广到采用其他接受规则的拒绝采样。
修正分布只在尚未补足的位置分配
先逐项计算max(pᵢ−qᵢ,0),本例得到[0.3,0,0];再除以它们的总和0.3,得到拒绝之后使用的条件分布[1,0,0]。这通常称为残差分布:意思就是还缺多少。拒绝发生后,本例只能用A替换;加回接受份额,最终恰好是[0.5,0.3,0.2]。
这个算式可以逐项证明。令R为总拒绝概率,它等于1−Σmin(pᵢ,qᵢ),也等于Σmax(pᵢ−qᵢ,0)。当R大于零,修正分布rᵢ=max(pᵢ−qᵢ,0)/R。最终概率为min(pᵢ,qᵢ)+R×rᵢ,也就是min(pᵢ,qᵢ)+max(pᵢ−qᵢ,0)=pᵢ。
CSV中的拒绝路径还会保留原草稿候选,让你看到“拒绝B,再补A”是一条联合事件。更换词表或概率后,缺额可能分布在多个词上;脚本第五组输入就有两个可补的位置,不能把本例只能补A误当成普遍规则。
两个零值边界不能靠除法硬算
如果某个qᵢ为零,草稿永远不会提出这个词,就不需要计算它的条件接受率。报告把该位置记为null,而不是强行计算pᵢ/0。目标仍可给这个词正概率,缺少的份额可以通过修正分支补回;因此草稿不覆盖某个目标词,并不自动使这个单步概率构造失效。
如果R为零,则p与q逐项相同,全部草稿都会接受。此时修正分布不需要存在,报告同样记为null,程序不会把全零缺额再除以零。无效输入则是另一回事:负概率、总和不为一、长度错误、无法解析为有理数的字符串和非字符串输入会被拒绝,不会被悄悄改成某个合法分布。
运行精确账本,检查每条路径怎样合起来
把下面完整脚本保存到空目录,用Python 3运行。它只用标准库Fraction保存有理数,所以0.3以3/10参与计算,不需要用浮点容差判断相等。五组输入覆盖主要反例、完全相同、草稿有零概率、非零位置完全分开,以及两个残差候选。
probability-inputs.json记录原始概率;sampling-ledger-report.json给出条件接受率、接受份额、拒绝概率和两种最终结果;probability-branches.csv列出所有正概率路径。零概率路径不写入CSV,但完整词表仍在汇总中。每组输入、每种方法的路径总和都必须为一。
验收时从CSV独立按最终输出词求和,再与目标表比较,而不是只看程序打印PASS。它枚举的是有限词表中的概率分支,既没有执行随机模拟,也没有测量真实模型的抽样频率;这些结果的证据范围应与运行方式保持一致。
展开收起代码 · 小实验
长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。
"""Exact one-token probability ledger; no model calls or random simulation."""
import csv
import json
from fractions import Fraction as F
from pathlib import Path
TOKENS = ("A", "B", "C")
CASES = (
("main", ("1/2", "3/10", "1/5"), ("1/5", "3/5", "1/5")),
("equal", ("1/2", "3/10", "1/5"), ("1/2", "3/10", "1/5")),
("draft_zero", ("1/2", "3/10", "1/5"), ("0", "4/5", "1/5")),
("disjoint", ("1", "0", "0"), ("0", "1", "0")),
("two_residual_tokens", ("1/2", "2/5", "1/10"), ("1/5", "1/5", "3/5")),
)
def distribution(values):
if len(values) != len(TOKENS) or any(type(x) is not str for x in values):
raise ValueError("provide exactly three rational-number strings")
result = tuple(F(x) for x in values)
if any(x < 0 for x in result) or sum(result) != 1:
raise ValueError("probabilities must be nonnegative and sum exactly to one")
return result
def exact_strings(values):
return [None if x is None else str(x) for x in values]
def enumerate_case(name, target_values, draft_values):
p, q = distribution(target_values), distribution(draft_values)
acceptance = tuple(min(F(1), pi / qi) if qi else None for pi, qi in zip(p, q))
accepted = tuple(qi * ai if ai is not None else F(0) for qi, ai in zip(q, acceptance))
rejected = tuple(qi - ai for qi, ai in zip(q, accepted))
rejection = sum(rejected)
deficit = tuple(max(F(0), pi - qi) for pi, qi in zip(p, q))
assert sum(deficit) == rejection
residual = tuple(x / rejection for x in deficit) if rejection else None
rows, totals = [], {}
for method in ("correct_residual", "wrong_redraw_target"):
final = [F(0) for _ in TOKENS]
for i, candidate in enumerate(TOKENS):
if accepted[i]:
final[i] += accepted[i]
rows.append({"case": name, "method": method, "candidate": candidate,
"path": "accept", "output": candidate, "mass": str(accepted[i])})
if rejected[i]:
fallback = residual if method == "correct_residual" else p
assert fallback is not None
for j, output in enumerate(TOKENS):
joint_mass = rejected[i] * fallback[j]
if joint_mass:
final[j] += joint_mass
rows.append({"case": name, "method": method, "candidate": candidate,
"path": "reject_then_replace", "output": output,
"mass": str(joint_mass)})
assert sum(final) == 1
totals[method] = tuple(final)
assert totals["correct_residual"] == p
assert totals["wrong_redraw_target"] == tuple(a + rejection * pi for a, pi in zip(accepted, p))
report = {"case": name, "tokens": list(TOKENS), "target_p": exact_strings(p),
"draft_q": exact_strings(q), "conditional_acceptance": exact_strings(acceptance),
"accepted_mass": exact_strings(accepted), "rejected_mass_by_candidate": exact_strings(rejected),
"rejection_probability": str(rejection), "positive_deficit": exact_strings(deficit),
"residual_given_rejection": exact_strings(residual) if residual is not None else None,
"correct_final": exact_strings(totals["correct_residual"]),
"wrong_final": exact_strings(totals["wrong_redraw_target"]),
"wrong_equals_target_in_this_case": totals["wrong_redraw_target"] == p}
return report, rows
def expect_invalid(values):
try:
distribution(values)
except (ValueError, ZeroDivisionError):
return
raise AssertionError("invalid distribution was accepted")
def main():
reports, rows = [], []
for name, target, draft in CASES:
report, case_rows = enumerate_case(name, target, draft)
reports.append(report)
rows.extend(case_rows)
main_report = reports[0]
assert main_report["accepted_mass"] == ["1/5", "3/10", "1/5"]
assert main_report["rejection_probability"] == "3/10"
assert main_report["residual_given_rejection"] == ["1", "0", "0"]
assert main_report["wrong_final"] == ["7/20", "39/100", "13/50"]
assert reports[1]["residual_given_rejection"] is None
assert reports[2]["conditional_acceptance"][0] is None
assert reports[3]["rejection_probability"] == "1"
assert reports[3]["wrong_equals_target_in_this_case"]
assert reports[4]["residual_given_rejection"] == ["3/5", "2/5", "0"]
for invalid in (("0", "0", "0"), ("-1", "1", "1"), ("1/2", "1/2"),
("1/2", "1/2", 0), ("nan", "0", "1"), ("1/0", "0", "0")):
expect_invalid(invalid)
# Independently sum all exported branches by case, method, and final token.
for report in reports:
for method, key in (("correct_residual", "correct_final"), ("wrong_redraw_target", "wrong_final")):
subset = [r for r in rows if r["case"] == report["case"] and r["method"] == method]
reconstructed = [sum((F(r["mass"]) for r in subset if r["output"] == t), F(0)) for t in TOKENS]
assert exact_strings(reconstructed) == report[key]
output = Path.cwd()
(output / "probability-inputs.json").write_text(json.dumps([
{"case": name, "target_p": list(p), "draft_q": list(q)} for name, p, q in CASES
], indent=2) + "\n", encoding="utf-8")
(output / "sampling-ledger-report.json").write_text(json.dumps({
"scope": "Exact one-token finite probability enumeration; no random draws or model inference.",
"zero_convention": "null acceptance means q=0; null residual means rejection probability is zero.",
"cases": reports
}, indent=2) + "\n", encoding="utf-8")
with (output / "probability-branches.csv").open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=("case", "method", "candidate", "path", "output", "mass"))
writer.writeheader()
writer.writerows(rows)
print(json.dumps({"status": "PASS", "cases": len(reports), "branch_rows": len(rows),
"main_correct": main_report["correct_final"], "main_wrong": main_report["wrong_final"]}))
if __name__ == "__main__":
main()
怎样运行:保存为 speculative_ledger.py,在空练习目录运行 python speculative_ledger.py。只用Python 3标准库,在当前目录生成两个JSON和一个CSV。 2026-10-15 已用 3.14.0 执行这份示例并通过自带断言;下方为当次输出。用精确分数枚举单步、三项分布的概率路径;不执行真实语言模型,也不测量推理速度或多步缓存。
运行后对照结果
{"status": "PASS", "cases": 5, "branch_rows": 37, "main_correct": ["1/2", "3/10", "1/5"], "main_wrong": ["7/20", "39/100", "13/50"]}概率保持成立,不等于已经证明推理更快
Leviathan等人的论文把这个修正放进多词推测解码:草稿先提出一段续写,目标模型验证各位置,首次拒绝处修正;如果全部接受,还会从后续目标分布产生额外词。本文只抽出固定上下文下一个候选的概率核心,没有实现这些序列控制、缓存维护或额外词步骤。
Transformers v4.57.1的_speculative_sampling函数也包含接受比值、首次拒绝和非负差值归一化。这里固定阅读这个版本的源码,不把它说成最新版本,也没有实际调用框架函数。真实实现还要处理数值精度、停止条件及模型输入,这个分数账本不能替代相关测试。
保持分布表示每种结果的概率符合目标,并不保证使用同一个随机种子时逐词输出完全相同。实际是否加速,还取决于草稿成本、接受率、目标验证、硬件与调度。我们的实验只验证一个概率规则,不输出任何吞吐或延迟结论。
记住这些要点
- 条件接受率乘上草稿概率,才是全流程中的接受份额。
- 拒绝后应补剩余缺额;直接按目标重抽可能重复分配概率。
- q为零的位置不用计算接受比值,拒绝率为零时不用构造修正分布。
- 保持输出分布不等于相同随机种子下输出逐词相同。
- 单步精确枚举说明概率正确性,不证明真实推理加速。
别把例子用过头
- 把条件接受率与全流程接受概率混为一谈。
- 只验证输出概率总和为一,却不逐项比较目标。
- 对零草稿概率或零拒绝率直接做除法。
- 把两篇论文中的p、q字母照抄,却忽略目标和草稿的角色相反。
- 把标准库概率账本称为完整推测解码实现或大模型性能测试。
自己试一小步
换成四种词后,你能用独立账本审查两种回退规则吗?
- 在独立目录自行选定两个四项概率分布,至少让两个位置存在正缺额;先写清目标和草稿各是哪一组。
- 扩展输入检查与分支导出,分别保存修正回退和直接重抽目标的所有正概率路径。
- 另写一个只读取CSV的核对程序,按最终词汇总并逐项比较目标;同时保留一个无拒绝边界例。
- 提交概率表、两个程序、CSV和结论,说明哪些发现来自精确枚举,哪些真实解码环节没有实现。
做到这些,就算完成
- 四项输入先固定,核对程序不依赖主程序的汇总字段。
- CSV能追溯草稿候选、接受或拒绝路径、最终词与联合概率。
- 结论不把某一组巧合正确推成普遍正确,不包含未测量的加速数字。
查原始资料
想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。
- Fast Inference from Transformers via Speculative Decoding (新标签页)↗
正文2.2与2.3的采样分布、接受规则及残差修正,算法1中的多词控制背景。
- Accelerating Large Language Model Decoding with Speculative Sampling (新标签页)↗
算法2和4.2节中的拒绝修正、归一化与概率保持边界。
- Transformers v4.57.1:推测采样函数源码 (新标签页)↗
_speculative_sampling中p/q接受比值和首次拒绝后clamp(p−q)再归一化的实现。