15.2 求值与遍历的开销

#!/usr/bin/env python3
# file: sim/ne_15_2_cost.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_15_2_cost.py
"""卷二 15.2 求值与遍历的开销（S 层）：量条件求值的单位成本、一次完整 BFS 的代价，
以及求值次数与规格规模的比值。

三段读数：
  ① 单次 eval_cond 求值耗时——取规格里全部**真实** when 条件，每轮求值 BATCH 次、
     共 ROUNDS 轮，报每轮「单次成本」的**中位数**（单次计时会被计时器抖动吃掉）。
  ② 一次完整 BFS（无深度上限，预算 12000000）——层数、可达状态数、峰值 frontier、
     峰值内存 maxrss、耗时。
  ③ 比值——求值总次数 / 可达状态数、/ 选项总数、/ 节点数。

**口径纪律**：所有耗时数字都是本机单线程的 **proxy**（受缓存与 GC 影响，逐轮波动），
只做**相对比较**，不得读成绝对性能，也不得写成「提升 N 倍」。BFS 与
sim/ne_05_2_reachable_enumeration.py 同口径（同起点、同预算、同打包去重），可交叉核对。
只读 config/story_spec.json。产出 out/ne_15_2_cost.json。exit 0 表示 BFS 未超预算。
"""
from __future__ import annotations

import json
import os
import resource
import statistics
import sys
import time

import ne_05_1_scale as s1

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
STATE_BUDGET = 12000000
BATCH, ROUNDS = 20000, 101


def collect_conds(spec: dict) -> tuple:
    """规格里全部真实 when 条件（去重、按出现顺序），以及没有 when 的选项数。"""
    seen, real, n_none = set(), [], 0
    for n in spec["dialogue"]:
        for c in (n.get("choices") or []):
            if c.get("when") is None:
                n_none += 1
                continue
            k = json.dumps(c["when"], ensure_ascii=False, sort_keys=True)
            if k not in seen:
                seen.add(k)
                real.append(c["when"])
    return real, n_none


def init_state(spec: dict) -> dict:
    """起点处的初始状态（好感度取 characters[].affinity.init，flag / item 全 0）。"""
    VARS = sorted(spec["state"]["vars"])
    return {"aff": {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
                    for v in VARS},
            "flags": {f: 0 for f in sorted(spec["state"]["flags"])},
            "items": {i: 0 for i in sorted(spec["state"]["items"])}}


