7.2 剪枝与采样上界

路径数有两层上界:朴素乘积上界 5 的 31 次方约 4.66×10 的 21 次方,真实结构路径数 450208471680,剪枝后 BFS 可达边 115/115 即 1000‰。本节推导优惠券收集器的采样上界(覆盖 115 条边约 613 条路径),并实测随机与贪心两种策略在 N=100/1000/10000/20000 下的覆盖率收敛曲线。

7.2 剪枝与采样上界

玩法机制与可执行规格 那套对机制的组合爆炸用仿真加守恒不变量控制,对象是数值空间。本章控制的是分支的组合爆炸,对象是对话图。上一节量了「每步多宽」,这一节把它乘起来:一张 48 节点、每步 2 到 3 个选项的图,一共能拼出多少条路? 答案分三层——朴素乘积上界、真实结构路径数、剪枝后可达边集——再给出用采样去覆盖它的上界。

两个上界:路径数与采样

路径数的上界用乘积形式。 如果每一步都取最宽的节点、且都走到最深,路径数不会超过 dmax ^ depth:最大出度 5、最长深度 31,于是 5³¹ ≈ 4.66×10²¹。这是上界不是真值——它假设每个节点都能走到最深,而实际上很多节点离结局只有几步。

采样的上界用优惠券收集器。 要把 E 条边全部采到,均匀随机下期望需要 E · H_E ≈ E · (ln E + γ) 条路径(H 是调和数,γ 是欧拉常数)。这是「采样多少条才够」的理论刻度,下面用实测收敛曲线去核对它。

S 层:剪枝枚举 + 两种策略的收敛曲线

脚本 sim/ne_07_2_sampling.py 做四件事:① DAG 上 DP 精确算真实结构路径数;② 用投影 BFS(每步只走 when 成立的边)求剪枝后的可达边集;③ 推导采样上界;④ 用随机与贪心两种策略跑 N = 100/1000/10000/20000 的覆盖率收敛曲线,输出 CSV。

#!/usr/bin/env python3
# file: sim/ne_07_2_sampling.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_07_2_sampling.py
"""卷二 7.2 剪枝与采样上界(S 层):路径数的上界与覆盖率的采样上界。

路径数:朴素乘积上界 dmax^depth 与真实结构路径数(DAG 上 DP 精确计数,忽略 when)对照。
覆盖率:采样上界(优惠券收集器 E·H_E)与实测收敛曲线对照——随机(每步等概率)与贪心
(优先走全局访问次数最少的可用边)两种策略,N = 100/1000/10000/20000。
剪枝规则:只枚举可达状态(投影到条件可读变量的 BFS,每步只走 when 成立的边),
BFS 口径覆盖率应为 1000‰(115/115 条对话边全部可达)。产出 out/ne_07_2_sampling.csv。
只读 config/story_spec.json。exit 0 表示 BFS 覆盖 115/115。
"""
from __future__ import annotations

import csv
import json
import math
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
SEED = 20261016
NS = [100, 1000, 10000, 20000]

