推論時拡張を深掘り：思考予算、適応的サンプリング、並列探索 — AI開発チュートリアル

1. なぜ「テスト時計算」は単独の研究軸として育つのか

この2年間、モデル能力の向上を語る物語はほぼすべて「訓練時スケーリング」に乗っていた。より大きなパラメータ数、より多くの事前学習トークン、より長い強化学習のループ。ところが2025〜2026年に業界には、ほぼ誰も否定できない共通認識が現れた。同じモデルの重みであっても、推論フェーズで計算を少しだけ増やしただけで、精度は桁違いに変化する。どれだけ変わるのか。その変化の大きさは、推論予算が研究技法から製品契約へと変わるほどだった。OpenAI は o シリーズで「どれだけ考えるか」を推論強度（reasoning effort）の段階表現にし、Anthropic の extended thinking は思考予算トークンを明示的に提示し、Google は Gemini 2.5 で思考予算パラメータを公開した（0 なら思考を off、動的な値ならモデル自身に判断を委ねる）。Qwen3 はハイブリッド推論モデルで思考スイッチを think と non-think の2モードとして実装した。frontier 製品の競争軸は「どのモデルが大きいか」から、部分的に「誰が推論予算の配分を上手できるか」へ移った。

このことの本質はこういうことだ。推論は一度きりの順方向伝播ではなく、オンラインの探索とリソース配分の問題になった。手元には一定の計算力と一定のミリ秒数がある。その範囲のなかで、1つの問題にできるだけ正しい答えを返さなければならない。8本の独立した推論チェーンをサンプリングして多数決に回すこともできるし、チェーンを2本だけ選んで1トークンずつ深く探索することもできる。まず200トークン使って難易度を探り、追加投資をするかどうかを決めてから動くこともできる。配り方だけが違うのに使う予算は同じなのに、手に入る精度は10ポイント以上も違う。これが、テスト時計算のスケーリングが独立した研究軸として立ち上がった根本の理由である。

1.1 限界収益の構造がまったく違う

訓練時スケーリングの収益曲線はきわめて明確で、しかも絶望的だ。パラメータ数を2倍にしても得られる収益はほぼべき乗則に従う。しかも初期は急勾配、後期にはきわめて平坦になる。テスト時スケーリングの収益曲線は形が違う。低予算域ではきわめて急勾配だ——もともと1回の通過率が4割しかない問題でも、同じモデルで16回サンプリングして多数決を取れば7割まで跳ね上がることがある。当然、飽和もする。ただし飽和する位置が違う。この曲線の傾き・変曲点・天井を左右するのは、まったく異なる三つの要素である。ベース方策が正解をサンプルできる確率、つまりカバレッジ。および、いったん出た候補の中から正解を見分けられるかどうか、つまり判別である。

この二つを分けて考えることが、本文のすべての方法論の出発点になる。「テスト時スケーリングが効かない」と「検証器が弱い」を混同して、いたずらにサンプリングを増やしている人が多い。結果はコストを使うだけで、精度は一向に変わらない。

1.2 予算はすでに一等市民のパラメータになっている

各社のAPIにある予算ノブを横並びに見ると、これはきわめて明確な成熟度のシグナルだとわかる。ある機能が安定したうえでパラメータとして出せるほど、その機能は研究技法から製品契約に変わったということになる。

プラットフォームとモデル	予算パラメータ	セマンティクス	実践上のポイント
Gemini 2.5 シリーズ	思考予算	内部の思考トークン上限。動的な値はモデルに判断を委ね、0 は思考を off にする意味になる	低予算では推論チェーンが明らかに短くなり、複雑な数学・コードタスクの収益が真っ先に消える
Claude extended thinking	思考予算トークン	思考に使えるトークン数。最小値の制約がある	最大出力トークンとは別の予算なのでセットで計画しないと、思考が出力枠を食ってしまう
OpenAI o シリーズ以降の推論モデル	推論強度の段階	低・中・高の離散段階を内部の思考予算にマッピングする	製品としての抽象化であり、精密な計算量の制御値として使ってはいけない
Qwen3 などのハイブリッド推論モデル	思考スイッチと思考の長さ	同じ重みのうえで「思考する」挙動と「思考しない」挙動を切り替える	ハイブリッドモードは難易度ルーティングの最安の落としどころ。簡単なリクエストは非思考、難しいリクエストは思考で通す

この表自体が、本文で論じたい問いを示している。予算管理は論文の中の小さな手法ではなく、推論サービスが本番に載せたあとに必ず向き合う、日常のエンジニアリング課題なのだ。

1.3 テスト時スケーリングを使ってはいけないケース

同様に重要なのは境界を引くことだ。次の3つのケースでは、予算を増やすのはほぼ無駄になる。

タスクが判定できない。オープンエンドなライティング、クリエイティブ生成、標準解のない分析タスクは信頼できる検証シグナルが乏しく、サンプルを増やしてもその多様性を正答へと収束させない。正しさの側に投資すべきで、サンプリングを増やすべきではない。
カバレッジがすでに飽和している。1回の通過率がすでに十分に高い、あるいは複数回サンプリングした答えが完全に一致するなら、予算を増やしたときの限界収益はほぼゼロになる。このときに意味のある唯一の行動は、モデルを変えるかプロンプトを変えるかだ。
レイテンシの予算が極端に厳しい。対話型シナリオでは数十ミリ秒の遅延でもう使い物にならないので、あらゆる拡張方案が最初から成立しない。アーキテクチャを変えるしかない。

この3つを設計ドキュメントのいちばん前に書いておけば、無駄な実験を大量に省ける。

2. 形式化：「推論予算」とは結局何なのか

配分の話をするには、まず度量の話から始める必要がある。この節では推論予算を計算可能な形に分解していく。以降のすべての方法は、この一組の変数をどう取るかという異なる戦略として見ることができる。

2.1 三つの予算軸

もっとも粗い説明は「モデルにもう少しトークンを与えればもっと正しくなる」で、これでは足りない。テスト時計算を実際に制御しているのは三つの独立したノブだ。

幅：独立にサンプリングする推論チェーンの本数、あるいは探索木で同時に保持する分岐の本数。
深さ：各チェーンが生成してよい最大トークン数。
判別コスト：検証器を呼び出す回数と、1回の採点で読み込む必要がある入力長。採点は通常チェーン全体をもう一度読み直す必要があり、この分のオーバーヘッドは実践では深刻に過小評価されている。

したがって総予算は、おおむね「幅×深さ×生成単価＋判別呼び出し回数×採点単価」と書ける。実際にコストを決めるのは総トークン量で、実際にユーザー体験を決めるのは壁時計時間だ。この2つは別のものだ。テスト時スケーリングでよく混同される。テスト時スケーリングで最も混同されやすいポイントである。

2.2 コストは総量、レイテンシは最長チェーン

16チェーンを並列サンプリングして各2,000トークンなら、生成トークンは合計32,000。しかしハードウェアが一度に8チェーンしか並列処理できないなら、ユーザーが待つ時間は2ラウンドのうち遅いほうのチェーンで決まる。同じ予算を8チェーン×4,000トークンに変えても総量は同じなのに、壁時計時間は逆に長くなることもある。深さはラウンド内では並列化できないからだ。逆に、スケジューラうまく動けば幅は完全にバッチスループットに吸収できるが、深さは直列でしかない。つまり幅と深さはレイテンシへの代金がまったく異なり、深さのほうが通常は高くつく。

