57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import asyncio
|
||
import base64
|
||
import os
|
||
import sys
|
||
|
||
# 确保能找到backend模块
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||
|
||
from backend.tts_asr import _tts_sync_with_retry, _load_asr_pipeline_with_retry, _get_asr_pipeline
|
||
|
||
async def verify_tts_asr_cross():
|
||
print("====================================")
|
||
print(" 开始严格交叉验证: TTS 生成 -> ASR 解析")
|
||
print("====================================")
|
||
|
||
test_text = "苹果设备支持离线大模型运算"
|
||
print(f"\n[1] 正在调用 TTS 引擎 (suno/bark-small)...")
|
||
print(f"目标文本: '{test_text}'")
|
||
|
||
try:
|
||
# TTS生成
|
||
audio_bytes, sr = await _tts_sync_with_retry(test_text, rate=1.0)
|
||
print(f"-> TTS 成功生成音频数据，大小: {len(audio_bytes)} Bytes, 采样率: {sr}Hz")
|
||
except Exception as e:
|
||
print(f"-> TTS 失败: {str(e)}")
|
||
sys.exit(1)
|
||
|
||
print("\n[2] 正在调用 ASR 引擎 (Whisper)...")
|
||
try:
|
||
loaded = await _load_asr_pipeline_with_retry()
|
||
if not loaded:
|
||
print("-> ASR 核心加载失败！")
|
||
sys.exit(1)
|
||
|
||
print("-> ASR 加载成功，开始解析音频...")
|
||
|
||
# 将生成的wav bytes传递给ASR进行语音识别
|
||
asr_pipeline = _get_asr_pipeline()
|
||
result = asr_pipeline(audio_bytes, generate_kwargs={"task": "transcribe"})
|
||
recognized_text = result.get('text', '')
|
||
print(f"-> ASR 识别结果: '{recognized_text.strip()}'")
|
||
|
||
if len(recognized_text.strip()) > 0:
|
||
print("\n结论: ✅ 验证成功！TTS和ASR模块功能链路闭环完成。")
|
||
else:
|
||
print("\n结论: ❌ ASR输出为空字符，闭环失败。")
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
print(f"-> ASR 分析阶段失败: {str(e)}")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(verify_tts_asr_cross())
|