mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-11-01 05:06:58 +08:00
perf(parser): add reproducible tokenizer benchmark
This commit is contained in:
Estrella Pan
114
backend/scripts/benchmark_tokenizer.py
Normal file
114
backend/scripts/benchmark_tokenizer.py
Normal file
@@ -0,0 +1,114 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Benchmark the generic release-title tokenizer.
|
||||
|
||||
Run from ``backend`` with::
|
||||
|
||||
uv run python scripts/benchmark_tokenizer.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_BACKEND_ROOT = Path(__file__).resolve().parent.parent
|
||||
_REPO_ROOT = _BACKEND_ROOT.parent
|
||||
_SRC_ROOT = _BACKEND_ROOT / "src"
|
||||
sys.path.insert(0, str(_SRC_ROOT))
|
||||
|
||||
from module.parser.analyser.tokenizer.benchmark import ( # noqa: E402
|
||||
load_corpus,
|
||||
render_text,
|
||||
run_benchmark,
|
||||
)
|
||||
|
||||
_DEFAULT_CORPUS = _SRC_ROOT / "test" / "fixtures" / "release_title_golden.json"
|
||||
|
||||
|
||||
def _positive_int(value: str) -> int:
|
||||
parsed = int(value)
|
||||
if parsed <= 0:
|
||||
raise argparse.ArgumentTypeError("must be a positive integer")
|
||||
return parsed
|
||||
|
||||
|
||||
def _non_negative_int(value: str) -> int:
|
||||
parsed = int(value)
|
||||
if parsed < 0:
|
||||
raise argparse.ArgumentTypeError("must be a non-negative integer")
|
||||
return parsed
|
||||
|
||||
|
||||
def _positive_float(value: str) -> float:
|
||||
parsed = float(value)
|
||||
if parsed <= 0:
|
||||
raise argparse.ArgumentTypeError("must be positive")
|
||||
return parsed
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Benchmark plain and traced parsing on the stable golden corpus. "
|
||||
"Timing never changes the exit status or acts as a CI threshold."
|
||||
)
|
||||
)
|
||||
parser.add_argument("--mode", choices=("plain", "trace", "both"), default="both")
|
||||
parser.add_argument("--suite", choices=("real", "stress", "all"), default="real")
|
||||
parser.add_argument("--corpus", type=Path, default=_DEFAULT_CORPUS)
|
||||
parser.add_argument("--warmup-rounds", type=_non_negative_int, default=20)
|
||||
parser.add_argument("--samples", type=_positive_int, default=7)
|
||||
parser.add_argument("--min-sample-ms", type=_positive_float, default=250.0)
|
||||
parser.add_argument(
|
||||
"--loops",
|
||||
type=_positive_int,
|
||||
help="fixed batch loop count; by default it is calibrated per group",
|
||||
)
|
||||
parser.add_argument("--format", choices=("text", "json"), default="text")
|
||||
parser.add_argument("--output", type=Path, help="write the report to this path")
|
||||
parser.add_argument(
|
||||
"--quiet", action="store_true", help="hide calibration progress on stderr"
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
modes = ("plain", "trace") if args.mode == "both" else (args.mode,)
|
||||
progress = None if args.quiet else lambda message: print(message, file=sys.stderr)
|
||||
|
||||
try:
|
||||
corpus = load_corpus(args.corpus)
|
||||
report = run_benchmark(
|
||||
corpus,
|
||||
repo_root=_REPO_ROOT,
|
||||
modes=modes,
|
||||
suite=args.suite,
|
||||
warmup_rounds=args.warmup_rounds,
|
||||
samples=args.samples,
|
||||
target_ns=int(args.min_sample_ms * 1_000_000),
|
||||
loops=args.loops,
|
||||
progress=progress,
|
||||
)
|
||||
except (OSError, RuntimeError, ValueError) as exc:
|
||||
print(f"benchmark failed: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
output = (
|
||||
json.dumps(report, ensure_ascii=False, indent=2)
|
||||
if args.format == "json"
|
||||
else render_text(report)
|
||||
)
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(output + "\n", encoding="utf-8")
|
||||
print(f"wrote benchmark report to {args.output}", file=sys.stderr)
|
||||
else:
|
||||
print(output)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
543
backend/src/module/parser/analyser/tokenizer/benchmark.py
Normal file
543
backend/src/module/parser/analyser/tokenizer/benchmark.py
Normal file
@@ -0,0 +1,543 @@
|
||||
"""Reusable benchmark helpers for the release-title tokenizer.
|
||||
|
||||
The benchmark deliberately uses only the standard library. It measures whole
|
||||
corpus batches so the timer overhead stays negligible and never treats a noisy
|
||||
wall-clock result as a correctness or CI pass/fail condition.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import gc
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import platform
|
||||
import statistics
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from collections import Counter
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .parser import parse_release_title, parse_release_title_with_trace
|
||||
|
||||
Parser = Callable[[str], object]
|
||||
Clock = Callable[[], int]
|
||||
Progress = Callable[[str], None]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class BenchmarkCase:
|
||||
"""One stable resource name and its semantic corpus category."""
|
||||
|
||||
id: str
|
||||
category: str
|
||||
raw: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class BenchmarkCorpus:
|
||||
"""Validated corpus metadata and cases."""
|
||||
|
||||
path: str
|
||||
schema_version: int
|
||||
sha256: str
|
||||
cases: tuple[BenchmarkCase, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class BenchmarkGroup:
|
||||
"""Cases timed together as one calibrated benchmark."""
|
||||
|
||||
name: str
|
||||
description: str
|
||||
cases: tuple[BenchmarkCase, ...]
|
||||
|
||||
|
||||
REAL_GROUPS: tuple[tuple[str, str, frozenset[str]], ...] = (
|
||||
("episode", "ordinary and season-aware episodes", frozenset({"episode"})),
|
||||
(
|
||||
"cardinality",
|
||||
"ranges, batches, and collections",
|
||||
frozenset({"range", "batch", "collection"}),
|
||||
),
|
||||
(
|
||||
"media",
|
||||
"movies, specials, and auxiliary media",
|
||||
frozenset({"movie", "special", "auxiliary"}),
|
||||
),
|
||||
(
|
||||
"ambiguous",
|
||||
"ambiguous, title-only, negative, and compatibility boundaries",
|
||||
frozenset({"ambiguity", "title-only", "negative", "legacy-boundary"}),
|
||||
),
|
||||
("technical", "technical-metadata-heavy titles", frozenset({"technical"})),
|
||||
)
|
||||
|
||||
|
||||
def load_corpus(path: Path) -> BenchmarkCorpus:
|
||||
"""Load and validate the release-title golden corpus."""
|
||||
try:
|
||||
raw_bytes = path.read_bytes()
|
||||
payload = json.loads(raw_bytes)
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"cannot load benchmark corpus {path}: {exc}") from exc
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("benchmark corpus root must be an object")
|
||||
schema_version = payload.get("schema_version")
|
||||
raw_cases = payload.get("cases")
|
||||
if not isinstance(schema_version, int):
|
||||
raise ValueError("benchmark corpus schema_version must be an integer")
|
||||
if not isinstance(raw_cases, list) or not raw_cases:
|
||||
raise ValueError("benchmark corpus cases must be a non-empty list")
|
||||
|
||||
cases: list[BenchmarkCase] = []
|
||||
for index, item in enumerate(raw_cases):
|
||||
if not isinstance(item, dict):
|
||||
raise ValueError(f"benchmark corpus case {index} must be an object")
|
||||
case_id = item.get("id")
|
||||
category = item.get("category")
|
||||
raw = item.get("raw")
|
||||
if not (
|
||||
isinstance(case_id, str)
|
||||
and case_id
|
||||
and isinstance(category, str)
|
||||
and category
|
||||
and isinstance(raw, str)
|
||||
and raw
|
||||
):
|
||||
raise ValueError(
|
||||
f"benchmark corpus case {index} needs non-empty id/category/raw"
|
||||
)
|
||||
cases.append(BenchmarkCase(case_id, category, raw))
|
||||
|
||||
ids = [case.id for case in cases]
|
||||
titles = [case.raw for case in cases]
|
||||
if len(ids) != len(set(ids)):
|
||||
raise ValueError("benchmark corpus case ids must be unique")
|
||||
if len(titles) != len(set(titles)):
|
||||
raise ValueError("benchmark corpus raw titles must be unique")
|
||||
|
||||
return BenchmarkCorpus(
|
||||
path=str(path.resolve()),
|
||||
schema_version=schema_version,
|
||||
sha256=hashlib.sha256(raw_bytes).hexdigest(),
|
||||
cases=tuple(cases),
|
||||
)
|
||||
|
||||
|
||||
def real_groups(corpus: BenchmarkCorpus) -> tuple[BenchmarkGroup, ...]:
|
||||
"""Build a complete real-corpus group plus stable semantic partitions."""
|
||||
groups = [
|
||||
BenchmarkGroup(
|
||||
name="all-real",
|
||||
description="all real golden-corpus resource names",
|
||||
cases=corpus.cases,
|
||||
)
|
||||
]
|
||||
covered: set[str] = set()
|
||||
for name, description, categories in REAL_GROUPS:
|
||||
cases = tuple(case for case in corpus.cases if case.category in categories)
|
||||
if cases:
|
||||
groups.append(BenchmarkGroup(name, description, cases))
|
||||
covered.update(case.category for case in cases)
|
||||
|
||||
all_categories = {case.category for case in corpus.cases}
|
||||
uncovered = all_categories - covered
|
||||
if uncovered:
|
||||
cases = tuple(case for case in corpus.cases if case.category in uncovered)
|
||||
groups.append(
|
||||
BenchmarkGroup(
|
||||
name="other",
|
||||
description="new corpus categories not in the stable partitions",
|
||||
cases=cases,
|
||||
)
|
||||
)
|
||||
return tuple(groups)
|
||||
|
||||
|
||||
def _technical_stress_title(tag_count: int) -> str:
|
||||
tags = ("1080p", "WEB-DL", "HEVC", "AAC")
|
||||
suffix = " ".join(f"[{tags[index % len(tags)]}]" for index in range(tag_count))
|
||||
return f"[Stress] Benchmark Anime S02E05 {suffix}"
|
||||
|
||||
|
||||
def stress_groups() -> tuple[BenchmarkGroup, ...]:
|
||||
"""Return fixed synthetic cases for observing parser scaling."""
|
||||
groups: list[BenchmarkGroup] = []
|
||||
for tag_count in (1, 4, 16, 64):
|
||||
case = BenchmarkCase(
|
||||
id=f"technical_tags_{tag_count}",
|
||||
category="stress",
|
||||
raw=_technical_stress_title(tag_count),
|
||||
)
|
||||
groups.append(
|
||||
BenchmarkGroup(
|
||||
name=f"stress-tags-{tag_count}",
|
||||
description=f"one title containing {tag_count} technical tags",
|
||||
cases=(case,),
|
||||
)
|
||||
)
|
||||
|
||||
conflicts = BenchmarkCase(
|
||||
id="conflicting_markers",
|
||||
category="stress",
|
||||
raw=(
|
||||
"[Stress] Benchmark Anime S02E05 E06 07 OVA08 OAD09 SP10 PV11 "
|
||||
"Movie 2024 [1080p WEB-DL HEVC AAC]"
|
||||
),
|
||||
)
|
||||
groups.append(
|
||||
BenchmarkGroup(
|
||||
name="stress-conflicts",
|
||||
description="one title containing overlapping structural/media markers",
|
||||
cases=(conflicts,),
|
||||
)
|
||||
)
|
||||
return tuple(groups)
|
||||
|
||||
|
||||
def validate_parser_equivalence(groups: Sequence[BenchmarkGroup]) -> None:
|
||||
"""Fail before timing if plain and diagnostic APIs disagree."""
|
||||
seen: set[str] = set()
|
||||
for group in groups:
|
||||
for case in group.cases:
|
||||
if case.raw in seen:
|
||||
continue
|
||||
seen.add(case.raw)
|
||||
plain = parse_release_title(case.raw)
|
||||
traced = parse_release_title_with_trace(case.raw)
|
||||
if traced.result != plain:
|
||||
raise ValueError(
|
||||
f"plain/trace result mismatch for benchmark case {case.id!r}"
|
||||
)
|
||||
|
||||
|
||||
def _execute_batch(parser: Parser, titles: Sequence[str], loops: int) -> object:
|
||||
last_result: object = None
|
||||
for _ in range(loops):
|
||||
for title in titles:
|
||||
last_result = parser(title)
|
||||
return last_result
|
||||
|
||||
|
||||
def warm_up(parser: Parser, titles: Sequence[str], rounds: int) -> None:
|
||||
"""Warm regex caches and interpreter paths outside the timed samples."""
|
||||
if rounds < 0:
|
||||
raise ValueError("warm-up rounds cannot be negative")
|
||||
_execute_batch(parser, titles, rounds)
|
||||
|
||||
|
||||
def calibrate_loops(
|
||||
parser: Parser,
|
||||
titles: Sequence[str],
|
||||
target_ns: int,
|
||||
*,
|
||||
clock: Clock = time.perf_counter_ns,
|
||||
maximum_loops: int = 1_048_576,
|
||||
) -> int:
|
||||
"""Find a batch loop count that reaches the requested sample duration."""
|
||||
if not titles:
|
||||
raise ValueError("cannot calibrate an empty benchmark group")
|
||||
if target_ns <= 0:
|
||||
raise ValueError("target sample duration must be positive")
|
||||
|
||||
loops = 1
|
||||
while True:
|
||||
gc.collect()
|
||||
started = clock()
|
||||
_execute_batch(parser, titles, loops)
|
||||
elapsed = clock() - started
|
||||
if elapsed >= target_ns:
|
||||
return loops
|
||||
if loops >= maximum_loops:
|
||||
raise RuntimeError("benchmark calibration exceeded maximum loop count")
|
||||
|
||||
multiplier = max(2, min(16, math.ceil(target_ns / max(elapsed, 1))))
|
||||
loops = min(maximum_loops, loops * multiplier)
|
||||
|
||||
|
||||
def run_samples(
|
||||
parser: Parser,
|
||||
titles: Sequence[str],
|
||||
*,
|
||||
loops: int,
|
||||
samples: int,
|
||||
clock: Clock = time.perf_counter_ns,
|
||||
) -> tuple[int, ...]:
|
||||
"""Measure calibrated whole-corpus samples with GC enabled."""
|
||||
if not titles:
|
||||
raise ValueError("cannot benchmark an empty title sequence")
|
||||
if loops <= 0 or samples <= 0:
|
||||
raise ValueError("loops and samples must be positive")
|
||||
|
||||
elapsed_samples: list[int] = []
|
||||
for _ in range(samples):
|
||||
gc.collect()
|
||||
started = clock()
|
||||
_execute_batch(parser, titles, loops)
|
||||
elapsed = clock() - started
|
||||
if elapsed < 0:
|
||||
raise RuntimeError("benchmark clock moved backwards")
|
||||
elapsed_samples.append(elapsed)
|
||||
return tuple(elapsed_samples)
|
||||
|
||||
|
||||
def _nearest_rank(values: Sequence[float], percentile: float) -> float:
|
||||
if not values:
|
||||
raise ValueError("cannot summarize empty benchmark samples")
|
||||
if not 0 < percentile <= 1:
|
||||
raise ValueError("percentile must be within (0, 1]")
|
||||
ordered = sorted(values)
|
||||
rank = max(1, math.ceil(percentile * len(ordered)))
|
||||
return ordered[rank - 1]
|
||||
|
||||
|
||||
def summarize_samples(
|
||||
elapsed_samples_ns: Sequence[int], operations_per_sample: int
|
||||
) -> dict[str, Any]:
|
||||
"""Summarize raw batch timings as stable per-title statistics."""
|
||||
if operations_per_sample <= 0:
|
||||
raise ValueError("operations_per_sample must be positive")
|
||||
if not elapsed_samples_ns:
|
||||
raise ValueError("elapsed_samples_ns cannot be empty")
|
||||
|
||||
per_title = [elapsed / operations_per_sample for elapsed in elapsed_samples_ns]
|
||||
median = statistics.median(per_title)
|
||||
mad = statistics.median(abs(value - median) for value in per_title)
|
||||
return {
|
||||
"sample_elapsed_ns": list(elapsed_samples_ns),
|
||||
"ns_per_title": {
|
||||
"min": min(per_title),
|
||||
"mean": statistics.fmean(per_title),
|
||||
"median": median,
|
||||
"p95": _nearest_rank(per_title, 0.95),
|
||||
"max": max(per_title),
|
||||
"mad": mad,
|
||||
"mad_percent": (mad / median * 100) if median else 0.0,
|
||||
},
|
||||
"titles_per_second": (1_000_000_000 / median) if median else None,
|
||||
}
|
||||
|
||||
|
||||
def benchmark_group(
|
||||
parser: Parser,
|
||||
group: BenchmarkGroup,
|
||||
*,
|
||||
warmup_rounds: int,
|
||||
samples: int,
|
||||
target_ns: int,
|
||||
loops: int | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Warm, calibrate, and measure one parser/group pair."""
|
||||
if loops is not None and loops <= 0:
|
||||
raise ValueError("fixed loops must be positive")
|
||||
titles = tuple(case.raw for case in group.cases)
|
||||
warm_up(parser, titles, warmup_rounds)
|
||||
calibrated_loops = (
|
||||
loops if loops is not None else calibrate_loops(parser, titles, target_ns)
|
||||
)
|
||||
elapsed = run_samples(
|
||||
parser,
|
||||
titles,
|
||||
loops=calibrated_loops,
|
||||
samples=samples,
|
||||
)
|
||||
operations = calibrated_loops * len(titles)
|
||||
return {
|
||||
"description": group.description,
|
||||
"case_count": len(titles),
|
||||
"loops": calibrated_loops,
|
||||
"operations_per_sample": operations,
|
||||
**summarize_samples(elapsed, operations),
|
||||
}
|
||||
|
||||
|
||||
def runtime_metadata(repo_root: Path) -> dict[str, Any]:
|
||||
"""Collect comparison context without making benchmark success depend on Git."""
|
||||
metadata: dict[str, Any] = {
|
||||
"python": platform.python_version(),
|
||||
"implementation": platform.python_implementation(),
|
||||
"executable": sys.executable,
|
||||
"platform": platform.platform(),
|
||||
"machine": platform.machine(),
|
||||
"processor": platform.processor() or None,
|
||||
"python_hash_seed": os.environ.get("PYTHONHASHSEED"),
|
||||
}
|
||||
try:
|
||||
commit = subprocess.run(
|
||||
["git", "rev-parse", "--short=12", "HEAD"],
|
||||
cwd=repo_root,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
).stdout.strip()
|
||||
dirty = bool(
|
||||
subprocess.run(
|
||||
["git", "status", "--porcelain"],
|
||||
cwd=repo_root,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
).stdout
|
||||
)
|
||||
metadata.update({"git_commit": commit, "git_dirty": dirty})
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
metadata.update({"git_commit": None, "git_dirty": None})
|
||||
return metadata
|
||||
|
||||
|
||||
def run_benchmark(
|
||||
corpus: BenchmarkCorpus,
|
||||
*,
|
||||
repo_root: Path,
|
||||
modes: Sequence[str],
|
||||
suite: str,
|
||||
warmup_rounds: int,
|
||||
samples: int,
|
||||
target_ns: int,
|
||||
loops: int | None = None,
|
||||
progress: Progress | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Run selected suites and return a JSON-serializable versioned report."""
|
||||
if suite not in {"real", "stress", "all"}:
|
||||
raise ValueError(f"unsupported benchmark suite: {suite}")
|
||||
parsers: dict[str, Parser] = {
|
||||
"plain": parse_release_title,
|
||||
"trace": parse_release_title_with_trace,
|
||||
}
|
||||
unknown_modes = set(modes) - parsers.keys()
|
||||
if not modes or unknown_modes:
|
||||
raise ValueError(f"unsupported benchmark modes: {sorted(unknown_modes)}")
|
||||
|
||||
groups: list[BenchmarkGroup] = []
|
||||
if suite in {"real", "all"}:
|
||||
groups.extend(real_groups(corpus))
|
||||
if suite in {"stress", "all"}:
|
||||
groups.extend(stress_groups())
|
||||
validate_parser_equivalence(groups)
|
||||
|
||||
results: dict[str, dict[str, Any]] = {}
|
||||
for mode in modes:
|
||||
mode_results: dict[str, Any] = {}
|
||||
for group in groups:
|
||||
if progress:
|
||||
progress(f"{mode}: {group.name} ({len(group.cases)} cases)")
|
||||
mode_results[group.name] = benchmark_group(
|
||||
parsers[mode],
|
||||
group,
|
||||
warmup_rounds=warmup_rounds,
|
||||
samples=samples,
|
||||
target_ns=target_ns,
|
||||
loops=loops,
|
||||
)
|
||||
results[mode] = mode_results
|
||||
|
||||
comparison: dict[str, Any] = {}
|
||||
if {"plain", "trace"}.issubset(results):
|
||||
for group in groups:
|
||||
plain_ns = results["plain"][group.name]["ns_per_title"]["median"]
|
||||
trace_ns = results["trace"][group.name]["ns_per_title"]["median"]
|
||||
comparison[group.name] = {
|
||||
"trace_overhead_ratio": trace_ns / plain_ns if plain_ns else None,
|
||||
"trace_overhead_percent": (
|
||||
(trace_ns - plain_ns) / plain_ns * 100 if plain_ns else None
|
||||
),
|
||||
}
|
||||
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"benchmark": "release-title-tokenizer",
|
||||
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||||
"corpus": {
|
||||
"path": corpus.path,
|
||||
"schema_version": corpus.schema_version,
|
||||
"case_count": len(corpus.cases),
|
||||
"sha256": corpus.sha256,
|
||||
"categories": dict(
|
||||
sorted(Counter(case.category for case in corpus.cases).items())
|
||||
),
|
||||
},
|
||||
"runtime": runtime_metadata(repo_root),
|
||||
"config": {
|
||||
"modes": list(modes),
|
||||
"suite": suite,
|
||||
"warmup_rounds": warmup_rounds,
|
||||
"samples": samples,
|
||||
"target_sample_ms": target_ns / 1_000_000,
|
||||
"fixed_loops": loops,
|
||||
"gc_enabled": gc.isenabled(),
|
||||
},
|
||||
"results": results,
|
||||
"comparison": comparison,
|
||||
}
|
||||
|
||||
|
||||
def render_text(report: dict[str, Any]) -> str:
|
||||
"""Render a compact human-readable table from a benchmark report."""
|
||||
runtime = report["runtime"]
|
||||
corpus = report["corpus"]
|
||||
config = report["config"]
|
||||
timing = (
|
||||
f"fixed loops {config['fixed_loops']}"
|
||||
if config["fixed_loops"] is not None
|
||||
else f"target {config['target_sample_ms']:g} ms"
|
||||
)
|
||||
lines = [
|
||||
"AutoBangumi release-title tokenizer benchmark",
|
||||
(
|
||||
f"Python {runtime['python']} ({runtime['implementation']}) | "
|
||||
f"{runtime['machine']} | commit {runtime.get('git_commit') or 'unknown'}"
|
||||
),
|
||||
(
|
||||
f"Corpus {corpus['case_count']} real cases | suite {config['suite']} | "
|
||||
f"warmup {config['warmup_rounds']} | samples {config['samples']} | "
|
||||
f"{timing}"
|
||||
),
|
||||
]
|
||||
if runtime.get("git_dirty"):
|
||||
lines.append("Warning: worktree is dirty; record this when comparing runs.")
|
||||
|
||||
for mode, groups in report["results"].items():
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
f"{mode.upper()}",
|
||||
(
|
||||
f"{'group':<22} {'cases':>5} {'loops':>7} "
|
||||
f"{'median us':>11} {'p95 us':>10} {'MAD %':>8} {'titles/s':>10}"
|
||||
),
|
||||
]
|
||||
)
|
||||
for name, result in groups.items():
|
||||
stats = result["ns_per_title"]
|
||||
lines.append(
|
||||
f"{name:<22} {result['case_count']:>5} {result['loops']:>7} "
|
||||
f"{stats['median'] / 1_000:>11.2f} "
|
||||
f"{stats['p95'] / 1_000:>10.2f} "
|
||||
f"{stats['mad_percent']:>8.2f} "
|
||||
f"{result['titles_per_second']:>10.0f}"
|
||||
)
|
||||
|
||||
if report["comparison"]:
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"TRACE OVERHEAD",
|
||||
f"{'group':<22} {'ratio':>9} {'overhead':>10}",
|
||||
]
|
||||
)
|
||||
for name, comparison in report["comparison"].items():
|
||||
lines.append(
|
||||
f"{name:<22} {comparison['trace_overhead_ratio']:>8.2f}x "
|
||||
f"{comparison['trace_overhead_percent']:>9.1f}%"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
173
backend/src/test/test_tokenizer_benchmark.py
Normal file
173
backend/src/test/test_tokenizer_benchmark.py
Normal file
@@ -0,0 +1,173 @@
|
||||
"""Fast correctness tests for the tokenizer benchmark harness.
|
||||
|
||||
These tests validate corpus handling and statistics only. They intentionally
|
||||
contain no wall-clock threshold and do not turn normal pytest runs into a
|
||||
performance gate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from module.parser.analyser.tokenizer.benchmark import (
|
||||
BenchmarkCase,
|
||||
BenchmarkGroup,
|
||||
benchmark_group,
|
||||
calibrate_loops,
|
||||
load_corpus,
|
||||
real_groups,
|
||||
run_benchmark,
|
||||
run_samples,
|
||||
stress_groups,
|
||||
summarize_samples,
|
||||
)
|
||||
|
||||
_BACKEND_ROOT = Path(__file__).parents[2]
|
||||
_REPO_ROOT = _BACKEND_ROOT.parent
|
||||
_CORPUS_PATH = Path(__file__).parent / "fixtures" / "release_title_golden.json"
|
||||
|
||||
|
||||
def test_load_corpus_and_real_group_partitions() -> None:
|
||||
corpus = load_corpus(_CORPUS_PATH)
|
||||
groups = {group.name: group for group in real_groups(corpus)}
|
||||
|
||||
assert corpus.schema_version == 1
|
||||
assert len(corpus.cases) == 71
|
||||
assert len(corpus.sha256) == 64
|
||||
assert {name: len(group.cases) for name, group in groups.items()} == {
|
||||
"all-real": 71,
|
||||
"episode": 21,
|
||||
"cardinality": 10,
|
||||
"media": 22,
|
||||
"ambiguous": 15,
|
||||
"technical": 3,
|
||||
}
|
||||
|
||||
|
||||
def test_load_corpus_rejects_duplicate_resource_names(tmp_path: Path) -> None:
|
||||
corpus_path = tmp_path / "duplicate.json"
|
||||
corpus_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"cases": [
|
||||
{"id": "first", "category": "episode", "raw": "Same"},
|
||||
{"id": "second", "category": "movie", "raw": "Same"},
|
||||
],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="raw titles must be unique"):
|
||||
load_corpus(corpus_path)
|
||||
|
||||
|
||||
def test_calibrate_loops_grows_until_target_duration() -> None:
|
||||
ticks = iter((0, 10, 0, 80, 0, 160))
|
||||
calls: list[str] = []
|
||||
|
||||
loops = calibrate_loops(
|
||||
lambda title: calls.append(title),
|
||||
("title",),
|
||||
100,
|
||||
clock=lambda: next(ticks),
|
||||
)
|
||||
|
||||
assert loops == 20
|
||||
assert len(calls) == 31
|
||||
|
||||
|
||||
def test_run_samples_normalizes_whole_batch_operations() -> None:
|
||||
ticks = iter((10, 110, 200, 500))
|
||||
calls: list[str] = []
|
||||
|
||||
elapsed = run_samples(
|
||||
lambda title: calls.append(title),
|
||||
("first", "second"),
|
||||
loops=2,
|
||||
samples=2,
|
||||
clock=lambda: next(ticks),
|
||||
)
|
||||
|
||||
assert elapsed == (100, 300)
|
||||
assert calls == ["first", "second"] * 4
|
||||
|
||||
|
||||
def test_benchmark_group_rejects_non_positive_fixed_loops() -> None:
|
||||
group = BenchmarkGroup(
|
||||
name="invalid",
|
||||
description="invalid fixed loop count",
|
||||
cases=(BenchmarkCase("case", "episode", "Anime - 01"),),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="fixed loops must be positive"):
|
||||
benchmark_group(
|
||||
lambda _title: None,
|
||||
group,
|
||||
warmup_rounds=0,
|
||||
samples=1,
|
||||
target_ns=1,
|
||||
loops=0,
|
||||
)
|
||||
|
||||
|
||||
def test_summarize_samples_uses_nearest_rank_p95() -> None:
|
||||
summary = summarize_samples((200, 400, 600, 800), operations_per_sample=2)
|
||||
|
||||
assert summary["sample_elapsed_ns"] == [200, 400, 600, 800]
|
||||
assert summary["ns_per_title"] == {
|
||||
"min": 100,
|
||||
"mean": 250,
|
||||
"median": 250.0,
|
||||
"p95": 400,
|
||||
"max": 400,
|
||||
"mad": 100.0,
|
||||
"mad_percent": 40.0,
|
||||
}
|
||||
assert summary["titles_per_second"] == 4_000_000
|
||||
|
||||
|
||||
def test_stress_groups_cover_scaling_and_conflicts() -> None:
|
||||
groups = stress_groups()
|
||||
|
||||
assert [group.name for group in groups] == [
|
||||
"stress-tags-1",
|
||||
"stress-tags-4",
|
||||
"stress-tags-16",
|
||||
"stress-tags-64",
|
||||
"stress-conflicts",
|
||||
]
|
||||
lengths = [len(group.cases[0].raw) for group in groups[:4]]
|
||||
assert lengths == sorted(lengths)
|
||||
assert len(set(lengths)) == 4
|
||||
|
||||
|
||||
def test_fixed_loop_report_is_json_serializable_without_speed_assertions() -> None:
|
||||
corpus = load_corpus(_CORPUS_PATH)
|
||||
|
||||
report = run_benchmark(
|
||||
corpus,
|
||||
repo_root=_REPO_ROOT,
|
||||
modes=("plain", "trace"),
|
||||
suite="stress",
|
||||
warmup_rounds=0,
|
||||
samples=1,
|
||||
target_ns=1,
|
||||
loops=1,
|
||||
)
|
||||
|
||||
assert report["schema_version"] == 1
|
||||
assert report["benchmark"] == "release-title-tokenizer"
|
||||
assert set(report["results"]) == {"plain", "trace"}
|
||||
assert set(report["comparison"]) == {
|
||||
"stress-tags-1",
|
||||
"stress-tags-4",
|
||||
"stress-tags-16",
|
||||
"stress-tags-64",
|
||||
"stress-conflicts",
|
||||
}
|
||||
json.dumps(report)
|
||||
@@ -164,6 +164,36 @@ Claims 保证 `S02E05`、`OVA01`、range 不会被拆成互相矛盾的半份结
|
||||
技术标签和未知标签，保证确定性与不抛异常。
|
||||
6. 下游测试验证 range/PV 不进入单集去重，OVA 不污染 Offset，Movie 的 LLM/确定性投影一致。
|
||||
|
||||
## 性能基准
|
||||
|
||||
Tokenizer benchmark 是纯标准库 CLI，不引入 `pytest-benchmark`，也不在普通测试或
|
||||
CI 中设置机器相关的速度阈值：
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
|
||||
# 71 条真实语料；分别测 plain、Trace 和 Trace 开销
|
||||
uv run python scripts/benchmark_tokenizer.py
|
||||
|
||||
# 加入固定 stress 语料，并保存可复算的原始样本
|
||||
uv run python scripts/benchmark_tokenizer.py \
|
||||
--suite all --format json --output tokenizer-benchmark.json
|
||||
|
||||
# 只检查 benchmark 流程，不作为有效性能数据
|
||||
uv run python scripts/benchmark_tokenizer.py \
|
||||
--suite all --warmup-rounds 1 --samples 1 --loops 1
|
||||
```
|
||||
|
||||
默认先 warm up 20 轮，再为每个分组自动校准 loops，使每个 sample 至少持续 250ms，
|
||||
最后采集 7 个 sample。主指标是每个资源名的 median，辅助报告 p95、MAD、吞吐量和
|
||||
完整 `sample_elapsed_ns`。真实语料划分为 episode、cardinality、media、ambiguous、
|
||||
technical；stress 套件单独观察 1/4/16/64 个技术标签以及冲突 marker，避免合成输入
|
||||
污染真实语料总分。
|
||||
|
||||
比较两次结果时至少要核对 Python、平台、commit、worktree dirty 状态和 corpus
|
||||
SHA-256。Plain 与 Trace 的结果一致性在计时区外验证；语料加载、GC collect、报告
|
||||
序列化和输出均不计入样本时间。
|
||||
|
||||
## 迁移与回滚
|
||||
|
||||
先以现有 parser 结果作为 characterization 基线，实现 Candidate/Resolver 后跑全量
|
||||
|
||||
Reference in New Issue
Block a user