feat(transcriber): 添加 MLX Whisper 转录器支持

- 新增 MLXWhisperTranscriber 类，支持在 Apple 平台上进行转录。 - 更新 transcriber_provider.py，动态导入 MLX Whisper 转录器并添加相应的环境变量检查。 - 修改 .env.example 文件，更新 TRANSCRIBER_TYPE 配置说明以包含 mlx-whisper 选项。
2026-09-07 00:27:37 +08:00 · 2025-04-20 00:37:48 +08:00
parent 0b7f6ca4ee
commit 369de19572
3 changed files with 122 additions and 4 deletions
@@ -0,0 +1,82 @@
+import mlx_whisper
+from pathlib import Path
+import os
+import platform
+
+from app.decorators.timeit import timeit
+from app.models.transcriber_model import TranscriptSegment, TranscriptResult
+from app.transcriber.base import Transcriber
+from app.utils.logger import get_logger
+from app.utils.path_helper import get_model_dir
+from events import transcription_finished
+
+logger = get_logger(__name__)
+
+class MLXWhisperTranscriber(Transcriber):
+    def __init__(
+            self,
+            model_size: str = "base",
+            device: str = None,  # MLX 会自动选择最佳设备
+    ):
+        # 检查平台
+        if platform.system() != "Darwin":
+            raise RuntimeError("MLX Whisper 仅支持 Apple 平台")
+            
+        # 检查环境变量
+        if os.environ.get("TRANSCRIBER_TYPE") != "mlx-whisper":
+            raise RuntimeError("必须设置环境变量 TRANSCRIBER_TYPE=mlx-whisper 才能使用 MLX Whisper")
+            
+        self.model_size = model_size
+        self.model_name = f"whisper-{model_size}-mlx"
+        self.model_path = None
+        
+        # 设置模型路径
+        model_dir = get_model_dir("mlx-whisper")
+        self.model_path = os.path.join(model_dir, self.model_name)
+        
+        # 确保模型目录存在
+        Path(model_dir).mkdir(parents=True, exist_ok=True)
+        
+        logger.info(f"初始化 MLX Whisper 转录器，模型：{self.model_name}")
+
+    @timeit
+    def transcript(self, file_path: str) -> TranscriptResult:
+        try:
+            # 使用 MLX Whisper 进行转录
+            result = mlx_whisper.transcribe(
+                file_path,
+                path_or_hf_repo=f"mlx-community/{self.model_name}"
+            )
+            
+            # 转换为标准格式
+            segments = []
+            full_text = ""
+            
+            for segment in result["segments"]:
+                text = segment["text"].strip()
+                full_text += text + " "
+                segments.append(TranscriptSegment(
+                    start=segment["start"],
+                    end=segment["end"],
+                    text=text
+                ))
+            
+            transcript_result = TranscriptResult(
+                language=result.get("language", "unknown"),
+                full_text=full_text.strip(),
+                segments=segments,
+                raw=result
+            )
+            
+            self.on_finish(file_path, transcript_result)
+            return transcript_result
+            
+        except Exception as e:
+            logger.error(f"MLX Whisper 转写失败：{e}")
+            raise e
+
+    def on_finish(self, video_path: str, result: TranscriptResult) -> None:
+        logger.info("MLX Whisper 转写完成")
+        transcription_finished.send({
+            "file_path": video_path,
+        })