"""语音识别服务:faster-whisper,输出带词级时间戳的 segments。 CPU dev: tiny.en + int8;GPU prod: large-v3-turbo + float16。同一份代码。 """ from __future__ import annotations import logging from pathlib import Path from ..config import get_settings from .model_manager import get_model_manager from .types import Segment, Word logger = logging.getLogger("audio2text.asr") def transcribe(wav_path: Path) -> list[Segment]: """转写 wav,返回 segments(含词级时间戳)。 Args: wav_path: 16kHz mono PCM wav Returns: list[Segment],每个 Segment 带词级 words(若 word_timestamps 启用)。 """ s = get_settings().asr if not wav_path.is_file(): raise FileNotFoundError(f"音频不存在:{wav_path}") model = get_model_manager().get_asr() logger.debug("开始转写 %s(model=%s language=%s batch_size=%d)", wav_path.name, s.model, s.language, s.batch_size) segments_gen, info = model.transcribe( str(wav_path), language=s.language, word_timestamps=s.word_timestamps, vad_filter=s.vad_filter, beam_size=s.beam_size, # beam search 宽度(config 可调);GPU turbo 建议降到 2 加速 batch_size=s.batch_size, # 批量解码:多音频块一次性送 GPU without_timestamps=False, # BatchedInferencePipeline 默认 True,需显式关闭以生成段级时间戳 ) logger.debug( "音频时长 %.1fs,检测语言=%s(置信度 %.2f)", info.duration, info.language, info.language_probability, ) segments: list[Segment] = [] for seg in segments_gen: words: list[Word] = [] if s.word_timestamps and getattr(seg, "words", None): for w in seg.words: words.append(Word( text=w.word.strip(), start=float(w.start), end=float(w.end), probability=float(getattr(w, "probability", 1.0)), )) segments.append(Segment( text=seg.text.strip(), start=float(seg.start), end=float(seg.end), words=words, )) logger.debug("转写完成:%d 段,%d 词。", len(segments), sum(len(s.words) for s in segments)) return segments