Skip to content

Add repeatable local audio performance baseline #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
FANNG1 merged 3 commits into from
Jul 22, 2026
Merged

Add repeatable local audio performance baseline #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
098e59f
Add repeatable local audio baseline
Jul 16, 2026
c3f0eaf
Preserve failed baseline run summaries
Jul 16, 2026
fbc9f2b
Translate benchmark README to Chinese
Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 24 additions & 18 deletions benchmark/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,32 +1,39 @@
# Daft audio stability benchmark
# Daft 音频稳定性基准测试

For complete local and multi-node Ray instructions, sizing guidance, metrics,
and extension points, see the [audio benchmark guide](../docs/audio-benchmark.md).
完整的本地与多节点 Ray 操作说明、容量规划、指标口径和扩展点，见
[音频基准测试指南](../docs/audio-benchmark.md)。

This package exercises the real audio path without calling a real LLM:
本 package 在不调用真实 LLM 的前提下跑通完整的音频链路：

```
MinIO -> Daft/Ray download -> SenseVoice ASR -> HTTP mock LLM -> Lance blob v2
MinIO -> Daft/Ray 下载 -> SenseVoice ASR -> HTTP mock LLM -> Lance blob v2
```

It always uses the MinIO credentials and endpoint from the repository `.env`.
It never starts, stops, or deletes the configured MinIO service. Each run writes
under its own `s3://<bucket>/<prefix>/<run-id>/` prefix.
它始终使用仓库 `.env` 中的 MinIO 凭据与 endpoint，绝不会启动、停止或删除已配置的
MinIO 服务。每次运行写入各自独立的 `s3://<bucket>/<prefix>/<run-id>/` 前缀下。

## Local smoke test
## 本地冒烟测试

Prerequisites: the configured MinIO is running, the SenseVoice and FSMN VAD
models are cached or downloadable, and `data/audio/` contains seed audio.
前置条件：配置的 MinIO 已在运行，SenseVoice 与 FSMN VAD 模型已缓存或可下载，
且 `data/audio/` 中有种子音频。

```bash
uv run python -m benchmark.audio local-smoke --count 4
```

Results are written to `.benchmarks/<run-id>/`, including `metadata.json`,
`summary.json`, `resources.csv`, `report.json`, `report.md`, the Daft physical
plan, and Daft event logs.
如果需要可重复的本地性能 baseline —— 生成 50 条固定 60 秒音频，预热一次模型/缓存，
再用同一份 manifest 跑两轮：

## Separate commands
```bash
uv run python -m benchmark.audio local-baseline
```

比较结果写入 `.benchmarks/<run-id>/baseline-summary.{json,md}`。

运行产物写入 `.benchmarks/<run-id>/`，包含 `metadata.json`、`summary.json`、
`resources.csv`、`report.json`、`report.md`、Daft 物理执行计划以及 Daft 事件日志。

## 分步命令

```bash
uv run python -m benchmark.audio generate --profile smoke --count 4
Expand All @@ -47,6 +54,5 @@ uv run python -m benchmark.audio submit \
--ray-address auto --max-minutes 60 --wait
```

Use `generate --profile fixed` for controlled fixed-duration data and
`--profile mixed` for a 70% short / 25% medium / 5% long distribution. Data
generation happens before the timed benchmark.
`generate --profile fixed` 用于生成受控的固定时长数据，`--profile mixed` 生成
70% 短 / 25% 中 / 5% 长 的分布。数据生成发生在计时的基准测试之前。
227 changes: 221 additions & 6 deletions benchmark/audio/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,9 @@
from .config import BenchmarkConfig, DEFAULT_RUN_ROOT, new_run_id


def _add_run_arguments(parser: argparse.ArgumentParser, *, ray_default: str = "local") -> None:
def _add_run_arguments(
parser: argparse.ArgumentParser, *, ray_default: str = "local"
) -> None:
parser.add_argument("--run-id", default=None)
parser.add_argument("--manifest", required=True)
parser.add_argument("--lance-uri", required=True)
Expand Down Expand Up @@ -71,7 +73,9 @@ def _cmd_generate(args) -> None:
source_dir=Path(args.source_dir) if args.source_dir else None,
seed=args.seed,
)
print(json.dumps({"run_id": run_id, **asdict(result)}, ensure_ascii=False, indent=2))
print(
json.dumps({"run_id": run_id, **asdict(result)}, ensure_ascii=False, indent=2)
)


def _cmd_run(args) -> None:
Expand All @@ -87,7 +91,11 @@ def _cmd_run(args) -> None:
raise
else:
report = build_report(cfg.output_dir)
print(json.dumps({"summary": summary, "report": report}, ensure_ascii=False, indent=2))
print(
json.dumps(
{"summary": summary, "report": report}, ensure_ascii=False, indent=2
)
)


def _cmd_submit(args) -> None:
Expand Down Expand Up @@ -154,7 +162,9 @@ def _cmd_local_smoke(args) -> None:
raise
report = build_report(cfg.output_dir)
if summary.get("rows") != dataset.rows:
raise RuntimeError(f"row count mismatch: manifest={dataset.rows}, lance={summary.get('rows')}")
raise RuntimeError(
f"row count mismatch: manifest={dataset.rows}, lance={summary.get('rows')}"
)
if summary.get("unique_doc_ids") != dataset.rows:
raise RuntimeError("output doc_id values are not unique")
print(json.dumps(report, ensure_ascii=False, indent=2))
Expand All @@ -167,6 +177,178 @@ def _cmd_local_smoke(args) -> None:
ray.shutdown()


def _cmd_local_baseline(args) -> None:
import ray

from .data import generate_dataset
from .mock_llm import start_in_thread
from .pipeline import run_pipeline
from .report import build_baseline_report, build_report
from .storage import preflight

if args.repeats < 2:
raise ValueError("--repeats must be at least 2")
standard_baseline = (
40 <= args.count <= 60
and args.duration_s == 60.0
and 0 < args.max_minutes <= 45.0
)
if not standard_baseline:
print(
"[warning] non-standard diagnostic baseline: the comparable baseline uses "
"40-60 files, 60s duration, and a per-repeat limit no greater than 45 minutes"
)
base_run_id = args.run_id or f"baseline-{new_run_id()}"
base_dir = DEFAULT_RUN_ROOT / base_run_id
print(f"[preflight] checking configured MinIO bucket {args.bucket}")
print(json.dumps(preflight(args.bucket), ensure_ascii=False))
print(
f"[generate] creating {args.count} fixed-duration objects "
f"({args.duration_s:.0f}s each)"
)
dataset = generate_dataset(
run_id=base_run_id,
bucket=args.bucket,
prefix=args.prefix,
profile="fixed",
count=args.count,
fixed_duration_s=args.duration_s,
source_dir=Path(args.source_dir) if args.source_dir else None,
seed=args.seed,
)
warmup_dataset = None
if not args.skip_warmup:
print("[generate] creating one warm-up object")
warmup_dataset = generate_dataset(
run_id=f"{base_run_id}-warmup",
bucket=args.bucket,
prefix=args.prefix,
profile="smoke",
count=1,
source_dir=Path(args.source_dir) if args.source_dir else None,
seed=args.seed,
)
server, thread = start_in_thread(port=0, profile=args.mock_profile)
port = server.server_address[1]
reports = []
repeat_error: Exception | None = None
try:
if warmup_dataset is not None:
warmup_cfg = BenchmarkConfig(
run_id=f"{base_run_id}-warmup",
manifest_uri=warmup_dataset.manifest_uri,
lance_uri=(
f"s3://{args.bucket}/{args.prefix.strip('/')}/{base_run_id}/warmup.lance"
),
mock_url=f"http://127.0.0.1:{port}",
ray_address="local",
max_minutes=min(args.max_minutes, 15.0),
asr_actor_cpus=args.asr_actor_cpus,
asr_actor_concurrency=args.asr_actor_concurrency,
asr_batch_size=args.asr_batch_size,
llm_concurrency=args.llm_concurrency,
llm_timeout_s=15.0,
sample_interval_s=args.sample_interval_s,
num_partitions=1,
run_dir=str(base_dir / "warmup"),
)
print(f"[warmup] {warmup_dataset.manifest_uri} -> {warmup_cfg.lance_uri}")
warmup_summary = run_pipeline(warmup_cfg)
build_report(warmup_cfg.output_dir)
if warmup_summary.get("rows") != 1 or warmup_summary.get(
"status_counts"
) != {"ok": 1}:
raise RuntimeError(f"warm-up validation failed: {warmup_summary}")
for repeat in range(1, args.repeats + 1):
repeat_name = f"repeat-{repeat:02d}"
repeat_run_id = f"{base_run_id}-{repeat_name}"
lance_uri = (
f"s3://{args.bucket}/{args.prefix.strip('/')}/{base_run_id}/"
f"output-{repeat_name}.lance"
)
cfg = BenchmarkConfig(
run_id=repeat_run_id,
manifest_uri=dataset.manifest_uri,
lance_uri=lance_uri,
mock_url=f"http://127.0.0.1:{port}",
ray_address="local",
max_minutes=args.max_minutes,
asr_actor_cpus=args.asr_actor_cpus,
asr_actor_concurrency=args.asr_actor_concurrency,
asr_batch_size=args.asr_batch_size,
llm_concurrency=args.llm_concurrency,
llm_timeout_s=15.0,
sample_interval_s=args.sample_interval_s,
num_partitions=args.num_partitions,
run_dir=str(base_dir / repeat_name),
)
print(
f"[run {repeat}/{args.repeats}] {dataset.manifest_uri} -> {lance_uri}"
)
try:
summary = run_pipeline(cfg)
except Exception as exc:
if (cfg.output_dir / "summary.json").exists():
reports.append(build_report(cfg.output_dir))
else:
reports.append(
{
"run_id": repeat_run_id,
"state": "failed",
"summary": {"elapsed_s": 0.0, "error": repr(exc)},
}
)
repeat_error = exc
break
report = build_report(cfg.output_dir)
if (
summary.get("rows") != dataset.rows
or summary.get("unique_doc_ids") != dataset.rows
):
repeat_error = RuntimeError(
f"repeat {repeat} row validation failed: expected={dataset.rows}, "
f"rows={summary.get('rows')}, unique={summary.get('unique_doc_ids')}"
)
reports.append(report)
if repeat_error is not None:
break
comparison = build_baseline_report(
base_dir,
reports,
{
"run_id": base_run_id,
"manifest_uri": dataset.manifest_uri,
"rows": dataset.rows,
"duration_s": args.duration_s,
"total_bytes": dataset.total_bytes,
"total_audio_seconds": dataset.total_audio_seconds,
"planned_repeats": args.repeats,
"seed": args.seed,
"source_dir": args.source_dir,
"mock_profile": args.mock_profile,
"warmup_enabled": not args.skip_warmup,
"standard_baseline": standard_baseline,
"max_minutes_per_repeat": args.max_minutes,
"asr_actor_cpus": args.asr_actor_cpus,
"asr_actor_concurrency": args.asr_actor_concurrency,
"asr_batch_size": args.asr_batch_size,
"llm_concurrency": args.llm_concurrency,
"sample_interval_s": args.sample_interval_s,
"num_partitions": args.num_partitions,
},
)
print(json.dumps(comparison, ensure_ascii=False, indent=2))
print(f"[summary] comparison: {base_dir / 'baseline-summary.md'}")
if repeat_error is not None:
raise repeat_error
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
if ray.is_initialized():
ray.shutdown()