def load_spec() -> dict:
    with open(os.path.join(ROOT, "config/story_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)

def eval_cond(c, st: dict) -> bool:
    if c is None:
        return True
    k = c[0]
    if k == "flag":
        return st["flags"].get(c[1], 0) == c[2]
    if k == "aff":
        v, op, n = st["aff"].get(c[1], 0), c[2], c[3]
        return {">=": v >= n, "<": v < n, "==": v == n, "!=": v != n}[op]
    if k == "item":
        return st["items"].get(c[1], 0) == c[2]
    if k == "and":
        return eval_cond(c[1], st) and eval_cond(c[2], st)
    if k == "or":
        return eval_cond(c[1], st) or eval_cond(c[2], st)
    return not eval_cond(c[1], st)

def apply_effects(eff: dict, st: dict) -> dict:
    aff, flags, items = dict(st["aff"]), dict(st["flags"]), dict(st["items"])
    for k, v in (eff.get("set") or {}).items():
        flags[k] = v
    for k, v in (eff.get("add") or {}).items():
        aff[k] = max(0, min(100, aff.get(k, 0) + v))
    for k, v in (eff.get("grant") or {}).items():
        items[k] = v
    for k in (eff.get("revoke") or {}):
        items[k] = 0
    return {"aff": aff, "flags": flags, "items": items}

def reachable_edges(spec: dict, by: dict) -> tuple[set, set]:
    """剪枝规则:投影 BFS(整数打包去重),每步只走 when 成立的边。返回可达边集与跳转集。"""
    rf: set = set()
    ri: set = set()
    ra: set = set()

    def walk(c) -> None:
        if not c:
            return
        k = c[0]
        if k in ("and", "or"):
            walk(c[1])
            walk(c[2])
        elif k == "not":
            walk(c[1])
        elif k == "flag":
            rf.add(c[1])
        elif k == "item":
            ri.add(c[1])
        elif k == "aff":
            ra.add(c[1])

    for n in spec["dialogue"]:
        walk(n.get("line_when"))
        for c in (n.get("choices") or []):
            walk(c.get("when"))
    RA, RF, RI = sorted(ra), sorted(rf), sorted(ri)
    AR = {v: i for i, v in enumerate(RA)}
    FR = {v: i for i, v in enumerate(RF)}
    IR = {v: i for i, v in enumerate(RI)}
    NV, nidx = len(RA), {n["id"]: i for i, n in enumerate(spec["dialogue"])}

    def cond(c, code, fl, it) -> bool:
        if c is None:
            return True
        k = c[0]
        if k == "flag":
            return ((fl >> FR[c[1]]) & 1) == c[2]
        if k == "item":
            return ((it >> IR[c[1]]) & 1) == c[2]
        if k == "aff":
            v = (code // (101 ** (NV - 1 - AR[c[1]]))) % 101
            return {">=": v >= c[3], "<": v < c[3], "==": v == c[3], "!=": v != c[3]}[c[2]]
        if k == "and":
            return cond(c[1], code, fl, it) and cond(c[2], code, fl, it)
        if k == "or":
            return cond(c[1], code, fl, it) or cond(c[2], code, fl, it)
        return not cond(c[1], code, fl, it)

    def step(eff, code, fl, it):
        for k, v in (eff.get("set") or {}).items():
            if k in FR:
                fl = (fl | (1 << FR[k])) if v else (fl & ~(1 << FR[k]))
        for k, v in (eff.get("add") or {}).items():
            if k in AR:
                sh = 101 ** (NV - 1 - AR[k])
                code += (max(0, min(100, (code // sh) % 101 + v)) - (code // sh) % 101) * sh
        for k, v in (eff.get("grant") or {}).items():
            if k in IR:
                it = (it | (1 << IR[k])) if v else (it & ~(1 << IR[k]))
        for k in (eff.get("revoke") or {}):
            if k in IR:
                it &= ~(1 << IR[k])
        return code, fl, it

    code0 = 0
    for v in RA:
        code0 = code0 * 101 + next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
    start = spec["meta"]["start"]
    seen = {(nidx[start], code0, 0, 0)}
    frontier = [(start, code0, 0, 0)]
    edges: set = set()
    jumps: set = set()
    while frontier:
        nxt = []
        for node, code, fl, it in frontier:
            if by[node].get("ending"):
                continue
            for c in (by[node].get("choices") or []):
                if not cond(c.get("when"), code, fl, it):
                    continue
                edges.add((node, c["id"]))
                jumps.add((node, c["goto"]))
                c2, f2, i2 = step(c.get("effects") or {}, code, fl, it)
                k = (nidx[c["goto"]], c2, f2, i2)
                if k not in seen:
                    seen.add(k)
                    nxt.append((c["goto"], c2, f2, i2))
        frontier = nxt
    return edges, jumps

def coupon(n: int) -> float:
    return n * sum(1.0 / i for i in range(1, n + 1))

def main() -> int:
    spec = load_spec()
    nodes = spec["dialogue"]
    by = {n["id"]: n for n in nodes}
    start = spec["meta"]["start"]
    E = len({(n["id"], c["id"]) for n in nodes for c in (n.get("choices") or [])})
    J = len({(n["id"], c["goto"]) for n in nodes for c in (n.get("choices") or [])})
    memo: dict = {}

    def dp(nid: str, field: str) -> int:
        if (nid, field) in memo:
            return memo[(nid, field)]
        ch = by[nid].get("choices") or []
        if not ch:
            v = 1 if field == "paths" else 0
        elif field == "paths":
            v = sum(dp(c["goto"], "paths") for c in ch)
        elif field == "short":
            v = 1 + min(dp(c["goto"], "short") for c in ch)
        else:
            v = 1 + max(dp(c["goto"], "long") for c in ch)
        memo[(nid, field)] = v
        return v

    dmax = max(len(n.get("choices") or []) for n in nodes)
    paths_exact, dmin, dlong = dp(start, "paths"), dp(start, "short"), dp(start, "long")
    naive = dmax ** dlong
    bfs_edges, bfs_jumps = reachable_edges(spec, by)

    aff0 = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
            for v in sorted(spec["state"]["vars"])}
    init = {"aff": aff0, "flags": {f: 0 for f in sorted(spec["state"]["flags"])},
            "items": {i: 0 for i in sorted(spec["state"]["items"])}}

    def run(strategy: str, n_paths: int) -> dict:
        rng, visits, es, js = SEED, {}, set(), set()

        def rnd(mod: int) -> int:
            nonlocal rng
            rng = (rng * 48271) % 2147483647
            return rng % mod

        for _ in range(n_paths):
            node, st = start, init
            for _step in range(64):
                if by[node].get("ending"):
                    break
                opts = [c for c in (by[node].get("choices") or []) if eval_cond(c.get("when"), st)]
                if not opts:
                    break
                if strategy == "greedy":
                    m = min(visits.get((node, x["id"]), 0) for x in opts)
                    cand = [x for x in opts if visits.get((node, x["id"]), 0) == m]
                    c = cand[rnd(len(cand))]
                else:
                    c = opts[rnd(len(opts))]
                visits[(node, c["id"])] = visits.get((node, c["id"]), 0) + 1
                es.add((node, c["id"]))
                js.add((node, c["goto"]))
                st = apply_effects(c.get("effects") or {}, st)
                node = c["goto"]
        return {"edges": len(es), "permille": len(es) * 1000 // E, "jumps": len(js),
                "jump_permille": len(js) * 1000 // J}

    rows = [{"strategy": s, "n": n, **run(s, n)} for s in ("random", "greedy") for n in NS]
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_07_2_sampling.csv"), "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=list(rows[0]))
        w.writeheader()
        w.writerows(rows)
    with open(os.path.join(OUT, "ne_07_2_sampling.json"), "w", encoding="utf-8") as fh:
        json.dump({"edges_total": E, "jumps_total": J, "naive_product_bound": naive,
                   "paths_exact": paths_exact, "bfs_edges": len(bfs_edges), "bfs_jumps": len(bfs_jumps),
                   "bfs_permille": len(bfs_edges) * 1000 // E, "coupon_edges": coupon(E),
                   "coupon_jumps": coupon(J), "convergence": rows}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")

    print(f"图:{len(nodes)} 节点  {E} 条对话边(选项)  去重跳转 {J} 条")
    print(f"深度:最短 {dmin}  最长 {dlong}  最大出度 dmax={dmax}")
    print(f"朴素乘积上界 dmax^depth = {dmax}^{dlong} = {naive}")
    print(f"真实结构路径数(DP 精确,忽略 when)= {paths_exact}"
          f"  乘积上界 / 精确 = {naive / paths_exact:.3e}")
    print(f"BFS 剪枝后可达边 {len(bfs_edges)}/{E} = {len(bfs_edges) * 1000 // E}‰"
          f"  可达跳转 {len(bfs_jumps)}/{J}")
    print(f"采样上界(优惠券收集器):覆盖 {E} 条边 ≈ {coupon(E):.0f} 条路径;"
          f"覆盖 {J} 条跳转 ≈ {coupon(J):.0f} 条")
    print("收敛曲线(覆盖率千分比):")
    print(f"   {'策略':<8}{'N':>7}{'边':>6}{'‰':>7}{'跳转':>7}{'‰':>7}")
    for r in rows:
        print(f"   {r['strategy']:<8}{r['n']:>7}{r['edges']:>6}{r['permille']:>7}"
              f"{r['jumps']:>7}{r['jump_permille']:>7}")
    print("wrote out/ne_07_2_sampling.csv / .json")
    if len(bfs_edges) != E:
        print(f"FAIL BFS 覆盖率 {len(bfs_edges)}/{E}")
        return 1
    print(f"OK ne_07_2_sampling:BFS 覆盖 {len(bfs_edges)}/{E} = 1000‰,"
          f"乘积上界与精确路径数差 {math.log10(naive / paths_exact):.1f} 个数量级")
    return 0

if __name__ == "__main__":
    sys.exit(main())

本机实跑 /opt/homebrew/bin/python3 sim/ne_07_2_sampling.py:

图:48 节点  115 条对话边(选项)  去重跳转 56 条
深度:最短 22  最长 31  最大出度 dmax=5
朴素乘积上界 dmax^depth = 5^31 = 4656612873077392578125
真实结构路径数(DP 精确,忽略 when)= 450208471680  乘积上界 / 精确 = 1.034e+10
BFS 剪枝后可达边 115/115 = 1000‰  可达跳转 56/56
采样上界(优惠券收集器):覆盖 115 条边 ≈ 613 条路径;覆盖 56 条跳转 ≈ 258 条
收敛曲线(覆盖率千分比):
   策略            N     边      ‰     跳转      ‰
   random      100   112    973     55    982
   random     1000   115   1000     56   1000
   random    10000   115   1000     56   1000
   random    20000   115   1000     56   1000
   greedy      100   115   1000     56   1000
   greedy     1000   115   1000     56   1000
   greedy     10000   115   1000     56   1000
   greedy     20000   115   1000     56   1000
wrote out/ne_07_2_sampling.csv / .json
OK ne_07_2_sampling:BFS 覆盖 115/115 = 1000‰,乘积上界与精确路径数差 10.0 个数量级

判读:上界是乘法,实测是采样

乘积上界比真实路径数大 10 个数量级。 5³¹ = 4656612873077392578125(约 4.66×10²¹),而 DP 精确算出的结构路径数是 450208471680(约 4.50×10¹¹),比值 1.034×10¹⁰。差距来自「不是每个节点都能走到最深」——d_dq_2 这类收束点离结局只有一步,乘积上界却假设它后面还有 30 步可走。上界是乘法给的,真值是图给的。

剪枝后的可达边是 115/115,覆盖率 1000‰。 这是 when 全部代入求值、只走条件成立边之后的真实边集。换句话说:没有一条选项是死边——每一条都能在某条真实路径上被走到。这是覆盖率的分母,也是下一节扰动扫描的基线。

采样上界:覆盖 115 条边约需 613 条路径,覆盖 56 条跳转约需 258 条。 用优惠券收集器公式 E·H_E:115·H₁₁₅ ≈ 613、56·H₅₆ ≈ 258。实测与之吻合——随机采样在 N=1000 时已经 115/115(1000‰),落在 613 这条刻度之后不远。

两种策略的差别在「多快」。 N=100 时,随机只采到 112/115 条边(973‰)、55/56 条跳转(982‰);贪心已经 115/115。原因是贪心每步优先走全局访问次数最少的边,等价于不停地往「还没走过的角落」推,收敛更快。到 N=1000 两者都封顶——115/115 是硬上界,加无可加。

剪枝规则只有两条,但都是硬的。 ①只枚举可达状态:从起点出发做 BFS,投影到条件可读变量(未被任何 when 读到的 flag / item 不影响转移,投影掉不改变结论);②按条件剪枝:每一步只走 when 成立的边,不成立的边连进都不进。两条规则合起来,把「每个节点 × 每个状态」的乘积空间收缩成一条条真实走出来的路径。这也解释了为什么覆盖率能到 1000‰ 而不是 56/115:剪枝不是砍边,是只承认走得通的边。

检查项合格标准
上界可复算乘积上界 dmax^depth 由声明量直接算出
真值真跑结构路径数来自 DP 精确计数,非估计
剪枝口径只枚举可达状态、每步按 when 剪枝,覆盖率 = 1000‰
采样上界优惠券收集器公式给出,且与实测收敛吻合
双策略随机与贪心同时报,说明收敛速度差异

上界与采样都量完了:路径数被乘积上界压在天花板下,覆盖率被剪枝与采样收敛到 1000‰。但这些都是基线上的数。如果把这张图的宽度人为放大,路径数与覆盖率会怎么变?下一节用参数扰动把「组合爆炸」真的引爆一次。

阅读导航:上一节:7.1 分支宽度的度量 · 下一节:7.3 组合爆炸的控制 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「narrative」更多文章

  1. 15.3 规模告警
  2. 15.2 求值与遍历的开销
  3. 15.1 规格规模的增长