gahow/aituner
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add reproducible CollectiveSpec opportunity screen

Browse Source
This commit is contained in:
Gahow Wang
2026-07-13 15:09:01 +08:00
parent 6db7308558
commit 9b3a2eab80
6 changed files with 1055 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
docs/collectivespec-pilot-design-20260713.md Normal file
View File

@@ -0,0 +1,157 @@
# CollectiveSpec：先做机会判定的实验设计
## 结论先行
当前不应先实现一个“按请求动态 K、再做一次 DP collective 同步”的原型。该路径的
工程修复很薄，而且相邻公开工作已经覆盖了 request-level dynamic speculation 与
ragged verification 的大量空间。CollectiveSpec 只有在
一个更强、可证伪的事实成立时才值得继续：**在 wide-EP MoE、DP>1 的生产负载中，
不同 DP rank / request 所需的投机深度确实不同，并且全局 static K 明显浪费了 SLO
可行 goodput。**
本文件把第一轮实验定义为一个机会判定（opportunity gate），不是最终性能主张。
## 固定条件
- Host: `dash0`, 8x NVIDIA H20。
- Model: Qwen3-235B-A22B FP8；draft: EAGLE3。
- Deployment: TP=4, DP=2, EP=8，`VLLM_MOE_USE_DEEPEP=1`。
- Trace: `thinking_w20260327_1000`，600 秒 decode-only 窗口。
- SLO: TPOT <= 40 ms，pass rate >= 0.95。
- 同一 engine revision、同一模型/trace 路径、同一环境变量；实验串行执行，避免 GPU
互相干扰。
这里的 resolved topology 来自远端实际 StudySpec，而不是仓库 README 中可能已过期的
配置描述。
## 假设与可证伪指标
### G0：static-K 是否有足够可利用的空间？
- H0：在该固定拓扑和负载下，NoSpec/K=1/2/3 的 SLO-goodput 差异很小；最佳固定 K 已经
足够好。此时停止 CollectiveSpec。
- H1：不同 static K 的可行前沿存在实质差异，且最优 K 对负载区间敏感。只有 H1 才
说明 dynamic policy 可能有直接性能价值。
主要指标：每个 K 在相同 SLO 下可达到的最大 `sampling_u`，以及对应请求率、TPOT
pass rate、p50/p95 TPOT、成功/失败原因。`sampling_u` 是现有 replay 使用的一致 trace
抽样旋钮，因此只能作为此 trace 的 SLO-goodput proxy，不能直接外推为线上 QPS。
本 trace 的 output-length 分布很重尾；replayer 的 drain deadline 可能在长输出尚未完成
时终止 probe。故每个结果必须同时报告 completion-success count 和 deadline failure；不能
只因 TPOT pass rate 达标就把截断 request 当作“无成本”。本轮的原始 static screen 仍沿用
现有 SLO 以便和项目 baseline 可比，但它不能替代完整 completion 的确认实验。
判定门槛（预注册）：
1. 在 K=1/2/3 之间，最优 K 相对次优 K 的最大可行 `sampling_u` 小于 5%，或
置信区间/重复实验重叠很大：
**停止**。
2. 最优 speculative K 比次优 speculative K 至少高 10%，且在两个独立重复中方向一致：
进入 G1。NoSpec 仅作为“是否值得用 draft model”的部署对照，不能替代这条判据。
3. 如果 K=3 不受当前 engine 支持、任一配置启动失败，记录为兼容性结果，不把它误作
性能差。
## 第一阶段：static-K screening
配置为 `{NoSpec, K=1, K=2, K=3}`。NoSpec 会删除 `--speculative-config`，而不是传
非法的 `num_speculative_tokens=0`。它释放 draft model 相关资源，因此不等于
same-stack 的 logical K=0；后者在 EAGLE 类实现中仍可能需要一次 draft forward 来保持
KV 同步。当前 dash0 binary 的 MLA indexer 明确限制 `num_speculative_tokens <= 3`，
故这已穷尽该 binary 的合法 static horizon。原计划每个配置：
- 搜索范围 `sampling_u in [0.005, 0.020]`；
- 最多 3 次 probe、tolerance=0.003；
- 每个 probe 使用完整 600 秒 trace replay（不会使用 `max_requests_per_probe` 的
截断模式）；
- 启动顺序 `2,1,3,0`，降低冷启动或时间漂移与 K 单调对应的风险；
- 每个 K 都使用独立 Store、不可变派生 StudySpec、完整 stdout/stderr log。
这是一轮筛选而非 final frontier。它一旦显示值得继续，才对 top-2 K 做交叉顺序的完整
搜索与至少两次重复。
### 2026-07-13 数据质量修正：controlled screen
原始 trace 第一个 K=2 probe 暴露了一个不应隐藏的 measurement red flag：worker 的
drain deadline 按 selected set 的 p99 output length 计算，而该 set 含一个 36,034-token
completion。该 request 因 deadline 被裁掉；尽管 TPOT-only pass rate 仍可达标，censoring
会随 `sampling_u` 改变 selected set，不能用于比较 static-K frontier。
因此原始 run 只保留为诊断 artifact，停止后改跑一个明确标为 **controlled** 的 screen：
保持相同 arrival、prompt、sampling seed 和 topology，但把每个 request 的
`min_tokens=max_tokens=4096`。4096 接近原始 output mean 3,924.6，且使 p99 deadline 覆盖
每个 request 的完整 completion。它回答的是“在相同输入/到达条件下，static K 是否留下
可利用空间”，不是 production trace 的最终 goodput 结论。最终论文实验必须同时有：
1. 该受控 curve（机制和可重复性）；
2. 原始长度 trace 的完整-completion 版本（不能使用 p99 censoring）；
3. 至少一个 held-out window。
### 2026-07-13 数据质量修正 #2：fresh-engine fixed grid
受控 screen 的第一条 K=2、`u=0.0125` probe 本身通过了完整性检查（152/152 success、
usage 返回的 completion token 均精确为 4096、无 early stop）。但随后发现原二分搜索会
在同一 engine 内连续执行多个 probe；第二、三 probe 继承前一 probe 的 prefix/KV cache，
也继承全局 RNG 的已消耗状态。不同 K 的二分分支/中止路径不同，因此不能把该搜索输出的
`best_sampling_u` 差异直接归因于 K。该 run 在第二 probe 中主动停止，**不作为 G0 结果**。
替代协议是每个 `(offered-load, K)` 只运行一次、每次均启动一个 fresh engine。两个固定
负载由原始 immutable trace 的统一 `sampling_u` 阈值物化：`u=0.0125`（152 requests,
0.2533 req/s）和 `u=0.0200`（263 requests, 0.4383 req/s）。物化后的 request 仍保留原
prompt、arrival 和 sampling provenance，但强制 `temperature=0`、显式 engine `seed=0`，
并用统一的 4096 completion override。每个 K 只有一个 probe，所以 accelerator KV/prefix
cache 为空且 RNG 从相同 seed 开始；每次都从 `probe_details.jsonl` 验证：
1. `early_stopped=false`；
2. outcome count = selected count，且每个 request success；
3. TTFT/TPOT 均非空；
4. completion token 的 source 为 usage，且实际/预期均为 4096；
5. result 无 partial-probe failure，且只包含一个 primary probe。
运行顺序在两个负载间反向（ABBA），避免 K 与时间漂移完全共线。这个 grid 仍然只回答
static-K 是否存在足够大的机会；它不估计 production sampling goodput，也不证明 rank-local
K 的上界。当前 vLLM deployment 的 `reasoning_parser=''`；其 SSE 实现将生成文本放在
`delta.content`，所以本协议中的 token-time 定义覆盖当前 `<think>` 输出。若以后启用
reasoning parser，客户端必须同时记录 `reasoning_content` 后才可复用此指标。
## G1：只有在 G0 通过后才做的直接验证
目标不是“不同请求有不同 K”这种已经很常见的说法，而是验证下面的系统命题：
> 在 DP+EP MoE 下，局部独立的 K 决策会让 collective 序列分歧；把它们编译为
> rank-agreement 的 ragged execution plan，可保留异质请求的计算节省，同时不改变
> collective order。
当前 dash0 vLLM 已经有一个很好的切入点：每个 DP step 会 all-reduce 一段 metadata，
并把各 rank 的 total token count padding 到最大值；CUDA graph mode 也会取跨 rank 的
共同模式。这说明论文的最小机制不应另造一个 scheduler，而应把现有 scalar
`(num_tokens, num_reqs, graph_mode)` agreement 扩展为 canonical speculative-plan header。
关键增量是让 header 描述真实 active frontiers，并保证后续 verifier/EP split vector 的
collective ordinal 相同；若最后仍 padding 到 global max，就没有可主张的性能机制。
需要实现/测量：
1. **oracle trace replayer**：利用 G0 的 per-K service curve，为每个到达时刻选择
SLO-feasible K，比较 best-static K 与 oracle 的 upper bound。若 oracle gain <10%，
停止，避免把噪声当论文方向。
2. **collective trace**：按 DP rank 记录每个 decode step 的 collective 序列、token
shape、active-sequence mask、MoE all-to-all bytes 和 rank idle time。验证“local K
不同”是否真的导致 sequence divergence，而不是仅是一个 API 限制。
3. **CollectiveSpec prototype**：固定 collective order，用全局 agreement header 和
ragged/padded verification plan；对比 `best static K`、global-max-K、oracle 和当前
upstream dynamic-spec baseline（包括 DSpark/FASER 能实现的部分）。
4. **ablation**：去掉 agreement、去掉 ragged packing、去掉 queue/SLO policy；报告
goodput、p50/p95/p99 TPOT、acceptance、MoE communication bytes、GPU SM/HBM util、
rank skew。
## 主要风险
- 最新 upstream 动态投机对 DP>1 的处理可能本身只需一个 global-K broadcast；那是
feature patch，不构成研究贡献。
- 当前 dash0 runtime 已验证 DP=2 + static EAGLE 可以工作；尚未在这个 binary 上证明
“local dynamic K 会 deadlock”。因此研究动机必须写成固定 EAGLE horizon 的执行限制，
不能把未运行的 dynamic-K 路径当作既成故障。
- FASER/DSpark 等相邻工作会把“dynamic K + ragged verify”作为强 baseline；必须在
做任何大实现前进行逐项复现/排除。
- trace 的 `sampling_u` 是 proxy；最终结论必须在固定 arrival trace、真实请求长度和
至少一个不同 workload 上复现。