"""把预算拆成宽度 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

結論はきわめて直截的だ。同じ計算予算を幅に配るか深さに配るかで、壁時計レイテンシは4〜5倍も変わる。これは理屈の好き嫌いではなく、「計算を追加する」ことをユーザーが許容できる機能にできるかどうかを直接決める問題である。

2.3 カバレッジと意思決定：二つの指標の乖離

テスト時スケーリングの収益は一対の指標に分解して見ることができる。カバレッジは正しくできる可能性を測る。何度かサンプリングして、少なくとも1回は正解が出る確率。意思決定の質は実際に正しくできるかどうかを測る。システムが最終出力する答えの正答率である。

カバレッジが高く、意思決定が低い：モデルは正解をサンプルできるのに、投票や採点では選べない。ボトルネックは判別シグナルにある。より強い検証器が必要か、あるいは問題そのものの判定可能性が悪すぎる。
カバレッジが低い：予算をどう配っても無駄になる。ボトルネックはベース方策の探索空間にある。訓練側に戻るか、深さ優先探索に切り替えて、モデルに「別の道を歩く」機会を与えるしかない。
どちらも高い：予算はすでに飽和している。これ以上コストを増やすのは無駄で、モデルを変えるかプロンプトを変えるかを考えるべきだ。

この乖離をデバッグの入口に据えると、最終精度を睨みながら推測するよりよほど効く。さらに、これを毎回の実験の固定出力にしておくべきである。乖離の向きが、次に何を投資すべきか（サンプラーか検証器か）を直接決めてしまう。

2.4 予算には3種類の通貨がある

最後に、よく忘れられている階層を1つ挙げておく。予算はトークンだけではない。現実のシステムには、少なくとも互いに矛盾する3種類の通貨が存在する——

お金：主に総生成トークンと判別呼び出しで決まり、同時実行数には依存しない。
レイテンシ：最長チェーンとバッチラウンド数で決まり、ユーザーが感じ取る硬い制約である。
クォータ：レート制限のあるサービスでは、本当に希少なのは1分あたりのリクエスト数と同時実行数であり、このときバッチ効率は単一リクエスト効率より重要になる。

この3つの最適な配分はしばしば違う。夜間バッチ処理ではレイテンシを完全に無視して、予算をすべて幅に注いで精度を買ってもよい。オンライン対話では逆で、予算を浅すぎる深さと小さい幅の高速経路に絞り込まなければならない。すべての場面で最適な予算戦略は存在せず、特定の制約のもとで最適な戦略があるだけである。

3. 方法族その1：並列サンプリングと自己整合性

もっとも古くて、もっとも堅実なテスト時スケーリングの方法である。Wang らが2022年に提案した自己整合性（self-consistency）の仕組みはごく素朴だ。同じ問題に対して複数の推論パスをサンプリングし、それぞれの最終答えを取り出して、正規化したうえで多数決を取る。その価値は外部の検証器がまったく要らないことにある。モデルがサンプリングでき、答えを比較できれば使える。だから本番環境での普及率がもっとも高い。余計な部品がなければ、余計な故障点もない。

3.1 なぜ多数決が効くのか

自己整合性が効く仕組みはこうだ。chain of thought の失敗はランダムではなく、いくつかの浅い近道に集中する傾向がある。条件を1つ落とす、符号を逆にする、掛け算を1回間違えて最後まで崩れる、などの誤りは、相当数のサンプルを引き寄せる。一方、正しい道は多くあっても、より深い共通構造を共有している。サンプル数が十分になると、正しいアトラクタが答え分布の上で最頻値として蓄積し、浅い誤りの票を押し流す。

この仕組みには重要な帰結が1つある。自己整合性の上限は「正解がサンプルされたか」ではなく、「正解が答え分布の最頻値になれるか」で決まる。複数の候補がどれも同じ正解なのに正規化していなかった場合、正解の票は薄められ、投票は機能しなくなる。これは繰り返し観測されるオンライン障害を説明する。モデルは正解を出しているのに、答えの書式がそろっていなくて少数票に沈められた、というものだ。

3.2 サンプリングのハイパーパラメータ探索：温度と切り詰めの関係

並列サンプリングには温度が要るが、温度は同時に一本のチェーンの質を損なう。これがテスト時スケーリングでもっとも核心にある緊張関係だ。低い温度は単一サンプルの正答率を上げ、高い温度は候補どうしの差を上げる。実務では0.6〜0.8が数学やコードのタスクでの主流なレンジで、それより上げると構文エラーや反復による劣化が大量に増える。

もう一つ、よく見落とされている細部がある。強化学習で学習された推論モデルは、温度に対する感度がベースモデルより明らかに高い。この種のモデルは温度ゼロでも長く、結構の整った推論チェーンが出るので、まず温度ゼロの「アンカーサンプル」を1本取れば質がもっとも高い。探索的なサンプリングで温度を上げるのはその後でいい。すべてのサンプルを高温にしてしまうと、無料で手に入る正答率を捨てるようなものだ。

切り詰めの長さも、温度と合わせて調整する必要があります。高い温度は生成長を伸ばします。切り詰め上限を低めに設定すると、途中で推理が切れたサンプルが大量に発生します。そのサンプルは答えを取り出せないので、予算を捨てるだけでなく、投票の分布を乱してしまいます。

稳妥な方法は、生成長の分布の95パーセンタイルより上に切り詰め上限を置くことです。切り詰め率も明示的に集計します。切り詰め率が数パーセントを超えていれば、そのパラメータの組み合わせに問題があります。

3.3 答えの正規化：最も過小評価されがちなエンジニアリングの詳細

投票が成功するかどうかは、正規化がどれだけ丁寧かにかかわっている。この点だけは最低限心存しておく。

空白と全角文字を統一する。
数式の包装形式を統一する。分数表記と小数表記のどちらでも、ドル記号の有無は同一視する。
数値の答えは文字列比較ではなく、浮動小数点の許容範囲で比較する。
単位とパーセント記号を取り除く。
構造化された答えは、フィールドごとに並べ替えてから比較する。
答えの中の「時」「分」のような同義の表現を畳み込む。

この作業量は大したものではありません。しかし、投票を使えるかどうかを直接左右します。

もう一つ、隠れた落とし穴があります。答えの中に推理の残骸が残っているということです。多くのモデルは、答えの後ろに説明や自己点検を付け足します。取り出し関数を「最後の数値」や「最後の1行」で切るように書くと、一部のサンプルではまったく別のものを取り出してしまいます。取り出し規則は、複数言語・複数形式・複数長に分けて検証しなければなりません。数個の手選びサンプルで確認しただけでは足りません。

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 自己整合性が失敗する3つの場面

自己整合性は万能ではありません。次の3つの状況では明確に失敗します。エンジニアリングの段階で前もって見分けられないと、これらの場面で予算を永続的に無駄にします。

答えの空間が大きい、またはオープン形式：ライティング、要約、複数ターンの対話、動作する関数の生成。投票できる離散的な答えがほとんど存在せず、無作為なサンプリングに退化します。この場面では「投票」という操作自体に意味がありません。
複数解が同値だが正規化していない：正答率が見かけ上、薄められてしまいます。これはすでに詳しく述べました。典型的な症状は、投票を追加したのに正答率が下がり始めることです。
より強い検証器がある場合：カバレッジがすでに高い（たとえば9割）一方で、検証器の判別力が9割5なら、検証器による再順位付けは多数決を明らかに上回ります。このときにサンプリングを増やすのは無駄で、予算を採点側に移すべきです。

判断の合言葉はこうです。答えを離散的に正規化できるなら、投票する価値がある。できなければ投票せず、検証するか評価を改善する。この合言葉の実用的な価値は、設計段階で注定して無効な案をまとめて除外できることにあります。

4. 方法族その2：検証器による再順位付け

自己整合性は頻度で正誤を代用していました。検証器による再順位付けは、別の信号に置き換えます。採点器を学習または構築し、各候補チェーンにスコアをつけ、最高得点のものを選びます。表面上は選択規則の交換にすぎませんが、この変更はシステムのボトルネックをまるごともう一方へ移します。以後、「正解がちょうど最頻値である」ことを前提とする必要はなくなり、正解を見分けられる判別器が必要になります。

4.1 2つの選択規則の本質的な違い

自己整合性と検証器による再順位付けの違いを1つの表にまとめておいたほうが、どちらが優れているかという定性的な議論よりずっと有用です。

軸	自己整合性の多数決	検証器による再順位付け
外部の採点器が必要か	不要	必要。しかもシステムの中核部品になる
意思決定の上限を決めるもの	答え分布の形。正解が最頻値になるかどうか	採点器の判別能力の上限
典型的な失敗	正しい同値の答えが薄められる。全員が同じ誤りをする	採点器がそれっぽい誤りに高得点をつける
コスト構造	コストはほぼ生成に集中する	生成と採点の両方。採点はチェーン全体を読み直すため、長いチェーンでは無視できない
探索との互換性	弱い。本来は幅戦略	強い。末端の再順位付けにも探索中の枝刈りの启发式にも使える
説明可能性	高い。票の分布がそのまま答えの確度を示す	低い。スコアは較正されていないことが多く、絶対値を説明できない

ここには見落とされがちな現実的な問題があります。採点は無料ではない。採点器はすべての候補チェーンを読み通さなければならず、長いチェーンなら数千トークンに達します。これは純粋な順方向計算です。長いチェーンの推論タスクでは、判別のオーバーヘッドが総予算の無視できない割合を占めることがあります。そのため「検証器を入れるべきか」を決めるときは、「幅×（生成長＋採点の入力長）」で計算すべきで、生成だけを見るべきではありません。

4.2 採点器の4つの実装形態

結果報酬モデル：最終的な答えだけを対象に採点器を学習します。データは安価で、注視すべきは終点だけなので、最終的な正確さに敏感に反応します。弱点はあります。長いチェーンに対する信用配分が極端に悪いのです。50ステップの推論チェーンが3ステップ目で間違っていたなら、学習されるのは「このチェーンは間違っている」ことだけで、「どのステップが間違っていた」ことではありません。
過程報酬モデル：推論の各ステップに局所的な正しさを加点します。探索の枝刈りにそのまま使えるため、信用配分が細かくできます。代償は注釈が高価なことです。ステップごとのラベル付けが必要で、しかも「このステップ自体は正しいが、その文脈で最適か」についての判断を統一するのは困難です。
生成式検証器：独立した採点モデルを持たず、プロンプトの末尾でモデル自身に「これは正解だったか、間違いだったか」という判定トークンを出させます。検証を次のトークン予測という形に置き換えたものです。利点は検証器と生成器が表現を共有し、エンドツーエンドで微調整できることです。短所は生成器の偏見をそのまま受け継ぐことです。
手続き的な検証：数学なら記号計算、コードなら単体テスト、論理問題なら制約ソルバー、データ分析ならクエリの再実行を使います。これは本当に信頼できる唯一の検証器の分類です。だからこそ競技数学とコードがテスト時スケーリング研究の主戦場になっているのです。その2つのタスクが難しいからではなく、逆にその答えを機械で確認できるからです。

4.3 なぜ強い推論モデルは過程監視を敬遠する傾向があるのか

DeepSeek-R1 の技術報告には、繰り返し読む価値のある判断があります。訓練段階に過程報酬モデルを持ち込まず、検証可能な結果報酬で強化学習を続けることを選んだのです。その理由の1つは、過程注釈が最終方策のデータ分布と一致している必要があるのに対し、方策は強化学習の過程で絶えずドリフトし、過程注釈の分布も一緒にドリフトし、最終的に過程報酬モデルが逆に精度向上のボトルネックになったからです。

これは大規模な実践から得られた直感に反する結論です。過程監視が無用だという意味ではなく、その保守コストが長時間の学習の中で過小評価されていたことを意味しています。テスト時スケーリングの言葉に翻訳すると、過程報酬モデルの信号は方策の能力が上がるにつれて失効します。ラベルを貼り直し、適応させ、対抗的にテストし続けるための資源を常に入れ続ける必要があります。この費用は、得られる便益を上回ることがよくあります。実用的な結論はこうです。末端の再順位付けだけなら結果報酬で十分であり、本当に多段探索を行って中間の枝刈り信号が必要なときになって初めて、過程監視は代替できなくなるのです。

4.4 検証器で最も危険な失敗：相関した誤り

検証器と生成器が同じ訓練データ、同じアーキテクチャ、あるいは単に同じベースモデルから微調整されているなら、両者は相関した誤りを犯します。この種のモデルが系統的に間違える問題では、生成器が1回間違えれば検証器ももう1回間違えます。再順位付けは効果がないどころか、多数決より悪化しうるのです。多数決には少なくとも「集団の一致」という弱い信号が保険としてあるのに対し、検証器は正確そうに見える誤りの信号を提供してしまうからです。

実務的に使える緩和手段が3つあります。

検証器と生成器を異質にする。異なるベースモデル、異なる訓練データ、あるいは単純に手続き的な検証を使います。もっとも徹底的な方法ですが、もっとも費用もかかります。
スコアを較正する。未較正のスコアをそのまま比較することはできません。答えが既知のサンプルを一批検証器に通し、等渗回帰やロジスティック回帰を行って、スコアを確率に変換します。較正されると、スコア自体が使える停止信号になります。
検証器を判官ではなく特徴量として扱う。複数の信号が使える場合（多数票、検証スコア、長さ、答えの出現位置、形式の整い方）、小さなロジスティック回帰を1つ積むほうが、単一の検証器を信頼するより安定します。

4.5 階層的な再順位付け：予算を要所に使

最後は純粋なエンジニアリングの最適化点です。検証器による再順位付けのコストは候補数に対して線形に増えますが、ほとんどの候補について採点は不要です。2つの層に分けられます。まず安価なモデルで大量に生成し、安いヒューリスティックで粗く並べます。その後、重複を除いた上位の異なる答えについてのみ、強いモデルで再生成し、高精度に採点します。これにより判別コストを大きく圧縮でき、最終的な精度への影響はごく小さくなります。

"""验证器重排：分层生成 + 并行打分 + 校准阈值停止。

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 に注目してください。1位と2位の較正済み確率の差が0.02しかなければ、採点器はこの問題に対してほぼ判別力を持っていないことを意味します。このとき「スコアが高い」ことを根拠とする早期終了は信頼できません。高スコアは採点器全体が足元を基準に高めに置かれているだけかもしれないからです。これは実用的で、ほぼゼロコストの較正セルフチェック指標です。モデルバージョンごとに一度実行することを勧めます。

