昨天我成功讓 Whisper 在本地跑起來了,但是馬上發現了一個問題,就是每次執行時都要重新載入一次 medium
模型,都需要花費 10 幾秒!如果要讓 Agent 處理多個任務,這種方式完全沒辦法。
因此我今天的主要任務就是進行「服務化」重構,目標是
WhisperService
類別我為了要做到關注點分離,因此首先
src/
src/
下建立 whisper_service.py
和 test_whisper_service.py
M2A Agent/
├─ src/
│ ├─ whisper_service.py
│ └─ test_whisper_service.py
├─ recording/
│ ├─ 小妹妹介紹她的玩偶.m4a
│ └─ meeting_demo.mp3
└─ venv/
src
裡面是我們專案的核心功能,而像測試腳本、文件、音訊檔案這些輔助性的東西就放在外面。
這樣的好處是讓專案更整潔,未來要打包或部署時也更方便!
我為了將語音轉錄流程整理成一個易於重複使用的類別,因此我封裝了 WhisperService
的類別,寫在 src/whisper_service.py
裡。
在構思這個類別時,我希望可以達成兩個目標
whisper.load_model()
放在 __init__
建構函式裡。這樣一來,只要服務實體還活著,模型就一直在記憶體裡待命。try...except
來捕捉任何可能發生的錯誤。import whisper
import time
import os
class WhisperService:
# 載入 Whisper 模型,只載入一次
def __init__(self, model_name="medium"):
print(f"[WhisperService] 載入模型 {model_name} 中…")
self.model = whisper.load_model(model_name)
print(f"[WhisperService] 模型 {model_name} 載入完成")
# 轉錄單一音訊,回傳結果字典
def transcribe(self, audio_path: str, language: str = "zh") -> dict:
if not os.path.isfile(audio_path):
return {"error": f"找不到檔案:{audio_path}"}
print(f"[WhisperService] 開始轉錄:{audio_path}")
start = time.time()
try:
result = self.model.transcribe(audio_path, language=language)
elapsed = time.time() - start
# 成功時回傳
return {
"audio_path": audio_path,
"text": result.get("text", ""),
"segments": result.get("segments", []),
"duration": result.get("duration", 0),
"transcribe_time": elapsed,
"error": None,
}
except Exception as e:
# 失敗時回傳,但 error 不為 None
return {
"audio_path": audio_path,
"text": "",
"segments": [],
"duration": 0,
"transcribe_time": 0,
"error": str(e),
}
一個好的服務,需要一個清晰的 API 介面,讓呼叫者能用同樣的方式處理結果。
audio_path
(字串):告訴它音訊檔案放在哪裡。它支援常見的格式為 .wav
, .mp3
, .m4a
等。language
(字串, 選填):可以指定音訊的語言,例如 'zh'
代表中文。如果省略,它會試著自己猜測。無論轉錄成功或失敗,它都會回傳一個 Python 字典 (Dictionary)。我這樣設計,是為了讓呼叫者能用同樣的方式處理結果,只需要檢查一個欄位就知道發生了什麼事。
格式如下
"text"
:完整語音轉文字結果"segments"
:分段時間戳"duration"
:音訊長度"transcribe_time"
:處理所需秒數"audio_path"
:原始路徑"error"
:如果發生例外,給出錯誤訊息{
"text": "大家好...", // 完整轉錄文字
"segments": [ ... ], // 帶有時間戳的詳細分段
"duration": 31, // 音訊的總長度,單位為【秒】
"transcribe_time": 30.08, // 這次轉錄花了多少時間
"audio_path": "recording/...", // 原始檔案路徑
"error": null // null 代表成功轉錄!
}
{
"text": "",
"segments": [],
"duration": 0,
"transcribe_time": 0,
"audio_path": "recording/not_exist.mp3",
"error": "找不到檔案:recording/not_exist.mp3" // 這裡會清楚說明錯誤原因
}
這樣統一的設計,讓未來的 MCP Agent 在與我的 WhisperService 溝通時,可以更穩定,因為它們知道該如何解析回傳的結果。
為了驗證我們剛剛寫的 WhisperService 類別,因此我在 src
底下寫了一個測試程式 test_whisper_service.py
。
這次我不只測一個檔案,我準備了一個清單,裡面包含兩個正常的音訊檔,還有一個故意寫錯檔名的檔案,來測試我的錯誤處理機制是否正常運作。
from whisper_service import WhisperService
import os
def main():
# 初始化服務
service = WhisperService(model_name="medium")
# 準備測試的音訊檔
audio_files = [
"recording/小妹妹介紹他的玩偶.m4a",
"recording/meeting_demo.mp3",
"recording/not_exist.mp3", # 用於測試錯誤處理
]
# 逐一測試
for path in audio_files:
result = service.transcribe(path, language="zh")
print(f"\n--- 測試檔案:{path} ---")
if result["error"]:
print(f"❌ 轉錄失敗:{result['error']}")
else:
print(f"✅ 轉錄成功,時間:{result['transcribe_time']:.2f}s")
print(f"檔案長度:{result['duration']:.2f}s")
print(f"段落數:{len(result['segments'])}")
print(f"文字內容:{result['text'][:80]}…")
if __name__ == "__main__":
main()
執行後,結果完全符合我的預期!正常的檔案都成功轉錄,而且那個不存在的檔案也回報了「找不到檔案」的錯誤,整個程式並沒有因此中斷!
✅ 完成項目
我執行了以上的測試後,確認了 WhisperService 類別 已經可以穩定的處理文字轉錄,並且能透過測試程式來批次檔案驗證效能與穩定性,這為後續 MCP Agent 的整合打下了紮實的地基。
🎯 明天計劃
開始打造我們 M2A Agent 的「大腦」,設計 MCP Agent 的核心協調邏輯,並讓它學會如何使用我們今天打造的 WhisperService
。