88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
import asyncio
|
||
import os
|
||
import sys
|
||
import uuid
|
||
|
||
# 添加路径以导入其他模块
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
from T1_LLM import stream_and_split_text
|
||
from T2_StreamingVolanoTTS import StreamingVolcanoTTS
|
||
|
||
|
||
async def streaming_tts_pipeline(prompt, audio_callback):
|
||
"""
|
||
流式TTS管道：获取LLM流式输出并断句，然后使用TTS合成语音
|
||
|
||
Args:
|
||
prompt: 提示文本
|
||
audio_callback: 音频数据回调函数
|
||
"""
|
||
# 1. 获取LLM流式输出并断句
|
||
text_stream = stream_and_split_text(prompt)
|
||
|
||
# 2. 初始化TTS处理器
|
||
tts = StreamingVolcanoTTS()
|
||
|
||
# 3. 流式处理文本并生成音频
|
||
await tts.synthesize_stream(text_stream, audio_callback)
|
||
|
||
|
||
def save_audio_callback(output_dir=None):
|
||
"""
|
||
创建一个音频回调函数，用于保存音频数据到文件
|
||
|
||
Args:
|
||
output_dir: 输出目录，默认为当前文件所在目录下的output文件夹
|
||
|
||
Returns:
|
||
音频回调函数
|
||
"""
|
||
if output_dir is None:
|
||
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
|
||
|
||
# 确保输出目录存在
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
def callback(audio_data):
|
||
# 生成文件名
|
||
filename = f"pipeline_tts_{uuid.uuid4().hex[:8]}.wav"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
# 保存音频文件
|
||
with open(filepath, "wb") as f:
|
||
f.write(audio_data)
|
||
|
||
print(f"音频片段已保存到: {filepath} ({len(audio_data)} 字节)")
|
||
|
||
return callback
|
||
|
||
|
||
async def test_pipeline():
|
||
"""
|
||
测试流式TTS管道
|
||
"""
|
||
# 创建音频回调函数
|
||
audio_handler = save_audio_callback()
|
||
|
||
# 测试提示
|
||
prompt = "请详细解释一下量子力学的基本原理，包括波粒二象性、不确定性原理和薛定谔方程。"
|
||
|
||
print("开始测试流式TTS管道...")
|
||
print(f"测试提示: {prompt}")
|
||
print("等待LLM生成文本并转换为语音...")
|
||
|
||
# 运行管道
|
||
await streaming_tts_pipeline(prompt, audio_handler)
|
||
|
||
print("流式TTS管道测试完成！")
|
||
|
||
|
||
def main():
|
||
"""
|
||
主函数，运行测试
|
||
"""
|
||
asyncio.run(test_pipeline())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |