15.3 规模告警

给规模定四条软告警阈值:可达状态数 > 1e7、单次 BFS > 120 秒、峰值内存 > 4 GB、单条路径步数 > 40。当前规格四条全未越(9895423 状态、58.6 秒、2147.1 MB、31 步);把 flag 放大到 30 后越过三条,而内存那条在两次复跑里一次越一次未越,脚本恒 exit 0。

15.3 规模告警

玩法机制与可执行规格 那套量的是数值仿真的规模上限——回合数、实体数、每帧开销各自撞到哪条线;本节量的是叙事引擎的规模上限。15.1 量的是「规模涨一档,可达状态数涨多少」,15.2 量的是「一次求值与一次遍历各要多少成本」,这一节把两者合起来,回答最后一个问题:多贵算贵。

四条阈值各自对应一个真实后果

阈值不能拍脑袋定,每条都要能指到一个具体的失败模式:

指标阈值跨过之后会发生什么
可达状态数> 10⁷完整枚举走不完,必须退回采样(7.2 的采样上界)
单次 BFS 耗时> 120 s一次回归跑不动 CI,只能改成按需触发
峰值内存> 4096 MB普通开发机与 CI runner 上会 OOM
单条路径步数> 40玩家一次通关要走四十多步,节奏与可玩性先出问题

前三条是机器的线,第四条是玩家的线——它和前三条的来源完全不同:前三条随状态空间涨,第四条只随图的结构深度涨。

S 层:规模指标探针

先把四个指标量出来,判据留给下一小节。前三项来自一次完整 BFS(复用 15.2 的内核与同一预算 12000000),第四项是 DAG 最长路径(精确,忽略 when 与状态)。两个场景各在独立子进程里跑,这样 ru_maxrss 只反映本场景,不被上一场景的高水位污染。

#!/usr/bin/env python3
# file: sim/ne_15_3_scale_probe.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_15_3_scale_probe.py
"""卷二 15.3 规模指标探针(S 层):把四个规模指标在 base 与放大两个场景下量出来。

指标:可达状态数、单次 BFS 耗时(proxy)、峰值内存(proxy)、单条路径步数。前三个来自
一次完整 BFS(复用 sim/ne_15_2_cost.py 的内核与同一预算),第四个是 DAG 最长路径(精确,
忽略 when 与状态)。放大场景把 flag 18→30,挂载点复用 15.1 的流量口径。

每个场景在**独立子进程**里跑,这样 ru_maxrss 只反映该场景,不被上一场景的高水位污染。
本脚本只**测量**:判据与告警在 sim/ne_15_3_scale_warning.py(B 层)。只读
config/story_spec.json。产出 out/ne_15_3_scale_probe.json。exit 0 表示两个场景都出了读数。
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from collections import Counter

import ne_05_1_scale as s1
import ne_15_1_scale_scan as s15
import ne_15_2_cost as s152

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
DEPTH_CAP = 20
SCENARIOS = ("base", "amp")


def longest_path(spec: dict) -> int:
    """DAG 上从起点出发的最长路径步数(精确,忽略 when 与状态)。"""
    by_id = {n["id"]: n for n in spec["dialogue"]}
    memo: dict = {}

    def dp(nid: str) -> int:
        if nid not in memo:
            ch = by_id[nid].get("choices") or []
            memo[nid] = 0 if not ch else 1 + max(dp(c["goto"]) for c in ch)
        return memo[nid]

    return dp(spec["meta"]["start"])


def build(scenario: str) -> dict:
    """场景 → 规格:base 原样;amp 把 flag 18→30,挂载点复用 15.1 的流量口径。"""
    spec = s1.load_spec()
    if scenario == "base":
        return spec
    traffic: Counter = Counter()
    s15.walk(spec, DEPTH_CAP, 10 ** 9, traffic)
    return s15.amplify(spec, s15.busiest_slots(traffic, 16), 30, 4, 9, 48)


def measure(scenario: str) -> dict:
    """跑一个场景:完整 BFS(与 15.2 同内核同预算)+ 最长路径 + 规格规模。"""
    spec = build(scenario)
    res = s152.bfs(spec)
    res.update({"scenario": scenario, "max_steps": longest_path(spec),
                "nodes": len(spec["dialogue"]),
                "flags": len(spec["state"]["flags"])})
    return res


def main() -> int:
    results = []
    for sc in SCENARIOS:
        proc = subprocess.run([sys.executable, os.path.abspath(__file__), "--scenario", sc],
                              capture_output=True, text=True, check=False)
        if proc.returncode != 0 or not proc.stdout.strip():
            print(f"FAIL 场景 {sc} 子进程未产出读数(退出码 {proc.returncode})")
            return 1
        results.append(json.loads(proc.stdout.strip().split("\n")[-1]))

    for res in results:
        print(f"场景 {res['scenario']}({res['nodes']} 节点 / {res['flags']} flag)"
              f"{'  超预算截断' if res['budget_hit'] else ''}")
        print(f"  可达状态数 {res['states']}  层数 {res['levels']}"
              f"  峰值 frontier {res['peak_frontier']}")
        print(f"  单次 BFS 耗时 {res['elapsed_s']}s  峰值内存 {res['maxrss_mb']} MB(均为 proxy)")
        print(f"  单条路径步数 {res['max_steps']}  eval_cond 调用 {res['evals']}")
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_15_3_scale_probe.json"), "w", encoding="utf-8") as fh:
        json.dump({"depth_cap": DEPTH_CAP, "scenarios": results}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print("wrote out/ne_15_3_scale_probe.json")
    print(f"OK ne_15_3_scale_probe:{len(results)} 个场景的四个规模指标已量出")
    return 0


if __name__ == "__main__":
    if len(sys.argv) > 2 and sys.argv[1] == "--scenario":
        print(json.dumps(measure(sys.argv[2]), ensure_ascii=False))
        sys.exit(0)
    sys.exit(main())

B 层:软告警

按 B 层的口径,跨过阈值只打印 WARN 并计入 JSON,恒 exit 0——它量的是「扰动参数后哪里相变」,不是 CI 门禁。判据脚本自己不做 BFS,只读探针产物;产物缺失也只 WARN 不失败。

#!/usr/bin/env python3
# file: sim/ne_15_3_scale_warning.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/ne_15_3_scale_warning.py
"""卷二 15.3 规模告警(B 层,软告警不硬失败,恒 exit 0)。

