LLM和TTS采用适配器模式进行解耦，方便扩展更多服务平台调用

core/utils/tts.py

@@ -1,9 +1,11 @@
 import asyncio
 import logging
 import os
+import sys
 import json
 import uuid
 import base64
+import importlib
 from datetime import datetime
 import edge_tts
 import numpy as np
@@ -16,150 +18,15 @@ from abc import ABC, abstractmethod
 logger = logging.getLogger(__name__)
 
 
-class TTS(ABC):
-    def __init__(self, config, delete_audio_file):
-        self.delete_audio_file = delete_audio_file
-        self.output_file = config.get("output_file")
-
-    @abstractmethod
-    def generate_filename(self):
-        pass
-
-    def to_tts(self, text):
-        tmp_file = self.generate_filename()
-        try:
-            max_repeat_time = 5
-            while not os.path.exists(tmp_file) and max_repeat_time > 0:
-                asyncio.run(self.text_to_speak(text, tmp_file))
-                if not os.path.exists(tmp_file):
-                    max_repeat_time = max_repeat_time - 1
-                    logger.error(f"语音生成失败: {text}:{tmp_file}，再试{max_repeat_time}次")
-
-            return tmp_file
-        except Exception as e:
-            logger.info(f"Failed to generate TTS file: {e}")
-            return None
-
-    @abstractmethod
-    async def text_to_speak(self, text, output_file):
-        pass
-
-    def wav_to_opus_data(self, wav_file_path):
-        # 使用pydub加载PCM文件
-        # 获取文件后缀名
-        file_type = os.path.splitext(wav_file_path)[1]
-        if file_type:
-            file_type = file_type.lstrip('.')
-        audio = AudioSegment.from_file(wav_file_path, format=file_type)
-
-        duration = len(audio) / 1000.0
-
-        # 转换为单声道和16kHz采样率（确保与编码器匹配）
-        audio = audio.set_channels(1).set_frame_rate(16000)
-
-        # 获取原始PCM数据（16位小端）
-        raw_data = audio.raw_data
-
-        # 初始化Opus编码器
-        encoder = opuslib.Encoder(16000, 1, opuslib.APPLICATION_AUDIO)
-
-        # 编码参数
-        frame_duration = 60  # 60ms per frame
-        frame_size = int(16000 * frame_duration / 1000)  # 960 samples/frame
-
-        opus_datas = []
-        # 按帧处理所有音频数据（包括最后一帧可能补零）
-        for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
-            # 获取当前帧的二进制数据
-            chunk = raw_data[i:i + frame_size * 2]
-
-            # 如果最后一帧不足，补零
-            if len(chunk) < frame_size * 2:
-                chunk += b'\x00' * (frame_size * 2 - len(chunk))
-
-            # 转换为numpy数组处理
-            np_frame = np.frombuffer(chunk, dtype=np.int16)
-
-            # 编码Opus数据
-            opus_data = encoder.encode(np_frame.tobytes(), frame_size)
-            opus_datas.append(opus_data)
-
-        return opus_datas, duration
-
-
-class EdgeTTS(TTS):
-    def __init__(self, config, delete_audio_file):
-        super().__init__(config, delete_audio_file)
-        self.voice = config.get("voice")
-
-    def generate_filename(self, extension=".mp3"):
-        return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
-
-    async def text_to_speak(self, text, output_file):
-        communicate = edge_tts.Communicate(text, voice=self.voice)  # Use your preferred voice
-        await communicate.save(output_file)
-
-
-class DoubaoTTS(TTS):
-    def __init__(self, config, delete_audio_file):
-        super().__init__(config, delete_audio_file)
-        self.appid = config.get("appid")
-        self.access_token = config.get("access_token")
-        self.cluster = config.get("cluster")
-        self.voice = config.get("voice")
-
-        self.host = "openspeech.bytedance.com"
-        self.api_url = f"https://{self.host}/api/v1/tts"
-        self.header = {"Authorization": f"Bearer;{self.access_token}"}
-
-    def generate_filename(self, extension=".wav"):
-        return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
-
-    async def text_to_speak(self, text, output_file):
-        request_json = {
-            "app": {
-                "appid": self.appid,
-                "token": "access_token",
-                "cluster": self.cluster
-            },
-            "user": {
-                "uid": "1"
-            },
-            "audio": {
-                "voice_type": self.voice,
-                "encoding": "wav",
-                "speed_ratio": 1.0,
-                "volume_ratio": 1.0,
-                "pitch_ratio": 1.0,
-            },
-            "request": {
-                "reqid": str(uuid.uuid4()),
-                "text": text,
-                "text_type": "plain",
-                "operation": "query",
-                "with_frontend": 1,
-                "frontend_type": "unitTson"
-            }
-        }
-
-        resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header)
-        if "data" in resp.json():
-            data = resp.json()["data"]
-            file_to_save = open(output_file, "wb")
-            file_to_save.write(base64.b64decode(data))
-
-
 def create_instance(class_name, *args, **kwargs):
-    # 获取类对象
-    cls_map = {
-        "DoubaoTTS": DoubaoTTS,
-        "EdgeTTS": EdgeTTS,
-        # 可扩展其他TTS实现
-    }
+    # 创建TTS实例
+    if os.path.exists(os.path.join('core', 'providers', 'tts', f'{class_name}.py')):
+        lib_name = f'core.providers.tts.{class_name}'
+        if lib_name not in sys.modules:
+            sys.modules[lib_name] = importlib.import_module(f'{lib_name}')
+            return sys.modules[lib_name].TTSProvider(*args, **kwargs)
 
-    if cls := cls_map.get(class_name):
-        return cls(*args, **kwargs)
-    raise ValueError(f"不支持的TTS类型: {class_name}")
+    raise ValueError(f"不支持的TTS类型: {class_name}，请检查该配置的type是否设置正确")
 
 
 if __name__ == "__main__":
@@ -168,7 +35,10 @@ if __name__ == "__main__":
     """
     config = read_config(get_project_dir() + "config.yaml")
     tts = create_instance(
-        config["selected_module"]["TTS"],
+        config["selected_module"]["TTS"]
+        if not 'type' in config["TTS"][config["selected_module"]["TTS"]]
+        else
+        config["TTS"][config["selected_module"]["TTS"]]["type"],
         config["TTS"][config["selected_module"]["TTS"]],
         config["delete_audio"]
     )