5. 方法族その3：過程報酬と結果報酬のどちらを取るか

この節は単独で取り出す価値があります。しばしば誤解されるからです。「過程監視は結果監視より先進だ」と既定で考え、何でも過程報酬モデルを使い始め、保守コストに潰されます。2つの信号の本質的な違いは1文に圧縮できます。結果報酬はこの道が通れるか教えてくれ、過程報酬は次のステップをどちらに曲がるべきか教えてくれる。前者は安価で頑健、後者は精密だが育てにくい。

5.1 項目ごとの比較

軸	結果報酬モデル	過程報酬モデル
教師信号の出所	最終的な答えが正誤か	各ステップが正しいか、有効に前進しているか
データコスト	低い：軌跡は自動生成でき、終点だけをラベル付けする	高い：ステップごとのラベル付けが必要で、人手密集かつ標準の統一が難しい
信用配分の粒度	チェーン全体に1つの信号。粒度はきわめて粗い	ステップ級。長いチェーンと枝刈りに適している
分布ドリフトへの耐性	強い：終点だけを問題視せず、推論のスタイルは気にしない	弱い：ラベルが特定の推論スタイルと強く結び付いていて、方策が変われば古びる
探索における役割	末端の再順位付け	各ステップの枝刈りにおける局所ヒューリスティック
典型的な失敗	「1歩間違えたが正解」と「遠回りしたが正解」を区別できず、表面的なパターンを学ぶ	報酬攻撃：採点器を満足させる形式を出力することを学び、本物の推論ではなくなる