def _cmd_serve_mock(args) -> None:
from .mock_llm import serve

Expand All @@ -181,7 +363,9 @@ def build_parser() -> argparse.ArgumentParser:
generate.add_argument("--run-id", default=None)
generate.add_argument("--bucket", default="benchmark")
generate.add_argument("--prefix", default="audio")
generate.add_argument("--profile", choices=("smoke", "fixed", "mixed"), default="smoke")
generate.add_argument(
"--profile", choices=("smoke", "fixed", "mixed"), default="smoke"
)
generate.add_argument("--count", type=int, default=4)
generate.add_argument("--duration-s", type=float, default=30.0)
generate.add_argument("--source-dir", default=None)
Expand All @@ -208,7 +392,9 @@ def build_parser() -> argparse.ArgumentParser:
report.add_argument("--run-dir", required=True)
report.set_defaults(func=_cmd_report)

smoke = commands.add_parser("local-smoke", help="run a real local end-to-end smoke test")
smoke = commands.add_parser(
"local-smoke", help="run a real local end-to-end smoke test"
)
smoke.add_argument("--run-id", default=None)
smoke.add_argument("--bucket", default="benchmark")
smoke.add_argument("--prefix", default="audio")
Expand All @@ -217,6 +403,35 @@ def build_parser() -> argparse.ArgumentParser:
smoke.add_argument("--max-minutes", type=float, default=15.0)
smoke.add_argument("--asr-actor-cpus", type=float, default=4.0)
smoke.set_defaults(func=_cmd_local_smoke)

baseline = commands.add_parser(
"local-baseline",
help="run repeated fixed-duration local performance baselines",
)
baseline.add_argument("--run-id", default=None)
baseline.add_argument("--bucket", default="benchmark")
baseline.add_argument("--prefix", default="audio")
baseline.add_argument("--count", type=int, default=50)
baseline.add_argument("--duration-s", type=float, default=60.0)
baseline.add_argument("--repeats", type=int, default=2)
baseline.add_argument("--source-dir", default=None)
baseline.add_argument("--seed", type=int, default=20260715)
baseline.add_argument(
"--mock-profile", choices=("fast", "standard"), default="fast"
)
baseline.add_argument(
"--skip-warmup",
action="store_true",
help="skip the one-row model/cache warm-up before measured repeats",
)
baseline.add_argument("--max-minutes", type=float, default=45.0)
baseline.add_argument("--asr-actor-cpus", type=float, default=4.0)
baseline.add_argument("--asr-actor-concurrency", type=int, default=1)
baseline.add_argument("--asr-batch-size", type=int, default=1)
baseline.add_argument("--llm-concurrency", type=int, default=8)
baseline.add_argument("--sample-interval-s", type=float, default=5.0)
baseline.add_argument("--num-partitions", type=int, default=None)
baseline.set_defaults(func=_cmd_local_baseline)
return parser


Expand Down
Loading
Loading