推理队列封装import asyncio import edge_tts import time from threading import Thread from queue import Queue, Empty import os def tts_sync_stream(text, voicezh-CN-XiaoxiaoNeural): 同步流式生成内部用 async async def _run(): communicate edge_tts.Communicate(text, voice) audio_chunks [] first_chunk_time None start_time time.time() chunk_count 0 async for chunk in communicate.stream(): if chunk[type] audio: chunk_count 1 if first_chunk_time is None: first_chunk_time time.time() - start_time audio_chunks.append(chunk[data]) total_time time.time() - start_time return audio_chunks, first_chunk_time, total_time, chunk_count return asyncio.run(_run()) # Worker def worker(queue: Queue, out_diroutput): os.makedirs(out_dir, exist_okTrue) while True: try: text, idx queue.get(timeout1) except Empty: break print(f\n 开始合成: {text}) audio_chunks, ttfb, total_time, chunk_count tts_sync_stream(text) # 保存音频 output_path os.path.join(out_dir, fout_{idx:02d}.mp3) with open(output_path, wb) as f: for c in audio_chunks: f.write(c) # 估算音频时长简单估计 audio_size sum(len(c) for c in audio_chunks) audio_duration audio_size / (16000 * 2) # 粗略估计16k 16bit rtf total_time / audio_duration if audio_duration 0 else 0 print(f 保存: {output_path}) print(fTTFB: {ttfb:.3f}s | Total: {total_time:.3f}s | RTF: {rtf:.3f} | chunks: {chunk_count}) if __name__ __main__: texts [ 繁荣的商务区如金融街, 完善的城市基础设施, 服务业是主要经济支柱之一, 这里有兵马俑, 曾举办过2008年夏季奥运会, 北京是一座兼具悠久历史, 深厚文化底蕴和现代化城市功能的国际化大都市, ] queue Queue() for i, t in enumerate(texts): queue.put((t, i)) start_all time.time() # 单 worker你要测纯推理速度 t Thread(targetworker, args(queue,)) t.start() t.join() print(\n 总耗时:, time.time() - start_all)推理耗时20s。