この2つの選択は1つの判断ルールに単純化できます。末端の再順位付けだけなら結果報酬で十分であり、多段探索を行う場合だけ過程報酬が必要になる。このルールは绝大多数の実際のシステムで正しい病了

5.2 過程報酬が正解になる3つの場面

過程報酬が無駄だということは、成本対効果が高いのは特定の条件下だということです。

探索木の枝刈り。多段探索を行う場合、中間状態を評価する信号は過程報酬からしか得られません。ある分岐の拡張コストが数百トークンなら、すべての中間ノードで完全な検証を実行することは不可能です。
長いチェーンの推論における失敗の特定。100ステップの推論チェーンが間違えたとき、どのステップで曲がったのかを知る必要があります。それを与えるのは過程報酬だけです。これはデータの洗浄や難例の採掘に特に有用です。
蒸留と教師データの絞り込み。過程スコアで高品質な中間ステップを絞る作業は、最終的な答えだけで絞るより、きれいな教師データを得やすくなります。

一方、末端の再順位付けだけを行い、チェーンも長くなく、大量の自動ラベル付けができる状況なら、結果報酬が圧倒的に正しい選択です。

5.3 ステップスコアを1つの信号に集約する

過程報酬が与えるのはステップ級のスコアですが、探索はスカラー値を必要とします。集約方法は枝刈りの挙動に直接影響します。最小値が最も保守的で、どのステップかが崩れたら全体が崩れます。ハードな枝刈りに適します。積は長いチェーンを構造的に不利にするため、正規化後の並べ替えに適します。最初の誤りへのペナルティは「初期の誤り」と「後期の誤り」を明示的に区別します。前者のほうが通常は救い難いです。実務でもっとも安定した組み合わせは、最小値を枝刈りの閾値に、対数積を並べ替えに使うということです。役割がはっきり分かれます。

