diff --git a/eval/.gitignore b/eval/.gitignore new file mode 100644 index 0000000..ae23926 --- /dev/null +++ b/eval/.gitignore @@ -0,0 +1,14 @@ +# 数据集、模型权重、结果、缓存全部不入库（体积大） +# 注意：锚定到 eval/ 根（前导 /），否则会误伤 asr_eval/datasets/ 源码包 +/datasets/ +/models/ +/results/ +*.wav +*.mp3 +*.flac +*.aiff +__pycache__/ +*.egg-info/ +.venv/ +venv/ +.pytest_cache/ diff --git a/eval/README.md b/eval/README.md new file mode 100644 index 0000000..25e76e7 --- /dev/null +++ b/eval/README.md @@ -0,0 +1,63 @@ +# dudu ASR 模型评估框架 + +横向对比**云端 gummy**（DashScope `gummy-realtime-v1`）与**本地开源模型**（faster-whisper / +SenseVoice / Paraformer），重点覆盖**中英混说（code-switching）**，产出三类指标： + +| 维度 | 指标 | +|---|---| +| 准确率 | **MER**（中英混合错误率，headline）/ CER（字级）/ WER（词级） | +| 速度 | 首包延迟、定稿延迟、RTF（处理时长 / 音频时长） | +| 资源 | 本地：峰值内存 / 平均 CPU / 模型大小；云端：成本(¥/分钟) | + +## 安装 + +```bash +cd eval +python -m venv .venv && source .venv/bin/activate +pip install -e . # 核心管线（含云端 gummy 评估能力） +pip install -e '.[whisper]' # 加 faster-whisper 本地引擎 +pip install -e '.[funasr]' # 加 SenseVoice / Paraformer（含 torch） +pip install -e '.[all]' # 全部本地引擎 +``` + +## 跑评估 + +```bash +# 冒烟：每个数据集取 5 条，只跑 gummy + whisper-small +rbw get dashscope-api-key | python -m asr_eval run --sample-limit 5 --engines gummy,whisper-small + +# 只验证 gummy 连通 +rbw get dashscope-api-key | python -m asr_eval run --engines gummy --datasets custom --sample-limit 3 + +# 全量（放开 sample_limit、在 config.yaml 里 enable 想跑的引擎/数据集） +rbw get dashscope-api-key | python -m asr_eval run --config config.yaml +``` + +> gummy 的 API key 优先读环境变量 `DASHSCOPE_API_KEY`，否则读 stdin 首行—— +> 遵全局规则用 `rbw get dashscope-api-key | ...` 管道传入，不落盘、不写 env 文件。 + +报告写到 `results/<时间戳>/`：`report.md`（终端友好）、`report.html`（表格）、 +`report.json`（机器可读）、`results.jsonl`（每条样本明细，便于 debug 个案）。 + +## 配置（`config.yaml`） + +- **数据集**：公共集统一 `type: hf`，差异全在配置（`hf_id`/`split`/`text_field`/`lang`）—— + 广覆盖 / 换数据集只改配置不动代码。预置 ASCEND（中英混说）、AISHELL-1（纯中）、LibriSpeech（纯英）。 +- **自定义集**：`type: custom` 指向 JSONL manifest，把自录/业务音频按 `manifests/custom.example.jsonl` + 格式丢进来即可与公共集同管线评估。 +- **引擎**：`enabled` 开关；本地引擎可选 `device`（cpu / 苹果芯片可试 mps）、whisper 的 `compute_type`。 + +## 设计要点 + +- **gummy 协议**对照后端 `server/internal/asr/gummy.go` 移植（run-task/finish-task、`transcription` + 与 `sentence` 双字段、`sentence_end` 的 bool/字符串兼容、3200B/16k/pcm 分帧）。 +- **错误率语料级聚合**：累加 S/D/I/N 再求率，不对每条样本求率再平均（短句不会被放大）。 +- **评分前归一化**：去标点、英文小写、繁→简、全/半角统一，否则中英混说分数失真。 +- **公平性声明**：流式(gummy) vs 离线(本地)延迟口径不同（gummy RTF 含推流倍速与网络）； + 资源仅本地引擎测量，含 Python 进程基底——报告会标注硬件环境。 + +## 已知约束 + +- ASCEND 在 HuggingFace 开放；AISHELL/LibriSpeech 首次会下载（数 GB，已 gitignore）。 +- SEAME（更权威的中英混说集）需 LDC 付费授权，默认不含；有授权可加一个 hf/custom 适配条目。 +- 本地模型权重大（whisper-large ≈1.5GB、SenseVoice ≈900MB），按需 enable。 diff --git a/eval/asr_eval/__init__.py b/eval/asr_eval/__init__.py new file mode 100644 index 0000000..634401b --- /dev/null +++ b/eval/asr_eval/__init__.py @@ -0,0 +1,7 @@ +"""dudu ASR 模型评估框架。 + +云端 gummy ↔ 本地开源模型横向对比，覆盖中英混说，产出准确率/速度/资源三类指标。 +入口：python -m asr_eval run --config config.yaml +""" + +__version__ = "0.1.0" diff --git a/eval/asr_eval/__main__.py b/eval/asr_eval/__main__.py new file mode 100644 index 0000000..be9b08a --- /dev/null +++ b/eval/asr_eval/__main__.py @@ -0,0 +1,4 @@ +from asr_eval.cli import main + +if __name__ == "__main__": + main() diff --git a/eval/asr_eval/audio.py b/eval/asr_eval/audio.py new file mode 100644 index 0000000..2e590c4 --- /dev/null +++ b/eval/asr_eval/audio.py @@ -0,0 +1,42 @@ +"""音频加载与重采样工具：统一到 16k/mono，供 gummy(pcm16) 与本地引擎(float) 使用。""" + +from __future__ import annotations + +import numpy as np +import soundfile as sf + +TARGET_SR = 16000 + + +def _resample(data: np.ndarray, sr: int, target_sr: int) -> np.ndarray: + if sr == target_sr: + return data + import soxr + + return soxr.resample(data, sr, target_sr) + + +def load_pcm16(path: str, target_sr: int = TARGET_SR) -> tuple[np.ndarray, int]: + """读为 int16 单声道 PCM（gummy 推流用）。""" + data, sr = sf.read(path, dtype="float32", always_2d=False) + if data.ndim > 1: + data = data.mean(axis=1) + data = _resample(data, sr, target_sr) + # float32(-1,1) -> int16 + pcm = np.clip(data, -1.0, 1.0) + pcm = (pcm * 32767.0).astype(np.int16) + return pcm, target_sr + + +def load_float(path: str, target_sr: int = TARGET_SR) -> tuple[np.ndarray, int]: + """读为 float32 单声道（本地引擎用）。""" + data, sr = sf.read(path, dtype="float32", always_2d=False) + if data.ndim > 1: + data = data.mean(axis=1) + data = _resample(data, sr, target_sr) + return data.astype(np.float32), target_sr + + +def duration_sec(path: str) -> float: + info = sf.info(path) + return info.frames / info.samplerate diff --git a/eval/asr_eval/cli.py b/eval/asr_eval/cli.py new file mode 100644 index 0000000..1a152fb --- /dev/null +++ b/eval/asr_eval/cli.py @@ -0,0 +1,200 @@ +"""命令行入口：python -m asr_eval run --config config.yaml + +流程：载 config → 物料化样本 → 逐引擎逐样本识别(本地引擎包资源探针) → 算分 → 出报告。 +gummy 的 API key：优先 env DASHSCOPE_API_KEY，否则读 stdin 首行 +（遵全局规则：rbw get dashscope-api-key | python -m asr_eval run ...）。 +""" + +from __future__ import annotations + +import argparse +import os +import sys +from datetime import datetime + +import yaml +from rich.console import Console +from rich.progress import BarColumn, Progress, TextColumn, TimeElapsedColumn + +from .datasets import build_dataset +from .engines import build_engine +from .metrics.err import score_sample +from .metrics.resource import ResourceProbe +from .report import aggregate, render_markdown, write_reports + +console = Console() + + +def _resolve_api_key() -> str: + key = os.environ.get("DASHSCOPE_API_KEY", "").strip() + if key: + return key + if not sys.stdin.isatty(): + line = sys.stdin.readline().strip() + if line: + return line + return "" + + +def _load_samples(cfg: dict, only: set[str] | None, limit: int | None) -> list: + audio_cache = cfg.get("audio_cache", "datasets/_audio_cache") + samples = [] + for dcfg in cfg.get("datasets", []): + if only and dcfg["name"] not in only: + continue + ds = build_dataset(dcfg, audio_cache) + console.print(f"[cyan]载入数据集[/] {dcfg['name']} (type={dcfg['type']}) ...") + try: + got = list(ds.samples(limit=limit)) + except Exception as ex: + console.print(f"[red]数据集 {dcfg['name']} 载入失败，跳过：{ex}[/]") + continue + console.print(f" → {len(got)} 条") + samples.extend(got) + return samples + + +def _enabled_engines(cfg: dict, only: set[str] | None, api_key: str) -> list[dict]: + out = [] + for ecfg in cfg.get("engines", []): + if not ecfg.get("enabled", True): + continue + if only and ecfg["name"] not in only: + continue + if ecfg["type"] == "gummy": + ecfg = {**ecfg, "api_key": api_key} + out.append(ecfg) + return out + + +def cmd_run(args: argparse.Namespace) -> int: + with open(args.config, encoding="utf-8") as f: + cfg = yaml.safe_load(f) + + limit = args.sample_limit if args.sample_limit is not None else cfg.get("sample_limit") + only_engines = set(args.engines.split(",")) if args.engines else None + only_datasets = set(args.datasets.split(",")) if args.datasets else None + + engine_cfgs = _enabled_engines(cfg, only_engines, "") + need_key = any(e["type"] == "gummy" for e in engine_cfgs) + api_key = _resolve_api_key() if need_key else "" + if need_key and not api_key: + console.print("[red]gummy 已启用但未取到 API key（设 DASHSCOPE_API_KEY 或 stdin 传入）[/]") + return 2 + engine_cfgs = _enabled_engines(cfg, only_engines, api_key) + + samples = _load_samples(cfg, only_datasets, limit) + if not samples: + console.print("[red]没有可评估的样本，检查 datasets 配置或 manifest[/]") + return 1 + + rows: list[dict] = [] + engines_meta: dict[str, dict] = {} + + for ecfg in engine_cfgs: + name = ecfg["name"] + console.print(f"\n[bold magenta]引擎 {name}[/] (type={ecfg['type']}) 加载中 ...") + engine = build_engine(ecfg) + try: + engine.load() + except Exception as ex: + console.print(f"[red]引擎 {name} 加载失败，跳过：{ex}[/]") + continue + + engines_meta[name] = { + "is_local": engine.is_local, + "kind": engine.kind, + "model_size_mb": engine.model_size_mb(), + "cost_per_min": engine.cost_per_min(), + } + + with Progress( + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TextColumn("{task.completed}/{task.total}"), + TimeElapsedColumn(), + console=console, + ) as prog: + task = prog.add_task(f"{name}", total=len(samples)) + for s in samples: + tr = _transcribe_one(engine, s.audio_path) + sc = score_sample(s.ref_text, tr.text) + rows.append({ + "engine": name, + "dataset": s.dataset, + "category": s.category, + "lang": s.lang, + "domain": s.domain, + "sample_id": s.id, + "ref": s.ref_text, + "hyp": tr.text, + "counts": sc, + "audio_sec": tr.audio_sec, + "proc_sec": tr.proc_sec, + "rtf": tr.rtf, + "first_partial_sec": tr.first_partial_sec, + "finalize_sec": tr.finalize_sec, + "peak_rss_mb": tr.peak_rss_mb, + "avg_cpu": tr.avg_cpu, + "error": tr.error, + }) + prog.advance(task) + engine.unload() + + if not rows: + console.print("[red]无结果（所有引擎都失败了？）[/]") + return 1 + + report = aggregate(rows, engines_meta) + out_dir = os.path.join(cfg.get("output_dir", "results"), datetime.now().strftime("%Y%m%d-%H%M%S")) + paths = write_reports(report, out_dir) + _dump_raw(rows, out_dir) + + console.print("\n" + render_markdown(report)) + console.print(f"\n[green]报告已写入[/] {out_dir}/ (report.json / report.md / report.html)") + return 0 + + +def _transcribe_one(engine, audio_path: str): + """本地引擎包资源探针；云端引擎直接调用。""" + if engine.is_local: + with ResourceProbe() as probe: + tr = engine.transcribe(audio_path) + if tr.peak_rss_mb is None: + tr.peak_rss_mb = probe.peak_rss_mb + if tr.avg_cpu is None: + tr.avg_cpu = probe.avg_cpu + return tr + return engine.transcribe(audio_path) + + +def _dump_raw(rows: list[dict], out_dir: str) -> None: + import json + + path = os.path.join(out_dir, "results.jsonl") + with open(path, "w", encoding="utf-8") as f: + for r in rows: + rec = {k: v for k, v in r.items() if k != "counts"} + rec["cer"] = r["counts"]["cer"].rate + rec["wer"] = r["counts"]["wer"].rate + rec["mer"] = r["counts"]["mer"].rate + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + + +def main() -> None: + ap = argparse.ArgumentParser(prog="asr_eval", description="dudu ASR 模型评估框架") + sub = ap.add_subparsers(dest="cmd", required=True) + + run = sub.add_parser("run", help="跑评估") + run.add_argument("--config", default="config.yaml") + run.add_argument("--engines", help="只跑这些引擎（逗号分隔，覆盖 enabled）") + run.add_argument("--datasets", help="只跑这些数据集（逗号分隔）") + run.add_argument("--sample-limit", type=int, default=None, help="每个数据集采样上限（覆盖 config）") + run.set_defaults(func=cmd_run) + + args = ap.parse_args() + sys.exit(args.func(args)) + + +if __name__ == "__main__": + main() diff --git a/eval/asr_eval/datasets/__init__.py b/eval/asr_eval/datasets/__init__.py new file mode 100644 index 0000000..e5a8a87 --- /dev/null +++ b/eval/asr_eval/datasets/__init__.py @@ -0,0 +1,25 @@ +"""数据集注册表：config 的 type(hf|custom) 映射到适配器。 + +公共集（ASCEND/AISHELL/LibriSpeech）统一走 HFDataset，差异全在 config.yaml 里描述 +（hf_id/split/text_field/lang），所以"广覆盖 + 自定义"只需改配置，不必加代码。 +""" + +from __future__ import annotations + +from .base import Dataset + + +def build_dataset(cfg: dict, audio_cache: str) -> Dataset: + dtype = cfg["type"] + if dtype == "hf": + from .hf import HFDataset + + return HFDataset(cfg, audio_cache) + if dtype == "custom": + from .custom import CustomDataset + + return CustomDataset(cfg) + raise ValueError(f"未知数据集类型: {dtype}") + + +__all__ = ["Dataset", "build_dataset"] diff --git a/eval/asr_eval/datasets/base.py b/eval/asr_eval/datasets/base.py new file mode 100644 index 0000000..11a6eef --- /dev/null +++ b/eval/asr_eval/datasets/base.py @@ -0,0 +1,34 @@ +"""数据集适配器基类与音频物料化工具。""" + +from __future__ import annotations + +import os +from collections.abc import Iterator + +import numpy as np +import soundfile as sf + +from ..manifest import Sample + +TARGET_SR = 16000 + + +class Dataset: + name: str = "base" + + def samples(self, limit: int | None = None) -> Iterator[Sample]: + raise NotImplementedError + + +def write_wav_16k(array: np.ndarray, sr: int, out_path: str) -> None: + """把（可能任意采样率的）单/多声道数组写成 16k/mono/16bit wav。""" + os.makedirs(os.path.dirname(out_path), exist_ok=True) + data = np.asarray(array, dtype=np.float32) + if data.ndim > 1: + data = data.mean(axis=1) + if sr != TARGET_SR: + import soxr + + data = soxr.resample(data, sr, TARGET_SR) + data = np.clip(data, -1.0, 1.0) + sf.write(out_path, (data * 32767.0).astype(np.int16), TARGET_SR, subtype="PCM_16") diff --git a/eval/asr_eval/datasets/custom.py b/eval/asr_eval/datasets/custom.py new file mode 100644 index 0000000..c943470 --- /dev/null +++ b/eval/asr_eval/datasets/custom.py @@ -0,0 +1,62 @@ +"""自定义集：JSONL manifest，每行一条样本。 + +格式（audio 路径相对 manifest 文件所在目录解析）： + {"audio": "clips/a.wav", "text": "帮我把这份 weekly report 整理一下", "lang": "zh-en", "domain": "tech"} + +lang 缺省按文本自动猜（含中文且含 ASCII 字母→zh-en；纯 ASCII→en；否则 zh）。 +""" + +from __future__ import annotations + +import json +import os +import re +from collections.abc import Iterator + +from ..manifest import Sample + +_CJK = re.compile(r"[一-鿿]") +_ASCII_ALPHA = re.compile(r"[A-Za-z]") + + +def _guess_lang(text: str) -> str: + has_cjk = bool(_CJK.search(text)) + has_en = bool(_ASCII_ALPHA.search(text)) + if has_cjk and has_en: + return "zh-en" + if has_en and not has_cjk: + return "en" + return "zh" + + +class CustomDataset: + def __init__(self, cfg: dict): + self.name = cfg.get("name", "custom") + self.manifest = cfg["manifest"] + self.domain_default = cfg.get("domain", "general") + self._base = os.path.dirname(os.path.abspath(self.manifest)) + + def samples(self, limit: int | None = None) -> Iterator[Sample]: + with open(self.manifest, encoding="utf-8") as f: + n = 0 + for i, line in enumerate(f): + line = line.strip() + if not line or line.startswith("#"): + continue + if limit is not None and n >= limit: + break + row = json.loads(line) + text = (row.get("text") or "").strip() + audio = row["audio"] + if not os.path.isabs(audio): + audio = os.path.join(self._base, audio) + lang = row.get("lang") or _guess_lang(text) + n += 1 + yield Sample( + id=row.get("id", f"{self.name}-{i:06d}"), + audio_path=audio, + ref_text=text, + lang=lang, + domain=row.get("domain", self.domain_default), + dataset=self.name, + ) diff --git a/eval/asr_eval/datasets/hf.py b/eval/asr_eval/datasets/hf.py new file mode 100644 index 0000000..1c2dfe4 --- /dev/null +++ b/eval/asr_eval/datasets/hf.py @@ -0,0 +1,89 @@ +"""通用 HuggingFace 数据集适配器（流式）。 + +config 条目示例： + - name: ascend + type: hf + hf_id: CAiRE/ASCEND + split: test + text_field: transcription + audio_field: audio # 默认 audio + lang: zh-en + domain: general + trust_remote_code: false + +流式（streaming=True）避免为跑小样而下载整个数据集；每条把音频物料化为 16k wav 缓存。 +""" + +from __future__ import annotations + +import io +import os +from collections.abc import Iterator + +import soundfile as sf + +from ..manifest import Sample +from .base import Dataset, write_wav_16k + + +class HFDataset(Dataset): + def __init__(self, cfg: dict, audio_cache: str): + self.name = cfg["name"] + self.hf_id = cfg["hf_id"] + self.hf_config = cfg.get("hf_config") + self.split = cfg.get("split", "test") + self.text_field = cfg["text_field"] + self.audio_field = cfg.get("audio_field", "audio") + self.lang = cfg.get("lang", "zh") + self.domain = cfg.get("domain", "general") + self.trust_remote_code = cfg.get("trust_remote_code", False) + self.cache_dir = os.path.join(audio_cache, self.name) + + def samples(self, limit: int | None = None) -> Iterator[Sample]: + from datasets import Audio, load_dataset + + ds = load_dataset( + self.hf_id, + self.hf_config, + split=self.split, + streaming=True, + trust_remote_code=self.trust_remote_code, + ) + # 关闭自动解码（新版 datasets 需 torchcodec）：拿原始 bytes/path 自己用 soundfile 读，避开重依赖。 + ds = ds.cast_column(self.audio_field, Audio(decode=False)) + n = 0 + for i, row in enumerate(ds): + if limit is not None and n >= limit: + break + text = (row.get(self.text_field) or "").strip() + if not text: + continue + audio = row.get(self.audio_field) + if not audio: + continue + wav_path = os.path.join(self.cache_dir, f"{i:06d}.wav") + if not os.path.exists(wav_path): + array, sr = self._decode(audio) + if array is None: + continue + write_wav_16k(array, sr, wav_path) + n += 1 + yield Sample( + id=f"{self.name}-{i:06d}", + audio_path=wav_path, + ref_text=text, + lang=self.lang, + domain=self.domain, + dataset=self.name, + ) + + @staticmethod + def _decode(audio: dict): + """从未解码的 HF 音频列读出 (array, sr)：优先 bytes，其次本地 path。""" + if audio.get("bytes"): + data, sr = sf.read(io.BytesIO(audio["bytes"]), dtype="float32", always_2d=False) + return data, sr + if audio.get("path") and os.path.exists(audio["path"]): + data, sr = sf.read(audio["path"], dtype="float32", always_2d=False) + return data, sr + return None, None diff --git a/eval/asr_eval/engines/__init__.py b/eval/asr_eval/engines/__init__.py new file mode 100644 index 0000000..3b92256 --- /dev/null +++ b/eval/asr_eval/engines/__init__.py @@ -0,0 +1,45 @@ +"""引擎注册表：config.yaml 里的 type 映射到具体引擎类（懒加载，避免未装的重依赖被导入）。""" + +from __future__ import annotations + +from .base import Engine, Transcript + + +def build_engine(cfg: dict) -> Engine: + """按 config 的单个 engine 条目构造引擎实例。""" + etype = cfg["type"] + name = cfg.get("name", etype) + + if etype == "gummy": + from .gummy import GummyEngine + + return GummyEngine( + name=name, + api_key=cfg["api_key"], + model=cfg.get("model", "gummy-realtime-v1"), + cost_per_min=cfg.get("cost_per_min"), + realtime_factor=cfg.get("realtime_factor", 2.0), + ) + if etype == "whisper": + from .whisper import WhisperEngine + + return WhisperEngine( + name=name, + model_size=cfg.get("model_size", "small"), + device=cfg.get("device", "cpu"), + compute_type=cfg.get("compute_type", "int8"), + model_dir=cfg.get("model_dir"), + ) + if etype == "sensevoice": + from .funasr import SenseVoiceEngine + + return SenseVoiceEngine(name=name, device=cfg.get("device", "cpu"), model_dir=cfg.get("model_dir")) + if etype == "funasr": + from .funasr import ParaformerEngine + + return ParaformerEngine(name=name, device=cfg.get("device", "cpu"), model_dir=cfg.get("model_dir")) + + raise ValueError(f"未知引擎类型: {etype}") + + +__all__ = ["Engine", "Transcript", "build_engine"] diff --git a/eval/asr_eval/engines/base.py b/eval/asr_eval/engines/base.py new file mode 100644 index 0000000..b0b6f54 --- /dev/null +++ b/eval/asr_eval/engines/base.py @@ -0,0 +1,53 @@ +"""引擎接口与单次识别结果。 + +两类引擎统一到 transcribe()，但计时口径不同： + - streaming（gummy）：有首包延迟 first_partial_sec、定稿延迟 finalize_sec。 + - offline（本地）：只有总处理时长与 RTF；资源探针在 runner 层包裹（仅本地）。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class Transcript: + text: str + audio_sec: float + proc_sec: float # transcribe 墙钟耗时 + first_partial_sec: float | None = None + finalize_sec: float | None = None + peak_rss_mb: float | None = None # 由 runner 的 ResourceProbe 回填（本地引擎） + avg_cpu: float | None = None + error: str | None = None + + @property + def rtf(self) -> float | None: + if not self.audio_sec: + return None + return self.proc_sec / self.audio_sec + + +class Engine: + """引擎基类。子类至少实现 transcribe()；load()/unload() 处理重模型生命周期。""" + + name: str = "base" + kind: str = "offline" # "offline" | "streaming" + is_local: bool = True + + def load(self) -> None: + """加载模型（本地引擎重操作）。云端引擎可空实现。""" + + def transcribe(self, audio_path: str) -> Transcript: + raise NotImplementedError + + def unload(self) -> None: + """释放模型，便于多引擎串行评估时回收内存。""" + + def model_size_mb(self) -> float | None: + """磁盘上模型权重大小（MB）；云端返回 None。""" + return None + + def cost_per_min(self) -> float | None: + """每分钟成本（云端）；本地返回 None。""" + return None diff --git a/eval/asr_eval/engines/funasr.py b/eval/asr_eval/engines/funasr.py new file mode 100644 index 0000000..0774bf5 --- /dev/null +++ b/eval/asr_eval/engines/funasr.py @@ -0,0 +1,87 @@ +"""本地 FunASR 引擎：SenseVoice-Small（多语，中英混说强）与 Paraformer-zh（纯中 SOTA）。 + +依赖：pip install '.[funasr]' +两者都走 funasr.AutoModel，离线整段识别。SenseVoice 输出带 <|zh|><|EMO|> 等富标签，需剥离。 +""" + +from __future__ import annotations + +import os +import re +import time + +from ..audio import duration_sec +from ..metrics.resource import dir_size_mb +from .base import Engine, Transcript + +_TAG = re.compile(r"<\|[^|]*\|>") + + +def _strip_tags(text: str) -> str: + return _TAG.sub("", text).strip() + + +class _FunASRBase(Engine): + kind = "offline" + is_local = True + repo: str = "" + + def __init__(self, name: str, device: str = "cpu", model_dir: str | None = None): + self.name = name + self.device = device + self.model_dir = model_dir + self.model = None + self._weights_path: str | None = None + + def load(self) -> None: + from funasr import AutoModel + + kwargs = {"model": self.repo, "disable_update": True, "device": self.device} + if self.model_dir: + kwargs["cache_dir"] = self.model_dir + self.model = AutoModel(**kwargs) + self._resolve_weights_path() + + def _resolve_weights_path(self) -> None: + # funasr 默认从 modelscope 下载；尝试常见属性与缓存目录（best-effort） + for attr in ("model_path", "model_pth", "kwargs"): + val = getattr(self.model, attr, None) + if isinstance(val, str) and os.path.exists(val): + self._weights_path = val + return + if isinstance(val, dict) and isinstance(val.get("model_path"), str): + self._weights_path = val["model_path"] + return + self._weights_path = self.model_dir + + def transcribe(self, audio_path: str) -> Transcript: + if self.model is None: + self.load() + t0 = time.monotonic() + res = self.model.generate(input=audio_path) + proc = time.monotonic() - t0 + text = "" + if res and isinstance(res, list) and res[0].get("text"): + text = _strip_tags(res[0]["text"]) + return Transcript(text=text, audio_sec=duration_sec(audio_path), proc_sec=proc) + + def model_size_mb(self) -> float | None: + size = dir_size_mb(self._weights_path) if self._weights_path else 0.0 + return size or None + + def unload(self) -> None: + self.model = None + + +class SenseVoiceEngine(_FunASRBase): + repo = "iic/SenseVoiceSmall" + + def __init__(self, name: str = "sensevoice", device: str = "cpu", model_dir: str | None = None): + super().__init__(name=name, device=device, model_dir=model_dir) + + +class ParaformerEngine(_FunASRBase): + repo = "paraformer-zh" + + def __init__(self, name: str = "paraformer-zh", device: str = "cpu", model_dir: str | None = None): + super().__init__(name=name, device=device, model_dir=model_dir) diff --git a/eval/asr_eval/engines/gummy.py b/eval/asr_eval/engines/gummy.py new file mode 100644 index 0000000..7fa0091 --- /dev/null +++ b/eval/asr_eval/engines/gummy.py @@ -0,0 +1,180 @@ +"""云端 gummy 引擎：阿里云百炼 DashScope 流式 ASR（gummy-realtime-v1）。 + +协议对照 server/internal/asr/gummy.go： + run-task → 等 task-started → 推二进制 PCM(3200B/16k/pcm) → finish-task → 收 task-finished。 +下行解析 payload.output.transcription(gummy) 或 sentence(paraformer)， +sentence_end 兼容 bool 与字符串 "true"/"false"（gummy.go 的 flexBool 怪癖）。 +""" + +from __future__ import annotations + +import json +import threading +import time +import uuid + +import websocket # websocket-client + +from ..audio import load_pcm16 +from .base import Engine, Transcript + +WS_URL = "wss://dashscope.aliyuncs.com/api-ws/v1/inference" +FRAME = 3200 # 1600 samples = 100ms @16k + + +def _flex_bool(v) -> bool: + return v is True or v == "true" + + +class GummyEngine(Engine): + name = "gummy" + kind = "streaming" + is_local = False + + def __init__( + self, + name: str = "gummy", + api_key: str = "", + model: str = "gummy-realtime-v1", + cost_per_min: float | None = None, + realtime_factor: float = 2.0, + ): + self.name = name + self.api_key = api_key + self.model = model + self._cost = cost_per_min + self.rt = realtime_factor # 推流倍速；2.0 = 2x 实时（同 gummycheck） + + def cost_per_min(self) -> float | None: + return self._cost + + def _run_task_msg(self, task_id: str, sample_rate: int) -> str: + return json.dumps( + { + "header": {"action": "run-task", "task_id": task_id, "streaming": "duplex"}, + "payload": { + "task_group": "audio", + "task": "asr", + "function": "recognition", + "model": self.model, + "parameters": { + "sample_rate": sample_rate, + "format": "pcm", + "transcription_enabled": True, + "translation_enabled": False, + }, + "input": {}, + }, + } + ) + + def _finish_task_msg(self, task_id: str) -> str: + return json.dumps( + { + "header": {"action": "finish-task", "task_id": task_id, "streaming": "duplex"}, + "payload": {"input": {}}, + } + ) + + def transcribe(self, audio_path: str) -> Transcript: + pcm, sr = load_pcm16(audio_path, 16000) + audio = pcm.tobytes() + audio_sec = len(pcm) / sr + task_id = uuid.uuid4().hex + + started = threading.Event() + finished = threading.Event() + results: list[tuple[float, str, bool]] = [] # (rel_t, text, is_final) + err: list[str | None] = [None] + first_partial: list[float | None] = [None] + t0 = time.monotonic() + + try: + ws = websocket.create_connection( + WS_URL, + header=[ + f"Authorization: bearer {self.api_key}", + "X-DashScope-DataInspection: enable", + ], + timeout=20, + ) + except Exception as ex: # 连接失败 + return Transcript(text="", audio_sec=audio_sec, proc_sec=time.monotonic() - t0, error=f"dial: {ex}") + + def reader() -> None: + try: + while True: + msg = ws.recv() + if not msg: + continue + ev = json.loads(msg) + h = ev.get("header", {}) + event = h.get("event") + if event == "task-started": + started.set() + elif event == "result-generated": + out = ev.get("payload", {}).get("output", {}) + sen = out.get("transcription") or out.get("sentence") + if not sen or not sen.get("text"): + continue + is_final = _flex_bool(sen.get("sentence_end")) or _flex_bool(sen.get("is_sentence_end")) + if first_partial[0] is None: + first_partial[0] = time.monotonic() - t0 + results.append((time.monotonic() - t0, sen["text"], is_final)) + elif event == "task-finished": + finished.set() + break + elif event == "task-failed": + err[0] = h.get("error_message", "task-failed") + finished.set() + break + except Exception as ex: + if not finished.is_set(): + err[0] = str(ex) + finished.set() + + rt = threading.Thread(target=reader, daemon=True) + rt.start() + + try: + ws.send(self._run_task_msg(task_id, sr)) + if not started.wait(timeout=10): + ws.close() + return Transcript( + text="", audio_sec=audio_sec, proc_sec=time.monotonic() - t0, error="task-started timeout" + ) + + # 推流：每帧 100ms 音频，按倍速 sleep（2x → 50ms） + per_frame_sleep = 0.1 / self.rt + for off in range(0, len(audio), FRAME): + ws.send_binary(audio[off : off + FRAME]) + time.sleep(per_frame_sleep) + last_audio_t = time.monotonic() + + ws.send(self._finish_task_msg(task_id)) + finished.wait(timeout=30) + except Exception as ex: + err[0] = err[0] or str(ex) + finally: + try: + ws.close() + except Exception: + pass + + # gummy 按句给 final；拼接所有 final，无 final 则回退最后一条 partial + finals = [t for _, t, f in results if f] + text = "".join(finals) if finals else (results[-1][1] if results else "") + + finalize_sec = None + final_rel_times = [rel for rel, _, f in results if f] + if final_rel_times: + finalize_sec = max(0.0, (t0 + final_rel_times[-1]) - last_audio_t) + + return Transcript( + text=text, + audio_sec=audio_sec, + proc_sec=time.monotonic() - t0, + first_partial_sec=first_partial[0], + finalize_sec=finalize_sec, + error=err[0], + ) diff --git a/eval/asr_eval/engines/whisper.py b/eval/asr_eval/engines/whisper.py new file mode 100644 index 0000000..f15fbad --- /dev/null +++ b/eval/asr_eval/engines/whisper.py @@ -0,0 +1,70 @@ +"""本地 faster-whisper 引擎（CTranslate2，CPU-first）。多语，对中英混说有基础能力。 + +依赖：pip install '.[whisper]' +""" + +from __future__ import annotations + +import time + +from ..metrics.resource import dir_size_mb +from .base import Engine, Transcript + + +class WhisperEngine(Engine): + kind = "offline" + is_local = True + + def __init__( + self, + name: str = "whisper-small", + model_size: str = "small", + device: str = "cpu", + compute_type: str = "int8", + model_dir: str | None = None, + ): + self.name = name + self.model_size = model_size + self.device = device + self.compute_type = compute_type + self.model_dir = model_dir + self.model = None + self._weights_path: str | None = None + + def load(self) -> None: + from faster_whisper import WhisperModel + + self.model = WhisperModel( + self.model_size, + device=self.device, + compute_type=self.compute_type, + download_root=self.model_dir, + ) + self._resolve_weights_path() + + def _resolve_weights_path(self) -> None: + # 解析磁盘权重路径用于报模型大小（best-effort） + try: + from huggingface_hub import snapshot_download + + repo = f"Systran/faster-whisper-{self.model_size}" + self._weights_path = snapshot_download(repo, local_files_only=True, cache_dir=self.model_dir) + except Exception: + self._weights_path = self.model_dir + + def transcribe(self, audio_path: str) -> Transcript: + if self.model is None: + self.load() + t0 = time.monotonic() + # language=None → 自动检测；中英混说交给模型自身 + segments, info = self.model.transcribe(audio_path, language=None, beam_size=5) + text = "".join(seg.text for seg in segments) # 生成器在此 join 时才真正解码 + proc = time.monotonic() - t0 + return Transcript(text=text.strip(), audio_sec=float(info.duration), proc_sec=proc) + + def model_size_mb(self) -> float | None: + size = dir_size_mb(self._weights_path) if self._weights_path else 0.0 + return size or None + + def unload(self) -> None: + self.model = None diff --git a/eval/asr_eval/manifest.py b/eval/asr_eval/manifest.py new file mode 100644 index 0000000..99c2531 --- /dev/null +++ b/eval/asr_eval/manifest.py @@ -0,0 +1,32 @@ +"""统一样本模型：所有数据集适配器都产出 Sample，下游引擎/指标只认它。""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class Sample: + """一条评估样本。 + + lang 取值约定（决定 breakdown 归类）： + - "zh" 纯普通话 + - "en" 纯英文 + - "zh-en" 中英混说（code-switching，本框架重点） + """ + + id: str + audio_path: str # 16k/mono/16bit wav 优先；非此规格引擎侧会重采样 + ref_text: str # 参考（标注）文本 + lang: str = "zh" + domain: str = "general" + dataset: str = "custom" + + @property + def category(self) -> str: + """报告里的粗分类。""" + if self.lang == "zh-en": + return "code-switch" + if self.lang == "en": + return "pure-en" + return "pure-zh" diff --git a/eval/asr_eval/metrics/__init__.py b/eval/asr_eval/metrics/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/eval/asr_eval/metrics/err.py b/eval/asr_eval/metrics/err.py new file mode 100644 index 0000000..46dd60b --- /dev/null +++ b/eval/asr_eval/metrics/err.py @@ -0,0 +1,98 @@ +"""准确率指标：CER（字级）/ WER（词级）/ MER（中英混合）。 + +分词策略： + - CER : 去空格后逐字符（中文每字、英文每字母各一 token）。 + - WER : 按空白切词（主要对英文有意义）。 + - MER : 中文按【字】、英文/数字按【词】，混合 token 序列算编辑距离——这是中英混说的 headline 指标。 + +聚合：错误率必须在语料级聚合（累加 S/D/I/N 后再求率），不能对每条样本的率求平均， +否则短句权重被放大、结果失真。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +import jiwer + +from .text_norm import normalize + +_CJK = re.compile(r"[一-鿿㐀-䶿]") + + +@dataclass +class Counts: + """一组编辑距离统计。N = 参考长度 = S + D + H。""" + + S: int = 0 # 替换 + D: int = 0 # 删除 + I: int = 0 # 插入 + N: int = 0 # 参考 token 数 + + def __add__(self, o: "Counts") -> "Counts": + return Counts(self.S + o.S, self.D + o.D, self.I + o.I, self.N + o.N) + + @property + def rate(self) -> float: + if self.N == 0: + return 1.0 if (self.S + self.D + self.I) > 0 else 0.0 + return (self.S + self.D + self.I) / self.N + + +def char_tokens(text: str) -> list[str]: + return [c for c in text if not c.isspace()] + + +def word_tokens(text: str) -> list[str]: + return text.split() + + +def mixed_tokens(text: str) -> list[str]: + """中文按字、其余按词。""" + tokens: list[str] = [] + buf = "" + for ch in text: + if _CJK.match(ch): + if buf.strip(): + tokens.extend(buf.split()) + buf = "" + tokens.append(ch) + else: + buf += ch + if buf.strip(): + tokens.extend(buf.split()) + return tokens + + +def _counts(ref_tokens: list[str], hyp_tokens: list[str]) -> Counts: + if not ref_tokens: + return Counts(S=0, D=0, I=len(hyp_tokens), N=0) + out = jiwer.process_words(" ".join(ref_tokens), " ".join(hyp_tokens)) + return Counts( + S=out.substitutions, + D=out.deletions, + I=out.insertions, + N=out.substitutions + out.deletions + out.hits, + ) + + +def score_sample(ref_text: str, hyp_text: str) -> dict[str, Counts]: + """对一条样本算 CER/WER/MER 的原始计数（供语料级聚合）。""" + ref = normalize(ref_text) + hyp = normalize(hyp_text) + return { + "cer": _counts(char_tokens(ref), char_tokens(hyp)), + "wer": _counts(word_tokens(ref), word_tokens(hyp)), + "mer": _counts(mixed_tokens(ref), mixed_tokens(hyp)), + } + + +def empty_counts() -> dict[str, Counts]: + return {"cer": Counts(), "wer": Counts(), "mer": Counts()} + + +def add_counts(acc: dict[str, Counts], one: dict[str, Counts]) -> dict[str, Counts]: + for k in acc: + acc[k] = acc[k] + one[k] + return acc diff --git a/eval/asr_eval/metrics/resource.py b/eval/asr_eval/metrics/resource.py new file mode 100644 index 0000000..567c400 --- /dev/null +++ b/eval/asr_eval/metrics/resource.py @@ -0,0 +1,85 @@ +"""资源探针：本地引擎推理时采样峰值内存/CPU；模型大小读磁盘。 + +注意：psutil 测的是整个 Python 进程 RSS（含解释器与已加载库），用于本地模型横向对比是 +合理代理量；报告会标注硬件环境，且只在本地引擎启用，云端 gummy 不测（填 N/A）。 +""" + +from __future__ import annotations + +import os +import threading +import time + +import psutil + + +class ResourceProbe: + """上下文管理器：with 块内后台采样进程 RSS / CPU%。""" + + def __init__(self, interval: float = 0.05): + self.interval = interval + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._proc = psutil.Process(os.getpid()) + self.peak_rss = 0 + self._cpu: list[float] = [] + + def __enter__(self) -> "ResourceProbe": + self._proc.cpu_percent(None) # 预热：首次调用返回 0，丢弃 + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + return self + + def _run(self) -> None: + while not self._stop.is_set(): + try: + self.peak_rss = max(self.peak_rss, self._proc.memory_info().rss) + c = self._proc.cpu_percent(None) + if c > 0: + self._cpu.append(c) + except Exception: + pass + time.sleep(self.interval) + + def __exit__(self, *_exc) -> None: + self._stop.set() + if self._thread: + self._thread.join(timeout=1.0) + + @property + def peak_rss_mb(self) -> float: + return self.peak_rss / 1024 / 1024 + + @property + def avg_cpu(self) -> float: + return sum(self._cpu) / len(self._cpu) if self._cpu else 0.0 + + +def dir_size_mb(path: str) -> float: + """目录下所有文件体积之和（MB）；路径不存在返回 0。""" + total = 0 + if not path or not os.path.exists(path): + return 0.0 + if os.path.isfile(path): + return os.path.getsize(path) / 1024 / 1024 + for root, _dirs, files in os.walk(path): + for f in files: + try: + total += os.path.getsize(os.path.join(root, f)) + except OSError: + pass + return total / 1024 / 1024 + + +def hardware_info() -> dict: + """报告里标注的硬件环境。""" + import platform + + vm = psutil.virtual_memory() + return { + "platform": platform.platform(), + "machine": platform.machine(), + "cpu_count": psutil.cpu_count(logical=True), + "cpu_count_physical": psutil.cpu_count(logical=False), + "total_mem_gb": round(vm.total / 1024**3, 1), + } diff --git a/eval/asr_eval/metrics/text_norm.py b/eval/asr_eval/metrics/text_norm.py new file mode 100644 index 0000000..74a4a20 --- /dev/null +++ b/eval/asr_eval/metrics/text_norm.py @@ -0,0 +1,53 @@ +"""评分前文本归一化。 + +中英混说不归一化就算分会严重失真（标点、繁简、全半角、大小写都会被算成错）。 +本模块做最小必要归一化；繁→简依赖 opencc，缺失时自动降级（仅告警一次）。 +""" + +from __future__ import annotations + +import re +import sys +import unicodedata + +_cc = None +_cc_warned = False + + +def _converter(): + global _cc, _cc_warned + if _cc is not None: + return _cc + try: + from opencc import OpenCC + + _cc = OpenCC("t2s") + except Exception: + _cc = False # 标记尝试过且失败 + if not _cc_warned: + print("[text_norm] 警告：opencc 不可用，跳过繁→简归一化", file=sys.stderr) + _cc_warned = True + return _cc + + +# 中英文常见标点（评分时整体移除） +_PUNCT = set( + "，。！？、；：“”‘’（）《》【】〔〕…—～·.,!?;:\"'()<>[]{}~`@#$%^&*-_=+|\\/。．" +) +_WS = re.compile(r"\s+") + + +def normalize(text: str, *, t2s: bool = True, lower: bool = True) -> str: + if not text: + return "" + # 全角→半角 + Unicode 兼容分解（NFKC 会把 ＡＢＣ→ABC、１２３→123） + text = unicodedata.normalize("NFKC", text) + if t2s: + cc = _converter() + if cc: + text = cc.convert(text) + if lower: + text = text.lower() + text = "".join(ch for ch in text if ch not in _PUNCT) + text = _WS.sub(" ", text).strip() + return text diff --git a/eval/asr_eval/report.py b/eval/asr_eval/report.py new file mode 100644 index 0000000..2bbce73 --- /dev/null +++ b/eval/asr_eval/report.py @@ -0,0 +1,234 @@ +"""汇总每条样本结果 → 语料级指标 + 延迟分位 + 资源，产出 JSON / Markdown / HTML。 + +错误率在语料级聚合（累加 S/D/I/N 再求率）。延迟取 p50/p95。资源取均值。 +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime + +import numpy as np + +from .metrics.err import Counts +from .metrics.resource import hardware_info + + +def _pct(values: list[float], p: float) -> float | None: + vals = [v for v in values if v is not None] + if not vals: + return None + return float(np.percentile(vals, p)) + + +def _mean(values: list[float]) -> float | None: + vals = [v for v in values if v is not None] + if not vals: + return None + return float(np.mean(vals)) + + +def _group_stats(rows: list[dict], cost_per_min: float | None) -> dict: + acc = {"cer": Counts(), "wer": Counts(), "mer": Counts()} + for r in rows: + for k in acc: + acc[k] = acc[k] + r["counts"][k] + audio_total = sum(r["audio_sec"] for r in rows) + return { + "samples": len(rows), + "errors": sum(1 for r in rows if r["error"]), + "cer": acc["cer"].rate, + "wer": acc["wer"].rate, + "mer": acc["mer"].rate, + "rtf_mean": _mean([r["rtf"] for r in rows]), + "first_partial_p50": _pct([r["first_partial_sec"] for r in rows], 50), + "first_partial_p95": _pct([r["first_partial_sec"] for r in rows], 95), + "finalize_p50": _pct([r["finalize_sec"] for r in rows], 50), + "finalize_p95": _pct([r["finalize_sec"] for r in rows], 95), + "peak_rss_mb_mean": _mean([r["peak_rss_mb"] for r in rows]), + "avg_cpu_mean": _mean([r["avg_cpu"] for r in rows]), + "audio_sec_total": audio_total, + "cost_total": (audio_total / 60.0 * cost_per_min) if cost_per_min else None, + } + + +def aggregate(rows: list[dict], engines_meta: dict) -> dict: + report = { + "generated_at": datetime.now().isoformat(timespec="seconds"), + "hardware": hardware_info(), + "engines": {}, + } + by_engine: dict[str, list[dict]] = {} + for r in rows: + by_engine.setdefault(r["engine"], []).append(r) + + for eng, erows in by_engine.items(): + meta = engines_meta.get(eng, {}) + cost = meta.get("cost_per_min") + cats: dict[str, list[dict]] = {} + dsets: dict[str, list[dict]] = {} + for r in erows: + cats.setdefault(r["category"], []).append(r) + dsets.setdefault(r["dataset"], []).append(r) + report["engines"][eng] = { + "is_local": meta.get("is_local"), + "kind": meta.get("kind"), + "model_size_mb": meta.get("model_size_mb"), + "cost_per_min": cost, + "overall": _group_stats(erows, cost), + "by_category": {c: _group_stats(rs, cost) for c, rs in sorted(cats.items())}, + "by_dataset": {d: _group_stats(rs, cost) for d, rs in sorted(dsets.items())}, + } + return report + + +# ---------- 渲染 ---------- + +def _pct_str(v: float | None) -> str: + return f"{v * 100:.2f}%" if v is not None else "—" + + +def _num(v: float | None, fmt: str = "{:.2f}") -> str: + return fmt.format(v) if v is not None else "—" + + +def render_markdown(report: dict) -> str: + hw = report["hardware"] + lines = [ + "# ASR 模型评估报告", + "", + f"- 生成时间：{report['generated_at']}", + f"- 硬件：{hw['platform']} · {hw['machine']} · {hw['cpu_count']} vCPU · {hw['total_mem_gb']} GB", + "", + "## 总览（各引擎全量）", + "", + "| 引擎 | 本地 | 样本 | 错误 | MER | CER | WER | RTF | 首包 p50 | 定稿 p50 | 峰值内存MB | 模型大小MB | 成本 |", + "|---|---|--:|--:|--:|--:|--:|--:|--:|--:|--:|--:|--:|", + ] + for eng, e in report["engines"].items(): + o = e["overall"] + lines.append( + "| {eng} | {loc} | {n} | {err} | {mer} | {cer} | {wer} | {rtf} | {fp} | {fin} | {mem} | {sz} | {cost} |".format( + eng=eng, + loc="是" if e["is_local"] else "云", + n=o["samples"], + err=o["errors"], + mer=_pct_str(o["mer"]), + cer=_pct_str(o["cer"]), + wer=_pct_str(o["wer"]), + rtf=_num(o["rtf_mean"]), + fp=_num(o["first_partial_p50"], "{:.2f}s") if o["first_partial_p50"] is not None else "—", + fin=_num(o["finalize_p50"], "{:.2f}s") if o["finalize_p50"] is not None else "—", + mem=_num(o["peak_rss_mb_mean"], "{:.0f}"), + sz=_num(e["model_size_mb"], "{:.0f}"), + cost=_num(o["cost_total"], "¥{:.4f}") if o["cost_total"] is not None else "—", + ) + ) + + # 中英混说专项 + lines += ["", "## 中英混说（code-switch）专项 MER", "", "| 引擎 | 样本 | MER | CER | WER |", "|---|--:|--:|--:|--:|"] + for eng, e in report["engines"].items(): + cs = e["by_category"].get("code-switch") + if not cs: + continue + lines.append( + f"| {eng} | {cs['samples']} | {_pct_str(cs['mer'])} | {_pct_str(cs['cer'])} | {_pct_str(cs['wer'])} |" + ) + + # 按数据集 + lines += ["", "## 按数据集 MER", "", "| 引擎 | 数据集 | 样本 | MER | CER | WER |", "|---|---|--:|--:|--:|--:|"] + for eng, e in report["engines"].items(): + for ds, g in e["by_dataset"].items(): + lines.append( + f"| {eng} | {ds} | {g['samples']} | {_pct_str(g['mer'])} | {_pct_str(g['cer'])} | {_pct_str(g['wer'])} |" + ) + lines.append("") + lines.append( + "> 注：流式(gummy)与离线(本地)延迟口径不同——gummy 的 RTF/延迟含 2x 推流与网络，" + "非纯算力；本地引擎为整段离线处理。资源(内存/CPU)仅本地引擎测量，含 Python 进程基底。" + ) + return "\n".join(lines) + + +_HTML_TMPL = """ +ASR 评估报告 + +

