5.2 可达状态的枚举

逐层 BFS 把 9895423 个可达状态一层层展开:给出每一层的新增与累计、峰值 frontier,以及 4 个好感度变量的实际值域。去重键是状态的规范串,层号就是最短步数。本节贴出 32 层完整输出,并给出预算 12000000 与峰值内存 3132.8 MB 的口径。

5.2 可达状态的枚举

Phaser 分支叙事 关心的是分支在引擎里怎么切;本节关心的是切出来的所有状态里哪些真的走得到。第 5.1 节算出了总量 9895423,却没解释这个数怎么来、也没说清每个好感度变量到底取了哪些值。这一节把 BFS 逐层展开:每一层新增多少、累计到多少,以及 4 个好感度变量的实际值域。

逐层 BFS 与规范串去重

去重键是状态的规范串 node|yan=..,hui=..,shen=..,mu=..|置位flag|持有物品——与引擎 canonical() 只差最后一段 path。枚举的是状态而不是「到达它的路径」:同一个状态可能由多条路径走到,但只算一次;逐层 BFS 保证它第一次出现时落在最短步数那一层。

状态在 frontier 里打包成整数(好感度 7 位一段、flag 与 item 各一位图),只在去重时展开成规范串。这一步不是花活:990 万条规范串同时驻留会把内存顶穿,打包后峰值内存落到 3 GB 量级。

S 层:逐层枚举

#!/usr/bin/env python3
# file: sim/ne_05_2_reachable_enumeration.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_05_2_reachable_enumeration.py
"""卷二 5.2 可达状态的枚举(S 层):BFS 逐层扩展,用规范串去重。

一个「状态」= 当前节点 × 4 个好感度 × 18 个 flag × 9 个物品。本节从起点出发做
**逐层 BFS**,每一层只扩展上一层新出现的状态;去重键是状态的**规范串**
`node|yan=..,hui=..,shen=..,mu=..|置位flag|持有物品`(不含 path,故与引擎
`canonical()` 只差最后一段)。给出三样东西:

  1) 每一层的新增状态数与累计状态数(落 out/ne_05_2_layers.csv);
  2) 4 个好感度变量**实际取到的值域**(各取到多少个不同值,远小于 101);
  3) 枚举预算与内存口径(峰值 frontier、maxrss)。

预算 12000000,与 sim/na_00_spec_check.py 同口径;超限即非零退出。
状态本身在 frontier 里打包成整数(好感度 7 位一段、flag / item 各一位图),
只在去重时展开成规范串——这是把峰值内存压到 3 GB 量级的关键。
"""
from __future__ import annotations

import csv
import json
import os
import resource
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
CSV_PATH = os.path.join(OUT, "ne_05_2_layers.csv")
JSON_PATH = os.path.join(OUT, "ne_05_2_reachable.json")

AFF_LO, AFF_HI = 0, 100
STATE_BUDGET = 12000000


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