"""过程奖励信号的聚合与提前终止。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 過程報酬を唯一の真実としない

過程報酬を使う場合でも、受け皿となる経路を1つ残しておくことを勧めます。チェーンを終点まで走らせて、手続き的にあるいは結果で検証するのです。ステップのラベルは必ずカバレッジが欠け、必ずノイズを含むのに対し、終点の検証は客観的だからです。安定した組み合わせはこうです。過程報酬は過程の枝刈りを担当し、終点の検証器が最終選択を担当する。これは構成上、ちょうど「深い探索に幅の再順位付けを重ねた」ハイブリッド戦略に対応します。次節で扱う内容です。

6. 適応的な難易度ルーティング：予算を要所に使

これまでの各節では、与えられた予算をどう配るかを論じてきました。この節では、もう1つ手前の問いを扱います。予算そのものをいくら与えるべきか。すべての問題に同じ予算を与えるなら、平均的には必ず無駄があります。問題によって収益曲線の傾きがまるで違うからです。

6.1 難易度と最適なサンプリング数の経験的な関係

これはテスト時スケーリングでもっとも実用的な経験則です。簡単な問題は少数のサンプリングで飽和し、難しい問題だけが大量のサンプリングを必要とする。モデルがもともと解ける問題を4回サンプリングしても64回サンプリングしても精度はほぼ同じです。一方、8回に1回だけ当たる問題なら、64回サンプリングすると精度が1倍以上に跳ね上がることがあります。したがって「簡単な問題は4サンプル、難しい問題は64サンプル」のような不均等な配分は、「全部64サンプル」よりはるかに大きな成果が得られる。

この法則の直接の帰結はこうです。テスト時計算の最適な配分は、問題ごとに行うもので、全体で一様に配るものではない。これが、固定されたサンプリング数の方式に常に改善の余地がある理由です。その方式は、簡単な問題に系統的な過剰投資をし、難しい問題に不足している。

6.2 3つの実装経路

カスケードと早期終了：もっとも単純でもっとも頑健です。まず小さい予算で1ラウンド回し、答えが一致するか検証スコアが十分に高ければそのまま返します。閾値を発火したときだけ予算を引き上げます。実装の複雑さは低く、遮断の意味を自然に備えており、本番環境の第一選択です。
難易度の予測：難易度分類器を学習するか、少数のサンプルから得られるカバレッジの傾き、最初のトークンのエントロピー、答えの長さといった安価な特徴で難易度を予測し、それに基づいてサンプリング数をあらかじめ割り当てます。理論上は最適ですが、難易度予測そのものは精度が出しにくいのです。モデルが「難しい」と感じるものと「実際に難しい」ものとの間には系統的なバイアスがあり、しかもこのバイアスはモデルのバージョンとともにドリフトします。
オンラインスケジューリング：予算をキュー上の資源として扱い、問題が 들어きたらまずもっとも安いプローブを当て、サンプリング途中の不確実さ（票の分布がどれだけ尖っているか、候補がどれだけ分かれているか）によって追加投資するかを決めます。これは実際のサービスにもっとも合った形態です。将来を見通す必要がなく、現在のバッチの情報だけで判断できるからです。

6.3 オンラインスケジューリングの落とし穴：自信を持った誤り

早期終了には実際のリスクがあります。モデルがまったく確信した状態で間違っている場合（すべてのサンプリングが一貫して同じ誤った答えを返す）、早期終了はその誤りをそのまま固定してしまう。そして一貫した誤りは、推論モデルが最も難しい状況のひとつです。訓練データの誤りパターンが一般に起きるとき、モデルはそのモードで極端に一貫して間違えます。

緩和手段が2つあり、どちらも実装する価値があります。

早期終了に最小予算の床を設ける：投票がどれだけ確定していても、少なくとも数本のサンプルを生成してから返すことを許す。簡単な問題での効率を少し犠牲にすることで、自分では回復できない種類の誤りを取り除けます。
早期終了には二重の閾値を使う：投票の一致度と検証スコアの両方が閾値に同時に達しなければなりません。単一の信号（票数でもスコアでも）は一貫性の誤りに騙され得ますが、2つの独立した信号の複合閾値のほうがはるかに信頼できます。

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

このスケジューラの最も重要な性質は、コスト分布の右偏を平坦化していることです。簡単な問題にはほとんど費用がかからず、予算は自然と難しい問題へ流れます。同じ平均予算であれば、「すべての問題にもっと大きなサンプリング数を与える」よりもはるかに効果が高く、後者は簡単な問題に使った1円ごとに限界収益がゼロだからです。

6.5 キューレベルでの予算の公平性

単一リクエストのスケジューリングでは解けない別の問題があります。一束のリクエストが1つの予算を共有しているとき、どのように配分するかです。1つの難しいリクエストがバッチ全体の予算を消費しながら、並行して簡単なリクエストが並んでいることがあります。エンジニアリングでは3つの戦略が一般的です。推定難易度による事前配分（もっとも楽だが、難易度予測に依存する）、先着順に加えて硬い上限（もっとも単純で予測可能性が最も高いが、簡単なリクエストが多いと予算を使い切れない）、残りのキューの長さによる動的調整（予算の利用率がもっとも高いが、レイテンシの予測可能性が最も悪い）。

実務での経験はこうです。レイテンシに敏感なサービスでは先着順と硬い上限を、オフラインのバッチ処理では動的調整を使う。前者はユーザーが知覚するのがレイテンシの安定性であり、後者はユーザーがそもそもオフラインなので、スループットだけが唯一の目標だからです。

七、幅優先と深さ優先：探索トポロジの選び方

ここまでの戦略はすべて幅優先だった。横方向に向かって独立した経路を並べていく方式である。だとすると縦に掘り下げるほうがいいのか。これは本物のトレードオフであり、無料の答えはない。

7.1 二つのトポロジの比較

観点	幅優先（独立サンプリング＋投票または再ランキング）	深さ優先（ビーム探索、ツリー探索、モンテカルロ木探索）
経路の関係	各チェーンは完全に独立	ノードは部分解、エッジは継続生成で、接頭辞を共有できる
トークン効率	低い：複数のチェーンが同じ接頭辞を重複生成する	高い：接頭辞の共有が重複計算を薄める
レイテンシ特性	良好：実時間はバッチの段数で決まる	悪い：ツリーの深さは直列で、テールレイテンシが高い
実装の複雑さ	低い：ステートレスで、水平スケールしやすい	高い：ノードキャッシュ・バックトラック・枝刈り・並行状態が要る
スコアリングへの依存	弱い：末端での再ランキングだけでよい	強い：毎ステップの枝刈りがスコアの質に依存する
得意分野	探索：局所最適から抜け、多様な解法をカバーする	利用：一つの解法に沿って深く掘り下げる
主な失敗	正しい解の方向に届かない	初期の誤った枝刈り。しかも深いチェーン上の誤りが見えない

OpenAI のテスト時計算の最適スケーリングに関する論文は、このトレードオフをかなり整理して示している。簡単な問題では独立サンプリングで十分であり、難しい問題でのみ探索が正味の利益をもたらす。ただしその利益はスコアラの強度によって左右され、スコアラが弱ければ深掘りはむしろ幅広探索に負ける。論文にはさらに直感に反する観察もある。計算量が十分にある場合でも、ビーム探索は十分に強いスコアラに支えられていなければ、単純な独立サンプリング＋再ランキングに조차及ばない。

これはよくある観測を説明する。同じ探索アルゴリズムが弱いモデルでは平凡なのに、より強い推論モデルに変えると突然うまくいくのはなぜか。理由は、モデルの1ステップの質が上がると、深掘りが各ステップで誤る確率が下がり、誤った枝刈りの代償もそれに比例して下がるためだ。探索アルゴリズムと基盤モデルは不可分な1組のシステムであり、モデルから切り離して評価することはできない。

7.2 ハイブリッド戦略：まず散らし、後で集める

本番システムで二者択一になることは珍しく、ほとんどはハイブリッドだ。幅で探索し、幅で収束させ、深さで精密化する。まず少数の独立サンプリングで問題空間を把握する。どの型の解法が有効か、答えは一意か、接頭辞は安定か。次に最も有望な解法に対して深い検証や局所探索を行う。この「まず散らし、後で集める」形は純粋な深掘りより頑健だ。深掘りの最初の一歩がもっとも高く、一旦方向を間違えれば以降のすべてのトークンが無駄になるためである。

7.3 深さ優先の最小実装

ビーム探索は深さ優先の中で最も実用的な形だ。ポイントは接頭辞の共有である。ツリー探索のトークン効率上の優位はほとんどここから来る。よい実装は部分解を接頭辞木としてまとめ、共有接頭辞のKVキャッシュを再利用する。ノードごとに独立に再実行してはならない。

"""束搜索（束宽度 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回サンプリングで self-consistency を取っている」という一文自体に巨大な最適化余地が潜んでいることを意味する。

8.2 法則二：verifier の質が天井を決める

この法則は次のように言い換えられる。テスト時スケーリングの上限は verifier で決まる。しかも verifier がもたらす利得はサンプラより急峻だ。直感的に言うとこうなる。サンプリングはモデル能力の範囲内の答えしか採れないが、verifier はその中の正しい答えを識別できるかどうかを決める。カバー率が十分に高くなれば、唯一のボトルネックは判別だけになる。この時点でもサンプリングを増やしても全く意味がなく、予算を verifier に集中投入するのが正解だ。

これはプログラム検証がテスト時スケーリングの研究の中で特別な地位にある理由も説明している。verifier を確率モデルから決定論的判定へ変える、つまりボトルネックそのものを直接取り除いてしまうからだ。数学コンペティションが主戦場になったのも、正解が記号計算で確認できるからである。

8.3 法則三：利得の逓減は構造的なもので、確率的なゆらぎではない

精度がサンプリング数とともに向上していく様子はおおよそ対数線形的で、サンプリングを倍に増やすたびに得られる増分は下がり続ける。しかし「飽和する」こと以上に注目すべきなのは飽和の形だ。self-consistency の飽和は多くの人の期待より早く訪れる。正解が最多数になっていない場合、あと100回サンプリングしても取り戻せないからだ。一方、verifier による再ランキングの飽和はより遅い。最多数には依存せず、スコアラーが識別できるかどうかだけだからです。この2つの飽和点の位置が違うという事実こそ、実運用で手法を選ぶ際の主な判断材料になる。

8.4 法則四：訓練時スケーリングとテスト時スケーリングには乗算関係がある