175
scripts/collectivespec/create_controlled_greedy_window.py Normal file
View File

@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""Materialize a fixed-load, greedy subset of an immutable replay window.
The output preserves prompt/arrival/sampling provenance while setting every
selected request's temperature to zero. It is intentionally a controlled
mechanism workload, not a replacement for a production sampling trace.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
from pathlib import Path
from typing import Any
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def resolve_trace_path(windows_path: Path, trace_file: str) -> Path:
candidate = Path(trace_file)
if candidate.is_absolute():
return candidate
direct = (windows_path.parent / candidate).resolve()
if direct.exists():
return direct
if candidate.parts and candidate.parts[0] == "trace_windows":
return (windows_path.parent / Path(*candidate.parts[1:])).resolve()
return direct
def numeric(value: Any, *, field: str, row_index: int) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise SystemExit(f"row {row_index}: {field} must be numeric")
value = float(value)
if not math.isfinite(value):
raise SystemExit(f"row {row_index}: {field} must be finite")
return value
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--windows-path", type=Path, required=True)
parser.add_argument("--window-id", required=True)
parser.add_argument("--sampling-u-threshold", type=float, required=True)
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--output-window-id", required=True)
args = parser.parse_args()
if not args.windows_path.is_file():
raise SystemExit(f"windows file does not exist: {args.windows_path}")
threshold = float(args.sampling_u_threshold)
if not math.isfinite(threshold) or not 0.0 <= threshold <= 1.0:
raise SystemExit("--sampling-u-threshold must be finite and in [0, 1]")
source_windows = json.loads(args.windows_path.read_text(encoding="utf-8"))
windows = source_windows.get("windows") if isinstance(source_windows, dict) else source_windows
if not isinstance(windows, list):
raise SystemExit("windows payload must contain a list")
source_window = next(
(
item
for item in windows
if isinstance(item, dict) and item.get("window_id") == args.window_id
),
None,
)
if source_window is None:
raise SystemExit(f"window id not found: {args.window_id}")
trace_file = source_window.get("trace_file")
if not isinstance(trace_file, str) or not trace_file:
raise SystemExit(f"window {args.window_id} has no trace_file")
source_trace = resolve_trace_path(args.windows_path, trace_file)
if not source_trace.is_file():
raise SystemExit(f"source trace does not exist: {source_trace}")
args.output_dir.mkdir(parents=True, exist_ok=True)
output_trace = args.output_dir / f"{args.output_window_id}.jsonl"
selected = 0
sampling_values: list[float] = []
timestamps: list[float] = []
with source_trace.open("r", encoding="utf-8") as src, output_trace.open(
"w", encoding="utf-8"
) as dest:
for row_index, line in enumerate(src):
if not line.strip():
continue
row = json.loads(line)
if not isinstance(row, dict):
raise SystemExit(f"row {row_index}: expected an object")
sampling_u = numeric(row.get("sampling_u"), field="sampling_u", row_index=row_index)
if sampling_u > threshold:
continue
timestamp = numeric(row.get("timestamp"), field="timestamp", row_index=row_index)
row["temperature"] = 0.0
dest.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n")
selected += 1
sampling_values.append(sampling_u)
timestamps.append(timestamp)
if selected == 0:
output_trace.unlink(missing_ok=True)
raise SystemExit("threshold selected zero requests")
output_window = dict(source_window)
output_window.update(
{
"window_id": args.output_window_id,
"trace_file": str(output_trace.resolve()),
"num_requests": selected,
"sampling_strategy": "fixed_uniform_score_then_greedy_temperature",
"collectivespec_source_window_id": args.window_id,
"collectivespec_sampling_u_threshold": threshold,
"collectivespec_temperature_override": 0.0,
"collectivespec_source_trace_path": str(source_trace),
"collectivespec_source_trace_sha256": sha256(source_trace),
}
)
output_windows = {
"kind": "collectivespec_controlled_greedy_window",
"source_windows_path": str(args.windows_path.resolve()),
"source_window_id": args.window_id,
"sampling_u_threshold": threshold,
"temperature_override": 0.0,
"windows": [output_window],
}
output_manifest = args.output_dir / "windows.json"
output_manifest.write_text(
json.dumps(output_windows, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
sanity = {
"n": selected,
"sampling_u": {
"min": min(sampling_values),
"max": max(sampling_values),
"distinct_value_count": len(set(sampling_values)),
"all_at_or_below_threshold": all(value <= threshold for value in sampling_values),
},
"timestamp": {
"min": min(timestamps),
"max": max(timestamps),
"non_negative": all(value >= 0 for value in timestamps),
"nondecreasing": all(
timestamps[index] >= timestamps[index - 1]
for index in range(1, len(timestamps))
),
},
"temperature": {"distinct_value_count": 1, "all_zero": True},
"source_trace_sha256": sha256(source_trace),
"output_trace_sha256": sha256(output_trace),
}
print(
json.dumps(
{
"windows_path": str(output_manifest),
"trace_path": str(output_trace),
"data_sanity": sanity,
},
ensure_ascii=False,
indent=2,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())

237
scripts/collectivespec/run_static_k_pilot.py Normal file
View File

@@ -0,0 +1,237 @@
#!/usr/bin/env python3
"""Run an isolated, reproducible static-speculation screening study.
This driver deliberately does not implement CollectiveSpec. It establishes
whether the current static-K deployment has enough headroom to justify such a
system. Each K gets an immutable derived StudySpec and a separate store.
"""
from __future__ import annotations
import argparse
import datetime as dt
import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-spec", type=Path, required=True)
parser.add_argument("--source-root", type=Path, required=True)
parser.add_argument("--output-root", type=Path, required=True)
parser.add_argument("--run-id", required=True)
parser.add_argument("--k-values", nargs="+", type=int, default=[0, 1, 2, 3])
parser.add_argument(
"--order",
nargs="+",
type=int,
help="Sequential launch order. Defaults to --k-values order.",
)
parser.add_argument("--sampling-u-low", type=float, default=0.005)
parser.add_argument("--sampling-u-high", type=float, default=0.020)
parser.add_argument("--tolerance", type=float, default=0.003)
parser.add_argument("--max-probes", type=int, default=3)
parser.add_argument(
"--completion-tokens-override",
type=int,
help="Force every request to this exact completion length for a controlled screen.",
)
parser.add_argument(
"--trace-windows-path",
type=Path,
help="Override trace.windows_path in every derived StudySpec.",
)
parser.add_argument(
"--trace-window-id",
help="Override trace.window_id in every derived StudySpec.",
)
parser.add_argument(
"--seed",
type=int,
help="Explicit engine seed recorded in every derived StudySpec.",
)
parser.add_argument(
"--port",
type=int,
help="Serving port. Defaults to the base StudySpec's engine/base flag port.",
)
parser.add_argument(
"--continue-on-error",
action="store_true",
help="Record a failed K and proceed to the next independent K.",
)
parser.add_argument("--dry-run", action="store_true")
return parser.parse_args()
def derived_spec(
base: dict[str, Any], args: argparse.Namespace, k: int
) -> dict[str, Any]:
spec = json.loads(json.dumps(base))
label = f"collectivespec-static-k{k}-screen-{args.run_id}"
spec["study_id"] = label
spec["engine"]["cwd"] = str(args.source_root)
flags = spec["engine"]["base_flags"]
port = args.port
if port is None:
port = int(flags.get("port", spec["engine"]["port"]))
spec["engine"]["port"] = port
flags["port"] = port
if k == 0:
# vLLM's EAGLE config rejects num_speculative_tokens=0. Removing the
# entire flag is the real no-speculation baseline.
flags.pop("speculative-config", None)
else:
raw_config = flags.get("speculative-config")
if raw_config is None:
raise ValueError("base spec has no speculative-config for K > 0")
config = json.loads(raw_config)
config["num_speculative_tokens"] = k
flags["speculative-config"] = json.dumps(config, separators=(",", ":"))
search = spec["search"]
search["low"] = args.sampling_u_low
search["high"] = args.sampling_u_high
search["tolerance"] = args.tolerance
search["max_probes"] = args.max_probes
if args.completion_tokens_override is not None:
spec["trace"]["completion_tokens_override"] = args.completion_tokens_override
if args.trace_windows_path is not None:
spec["trace"]["windows_path"] = str(args.trace_windows_path)
if args.trace_window_id is not None:
spec["trace"]["window_id"] = args.trace_window_id
if args.seed is not None:
flags["seed"] = args.seed
return spec
def main() -> int:
args = parse_args()
if not args.base_spec.is_file():
raise SystemExit(f"base spec does not exist: {args.base_spec}")
if not args.source_root.is_dir():
raise SystemExit(f"source root does not exist: {args.source_root}")
if any(k < 0 for k in args.k_values):
raise SystemExit("K must be non-negative")
if not 0 <= args.sampling_u_low < args.sampling_u_high <= 1:
raise SystemExit("sampling-u bounds must satisfy 0 <= low < high <= 1")
if args.max_probes < 1:
raise SystemExit("max-probes must be positive")
if args.completion_tokens_override is not None and args.completion_tokens_override < 0:
raise SystemExit("completion token override must be non-negative")
if (args.trace_windows_path is None) != (args.trace_window_id is None):
raise SystemExit(
"--trace-windows-path and --trace-window-id must be provided together"
)
if args.trace_windows_path is not None and not args.trace_windows_path.is_file():
raise SystemExit(f"trace windows file does not exist: {args.trace_windows_path}")
order = args.order or args.k_values
if sorted(order) != sorted(args.k_values) or len(order) != len(args.k_values):
raise SystemExit("--order must be a permutation of --k-values")
base = json.loads(args.base_spec.read_text())
args.output_root.mkdir(parents=True, exist_ok=True)
specs_dir = args.output_root / "specs"
specs_dir.mkdir(exist_ok=True)
manifest = {
"kind": "collectivespec_static_k_screen",
"created_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
"base_spec": str(args.base_spec),
"base_spec_sha256": sha256(args.base_spec),
"source_root": str(args.source_root),
"run_id": args.run_id,
"k_values": args.k_values,
"launch_order": order,
"sampling_u": [args.sampling_u_low, args.sampling_u_high],
"tolerance": args.tolerance,
"max_probes": args.max_probes,
"completion_tokens_override": args.completion_tokens_override,
"trace_windows_path": (
str(args.trace_windows_path) if args.trace_windows_path is not None else None
),
"trace_window_id": args.trace_window_id,
"seed": args.seed,
"port": args.port
if args.port is not None
else int(base["engine"]["base_flags"].get("port", base["engine"]["port"])),
"python": sys.executable,
}
(args.output_root / "manifest.json").write_text(
json.dumps(manifest, indent=2, sort_keys=True) + "\n"
)
failures: list[dict[str, Any]] = []
for ordinal, k in enumerate(order, start=1):
spec = derived_spec(base, args, k)
spec_path = specs_dir / f"{ordinal:02d}_k{k}.json"
spec_path.write_text(json.dumps(spec, indent=2, sort_keys=True) + "\n")
store_root = args.output_root / "stores" / f"k{k}"
command = [
sys.executable,
"-m",
"aituner.cli",
"study",
"tune",
"--spec",
str(spec_path),
"--store-root",
str(store_root),
"--max-trials",
"1",
]
print(f"[{ordinal}/{len(order)}] K={k} command={json.dumps(command)}", flush=True)
if args.dry_run:
continue
log_path = args.output_root / "logs" / f"{ordinal:02d}_k{k}.log"
log_path.parent.mkdir(exist_ok=True)
env = os.environ.copy()
env["PYTHONPATH"] = str(args.source_root / "src")
env["PYTHONDONTWRITEBYTECODE"] = "1"
with log_path.open("w") as log:
completed = subprocess.run(
command,
cwd=args.source_root,
env=env,
text=True,
stdout=log,
stderr=subprocess.STDOUT,
)
print(
f"[{ordinal}/{len(order)}] K={k} returncode={completed.returncode} log={log_path}",
flush=True,
)
if completed.returncode:
failures.append({"k": k, "returncode": completed.returncode, "log": str(log_path)})
if not args.continue_on_error:
break
result = {
"finished_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
"failures": failures,
"status": "ok" if not failures else "completed_with_failures",
}
(args.output_root / "driver_result.json").write_text(
json.dumps(result, indent=2, sort_keys=True) + "\n"
)
return 0 if not failures else 1
if __name__ == "__main__":
raise SystemExit(main())

135
scripts/collectivespec/summarize_engine_metrics.py Normal file
View File

@@ -0,0 +1,135 @@
#!/usr/bin/env python3
"""Extract per-DP-window speculative metrics from a vLLM engine log.
vLLM emits an ``Engine NNN`` throughput line followed by the corresponding
``SpecDecoding metrics`` line. This script keeps that adjacency explicit and
does not infer request-level outcomes from the aggregate metrics.
"""
from __future__ import annotations
import argparse
import json
import math
import re
import statistics
from collections import defaultdict
from pathlib import Path
from typing import Any
ENGINE = re.compile(
r"INFO (?P<clock>\d\d-\d\d \d\d:\d\d:\d\d\.\d+) .*?"
r"Engine (?P<engine>\d+): .*?"
r"Avg decode step duration: (?P<decode_ms>[0-9.]+) ms .*?"
r"Running: (?P<running>\d+) reqs, Waiting: (?P<waiting>\d+) reqs"
)
SPEC = re.compile(
r"INFO (?P<clock>\d\d-\d\d \d\d:\d\d:\d\d\.\d+) .*?"
r"SpecDecoding metrics: Mean acceptance length: (?P<mean_accept>[0-9.]+), .*?"
r"Avg Draft acceptance rate: (?P<accept_pct>[0-9.]+)%"
)
def percentile(values: list[float], q: float) -> float | None:
if not values:
return None
values = sorted(values)
index = (len(values) - 1) * q
lower, upper = math.floor(index), math.ceil(index)
if lower == upper:
return values[lower]
return values[lower] + (values[upper] - values[lower]) * (index - lower)
def describe(values: list[float]) -> dict[str, Any]:
return {
"n": len(values),
"mean": statistics.fmean(values) if values else None,
"min": min(values) if values else None,
"p50": percentile(values, 0.5),
"p95": percentile(values, 0.95),
"max": max(values) if values else None,
"distinct_value_count": len(set(values)),
}
def extract(path: Path) -> dict[str, Any]:
pending: tuple[str, dict[str, Any]] | None = None
rows: list[dict[str, Any]] = []
for line in path.read_text(errors="replace").splitlines():
engine = ENGINE.search(line)
if engine:
pending = (
engine.group("clock")[:17],
{
"clock": engine.group("clock"),
"engine": int(engine.group("engine")),
"decode_ms": float(engine.group("decode_ms")),
"running": int(engine.group("running")),
"waiting": int(engine.group("waiting")),
},
)
continue
spec = SPEC.search(line)
if not spec or pending is None:
continue
clock, row = pending
if spec.group("clock")[:17] != clock:
pending = None
continue
row["mean_accept_length"] = float(spec.group("mean_accept"))
row["accept_pct"] = float(spec.group("accept_pct"))
rows.append(row)
pending = None
by_engine: dict[int, list[dict[str, Any]]] = defaultdict(list)
for row in rows:
by_engine[row["engine"]].append(row)
summaries: list[dict[str, Any]] = []
for engine, items in sorted(by_engine.items()):
summaries.append(
{
"engine": engine,
"window_count": len(items),
"accept_pct": describe([item["accept_pct"] for item in items]),
"mean_accept_length": describe([item["mean_accept_length"] for item in items]),
"decode_ms": describe([item["decode_ms"] for item in items]),
"running_requests": describe([float(item["running"]) for item in items]),
}
)
acceptance = [row["accept_pct"] for row in rows]
return {
"source": str(path),
"records": rows,
"per_engine": summaries,
"data_sanity": {
"accept_pct": {
"n": len(acceptance),
"min": min(acceptance) if acceptance else None,
"max": max(acceptance) if acceptance else None,
"distinct_value_count": len(set(acceptance)),
"within_0_100": all(0 <= item <= 100 for item in acceptance),
},
"invariants": {
"non_negative_decode_ms": all(row["decode_ms"] >= 0 for row in rows),
"non_negative_running": all(row["running"] >= 0 for row in rows),
"all_records_have_engine": all("engine" in row for row in rows),
},
},
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--engine-log", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
result = extract(args.engine_log)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print(json.dumps({"per_engine": result["per_engine"], "data_sanity": result["data_sanity"]}, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())

215
scripts/collectivespec/summarize_fixed_grid.py Normal file
View File

@@ -0,0 +1,215 @@
#!/usr/bin/env python3
"""Summarize a one-probe-per-K fresh-engine static-K grid conservatively.
Unlike the frontier helper, this reader treats ``probe_details.jsonl`` as the
source of completion and metric-integrity checks. A missing or partial probe
is reported as invalid rather than silently summarized.
"""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
from typing import Any
def load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def first(path: Path, pattern: str) -> Path | None:
values = sorted(path.glob(pattern))
return values[0] if values else None
def num(value: Any) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
value = float(value)
return value if math.isfinite(value) else None
def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]:
record: dict[str, Any] = {"k": k, "label": "NoSpec" if k == 0 else f"K={k}"}
store = root / "stores" / f"k{k}"
result_path = first(store, "*/trials/trial-*/result.json")
history_path = first(store, "*/trials/trial-*/probe_history.json")
details_path = first(store, "*/trials/trial-*/probe_details.jsonl")
record.update(
{
"result_path": str(result_path) if result_path else None,
"probe_history_path": str(history_path) if history_path else None,
"probe_details_path": str(details_path) if details_path else None,
}
)
failures: list[str] = []
if result_path is None or history_path is None or details_path is None:
failures.append("missing_artifact")
record.update({"valid": False, "integrity_failures": failures})
return record
result = load_json(result_path)
history = load_json(history_path)
detail_rows = [json.loads(line) for line in details_path.read_text(encoding="utf-8").splitlines() if line]
if result.get("status") != "completed":
failures.append(f"result_status={result.get('status')}")
if result.get("best_source") != "primary_search":
failures.append(f"best_source={result.get('best_source')}")
if result.get("completed_with_probe_failure"):
failures.append("completed_with_probe_failure")
if not isinstance(history, list) or len(history) != 1:
failures.append("history_not_exactly_one_probe")
if len(detail_rows) != 1:
failures.append("details_not_exactly_one_probe")
if failures:
record.update({"valid": False, "integrity_failures": failures})
return record
probe = history[0]
details = detail_rows[0]
outcomes = details.get("outcomes") if isinstance(details.get("outcomes"), list) else []
request_count = int(probe.get("request_count", -1))
if details.get("early_stopped") or probe.get("early_stopped"):
failures.append("early_stopped")
if len(outcomes) != request_count:
failures.append(f"outcome_count={len(outcomes)}_expected={request_count}")
successful = sum(bool(item.get("success")) for item in outcomes if isinstance(item, dict))
tpot_present = sum(item.get("tpot_ms") is not None for item in outcomes if isinstance(item, dict))
ttft_present = sum(item.get("ttft_ms") is not None for item in outcomes if isinstance(item, dict))
tokens_verified = all(
isinstance(item, dict)
and item.get("completion_tokens_source") == "usage"
and item.get("completion_tokens") == expected_tokens
and item.get("expected_completion_tokens") == expected_tokens
for item in outcomes
)
if successful != request_count:
failures.append(f"success_count={successful}_expected={request_count}")
if tpot_present != request_count:
failures.append(f"tpot_count={tpot_present}_expected={request_count}")
if ttft_present != request_count:
failures.append(f"ttft_count={ttft_present}_expected={request_count}")
if expected_tokens is not None and not tokens_verified:
failures.append("completion_tokens_not_all_usage_verified")
latency = probe.get("latency_summary") if isinstance(probe.get("latency_summary"), dict) else {}
tpot = latency.get("tpot_ms") if isinstance(latency.get("tpot_ms"), dict) else {}
record.update(
{
"valid": not failures,
"integrity_failures": failures,
"threshold": num(probe.get("threshold")),
"request_count": request_count,
"request_rate": num(probe.get("request_rate")),
"feasible": bool(probe.get("feasible")),
"pass_rate": num(probe.get("pass_rate")),
"success_count": successful,
"ttft_count": ttft_present,
"tpot_count": tpot_present,
"tpot_ms": {name: num(tpot.get(name)) for name in ("mean", "p50", "p90", "p95", "p99")},
}
)
return record
def data_sanity(records: list[dict[str, Any]]) -> dict[str, Any]:
valid = [record for record in records if record.get("valid")]
p95 = [num(record.get("tpot_ms", {}).get("p95")) for record in valid]
p95 = [value for value in p95 if value is not None]
pass_rates = [num(record.get("pass_rate")) for record in valid]
pass_rates = [value for value in pass_rates if value is not None]
request_rates = [num(record.get("request_rate")) for record in valid]
request_rates = [value for value in request_rates if value is not None]
return {
"n_configurations": len(records),
"n_valid": len(valid),
"all_valid": len(valid) == len(records),
"p95_tpot_ms": {
"n": len(p95),
"min": min(p95) if p95 else None,
"max": max(p95) if p95 else None,
"distinct_value_count": len(set(p95)),
"non_negative": all(value >= 0 for value in p95),
},
"pass_rate": {
"n": len(pass_rates),
"min": min(pass_rates) if pass_rates else None,
"max": max(pass_rates) if pass_rates else None,
"distinct_value_count": len(set(pass_rates)),
"within_0_1": all(0 <= value <= 1 for value in pass_rates),
},
"request_rate": {
"n": len(request_rates),
"min": min(request_rates) if request_rates else None,
"max": max(request_rates) if request_rates else None,
"distinct_value_count": len(set(request_rates)),
"non_negative": all(value >= 0 for value in request_rates),
},
"invariants": {
"one_record_per_k": len({record["k"] for record in records}) == len(records),
"all_full_completions": all(
record.get("success_count") == record.get("request_count")
for record in valid
),
"all_latency_values_present": all(
record.get("ttft_count") == record.get("request_count")
and record.get("tpot_count") == record.get("request_count")
for record in valid
),
},
}
def markdown(records: list[dict[str, Any]], sanity: dict[str, Any]) -> str:
lines = [
"# CollectiveSpec fresh-engine fixed-grid summary",
"",
"| configuration | valid | feasible | offered req/s | completed | pass rate | p95 TPOT (ms) | p99 TPOT (ms) | integrity failures |",
"|---|---:|---:|---:|---:|---:|---:|---:|---|",
]
for record in records:
tpot = record.get("tpot_ms") or {}
fmt = lambda value: "—" if value is None else f"{value:.6g}"
lines.append(
"| {label} | {valid} | {feasible} | {rate} | {completed}/{count} | {pass_rate} | {p95} | {p99} | {failures} |".format(
label=record["label"],
valid=record.get("valid", False),
feasible=record.get("feasible", "—"),
rate=fmt(record.get("request_rate")),
completed=record.get("success_count", "—"),
count=record.get("request_count", "—"),
pass_rate=fmt(record.get("pass_rate")),
p95=fmt(tpot.get("p95")),
p99=fmt(tpot.get("p99")),
failures=", ".join(record.get("integrity_failures") or []) or "—",
)
)
lines.extend(["", "## Data sanity", "", "```json", json.dumps(sanity, indent=2, sort_keys=True), "``", ""])
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--output-json", type=Path, required=True)
parser.add_argument("--output-md", type=Path, required=True)
args = parser.parse_args()
manifest = load_json(args.root / "manifest.json")
expected_tokens = manifest.get("completion_tokens_override")
if isinstance(expected_tokens, bool) or not isinstance(expected_tokens, int):
expected_tokens = None
records = [
read_one(args.root, int(k), expected_tokens)
for k in sorted(int(k) for k in manifest["k_values"])
]
sanity = data_sanity(records)
output = {"manifest": manifest, "records": records, "data_sanity": sanity}
args.output_json.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n", encoding="utf-8")
args.output_md.write_text(markdown(records, sanity), encoding="utf-8")
print(json.dumps(output, indent=2, sort_keys=True))
return 0 if sanity["all_valid"] else 2
if __name__ == "__main__":
raise SystemExit(main())

136
scripts/collectivespec/summarize_static_k_pilot.py Normal file
View File

@@ -0,0 +1,136 @@
#!/usr/bin/env python3
"""Summarize immutable artifacts from ``run_static_k_pilot.py``.
The output intentionally distinguishes an absent/failed result from an
SLO-infeasible probe, and includes a small data-sanity block for review.
"""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
from typing import Any
def number(value: Any) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
value = float(value)
return value if math.isfinite(value) else None
def first_result(store: Path) -> Path | None:
candidates = sorted(store.glob("*/trials/trial-*/result.json"))
return candidates[0] if candidates else None
def summarize_result(k: int, path: Path | None) -> dict[str, Any]:
record: dict[str, Any] = {"k": k, "label": "NoSpec" if k == 0 else f"K={k}"}
if path is None:
record.update({"status": "missing_result"})
return record
payload = json.loads(path.read_text())
probes = payload.get("probes") if isinstance(payload.get("probes"), list) else []
feasible = [probe for probe in probes if isinstance(probe, dict) and probe.get("feasible")]
best_payload = feasible[-1].get("payload", {}) if feasible else {}
latency = best_payload.get("latency_summary", {}) if isinstance(best_payload, dict) else {}
tpot = latency.get("tpot_ms", {}) if isinstance(latency, dict) else {}
record.update(
{
"status": str(payload.get("status", "unknown")),
"result_path": str(path),
"best_sampling_u": number(payload.get("best_sampling_u")),
"best_request_rate": number(payload.get("best_request_rate")),
"best_pass_rate": number(payload.get("best_pass_rate")),
"best_request_count": number(payload.get("best_request_count")),
"probe_count": len(probes),
"feasible_probe_count": len(feasible),
"probe_thresholds": [number(probe.get("threshold")) for probe in probes if isinstance(probe, dict)],
"best_tpot_ms": {
metric: number(tpot.get(metric)) for metric in ("mean", "p50", "p90", "p95", "p99")
},
}
)
return record
def sanity(records: list[dict[str, Any]]) -> dict[str, Any]:
completed = [r for r in records if r.get("status") == "completed"]
values = [r["best_sampling_u"] for r in completed if number(r.get("best_sampling_u")) is not None]
pass_rates = [r["best_pass_rate"] for r in completed if number(r.get("best_pass_rate")) is not None]
distinct = len(set(values))
return {
"n_configurations": len(records),
"n_completed": len(completed),
"sampling_u": {
"n": len(values),
"min": min(values) if values else None,
"max": max(values) if values else None,
"distinct_value_count": distinct,
"all_identical": len(values) > 1 and distinct == 1,
"non_negative": all(value >= 0 for value in values),
},
"pass_rate": {
"n": len(pass_rates),
"min": min(pass_rates) if pass_rates else None,
"max": max(pass_rates) if pass_rates else None,
"distinct_value_count": len(set(pass_rates)),
"within_0_1": all(0 <= value <= 1 for value in pass_rates),
},
"invariants": {
"one_result_per_k": len({r["k"] for r in records}) == len(records),
"all_best_request_rates_non_negative": all(
(value := number(r.get("best_request_rate"))) is None or value >= 0 for r in completed
),
},
}
def to_markdown(records: list[dict[str, Any]], checks: dict[str, Any]) -> str:
lines = [
"# CollectiveSpec static-K screening summary",
"",
"| configuration | status | max feasible sampling_u | request rate | pass rate | p95 TPOT (ms) | probes |",
"|---|---:|---:|---:|---:|---:|---:|",
]
for r in records:
tpot = r.get("best_tpot_ms") or {}
fmt = lambda value: "—" if value is None else f"{value:.6g}"
lines.append(
"| {label} | {status} | {u} | {rate} | {pass_rate} | {p95} | {probes} |".format(
label=r["label"],
status=r["status"],
u=fmt(r.get("best_sampling_u")),
rate=fmt(r.get("best_request_rate")),
pass_rate=fmt(r.get("best_pass_rate")),
p95=fmt(tpot.get("p95")),
probes=r.get("probe_count", "—"),
)
)
lines.extend(["", "## Data sanity", "", "```json", json.dumps(checks, indent=2, sort_keys=True), "``", ""])
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--output-json", type=Path, required=True)
parser.add_argument("--output-md", type=Path, required=True)
args = parser.parse_args()
manifest = json.loads((args.root / "manifest.json").read_text())
k_values = [int(k) for k in manifest["k_values"]]
records = [summarize_result(k, first_result(args.root / "stores" / f"k{k}")) for k in sorted(k_values)]
checks = sanity(records)
output = {"manifest": manifest, "records": records, "data_sanity": checks}
args.output_json.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n")
args.output_md.write_text(to_markdown(records, checks))
print(json.dumps(output, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())