2.1 采样策略:随机、贪心、穷举

一张 48 节点、115 个选项的故事图,一条确定性路径只走 25 条边(446‰),远低于 850‰ 的目标。本节把随机、贪心、穷举三种采样策略放进同一路径预算对照,量出各自覆盖多少条对话边、到达几个结局,并解释为什么穷举是覆盖率的基准而不是一种采样。

2.1 采样策略:随机、贪心、穷举

独立游戏的叙事设计与分支对话 讲的是分支怎么写、选项怎么摆——那是散文式的经验,配清单与示意图,代码不可跑。本节把同一个问题变成一个可算的量:一张 48 节点、115 个选项的故事图,一条路径能走多少条边?「走到多少条边才算够」这句话,在散文里只能靠手感,在这里可以逐条数出来。

为什么必须采样

目录篇已经量过:用「每步选第一个可用选项」的确定性策略走一条路径,只覆盖 25 条对话边,占全量的 446‰——还不到目标 850‰ 的一半,只到达 1/5 个结局。一条路径的覆盖远远不够,所以必须走很多条。问题随之而来:多走几条、怎么走?下面三种走法各有各的代价,本节把它们放在同一路径预算下对照。先约定口径——本节所有覆盖率都按去重对话边计(把指向同一目标的并行选项合并成一条图边,共 56 条;规格声明的 115 个选项去重后就是这 56 条),这与目录篇的 446‰ 同口径。

策略每一步怎么选代价用途
随机 random在可用选项里等概率选覆盖靠重复撞,收敛慢无偏的路径分布
贪心 greedy优先走全局访问次数最少的可用边路径分布被人为抹平用最少路径数把边铺开
穷举 exhaustive按条件剪枝的 BFS,收齐所有可达边不是「采样」,是基准覆盖率的定义上界

S 层:三种策略在同一预算下对照

下面这段 S 层脚本 sim/nm_02_1_sampling.py 把确定性下界、随机、贪心、穷举放在一起。随机与贪心各扫 N = 10 / 25 / 50 / 100 条路径,穷举不消耗预算、直接收齐可达边;每一步只走 when 成立的选项,PRNG 为自实现的 Lehmer MINSTD(rng = rng * 48271 % 2147483647),同种子同序列。

#!/usr/bin/env python3
# file: sim/nm_02_1_sampling.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/nm_02_1_sampling.py
"""卷三 2.1 采样策略:随机、贪心、穷举(S 层)。

同一个问题——「一张故事图上走到多少条边才算够」——有三种走法:

  随机(random)  每步在可用选项里等概率选,不看历史;覆盖靠重复撞。
  贪心(greedy)  每步优先走**全局访问次数最少**的可用边;用最少的路径数把边
                  铺开,代价是路径分布被人为抹平(故只用于覆盖,不做分布结论)。
  穷举(exhaustive)按条件剪枝的 BFS,把**所有可达边**一次收齐;这是覆盖率
                  的**基准**(56/56 = 1000‰),不是「采样」。

覆盖率口径:**走过的去重对话边 / 全部去重对话边**。规格声明 115 个选项,但指向
同一目标的并行选项去重后是 56 条边;本节一律以去重边为分母(目录篇的 446‰ 即此
口径)。对照确定性下界(每步选第一个可用选项)与三种策略在若干路径预算下的覆盖。
全部整数、无浮点;PRNG 为 Lehmer MINSTD,同种子同序列。只读 config/story_spec.json。
产出 out/nm_02_1_sampling.csv。
"""
from __future__ import annotations

import csv
import json
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 = [10, 25, 50, 100]
STEP_CAP = 64


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)          # not


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 init_state(spec: dict) -> dict:
    aff = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
           for v in sorted(spec["state"]["vars"])}
    return {"aff": aff, "flags": {f: 0 for f in sorted(spec["state"]["flags"])},
            "items": {i: 0 for i in sorted(spec["state"]["items"])}}