見落とされがちな点だが極めて重要だ。強化学習学習で訓練された推論モデルは、テスト時スケーリングの利得が基盤モデルより显著に高い。理由はテスト時スケーリングの空間が基盤方策のカバー能力で決まるためだ。モデルがそもそも正解を採れなければ、サンプリングをいくら増やしても意味がない。推論モデルは訓練によって単一サンプリングの質も多様性も大きく上がり、探索できる空間が自然と広がる。

言い換えれば、訓練時スケーリングとテスト時スケーリングは競合関係ではなく乗算関係である。訓練が探索空間の大きさを決め、テスト時スケーリングがその空間のなかでどれだけ良く探索できるかを決める。这也意味している。テスト時スケーリングだけに力を入れて訓練の質を無視すれば、天井はすぐに来る。逆に、mediocre な基盤モデルは予算をどれだけ増やしても、強い推論モデルの単純な呼び出しに追いつくのは難しい。

8.5 より完全な利得曲線のモデル

ここまでの性質を束ねると、素朴な利得モデルはこうなる。精度の向上量は、カバー率の増分に verifier が候補から正解を引き抜く確率を掛け合わせたものだ。この分解の意義は、どこに予算を足すべきかを直接教えてくれることにある。

"""测试时扩展的收益分解模型。

准确率(宽度 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 利得モデルにおける3つの飽和パターン

上の曲線を形状ごとに分類すると、実用的な診断ツールが得られる。

急速飽和型：カバー率は低いが self-consistency は速く伸び、すぐに横ばいになる。簡単なタスクや強い基盤モデルが典型だ。10回以内で飽和するという特徴があり、それ以上のサンプリングは純粋な無駄だ。
緩慢上昇型：カバー率は伸び続けるが、決定の質が上がらない。「採れたが選べない」のが典型だ。self-consistency と verifier 再ランキングの双方がカバー率の上限から遠いことが特徴で、verifier に投資すべきだ。
天井制約型：2本の曲線とも verifier の上限に張り付く。verifier の質が不足している場合の典型だ。再ランキングの結果が実質的に「カバー率×一定の天井係数」に一致するのが特徴である。

実験のたびにどれに当たるか判断すれば、無駄に予算を積み上げる時間を大幅に省ける。

九、エンジニアリング実装：予算制御、タイムアウト、サーキットブレーカー

理論から実地までの間にあるのはすべてエンジニアリングの問題だ。オフライン評価でしか動かない Test-Time Compute Scaling の方式は、実運用に出すとだいたいタイムアウト、コストの暴走、連鎖障害で死ぬ。この節で扱うのはすべて「生き残る」ために必要なものだ。

9.1 予算制御の三つの階層

グローバル予算：一連のリクエストに配分する合計計算量。スケジューラがリクエスト間で公平に配分しない限り、二つの難しいリクエストがバッチ全体の予算を飲み干む。
単一リクエスト予算：一つの問題が悪用できるトークンとミリ秒の上限。これは唯一のハード制約で、呼び出し側かサービスが保証しなければならない。
段階予算：単一リクエスト予算の内側で、各段階（プローブ、格上げ、最終検証）に子予算を予約し、ある段階が無制限に時間を飲み込むことを防ぐ。

もっともよくある誤りは、グローバル予算だけを設定して段階予算を設定しないことだ。典型的な結果はこうなる。サンプリングの段度を上げると1チェーンが長くなり、1チェーンのデコード時間が残りの予算をすべて消費し、最終的に切り詰められた不完全な推論チェーンしか返せない。これは有用でないどころか、拡張しない場合より悪い。正しい答えを出せたはずの問題を、切り詰め失敗に変えてしまうからだ。切り詰めによる失敗はオフライン評価ではほとんど起こらない。低いレイテンシの開発機上で動いているからだ。

9.2 タイムアウト時の縮退：三段フォールバック

タイムアウトの処理は単純に「タイムアウトしたらエラー」ではない。明示的な縮退の連鎖を持ち、どの段階でも使えるものを返さなければならない。

フル戦略：プローブ、格上げ、検証の完全な流れ。
収束したら返す：格上げをやめ、既存のサンプリングで多数決して返す。もっともよく使われる段階で、大部分のリクエストはここで解決する。
切り詰めたら返す：単一チェーンのデコードがタイムアウトしたら強制停止し、すでに生成された部分を出力する。この部分の中にすでに答えが含まれていれば（多くの推論モデルは答えを終了マーカーの後ろに置く）、そのまま使える。

重要な設計原則がある。縮退パスはテストの中で実際に発動させなければならない。多くのチームの欠陥は「縮退ロジックを書き間違えた」ことではなく、「縮退ロジックが一度も発動していない。テスト環境のレイテンシが低すぎるから」ことだ。遅延を注入した環境で縮退テストを意図的に一度実行し、タイムアウト閾値を臨時にきわめて小さくして確かめよう。

9.3 サーキットブレーカー：自動的に切るべき五つの故障

Test-Time Compute Scaling は長時間・大容量・再帰的な呼び出しをもたらすので、もともとブレーカーを導入するのに向いた場面だ。

コストブレーカー：単一リクエストのトークン数、単一ユーザー累積トークン、バッチ全体の総予算のいずれかが上限を超えたら格上げを止める。
時間ブレーカー：実時間の上限。すべてのループの層で必ず確認しなければならず、入口だけでチェックしてはいけない。深さ優先の探索ループが毎ステップで確認しないと、タイムアウトが無限のハングになってしまう。
品質ブレーカー：推論チェーンが退化のループに入ったときに切断する。これは Test-Time Compute Scaling が純粋にもたらした新しい故障モードだ。単発の生成は無限に繰り返さないが、「生成、スコア付け、再生成」のループは陥落入る。
依存関係ブレーカー：verifier サービスがタイムアウトや異常を返したら、純粋な多数決へ縮退する。ここでは必ず「縮退して続行」でなければならず「全体の失敗」にしてはいけない。投票経路は verifier に依存しないからだ。
同時実行ブレーカー：拡張は単一リクエストの並列度を数倍に拡大するため、トラフィックのピーク時に推論クラスタを飽和させやすい。これはゲートウェイ層での同時実行のシェーピングが必要だ。

"""带完整熔断与降级的推理执行框架。

