From d9219b9eb11fe09281551c76c05faee292fbdf7c Mon Sep 17 00:00:00 2001 From: KISEKI Date: Mon, 6 Jul 2026 23:50:25 +0800 Subject: [PATCH] Initial commit: mlx-whisper enhanced transcription with hallucination detection and cold-start retry --- .env.example | 3 + .gitignore | 21 + README.md | 104 ++++ SKILL.md | 85 ++++ references/empty-segment-gap.md | 59 +++ references/subtitle-timing-deep-dive.md | 162 +++++++ references/token-repetition-loop.md | 127 +++++ scripts/whisper_transcribe.py | 611 ++++++++++++++++++++++++ 8 files changed, 1172 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 SKILL.md create mode 100644 references/empty-segment-gap.md create mode 100644 references/subtitle-timing-deep-dive.md create mode 100644 references/token-repetition-loop.md create mode 100644 scripts/whisper_transcribe.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e6257ae --- /dev/null +++ b/.env.example @@ -0,0 +1,3 @@ +# Whisper Transcribe — Environment Variables +# This script is self-contained; no external API keys required. +# mlx-whisper runs locally via Apple Silicon (MLX framework). diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ec76f8d --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +# macOS +.DS_Store + +# Python +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +.mypy_cache/ +.pytest_cache/ +.ruff_cache/ +dist/ + +# Test / generated artifacts +*.wav +*.mp3 +*.m4a +*.webm +*.txt +*.srt +*.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..1805306 --- /dev/null +++ b/README.md @@ -0,0 +1,104 @@ +# Whisper Transcribe + +[mlx-whisper](https://github.com/ml-explore/mlx-examples/tree/main/whisper) 增强版字幕生成工具。 +Apple Silicon 本地运行，无需 GPU 服务器或第三方 API。 + +## 特性 + +- **本地推理** — MLX 框架，Apple Silicon 原生加速，无外部 API 依赖 +- **逐字时间轴** — 每个词的精确起止时间，支持智能断句 +- **静音过滤** — ffmpeg `silencedetect` 预扫描，剔除音乐段的幽灵字幕 +- **幻觉清理** — 检测尾部复读/填充词，自动清除 +- **循环检测** — 检测单字/短句级 token loop，冷启动重转写修复 +- **空段修复** — 补回 Whisper 漏转写的空白段 +- **多种输出** — txt / srt / json + +## 依赖 + +- macOS (Apple Silicon) +- [mlx-whisper](https://pypi.org/project/mlx-whisper/) +- ffmpeg / ffprobe + +```bash +pip install mlx-whisper +brew install ffmpeg +``` + +## 用法 + +```bash +python3 whisper_transcribe.py <音频文件> \ + --initial-prompt "<提示词>" \ + --language ja \ + --output-format all +``` + +### 参数 + +| 参数 | 默认 | 说明 | +|------|------|------| +| `--model` | `turbo` | turbo / small / medium | +| `--language` | `ja` | 语言代码 | +| `--output-format` | `txt` | txt / srt / json / all | +| `--initial-prompt` | **必填** | 提示词，按场景列出关键词 | +| `--max-line-duration` | `5.0` | 单行字幕最大秒数 | +| `--no-word-ts` | off | 禁用逐字时间轴 | +| `--no-smart-split` | off | 禁用智能断句 | +| `--no-vad-filter` | off | 禁用静音过滤 | +| `--no-clean` | off | 禁用幻觉清理 | +| `--verbose` | off | 显示进度 | + +### 提示词 + +提示词决定转写质量。每次根据音频内容从零构建，不套模板。 + +**构建方法：** 从标题提取节目名、期号、嘉宾、主题 → 按场景补充术语 → 写成逗号分隔的 10-20 个关键词。 + +```bash +# 日语播客 +--initial-prompt "野良犬ラジオ、浅見春那、結婚式、ゲスト、トーク番組、声優" + +# 音乐现场 MC +--initial-prompt "ライブ、MC、バンド、メンバー紹介、曲紹介、観客、拍手" +``` + +> 专有名词务必写进 prompt。prompt 能大幅提高准确率，但不能完全覆盖非常用汉字。 + +### 视频预处理 + +```bash +ffmpeg -y -i 输入.webm \ + -af "loudnorm=I=-16:TP=-1.5:LRA=11" \ + -ar 16000 -ac 1 -f wav 输出.wav +``` + +## 输出 + +默认与输入文件同目录。支持格式： + +- **txt** — 纯文本，每行一段 +- **srt** — 标准字幕格式，含序号和时间轴 +- **json** — 完整数据结构，含逐字时间轴 +- **all** — 同时输出以上三种 + +## 字幕质量增强 + +| 功能 | 原理 | +|------|------| +| **逐字时间轴** | `word_timestamps=True` | +| **智能断句** | 词间停顿 >0.3s 拆分，`max_line_duration` 强制切断 | +| **静音过滤** | ffmpeg 预扫静音区，70%+ 在静音内的段丢弃 | +| **幻觉清理** | 从尾部向前检测重复段和填充词 | +| **循环检测** | 单字频率 / 字符串周期性 / n-gram 去重率，冷启动重转写 | +| **空段修复** | 15 秒+长空白且非 VAD 静音区才重试，仍为空则删除 | + +## 技术细节 + +- 内部使用 `mlx-community/whisper-large-v3-turbo`（fp16，~1.5GB） +- mlx-whisper 自带滑动窗口，无需外部手动分段 +- 预处理自动做 16kHz 降采样 + 响度标准化 +- 逆序遍历替换区间，防止索引漂移 Bug + +## License + +MIT diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..d80d520 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,85 @@ +--- +name: whisper-transcribe +description: "使用 mlx-whisper（Apple Silicon 优化）转写音频。支持逐字时间轴、智能断句、静音过滤、幻觉清理。提示词必填，按场景构建。" +--- + +# Whisper Transcribe (mlx-whisper) + +## 基本用法 + +```bash +PYTHONPATH="" ~/.local/venvs/default/bin/python3 \ + ~/.hermes/skills/productivity/whisper-transcribe/scripts/whisper_transcribe.py \ + <音频> --initial-prompt "<提示词>" --language ja --output-format all +``` + +> Hermes TUI 环境需加 `PYTHONPATH=""`，否则 Hermes 自身 venv 的 numpy C 扩展与 Python 3.12 不兼容。 + +视频先提音频再转写更稳定： +```bash +ffmpeg -y -i 输入.webm -af "loudnorm=I=-16:TP=-1.5:LRA=11" -ar 16000 -ac 1 -f wav 输出.wav +PYTHONPATH="" ~/.local/venvs/default/bin/python3 \ + ~/.hermes/skills/productivity/whisper-transcribe/scripts/whisper_transcribe.py \ + 输出.wav --initial-prompt "..." --language ja --output-format all +``` + +## 参数 + +| 参数 | 默认 | 说明 | +|------|------|------| +| `--model` | `turbo` | turbo / small / medium | +| `--language` | `ja` | 语言代码 | +| `--output-format` | `txt` | txt / srt / json / all | +| `--initial-prompt` | **必填** | 提示词，按场景列出关键词 | +| `--max-line-duration` | `5.0` | 单行字幕最大秒数 | +| `--no-word-ts` | off | 禁用逐字时间轴 | +| `--no-smart-split` | off | 禁用智能断句 | +| `--no-vad-filter` | off | 禁用静音过滤 | +| `--no-clean` | off | 禁用幻觉清理 | +| `--verbose` | off | 显示进度 | + +## 字幕质量增强（默认开启） + +| 功能 | 作用 | 原理 | +|------|------|------| +| **逐字时间轴** | 获得每个词的精确起止时间 | `word_timestamps=True` | +| **智能断句** | 修复黏连（一句字幕几秒不消失） | 检测词间停顿 >0.3s，达 `--max-line-duration` 强制切断 | +| **静音过滤** | 剔除音乐段的幽灵字幕 | ffmpeg `silencedetect` 预扫描，字幕 70% 以上在静音区则丢弃 | +| **幻觉清理** | 删除尾部复读幻觉 | 检测末尾连续重复 / 填充词 | +| **循环检测** | 修复单字 & 短句 token loop | 单字频率 + 字符串周期性检测，冷启动重转写；逆序遍历防索引漂移 [详情](references/token-repetition-loop.md) | +| **空段修复** | 补回 Whisper 漏转写的空白段 | 检测空文本段（≥2s），冷启动重转写；重试仍为空则删除 [详情](references/empty-segment-gap.md) | + +## 提示词定制 + +**每次根据音频内容从零构建，不要套模板。** 决定转写质量最关键的一步。 + +### 构建方法 + +1. 从标题提取：节目名、期号、嘉宾、主题关键词 +2. 按场景补充术语：声优访谈→作品/角色名，技术讲座→技术栈，音乐现场→乐队/曲名 +3. 写成逗号分隔的 10-20 个关键词 + +### 示例 + +```bash +# 日语播客（嘉宾浅見春那，主题婚礼） +--initial-prompt "野良犬ラジオ、浅見春那、結婚式、あいのなかで、ゲスト、トーク番組、声優、川瀬光平、パーソナリティ、対談" + +# 中文技术会议 +--initial-prompt "项目复盘、技术评审、架构设计、前端、后端、API、数据库、性能优化" + +# 日语音乐现场 MC +--initial-prompt "ライブ、MC、バンド、メンバー紹介、曲紹介、挨拶、観客、拍手" +``` + +> 专有名词（人名、乐队名）务必写进 prompt。例如「浅見春那」无 prompt → 「浅見春菜」。 +> **注意**：prompt 能大幅提高准确率，但非常用汉字（如「那」）仍可能被 Whisper 的声学模型覆盖——prompt 是引导而非强制。对此类人名需在后期手动修正。 + +## 关键技巧 + +- mlx-whisper 内部自带滑动窗口，长音频自动稳定处理，无需手动分段 +- 脚本自动做 16kHz 降采样 + 响度标准化 + 幻觉清理 +- 默认输出 SRT 已智能断句 + 静音过滤，可直接导入剪辑软件 +- 修改脚本注意陷阱：浮点取整用 `int(round(x * 1000))`、silencedetect 结尾配对不要用 `zip()`、除零防护、负数时间戳 clamp +- [Token Repetition Loop 诊断修复](references/token-repetition-loop.md) — 长音频中单字循环幻觉的根因与局部重转写方案 +- [Empty Segment Gap 诊断修复](references/empty-segment-gap.md) — Whisper 漏转写空白段的根因与修复方案 diff --git a/references/empty-segment-gap.md b/references/empty-segment-gap.md new file mode 100644 index 0000000..f76db2f --- /dev/null +++ b/references/empty-segment-gap.md @@ -0,0 +1,59 @@ +# Empty Segment Gap（空白段漏转写） + +## 症状 + +全长转写结果中，某一段（几秒到几十秒）输出为**空文本**： + +```json +{"start": 1376.42, "end": 1406.42, "text": ""} +``` + +该段音频实际包含清晰的对话，单独提取转写完全正常。 + +## 根因 + +与 [token repetition loop](token-repetition-loop.md) 同一类问题——**Whisper autoregressive decoder 的边界故障**，但表现不同： + +- 循环幻觉：低置信度边界 → 猜错第一个 token → 自激放大 → 重复输出 +- 空段：低置信度边界 → decoder 全程置信度不足 → 输出空字符串 + +两种故障都发生在长音频中后段（20+ 分钟），此时 KV cache 积累了大量上下文，在模糊发音、犹豫、或者声学域偏移（如不同录音源的拼接点）处模型失去对齐。 + +## 修复方案 + +与 token repetition loop 相同——**检测 → 提取 → 冷启动重转写**： + +### 检测 + +```python +def find_empty_gaps(segments: list[dict], min_duration: float = 1.0) -> list[int]: + """返回所有空文本且时长超过 min_duration 的 segment 索引""" + return [ + i for i, s in enumerate(segments) + if not s.get("text", "").strip() and s["end"] - s["start"] > min_duration + ] +``` + +### 合并连续空段 + +连续的空段合并为一个时间范围，避免对同一片音频发起多次重转写。 + +### 冷启动重写 + +ffmpeg 切出该段 → `mlx_whisper.transcribe()` 不带前序上下文 → 如果重转写得到有效文本，替换回原结果。 + +**如果重转写也无输出**（说明该段确实是静音/纯噪），则直接删除该段，不保留原始空字幕。这与旧版"保留 originals"的做法不同——避免静音段冷启动逼出幻听。 + +## 为什么 VAD 过滤没抓到 + +VAD（静音过滤）检查的是 segments 是否落在 ffmpeg `silencedetect` 标记的**静音区**。空段大多不是静音（有音乐、背景音甚至人声），只是 Whisper 的 decoder 没有输出文字——所以不会被 VAD 丢弃。 + +## 在脚本中的实现 + +空段检测已合并进 `fix_repetition_loops()` 函数（Mode B），与循环检测共享重转写逻辑骨架。位于 `whisper_transcribe.py` 的 Post-processing 阶段，在 `--no-clean` 未设置时运行。 + +核心行为： +- 空段 ≥2s → 标记为异常，冷启动重试 +- 重试成功 → 用新 segments 替换 +- 重试仍为空 → `del` 删除该段（不入库） +- 使用 `reversed(ranges)` 逆序遍历，防止多区间替换时的索引漂移 diff --git a/references/subtitle-timing-deep-dive.md b/references/subtitle-timing-deep-dive.md new file mode 100644 index 0000000..dbd0b7d --- /dev/null +++ b/references/subtitle-timing-deep-dive.md @@ -0,0 +1,162 @@ +# 字幕时间轴质量深度解析 + +本文档记录从 PyTorch 方案（stable-ts / whisper-timestamped）向 MLX 增强方案迁移的决策过程与技术细节。 + +## 为什么不用 stable-ts / whisper-timestamped + +| 框架 | 底层依赖 | 与 mlx-whisper 兼容性 | +|---|---|---| +| **stable-ts** | PyTorch + `openai-whisper` | ❌ 不兼容 | +| **whisper-timestamped** | PyTorch + `openai-whisper` | ❌ 不兼容 | +| **mlx-whisper 增强** | MLX 框架 | ✅ 原生兼容 | + +这两个工具都是 PyTorch 生态的插件，直接修改/包装的是 OpenAI 官方 Whisper 模型。mlx-whisper 使用的是 MLX 框架和不同的模型格式，两者推理代码完全不同，无法混用。 + +如果硬要引入 stable-ts，意味着： +- 安装 PyTorch + openai-whisper（~2GB 额外依赖） +- 放弃 Apple Silicon 的 MLX 加速 +- turbo 模型在 PyTorch 上跑得更慢 + +## mlx-whisper 内建方案：word_timestamps + +mlx-whisper 本身支持 `word_timestamps=True`，只是默认关闭。打开后可以获得每个词的精确起止时间，用来： + +1. 更精准的 SRT 断句（不在词中间切断） +2. 检测长停顿，自动拆分句子 +3. 修复"黏连"（一句字幕占好几秒不消失） + +### 出现黏连的根因 + +Whisper 的默认输出是 **segment-level**，一个 segment 可能包含多句话，但只有一个起止时间。当这个 segment 跨度很大时，字幕就会"黏在屏幕上"。 + +解决方法：用 word-level timestamps 重构断句逻辑。 + +## 智能断句算法 + +```python +# 核心逻辑 +for 每个词: + 检测词间停顿 pause = 下一个词.start - 当前词.end + + 如果 pause >= 0.3s 且当前块长度 >= 1.5s: + 在这里切断，形成新的字幕行 + 否则如果当前块长度 >= max_line_duration（默认 5s）: + 强制切断 +``` + +这比 stable-ts 的动态重构交叉注意力机制简单，但在大多数场景下足够有效。 + +## 幽灵字幕的根因与解决 + +**幽灵字幕**指背景音乐/静音期间出现的无意义字幕。 + +Whisper 是一个语音识别模型，它不区分"人声"和"背景音乐"，只是尝试预测文字。在音乐声音中，模型可能会"幻想"出一些文字。 + +### VAD 预处理方案 + +不需要 PyTorch 的 Silero VAD，用 ffmpeg 内置的 `silencedetect` 即可： + +```bash +ffmpeg -i 音频.wav -af silencedetect=noise=-40dB:d=0.5 -f null - +``` + +这会输出所有静音区域的起止时间。转写后，如果某段字幕 70% 以上落在静音区，则自动丢弃。 + +## 手动分段已移除 + +原先脚本里用 ffmpeg 做 10 分钟一段的手动分段 + 重叠合并，后来发现： + +- **mlx-whisper 内部自带滑动窗口机制**，会自动处理长音频，不需要外部切块 +- 手动分段导致 chunk 重叠区的**幻觉**（重复「うんうん」等） +- 额外 I/O 开销（ffmpeg 切 → 写盘 → 读回 → merge） + +移除了 `segment_audio()`、`merge_segments()` 及 `--segment-minutes` / `--overlap-seconds` / `--no-segment` 参数。现在流程更简洁：预处理 → mlx-whisper 整段转写 → 后处理。 + +## 残留依赖清理注意事项 + +如果曾经尝试装过 PyTorch 后又卸载，可能留下空目录： + +```bash +# 检查残留 +ls ~/.local/venvs/default/lib/python*/site-packages/ | grep torch + +# 清理 +rm -rf ~/.local/venvs/default/lib/python*/site-packages/torch \ + ~/.local/venvs/default/lib/python*/site-packages/functorch \ + ~/.local/venvs/default/lib/python*/site-packages/torchgen +``` + +uv pip uninstall 对非标准 wheel 的清理可能不完整，需要手动检查。 + +## 现场对比：Whisper vs Deepgram Nova-2（日语播客 #379） + +29 分钟日语广播节目，同一段音频，分别用 mlx-whisper turbo（有 prompt）和 Deepgram Nova-2（无提示）转写： + +| 对比项 | Deepgram (Nova-2, 裸跑) | Whisper (mlx turbo, 有 prompt) | +|---|---|---| +| 分段数 | 296 段（偏长） | 635 段（偏细，智能断句后） | +| 首段开始 | 41.67s（错过片头音乐） | 8.20s（抓到片头标签） | +| 末尾幻觉 | 无 | 有「ご視聴ありがとうございました」重复 | +| 标点符号 | ✅ 有、。 | ❌ 基本没有 | +| 嘉宾名 | ❌ 浅見**春奈**（错字） | ✅ 浅見**春菜**（prompt 起作用） | +| 整体识别率 | 很高，自然流畅 | 高，但有小错 | + +**结论：** 在人声清晰的广播场景下，mlx-whisper turbo + promt 并不比 Deepgram Nova-2 差。Deepgram 的优势在嘈杂环境、多人重叠、带 BGM 的现场录音。（注意：Nova-2 不支持 keyterm 提示，需 Nov-3 才有。） + +## 模型量化信息（mlx-whisper） + +`mlx-community/whisper-large-v3-turbo` 是 **fp16（float16 半精度）**，约 1.5GB。 + +| 量化 | 大小 | 说明 | +|---|---|---| +| 原始 fp32 | ~3GB | OpenAI 原版 | +| MLX fp16 | 1.5GB | Apple Silicon native 格式，计算效率最高 | +| 4-bit | ~400MB | 精度损失大，MLX 社区很少出 | + +Apple Silicon 上 fp16 是原生格式，不需要进一步量化。想要更小模型应换 `small`（~500MB）或 `medium`（~1.5GB）。 + +## 快速诊断清单 + +当用户抱怨字幕质量时，按以下流程排查： + +1. **黏连**：检查是否开启了 `word_timestamps` 和 `smart_split` +2. **幽灵字幕**：检查是否开启了 `vad_filter` +3. **幻觉复读**：检查是否开启了 `clean` +4. **断句不自然**：调整 `max_line_duration` 参数 +5. **识别准确度低**：检查 `initial_prompt` 是否包含专有名词 + +## 代码审查修复记录（2026-07-06） + +一轮完整代码审查发现并修复了 4 个问题： + +### Bug 1：浮点精度导致 SRT 毫秒偏移（🔴 高影响） + +**问题：** `format_timestamp_srt` 用 `int((seconds % 1) * 1000)` 提取毫秒。Python 浮点精度导致 `8.2 % 1 = 0.1999...`，`int()` 截断后变成 `199` 而不是 `200`。实测 642 段字幕中 623 段受影响（偏差 1ms）。 + +**修复：** 改用 `int(round(seconds * 1000))` 先转总毫秒再取模。 + +### Bug 2：detect_silence 尾部静音丢失（🔴 高影响） + +**问题：** 音频以静音结尾时，ffmpeg 只输出 `silence_start` 没有 `silence_end`。原代码用 `zip(starts, ends)` 配对，`zip` 静默截断不匹配项，导致最后一段静音被丢弃，VAD 过滤失效。 + +**修复：** 改用索引配对，未闭合的 `silence_start` 用 `audio_duration` 参数补上结尾。 + +### Bug 3：format_timestamp_srt 负数乱码（🟡 防御性） + +**问题：** `-0.1` 输出 `-1:59:59,900`。Whisper 正常不返回负数，但不防御。 + +**修复：** 开头加 `if seconds < 0: seconds = 0.0`。 + +### Bug 4：remove_vad_gaps 除零风险（🟡 防御性） + +**问题：** 零长度段（`start == end`）时 `overlap / seg_len` 除零崩溃。 + +**修复：** 加 `if seg_len <= 0: continue`。 + +### 清理：删除未使用的 `import math` + +### 教训 + +- **浮点取整陷阱：** 任何 `int(float_val * N)` 都应该用 `int(round(float_val * N))` 替代，避免 IEEE 754 精度问题 +- **zip 配对陷阱：** 当两个列表长度可能不匹配时，不要用 `zip()`，用索引配对并处理 unmatched 项 +- **代码审查应包含边界值测试：** 负数、零、空列表、unmatched 输入 diff --git a/references/token-repetition-loop.md b/references/token-repetition-loop.md new file mode 100644 index 0000000..3fac320 --- /dev/null +++ b/references/token-repetition-loop.md @@ -0,0 +1,127 @@ +# Token Repetition Loop（循环幻觉） + +## 症状 + +全长转写中，某一段（几秒到几十秒）陷入自激循环，有两种表现： + +**单字循环**——反复输出同一个字符： +``` +23:31.420 --> 23:59.760 秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀秀 +``` + +**短句循环**——反复念叨同一个词组： +``` +05:12.100 --> 05:25.300 谢谢大家谢谢大家谢谢大家谢谢大家谢谢大家谢谢大家 +``` + +单字循环可通过字符频率检测（占比 >65%），但短句循环（如"谢谢大家"每个字只出现 25%）会漏网——需增加周期性检测。 + +而将问题段音频**单独提取转写**时，识别结果完全正确。 + +## 根因 + +### 触发条件 + +Whisper 的 autoregressive decoder 在**低置信度边界**处进入 token 自激循环： + +1. 前一段结尾是犹豫、拖音或模糊发音（如「え…」） +2. KV cache 跨窗口传递时，这个低置信度状态污染了下一个处理窗口的初始预测 +3. 第一个 token 猜错（如「秀」的声学嵌入恰好匹配了边界特征） +4. 自激放大：输出「秀」→ 下一帧看到「秀」+ 模糊声学特征 → 继续输出「秀」→ ... + +持续到下一个清晰的语音边界才打断循环。 + +### 为什么单独跑片段就对了 + +**冷启动。** 没有前面 20+ 分钟的 KV cache 积累，模型在这个窗口开头没有低置信度的上下文污染，直接对准了语音。 + +### 为什么 turbo 模型更容易触发 + +`large-v3-turbo` 为了速度减少了 decoder 层数（4 层 vs large-v3 的 32 层），decoder 的纠错能力弱于完整模型，在模糊边界处进入循环的概率更高。 + +## 修复方案 + +不做手动分段（会引入切分边界丢字和重叠区幻觉），而是**后处理检测 + 局部重转写**： + +### 检测 + +两种并行的检测机制，覆盖单字和短句两种循环形态： + +**A1. 单字频率检测**——某字符占比超过 `char_threshold`（默认 0.7）： + +```python +if len(text) >= 5: + most_common = max(set(text), key=text.count) + if text.count(most_common) / len(text) > 0.7: + bad_indices.append(i) # → 判定为单字循环 +``` + +**A2. 短句周期性检测**——字符串由某个子串重复构成： + +```python +def is_phrase_repetition(text: str) -> bool: + """检测短语/词组级别的幻听死循环""" + t = text.strip() + if len(t) < 6: + return False + # (t+t).find(t,1) < len(t) → t 由子串周期性重复 + pos = (t + t).find(t, 1) + if pos != -1 and pos < len(t): + return True + # 前后两半完全相同或包含关系 + mid = len(t) // 2 + if t[:mid] == t[mid:] or (mid >= 3 and t[:mid] in t[mid:]): + return True + return False +``` + +例如 "谢谢大家谢谢大家谢谢大家"（12 字）： +- 单字检测：'谢' 出现 3/12 = 25%，远低于 70% 阈值 → 不触发 +- 周期性检测：`(t+t).find(t, 1)` = 3，3 < 12 → ✅ 命中 + +### 合并 + +把连续受影响的段合并成一个时间范围（`start` 到 `end`）。 + +### 局部重转写 + +用 ffmpeg 切出该段音频，调用 `transcribe()` 冷启动。关键流程： + +1. 从原始音频 `ffmpeg -ss/-to` 切片 + 响度标准化（`loudnorm` + 16kHz mono） +2. `mlx_whisper.transcribe()` 不带前序上下文（冷启动） +3. 重转写结果做二次过滤：空段丢弃、依然循环的文本也丢弃 +4. 重转写无有效输出 → 删除该段（静音/纯噪，不保留原始幻听） + +### 替换与索引安全 + +**⚠️ 关键陷阱：索引漂移（Index Shifting）** + +重转写后的 segments 数量可能与原始不同（2 个坏段 → 5 个好段，或 0 个好段 → 删除整段）。如果正向遍历 ranges，第一次替换改变了列表长度，后续 ranges 的索引全部错位——会覆盖到正确的字幕。 + +**必须逆序遍历**：`for seg_start_idx, seg_end_idx in reversed(ranges):` + +尾部改动不影响前面尚未处理的元素索引，彻底消灭索引漂移。 + +## 工程陷阱总结 + +| 陷阱 | 表现 | 修复 | +|------|------|------| +| **索引漂移** | 多个幻听区间替换后字幕张冠李戴 | `reversed(ranges)` 逆序遍历 | +| **短句循环漏检** | "谢谢大家谢谢大家" 不触发单字检测 | `is_phrase_repetition()` 周期性检测 | +| **空段重试反噬** | 静音段冷启动重写逼出幻听 | 重试无结果→ `del` 删除而非保留 | +| **ffprobe 冗余** | 循环内反复 `get_audio_duration()` | 外部传入 `audio_duration` 参数 | + +## 适用场景与限制 + +| 场景 | 效果 | +|------|------| +| 英文/日文清晰对谈 | ✅ 几乎 100% 修复 | +| 广告/赞助商段（BGM + 不同播音员） | 依赖音频复杂度 | +| 纯音乐/大量噪声 | 可能再次出现循环 | + +## 与手动分段的对比 + +| 方案 | 优点 | 缺点 | +|------|------|------| +| 手动分段（已移除） | 每段冷启，无循环 | 切分边界丢字、重叠区幻觉、额外 I/O | +| 后处理检测 + 局部重转写 | 只在必要时冷启，不影响全长 | 增加了 ~30s 的处理时间（调用重转写时） | diff --git a/scripts/whisper_transcribe.py b/scripts/whisper_transcribe.py new file mode 100644 index 0000000..17572d2 --- /dev/null +++ b/scripts/whisper_transcribe.py @@ -0,0 +1,611 @@ +#!/usr/bin/env python3 +""" +MLX Whisper enhanced transcription with precise subtitle timing. +Uses mlx-whisper (Apple Silicon optimized) + word-level timestamps + smart split. + +Usage: + python3 whisper_transcribe.py --initial-prompt "<提示词>" [选项] + +Options: + --model MODEL mlx-whisper model: turbo/small/medium (default: turbo) + --language LANG Language code (ja/en/zh/...) + --output-dir DIR Output directory (default: same as input) + --output-format FMT Output format: txt/srt/json/all (default: txt) + --no-clean Disable hallucination cleanup + --no-word-ts Disable word-level timestamps (revert to segment-level) + --no-smart-split Disable smart split on pauses (default: on) + --no-vad-filter Disable silence/VAD filtering (default: on) + --max-line-duration Max duration per subtitle line in seconds (default: 5.0) + --initial-prompt TEXT Initial prompt for context guidance (required) + --verbose Show progress +""" + +import argparse +import json +import os +import re +import subprocess +import sys +import time +from contextlib import contextmanager +from pathlib import Path +from typing import Optional + +import tempfile as tf_module +import shutil + +# mlx-whisper model repo mapping +MLX_MODELS = { + "turbo": "mlx-community/whisper-large-v3-turbo", + "large-v3-turbo": "mlx-community/whisper-large-v3-turbo", + "small": "mlx-community/whisper-small-mlx", + "medium": "mlx-community/whisper-medium-mlx", +} + + +# ── Audio Preprocessing ────────────────────────────────────────────────────── + +def preprocess_audio(audio_path: str, output_dir: str) -> str: + """Preprocess audio: downsample to 16kHz mono + loudness normalization.""" + out_path = os.path.join(output_dir, "_preprocessed.wav") + cmd = [ + "ffmpeg", "-y", "-i", str(audio_path), + "-af", "loudnorm=I=-16:TP=-1.5:LRA=11", + "-ar", "16000", "-ac", "1", + "-f", "wav", out_path, + ] + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError(f"ffmpeg preprocessing failed: {result.stderr}") + return out_path + + +def get_audio_duration(audio_path: str) -> float: + """Get audio duration in seconds using ffprobe.""" + cmd = [ + "ffprobe", "-v", "quiet", "-print_format", "json", + "-show_format", str(audio_path) + ] + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError(f"ffprobe failed: {result.stderr}") + info = json.loads(result.stdout) + return float(info["format"]["duration"]) + + +def detect_silence(audio_path: str, noise_db: int = -40, min_duration: float = 0.3, + audio_duration: float = None) -> list[tuple[float, float]]: + """Detect silent regions using ffmpeg silencedetect. + Returns list of (start, end) tuples for silent segments. + If audio ends in silence, ffmpeg emits silence_start without silence_end; + we use audio_duration to close the last unclosed silence region.""" + cmd = [ + "ffmpeg", "-i", str(audio_path), + "-af", f"silencedetect=noise={noise_db}dB:d={min_duration}", + "-f", "null", "-" + ] + result = subprocess.run(cmd, capture_output=True, text=True) + starts = re.findall(r"silence_start: ([\d.]+)", result.stderr) + ends = re.findall(r"silence_end: ([\d.]+)", result.stderr) + + starts = [float(s) for s in starts] + ends = [float(e) for e in ends] + + silences = [] + # Pair starts with ends; if there's an unmatched start at the end, + # close it with audio_duration + for i, s in enumerate(starts): + if i < len(ends): + silences.append((s, ends[i])) + elif audio_duration: + # Unclosed silence at end of audio + silences.append((s, audio_duration)) + + return silences + + +# ── Transcription ──────────────────────────────────────────────────────────── + +def transcribe_file(audio_path: str, model_repo: str, language: str, + initial_prompt: Optional[str] = None, + word_timestamps: bool = True, + verbose: bool = False) -> dict: + """Transcribe a single audio file with mlx-whisper.""" + from mlx_whisper import transcribe + + options = { + "path_or_hf_repo": model_repo, + "language": language, + "verbose": verbose, + "word_timestamps": word_timestamps, + } + if initial_prompt: + options["initial_prompt"] = initial_prompt + + return transcribe(audio_path, **options) + + +# ── Post-processing: Smart Split & Glue Fix ────────────────────────────────── + +def smart_split_segments(segments: list[dict], max_duration: float = 5.0, + min_pause: float = 0.3) -> list[dict]: + """Split long segments on natural pauses using word-level timestamps. + Fixes subtitle 'glue' where one line sticks on screen too long.""" + result = [] + + for seg in segments: + words = seg.get("words", []) + if not words: + result.append(seg) + continue + + # If segment is short enough, keep as-is + if seg["end"] - seg["start"] <= max_duration: + result.append(seg) + continue + + # Split on pauses between words + current_words = [] + current_start = words[0]["start"] + + for i, w in enumerate(words): + current_words.append(w) + + # Check for pause after this word + pause = 0.0 + if i < len(words) - 1: + pause = words[i + 1]["start"] - w["end"] + + # Split condition: pause long enough OR current chunk exceeds max_duration + chunk_duration = w["end"] - current_start + should_split = (pause >= min_pause and chunk_duration >= 1.5) or \ + chunk_duration >= max_duration + + if should_split and current_words: + text = "".join(wd["word"] for wd in current_words).strip() + if text: + result.append({ + "start": current_start, + "end": w["end"], + "text": text, + "words": current_words[:], + }) + current_words = [] + if i < len(words) - 1: + current_start = words[i + 1]["start"] + + # Flush remaining words + if current_words: + text = "".join(wd["word"] for wd in current_words).strip() + if text: + result.append({ + "start": current_start, + "end": current_words[-1]["end"], + "text": text, + "words": current_words[:], + }) + + return result + + +def remove_vad_gaps(segments: list[dict], silences: list[tuple[float, float]], + min_gap: float = 0.5) -> list[dict]: + """Remove segments that fall entirely within detected silent regions. + Prevents 'ghost subtitles' during music-only sections.""" + if not silences: + return segments + + def in_silence(start: float, end: float) -> bool: + for s_start, s_end in silences: + # If segment is mostly inside a silence region + overlap_start = max(start, s_start) + overlap_end = min(end, s_end) + if overlap_end > overlap_start: + overlap = overlap_end - overlap_start + seg_len = end - start + if seg_len <= 0: + continue # Skip zero-length segments + if overlap / seg_len > 0.7: # 70% or more in silence + return True + return False + + filtered = [] + for seg in segments: + if not in_silence(seg["start"], seg["end"]): + filtered.append(seg) + return filtered + + +def clean_hallucinations(segments: list[dict]) -> list[dict]: + """Remove trailing hallucinated repetitions.""" + if not segments: + return segments + + texts = [s["text"] for s in segments] + cut_point = len(texts) + seen_repeats = 0 + max_repeats = 5 + + for i in range(len(texts) - 1, 0, -1): + if texts[i] == texts[i - 1] and texts[i].strip(): + seen_repeats += 1 + if seen_repeats >= max_repeats: + cut_point = i + break + else: + seen_repeats = 0 + + filler_pattern = re.compile(r'^(あ+|啊+|um+|uh+|er+|ah+)$', re.IGNORECASE) + filler_count = 0 + for i in range(len(texts) - 1, max(cut_point - 10, 0), -1): + if filler_pattern.match(texts[i].strip()): + filler_count += 1 + else: + break + + if filler_count > 3: + cut_point = len(texts) - filler_count + + return segments[:cut_point] + + +def is_phrase_repetition(text: str) -> bool: + """检测短语/词组级别的幻听死循环。 + 只对大段重复敏感：精确周期需 ≥10 次重复，n-gram 需 ≥30 字符。 + 短文本如「うんうんうん」「うふふふ」正常放行。""" + t = text.strip() + if len(t) < 6: + return False + # 精确周期性检测: 如果 t 由某个子串重复多次构成 + pos = (t + t).find(t, 1) + if pos != -1 and pos < len(t): + # 至少重复 10 次以上才算循环（避免 "うん"×3 这种正常相槌） + if len(t) // pos >= 10: + return True + # 宽松检测：前后两半完全相同或包含关系（仅对较长文本） + if len(t) >= 16: + mid = len(t) // 2 + if t[:mid] == t[mid:] or (mid >= 3 and t[:mid] in t[mid:]): + return True + # 近似周期检测：对付长度不对齐的循环（如 "ABC"*55 + "AB"） + # n-gram 去重率极低 → 文本由少数短语反复拼接 + if len(t) >= 30: + for n in (3, 4, 5): + grams = [t[i:i + n] for i in range(len(t) - n + 1)] + unique_ratio = len(set(grams)) / len(grams) + if unique_ratio < 0.15: + return True + return False + + +def fix_repetition_loops(segments: list[dict], audio_path: str, + model_repo: str, language: str, + audio_duration: float, + initial_prompt: Optional[str] = None, + char_threshold: float = 0.7, + silences: Optional[list[tuple[float, float]]] = None) -> list[dict]: + """检测并重新转写 Whisper 陷入死循环或长空段的异常片段。 + + Detection modes: + 1. Single-char loop — one character dominates (e.g. '秀秀秀秀秀...') + 2. Phrase loop — text built from a repeating substring + 3. Empty segment — text empty AND segment ≥15s AND not in VAD silence + + If retry produces no valid text, the segment range is deleted. + Traverses ranges in REVERSE to avoid index shifting from list mutations. + """ + # Helper: check if a time range overlaps any VAD silence region + def _in_silence(seg_start: float, seg_end: float) -> bool: + if not silences: + return False + for s_start, s_end in silences: + if seg_start < s_end and seg_end > s_start: + return True + return False + + # 1. Detect problem segments + bad_indices = [] + for i, seg in enumerate(segments): + text = seg.get("text", "").strip() + + # Mode A1: single-char loop (only for very long repetitions) + if len(text) >= 20: + most_common = max(set(text), key=text.count) + if text.count(most_common) / len(text) > char_threshold: + bad_indices.append(i) + continue + + # Mode A2: phrase-level loop + if is_phrase_repetition(text): + bad_indices.append(i) + continue + + # Mode B: long empty segment not explainable by VAD silence + seg_len = seg["end"] - seg["start"] + if not text and seg_len >= 15.0 and not _in_silence(seg["start"], seg["end"]): + bad_indices.append(i) + + if not bad_indices: + return segments + + # 2. Group consecutive indices into continuous ranges + ranges = [] + start_i = bad_indices[0] + end_i = bad_indices[0] + for i in bad_indices[1:]: + if i == end_i + 1: + end_i = i + else: + ranges.append((start_i, end_i)) + start_i = end_i = i + ranges.append((start_i, end_i)) + + # 3. Re-transcribe each range with a cold start + # CRITICAL: reversed() prevents index shifting when list length changes + from mlx_whisper import transcribe + + result = list(segments) + total_replaced = 0 + + for seg_start_idx, seg_end_idx in reversed(ranges): + t_start = max(0, result[seg_start_idx]["start"] - 0.3) + t_end = min(result[seg_end_idx]["end"] + 0.3, audio_duration) + + print(f" Problem segment at {t_start:.1f}s-{t_end:.1f}s, " + f"re-transcribing ({seg_end_idx - seg_start_idx + 1} segs)...") + + tmp_dir = tf_module.mkdtemp() + try: + seg_path = os.path.join(tmp_dir, "_retry.wav") + subprocess.run([ + "ffmpeg", "-y", "-i", audio_path, + "-ss", str(t_start), + "-to", str(t_end), + "-af", "loudnorm=I=-16:TP=-1.5:LRA=11", + "-ar", "16000", "-ac", "1", + "-f", "wav", seg_path, + ], capture_output=True, text=True, check=True) + + # Cold-start transcribe (no preceding context) + retry_opts = { + "path_or_hf_repo": model_repo, + "language": language, + "verbose": False, + "word_timestamps": True, + } + if initial_prompt: + retry_opts["initial_prompt"] = initial_prompt + + retry_result = transcribe(seg_path, **retry_opts) + + # Build replacement segments with absolute timestamps + new_segs = [] + for s in retry_result.get("segments", []): + text_clean = s["text"].strip() + # Filter: no empty text, no repeated loops in retry + if not text_clean: + continue + if len(text_clean) >= 5: + mc = max(set(text_clean), key=text_clean.count) + if text_clean.count(mc) / len(text_clean) > char_threshold: + continue + if is_phrase_repetition(text_clean): + continue + # Density check: impossible char rate → hallucination + # Normal Japanese speech is ~7 chars/s; 20 is very generous + seg_dur = s["end"] - s["start"] + if seg_dur > 0 and len(text_clean) > seg_dur * 20: + continue + new_segs.append({ + "start": s["start"] + t_start, + "end": s["end"] + t_start, + "text": text_clean, + "words": [ + {"start": w["start"] + t_start, + "end": w["end"] + t_start, + "word": w["word"]} + for w in s.get("words", []) + ] + }) + + if new_segs: + result[seg_start_idx:seg_end_idx + 1] = new_segs + total_replaced += (seg_end_idx - seg_start_idx + 1) + print(f" Replaced with {len(new_segs)} corrected segments") + else: + # Retry produced nothing useful — delete (silence/noise/hallucination) + del result[seg_start_idx:seg_end_idx + 1] + print(f" Retry empty, removed {seg_end_idx - seg_start_idx + 1} segment(s)") + + except Exception as e: + print(f" Warning: retry failed ({e}), keeping original segments") + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + if total_replaced: + print(f" Repetition fix: replaced {total_replaced} hallucinated segments") + return result + + +# ── Output Formatting ──────────────────────────────────────────────────────── + +def format_timestamp_srt(seconds: float) -> str: + """Format seconds to SRT timestamp: HH:MM:SS,mmm + Uses round() to avoid floating-point precision errors where + 1.001 % 1 = 0.000999... → int() truncates to 000 instead of 001. + Clamps negative values to 0 (Whisper shouldn't return them, but be safe).""" + if seconds < 0: + seconds = 0.0 + total_ms = int(round(seconds * 1000)) + h = total_ms // 3600000 + m = (total_ms % 3600000) // 60000 + s = (total_ms % 60000) // 1000 + ms = total_ms % 1000 + return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" + + +def write_output(segments: list[dict], output_path: str, fmt: str): + """Write transcription output in specified format.""" + base = Path(output_path).with_suffix("") + + if fmt in ("txt", "all"): + txt_path = str(base) + ".txt" + with open(txt_path, "w", encoding="utf-8") as f: + for seg in segments: + f.write(seg["text"] + "\n") + print(f" Written: {txt_path}") + + if fmt in ("srt", "all"): + srt_path = str(base) + ".srt" + with open(srt_path, "w", encoding="utf-8") as f: + for i, seg in enumerate(segments, 1): + f.write(f"{i}\n") + f.write(f"{format_timestamp_srt(seg['start'])} --> " + f"{format_timestamp_srt(seg['end'])}\n") + f.write(f"{seg['text']}\n\n") + print(f" Written: {srt_path}") + + if fmt in ("json", "all"): + json_path = str(base) + ".json" + with open(json_path, "w", encoding="utf-8") as f: + json.dump(segments, f, ensure_ascii=False, indent=2) + print(f" Written: {json_path}") + + +# ── Main ───────────────────────────────────────────────────────────────────── + +@contextmanager +def tempfile_directory(): + d = tf_module.mkdtemp() + try: + yield d + finally: + shutil.rmtree(d, ignore_errors=True) + + +def main(): + parser = argparse.ArgumentParser( + description="MLX Whisper enhanced transcription with precise subtitles") + parser.add_argument("audio", help="Input audio file path") + parser.add_argument("--model", default="turbo", + choices=["turbo", "large-v3-turbo", "small", "medium"], + help="mlx-whisper model (default: turbo)") + parser.add_argument("--language", default="ja", + help="Language code (default: ja)") + parser.add_argument("--output-dir", default=None, + help="Output directory (default: same as input)") + parser.add_argument("--output-format", default="txt", + choices=["txt", "srt", "json", "all"], + help="Output format (default: txt)") + parser.add_argument("--no-clean", action="store_true", + help="Disable hallucination cleanup") + parser.add_argument("--no-word-ts", action="store_true", + help="Disable word-level timestamps") + parser.add_argument("--no-smart-split", action="store_true", + help="Disable smart split on pauses") + parser.add_argument("--no-vad-filter", action="store_true", + help="Disable silence/VAD filtering") + parser.add_argument("--max-line-duration", type=float, default=5.0, + help="Max seconds per subtitle line (default: 5.0)") + parser.add_argument("--initial-prompt", required=True, + help="Initial prompt for context guidance (required)") + parser.add_argument("--verbose", action="store_true", + help="Show progress") + + args = parser.parse_args() + + audio_path = Path(args.audio).resolve() + if not audio_path.exists(): + print(f"Error: File not found: {audio_path}", file=sys.stderr) + sys.exit(1) + + output_dir = Path(args.output_dir) if args.output_dir else audio_path.parent + output_dir.mkdir(parents=True, exist_ok=True) + + output_name = audio_path.stem + model_repo = MLX_MODELS[args.model] + + # Get duration + duration = get_audio_duration(str(audio_path)) + duration_min = duration / 60 + print(f"Audio: {audio_path.name} ({duration_min:.1f} min)") + print(f"Model: {args.model} ({model_repo})") + + # ── Preprocessing ────────────────────────────────────────────────────── + print("Preprocessing: 16kHz mono + loudness normalization...") + with tempfile_directory() as preprocess_dir: + preprocessed = preprocess_audio(str(audio_path), preprocess_dir) + print(f" Preprocessed: {Path(preprocessed).name}") + + # Optional: detect silence regions for VAD filtering + silences = [] + if not args.no_vad_filter: + print("Detecting silent regions...") + silences = detect_silence(preprocessed, noise_db=-40, min_duration=0.5, + audio_duration=duration) + print(f" Found {len(silences)} silent regions") + + # ── Transcribe (mlx-whisper handles long audio internally) ────────── + print(f"Transcribing with mlx-whisper ({args.model})...") + t0 = time.time() + result = transcribe_file( + preprocessed, model_repo, args.language, + args.initial_prompt, + word_timestamps=not args.no_word_ts, + verbose=args.verbose) + elapsed = time.time() - t0 + + segments_data = result.get("segments", []) + merged = [ + {"start": s["start"], "end": s["end"], + "text": s["text"].strip(), + "words": [ + {"start": w["start"], "end": w["end"], "word": w["word"]} + for w in s.get("words", []) + ]} + for s in segments_data + ] + print(f" Transcribed {len(merged)} segments in {elapsed:.1f}s") + + # ── Post-processing ──────────────────────────────────────────────────── + + # 1. VAD filter: remove ghost subtitles in silent regions + if not args.no_vad_filter and silences: + before = len(merged) + merged = remove_vad_gaps(merged, silences) + removed = before - len(merged) + if removed > 0: + print(f" VAD filter: removed {removed} segments in silent regions") + + # 2. Smart split: break long glued segments on natural pauses + if not args.no_smart_split and not args.no_word_ts: + before = len(merged) + merged = smart_split_segments(merged, max_duration=args.max_line_duration) + if len(merged) > before: + print(f" Smart split: {before} → {len(merged)} segments") + + # 3. Hallucination cleanup + if not args.no_clean: + before = len(merged) + merged = clean_hallucinations(merged) + removed = before - len(merged) + if removed > 0: + print(f" Cleaned: removed {removed} hallucinated segments") + + # 4. Repetition loop detection & local re-transcription + if not args.no_clean: + merged = fix_repetition_loops( + merged, str(audio_path), model_repo, + args.language, duration, args.initial_prompt, + silences=silences) + + # ── Output ───────────────────────────────────────────────────────────── + out_path = str(output_dir / output_name) + print("Output:") + write_output(merged, out_path, args.output_format) + + print("Done!") + + +if __name__ == "__main__": + main()