读 sim/ne_15_3_scale_probe.py 量出的四个规模指标,套四条阈值:

  可达状态数   > 1e7      枚举预算走不完,该转采样
  单次 BFS 耗时 > 120 s    proxy,本机单线程
  峰值内存     > 4096 MB   proxy
  单条路径步数 > 40        玩家一次通关要走多少步

阈值是**判据**,不是目标。跨过只打印 WARN 并计入 JSON,**恒 exit 0**——B 层量的是
「扰动参数后哪里相变」,不是 CI 门禁。测量与判据分离:本脚本只读探针产物,自己不做
BFS。产物缺失也只 WARN 不失败。只读 out/ne_15_3_scale_probe.json。产出
out/ne_15_3_scale_warning.json。
"""
from __future__ import annotations

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")
PROBE = os.path.join(OUT, "ne_15_3_scale_probe.json")
THRESHOLDS = [
    {"key": "states", "label": "可达状态数", "limit": 10 ** 7, "unit": "个"},
    {"key": "elapsed_s", "label": "单次 BFS 耗时", "limit": 120.0, "unit": "s"},
    {"key": "maxrss_mb", "label": "峰值内存", "limit": 4096.0, "unit": "MB"},
    {"key": "max_steps", "label": "单条路径步数", "limit": 40, "unit": "步"},
]


def fmt(v) -> str:
    return f"{v:.1f}" if isinstance(v, float) else str(v)


def main() -> int:
    if not os.path.exists(PROBE):
        print(f"WARN 缺少 {os.path.relpath(PROBE, ROOT)},请先跑 sim/ne_15_3_scale_probe.py")
        return 0
    with open(PROBE, encoding="utf-8") as fh:
        probe = json.load(fh)

    print("阈值表(判据,不是目标)")
    for t in THRESHOLDS:
        suffix = "(proxy)" if t["key"] in ("elapsed_s", "maxrss_mb") else ""
        print(f"  {t['label']:<12} > {t['limit']} {t['unit']}{suffix}")
    print()
    warns: list[str] = []
    for res in probe["scenarios"]:
        print(f"场景 {res['scenario']}({res['nodes']} 节点 / {res['flags']} flag)"
              f"{'  超预算截断' if res['budget_hit'] else ''}")
        for t in THRESHOLDS:
            v = res[t["key"]]
            over = v > t["limit"]
            print(f"  {t['label']:<12} {fmt(v):>12} {t['unit']:<4}"
                  f"{'越过' if over else '未越'}(阈值 {t['limit']})")
            if over:
                warns.append(f"场景 {res['scenario']}:{t['label']} {fmt(v)} "
                             f"越过阈值 {t['limit']} {t['unit']}")
        print()
    for w in warns:
        print(f"WARN {w}")
    if not warns:
        print("无软告警:所有场景均未跨过任何阈值")

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_15_3_scale_warning.json"), "w", encoding="utf-8") as fh:
        json.dump({"thresholds": THRESHOLDS, "scenarios": probe["scenarios"],
                   "warnings": warns}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print("wrote out/ne_15_3_scale_warning.json")
    print(f"OK ne_15_3_scale_warning:{len(probe['scenarios'])} 个场景,"
          f"{len(warns)} 条软告警,exit 0(B 层口径)")
    return 0


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

本机实跑

先跑探针:/opt/homebrew/bin/python3 sim/ne_15_3_scale_probe.py(exit 0,两个子进程,整轮约 198 秒):

场景 base(48 节点 / 18 flag)
  可达状态数 9895423  层数 32  峰值 frontier 2498621
  单次 BFS 耗时 58.6s  峰值内存 2147.1 MB(均为 proxy)
  单条路径步数 31  eval_cond 调用 19121205
场景 amp(48 节点 / 30 flag)  超预算截断
  可达状态数 12000001  层数 26  峰值 frontier 8236197
  单次 BFS 耗时 129.9s  峰值内存 4174.5 MB(均为 proxy)
  单条路径步数 31  eval_cond 调用 32798824
wrote out/ne_15_3_scale_probe.json
OK ne_15_3_scale_probe:2 个场景的四个规模指标已量出

再跑判据:/opt/homebrew/bin/python3 sim/ne_15_3_scale_warning.py(exit 0,读产物,瞬时):

阈值表(判据,不是目标)
  可达状态数        > 10000000 个
  单次 BFS 耗时    > 120.0 s(proxy)
  峰值内存         > 4096.0 MB(proxy)
  单条路径步数       > 40 步

场景 base(48 节点 / 18 flag)
  可达状态数             9895423 个   未越(阈值 10000000)
  单次 BFS 耗时            58.6 s   未越(阈值 120.0)
  峰值内存               2147.1 MB  未越(阈值 4096.0)
  单条路径步数                 31 步   未越(阈值 40)

场景 amp(48 节点 / 30 flag)  超预算截断
  可达状态数            12000001 个   越过(阈值 10000000)
  单次 BFS 耗时           129.9 s   越过(阈值 120.0)
  峰值内存               4174.5 MB  越过(阈值 4096.0)
  单条路径步数                 31 步   未越(阈值 40)

WARN 场景 amp:可达状态数 12000001 越过阈值 10000000 个
WARN 场景 amp:单次 BFS 耗时 129.9 越过阈值 120.0 s
WARN 场景 amp:峰值内存 4174.5 越过阈值 4096.0 MB
wrote out/ne_15_3_scale_warning.json
OK ne_15_3_scale_warning:2 个场景,3 条软告警,exit 0(B 层口径)

判读:三条线对规模敏感,一条只对结构敏感

第一,当前规格四条全未越,但状态数贴着边。 base 场景 9895423 个可达状态,离 10⁷ 只差 104577 个(余量 1.05%);58.6 秒对 120 秒还有近一倍余量;2147.1 MB 对 4096 MB 也有近一倍。上一节已经说明这 990 万个状态每个还要付约 1.93 次求值与一次打包去重,所以「离阈值还有 1%」并不等于安全。

第二,放大 flag 18→30 之后跨过三条线。 可达状态数在 12000001 处被预算截断(这个读数是下界,真实值更大),跨过 10⁷;耗时 129.9 秒,跨过 120 秒;峰值内存 4174.5 MB,跨过 4096 MB。三条告警同一个原因:状态数涨了,求值次数(19121205 → 32798824,×1.72)与打包去重次数跟着涨。注意 amp 的层数只有 26(base 是 32)——它是被截断在半路的,峰值 frontier 却有 8236197,是 base 峰值(2498621)的 3.3 倍:放大之后单层 frontier 本身就吃掉大部分内存。

第三,内存那条线在阈值附近会抖。 同一段测量逻辑的另一次复跑,amp 的峰值内存读数是 4083.9 MB——在 4096 MB 下方,三条告警只剩两条。两次读数差 90.6 MB(约 2%),跨不跨线完全取决于这一次运行。这正是 proxy 口径要提醒的:贴着阈值的读数不能当定论,要么把阈值抬高到留出抖动余量,要么连续跑几次取最坏值。

第四,单条路径步数对 flag 完全不敏感。 两个场景都是 31 步——放大 flag 只加状态维度,不加图深度。要跨过 40 步这条线,得往对话图里加层(新场景、新回合),而不是加 flag 或物品。这是 15.1 里「节点轴最钝」那条结论的另一面:节点计数不变深度,深度不变步数。

第五,阈值只是判据,不是目标。 这四条线的作用是让「该转采样了」「该拆规格了」有一个可复算的触发点;它们不构成 CI 门禁——判据脚本恒 exit 0,跨线只打 WARN。所有耗时与内存都是本机 proxy,换机器要重新标定阈值。

自检表

检查项合格标准
测量与判据分离S 层只出读数,B 层只套阈值,B 层自己不做 BFS
软告警不硬失败跨线只打印 WARN 并落 JSON,return 0 恒成立
子进程隔离每个场景独立进程,ru_maxrss 不跨场景污染
截断值标注超预算的场景显式标「超预算截断」,读数为下界
三处同口径放大复用 15.1 挂载点,BFS 复用 15.2 内核与预算

规模这条线到此为止。从下一章起,问题从「引擎能跑多大」换成「引擎装对了没有」——16.1 把全卷的检查项收成一张装配表。

阅读导航:上一节:15.2 求值与遍历的开销 · 下一节:16.1 引擎的装配检查表 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「narrative」更多文章

  1. 15.2 求值与遍历的开销
  2. 15.1 规格规模的增长
  3. 附录 A 情感模型公式、参数表与数据集索引