From 46bd809bf6013dde7030bbf9386d721c8e0cdc0b Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 19:05:28 +0800 Subject: [PATCH] docs(phase-03): research ASR processing domain --- .../03-processing-skills/03-RESEARCH.md | 478 ++++++++++++++++++ 1 file changed, 478 insertions(+) create mode 100644 .planning/phases/03-processing-skills/03-RESEARCH.md diff --git a/.planning/phases/03-processing-skills/03-RESEARCH.md b/.planning/phases/03-processing-skills/03-RESEARCH.md new file mode 100644 index 0000000..4490cb0 --- /dev/null +++ b/.planning/phases/03-processing-skills/03-RESEARCH.md @@ -0,0 +1,478 @@ +# Phase 3: Processing Skills - Research + +**Researched:** 2026-03-13 +**Domain:** ASR (Automatic Speech Recognition) 和字幕处理集成 +**Confidence:** HIGH + +## Summary + +Phase 3 的核心目标是将现有的分散 ASR 工具整合为统一的管道步骤。关键发现： + +1. **现有资产高度可复用**：`kb-whisper/whisper.py` 和 `kb-whisper-diarization/whisper_diarization.py` 已经实现了完整的功能（音频转换、ASR 识别、多格式输出），只需包装为管道步骤模式。 + +2. **决策已锁定**：OCR 明确排除，只关注 ASR/字幕处理。官方字幕优先，无字幕时使用 Whisper.cpp。HF_TOKEN 配置时自动启用说话者分离。 + +3. **架构模式清晰**：Phase 1/2 建立的 `StepInput`/`StepOutput` Schema、`WorkflowState` 状态管理、`retry_on_network_error` 重试机制可直接复用。 + +**Primary recommendation:** 创建 `ASRProcessor` 类，统一封装官方字幕下载和 Whisper.cpp 调用，复用现有代码逻辑，遵循 Phase 1/2 的管道模式。 + + +## User Constraints (from CONTEXT.md) + +### Locked Decisions + +#### ASR 引擎选择策略 +- **优先级顺序**：B站官方字幕 → Whisper.cpp 本地识别 +- 官方字幕质量最高且无计算成本，优先使用 +- 无官方字幕时，默认使用 Whisper.cpp 进行语音识别 +- FunASR 保留作为备选，不作为默认选项 + +#### 说话者分离 +- HF_TOKEN 配置时自动启用带说话者分离的 Whisper（pyannote.audio） +- 无需显式参数控制，根据环境变量自动判断 +- 说话者分离结果合并到标准输出格式中 + +#### 输出格式 +- 保留全部格式：txt（纯文本）+ srt（字幕）+ json（结构化数据） +- 与现有 ASR 工具输出保持一致 +- JSON 包含 segments（带时间戳的分段）和完整文本 + +#### OCR 处理 +- **不集成 OCR 功能** +- Phase 3 专注于 ASR/字幕处理 +- OCR 相关计划（03-02）移除 + +#### 工作流集成 +- ASR 作为管道步骤执行，完成后记录到 workflow_state.json +- B站字幕下载集成到 ASR 步骤（统一入口：先尝试官方字幕，无则 ASR） +- 输出文件与内容放在同一目录，不创建子目录 + +#### Schema 复用 +- 复用 Phase 1 定义的 StepInput/StepOutput 格式 +- ASR 输出符合统一的 JSON Schema 规范 +- 包含 success、tool、segments、text 等标准字段 + +### Claude's Discretion +- Whisper 模型默认选择（base/small/medium） +- 重试机制的具体参数（继承 Phase 1 的 tenacity 配置） +- 音频转换的临时文件处理策略 + +### Deferred Ideas (OUT OF SCOPE) +- OCR 视频帧文字识别 — 不需要，明确排除 +- 进度可见性（实时显示 ASR 进度）— v2 需求 +- 并行 ASR 处理 — v2 需求 + + + +## Standard Stack + +### Core +| Library | Version | Purpose | Why Standard | +|---------|---------|---------|--------------| +| whisper.cpp | v1.7.1+ | 本地语音识别引擎 | 零依赖、高性能、CPU 友好 | +| ffmpeg | 4.x+ | 音频格式转换 | 行业标准，支持所有格式 | +| pydantic | v2 | Schema 验证 | Phase 1 已采用，类型安全 | +| pydantic-settings | v2 | 配置管理 | Phase 1 已采用，环境变量加载 | + +### Supporting +| Library | Version | Purpose | When to Use | +|---------|---------|---------|-------------| +| pyannote.audio | 3.x | 说话者分离 | HF_TOKEN 配置时自动启用 | +| torch | 2.x | pyannote 依赖 | 说话者分离需要 | +| tenacity | 8.x | 重试机制 | 继承 Phase 1 配置 | +| aiohttp | 3.x | 异步 HTTP | B站字幕 API 调用 | + +### Alternatives Considered +| Instead of | Could Use | Tradeoff | +|------------|-----------|----------| +| Whisper.cpp | FunASR | FunASR 中文效果好，但需要额外依赖（modelscope），作为备选方案 | +| Whisper.cpp | OpenAI Whisper API | API 需要付费且依赖网络，本地处理更可靠 | + +**Installation:** +```bash +# Whisper.cpp（如果未安装） +git clone https://github.com/ggml-org/whisper.cpp.git ~/whisper.cpp +cd ~/whisper.cpp && make +./models/download-ggml-model.sh base + +# Python 依赖 +pip install pydantic pydantic-settings tenacity aiohttp + +# 说话者分离（可选，需要 HF_TOKEN） +pip install pyannote.audio torch +``` + +## Architecture Patterns + +### Recommended Project Structure +``` +kb_pipeline/ +├── processors/ # 处理器模块（Phase 3 新增） +│ ├── __init__.py +│ ├── base.py # BaseProcessor 基类 +│ ├── asr.py # ASRProcessor 实现 +│ └── subtitle.py # 字幕下载逻辑 +├── schemas.py # 已有 Schema 定义 +├── state.py # 已有状态管理 +├── config.py # 已有配置管理 +└── retry.py # 已有重试机制 +``` + +### Pattern 1: ASRProcessor 统一入口 + +**What:** 将官方字幕下载和 Whisper.cpp 识别统一为单一处理器 +**When to use:** 所有需要视频内容转文字的场景 + +**Example:** +```python +# Source: 基于 kb-whisper/whisper.py 和 kb-bilibili-content/bilibili_content.py +from kb_pipeline.schemas import StepInput, StepOutput +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.config import get_settings + +class ASRInput(StepInput): + """ASR 步骤输入""" + video_path: str # 视频文件路径 + content_type: str = "video" # video/zvideo + +class ASROutput(StepOutput): + """ASR 步骤输出""" + tool: str # "subtitle" | "whisper" | "whisper-diarization" + has_subtitle: bool + output_files: Dict[str, str] # {txt, srt, json} + segments: List[Dict] + text: str + duration_seconds: float + +class ASRProcessor: + """ASR 处理器 - 统一字幕下载和语音识别""" + + def __init__(self, state_manager: WorkflowStateManager = None): + self.settings = get_settings() + self.state_manager = state_manager + + async def process(self, input_data: ASRInput) -> ASROutput: + """执行 ASR 处理 + + 策略： + 1. 先尝试下载官方字幕 + 2. 无官方字幕则调用 Whisper.cpp + 3. HF_TOKEN 配置时自动启用说话者分离 + """ + # Step 1: 尝试官方字幕 + if input_data.platform == "bilibili": + subtitle_result = await self._try_official_subtitle(input_data) + if subtitle_result: + return subtitle_result + + # Step 2: Whisper.cpp 识别 + return await self._run_whisper(input_data) + + async def _try_official_subtitle(self, input_data: ASRInput) -> Optional[ASROutput]: + """尝试下载 B站官方字幕""" + # 复用 kb-bilibili-content/bilibili_content.py 的 download_subtitle 逻辑 + ... + + async def _run_whisper(self, input_data: ASRInput) -> ASROutput: + """执行 Whisper.cpp 识别""" + # 检查 HF_TOKEN 决定是否启用说话者分离 + hf_token = self.settings.hf_token.get_secret_value() + if hf_token: + return await self._run_whisper_with_diarization(input_data) + else: + return await self._run_whisper_basic(input_data) +``` + +### Pattern 2: 状态记录 + +**What:** ASR 完成后记录到 workflow_state.json +**When to use:** 每次 ASR 处理完成后 + +**Example:** +```python +# Source: Phase 1 状态管理模式 +def record_asr_result(state_manager: WorkflowStateManager, result: ASROutput): + """记录 ASR 结果到状态文件""" + if result.success: + state_manager.mark_step_completed( + "asr", + output_summary=f"tool={result.tool}, segments={len(result.segments)}" + ) + else: + state_manager.mark_step_failed("asr", result.error_message) +``` + +### Anti-Patterns to Avoid + +- **不要重新实现音频转换逻辑**：`kb-whisper/whisper.py` 的 `convert_to_wav()` 已经验证可用，直接复用 +- **不要硬编码 Whisper 模型路径**：使用 `WHISPER_CPP_PATH` 环境变量 +- **不要忽略临时文件清理**：`_audio.wav` 临时文件处理后必须删除 +- **不要在没有官方字幕时报告失败**：返回 `has_subtitle=False` 并继续 Whisper 流程 + +## Don't Hand-Roll + +| Problem | Don't Build | Use Instead | Why | +|---------|-------------|-------------|-----| +| 音频格式转换 | 自己写 ffmpeg 调用 | `kb-whisper/whisper.py:convert_to_wav()` | 已处理采样率检测、格式验证 | +| SRT 时间解析 | 正则表达式解析 | `kb-whisper/whisper.py:parse_srt_for_duration()` | 已验证的时间解析逻辑 | +| Whisper 模型检查 | 自己检查文件存在 | `kb-whisper/whisper.py:check_model()` | 已处理模型下载提示 | +| 说话者分离 | 自己实现分离逻辑 | `kb-whisper-diarization/whisper_diarization.py` | 已集成 pyannote.audio | +| B站字幕下载 | 自己调用 API | `kb-bilibili-content/bilibili_content.py:download_subtitle()` | 已处理 API 调用和 SRT 转换 | + +**Key insight:** 现有代码已经过验证，核心逻辑直接复用，只需包装为管道模式。 + +## Common Pitfalls + +### Pitfall 1: Whisper.cpp 路径未配置 +**What goes wrong:** 运行时报错 "未找到 whisper.cpp" +**Why it happens:** `WHISPER_CPP_PATH` 环境变量未设置，默认路径也不存在 +**How to avoid:** +```python +def get_whisper_cpp_path(): + path = os.environ.get('WHISPER_CPP_PATH') + if path and os.path.exists(path): + return os.path.expanduser(path) + default = os.path.expanduser('~/whisper.cpp') + if os.path.exists(default): + return default + raise RuntimeError( + "未找到 whisper.cpp。请设置 WHISPER_CPP_PATH 或安装到 ~/whisper.cpp" + ) +``` +**Warning signs:** FileNotFoundError 在启动时 + +### Pitfall 2: HF_TOKEN 未配置时尝试说话者分离 +**What goes wrong:** pyannote.audio 报错 "需要设置 HF_TOKEN 环境变量" +**Why it happens:** 说话者分离需要 HuggingFace 认证 +**How to avoid:** 检查 HF_TOKEN 后再决定是否启用说话者分离 +```python +hf_token = settings.hf_token.get_secret_value() +if not hf_token: + # 回退到普通 Whisper + return await self._run_whisper_basic(input_data) +``` +**Warning signs:** RuntimeError 包含 "HF_TOKEN" + +### Pitfall 3: 临时音频文件未清理 +**What goes wrong:** 磁盘空间逐渐减少，`*_audio.wav` 文件堆积 +**Why it happens:** ffmpeg 转换后的临时文件没有删除 +**How to avoid:** +```python +# 处理完成后清理 +if audio_path != input_file and audio_path.endswith('_audio.wav'): + try: + os.remove(audio_path) + except Exception: + pass # 静默失败，不阻塞主流程 +``` +**Warning signs:** 目录中存在多个 `*_audio.wav` 文件 + +### Pitfall 4: 超时设置不合理 +**What goes wrong:** 长视频处理超时中断 +**Why it happens:** Whisper 处理 1 小时视频可能需要 10+ 分钟 +**How to avoid:** 设置合理超时（现有代码使用 3600 秒） +```python +result = subprocess.run(cmd, timeout=3600) # 1 小时超时 +``` +**Warning signs:** TimeoutExpired 异常 + +## Code Examples + +### 官方字幕下载（复用现有代码） + +```python +# Source: kb-bilibili-content/bilibili_content.py +async def download_subtitle(self, bvid: str) -> Dict[str, Any]: + """下载字幕并转换为 SRT 格式""" + result = {"success": False, "has_subtitle": False} + + # 获取 CID + api_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}" + async with aiohttp.ClientSession() as session: + async with session.get(api_url) as resp: + data = await resp.json() + + pages = data.get('data', []) + if not pages: + result["success"] = True # 没有分P是正常的 + return result + + cid = pages[0].get('cid', '') + + # 获取字幕列表 + async with session.get(f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}") as resp: + sub_data = await resp.json() + + subtitles = sub_data.get('data', {}).get('subtitle', {}).get('list', []) + if not subtitles: + result["success"] = True # 没有字幕是正常的 + return result + + # 下载并转换为 SRT + # ... (转换逻辑) + result["has_subtitle"] = True + result["success"] = True + return result +``` + +### Whisper.cpp 调用（复用现有代码） + +```python +# Source: kb-whisper/whisper.py +def run_whisper(whisper_path, audio_path, model, output_dir, language='zh'): + """运行 Whisper 语音识别""" + cmd = [ + os.path.join(whisper_path, 'whisper-cli'), + '-m', os.path.join(whisper_path, 'models', f'ggml-{model}.bin'), + '-f', audio_path, + '-of', os.path.join(output_dir, output_basename), + '-l', language, + '--output-txt', '--output-srt', '--output-json', + ] + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) + return result +``` + +### 说话者分离（复用现有代码） + +```python +# Source: kb-whisper-diarization/whisper_diarization.py +def run_diarization(audio_path, num_speakers=None): + """运行 pyannote.audio 说话者分离""" + hf_token = os.environ.get('HF_TOKEN') + if not hf_token: + raise RuntimeError("需要设置 HF_TOKEN 环境变量") + + from huggingface_hub import login + login(token=hf_token) + + from pyannote.audio import Pipeline + pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1") + + import torch + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + pipeline.to(device) + + diarization = pipeline(audio_path, num_speakers=num_speakers) + + # 解析结果 + speakers = {} + for turn, _, speaker in diarization.itertracks(yield_label=True): + if speaker not in speakers: + speakers[speaker] = [] + speakers[speaker].append({ + "start": round(turn.start, 2), + "end": round(turn.end, 2) + }) + return speakers +``` + +### 输出格式（标准 JSON） + +```json +{ + "success": true, + "tool": "whisper", + "model": "base", + "language": "zh", + "input_file": "video.mp4", + "output_files": { + "txt": "video.txt", + "srt": "video.srt", + "json": "video.json" + }, + "duration_seconds": 120.5, + "segments": [ + { + "id": 0, + "start": 0.0, + "end": 3.2, + "text": "这是第一段字幕", + "speaker": "SPEAKER_00" + } + ], + "text": "完整文本内容..." +} +``` + +## State of the Art + +| Old Approach | Current Approach | When Changed | Impact | +|--------------|------------------|--------------|--------| +| 手动调用 whisper.py | 管道步骤封装 | Phase 3 | 统一入口，状态可追踪 | +| 硬编码模型选择 | 配置化模型选择 | Phase 3 | 灵活调整精度/速度 | +| 无说话者分离 | HF_TOKEN 自动启用 | Phase 3 | 访谈/对话视频支持 | + +**Deprecated/outdated:** +- `kb-workflow` 中的手动 Whisper 调用步骤：将被 `ASRProcessor` 统一封装 + +## Open Questions + +1. **Whisper 默认模型选择** + - What we know: 现有代码默认使用 `base` 模型 + - What's unclear: 是否需要根据视频时长自动选择模型（短视频用 tiny，长视频用 base） + - Recommendation: 保持 `base` 默认，Claude's Discretion 范围内可后续优化 + +2. **临时文件目录** + - What we know: 当前在视频同目录创建 `_audio.wav` + - What's unclear: 是否应该使用系统临时目录 + - Recommendation: 保持现有行为（同目录），便于调试和断点续传 + +## Validation Architecture + +### Test Framework +| Property | Value | +|----------|-------| +| Framework | pytest 8.x | +| Config file | pytest.ini | +| Quick run command | `pytest tests/test_processors/ -x -v` | +| Full suite command | `pytest tests/ -v` | + +### Phase Requirements → Test Map +| Req ID | Behavior | Test Type | Automated Command | File Exists? | +|--------|----------|-----------|-------------------|-------------| +| ASR-01 | 官方字幕下载成功 | unit | `pytest tests/test_processors/test_asr.py::test_official_subtitle -x` | ❌ Wave 0 | +| ASR-02 | 无字幕时调用 Whisper | unit | `pytest tests/test_processors/test_asr.py::test_whisper_fallback -x` | ❌ Wave 0 | +| ASR-03 | HF_TOKEN 启用说话者分离 | unit | `pytest tests/test_processors/test_asr.py::test_diarization_enabled -x` | ❌ Wave 0 | +| ASR-04 | 输出 txt/srt/json 三格式 | unit | `pytest tests/test_processors/test_asr.py::test_output_formats -x` | ❌ Wave 0 | +| ASR-05 | 状态记录到 workflow_state | unit | `pytest tests/test_processors/test_asr.py::test_state_recording -x` | ❌ Wave 0 | +| ASR-06 | 临时文件清理 | unit | `pytest tests/test_processors/test_asr.py::test_temp_file_cleanup -x` | ❌ Wave 0 | + +### Sampling Rate +- **Per task commit:** `pytest tests/test_processors/test_asr.py -x -v` +- **Per wave merge:** `pytest tests/ -v` +- **Phase gate:** Full suite green before `/gsd:verify-work` + +### Wave 0 Gaps +- [ ] `tests/test_processors/__init__.py` — 测试模块初始化 +- [ ] `tests/test_processors/conftest.py` — 共享 fixtures（mock Whisper、mock B站 API） +- [ ] `tests/test_processors/test_asr.py` — ASR 处理器测试 +- [ ] `tests/test_processors/test_subtitle.py` — 字幕下载测试 + +## Sources + +### Primary (HIGH confidence) +- kb-whisper/whisper.py - 已验证的 Whisper.cpp 封装实现 +- kb-whisper-diarization/whisper_diarization.py - 已验证的说话者分离实现 +- kb-bilibili-content/bilibili_content.py - 已验证的 B站字幕下载实现 +- kb_pipeline/schemas.py - Phase 1 定义的 Schema 体系 +- kb_pipeline/state.py - Phase 1 定义的状态管理 +- kb_pipeline/config.py - Phase 1 定义的配置管理 + +### Secondary (MEDIUM confidence) +- [whisper.cpp GitHub](https://github.com/ggml-org/whisper.cpp) - 官方仓库，确认安装方式 +- [pyannote.audio GitHub](https://github.com/pyannote/pyannote-audio) - 官方仓库，确认 HF_TOKEN 配置 + +### Tertiary (LOW confidence) +- 无 + +## Metadata + +**Confidence breakdown:** +- Standard stack: HIGH - 基于现有代码和已验证的依赖 +- Architecture: HIGH - 遵循 Phase 1/2 建立的模式 +- Pitfalls: HIGH - 基于现有代码中的错误处理逻辑 + +**Research date:** 2026-03-13 +**Valid until:** 30 days（稳定技术栈）