From 42a98bd7030913ec6573ca80ad7b6a90c2c9db6c Mon Sep 17 00:00:00 2001 From: Vladimir Bryzgalov Date: Sat, 15 Aug 2026 23:01:55 +0500 Subject: [PATCH] =?UTF-8?q?=D0=AD=D0=BD=D0=B4=D0=BF=D0=BE=D0=B8=D0=BD?= =?UTF-8?q?=D1=82=20=D0=BF=D0=BE=D0=B4=D0=B1=D0=BE=D1=80=D0=B0=20=D1=87?= =?UTF-8?q?=D0=B8=D1=81=D0=BB=D0=B0=20=D0=BF=D0=BE=D1=82=D0=BE=D0=BA=D0=BE?= =?UTF-8?q?=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Масштабирование ONNX зависит от процессора: замеры на Apple M4 не переносятся на Ryzen, а подбирать конфигурацию перезапусками мучительно. Теперь /v1/benchmark гоняет минуту записи на 1, 2, 4, 8 и 16 потоках и говорит, что поставить в config.toml. Co-Authored-By: Claude Opus 5 (1M context) --- app/bench.py | 63 +++++++++++++++++++++++++++++++++++++++ app/main.py | 25 ++++++++++++++++ app/version.py | 2 +- tests/test_concurrency.py | 23 ++++++++++++++ 4 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 app/bench.py diff --git a/app/bench.py b/app/bench.py new file mode 100644 index 0000000..af7e0bc --- /dev/null +++ b/app/bench.py @@ -0,0 +1,63 @@ +"""Подбор оптимального числа потоков на конкретной машине. + +Масштабирование ONNX сильно зависит от процессора: на Apple M4 оптимум +оказался в районе 4 потоков, на Ryzen картина другая. Гадать бесполезно, +поэтому сервис умеет померить сам. +""" +import time +from pathlib import Path + +import numpy as np + +from app.pipeline import SAMPLE_RATE, Pipeline + +__all__ = ["run_benchmark"] + +# Больше 16 потоков смысла мерить нет: везде, где замеряли, деградация +# начиналась гораздо раньше. +THREAD_OPTIONS = (1, 2, 4, 8, 16) + + +def run_benchmark(samples: np.ndarray, models_dir: Path, base_dir: Path, + replacements_path: Path, seconds: float = 60.0, + speakers: int = 2) -> dict: + """Гоняет один и тот же фрагмент на разном числе потоков.""" + clip = samples[: int(seconds * SAMPLE_RATE)] + duration = len(clip) / SAMPLE_RATE + rows = [] + + for threads in THREAD_OPTIONS: + pipe = Pipeline(models_dir=models_dir, threads=threads, + replacements_path=replacements_path, base_dir=base_dir) + pipe.warmup() + + t0 = time.time() + pipe._diarizer(speakers).process(clip) + diar = time.time() - t0 + + t0 = time.time() + pipe._asr.recognize(clip[: int(30 * SAMPLE_RATE)], sample_rate=SAMPLE_RATE) + asr = time.time() - t0 + + rows.append({ + "threads": threads, + "diarization_sec": round(diar, 2), + "diarization_x": round(duration / diar, 1) if diar else 0, + "asr_sec": round(asr, 2), + "asr_x": round(min(30.0, duration) / asr, 1) if asr else 0, + }) + + best_diar = max(rows, key=lambda r: r["diarization_x"]) + best_asr = max(rows, key=lambda r: r["asr_x"]) + return { + "clip_sec": round(duration, 1), + "results": rows, + "best_for_diarization": best_diar["threads"], + "best_for_asr": best_asr["threads"], + "recommended_threads": best_diar["threads"], + "hint": ( + f"Поставьте threads = {best_diar['threads']} в config.toml. " + "Разделение говорящих занимает большую часть времени, поэтому " + "ориентируемся на него." + ), + } diff --git a/app/main.py b/app/main.py index e2e8113..908892c 100644 --- a/app/main.py +++ b/app/main.py @@ -229,6 +229,31 @@ async def create_job( "queue_position": store.queue_position(job_id)} +@app.post("/v1/benchmark", dependencies=[Depends(guard)], summary="Подобрать число потоков", + description="Прогоняет минуту из присланной записи на разном числе потоков " + "и показывает, где быстрее. Занимает пару минут и на это время " + "нагружает процессор.") +async def benchmark(file: UploadFile = File(...)) -> dict: + from app.bench import run_benchmark + from app.pipeline import read_wav + + if not _state["ready"]: + raise HTTPException(status_code=503, detail=_state["error"] or "сервис ещё не готов") + + with tempfile.TemporaryDirectory() as tmp: + src = Path(tmp) / (file.filename or "audio") + with src.open("wb") as out: + while chunk := await file.read(1 << 20): + out.write(chunk) + wav = Path(tmp) / "audio.wav" + to_wav16k(src, wav, pipeline.ffmpeg) + samples = read_wav(wav) + + return run_benchmark(samples, models_dir=settings.models_dir, + base_dir=settings.base_dir, + replacements_path=settings.replacements_path) + + @app.get("/v1/jobs/{job_id}", dependencies=[Depends(guard)], summary="Забрать результат", description="Пока задача не готова, возвращает статус и место в очереди. " "Готовый результат содержит реплики с говорящими и тайм-кодами, " diff --git a/app/version.py b/app/version.py index 493f741..260c070 100644 --- a/app/version.py +++ b/app/version.py @@ -1 +1 @@ -__version__ = "0.3.0" +__version__ = "0.3.1" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 97cab63..779f6af 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -135,3 +135,26 @@ class TestLazyWarmup: assert "warmup" not in str(exc) except Exception: pass + + +class TestBenchmarkEndpoint: + def test_benchmark_requires_token(self, tmp_path, monkeypatch): + import sys + import types + + from fastapi.testclient import TestClient + + config = tmp_path / "config.toml" + config.write_text('[security]\ntoken="t"\nallow_ips=""\n', encoding="utf-8") + monkeypatch.setenv("TALKSCORE_ASR_HOME", str(tmp_path)) + for name in ("sherpa_onnx", "onnx_asr", "onnxruntime"): + monkeypatch.setitem(sys.modules, name, types.ModuleType(name)) + for mod in [m for m in list(sys.modules) if m.startswith("app.")]: + del sys.modules[mod] + import app.config as cfg + monkeypatch.setattr(cfg, "BASE_DIR", tmp_path) + import app.main as main + main._worker_stop.set() + + with TestClient(main.app) as c: + assert c.post("/v1/benchmark", files={"file": ("a.wav", b"x")}).status_code == 401