降级链条：完整策略 -> 收敛即返（多数投票）-> 截断即返 -> 失败
每一级都必须返回可用结果，只有最后一层才允许返回空。
"""

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 接頭辞キャッシュ：最も過小評価されている最適化

並列サンプリングには巨大な無駄がある。複数のチェーンが同じ問題接頭辞を共有しているのに、リクエストごとに毎回プリフィルし直している。長いシステムプロンプトに長文ドキュメントの検索を組み合わせた場合、このプリフィルコストはサンプリング数を掛け算するとデコード自体を超えることがある。主要な推論フレームワークはたいてい自動接頭辞キャッシュを提供しており、同じ接頭辞でKVキャッシュを再利用できる。並列サンプリングの場面では、これはほぼ無料の性能向上だ。

有効化のコストはゼロで、長いプロンプトの場面では効果が非常に大きい。これは本稿で推奨する優先度が最も高いエンジニアリング最適化であり、すべてのアルゴリズム変更より前に来る。

9.5 バッチ処理と並行性のトレードオフ

並列サンプリングは連続バッチ処理に自然に適している。複数のチェーンのデコード段階はいずれもメモリ帯域律速で、バッチにまとめれば単一 GPU のスループットが大きく上がる。ただし逆向きの力が二つある。一つはバッチ内の長いチェーンがバッチ全体の完了時刻を引きずる（テールレイテンシ問題）こと、もう一つはバッチが大きくなるほど GPU メモリの圧力が高まり、同時実行数を下げられること。実務上のチューニング方向はリクエスト内では小さなバッチを保ち、リクエスト間で大きなバッチを作ることである。同じ問題のサンプリングを一つのバッチにまとめ、異なる問題どうしを後からマージする。

9.6 コストの可観測性

忘れられがちな点がある。コストは平均値の問題ではなく、分布の裾の問題だ。典型的な分布は、大多数のリクエストがごく少しかかり、難しい問題がわずかな数で予算の大半を消費するというものだ。運用では平均値ではなく、リクエスト単位コストの95パーセンタイルと99パーセンタイルを監視し、高コストのリクエストには個別にクォータを設定すべきだ。単一リクエストのコスト上限を持たない Test-Time Compute Scaling のシステムは、トラフィックが動揺すれば必ず事故る。

9.7 推奨する監視指標

Test-Time Compute Scaling は同じ問題に対して異なる品質の答えを出すので、監視の次元には予算が含まれている必要がある。最低限、次のものを用意する。予算水準ごとに分けた精度、平均消費トークン、平均レイテンシ、縮退の発動率、退化の検出率、verifier 利用不能の比率、そしてリクエスト単位コストの裾の分布。全体の精度だけを見ていると、「簡単な問題で節約したお金」と「難しい問題で足したお金」が互いに問題を隠してしまう。

十、評価：Test-Time Compute Scaling をどう公正に測るか

この節はもっとも地味に見えるが、信頼できる結論を出せるかどうかを決める。Test-Time Compute Scaling の評価には特有の落とし穴があり、どれか一つでも踏めば結論は反転する。

10.1 必ず報告すべき指標

指標	何を測るか	なぜ省略できないか
カバー率	サンプリング中に正答が少なくとも一度現れる確率	他のすべての指標の上界であり、これを測らないとボトルネックがサンプリングか判別かを判断できない
self-consistency の精度	投票判断の実際の品質	実際にデプロイされる方式そのものの品質
verifier 再ランキングの精度	判別段階が介在した後の品質	投票と比較しないと verifier に投資する価値があるかがわからない
平均トークンとレイテンシ	コスト	コストを伴わない精度は比較にならない
飽和曲線	限界利得がゼロになる時点	予算をどこで止めるかを決める
縮退の発動率	システムが実際に実行した方式	ブレーカーで切り落とされたリクエストの品質も勘案する必要がある

もっとも先に確認すべきなのは、カバー率と決定品質との間の裂け目だ。裂け目が大きければ決定側に余地があるので verifier を直す。裂け目が小さければ決定側はほぼ飽和しているので、サンプリングか訓練を直す。

10.2 評価の落とし穴リスト

単発通過率と単発投票を混同する。この2つは定義上ほとんど同じだが、多くの実装で数字が合わない。多くは答えの抽出ロジックが一致していないためだ。
コストが比較不能。単発の呼び出しと32回投票の精度を並べながらトークン差を報告しないのは、「30倍の計算力で1割良くなった」と宣言するのと同じだ。これは意味のある結論ではない。
答えの等価判定が厳しすぎる。数学的に等価な形式、コードの等価な実装、等価なクエリ文がすべて誤判定されると、投票や再ランキングの効果を体系的に過小評価する。この誤りの影響は多くの場合、手法の差による影響より大きく、しかも方向は固定されている。常に拡張手法を過小評価する。
温度がそろっていない。ベースラインは通常ゼロ温度、複数サンプリングはより高い温度を使う。この差そのものが「向上」の相当部分を説明するので、別に報告する必要がある。
易しい問題だけを測る。難しい問題ほど Test-Time Compute Scaling の利得は大きい。簡単な問題だけで評価すると「拡張は無意味」という誤った結論になる。
verifier を同分布データで訓練し、同じ同分布で評価する。得られるのは大きく過大評価された verifier の質であり、verifier への過剰投資につながる。
コンテキストと接頭辞キャッシュの影響を無視する。キャッシュの有無でレイテンシの差は大きく、両方の結果を混ぜて比較してはいけない。

10.3 予算と精度のフロンティア曲線

単一の予算点での精度にはほとんど情報量がない。情報量があるのはフロンティア曲線だ。横軸が総予算、縦軸が精度になる。2本の曲線（異なる方式）の上側包絡線が、真の性能境界だ。同じ総予算で比較しなければならないことに注意する。同じサンプリング数ではない。スコアリングにはコストが伴うので、同じサンプリング数でも実際のコストは大きく違うことがある。

"""构建预算-准确率前沿曲线，比较不同策略的性价比。"""

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 二者の本質的な違い

観点	訓練時スケーリング	テスト時スケーリング
本質	能力を重みに押し込む	能力を一時的に借り出す
コスト構造	一括で大きな額、長期に償却される	呼び出しごとに支払い、限界コストは線形
償却性	呼び出し量が多いほど有利	呼び出し量が多いほど不利、償却されない
ラベルなしタスクに対応できるか	できる、これは訓練の本質でもある	できない、判定可能な検証信号が必要
失敗モード	過学習、能力の非汎化、訓練の不安定性	予算の無駄、レイテンシの飛騰、verifier の抜け道
反復サイクル	週単位	時間単位、その日にリリースできる
天井	訓練データと信号の質	カバー率×verifier

11.2 経済学による判断基準

判断基準はごく機械的だ。タスクの呼び出し量を Q、訓練1回のコストを C、1回の推論あたりの追加コストを c、両者がもたらす精度の増分をそれぞれ Δ訓練 と Δ推論 とすると、限界が等価になる条件は単純だ。訓練を償却した後の単位コストがテスト時の単位コストを下回るときだけ、訓練する価値がある。

呼び出し量が極めて大きい汎用能力（カスタマーサポート问答、汎用アシスタント）：償却効果が显著で、テスト時スケーリングの限界コストは永久に追いつかない。この場面では訓練を優先すべきだ。
ロングテールの難問、領域特化、タスク量が小さい：償却が成立せず、再訓練の固定費用を薄められないため、テスト時スケーリングのコスト性能が非常に高い。垂直領域の小規模チームが推論モデルとテスト時計算力の組み合わせで、自前の基盤モデルを多くのタスクで勝てるのはこのためだ。
決定論的な verifier が使える高価値タスク（数学の証明、コード修正、データクリーニング）：ここでは利得が最も確実だ。検証信号が完全かつ無料だからである。投資対効果がもっとも高い場面である。

もう一つ見落とされがちな次元がある。反復の速さだ。訓練時スケーリングのフィードバック周期は週、テスト時スケーリングは時間である。変化の速いニーズ当面して、今日は予算調整とスケジューラの変更だけで得られる利得は、3ヶ月後の大規模モデル訓練より価値があることが多い。これはエンジニアリング上の現実であり、技術的な優劣ではない。

11.3 理想なのは組み合わせて使うこと

もっとも有効なのは、両者を役割分担させることである。訓練はカバー率と単発の質を高め、テスト時スケーリングは能力を極限まで引き出す。前出の「乗算関係」がこれを指している。2026年現在のコンセンサスはすでに明確で、強い推論モデルに良いテスト時スケーリング戦略を組み合わせたほうが、片側だけの最適化よりはるかに良い。

十二、限界と失敗モード

どの方式にも境界がある。この節では Test-Time Compute Scaling の既知の失敗モードを列挙する。これらを書き明確にしておくことは、普遍性を主張することより価値がある。本番環境の事故はほとんどこのどれかが原因だからである。

12.1 verifier の抜け道

verifier が学習されたモデルなら、利用可能なパターンを持つ。Test-Time Compute Scaling で verifier のスコアを最大化することは、代理目的を最適化していることに等しく、系は自発的に「verifier を騙せるが正しくはない」解を見つけるように学習する。数学コンペティションデータではこれが特に顕著だ。モデルは検証を回避する表現を見つけ出す。したがって verifier の設計は継続的な敵対テストが必要であり、模倣不可能なプログラム検証を優先すべきである。

12.2 精度は上がらないのにコストは倍になる

もっともよくある「拡張が無効」の場面だ。投票の結果が非常に一致しているが一致して誤っている、あるいは verifier が難問で機能せず全候補のスコアが近接している。前者はサンプリングを増やしても全く利得が出ないという形（増分がちょうどゼロ）で現れる。後者は再ランキングがランダム選択に退行する形で現れる。正しい対処は予算をサンプリング側から判別側へ移すことであり、サンプリングを増やすことではない。判断基準は前述の裂け目だ。カバー率が低ければサンプリングか訓練を直し、カバー率が高くても決定が悪ければ verifier を直す。

12.3 レイテンシが受け容纳できない

深さ優先探索、大量のサンプリング、verifier の呼び出しはいずれもレイテンシを延ばす。対話型の場面では数十ミリ秒の遅延でも利用できないので、この場合は拡張戦略を再設計しなければならない。深さを圧縮するか、サンプリング数を制限するか、あるいは非同期で初期版を返してから反復更新する形に切り替えられる。最後の方法はプロダクトでは非常に一般的だ。まず迅速に、完全ではない可能性のある答えを返し、あとから拡張結果で上書きする。

12.4 長さのインフレとコストの暴走

隠れた仕組みが一つある。長さに関連するヒューリスティックで過程スコアを採点するか、総トークン数で予算配分すると、系はより長く書くように学習する。これは初期には有効だ（長い推論は実際に正確だ）。だが最終的に純粋なコストの無駄になる。監視するときは精度だけでなく、「正答1件あたりの平均トークン数」を別に見る必要がある。

12.5 難易度ルーティングの誤判断

適応的予算は難易度推定に依存する。推定が誤ると代償は二通りある。難問を簡単と誤判定した場合（金をかけるべき時にかけず、精度を直接失う）と、問題を難問と誤判定した場合（お金を無駄にする）だ。前者の方がはるかに深刻である。よって最小予算のフロアは必須であり、「安いモデルで先に探る」のは投資に値する。

12.6 検証不可能なタスクにおける天井

最終的な境界はこうだ。タスクに検証可能な答えも高品質なスコアモデルも存在しない場合、Test-Time Compute Scaling はほぼ無効になる。この場合に使えるのは並列サンプリングがもたらす「集合知」効果だけで、しかも改善は限定的だ。これは実装の問題ではなく、情報理論的な制約である。どの答えがより良いかを示す信号が何もなければ、候補を増やしても選択がよりランダムになるだけだ。

12.7 評価への過学習

見落とされがちなリスクがある。検証集の上で予算戦略を繰り返し調整すること自体が、その検証集への過拟合だ。特に、カバー率の裂け目でサンプリングに投資するか verifier に投資するかを決めると、verifier が強い方向に無意識に傾く。対策は、一度も調整に参加していないホールドアウト集合で最終確認を行い、予算戦略の変更は控えめに保つことである。

十三、実行可能な実施チェックリスト

ここまでの内容を、そのまま手順に圧縮する。順序は重要だ。各ステップは次のステップの前提であり、飛ばせばその後の努力は無駄になる。

まずベースラインを取る：単発サンプリングの精度、トークン数、レイテンシを測定し、すべての比較の基準にする。ベースラインのない比較は無意味である。
タスクが検証可能か確認する：ユニットテスト、記号計算、決定論的なルールで答えを判定できるか。できるなら直接プログラム検証を導入する。できないならまずスコーラを作る。この結論が以降のすべての投資方向を直接決める。
答えの正規化と self-consistency を実装する：コストが最も低く、利得が最も確実な一歩だ。多くのチームはこれを飛ばして複雑な方式へ進み、精度が上がらず戻って初めて正規化の問題だったと気づく。
接頭辞キャッシュを有効にする：コストゼロの利得で、長いプロンプトの場面では効果が大きい。すべてのアルゴリズム変更より前に来る。
コストと時間のブレーカーを実装する：拡張を有効にする前にブレーカーを用意する。さもないと最初、事故の代償が大きい。
その後に verifier 再ランキングを検討する：まずカバー率と self-consistency の精度の裂け目を測る。裂け目が大きい場合にのみ verifier への投資が報われる。
最後に適応的スケジューリングを行う：固定予算で動かし指標が信頼できるようになった後で、予算配分を最適化する。そうでなければノイズの上で調整していることになる。
全経路の監視：予算水準ごとに精度、トークン数、レイテンシ、縮退率を観察し、特にコストの分布の裾に注目する。
縮退パスの回帰テスト：遅延と verifier の障害を意図的に注入し、縮退連鎖が本当に使える結果を返せることを確認する。
保留データでの最終確認：すべての調整が終わった後、調整に一度も参加していない集合で最終確認を行い、評価への過学習を防ぐ。

結語：Test-Time Compute Scaling は思考の様式を変える

Test-Time Compute Scaling の核心的価値は、モデル能力を「重みにどれだけあるか」から「スケジューリングをどれだけうまくできているか」に変えたことにある。予算制御、難易度ルーティング、verifier の質、探索トポロジがすべて継続的に最適化できるとき、推論は推論の問題から探索とスケジューリングの問題へと変わる。この転換がもたらす利得は構造的なものだ。再訓練なしで実現でき、しかもスケジューリング戦略が改善するほど利得は積み上がる。

同時に境界も明確だ。判定可能な検証信号が必要で、コストは呼び出し量とともに線形に増え、verifier のないタスクではすぐに天井に達する。これらの境界を理解することは、いかなる具体的なテクニックを覚えることよりも重要である。

真のエンジニアリング的判断力が現れるのは、いつ予算を増やすべきか、いつ止めるべきか、いつ訓練に戻るべきかを理解しているかどうかだ。この3点が Test-Time Compute Scaling の実践の知恵のすべてを構成する。ひたすらサンプリングの閾値だけを増やすシステムでは、最終的にコスト曲線に突き当たる。精度が飽和したときに自発的に手を止めるシステムだけが、長期的に健全に動き続けられる。