def clamp(v: int, lo: int = AFF_LO, hi: int = AFF_HI) -> int:
    return lo if v < lo else hi if v > hi else v


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]
        if op == ">=":
            return v >= n
        if op == "<":
            return v < n
        if op == "==":
            return v == n
        return v != n
    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] = clamp(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 main() -> int:
    t0 = time.time()
    spec = load_spec()
    by_id = {n["id"]: n for n in spec["dialogue"]}
    VARS = sorted(spec["state"]["vars"])
    FLAGS = sorted(spec["state"]["flags"])
    ITEMS = sorted(spec["state"]["items"])
    NV = len(VARS)

    def canon(node: str, aff: dict, flags: dict, items: dict) -> str:
        a = ",".join(f"{k}={aff[k]}" for k in VARS)
        f = ",".join(k for k in FLAGS if flags[k])
        it = ",".join(k for k in ITEMS if items[k])
        return f"{node}|{a}|{f}|{it}"

    def pack(aff: dict, flags: dict, items: dict) -> tuple:
        code = 0
        for v in VARS:
            code = (code << 7) | aff[v]
        fb = 0
        for i, k in enumerate(FLAGS):
            if flags[k]:
                fb |= 1 << i
        ib = 0
        for i, k in enumerate(ITEMS):
            if items[k]:
                ib |= 1 << i
        return code, fb, ib

    def unpack(code: int, fb: int, ib: int) -> tuple:
        aff = {VARS[i]: (code >> (7 * (NV - 1 - i))) & 127 for i in range(NV)}
        flags = {FLAGS[i]: (fb >> i) & 1 for i in range(len(FLAGS))}
        items = {ITEMS[i]: (ib >> i) & 1 for i in range(len(ITEMS))}
        return aff, flags, items

    start = spec["meta"]["start"]
    aff0 = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v) for v in VARS}
    flags0 = {f: 0 for f in FLAGS}
    items0 = {i: 0 for i in ITEMS}
    c0, b0, g0 = pack(aff0, flags0, items0)

    frontier = [(start, c0, b0, g0)]
    seen = {canon(start, aff0, flags0, items0)}
    aff_vals = {v: set() for v in VARS}
    flag_bm: set = set()
    item_bm: set = set()
    layers: list[dict] = []
    states = 0
    peak_frontier = 1
    budget_hit = False
    level = 0
    while frontier:
        cumulative = states + len(frontier)
        layers.append({"level": level, "states": len(frontier), "cumulative": cumulative})
        peak_frontier = max(peak_frontier, len(frontier))
        nxt = []
        for node, code, fb, ib in frontier:
            states += 1
            aff, flags, items = unpack(code, fb, ib)
            for v in VARS:
                aff_vals[v].add(aff[v])
            flag_bm.add(fb)
            item_bm.add(ib)
            cur = by_id[node]
            if cur.get("ending"):
                continue
            st = {"aff": aff, "flags": flags, "items": items}
            for c in (cur.get("choices") or []):
                if not eval_cond(c.get("when"), st):
                    continue
                st2 = apply_effects(c.get("effects") or {}, st)
                k = canon(c["goto"], st2["aff"], st2["flags"], st2["items"])
                if k not in seen:
                    seen.add(k)
                    c2, b2, g2 = pack(st2["aff"], st2["flags"], st2["items"])
                    nxt.append((c["goto"], c2, b2, g2))
        frontier = nxt
        level += 1
        if states > STATE_BUDGET:
            budget_hit = True
            break

    elapsed = round(time.time() - t0, 1)
    maxrss_mb = round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e6, 1)
    aff_range = {v: {"distinct": len(aff_vals[v]),
                     "min": min(aff_vals[v]), "max": max(aff_vals[v])} for v in VARS}

    print(f"起点 {start}  层数 {len(layers)}  峰值 frontier {peak_frontier}")
    print(f"{'level':>6}{'states':>12}{'cumulative':>14}")
    for row in layers:
        print(f"{row['level']:>6}{row['states']:>12}{row['cumulative']:>14}")
    print(f"可达状态 {states}(BFS 完整枚举,预算 {STATE_BUDGET},超限={budget_hit})")
    print(f"好感度值域(声明域 0–100 共 101 个取值):")
    for v in VARS:
        print(f"  {v}: 取到 {aff_range[v]['distinct']} 个不同值  [{aff_range[v]['min']}, {aff_range[v]['max']}]")
    print(f"flag 位图 {len(flag_bm)} 种 / item 位图 {len(item_bm)} 种(远小于 2^18 / 2^9)")
    print(f"耗时 {elapsed}s  峰值内存 {maxrss_mb} MB")

    assert states == len(seen), "可达状态数应等于去重集大小"
    for v in VARS:
        assert AFF_LO <= aff_range[v]["min"] and aff_range[v]["max"] <= AFF_HI, "好感度越界"

    os.makedirs(OUT, exist_ok=True)
    with open(CSV_PATH, "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=["level", "states", "cumulative"])
        w.writeheader()
        w.writerows(layers)
    report = {
        "states": states,
        "state_budget": STATE_BUDGET,
        "state_budget_exceeded": budget_hit,
        "levels": len(layers),
        "peak_frontier": peak_frontier,
        "aff_range": aff_range,
        "flag_bitmaps": len(flag_bm),
        "item_bitmaps": len(item_bm),
        "elapsed_s": elapsed,
        "maxrss_mb": maxrss_mb,
        "layers": layers,
    }
    with open(JSON_PATH, "w", encoding="utf-8") as fh:
        json.dump(report, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print(f"wrote out/ne_05_2_layers.csv rows={len(layers)}")
    if budget_hit:
        print(f"FAIL 状态数 {states} 超预算 {STATE_BUDGET}")
        return 1
    print(f"OK ne_05_2_reachable_enumeration:{states} 个可达状态,去重键=规范串,未超预算")
    return 0


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

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

起点 d_pier_1  层数 32  峰值 frontier 2498621
 level      states    cumulative
     0           1             1
     1           2             3
     2           4             7
     3          10            17
     4          21            38
     5          25            63
     6          50           113
     7         114           227
     8         278           505
     9         353           858
    10         706          1564
    11        1159          2723
    12        1771          4494
    13        3442          7936
    14        5827         13763
    15       10369         24132
    16       13773         37905
    17       29061         66966
    18       62173        129139
    19       81157        210296
    20      147549        357845
    21      270963        628808
    22      443922       1072730
    23      451425       1524155
    24      882512       2406667
    25     1641572       4048239
    26      644236       4692475
    27     1124261       5816736
    28     2498621       8315357
    29      226849       8542206
    30      374706       8916912
    31      978511       9895423
可达状态 9895423(BFS 完整枚举,预算 12000000,超限=False)
好感度值域(声明域 0–100 共 101 个取值):
  hui: 取到 35 个不同值  [10, 97]
  mu: 取到 16 个不同值  [25, 100]
  shen: 取到 12 个不同值  [45, 100]
  yan: 取到 21 个不同值  [0, 100]
flag 位图 711 种 / item 位图 36 种(远小于 2^18 / 2^9)
耗时 95.3s  峰值内存 3132.8 MB
wrote out/ne_05_2_layers.csv rows=32
OK ne_05_2_reachable_enumeration:9895423 个可达状态,去重键=规范串,未超预算

判读:三层读数

第一,32 层、990 万状态、峰值 frontier 250 万。 层号就是「从起点到该状态的最少步数」;表里 states 是该层新增的状态数、cumulative 是累计。前十几层增长平缓(第 13 层才 3442),第 20 层起加速,第 25 层单层新增 164 万,第 28 层冲到 249 万——这就是组合增长的形状。它不是指数:末几层反而回落(第 29 层 22.7 万),因为越接近结局节点、可继续扩展的分支越少。

第二,好感度值域远小于声明域。

变量声明域实际取到实测区间
yan10121[0, 100]
hui10135[10, 97]
shen10112[45, 100]
mu10116[25, 100]

4 个变量合起来只有 21 × 35 × 12 × 16 = 141120 种取值组合,远小于 101⁴。这正是第 5.1 节 B1 那一层的来源:好感度不是自由变量,它的取值被路径上的 add 效果限制在少数几个点上。

第三,预算与内存口径。 预算 12000000(与 sim/na_00_spec_check.py 同口径),实测 9895423,未超限;峰值内存 3132.8 MB,耗时 95.3 秒。990 万这个规模让「穷举」这条路走得通,但也只是刚刚走得通——放大一档就会退化成采样(第 7 章 讲采样上界与剪枝)。

自检表

检查项合格标准
去重键无路径规范串只含状态、不含 path,同一状态只计一次
层号即最短步数逐层 BFS,首次出现即最小层
值域真跑每个变量的不同值数来自枚举,非估计
预算显式预算值与超限标志都落进 JSON
不变量成对状态数 = 去重集大小;好感度 ∈ [0,100]

知道了有哪些状态,下一个问题是:有没有哪个节点或选项,一次都没被访问过。这是下一节的事。

阅读导航:上一节:5.1 状态空间的规模 · 下一节:5.3 不可达节点与孤儿结局 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「narrative」更多文章

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