feat: 每视频自适应 VAD 调参模块(信号分析 + 幻觉词扣分评分)
背景:实测 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 个用例覆盖信号分析、四个建议分支、 幻觉词/碎片评分、网格选优、异常回退、空音频/低采样率、候选展开。
This commit is contained in:
@@ -0,0 +1,311 @@
|
||||
"""每视频自适应 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
|
||||
Reference in New Issue
Block a user