一、为什么"测试时算力"值得单独长成一根研究轴
过去两年，模型能力提升的叙事几乎全部压在训练时扩展上：更大的参数量、更多的预训练 token、更长的强化学习。但到了 2025—2026 年，行业里出现了一个几乎没人能否认的共识：同一份模型权重，仅仅因为在推理阶段多花一点算力，准确率就会发生量级上的变化。变化大到什么程度？大到推理预算已经从研究技巧变成了产品契约——OpenAI 在 o 系列上把"要思考多少"做成推理强度档位，Anthropic 的扩展思考要求显式给出思考预算 token，Google 在 Gemini 2.5 上暴露思考预算参数（设为 0 表示关闭思考，设为动态值表示交给模型自己决定），Qwen3 则在混合推理模型上把思考开关做成 think 与 non-think 双模式。整个前沿产品的竞争，已经从"谁的模型大"部分转移到了"谁更会分配推理预算"。
这件事的本质是：推理不再是一次性的前向传播，而是一次在线的搜索与资源分配问题。你手上有若干算力、若干毫秒，需要给一个问题一个尽量对的答案；你可以选择采样八条独立推理链然后投票，也可以只选两条链然后逐个词元深度搜索，可以先花两百个词元探一下难度再决定要不要加钱。不同的分配方式花的是同一份预算，拿到的准确率可以差出十几个百分点。这就是测试时计算扩展成为独立研究轴的根本原因。
1.1 边际收益结构的本质差异
训练时扩展的收益曲线非常明确且令人绝望：参数量翻倍带来的收益大致遵循幂律，而且是那种早期很陡、后期非常平缓的幂律。测试时扩展的收益曲线形状不一样：它在低预算区间极其陡峭——一个原本单次通过率只有四成的题，在同一模型上做十六次采样多数投票后可能跳到七成；但它同样会饱和，只是饱和的位置不同。这条曲线的斜率、拐点和天花板，由三个完全不同的因素决定：基础策略能采到正确答案的概率，也就是覆盖；以及你有没有办法从一堆候选里认出那个正确答案，也就是判别。
把这两件事分开看，是本文所有方法论的出发点。很多人把"测试时扩展无效"和"验证器不够强"混为一谈，然后盲目加采样，结果是钱花了准确率不动。
1.2 预算已经成了一等公民参数
把各家接口的预算旋钮并排看一眼，会发现这是一个非常清晰的成熟度信号：一个能力如果稳定到可以做成参数，就说明它已经从研究技巧变成了产品契约。
| 平台与模型 | 预算参数 | 语义 | 实践要点 |
|---|---|---|---|
| Gemini 2.5 系列 | 思考预算 | 内部思考词元上限；动态值表示交给模型判断，零表示关闭思考 | 低预算下推理链会明显变短，复杂数学与代码任务的收益最先消失 |
| Claude 扩展思考 | 思考预算词元 | 思考阶段可用的词元数，有最小值约束 | 与最大输出词元是两套预算，需要一起规划，否则思考会吃掉输出配额 |
| OpenAI o 系列及后续推理模型 | 推理强度档位 | 低、中、高等离散档位映射到内部思考预算 | 这是产品化抽象，不要拿它当精确的计算量控制量 |
| Qwen3 等混合推理模型 | 思考开关与思考长度 | 同一权重上在思考与非思考两种行为间切换 | 混合模式是难度路由最省事的落地形态：简单请求走非思考，复杂请求走思考 |
这张表本身就说明了本文要讨论的问题：预算管理不再是论文里的把戏，而是每个推理服务上线后都要面对的日常工程问题。
1.3 什么情况下测试时扩展不该用
同样重要的是划出边界。以下三种情况，加预算基本是浪费：
- 任务不可判定。开放式写作、创意生成、没有标准答案的分析任务，缺少可靠的验证信号，扩展带来的多样性无法被收敛成正确性。这类场景应该把预算花在更好的评价标准上，而不是更多采样。
- 覆盖已经饱和。如果单次通过率已经很高，或者多次采样的答案完全一致，加预算的边际收益接近零。此时唯一有意义的动作是换模型或换提示。
- 延迟预算极紧。交互式场景里几十毫秒的延迟就已经不可用，此时任何扩展方案都无从谈起，只能改架构。
把这三条写在设计文档最前面，能省掉大量无谓的实验。
二、形式化：推理预算到底是什么
要谈分配，先得谈度量。这一节把推理预算拆成可计算的形式，后面的所有方法都可以看成对这组变量的不同取值策略。
2.1 三个预算维度
最粗糙的描述是"给模型更多词元就更准"，这不够用。测试时算力实际上由三个独立旋钮控制：
- 宽度：独立采样的推理链数量，或者搜索树中同时保留的分支数。
- 深度：每条链允许生成的最大词元数。
- 判别成本：验证器被调用的次数，以及单次打分需要读入的输入长度。打分通常要把整条链重读一遍，这部分开销在实践中常被严重低估。
于是总预算可以粗略写成"宽度乘深度乘生成单价，加上判别调用次数乘打分单价"。真正决定花钱的是总词元量，真正决定用户体验的是墙钟时间，这两者不是一回事，这是测试时扩展里最容易被混淆的地方。
2.2 成本看总量，延迟看最长链
并行采样十六条链、每条两千词元，总计三万两千个生成词元；但如果硬件一次只能并行处理八条，用户等待的时间由两轮中最慢的那一条决定。如果把同样的预算改成八条链、每条四千词元，总量一样，墙钟时间却可能反而变长，因为深度无法在轮内并行。反过来，如果调度器做得好，宽度可以完全被批处理吞吐吸收，而深度是硬串行的。这意味着宽度和深度对延迟的价格完全不同，深度通常更贵。
"""把预算拆成宽度 x 深度，估算成本与墙钟延迟。
总词元决定钱，批轮次与最长链决定用户等待时间。
深度在批内无法并行，因此同样的词元量下墙钟时间可能差出数倍。
"""
GEN_RATE = 120.0 # 单序列解码速率（词元/秒）
BATCH_CAP = 8 # 单请求内可并行的序列数上限
BATCH_GAIN = 0.9 # 批内每多一路序列带来的吞吐增益
def estimate(k: int, t: int) -> dict:
"""给定宽度 k 与深度 t，返回成本与延迟估算。"""
if k < 1 or t < 1:
raise ValueError("k 与 t 必须为正整数")
rounds = -(-k // BATCH_CAP) # 向上取整的批轮次
per_round = min(k, BATCH_CAP)
speedup = 1.0 + BATCH_GAIN * (per_round - 1)
rate = GEN_RATE * speedup
return {
"k": k,
"t": t,
"tokens": k * t,
"cost_k": round(k * t / 1000.0, 1), # 成本单位：千词元
"latency_s": round(rounds * t / rate, 1),
}
if __name__ == "__main__":
print(f"{'k':>4}{'t':>7}{'tokens':>9}{'cost(k)':>9}{'latency(s)':>12}")
for k, t in [(1, 2000), (8, 2000), (16, 2000), (8, 4000), (32, 1000), (4, 8000)]:
r = estimate(k, t)
print(f"{r['k']:>4}{r['t']:>7}{r['tokens']:>9}{r['cost_k']:>9}{r['latency_s']:>12}")
# k t tokens cost(k) latency(s)
# 1 2000 2000 2.0 17.5 最便宜，但只看单次通过率
# 8 2000 16000 16.0 20.4 同词元量下并行更划算
# 16 2000 32000 32.0 37.9
# 8 4000 32000 32.0 56.5 同样词元，深度把延迟拉高
# 32 1000 32000 32.0 15.1 同样词元，宽度最省墙钟
# 4 8000 32000 32.0 74.0
结论很直观：同样的算力预算，分配成宽度还是深度，墙钟延迟可以差出四到五倍。这不是理论洁癖，它直接决定你能不能把"多花算力"做成用户愿意接受的功能。
2.3 覆盖与决策：两个指标的裂口
测试时扩展的收益可以用一对指标拆开看。覆盖率衡量有没有可能做对：采样若干次，至少出现一次正确答案的概率。决策质量衡量实际会不会做对：系统最终输出的那个答案的正确率。
- 覆盖高、决策低：模型能采到正确答案，但投票或打分选不出来。瓶颈在判别信号，需要更强的验证器，或者问题本身的可判定性太差。
- 覆盖低：无论怎么分配预算都白搭。瓶颈在基础策略的搜索空间，需要回到训练侧，或者转向深度优先搜索，让模型有机会"换一条路走"。
- 两者都高：预算已经饱和，继续加钱是浪费，应该考虑换模型或换提示。
把这个裂口当成调试入口，比盯着最终准确率猜要有效得多。而且它应该成为每次实验的固定输出——裂口的方向直接决定下一个要投入的是采样器还是验证器。
2.4 预算的三种货币
最后要提醒一个常被忽略的层次：预算不只是词元。在真实系统里至少存在三种互相冲突的货币——
- 钱：主要由总生成词元与判别调用决定，与并发度无关。
- 延迟：由最长链与批轮次决定，是用户能感知的硬约束。
- 配额：在速率受限的服务里，真正稀缺的是每分钟请求数与并发路数，此时批处理效率比单请求效率更重要。
三者的最优配置经常不同。夜间批处理任务可以完全不管延迟，把预算全压到宽度上换取准确率；在线交互场景则相反，必须把预算压到深度极浅、宽度也很小的快速路径上。不存在对所有场景都最优的预算策略，只有对特定约束下最优的策略。
三、方法族一：并行采样与自一致性
这是最古老也最稳健的测试时扩展方法。Wang 等人 2022 年提出的自一致性机制非常朴素：对同一个问题采样多条推理路径，提取各自的最终答案，归一化之后做多数投票。它的价值在于完全不需要任何外部验证器，只要模型能采样、能比对答案，就能用。这也是它在生产环境里普及率最高的原因——没有额外组件就没有额外故障点。
3.1 为什么多数投票有效
自一致性起作用的机制是这样的：思维链的失败往往不是随机的，而是集中在若干条浅层捷径上——漏掉一个条件、把负号看成正号、算错一次乘法后一路错到底。这些错误模式会吸引走相当比例的样本；而正确路径虽然可能有很多条，却共享着更深的共同结构。当样本数足够时，正确的吸引子会在答案分布上累积成众数，把那些浅层错误的票数淹没掉。
这个机制有一个重要推论：自一致性的上限由"正确答案是否成为答案分布的众数"决定，而不是由"正确答案是否被采样到"决定。当多个候选答案都等价正确却没有做归一化时，正确答案的票数会被稀释，投票就会失效。这解释了一个反复出现的线上事故：模型明明答对了，但因为答案格式不统一而被投成了少数票。
3.2 采样超参的搜索：温度与截断的耦合
并行采样必须要有温度，但温度同时也在损害单条链的质量。这就是测试时扩展里最核心的张力：低温度提高单样本正确率，高温度提高候选之间的差异度。实践中零点六到零点八是数学、代码类任务的主流区间，再往上会出现大量语法错误和重复退化。
另一个常被忽略的细节：经过强化学习训练的推理模型，对温度的敏感度明显高于基座模型。这类模型在温度为零时就能产生很长、结构良好的推理链，所以先采一条零温度的"锚样本"往往质量最高；后续的探索性采样才需要加热。如果你把所有样本都设成高温，实际上是白白扔掉了一个免费的准确率来源。
截断长度也需要和温度一起调。温度高会拉长生成长度，如果截断上限设得刚好卡在分布尾部，会出现大量"推理到一半被切断"的样本，这些样本的答案提取会失败，等于浪费预算还污染投票分布。稳妥做法是把截断上限设在生成长度分布的九十五分位以上，并显式统计截断率——截断率超过百分之几就说明这个参数组合有问题。
3.3 答案归一化：最容易被低估的工程细节
投票成败的一半取决于归一化做得好不好。至少要做这几件事：统一空白与全角字符；统一数学公式的包装形式（分数写法与小数写法、是否带美元符号应该被视为同一个答案）；数值答案做浮点容差比较而不是字符串比较；单位与百分号剥离；对结构化答案按字段排序后再比较；把答案里的小时、分钟这类同义表述折叠。这些工作量不大，但直接决定投票是否可用。
还有一个隐蔽的坑：答案里包含推理过程的残留。很多模型的输出在答案后面还带着解释或自我检查，如果提取函数按"最后一个数字"或者"最后一行"来截取，会在部分样本上取到完全不同的东西。提取规则必须在多语言、多格式、多长度上分别验证，不能只在几个手挑样例上确认。
3.4 流式早停：用票数领先度提前收工
一个很实用的优化：多数投票不需要等全部链跑完。当某个答案的票数已经超过剩余可能产生的票数时，结果就已经不可能被翻盘，可以立即停止未完成的采样并返回。形式化表述就是：若已得票数为 v，已完成采样数为 d，总预算为 k，那么提前退出的充要条件是 v > k - d。
在答案分布很尖的题目上，这能把平均采样条数砍掉一半以上；在分布很平的问题上则几乎不触发——这正是我们想要的行为：只为确定的问题付确定性的钱。实现上要注意这个早停是可证明安全的，不需要额外假设，所以在任何系统里都应该默认开启。
"""自一致性采样：答案归一化 + 可证明安全的流式早停。
只依赖标准库。真实使用时把 generate_one() 换成实际模型调用即可。
"""
from __future__ import annotations
import re
from collections import Counter
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from math import isclose
# ---------- 答案归一化 ----------
def normalize_answer(raw: str) -> str:
"""把等价答案折叠成同一个字符串。漏掉任何一步都会让投票失效。"""
s = raw.strip().lower()
s = s.replace("\\left", "").replace("\\right", "")
s = re.sub(r"\\[dt]frac\{([^{}]+)\}\{([^{}]+)\}", r"(\1)/(\2)", s)
s = s.replace("^\\circ", "").replace("^{\\circ}", "").replace("\\%", "")
s = s.replace("$", "").replace("\\text{}", "").replace("\\,", "")
s = re.sub(r"[^\w\s\-\.\+/=\(\)]", " ", s)
s = re.sub(r"\s+", " ", s).strip()
try: # 数值答案：转成定长浮点串做容差折叠
return f"{float(s):.6g}"
except ValueError:
return s
def answers_equal(a: str, b: str) -> bool:
try:
return isclose(float(a), float(b), rel_tol=1e-6, abs_tol=1e-9)
except (ValueError, TypeError):
return a == b
# ---------- 投票 ----------
def vote(candidates: list[str]) -> tuple[str | None, Counter]:
"""按等价簇投票：先合并互相等价的答案，再统计簇规模。"""
clusters: list[list[str]] = []
for c in candidates:
for cl in clusters:
if any(answers_equal(c, m) for m in cl):
cl.append(c)
break
else:
clusters.append([c])
counts: Counter[str] = Counter()
for cl in clusters:
counts[max(cl, key=len)] += len(cl) # 用簇内最长串做代表
if not counts:
return None, counts
return counts.most_common(1)[0][0], counts
def can_stop(counts: Counter, done: int, budget: int) -> bool:
"""可证明安全的早停：领先票数 > 剩余可能票数，结果已不可能被翻盘。"""
if not counts:
return False
top_votes = counts.most_common(1)[0][1]
return top_votes > budget - done
# ---------- 主流程 ----------
def self_consistency(question: str, budget: int, workers: int = 8) -> dict:
counts: Counter[str] = Counter()
done = 0
with ThreadPoolExecutor(max_workers=workers) as pool:
pending = {pool.submit(generate_one, question) for _ in range(budget)}
while pending:
finished, pending = wait(pending, return_when=FIRST_COMPLETED)
for fut in finished:
counts[normalize_answer(fut.result())] += 1
done += 1
if can_stop(counts, done, budget):
for f in pending:
f.cancel()
pending = set()
break
if not counts:
return {"answer": None, "samples": done}
top, n = counts.most_common(1)[0]
return {
"answer": top,
"votes": n,
"samples": done,
"early_exit": done < budget,
"agreement": round(n / max(done, 1), 3),
}
def generate_one(question: str) -> str:
from generator import sample # 真实推理调用（需带停止条件与截断上限）
return sample(question, temperature=0.7, max_tokens=4000)
if __name__ == "__main__":
print(self_consistency("求 17 乘 23 的结果", budget=12))
# 期望形如：answer='391' votes=9 samples=9 early_exit=True agreement=1.0
3.5 自一致性的三个失效场景
自一致性不是万能的，有三类情况它会明确失效，工程上要提前识别，否则会在这些场景上持续浪费预算：
- 答案空间大或开放式：写作、摘要、多轮对话、生成一个能跑的函数，几乎没有可投票的离散答案，退化成随机采样。这种场景下"投票"这个动作本身就没有意义。
- 多解等价但未归一化：正确率高估被稀释，前面已经详细说过。这类问题表现为：加上投票之后准确率不升反降。
- 存在更强验证器时：当覆盖率已经很高（比如九成）而验证器判别力有九成五，验证器重排会明显超过多数投票。此时继续堆采样是浪费，应该把预算搬到打分上。
判断口诀：答案可离散归一化就值得投票；不可就别投，去做验证或改进评价。这条口诀的实用价值在于，它能在设计阶段就排除掉一批注定无效的方案。
四、方法族二：验证器引导的重排序
自一致性用频次代替对错，验证器重排则换了一个信号：训练或构造一个打分器，对每条候选链打分，取最高分的那条。表面上只是换一个选择规则，但这个改动把系统的瓶颈整个搬到了另一个地方——从此你不再需要"正确答案恰好是众数"，你需要一个能认出正确答案的判别器。
4.1 两种选择规则的本质差异
用一张表把自一致性和验证器重排的差异摆清楚，比任何定性的"哪个更好"都更有用。
| 维度 | 自一致性多数投票 | 验证器重排 |
|---|---|---|
| 是否需要外部打分器 | 不需要 | 需要，而且是系统核心组件 |
| 决策上限由谁决定 | 答案分布的形状：正确答案是否为众数 | 打分器的判别能力上限 |
| 典型失效 | 正确的等价答案被稀释；一致的集体错误 | 打分器给"像模像样的错误"打高分 |
| 成本结构 | 成本几乎全在生成 | 生成加打分，而打分要重读整条链，长链场景下不可忽略 |
| 与搜索的兼容性 | 弱，天生是宽度策略 | 强，既能做末端重排，也能做搜索中的剪枝启发式 |
| 可解释性 | 高：票数分布直接说明答案有多确定 | 低：分数往往不可校准，无法解释绝对值 |
这里有个容易被忽略的现实问题：打分不是免费的。打分器必须读完所有候选链，而长链可能有好几千词元，这是一次纯粹的前向计算。对长链推理任务，判别开销在总预算里可能占到可观比例。所以在决定"要不要上验证器"时，应该算的是"宽度乘以（生成长度加打分输入长度）"，而不是只算生成。
4.2 打分器的四种实现形态
- 结果奖励模型：只对最终答案训练打分器，数据便宜，标注意见只有终点，对最终正确性敏感。弱点是长链的信用分配极差——一条五十步的推理链如果第三步就走偏了，它学到的只是"这条链错了"，而不是"哪一步错了"。
- 过程奖励模型：对推理的每个步骤打局部正确分，天然可以用于搜索剪枝，信用分配更细。代价是标注昂贵，需要逐步骤标注，而且标注者对"这一步本身对，但在那个上下文里是否最优"的判断很难统一。
- 生成式验证器：不做独立的打分模型，而是让模型在提示末尾直接生成"这道题我算对了还是错了"的判定词元，把验证重新表述成一次下一词元预测。好处是验证器与生成器共享表示、可以端到端微调；坏处是同样继承生成器的偏见。
- 程序化验证：数学用符号计算，代码用单元测试，逻辑题用约束求解器，数据分析用重新执行一遍查询。这是唯一一类真正可信的验证器，也是为什么竞赛数学和代码成为测试时扩展研究的主战场——不是因为这两类任务简单，恰恰相反，而是因为它们的答案可以被机器确认。
4.3 为什么强推理模型反而倾向于不做过程监督
DeepSeek-R1 的技术报告里有一个值得反复读的决定：他们在训练阶段没有引入过程奖励模型，坚持用可验证的结果奖励做强化学习，理由之一就是过程标注需要与最终策略的数据分布保持一致，而策略在强化学习过程中持续漂移，导致过程标注的分布也随之漂移，最终过程奖励模型反而成为准确率提升的瓶颈。
这是一个来自大规模实践的反直觉结论，它不代表过程监督无用，而是说明它的维护成本在长时间训练里被严重低估了。把它翻译成测试时扩展的语言：过程奖励模型的信号会随着策略能力的提升而失效，你需要持续投入资源去重新标注、去适配、去对抗性测试。这笔开销往往超过它带来的收益。务实的结论是——如果你的推理时扩展只做末端重排，结果奖励就够；只有当你真的在做多步搜索、需要中间剪枝信号时，过程监督才变得不可替代。
4.4 验证器最危险的失效：相关性错误
如果验证器和生成器共享同一套训练数据、同一个架构、甚至只是同一个基座微调出来的，它们就会犯相关的错误：一道恰好是这类模型系统性搞错的题，生成器错一次，验证器就再错一次，重排序不但无效，甚至可能比多数投票更差——因为多数投票至少还有"群体一致性"这个弱信号兜底，而验证器提供一个看起来很精确的错误信号。
三个可操作的缓解手段：
- 让验证器与生成器异源。不同基座、不同训练数据，或者干脆用程序化验证。这是最彻底但成本也最高的做法。
- 对分数做校准。未校准的分数不能直接比较——把一批已知答案的样本送进验证器，做等渗回归或逻辑回归，把分数映射成概率。校准之后还有一个额外好处：分数本身就成了可用的停止信号。
- 把验证器当特征而不是当判官。当多个信号可用时（多数票、验证分、长度、答案出现位置、格式规范性），堆一个小的逻辑回归往往比信任单一验证器更稳。
4.5 分层重排：把预算花在刀刃上
最后是一个纯工程的优化点。验证器重排的成本随候选数线性增长，但对绝大多数候选来说打分是不必要的。可以分两层：先用便宜模型大规模生成并用一个廉价启发式粗排，只对去重后的前若干个不同答案用强模型重生成和高精度打分。这能把判别成本压下来一大截，而且对最终准确率影响很小。
"""验证器重排：分层生成 + 并行打分 + 校准阈值停止。
1. 分层生成：便宜层开大 k，强模型只重跑粗排后得分最高的 m 个不同答案
2. 并行打分：判别是纯前向，天然可批处理
3. 校准停止：分数达到阈值直接返回，把剩余预算还给系统
"""
from __future__ import annotations
import math
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
@dataclass
class BoNConfig:
k_cheap: int = 16 # 便宜层采样条数
m: int = 6 # 强模型只重跑粗排后前 m 个不同答案
k_strong: int = 4 # 每个目标答案的强模型重生成条数
tau: float = 0.86 # 校准后的接受阈值
verify_workers: int = 8
@dataclass
class Candidate:
text: str
answer: str
tier: str
score: float = 0.0 # 原始打分
prob: float = 0.0 # 校准后的置信概率
def calibrate(raw: float) -> float:
"""把原始分映射成概率。真实做法是对标注样本做等渗回归，此处简化为逻辑函数。"""
return 1.0 / (1.0 + math.exp(-(raw - 3.0)))
def best_of_n(question: str, cfg: BoNConfig) -> dict:
cands: list[Candidate] = []
# ---- 第 1 层：便宜、宽、并行 ----
for text, ans in sample_many(question, cfg.k_cheap, tier="cheap"):
cands.append(Candidate(text=text, answer=ans, tier="cheap"))
# ---- 廉价粗排：只保留前若干个"不同答案"，避免在同一个答案上重复投入 ----
ranked = sorted(cands, key=lambda c: -c.score)
top_answers = []
for c in ranked:
if c.answer not in top_answers:
top_answers.append(c.answer)
if len(top_answers) >= cfg.m:
break
# ---- 第 2 层：强模型重生成，验证"这个答案是否真的站得住" ----
for ans in top_answers:
for text, a in sample_many(question, cfg.k_strong, tier="strong", hint=ans):
cands.append(Candidate(text=text, answer=a, tier="strong"))
# ---- 并行打分 + 校准 ----
with ThreadPoolExecutor(max_workers=cfg.verify_workers) as pool:
scores = list(pool.map(lambda c: raw_score(c, question), cands))
for c, s in zip(cands, scores):
c.score, c.prob = s, calibrate(s)
cands.sort(key=lambda c: -c.prob)
best = cands[0]
gap = best.prob - cands[1].prob if len(cands) > 1 else 1.0
# 双信号早停：不仅要分高，还要明显甩开第二名（否则说明打分器没区分度）
stopped = best.prob >= cfg.tau and gap >= 0.10
return {
"answer": best.answer,
"prob": round(best.prob, 4),
"runner_up_gap": round(gap, 4),
"stopped_early": stopped,
"candidates": len(cands),
}
def raw_score(cand: Candidate, question: str) -> float:
from verifier import score # 真实打分器（奖励模型/生成式验证器/单元测试）
return float(score(cand.text, question))
def sample_many(question: str, k: int, tier: str, hint: str | None = None):
from generator import sample
for _ in range(k):
text = sample(question, tier=tier, hint_answer=hint)
from scoring import extract_answer
yield text, extract_answer(text)
if __name__ == "__main__":
r = best_of_n("求 1234567 的各位数字之和", BoNConfig())
print(r)
# 关注 runner_up_gap：若它很小，说明打分器在这题上没区分能力，
# 此时即便绝对分很高也不应该早停
注意输出里的 runner_up_gap。如果第一名和第二名的校准概率只差零点零二，说明打分器在这道题上基本没有区分能力，此时基于"分数很高"的早停是不可靠的——高分可能只是打分器整体偏高，而不是真的确信。这是一个实用且几乎零成本的校准自检指标，建议每个模型版本都跑一遍。
五、方法族三：过程奖励与结果奖励的取舍
这一节值得单独拎出来，因为它常被误判。默认"过程监督比结果监督更先进"，于是什么都上过程奖励模型，然后被维护成本拖垮。两个信号的本质差异可以压缩成一句话：结果奖励告诉你这条路通不通，过程奖励告诉你这一步该往哪拐。前者便宜稳健，后者精细但难养。
5.1 逐项对比
| 维度 | 结果奖励模型 | 过程奖励模型 |
|---|---|---|
| 监督信号来源 | 最终答案对错 | 每一步是否正确、是否有效推进 |
| 数据成本 | 低：自动生成轨迹，只标注终点 | 高：需要逐步骤标注，人力密集且标准难统一 |
| 信用分配粒度 | 整条链一个信号，粒度极粗 | 步骤级，适合长链与剪枝 |
| 抗分布漂移 | 强：只关心终点，不关心推理风格 | 弱：标注与特定推理风格强耦合，策略一变标注就过时 |
| 在搜索中的角色 | 末端重排 | 每步剪枝的局部启发式 |
| 典型失效 | "错一步但答对"与"绕远路但答对"无法区分，学到表面模式 | 奖励攻击：学会输出让打分器满意的形式，而非真正推理 |
把这两者的选择简化成一条决策规则：只做末端重排就用结果奖励，要做多步搜索才需要过程奖励。这条规则在绝大多数实际系统中都是对的。
5.2 过程奖励确实是正解的三个场景
过程奖励不是没用，而是在特定条件下性价比更高：
- 搜索树的剪枝。做多步搜索时，中间状态的评价信号只能来自过程奖励，没有别的来源。如果一个分支的扩展代价是几百词元，你不可能给每个中间节点都跑一遍完整验证。
- 长链推理的失败定位。一条一百步的推理链答错，你需要知道是第几步歪的，只有过程奖励提供这个信号。这对数据清洗和难例挖掘尤其有用。
- 蒸馏与监督数据筛选。用过程分筛出高质量的中间步骤，比只筛最终答案更容易得到干净的监督数据。
而当你只做末端重排序、链长不长、又有大量自动标注能力时，结果奖励是压倒性的正确选择。
5.3 步骤分数如何聚合成一个信号
过程奖励给出的是步骤级分数，搜索需要一个标量。聚合方式直接影响剪枝行为：取最小值最保守，任何一步崩了整条就崩，适合做硬剪枝；乘积会让长链天然吃亏，适合做归一化后的排序；首错惩罚显式区分"早期就错"和"晚期才错"，前者通常更不可救。实践中最稳妥的组合是：用最小值做剪枝阈值，用对数乘积做排序，两者分工明确。
"""过程奖励信号的聚合与提前终止。step_scores 中 None 表示该步未打分。"""
from __future__ import annotations
from math import exp, log
def min_aggregate(step_scores: list[float | None]) -> float:
"""硬剪枝信号：任一步崩则整链崩。对推理早期的错误最敏感。"""
vals = [s for s in step_scores if s is not None]
return min(vals) if vals else 0.0
def logprod_aggregate(step_scores: list[float | None], eps: float = 1e-3) -> float:
"""排序信号：取对数乘积并对链长归一化，避免长链被系统性惩罚。"""
vals = [max(s, eps) for s in step_scores if s is not None]
return exp(sum(log(v) for v in vals) / len(vals)) if vals else 0.0
def first_error_index(step_scores: list[float | None], threshold: float = 0.5) -> int | None:
"""定位首错步骤：用于长链失败归因，也是"是否值得继续救"的判据。"""
for i, s in enumerate(step_scores):
if s is not None and s < threshold:
return i
return None
def prune_decision(step_scores: list[float | None], threshold: float = 0.35) -> dict:
"""在搜索树扩展前调用：决定这条部分解是否还值得继续扩展。"""
agg = min_aggregate(step_scores)
bad = first_error_index(step_scores)
return {
"expand": agg >= threshold,
"aggregate": round(agg, 4),
"first_error_at": bad,
# 早错通常不可救，晚错还有回退空间 —— 这个区分直接决定要不要继续投入
"recoverable": bad is None or bad >= len(step_scores) * 0.6,
}
if __name__ == "__main__":
cases = {
"健康": [0.9, 0.88, 0.91, 0.9, 0.87],
"早错": [0.9, 0.92, 0.2, 0.1, 0.05],
"晚错": [0.9, 0.9, 0.9, 0.88, 0.15],
}
for name, s in cases.items():
d = prune_decision(s)
print(f"{name}: 可扩展={d['expand']} 聚合分={d['aggregate']} "
f"首错位于第{d['first_error_at']}步 可救={d['recoverable']}")
# 早错：聚合分 0.05，首错在第 3 步，不可救 —— 应立即剪枝并回溯
# 晚错：聚合分 0.15，首错在第 5 步，可救 —— 聚合分几乎相同，
# 但只有这一条值得再投入预算去救。不区分两者就是在浪费算力
5.4 别把过程奖励当唯一真相
即使用了过程奖励，也建议保留一条兜底路径：让链走到终点做程序化或结果验证。因为步骤标注必然覆盖不全、必然有噪声，而终点验证是客观的。一个稳健的组合是：过程奖励负责过程剪枝，终点验证器负责最终选择。这在拓扑上恰好对应"深度搜索加宽度重排"的混合策略，也就是下一节要讲的内容。
六、自适应难度路由：把预算花在刀刃上
前面几节都在讨论给定预算怎么分配，这一节讨论一个更前置的问题：预算本身该给多少。如果所有问题都拿到同样的预算，那么平均来看你一定浪费了——因为不同问题的收益曲线斜率天差地别。
6.1 难度与最优采样的经验关系
这是测试时扩展里最实用的一条经验规律：简单问题用少量采样就饱和了，困难问题才需要大量采样。一道模型本来就会的题，采四次和采六十四次的准确率几乎一样；而一道八次里蒙对一次的题，采六十四次可能把准确率推高一倍多。所以"简单题四样本、难题六十四样本"这种非均匀分配，收益远超"全部题八样本"。
这条规律的直接推论是：测试时算力的最优分配是逐题目的，而不是全局均匀的。这也解释了为什么固定采样数的方案总有明显改进空间——它系统性地在简单题上超支、在难题上不足。
6.2 三种实现路径
- 级联与早停：最简单也最鲁棒。先用小预算跑一轮，如果答案一致或验证分够高就直接返回，只有触发阈值才升级预算。实现复杂度低，天然带熔断语义，是生产环境的第一选择。
- 难度预测：训练一个难度分类器，或者用小样本的覆盖率斜率、首个词元的熵、答案长度等廉价特征预测难度，再据此预先分配采样数。理论最优，但难度预测本身很难做准——模型"觉得难"和"实际难"之间存在系统性偏差，而且这种偏差会随模型版本漂移。
- 在线调度：把预算当成队列资源，题目进来先给最便宜的探针，用采集中途的不确定性（票数分布有多尖、分歧有多大）决定是否继续投资。这是最贴合真实服务的形态，因为它只需要当前批次的信息，不需要预知难度。
6.3 在线调度的一个陷阱：自信的错误
早停机制有一个真实的风险：如果模型错得非常自信（所有采样都一致地给出同一个错误答案），早停会立刻固化这个错误。而一致的错误恰恰是推理模型最难的情况之一——当训练数据里某个错误模式很常见时，模型会在该模式下高度一致地犯错。
缓解手段有两个，都值得实现：
- 给早停设一个最小预算地板：无论投票多确定，至少跑若干条样本再允许返回。这牺牲一点简单题上的效率，换掉一类无法自愈的错误。
- 早停用双门槛：投票一致性与验证分数必须同时达标。单一信号（无论票数还是分数）都可以被一致性的错误骗过，两个独立信号的联合门槛要可靠得多。
6.4 预算调度的完整实现
下面的调度器把上面的几条原则都落进了代码：探针优先、不确定性升级、最小地板、双门槛早停，以及一个贯穿全流程的硬时间限制。它的结构可以直接搬到真实推理服务里。
"""自适应预算调度器：探针 -> 不确定性判定 -> 升级。
1. 先用最小探针（便宜），只有"值得"的题才追加预算
2. 早停是双门槛（投票一致性 AND 校准分数），防自信错误
3. 任何情况下都设最小地板，避免单样本决定答案
4. deadline 是唯一的硬约束，贯穿每一轮循环
"""
from __future__ import annotations
import time
from collections import Counter
from dataclasses import dataclass, field
@dataclass
class ScheduleConfig:
probe_k: int = 2 # 第一档：便宜探针
medium_k: int = 6 # 第二档：升档后补采
final_k: int = 16 # 第三档：难题取满
min_floor: int = 2 # 最小地板：任何题至少采这么多
agreement_stop: float = 0.85 # 投票一致性早停线
verify_stop: float = 0.82 # 校准分数早停线
uncertain_gap: float = 0.25 # 前二名票差小于此值 = 不确定，需升档
deadline_sec: float = 25.0 # 硬预算
def should_escalate(counts: Counter, done: int, cfg: ScheduleConfig) -> tuple[bool, str]:
"""根据当前投票分布判断是否值得追加预算。"""
if done < cfg.min_floor:
return True, "尚未达到最小地板"
ranked = counts.most_common(2)
agree = ranked[0][1] / max(done, 1)
gap = (ranked[0][1] - ranked[1][1]) / max(done, 1) if len(ranked) > 1 else 1.0
if agree >= cfg.agreement_stop and gap >= cfg.uncertain_gap:
return False, f"票数已确定 一致率={agree:.2f} 领先={gap:.2f}"
if gap < cfg.uncertain_gap:
return True, f"候选接近、结果不确定 领先={gap:.2f}"
return True, f"一致性不足 一致率={agree:.2f}"
def adaptive_solve(question: str, cfg: ScheduleConfig | None = None) -> dict:
cfg = cfg or ScheduleConfig()
t0 = time.monotonic()
counts: Counter[str] = Counter()
stages: list[dict] = []
k_now = cfg.probe_k
while True:
left = cfg.deadline_sec - (time.monotonic() - t0)
if left <= 0.2:
stages.append({"stage": "deadline_cut"})
break
# 剩余时间不足以跑完当前档位 -> 降到跑得起的规模，绝不返回半截结果
k_now = max(1, min(k_now, int(left / 0.15)))
counts.update(sample_stage(question, k_now))
done = sum(counts.values())
stop, reason = should_escalate(counts, done, cfg)
top = counts.most_common(1)[0][0] if counts else None
prob = calibrated_score(top, question)
if stop and prob >= cfg.verify_stop:
stages.append({"stage": "early_stop", "reason": reason, "samples": done})
break
if done >= cfg.final_k or time.monotonic() - t0 > cfg.deadline_sec * 0.8:
break
k_now = cfg.medium_k if done < cfg.medium_k else cfg.final_k
if k_now <= done:
break
done = sum(counts.values())
ranked = counts.most_common()
return {
"answer": ranked[0][0] if ranked else None,
"agreement": round(ranked[0][1] / max(done, 1), 3) if ranked else 0.0,
"total_samples": done,
"saving": round(1 - done / cfg.final_k, 2) if cfg.final_k else 0.0,
"elapsed_sec": round(time.monotonic() - t0, 2),
"stages": stages,
}
def sample_stage(question: str, k: int) -> list[str]:
from generator import sample_batch
from scoring import normalize_answer
return [normalize_answer(a) for a in sample_batch(question, k, temperature=0.7)]
def calibrated_score(answer: str | None, question: str) -> float:
if answer is None:
return 0.0
from verifier import score
from scoring import calibrate
return calibrate(float(score(answer, question)))
if __name__ == "__main__":
print(adaptive_solve("求 17 乘 23 的结果"))
print(adaptive_solve("证明存在无穷多个素数"))
# 前者应在探针档就早停（saving 明显大于 0），后者应一路升档到 final_k
这个调度器最重要的性质是把成本分布的右偏压平了：简单题几乎不花钱，预算自然流向难题。这比"给所有题一个更大的采样数"在同等平均预算下效果好得多，因为后者在简单题上花的每一分钱都是零边际收益。
6.5 队列级的预算公平性
单请求层面的调度解决不了另一个问题：当一批请求共享一份预算时，如何分配。一个困难请求可能吃掉整批预算，而同期有一堆简单请求在排队。工程上有三种常见策略：按预估难度预分配（最省事，但依赖难度预测）；先到先得加硬上限（最简单，可预测性最好，但简单请求多时总预算用不掉）；按剩余队列长度动态调整（总预算利用率最高，但延迟可预测性最差）。
实践中的经验是：对延迟敏感的服务用先到先得加硬上限，对离线批量任务用动态调整。因为前者用户能感知的是延迟稳定性，后者用户根本不在线，吞吐才是唯一目标。
七、深度优先与宽度优先：搜索拓扑的选择
到这里为止的策略都是宽度优先——横向铺开多条独立路径。那么纵向深挖是否更好？这是一个真实的权衡，没有免费答案。
7.1 两种拓扑的对比
| 维度 | 宽度优先（独立采样加投票或重排） | 深度优先（束搜索、树搜索、蒙特卡洛树搜索） |
|---|---|---|
| 路径关系 | 各链完全独立 | 节点是部分解，边是续写，前缀可共享 |
| 词元效率 | 低：多条链重复生成相同前缀 | 高：共享前缀摊薄重复计算 |
| 延迟特征 | 好：墙钟时间由批轮次决定 | 差：树的深度串行，尾部延迟高 |
| 实现复杂度 | 低：无状态，易水平扩展 | 高：需要节点缓存、回溯、剪枝、并发状态 |
| 对打分的依赖 | 弱：只需末端重排 | 强：每步剪枝都依赖打分质量 |
| 擅长 | 探索：跳出局部最优、覆盖多种解法 | 利用：沿一条思路反复深入 |
| 主要失效 | 覆盖不到正确的解法方向 | 早期误剪枝，且错误在深链上不可见 |
OpenAI 那篇关于测试时计算最优扩展的论文把这个权衡讲得很清楚：对简单问题，独立采样就够了；对困难问题，搜索才带来净收益——但这个收益依赖于打分器的强度，打分器不够强时深搜反而不如宽搜。论文中还有一个更反直觉的观察：在算力充足时，束搜索若没有足够强的打分器支撑，效果反而不如朴素的独立采样加重排。
这解释了一个常见的观测：同一个搜索算法在较弱模型上表现平平，换成更强的推理模型后突然变好。原因是模型的单步质量提高后，深搜每一步走错的概率降低了，误剪枝的代价随之下降。搜索算法和基础模型是一对耦合的系统，不能脱离模型单独评估。
7.2 混合策略：先散后聚
生产系统很少二选一，几乎都是混合：宽度探索、宽度收敛、深度精化。先用少量独立采样摸清问题空间（哪类解法有效、答案是否唯一、前缀是否稳定），再对最有希望的解法做深度验证或局部搜索。这种"先散后聚"的形态比纯深搜鲁棒得多，因为深搜的第一步往往是最贵的一步，一旦选错方向，后面所有词元都白花。
7.3 深度优先的最小实现
束搜索是深度优先里最实用的形态。关键在于前缀共享——树搜索的词元效率优势几乎全部来自这里。好的实现会把部分解组织成前缀树，让共享前缀的键值缓存复用，而不是每个节点独立重跑。
"""束搜索（束宽度 b），演示深度优先骨架 + 前缀共享。
与验证器重排的本质区别：
重排：k 条独立链 -> 末端打分
束搜：b 个活跃节点，每步只扩展少数候选 -> 中间剪枝 -> 末端打分
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class Node:
text: str # 已生成的部分推理
step: int
score: float # 累计启发式分
parent: "Node | None" = None
dead: bool = False
def expand(node: Node, continuations: list[str], scorer) -> list[Node]:
"""把一个部分解扩展成若干子节点，并立即打分剪枝。"""
kids = []
for cont in continuations:
child = Node(text=node.text + cont, step=node.step + 1, parent=node)
child.score = scorer(child.text) # 局部打分，剪枝靠它
child.dead = child.score <= 0.0 # 死枝：剪掉，不再展开
kids.append(child)
return kids
def beam_search(problem, width: int, max_steps: int, continuations_fn, scorer) -> Node | None:
"""束搜索主循环。剪枝发生在每一步，而不是只在末端。"""
beam: list[Node] = [Node(text="", step=0, score=1.0)]
expanded = 0
for _ in range(max_steps):
candidates: list[Node] = []
for node in beam:
if node.dead:
continue
kids = expand(node, continuations_fn(problem, node), scorer)
expanded += len(kids)
candidates.extend(kids)
if not candidates:
break
# 叶子判定：某个候选已给出完整答案
finished = [c for c in candidates if problem.is_complete(c.text)]
if finished:
return max(finished, key=lambda c: c.score)
candidates.sort(key=lambda c: -c.score)
beam = candidates[:width]
return max(beam, key=lambda c: c.score) if beam else None
def compare_cost(k: int, width: int, steps: int, branch: int, prefix_share: float) -> dict:
"""对比两种策略在同一扩展预算下的词元消耗。
prefix_share 是"共享前缀能省下的比例"，它完全取决于实现质量，
而不是理论值 —— 这正是束搜索在实践中常被低估的原因。
"""
bon_tokens = k * steps # 独立采样：每条链从头生成
tree_tokens = width * branch * steps * (1 - prefix_share) # 束搜：共享前缀摊薄
return {
"bon_tokens": bon_tokens,
"tree_tokens": int(tree_tokens),
"saving": round(1 - tree_tokens / bon_tokens, 3),
}
if __name__ == "__main__":
# 关键：把 prefix_share 从理想值调到贴近现实值，结论会翻转
for share in [0.0, 0.15, 0.4]:
r = compare_cost(k=8, width=4, steps=20, branch=4, prefix_share=share)
print(f"前缀共享率={share:.0%}: 独立采样={r['bon_tokens']} 词元, "
f"束搜索={r['tree_tokens']} 词元, 节省={r['saving']:.0%}")
# 前缀共享率为 0（没有做请求内前缀树）时，束搜索没有任何词元优势，
# 只剩下"更早暴露错误"这一个好处 —— 而它依赖强打分器
7.4 一个反直觉的结论：前缀共享的收益比理论小得多
理论上束搜索靠前缀共享省词元，但实测中这个收益经常远小于预期。三个原因：
- 推理模型分叉得早且分叉深。模型通常在前十几步就决定了走哪条路，而每条路的后续词元占整链的大部分，能共享的前缀其实很短。
- 实现层面的门槛很高。很多推理服务的前缀缓存只在请求之间生效，同一次请求内的树内共享需要专门的前缀树数据结构支撑。做不对就是零收益，甚至因为额外调度开销变成负收益。
- 批处理会打断共享。为了吞吐把节点打散到不同批次，前缀就很难命中缓存。
所以务实的判断是：如果你没有做真正的请求内前缀树，束搜索的词元优势基本不存在，剩下的只有"中间剪枝更早暴露错误"这一个好处，而这个好处依赖强打分器。所以在做深度优先的决策之前，先诚实地评估自己的前缀复用能力。
八、计算最优的推理时扩展规律
前面讲的都是怎么做，这一节讲规律是什么。有几个结论已经相当稳固，值得作为设计决策的直接依据。
8.1 规律一：算力应该按难度非均匀分配
最核心的经验规律。对一个固定的题目，简单问题的最优采样数很小，困难问题的最优采样数很大；如果被迫用一个全局统一的采样数，准确率必然低于一个知道该给谁多分配的调度器。这也意味着"我们的系统用八次采样做自一致性"这句话本身就隐藏了巨大的优化空间。
8.2 规律二：验证器质量是天花板
这条规律可以这样表述：测试时扩展的上限由验证器决定，而且验证器带来的收益比采样器更陡。直觉解释是——采样最多只能采到模型能力范围内的答案，但验证器决定了这些答案里你能不能挑出对的那个；一旦覆盖足够高，唯一的瓶颈就变成了判别。此时继续加大采样毫无用处，把预算全投到验证器上才是正解。
这也解释了为什么程序化验证在测试时扩展的研究里地位特殊：它把验证器从概率模型变成了确定性判定，等于直接消除了整个瓶颈。竞赛数学之所以成为主战场，正是因为答案可以被符号计算确认。
8.3 规律三：收益递减是结构性的，不是随机波动
准确率随采样的提升大致是对数线性的，每多翻一倍采样带来的增量持续下降。但比"会饱和"更值得注意的是饱和的形状：多数投票的饱和往往比人们期望的早，因为当正确答案不是众数时再采一百条也救不回来；而验证器重排的饱和更晚，因为它不依赖众数，只依赖打分器能识别。这两种饱和点位置不同，正是实际选择方法时的主要依据。
8.4 规律四：训练与测试时扩展存在相乘关系
这一点常被忽略但极其重要：经过强化学习训练的推理模型，其测试时扩展的收益显著高于基座模型。原因在于测试时扩展的空间由基础策略的覆盖能力决定——如果模型压根采不到正确答案，采样再多也没用；而推理模型经过训练后，单次采样的质量和多样性都大幅提升，可搜索的空间自然变大。
换句话说，训练时扩展和测试时扩展不是竞争关系，而是相乘关系：训练决定搜索空间有多大，测试时扩展决定在空间里找得多好。这也意味着，只在测试时扩展上发力而忽视训练质量，天花板很快就到了。反过来，一个 mediocre 的基座模型无论怎么加预算，也很难追上一个强推理模型的简单调用。
8.5 一个更完整的收益曲线模型
把上面几条整合起来，一个朴素的收益模型是：准确率的提升等于覆盖率的增长乘以验证器从候选中挑出正确答案的概率。这个分解的价值在于它直接告诉你该往哪加预算。
"""测试时扩展的收益分解模型。
准确率(宽度 k) ≈ P(候选中至少有一次正确 | k) x P(判别环节挑出正确的 | 存在正确候选)
两个因子都随 k 单调递增，但增长速度不同，且第二个因子有硬上限。
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class StrategyParams:
name: str
p_first_correct: float # 单次采样就正确的概率（基础质量）
spread: float # 搜索性：值越大，多样性带来的覆盖增长越快
verifier_top1: float # 验证器在候选含正确答案时的挑出率（硬天花板）
maj_quality: float = 1.0 # 多数投票的上限系数：正确答案成为众数的概率
def coverage(k: int, p: float, spread: float) -> float:
"""候选集合含正确答案的概率。采样变多时单条变弱，故按多样性折算。"""
p_k = min(0.95, p * (1.0 + spread * (k ** 0.5 - 1.0) * 0.1))
return 1.0 - (1.0 - p_k) ** k
def majority_accuracy(k: int, prm: StrategyParams) -> float:
"""多数投票：正确答案必须是众数，且需要攒够基本票数。"""
cov = coverage(k, prm.p_first_correct, prm.spread)
need = max(2, int(k * 0.4)) # 成为众数所需的基本票数
return cov * prm.maj_quality * (1 - need / k * 0.3)
def bon_accuracy(k: int, prm: StrategyParams) -> float:
"""验证器重排：上限完全由验证器决定，与众数无关，因此饱和更晚。"""
return coverage(k, prm.p_first_correct, prm.spread) * prm.verifier_top1
def decide(k: int, prm: StrategyParams) -> dict:
maj, bon = majority_accuracy(k, prm), bon_accuracy(k, prm)
return {
"k": k,
"coverage": round(coverage(k, prm.p_first_correct, prm.spread), 3),
"maj": round(maj, 3),
"bon": round(bon, 3),
"winner": "验证器重排" if bon > maj else "多数投票",
}
if __name__ == "__main__":
base = StrategyParams("推理模型", 0.35, 1.2, 0.82, 0.78)
print(f"{'宽度':>5}{'覆盖':>8}{'多数投票':>10}{'验证器重排':>12} 赢家")
for k in [1, 2, 4, 8, 16, 32, 64, 128]:
r = decide(k, base)
print(f"{k:>5}{r['coverage']:>8.3f}{r['maj']:>10.3f}{r['bon']:>12.3f} {r['winner']}")
# 三个结构性观察：
# 1) 覆盖率永远低于 1，堆采样买到的边际覆盖极其有限
# 2) 多数投票的损失来自众数约束，在 k=2、k=4 时伤害最重
# 3) 两种方法的差距在 k=8~32 这段（默认预算区间）最大
weak = StrategyParams("弱验证器", 0.35, 1.2, 0.55, 0.78)
print(f"\n换成弱验证器后，宽度 16 的结论：{decide(16, weak)}")
# 验证器一变弱，多数投票反而胜出 —— 验证器质量直接改写方法选择结论
8.6 收益模型的三种典型饱和形态
把上面的曲线按形状归类，能得到一个实用的诊断工具：
- 快速饱和型：覆盖率低但多数投票涨得快，然后很快持平。典型是简单任务或强基座模型。特征是十次以内就饱和，说明继续采样纯属浪费。
- 慢速爬升型：覆盖率增长持续，但决策质量始终上不去。典型是"采到了但选不出"。特征是多数投票与验证器重排都远离覆盖率的上限，说明该投验证器。
- 天花板受限型：两条曲线都贴着验证器上限走。典型是验证器质量不足。特征是重排结果几乎等于覆盖率乘一个固定的天花板系数。
每次实验都判断一下属于哪一类，能省下大量盲目加预算的时间。
九、工程实现：预算控制、超时与熔断
理论到落地之间隔着的全是工程问题。一个只在离线评测里跑得通的测试时扩展方案，上线后通常死于超时、成本失控和级联故障。这一节讲的全部是"活下来"所需的东西。
9.1 预算控制的三个层次
- 全局预算：给一批请求分配的总算力。这需要调度器在请求之间做公平性分配，否则一两个困难请求会把整个批次的预算吃光。
- 单请求预算：每个问题允许消耗的词元与毫秒上限。这是唯一的硬约束，必须由调用方或服务保证。
- 阶段预算：在单请求预算内，为每个阶段（探针、升档、终局验证）预留子预算，防止某个阶段无节制地吃掉全部时间。
最容易犯的错误是只设全局预算不设阶段预算。典型后果是：升级到高档采样后单条链变长，一条链的解码时间就吃掉了全部剩余预算，最终只能返回一条截断的、不完整的推理链——这样的结果不仅没用，而且比不扩展更糟，因为它把一个本来会正常回答的问题变成了截断失败。截断导致的失败在离线评测里几乎不会出现，因为它跑在低延迟的开发机上。
9.2 超时降级：三级回退
超时处理不是简单地"超时就报错"，而应该有明确的降级链条，且每一级都必须返回可用的东西：
- 完整策略：完整的探针加升档加验证流程。
- 收敛即返：不再升档，用已有采样做多数投票返回。这是最常用的一级，绝大多数请求在这里就解决了。
- 截断即返：单链解码超时，强制停止并输出已生成的部分内容。如果这部分内容里已经出现了答案（很多推理模型把答案放在结尾标记之后），照样能用。
关键设计原则：降级路径必须在测试中被真实触发过。很多团队的缺陷不是"降级逻辑写错了"，而是"降级逻辑从没被触发过，因为测试环境的延迟太低"。要主动在注入延迟的环境里跑一次降级测试，把超时阈值临时调到极小值。
9.3 熔断：五类必须能自动切断的故障
测试时扩展引入了长时间、大批量、循环调用，天然是熔断友好的场景：
- 成本熔断：单请求词元数、单用户累计词元、单批次总预算超限即停止升级。
- 时间熔断：墙钟限制。必须在每一层循环里检查，不能只在入口检查——一个深度优先的搜索循环如果不在每步检查，超时会变成无限挂起。
- 质量熔断：推理链进入退化循环时切断。这是一类纯粹由测试时扩展引入的新故障模式：单次生成不会无限重复，但"生成、打分、再生成"的循环会陷进去。
- 依赖熔断：验证器服务超时或返回异常时，降级为纯多数投票。注意这里必须是"降级继续"而不是"整体失败"，因为投票路径不依赖验证器。
- 并发熔断：扩展会把单请求的并发路数放大数倍，在流量高峰时容易把推理集群打满。这个需要在网关层做并发整形。
"""带完整熔断与降级的推理执行框架。
降级链条：完整策略 -> 收敛即返（多数投票）-> 截断即返 -> 失败
每一级都必须返回可用结果，只有最后一层才允许返回空。
"""
from __future__ import annotations
import time
from collections import Counter
from dataclasses import dataclass, field
from enum import Enum
class Level(Enum):
FULL = "完整策略"
MAJORITY = "收敛即返"
TRUNCATED = "截断即返"
FAILED = "失败"
class Breaker(Enum):
OK = "正常"
COST = "成本超限"
TIME = "时间超限"
DEGENERATE = "推理退化"
DEPENDENCY = "依赖故障"
@dataclass
class Guard:
max_tokens: int = 60_000 # 单请求词元上限
deadline_sec: float = 30.0 # 墙钟限制
deg_window: int = 40 # 退化检测窗口（词元数）
deg_ratio: float = 0.75 # 窗口内重复四元组占比阈值
events: list[Breaker] = field(default_factory=list)
def trip(self, b: Breaker) -> None:
if b is not Breaker.OK:
self.events.append(b)
def has_degeneracy(text: str, window: int = 40, ratio: float = 0.75) -> bool:
"""检测推理退化：最近窗口内出现大量重复四元组。
这是测试时扩展特有的故障：单次解码不会这样，
但"生成->打分->再生成"的循环会陷进去。
"""
words = text.split()
if len(words) < window:
return False
tail = words[-window:]
grams = [tuple(tail[i:i + 4]) for i in range(len(tail) - 3)]
return len(set(grams)) / len(grams) < (1 - ratio)
@dataclass
class Result:
answer: str | None
level: Level
breakers: list[Breaker]
tokens_used: int
elapsed: float
degraded_reason: str | None = None
def solve_with_guards(question: str, guard: Guard) -> Result:
t0 = time.monotonic()
g = Guard(max_tokens=guard.max_tokens, deadline_sec=guard.deadline_sec,
deg_window=guard.deg_window, deg_ratio=guard.deg_ratio)
tokens = 0
candidates: list[str] = []
# ---------- 完整策略主循环 ----------
try:
for k in (2, 6, 16):
if time.monotonic() - t0 > g.deadline_sec:
g.trip(Breaker.TIME)
break
if tokens > g.max_tokens * 0.8:
g.trip(Breaker.COST)
break
batch = sample_with_deadline(question, k, g, t0)
if batch is None:
g.trip(Breaker.TIME)
break
cands, used = batch
tokens += used
candidates.extend(cands)
if has_degeneracy(" ".join(candidates[-3:]), g.deg_window, g.deg_ratio):
g.trip(Breaker.DEGENERATE)
break
# 验证器故障 -> 降级为纯投票，而不是整体失败
try:
score = verify(candidates)
except (TimeoutError, ConnectionError):
g.trip(Breaker.DEPENDENCY)
score = None
answer, confident = decide(candidates, score)
if confident:
return Result(answer, Level.FULL, g.events, tokens,
time.monotonic() - t0)
except Exception:
g.trip(Breaker.DEPENDENCY)
# ---------- 降级一：纯多数投票 ----------
if candidates:
return Result(majority(candidates), Level.MAJORITY, g.events, tokens,
time.monotonic() - t0, "降级为多数投票")
# ---------- 降级二：单链截断 ----------
partial = sample_truncated(question, max_tokens=1500, deadline=g.deadline_sec)
if partial:
return Result(extract_answer(partial), Level.TRUNCATED, g.events, tokens,
time.monotonic() - t0, "使用截断的部分推理")
return Result(None, Level.FAILED, g.events, tokens, time.monotonic() - t0)
# ---- 外部依赖（示意） ----
def sample_with_deadline(q, k, g, t0):
from generator import sample_batch
return sample_batch(q, k, deadline=g.deadline_sec - (time.monotonic() - t0))
def verify(cands):
from verifier import verify_batch
return verify_batch(cands)
def decide(cands, score):
from scoring import is_confident_enough
return majority(cands), (score is not None and is_confident_enough(cands, score))
def majority(cands):
return Counter(cands).most_common(1)[0][0] if cands else None
def sample_truncated(q, max_tokens, deadline):
from generator import sample_one_forced
return sample_one_forced(q, max_tokens, deadline)
def extract_answer(text):
from scoring import extract_answer as ex
return ex(text)
if __name__ == "__main__":
# 测试技巧：把 deadline 调到 0.01 秒，必然触发降级，逐级验证返回可用性
r = solve_with_guards("证明存在无穷多个素数", Guard(deadline_sec=30.0))
print(f"等级={r.level.value} 熔断={[b.value for b in r.breakers]} "
f"词元={r.tokens_used} 耗时={r.elapsed:.1f}秒 原因={r.degraded_reason}")
# 降级路径必须在注入延迟的测试里真实跑过，不能只测正常路径
9.4 前缀缓存：最被低估的优化
并行采样有一个巨大的浪费：多条链共享同一个问题前缀，但每次请求都重新预填充一遍。对于长系统提示加长文档的检索场景，这个预填充成本乘以采样数后可能超过解码本身。主流推理框架都提供了自动前缀缓存，让相同前缀复用键值缓存，这在并行采样场景下几乎是免费的性能提升。
启用它的成本是零，收益在长提示场景下非常可观。这是本文推荐优先级最高的一个工程优化——它排在所有算法改动之前。
9.5 批处理与并发的取舍
并行采样天然适合连续批处理：多条链的解码阶段都是访存受限的，批在一起能显著提升单卡吞吐。但有两个反作用力：一是批内一条长链会拖住整个批次的完成时间（尾延迟问题），二是批越大，显存压力越高，可并发路数下降。实践中的调优方向是在请求内保持小批、在跨请求上做大批：把同一个问题的采样放在一个批次里，不同问题之间再合并。
9.6 成本可观测性
有一个容易被漏掉的点：成本不是平均值问题，而是尾部问题。典型分布是绝大多数请求花很少的钱，少数难题吃掉大部分预算。运维上应该监控每请求成本的九十五和九十九分位而不是均值，并为高成本请求设置单独配额。一个没有单请求成本上限的测试时扩展系统，在流量波动时一定会出事故。
9.7 建议的监控指标
测试时扩展会让同一个问题产生不同质量的结果，所以监控维度必须包含预算。至少要有这几个：按预算档位分别统计的准确率、平均消耗词元、平均延迟、降级触发率、退化检出率、验证器不可用比例、以及每请求成本的尾部分布。如果只看总体准确率，你会在"简单题省下的钱"和"难题加的钱"之间互相掩盖问题。
十、评测：怎么公平地衡量测试时扩展
这一节看起来最枯燥，但它决定了你能不能得出可信的结论。测试时扩展的评测有几个特有的坑，踩中任何一个，结论都会反转。
10.1 必须报告的指标
| 指标 | 衡量什么 | 为什么不能少 |
|---|---|---|
| 覆盖率 | 采样中至少出现一次正确答案的概率 | 它是所有其他指标的上界，不测它就无法判断瓶颈在采样还是判别 |
| 多数投票准确率 | 投票决策的真实质量 | 实际会部署的策略的质量 |
| 验证器重排准确率 | 判别环节介入后的质量 | 与投票对比才知道验证器是否值得投入 |
| 平均词元与延迟 | 成本 | 没有成本的准确率是不可比的 |
| 饱和曲线 | 边际收益何时归零 | 决定该把预算停在哪里 |
| 降级触发率 | 系统的真实有效策略 | 被熔断砍掉的请求，其质量也要计入 |
其中最该先看的是覆盖率与决策质量之间的裂口，它直接告诉你下一步该做什么：裂口大说明决策端有空间，去改验证器；裂口小说明决策端接近饱和，去改采样或训练。
10.2 评测陷阱清单
- 单次通过率与单次投票混用。这两者在定义上几乎一样，但很多实现里数字对不上，通常是因为答案提取逻辑不一致。
- 成本不可比。拿单次调用和三十二次投票比准确率却不报告词元差异，等于宣称"多花三十倍算力能提升一成"——这不是有意义的结论。
- 答案等价判定过严。数学等价形式、代码等价实现、等价查询语句都被判错，会系统性低估投票和重排的效果。这一项的错误往往比方法差异造成的影响更大，而且方向固定——总是低估扩展方法。
- 温度不一致。基线通常用零温度，多采样用更高温度，这个差异本身贡献了相当一部分"提升"，需要单独报告。
- 只测容易题。难题上测试时扩展的收益最大，只在简单题上评测会得出"扩展没用"的错误结论。
- 验证器在同分布数据上训练又在同分布评测，得到的是被严重高估的验证器质量，进而导致过度投入验证器。
- 忽略上下文与前缀缓存的影响。开启缓存前后延迟差异很大，两组结果不能混在一起比较。
10.3 预算与准确率的前沿曲线
单一预算点的准确率几乎没有信息量。有信息量的是前沿曲线：横轴是总预算，纵轴是准确率。两条曲线（不同方法）的上包络才是真正的性能边界。注意必须用相同总预算比较，而不是相同采样数——因为打分有开销，相同采样数的实际成本可以差很多。
"""构建预算-准确率前沿曲线，比较不同策略的性价比。"""
from __future__ import annotations
def evaluate_policy(dataset, policy, budgets) -> list[dict]:
"""在多个预算档位上评测同一策略，返回 (预算, 准确率, 成本) 曲线。"""
curve = []
for b in budgets:
correct = cost = 0
for item in dataset:
out = policy(item["question"], budget=b)
correct += int(out["answer"] == item["gold"])
cost += out["tokens"]
curve.append({
"budget": b,
"accuracy": round(correct / len(dataset), 4),
"avg_tokens": round(cost / len(dataset), 1),
})
return curve
def frontier(curves: dict[str, list[dict]]) -> list[dict]:
"""取上包络：在每个预算档位上找出最准的策略。"""
budgets = sorted({p["budget"] for c in curves.values() for p in c})
envelope = []
for b in budgets:
best = None
for name, c in curves.items():
for p in c:
if p["budget"] == b and (best is None or p["accuracy"] > best[1]):
best = (name, p["accuracy"], p["avg_tokens"])
if best:
envelope.append({"budget": b, "best_strategy": best[0],
"accuracy": best[1], "avg_tokens": best[2]})
return envelope
def cost_normalized(curve: list[dict], baseline: float) -> list[dict]:
"""把准确率增益折算成每百万词元换来多少准确率，用于跨策略比性价比。"""
out = []
for p in curve:
gain = p["accuracy"] - baseline
out.append({**p, "gain": round(gain, 4),
"gain_per_Mtok": round(gain / (p["avg_tokens"] / 1e6), 3)})
return sorted(out, key=lambda x: -x["gain_per_Mtok"])
if __name__ == "__main__":
# 示意数据：投票准确率随预算快速饱和，重排饱和更晚但起步更贵
curves = {
"多数投票": [
{"budget": 1, "accuracy": 0.351, "avg_tokens": 1200},
{"budget": 8, "accuracy": 0.612, "avg_tokens": 9600},
{"budget": 32, "accuracy": 0.683, "avg_tokens": 38400},
{"budget": 128, "accuracy": 0.694, "avg_tokens": 153600},
],
"验证器重排": [
{"budget": 1, "accuracy": 0.351, "avg_tokens": 1200},
{"budget": 8, "accuracy": 0.701, "avg_tokens": 13400},
{"budget": 32, "accuracy": 0.768, "avg_tokens": 43000},
{"budget": 128, "accuracy": 0.771, "avg_tokens": 155000},
],
}
print("预算前沿（相同总预算下取上包络）：")
for p in frontier(curves):
print(f" 预算={p['budget']:>4} 最佳={p['best_strategy']:<10} "
f"准确率={p['accuracy']:.3f} 平均词元={p['avg_tokens']}")
print("\n性价比排序（每百万词元的增益）：")
for p in cost_normalized(curves["多数投票"], 0.351)[:3]:
print(f" 预算={p['budget']:<4} 增益={p['gain']:.3f} 性价比={p['gain_per_Mtok']}")
# 结论形态：低预算段两种策略性价比接近；中预算段重排反超；
# 高预算段都饱和，继续加钱是纯浪费 —— 预算应该停在中段
十一、测试时算力与训练时算力的对比
一个自然的追问：既然同样的算力可以花在训练上，为什么不直接训一个更强的模型？
11.1 两者的本质差异
| 维度 | 训练时扩展 | 测试时扩展 |
|---|---|---|
| 本质 | 把能力压进权重 | 把能力临时借出来 |
| 成本结构 | 一次性、大额、长期摊销 | 每次调用都付，边际成本线性 |
| 摊销性 | 调用量越大越划算 | 调用量越大越亏，没有摊销 |
| 能否处理无标签任务 | 能，这正是训练的本质 | 不能，需要可判定的验证信号 |
| 失败模式 | 过拟合、能力不泛化、训练不稳定 | 预算浪费、延迟飙升、验证器被钻空子 |
| 迭代周期 | 周级 | 小时级，可当天上线 |
| 天花板 | 训练数据与信号的质量 | 覆盖率乘以验证器 |
11.2 经济学的判断准则
决策其实相当机械。设任务调用量为 Q，训练一次成本为 C，每次推理的额外成本为 c，两者带来的准确率增益分别是 Δ训练 和 Δ推理，那么边际等价的判断式很简单：训练摊薄后的单位成本低于测试时的单位成本，才值得训。
- 调用量极大的通用能力（客服问答、通用助手）：摊薄效应显著，测试时扩展的边际成本永远追不上。这类场景应当优先投入训练。
- 长尾难题、领域定制、任务量小：摊薄不成立，重训的固定成本无法被摊薄，测试时扩展的性价比极高。这就是为什么垂直领域的小团队用推理模型加测试时算力，就能 在很多任务上打赢自训基座。
- 有确定性验证器的高价值任务（数学证明、代码修复、数据清洗）：测试时扩展在这里的收益最确定，因为验证信号完美且免费。这是投入产出比最高的场景。
还有一个常被忽略的维度：迭代速度。训练时扩展的反馈周期是周，测试时扩展是小时。在快速变化的需求面前，一个今天就能通过调预算和改调度器拿到的收益，往往比三个月后的大模型训练更有价值。这是工程现实，不是技术优劣。
11.3 它们最好是组合使用
最有效的路径是让两者分工：训练负责让覆盖率高、让单次质量高，测试时扩展负责把能力榨到极致。前面提到的"相乘关系"就是指这个。2026 年的共识已经很清楚：强推理模型加上好的测试时扩展策略，远好于任何单侧优化。
十二、局限与失效模式
任何方法都有边界。这一节列出测试时扩展已知的失效模式，把它们写清楚比宣称普适性更有价值，因为线上事故基本都由其中之一导致。
12.1 验证器被钻空子
如果验证器是学出来的模型，它就有可被利用的模式。用测试时扩展去最大化验证器得分，实际上是在优化一个代理目标，系统会自发地学会找到"能骗过验证器但不正确"的解。在数学竞赛数据上这一点尤其明显：模型会找到绕过验证的表述方式。所以验证器的设计必须持续对抗性测试，并优先选择不可被模仿的程序化验证。
12.2 准确率不涨，成本翻倍
最常见的"扩展无效"场景：投票高度一致但一致地错，或者验证器在难题上失效导致所有候选得分接近。前者表现为增加采样完全无收益（增量恰好为零）；后者表现为重排退化到随机选择。正确的应对是把预算从采样侧搬到判别侧，而不是继续加采样。判断依据是前面说的裂口：覆盖率低就去改采样或训练，覆盖率高但决策差就去改验证器。
12.3 延迟不可接受
深度优先搜索、大量采样、验证器调用都会拉长延迟。在交互式场景里几十毫秒的延迟就已经不可用，此时扩展策略必须重新设计：压缩深度、限制采样数，或者改为异步返回初版答案再迭代更新。最后一种做法在产品上很常见——先快速返回一个可能不那么完美的答案，然后用扩展结果覆盖。
12.4 长度通胀与成本失控
一个隐蔽的机制：如果用长度相关的启发式给过程分数打分，或者按总词元分配预算，系统会学会写得更长。这在早期是有效的（长推理确实更准），但最终变成纯粹的成本浪费。监控时要单独看"每个正确答案的平均词元数"，而不只是看准确率。
12.5 难度路由的误判
自适应预算依赖难度估计，估计错误有两种代价：难题被误判为简单（该加钱时没加，直接损失准确率），简单题被误判为难（浪费钱）。前者严重得多。所以最小预算地板是必要的，而"用便宜模型先探测一下"是值得的投入。
12.6 任务不可验证时的天花板
最终的边界：当任务既没有可验证的答案，也没有高质量的评分模型时，测试时扩展基本无效。这时可用的只有并行采样带来的"集思广益"效应，而且往往提升有限。这不是实现问题，是信息论层面的限制——如果没有任何信号能告诉你哪个答案更好，更多候选只是让选择变得更随机。
12.7 评测过拟合
一个容易被忽略的风险：在验证集上反复调预算策略，本身就是在过拟合这套验证集。特别是当你用覆盖率裂口来决定投采样还是投验证器时，会不自觉地朝着验证器强的方向倾斜。缓解办法是在一个从未参与调参的保留集上做最终确认，并对预算策略的改动保持克制。
十三、一份可落地的实施清单
把前面所有内容压缩成一个可以照着做的顺序。顺序很重要——每一步都是下一步的前提，跳步会让后面的努力白费。
- 先做基线：把单次采样的准确率、词元、延迟测清楚，作为一切比较的基准。没有基线的所有比较都是无意义的。
- 确认任务是否可验证：能否用单元测试、符号计算或确定性规则判定答案？能，直接上程序化验证；不能，先建打分器。这一步的结论直接决定后面所有投入的方向。
- 实现答案归一化与多数投票：成本最低、收益最确定的一步。很多团队跳过它直接上复杂方法，然后发现准确率不涨，回来才发现是归一化的问题。
- 开启前缀缓存：零成本收益，尤其是长提示场景。它排在所有算法改动之前。
- 实现成本与时间熔断：在扩展开启之前就先有熔断，否则第一次事故会很贵。
- 再考虑验证器重排：先测覆盖率与投票准确率的裂口，裂口大才值得投入验证器。
- 最后做自适应调度：在固定预算跑通并且指标可信之后，再优化预算分配。否则你是在噪声上调参。
- 全链路监控：按预算档位看准确率、词元、延迟、降级率，重点看成本的尾部分布。
- 回归测试降级路径：主动注入延迟和验证器故障，确保降级链路真的能返回可用结果。
- 在保留集上确认：所有调参结束后，在一个从未参与调参的集合上做最终确认，防止评测过拟合。
结语：测试时扩展是一种思维方式的转变
测试时扩展的核心价值在于，它把模型能力从"权重里有多少"变成了"调度做得多好"。当预算控制、难度路由、验证器质量和搜索拓扑都可以持续优化时，推理就从一个推理问题变成了一个搜索与调度问题。这个转变带来的收益是结构性的：它不需要重新训练就能兑现，而且随着调度策略变好，收益还能持续叠加。
但它同样有明确的边界：需要可判定的验证信号，成本随调用量线性增长，在无验证器的任务上很快触顶。理解这些边界，比记住任何一个具体技巧更重要。
真正的工程判断力体现在知道什么时候该加预算、什么时候该停、什么时候该回去改训练——这三点构成了测试时扩展的全部实践智慧。一个只会无脑加采样阈值的系统，最终会在成本曲线上撞墙；一个懂得在准确率饱和时主动停手的系统，才能长期健康地运行下去。