def bench_eval(conds: list, st: dict) -> dict:
    """单次 eval_cond 的成本：每轮 BATCH 次，取 ROUNDS 轮「单次成本」的中位数。"""
    per_round, sink = [], 0
    for r in range(ROUNDS):
        t0 = time.perf_counter_ns()
        for i in range(BATCH):
            sink += s1.eval_cond(conds[(r + i) % len(conds)], st)
        per_round.append((time.perf_counter_ns() - t0) / BATCH)
    per_round.sort()
    return {"batch": BATCH, "rounds": ROUNDS, "calls": BATCH * ROUNDS, "sink": sink,
            "median_ns": round(statistics.median(per_round), 1),
            "p10_ns": round(per_round[ROUNDS // 10], 1),
            "p90_ns": round(per_round[ROUNDS * 9 // 10], 1),
            "min_ns": round(per_round[0], 1), "max_ns": round(per_round[-1], 1)}


def bfs(spec: dict) -> dict:
    """一次完整 BFS：层数、可达状态数、峰值 frontier、eval_cond 调用次数与耗时。"""
    t0 = time.time()
    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"])

    def pack(st):
        code = 0
        for v in VARS:
            code = (code << 7) | st["aff"][v]
        return (code, sum(1 << i for i, f in enumerate(FLAGS) if st["flags"][f]),
                sum(1 << i for i, k in enumerate(ITEMS) if st["items"][k]))

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

    frontier = [(spec["meta"]["start"],) + pack(init_state(spec))]
    seen = set(frontier)
    states = evals = ending_states = 0
    peak, levels, budget_hit = 1, 0, False
    while frontier:
        levels += 1
        peak = max(peak, len(frontier))
        nxt = []
        for node, code, fb, ib in frontier:
            states += 1
            cur = by_id[node]
            if cur.get("ending"):
                ending_states += 1
                continue
            st = unpack(code, fb, ib)
            for c in (cur.get("choices") or []):
                evals += 1
                if not s1.eval_cond(c.get("when"), st):
                    continue
                k = (c["goto"],) + pack(s1.apply_effects(c.get("effects") or {}, st))
                if k not in seen:
                    seen.add(k)
                    nxt.append(k)
            if states > STATE_BUDGET:
                budget_hit = True
                break
        frontier = nxt
        if budget_hit:
            break
    return {"levels": levels, "states": states, "peak_frontier": peak, "evals": evals,
            "ending_states": ending_states, "dedup": len(seen), "budget_hit": budget_hit,
            "elapsed_s": round(time.time() - t0, 1),
            "maxrss_mb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e6, 1)}


def main() -> int:
    spec = s1.load_spec()
    nodes, choices = len(spec["dialogue"]), spec["meta"]["scale"]["choices"]
    real, n_none = collect_conds(spec)
    bench = bench_eval(real, init_state(spec))
    print(f"规格：{nodes} 节点 / {choices} 选项 / {len(spec['state']['flags'])} flag / "
          f"{len(spec['state']['vars'])} 好感度 / {len(spec['state']['items'])} 物品；"
          f"真实 when 条件 {len(real)} 条，另有 {n_none} 条选项无 when")
    print(f"① 单次 eval_cond 求值：{BATCH} 次/轮 × {ROUNDS} 轮，取每轮单次成本的中位数")
    print("   proxy：本机单线程，受缓存与 GC 影响，只做相对比较")
    print(f"   中位数 {bench['median_ns']} ns/次   p10 {bench['p10_ns']}  p90 {bench['p90_ns']}"
          f"   min {bench['min_ns']}  max {bench['max_ns']}   合计 {bench['calls']} 次调用")

    res = bfs(spec)
    print(f"② 一次完整 BFS（无深度上限，预算 {STATE_BUDGET}）")
    print(f"   层数 {res['levels']}  可达状态 {res['states']}  峰值 frontier {res['peak_frontier']}"
          f"  超预算={res['budget_hit']}")
    print(f"   eval_cond 调用 {res['evals']} 次  落在结局节点的状态 {res['ending_states']}"
          f"（{res['ending_states'] / res['states'] * 100:.1f}%，不产生求值）")
    print(f"   耗时 {res['elapsed_s']}s  峰值内存 {res['maxrss_mb']} MB（proxy）")

    ratio = {"evals_per_state": res["evals"] / res["states"],
             "evals_per_choice": res["evals"] / choices,
             "evals_per_node": res["evals"] / nodes,
             "states_per_s": res["states"] / res["elapsed_s"],
             "evals_per_s": res["evals"] / res["elapsed_s"]}
    print("③ 求值次数与规格规模的比值")
    print(f"   求值/状态 {ratio['evals_per_state']:.3f}   求值/选项 {ratio['evals_per_choice']:.1f}"
          f"   求值/节点 {ratio['evals_per_node']:.1f}")
    print(f"   遍历速率（proxy）：状态 {ratio['states_per_s']:.0f}/s   求值 {ratio['evals_per_s']:.0f}/s")
    assert res["states"] == res["dedup"], "可达状态数应等于去重集大小"

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_15_2_cost.json"), "w", encoding="utf-8") as fh:
        json.dump({"nodes": nodes, "choices": choices, "real_conds": len(real),
                   "choices_without_when": n_none, "bench": bench, "bfs": res,
                   "ratio": ratio}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print("wrote out/ne_15_2_cost.json")
    print(f"OK ne_15_2_cost：BFS {res['levels']} 层 {res['states']} 状态，"
          f"单次求值中位数 {bench['median_ns']} ns（proxy），未超预算")
    return 0


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

规格：48 节点 / 115 选项 / 18 flag / 4 好感度 / 9 物品；真实 when 条件 24 条，另有 81 条选项无 when
① 单次 eval_cond 求值：20000 次/轮 × 101 轮，取每轮单次成本的中位数
   proxy：本机单线程，受缓存与 GC 影响，只做相对比较
   中位数 169.9 ns/次   p10 166.4  p90 180.9   min 165.9  max 546.0   合计 2020000 次调用
② 一次完整 BFS（无深度上限，预算 12000000）
   层数 32  可达状态 9895423  峰值 frontier 2498621  超预算=False
   eval_cond 调用 19121205 次  落在结局节点的状态 4211116（42.6%，不产生求值）
   耗时 60.8s  峰值内存 2153.0 MB（proxy）
③ 求值次数与规格规模的比值
   求值/状态 1.932   求值/选项 166271.3   求值/节点 398358.4
   遍历速率（proxy）：状态 162754/s   求值 314494/s
wrote out/ne_15_2_cost.json
OK ne_15_2_cost：BFS 32 层 9895423 状态，单次求值中位数 169.9 ns（proxy），未超预算