ASR 模型评估报告

+

生成时间 {generated_at} · {hw}

+{body} +

流式(gummy)与离线(本地)延迟口径不同；资源仅本地引擎测量，含 Python 进程基底。

+""" + + +def render_html(report: dict) -> str: + hw = report["hardware"] + + def table(headers, rows): + h = "".join(f"{x}" for x in headers) + body = "" + for row in rows: + body += "" + "".join(f"{x}" for x in row) + "" + return f"{h}{body}
" + + parts = ["

总览（各引擎全量）

"] + rows = [] + for eng, e in report["engines"].items(): + o = e["overall"] + rows.append([ + eng, "是" if e["is_local"] else "云", o["samples"], o["errors"], + _pct_str(o["mer"]), _pct_str(o["cer"]), _pct_str(o["wer"]), + _num(o["rtf_mean"]), + _num(o["first_partial_p50"], "{:.2f}s") if o["first_partial_p50"] is not None else "—", + _num(o["peak_rss_mb_mean"], "{:.0f}"), _num(e["model_size_mb"], "{:.0f}"), + _num(o["cost_total"], "¥{:.4f}") if o["cost_total"] is not None else "—", + ]) + parts.append(table( + ["引擎", "本地", "样本", "错误", "MER", "CER", "WER", "RTF", "首包p50", "峰值内存MB", "模型MB", "成本"], rows + )) + + parts.append("

中英混说专项 MER

") + cs_rows = [] + for eng, e in report["engines"].items(): + cs = e["by_category"].get("code-switch") + if cs: + cs_rows.append([eng, cs["samples"], _pct_str(cs["mer"]), _pct_str(cs["cer"]), _pct_str(cs["wer"])]) + parts.append(table(["引擎", "样本", "MER", "CER", "WER"], cs_rows)) + + parts.append("

按数据集

") + ds_rows = [] + for eng, e in report["engines"].items(): + for ds, g in e["by_dataset"].items(): + ds_rows.append([eng, ds, g["samples"], _pct_str(g["mer"]), _pct_str(g["cer"]), _pct_str(g["wer"])]) + parts.append(table(["引擎", "数据集", "样本", "MER", "CER", "WER"], ds_rows)) + + return _HTML_TMPL.format( + generated_at=report["generated_at"], + hw=f"{hw['platform']} · {hw['cpu_count']} vCPU · {hw['total_mem_gb']}GB", + body="\n".join(parts), + ) + + +def write_reports(report: dict, out_dir: str) -> dict: + os.makedirs(out_dir, exist_ok=True) + paths = { + "json": os.path.join(out_dir, "report.json"), + "md": os.path.join(out_dir, "report.md"), + "html": os.path.join(out_dir, "report.html"), + } + with open(paths["json"], "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) + with open(paths["md"], "w", encoding="utf-8") as f: + f.write(render_markdown(report)) + with open(paths["html"], "w", encoding="utf-8") as f: + f.write(render_html(report)) + return paths diff --git a/eval/config.yaml b/eval/config.yaml new file mode 100644 index 0000000..2d1ad80 --- /dev/null +++ b/eval/config.yaml @@ -0,0 +1,74 @@ +# dudu ASR 评估配置。改这里即可调整跑哪些引擎/数据集、采样多少。 +# 跑法：rbw get dashscope-api-key | python -m asr_eval run --config config.yaml + +output_dir: results +audio_cache: datasets/_audio_cache +sample_limit: null # null=全量；整数=每个数据集采样上限（冒烟用小值） + +# ---------------- 数据集 ---------------- +# 公共集统一走 type=hf（差异全在配置里）；自定义集 type=custom 指向 JSONL manifest。 +datasets: + - name: ascend # 中英混说金标准（headline） + type: hf + hf_id: CAiRE/ASCEND + split: test + text_field: transcription + lang: zh-en + domain: general + + - name: aishell1 # 纯普通话 + type: hf + hf_id: AISHELL/AISHELL-1 + split: test + text_field: text + lang: zh + domain: general + trust_remote_code: true + + - name: librispeech # 纯英文 + type: hf + hf_id: openslr/librispeech_asr + hf_config: clean + split: test + text_field: text + lang: en + domain: general + trust_remote_code: true + + - name: custom # 你的自录/业务场景音频 + type: custom + manifest: manifests/custom.example.jsonl + +# ---------------- 引擎 ---------------- +# 云端 gummy 与本地开源横向对比。本地引擎需先装 extras：pip install '.[whisper]' 或 '.[funasr]' +engines: + - name: gummy + type: gummy + model: gummy-realtime-v1 + realtime_factor: 2.0 # 推流倍速（同 gummycheck）；影响延迟口径，不影响准确率 + cost_per_min: 0.09 # ⚠️占位：按 dudu 实付 DashScope 单价改（此处暂用售价 ¥9/100min） + enabled: true + + - name: whisper-small + type: whisper + model_size: small + device: cpu + compute_type: int8 + enabled: true + + - name: whisper-large-v3 + type: whisper + model_size: large-v3 + device: cpu + compute_type: int8 + enabled: false # 体积大(~1.5GB)，按需开 + + - name: sensevoice + type: sensevoice # 多语，中英混说强 + device: cpu + enabled: false + + - name: paraformer-zh + type: funasr # 纯中 SOTA，英文/混说较弱 + device: cpu + enabled: false diff --git a/eval/manifests/custom.example.jsonl b/eval/manifests/custom.example.jsonl new file mode 100644 index 0000000..8e24883 --- /dev/null +++ b/eval/manifests/custom.example.jsonl @@ -0,0 +1,6 @@ +# 自定义测试集示例。每行一个 JSON（# 开头的注释行会被忽略）。 +# audio 路径相对本 manifest 文件所在目录解析；lang 可省略（按文本自动猜）。 +# 把你自录/业务场景的 wav 放进来，与公共集同管线评估。 +{"id": "cs-001", "audio": "clips/weekly-report.wav", "text": "帮我把这份 weekly report 整理一下，重点突出本周的 milestone", "lang": "zh-en", "domain": "tech"} +{"id": "cs-002", "audio": "clips/deploy.wav", "text": "把这个 service 部署到 staging 环境然后跑一遍 e2e test", "lang": "zh-en", "domain": "tech"} +{"id": "zh-001", "audio": "clips/zhoubao.wav", "text": "帮我把这份周报整理一下重点突出本周的进展", "lang": "zh", "domain": "general"} diff --git a/eval/manifests/smoke.jsonl b/eval/manifests/smoke.jsonl new file mode 100644 index 0000000..812c5e7 --- /dev/null +++ b/eval/manifests/smoke.jsonl @@ -0,0 +1,3 @@ +{"id": "smoke-a", "audio": "clips/smoke_a.wav", "text": "帮我把这份周报整理一下，重点突出本周的进展", "lang": "zh", "domain": "general"} +{"id": "smoke-b", "audio": "clips/smoke_b.wav", "text": "今天下午三点开会，记得带上笔记本电脑", "lang": "zh", "domain": "general"} +{"id": "smoke-c", "audio": "clips/smoke_c.wav", "text": "请把这个文件发送到我的邮箱", "lang": "zh", "domain": "general"} diff --git a/eval/pyproject.toml b/eval/pyproject.toml new file mode 100644 index 0000000..38526d3 --- /dev/null +++ b/eval/pyproject.toml @@ -0,0 +1,34 @@ +[project] +name = "dudu-asr-eval" +version = "0.1.0" +description = "dudu ASR 模型评估框架：横向对比云端 gummy 与本地开源模型（准确率/速度/资源）" +requires-python = ">=3.10" + +# 核心依赖：评估管线本身（不含本地推理引擎，按需装 extras） +dependencies = [ + "jiwer>=3.0", # CER/WER/MER 编辑距离 + "soundfile>=0.12", # 读音频 + "soxr>=0.3", # 重采样到 16k + "numpy>=1.24", + "psutil>=5.9", # 资源探针（cpu/mem） + "pyyaml>=6.0", # config.yaml + "websocket-client>=1.6", # gummy 云端 WS（对照 gummy.go 移植） + "opencc>=1.1", # 繁→简归一化（缺失时自动降级） + "jinja2>=3.1", # HTML 报告 + "rich>=13.0", # 进度/表格 + "datasets>=2.18", # 公共集（ASCEND/AISHELL/LibriSpeech） + "huggingface_hub>=0.20", +] + +[project.optional-dependencies] +# 本地开源引擎按需安装，避免核心管线被重依赖拖累 +whisper = ["faster-whisper>=1.0"] # CTranslate2，CPU-first +funasr = ["funasr>=1.0", "torch>=2.0", "torchaudio>=2.0"] # SenseVoice / Paraformer +all = ["faster-whisper>=1.0", "funasr>=1.0", "torch>=2.0", "torchaudio>=2.0"] + +[project.scripts] +asr-eval = "asr_eval.cli:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["asr_eval*"]