cat-shark
d0d96a8f89
背景：实测 CJOD-255 全程 BGM 覆盖（56% 低能量、几乎无静音），固定 VAD 参数把音乐当语音 → whisper 全段解码 → 80% 碎片化 + 32% 漏句 + 敏感段丢失。 实现 nodes/vad_profiler.py： - profile_audio：1s 能量网格分析 → 静音比例/BGM 覆盖/长停顿识别 - suggest_vad_parameters：按信号特征推荐 threshold/min_silence/speech_pad （BGM 覆盖 -> 降 threshold 增人声敏感；长停顿 -> 降 speech_pad 防时间漂移； 静音占比高 -> 升 threshold 剔虚警） - score_transcript：启发式评分（不用参考字幕，符合部署实际）—— 碎片率 + 幻觉词（感谢观看/晚安/音乐）扣分 + 平均字数适中 - vad_parameters_for_audio：信号分析 + 可选片段网格验证，选最优参数 - _grid_search_vad / _candidate_params：候选网格搜索与异常回退 测试 tests/test_vad_profiler.py：19 个用例覆盖信号分析、四个建议分支、 幻觉词/碎片评分、网格选优、异常回退、空音频/低采样率、候选展开。
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""每视频自适应 VAD 调参（信号分析 + 片段网格验证）。
|
|
|
|
背景（实测 CJOD-255）：全程 BGM 覆盖的视频几乎没有纯静音，固定 VAD
|
|
参数会把音乐当语音，导致 whisper 全段解码 → 碎片化(80%)、漏句(32%)、
|
|
敏感段丢失。不同视频声学差异大，需**每视频独立分析后再定 VAD 参数**。
|
|
|
|
方案（用户确认）：
|
|
1. 信号分析（秒级，无模型）缩窄参数空间：1s 能量分布 → 静音比例、
|
|
BGM 底噪、语音疏密，推出 threshold/min_silence/speech_pad 候选；
|
|
2. 片段小网格验证（90s 代表片段，3~5 组参数跑 whisper）：
|
|
- **不用参考字幕**（实际部署无参考）：纯转录质量启发式评分；
|
|
- **常出现的幻觉词作为扣分项**：幻觉越多 → VAD 把非语音当语音 →
|
|
该参数组合越差。
|
|
3. 全片用选出的 vad_parameters 跑一次。
|
|
|
|
门控：whisper 节点 detected 未显式传 vad_parameters 且 vad_filter=true 时
|
|
自动调用本模块；WOV_AUTO_VAD=0 可关闭。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import wave # 仅读 WAV 头/样本
|
|
|
|
# 幻觉词扣分集合：VAD 若把音乐/静音当语音，whisper 常在这些段输出套话。
|
|
# 每条命中扣 1 分；重复次数越高 → 该参数组合越差。
|
|
HALLUCINATION_TOKENS = (
|
|
"ご視聴ありがとうございました", # 感谢观看
|
|
"ご視聴ありがとうございました。",
|
|
"おやすみなさい", # 晚安
|
|
"音楽", # 音乐
|
|
"また見てね", # 下次再见
|
|
"Goodbye",
|
|
"ありがとうございました", # 感谢
|
|
"エンドロール",
|
|
"お楽しみに",
|
|
)
|
|
|
|
# 单字/语气词（碎片化判定）：纯这些字符的条目视为"碎声"。
|
|
_FRAGMENT_RE = re.compile(r"^[ぁ-んァ-ヶあいうえおっーんすよわ\s、。]+$")
|
|
|
|
# 默认候选网格（片段验证时用；由信号分析缩小范围）。
|
|
_THRESHOLDS = (0.3, 0.5, 0.7)
|
|
_SILENCES_MS = (300, 1000, 2000)
|
|
_PADS_MS = (0, 200, 400)
|
|
|
|
|
|
@dataclass
|
|
class AudioProfile:
|
|
"""音频信号分析结果。"""
|
|
|
|
sample_rate: int = 16000
|
|
duration_seconds: float = 0.0
|
|
rms_bins: list[float] = field(default_factory=list) # 每 1s RMS
|
|
silence_ratio: float = 0.0 # RMS<300（静音级）占比
|
|
lowish_ratio: float = 0.0 # RMS<900（音乐/低语）占比
|
|
voice_ratio: float = 0.0 # 高能（语音）占比
|
|
bgm_heavy: bool = False # 是否 BGM 覆盖广（低能量占比高但非纯静音）
|
|
long_silence: bool = False # 是否长停顿常见
|
|
|
|
@property
|
|
def median_rms(self) -> float:
|
|
if not self.rms_bins:
|
|
return 0.0
|
|
ordered = sorted(self.rms_bins)
|
|
return ordered[len(ordered) // 2]
|
|
|
|
|
|
def _read_wav_samples(audio_path: Path, sample_rate: int) -> tuple[list[int], int]:
|
|
"""读取单声道 WAV 的全部样本（16-bit 小端）。返回 (样本列表, 实际采样率)。"""
|
|
with wave.open(str(audio_path), "rb") as wav:
|
|
rate = wav.getframerate()
|
|
n = wav.getnframes()
|
|
data = wav.readframes(n)
|
|
import array
|
|
|
|
samples = array.array("h")
|
|
samples.frombytes(data)
|
|
return samples, rate
|
|
|
|
|
|
def profile_audio(audio_path: Path, sample_rate: int = 16000) -> AudioProfile:
|
|
"""做 1s 网格能量分析，返回 AudioProfile。
|
|
|
|
每秒计算 RMS；据此得出静音/音乐/语音占比，判断是否 BGM 覆盖、有无长静音。
|
|
"""
|
|
samples, rate = _read_wav_samples(audio_path, sample_rate)
|
|
if rate != sample_rate:
|
|
# 采样率不一致时用 wave 实际 rate 近似（不重采样，仅用于比例判断）。
|
|
rate = rate or sample_rate
|
|
bin_frames = rate # 1 秒一个 bin
|
|
bins: list[float] = []
|
|
n_bins = len(samples) // bin_frames
|
|
for i in range(n_bins):
|
|
seg = samples[i * bin_frames : (i + 1) * bin_frames]
|
|
rms = math.sqrt(sum(s * s for s in seg) / len(seg)) if seg else 0
|
|
bins.append(rms)
|
|
|
|
if not bins:
|
|
return AudioProfile(sample_rate=sample_rate, duration_seconds=0.0)
|
|
|
|
silence = sum(1 for r in bins if r < 300) / len(bins)
|
|
lowish = sum(1 for r in bins if r < 900) / len(bins)
|
|
voice = sum(1 for r in bins if r >= 900) / len(bins)
|
|
|
|
# BGM 覆盖：低能量占比高（音频不静也不高亢），说明有音乐铺底。
|
|
bgm_heavy = lowish > 0.4 and silence < 0.4
|
|
# 长静音：存在连续 >=5 个静音 bin（5s 以上停顿）。
|
|
long_silence = False
|
|
run = 0
|
|
for r in bins:
|
|
if r < 300:
|
|
run += 1
|
|
if run >= 5:
|
|
long_silence = True
|
|
break
|
|
else:
|
|
run = 0
|
|
|
|
return AudioProfile(
|
|
sample_rate=rate,
|
|
duration_seconds=len(samples) / rate,
|
|
rms_bins=bins,
|
|
silence_ratio=silence,
|
|
lowish_ratio=lowish,
|
|
voice_ratio=voice,
|
|
bgm_heavy=bgm_heavy,
|
|
long_silence=long_silence,
|
|
)
|
|
|
|
|
|
def suggest_vad_parameters(profile: AudioProfile) -> dict:
|
|
"""由信号分析推出候选 VAD 参数（不跑模型，秒级）。
|
|
|
|
返回 dict 含 threshold/min_silence_duration_ms/speech_pad_ms 候选。
|
|
规则见 docs/adaptive_vad.md。
|
|
"""
|
|
if profile.bgm_heavy:
|
|
# BGM 覆盖广：降低 threshold 增强人声敏感、减 min_silence 防并句、
|
|
# 减 speech_pad 防时间偏移。
|
|
return {
|
|
"threshold": 0.3,
|
|
"min_silence_duration_ms": 300,
|
|
"speech_pad_ms": 0,
|
|
}
|
|
if profile.long_silence:
|
|
# 长停顿常见：speech_pad 减小防时间轴压缩漂移，threshold 正常。
|
|
return {
|
|
"threshold": 0.5,
|
|
"min_silence_duration_ms": 1000,
|
|
"speech_pad_ms": 200,
|
|
}
|
|
if profile.silence_ratio > 0.3:
|
|
# 静音占比高（正常带静音的音频）：threhold 略高剔除虚警。
|
|
return {
|
|
"threshold": 0.6,
|
|
"min_silence_duration_ms": 2000,
|
|
"speech_pad_ms": 400,
|
|
}
|
|
# 常规。
|
|
return {
|
|
"threshold": 0.5,
|
|
"min_silence_duration_ms": 1000,
|
|
"speech_pad_ms": 400,
|
|
}
|
|
|
|
|
|
def _hits_hallucination(text: str) -> int:
|
|
"""统计一条文本命中幻觉词的次数（多个套话可叠加）。"""
|
|
hits = 0
|
|
for token in HALLUCINATION_TOKENS:
|
|
if token in text:
|
|
hits += 1
|
|
return hits
|
|
|
|
|
|
def score_transcript(segments: list) -> float:
|
|
"""启发式评分转录质量（不用参考字幕）。
|
|
|
|
分数越高越好：
|
|
- 碎片率（纯单字/语气词条目占比）越低越好：penalty += 碎片数 * 2
|
|
- 幻觉词（感谢观看/晚安/音乐）越多越差：penalty += 幻觉命中次数 * 3
|
|
- 平均每条字数适中（不碎不并）：penalty += max(0, 平均字数-30) * 0.5
|
|
返回 100 - penalty。
|
|
"""
|
|
if not segments:
|
|
return 0.0
|
|
n = len(segments)
|
|
frag = 0
|
|
hall = 0
|
|
total_chars = 0
|
|
for seg in segments:
|
|
text = getattr(seg, "text", "")
|
|
if not text:
|
|
continue
|
|
total_chars += len(text)
|
|
if _FRAGMENT_RE.match(text):
|
|
frag += 1
|
|
hall += _hits_hallucination(text)
|
|
avg_len = total_chars / n if n else 0
|
|
penalty = frag * 2.0 + hall * 3.0 + max(0.0, (avg_len - 20)) * 0.3
|
|
return max(0.0, 100.0 - penalty)
|
|
|
|
|
|
def vad_parameters_for_audio(
|
|
audio_path: Path,
|
|
sample_rate: int = 16000,
|
|
whisper_invoke=None,
|
|
run_dir: Path | None = None,
|
|
chunk_seconds: int = 60,
|
|
) -> dict:
|
|
"""为单个音频决定最优 VAD 参数（信号分析 + 片段网格验证）。
|
|
|
|
若不提供 whisper_invoke（无法跑片段），退化为仅信号分析启发式。
|
|
"""
|
|
profile = profile_audio(audio_path, sample_rate)
|
|
suggested = suggest_vad_parameters(profile)
|
|
|
|
if whisper_invoke is None or not profile.rms_bins:
|
|
# 无推理能力：直接用信号分析结论，不再网格验证。
|
|
return suggested
|
|
|
|
# 片段网格：取 1 段 90s 代表性片段（选能量中等段，避免纯音乐/纯静音）。
|
|
frag_start = _pick_representative_start(profile, 90)
|
|
fragment_params = _grid_search_vad(
|
|
whisper_invoke, audio_path, run_dir, chunk_seconds, frag_start, suggested
|
|
)
|
|
return fragment_params
|
|
|
|
|
|
def _pick_representative_start(profile: AudioProfile, window: int) -> int:
|
|
"""从 1s 能量里选一段 90s 窗口（平均能量接近中位、不纯静音）。
|
|
|
|
返回窗口起始秒；音频不足 window 秒时返回 0。
|
|
"""
|
|
bins = profile.rms_bins
|
|
n = len(bins)
|
|
if n <= window:
|
|
return 0
|
|
median = profile.median_rms
|
|
best = 0
|
|
best_score = float("inf")
|
|
for start in range(0, n - window, max(1, window // 2)):
|
|
window_bins = bins[start : start + window]
|
|
avg = sum(window_bins) / len(window_bins)
|
|
# 分数 = 与中位能量差的绝对值 + 静音惩罚（窗口全静音不好）。
|
|
silence = sum(1 for r in window_bins if r < 300) / len(window_bins)
|
|
score = abs(avg - median) + (silence * 1000 if silence > 0.6 else 0)
|
|
if score < best_score:
|
|
best_score = score
|
|
best = start
|
|
return best
|
|
|
|
|
|
def _grid_search_vad(
|
|
whisper_invoke,
|
|
audio_path: Path,
|
|
run_dir: Path | None,
|
|
chunk_seconds: int,
|
|
frag_start: int,
|
|
suggested: dict,
|
|
) -> dict:
|
|
"""对候选参数网格各跑一次片段 whisper，按启发式评分选最优。
|
|
|
|
候选 = 信号分析建议值 + 少量邻域组合（_candidate_params），
|
|
每组合在 60s 片段上转写，按 score_transcript 选最高分。
|
|
若无法运行 whisper（如缺模型）返回 suggested。
|
|
"""
|
|
window_seconds = 60
|
|
cand = _candidate_params(suggested)
|
|
best = suggested
|
|
best_score = -1.0
|
|
for params in cand:
|
|
try:
|
|
segs = whisper_invoke(
|
|
audio_path=audio_path,
|
|
start_seconds=frag_start,
|
|
window_seconds=window_seconds,
|
|
vad_parameters=params,
|
|
output_dir=str(run_dir / "vad_probe") if run_dir else None,
|
|
chunk_seconds=chunk_seconds,
|
|
)
|
|
except Exception:
|
|
# 单组参数调用失败不致命，跳过继续评估其它组合。
|
|
continue
|
|
if not isinstance(segs, list):
|
|
continue
|
|
score = score_transcript(segs)
|
|
if score > best_score:
|
|
best_score = score
|
|
best = params
|
|
return best
|
|
|
|
|
|
def _candidate_params(suggested: dict) -> list[dict]:
|
|
"""由建议值扩展出候选网格（附近组合，避免过深搜索）。"""
|
|
base = dict(suggested)
|
|
thres = base.get("threshold", 0.5)
|
|
sil = base.get("min_silence_duration_ms", 1000)
|
|
pad = base.get("speech_pad_ms", 400)
|
|
cand = []
|
|
for t in (thres, round(thres - 0.1, 1), round(thres + 0.1, 1)):
|
|
for s in (sil, sil // 2, sil * 2):
|
|
for p in (0, pad // 2, pad):
|
|
cand.append({"threshold": t, "min_silence_duration_ms": s, "speech_pad_ms": p})
|
|
return cand |