def first_path(spec: dict, by: dict) -> tuple[set, int]:
    """确定性下界:每步选第一个可用选项,走一条路径。返回走过的边与到达的结局数。"""
    node, st, es = spec["meta"]["start"], init_state(spec), set()
    for _ in range(STEP_CAP):
        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
        c = opts[0]
        es.add((node, c["goto"]))
        st = apply_effects(c.get("effects") or {}, st)
        node = c["goto"]
    return es, 1 if by[node].get("ending") else 0


def sample(spec: dict, by: dict, strategy: str, n: int) -> dict:
    """采样 n 条路径。返回覆盖的去重边 / 到达的结局 / 不同的选项序列数。"""
    rng, visits = SEED, {}
    es, ends, trails = set(), set(), set()

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

    for _ in range(n):
        node, st, trail = spec["meta"]["start"], init_state(spec), []
        for _step in range(STEP_CAP):
            if by[node].get("ending"):
                ends.add(by[node]["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["goto"]))
            trail.append(c["id"])
            st = apply_effects(c.get("effects") or {}, st)
            node = c["goto"]
        trails.add(tuple(trail))
    return {"edges": es, "endings": ends, "distinct_trails": len(trails)}


def reachable_edges(spec: dict, by: dict) -> set:
    """穷举口径:投影 BFS(只保留条件可读变量),每步只走 when 成立的边,收齐可达边。"""
    RA, RF, RI = [], [], []
    for n in spec["dialogue"]:
        stack = [n.get("line_when")] + [c.get("when") for c in (n.get("choices") or [])]
        while stack:
            c = stack.pop()
            if not c:
                continue
            if c[0] in ("and", "or"):
                stack += [c[1], c[2]]
            elif c[0] == "not":
                stack.append(c[1])
            elif c[0] == "flag":
                RF.append(c[1])
            elif c[0] == "item":
                RI.append(c[1])
            elif c[0] == "aff":
                RA.append(c[1])
    RA, RF, RI = sorted(set(RA)), sorted(set(RF)), sorted(set(RI))
    AR = {v: i for i, v in enumerate(RA)}
    NV = len(RA)
    nidx = {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 >> RF.index(c[1])) & 1) == c[2]
        if k == "item":
            return ((it >> RI.index(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 RF:
                fl = (fl | (1 << RF.index(k))) if v else (fl & ~(1 << RF.index(k)))
        for k, v in (eff.get("add") or {}).items():
            if k in AR:
                sh = 101 ** (NV - 1 - AR[k])
                cur = (code // sh) % 101
                code += (max(0, min(100, cur + v)) - cur) * sh
        for k, v in (eff.get("grant") or {}).items():
            if k in RI:
                it = (it | (1 << RI.index(k))) if v else (it & ~(1 << RI.index(k)))
        for k in (eff.get("revoke") or {}):
            if k in RI:
                it &= ~(1 << RI.index(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()
    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["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


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

    det_edges, det_ends = first_path(spec, by)
    bf_edges = reachable_edges(spec, by)

    rows = [{"strategy": "deterministic", "n": 1, "edges": len(det_edges),
             "permille": len(det_edges) * 1000 // E, "endings": det_ends, "trails": 1}]
    for strat in ("random", "greedy"):
        for n in NS:
            r = sample(spec, by, strat, n)
            rows.append({"strategy": strat, "n": n, "edges": len(r["edges"]),
                         "permille": len(r["edges"]) * 1000 // E, "endings": len(r["endings"]),
                         "trails": r["distinct_trails"]})
    rows.append({"strategy": "exhaustive", "n": 0, "edges": len(bf_edges),
                 "permille": len(bf_edges) * 1000 // E, "endings": len(spec["endings"]),
                 "trails": 0})

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "nm_02_1_sampling.csv"), "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=["strategy", "n", "edges", "permille", "endings", "trails"])
        w.writeheader()
        w.writerows(rows)

    print(f"图:{len(nodes)} 节点  选项 {options}  去重对话边 {E}  目标覆盖率 {m['coverage_target_permille']}‰")
    print(f"   {'策略':<14}{'N':>5}{'边':>5}{'‰':>6}{'结局':>5}{'不同序列':>9}")
    for r in rows:
        print(f"   {r['strategy']:<14}{r['n']:>5}{r['edges']:>5}{r['permille']:>6}"
              f"{r['endings']:>5}{r['trails']:>9}")
    print(f"穷举基准 = {len(bf_edges)}/{E} = {len(bf_edges) * 1000 // E}‰(覆盖率定义上界,非采样)")
    print(f"确定性下界 = {len(det_edges)}/{E} = {len(det_edges) * 1000 // E}‰(一条路径,远低于目标 {m['coverage_target_permille']}‰)")

    assert len(bf_edges) == E, "穷举应覆盖全部去重对话边"
    assert 0 < len(det_edges) < len(bf_edges), "单条路径应严格小于全量"
    assert all(0 <= r["permille"] <= 1000 for r in rows), "覆盖率须落在 [0,1000]‰"
    print("wrote out/nm_02_1_sampling.csv")
    print(f"OK nm_02_1_sampling:三种策略已对照,穷举基准 {len(bf_edges)}/{E} = {len(bf_edges) * 1000 // E}‰")
    return 0


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

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

图:48 节点  选项 115  去重对话边 56  目标覆盖率 850‰
   策略                N    边     ‰   结局     不同序列
   deterministic     1   25   446    1        1
   random           10   54   964    4       10
   random           25   55   982    4       25
   random           50   55   982    4       50
   random          100   55   982    4      100
   greedy           10   55   982    4       10
   greedy           25   56  1000    5       25
   greedy           50   56  1000    5       50
   greedy          100   56  1000    5      100
   exhaustive        0   56  1000    5        0
穷举基准 = 56/56 = 1000‰(覆盖率定义上界,非采样)
确定性下界 = 25/56 = 446‰(一条路径,远低于目标 850‰)
wrote out/nm_02_1_sampling.csv
OK nm_02_1_sampling:三种策略已对照,穷举基准 56/56 = 1000‰

判读:三种策略的代价与覆盖能力

第一,确定性下界是 446‰,只有穷举基准的不到一半。 这条路径走了 25 条边、1 个结局,无论重复多少次都还是这 25 条——确定性策略的覆盖率不随预算增长。这正是目录篇说「必须采样」的量化依据。

第二,随机采样靠「撞」。 N=10 就到 964‰、N=25 起稳定在 982‰(55/56 条边),但到 N=100 仍是 55 条——有一条边随机撞不到。这条没撞到的边就是第 2.3 节要定位的覆盖缺口。

第三,贪心用更少的路径数把边铺开。 它每步优先走全局访问次数最少的可用边,N=10 就追平随机的 55 条,N=25 到 1000‰(56/56),并且比随机早一步到达第 5 个结局(真相)。代价是它的路径分布被人为抹平:随机 N=100 走出 100 条不同序列,而贪心的路径形态高度趋同——所以贪心只用来做覆盖,不能拿它的路径分布下任何结论。

第四,穷举不是一种采样,是覆盖率的定义上界。 它不消耗路径预算,直接给出 56/56 = 1000‰——「全部可达边」这个数本身就是分母的来源;把它和采样并列,是为了回答「采样离基准还有多远」。一句话:随机无偏但收敛慢,贪心铺得快但有偏,穷举是基准不是策略。

本节所有覆盖率都是代理指标:它只度量「走过多少条结构边」,不声称这条路径的叙事体验更好或更完整。它只做相对比较(随机 vs 贪心 vs 基准),不做绝对判断。

自检表

检查项合格标准
口径一致三种策略同分母(去重对话边 56),千分比整数地板除
下界与基准真跑确定性 25/56 = 446‰(同目录篇);穷举 56/56 = 1000‰
整数不变量覆盖率 ∈ [0,1000]‰;无浮点
有偏声明贪心的路径分布被抹平,已显式声明不用于分布结论

知道了三种策略各能覆盖多少边,下一步要把「覆盖率」本身写成一个可复现的度量——它到底以什么为分母、单条路径与采样与穷举怎么放在一张表里比。这是下一节的事。

阅读导航:上一节:1.3 路径的回放与复现 · 下一节:2.2 分支覆盖率的度量 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「narrative」更多文章

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