diff --git a/TODO.md b/TODO.md index 6d6ff7b..ed05051 100644 --- a/TODO.md +++ b/TODO.md @@ -19,12 +19,13 @@ - 配置命令: `openclaw config set channels.wecom.voice.type "aliyun"` ### 字幕增强方案（待实现） -- [ ] **编译 Whisper.cpp** - 需要 sudo 权限 - - 命令: `sudo apt-get install cmake && cd ~/whisper.cpp && make` - - 替换 FunASR，获得更好的中文识别效果 -- [x] ~~语音识别(ASR)方案~~ - 已完成基础功能 - - FunASR 备用方案已可用 - - Whisper 编译完成后可提升准确率 +- [x] ~~编译 Whisper.cpp~~ - 已完成基础功能 + - Whisper 编译完成，集成到 KB Workflow +- [ ] **集成 pyannote.audio 说话者分离** + - 创建 kb-whisper-diarization 技能 + - 安装依赖: `pip install pyannote.audio torch` + - 设置 HF_TOKEN 环境变量 + - 申请模型权限: https://huggingface.co/pyannote/speaker-diarization --- diff --git a/skills/kb-whisper-diarization/.env.example b/skills/kb-whisper-diarization/.env.example new file mode 100644 index 0000000..108647b --- /dev/null +++ b/skills/kb-whisper-diarization/.env.example @@ -0,0 +1,9 @@ +# KB Whisper Diarization 环境变量 + +# Whisper.cpp 安装路径 +WHISPER_CPP_PATH=~/whisper.cpp + +# HuggingFace Token（必需，用于下载 pyannote 模型） +# 获取方式: https://huggingface.co/settings/tokens +# 注意: 需要先申请 pyannote/speaker-diarization 模型访问权限 +HF_TOKEN=hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx diff --git a/skills/kb-whisper-diarization/SKILL.md b/skills/kb-whisper-diarization/SKILL.md new file mode 100644 index 0000000..4f705d7 --- /dev/null +++ b/skills/kb-whisper-diarization/SKILL.md @@ -0,0 +1,127 @@ +# KB Whisper Diarization Skill + +**kb-whisper-diarization** - 使用 Whisper + pyannote.audio 实现语音识别和说话者分离。 + +## 什么时候使用 + +当用户需要： +- 区分多个说话者（如访谈、对话） +- 获取带说话者标签的字幕 +- 分析对话中谁说了什么 + +## 环境变量 + +在使用之前，必须检查环境变量是否已经设置。 + +| 变量名 | 用途 | 示例 | +|--------|------|------| +| `WHISPER_CPP_PATH` | whisper.cpp 安装路径 | `~/whisper.cpp` | +| `HF_TOKEN` | HuggingFace Token | `hf_xxx...` | + +### 获取 HuggingFace Token + +pyannote.audio 需要从 HuggingFace 下载模型，需要先： +1. 注册 HuggingFace 账号：https://huggingface.co +2. 申请模型访问权限：https://huggingface.co/pyannote/speaker-diarization +3. 创建 Token：https://huggingface.co/settings/tokens + +## 使用方法 + +```bash +# 基本用法（带说话者分离） +python whisper_diarization.py --input video.mp4 + +# 说话者分离 +python whisper_diarization.py --input interview.mp4 --diarize + +# 指定说话者数量 +python whisper_diarization.py --input meeting.mp4 --diarize --num-speakers 2 + +# 指定语言 +python whisper_diarization.py --input video.mp4 --language zh +``` + +## 命令参数 + +| 参数 | 说明 | 默认值 | +|------|------|--------| +| `--input` | 输入音视频文件路径 | 必需 | +| `--output` | 输出目录 | 文件所在目录 | +| `--diarize` | 启用说话者分离 | False | +| `--num-speakers` | 估计说话者数量 | None（自动检测） | +| `--language` | 语言 | 自动检测 | +| `--model` | Whisper 模型 | base | +| `--background` | 后台异步执行 | False | + +## 模型选项 + +- **Whisper 模型**: tiny, base, small, medium, large +- **说话者分离**: pyannote/speaker-diarization (需申请权限) + +## 输出文件 + +对于输入 `interview.mp4`，将生成： + +- `interview.mp4.json` - 完整结果（包含说话者标签） +- `interview.mp4.txt` - 纯文本 +- `interview.mp4.srt` - 字幕格式（带说话者标签） + +## 输出格式示例 + +### JSON 格式 + +```json +{ + "success": true, + "input_file": "interview.mp4", + "diarization": true, + "speakers": ["SPEAKER_00", "SPEAKER_01"], + "segments": [ + { + "id": 0, + "speaker": "SPEAKER_00", + "start": 0.0, + "end": 5.2, + "text": "大家好，欢迎来到访谈节目" + }, + { + "id": 1, + "speaker": "SPEAKER_01", + "start": 5.2, + "end": 10.8, + "text": "谢谢邀请，我很荣幸" + } + ] +} +``` + +### SRT 字幕格式 + +``` +1 +00:00:00,000 --> 00:00:05,200 +[SPEAKER_00] 大家好，欢迎来到访谈节目 + +2 +00:00:05,200 --> 00:00:10,800 +[SPEAKER_01] 谢谢邀请，我很荣幸 +``` + +## 安装 + +```bash +# 1. 安装依赖 +pip install pyannote.audio torch + +# 2. 设置 HuggingFace Token +export HF_TOKEN="hf_xxx..." + +# 3. 首次运行会自动下载模型 +python whisper_diarization.py --input test.mp4 +``` + +## 注意事项 + +- **说话者分离需要申请权限**: https://huggingface.co/pyannote/speaker-diarization +- **处理时间较长**: 说话者分离会增加处理时间 +- **对立体声效果更好**: 多声道音频的分离效果更佳 diff --git a/skills/kb-whisper-diarization/requirements.txt b/skills/kb-whisper-diarization/requirements.txt new file mode 100644 index 0000000..d465929 --- /dev/null +++ b/skills/kb-whisper-diarization/requirements.txt @@ -0,0 +1,8 @@ +# KB Whisper Diarization 依赖 + +# 核心依赖 +torch>=2.0.0 +pyannote.audio>=1.0.0 + +# 可选: GPU 加速 +# torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 diff --git a/skills/kb-whisper-diarization/whisper_diarization.py b/skills/kb-whisper-diarization/whisper_diarization.py new file mode 100644 index 0000000..a73ea0c --- /dev/null +++ b/skills/kb-whisper-diarization/whisper_diarization.py @@ -0,0 +1,337 @@ +#!/usr/bin/env python3 +""" +KB Whisper Diarization - 语音识别 + 说话者分离 +使用 Whisper 进行语音识别，pyannote.audio 进行说话者分离 +""" + +import argparse +import json +import os +import subprocess +import sys +import tempfile +from datetime import datetime +from pathlib import Path +import re + + +def parse_arguments(): + """解析命令行参数""" + parser = argparse.ArgumentParser( + description='KB Whisper Diarization - 语音识别 + 说话者分离', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + python whisper_diarization.py --input video.mp4 + python whisper_diarization.py --input interview.mp4 --diarize + python whisper_diarization.py --input meeting.mp4 --diarize --num-speakers 2 + """ + ) + parser.add_argument('--input', required=True, help='输入音视频文件路径') + parser.add_argument('--output', help='输出目录 (默认: 文件所在目录)') + parser.add_argument('--diarize', action='store_true', help='启用说话者分离') + parser.add_argument('--num-speakers', type=int, help='估计说话者数量') + parser.add_argument('--language', help='指定语言 (默认: 自动检测)') + parser.add_argument('--model', default='base', choices=['tiny', 'base', 'small', 'medium', 'large'], + help='Whisper 模型 (默认: base)') + parser.add_argument('--background', action='store_true', help='后台异步执行') + return parser.parse_args() + + +def get_whisper_path(): + """获取 whisper.cpp 路径""" + path = os.environ.get('WHISPER_CPP_PATH') + if path: + path = os.path.expanduser(path) + if os.path.exists(path): + return path + return os.path.expanduser('~/whisper.cpp') + + +def check_whisper_model(whisper_path, model): + """检查 Whisper 模型""" + model_path = os.path.join(whisper_path, 'models', f'ggml-{model}.bin') + if not os.path.exists(model_path): + raise FileNotFoundError(f"Whisper 模型不存在: {model_path}") + return model_path + + +def convert_to_wav(input_path, output_dir): + """转换为 16kHz WAV""" + if input_path.lower().endswith('.wav'): + try: + result = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', 'stream=sample_rate', + '-of', 'default=noprint_wrappers=1:nokey=1', input_path], + capture_output=True, text=True + ) + if result.returncode == 0 and int(result.stdout.strip()) == 16000: + return input_path + except Exception: + pass + + output_path = os.path.join(output_dir, f"{Path(input_path).stem}_audio.wav") + print(f"转换音频: {input_path} -> {output_path}") + + subprocess.run([ + 'ffmpeg', '-y', '-i', input_path, + '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', + '-loglevel', 'error', output_path + ], check=True, capture_output=True) + + print("✓ 音频转换完成") + return output_path + + +def run_whisper(whisper_path, audio_path, model, output_dir, language=None): + """运行 Whisper 识别""" + basename = Path(audio_path).stem.replace('_audio', '') + output_file = os.path.join(output_dir, basename) + + cmd = [ + os.path.join(whisper_path, 'whisper-cli'), + '-m', os.path.join(whisper_path, 'models', f'ggml-{model}.bin'), + '-f', audio_path, + '-of', output_file, + '-l', language or 'auto', + '--output-txt', '--output-srt', '--output-json', + ] + + print(f"执行 Whisper: {' '.join(cmd[:6])} ...") + + result = subprocess.run( + cmd, capture_output=True, text=True, cwd=whisper_path, timeout=3600 + ) + + if result.returncode != 0 and not os.path.exists(f"{output_file}.txt"): + raise RuntimeError(f"Whisper 失败: {result.stderr}") + + return basename + + +def run_diarization(audio_path, num_speakers=None): + """运行 pyannote.audio 说话者分离""" + print("\n🎙️ 步骤2: 说话者分离...") + + try: + import torch + from pyannote.audio import Pipeline + except ImportError as e: + print(f"✗ 缺少依赖: {e}") + print("请安装: pip install pyannote.audio torch") + raise + + # 检查 HuggingFace Token + hf_token = os.environ.get('HF_TOKEN') + if not hf_token: + raise RuntimeError("需要设置 HF_TOKEN 环境变量\n" + "获取方式: https://huggingface.co/settings/tokens") + + print("加载 pyannote 流水线...") + pipeline = Pipeline.from_pretrained( + "pyannote/speaker-diarization", + use_auth_token=hf_token + ) + + # 设置设备 + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"使用设备: {device}") + pipeline.to(device) + + # 运行分离 + print("执行说话者分离...") + 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) + }) + + print(f"✓ 检测到 {len(speakers)} 个说话者: {list(speakers.keys())}") + return speakers + + +def merge_results(whisper_txt, whisper_srt, diarization, output_dir, basename): + """合并 Whisper 和说话者分离结果""" + print("\n📝 步骤3: 合并结果...") + + # 读取 Whisper 文本 + text_segments = [] + if os.path.exists(whisper_txt): + with open(whisper_txt, 'r', encoding='utf-8') as f: + content = f.read() + + # 简单按行分段 + for i, line in enumerate(content.strip().split('\n')): + if line.strip(): + text_segments.append({ + "id": i, + "text": line.strip(), + "start": i * 3.0, # 估算 + "end": (i + 1) * 3.0 + }) + + # 读取 SRT 获取精确时间 + srt_segments = [] + if os.path.exists(whisper_srt): + with open(whisper_srt, 'r', encoding='utf-8') as f: + content = f.read() + + # 解析 SRT + pattern = r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3}) --> (\d{2}:\d{2}:\d{2},\d{3})\n(.+?)(?=\n\n|\Z)' + matches = re.findall(pattern, content, re.DOTALL) + + def parse_time(t): + h, m, s_ms = t.split(':') + s, ms = s_ms.split(',') + return float(h) * 3600 + float(m) * 60 + float(s) + float(ms) / 1000 + + for idx, num, start, end, text in matches: + srt_segments.append({ + "id": int(idx), + "start": parse_time(start), + "end": parse_time(end), + "text": text.strip() + }) + + # 为每个 Whisper 段落分配说话者 + merged_segments = [] + for seg in srt_segments: + speaker = assign_speaker(seg['start'], seg['end'], diarization) + merged_segments.append({ + "id": len(merged_segments), + "speaker": speaker, + "start": seg['start'], + "end": seg['end'], + "text": seg['text'] + }) + + # 生成带说话者的 SRT + output_srt = os.path.join(output_dir, f"{basename}_diarized.srt") + with open(output_srt, 'w', encoding='utf-8') as f: + for seg in merged_segments: + start = format_time(seg['start']) + end = format_time(seg['end']) + f.write(f"{seg['id'] + 1}\n") + f.write(f"{start} --> {end}\n") + f.write(f"[{seg['speaker']}] {seg['text']}\n\n") + + # 生成合并的 TXT + output_txt = os.path.join(output_dir, f"{basename}_diarized.txt") + with open(output_txt, 'w', encoding='utf-8') as f: + for seg in merged_segments: + f.write(f"[{seg['speaker']}] {seg['text']}\n") + + # 生成 JSON + output_json = os.path.join(output_dir, f"{basename}_diarized.json") + json_data = { + "success": True, + "tool": "whisper-diarization", + "diarization": True, + "speakers": list(diarization.keys()), + "input_file": basename, + "output_files": { + "txt": f"{basename}_diarized.txt", + "srt": f"{basename}_diarized.srt", + "json": f"{basename}_diarized.json" + }, + "segments": merged_segments + } + with open(output_json, 'w', encoding='utf-8') as f: + json.dump(json_data, f, ensure_ascii=False, indent=2) + + print(f"✓ 生成文件:") + print(f" - {output_srt}") + print(f" - {output_txt}") + print(f" - {output_json}") + + return json_data + + +def assign_speaker(start, end, diarization): + """将时间段分配给说话者""" + best_speaker = "UNKNOWN" + max_overlap = 0 + + for speaker, segments in diarization.items(): + for seg in segments: + overlap = min(end, seg['end']) - max(start, seg['start']) + if overlap > max_overlap: + max_overlap = overlap + best_speaker = speaker + + return best_speaker + + +def format_time(seconds): + """秒数转换为 SRT 时间格式""" + hours = int(seconds // 3600) + minutes = int((seconds % 3600) // 60) + secs = int(seconds % 60) + ms = int((seconds % 1) * 1000) + return f"{hours:02d}:{minutes:02d}:{secs:02d},{ms:03d}" + + +def main(): + args = parse_arguments() + + input_file = os.path.abspath(args.input) + if not os.path.exists(input_file): + print(f"✗ 文件不存在: {input_file}") + sys.exit(1) + + output_dir = os.path.abspath(args.output) if args.output else os.path.dirname(input_file) + os.makedirs(output_dir, exist_ok=True) + + basename = Path(input_file).stem + + print(f"\n📁 输入: {input_file}") + print(f"📁 输出: {output_dir}") + print(f"🎙️ 说话者分离: {'是' if args.diarize else '否'}") + + try: + # 步骤1: Whisper 识别 + print(f"\n🎙️ 步骤1: 语音识别...") + whisper_path = get_whisper_path() + check_whisper_model(whisper_path, args.model) + + audio_path = convert_to_wav(input_file, output_dir) + run_whisper(whisper_path, audio_path, args.model, output_dir, args.language) + + # 步骤2: 说话者分离 + if args.diarize: + diarization = run_diarization(audio_path, args.num_speakers) + else: + diarization = {} + + # 步骤3: 合并结果 + whisper_txt = os.path.join(output_dir, f"{basename}.txt") + whisper_srt = os.path.join(output_dir, f"{basename}.srt") + result = merge_results(whisper_txt, whisper_srt, diarization, output_dir, basename) + + # 清理临时音频 + if audio_path != input_path and audio_path.endswith('_audio.wav'): + try: + os.remove(audio_path) + except Exception: + pass + + print(f"\n✅ 处理完成!") + print(f" 说话者数量: {len(diarization) if args.diarize else 1}") + print(f" 段落数量: {len(result['segments'])}") + + except Exception as e: + print(f"\n✗ 错误: {e}") + sys.exit(1) + + +if __name__ == '__main__': + main()