Merge branch 'main' into mqtt

main/xiaozhi-server/core/connection.py

@@ -511,10 +511,14 @@ class ConnectionHandler:
         try:
             voiceprint_config = self.config.get("voiceprint", {})
             if voiceprint_config:
-                self.voiceprint_provider = VoiceprintProvider(voiceprint_config)
-                self.logger.bind(tag=TAG).info("声纹识别功能已在连接时动态启用")
+                voiceprint_provider = VoiceprintProvider(voiceprint_config)
+                if voiceprint_provider is not None and voiceprint_provider.enabled:
+                    self.voiceprint_provider = voiceprint_provider
+                    self.logger.bind(tag=TAG).info("声纹识别功能已在连接时动态启用")
+                else:
+                    self.logger.bind(tag=TAG).warning("声纹识别功能启用但配置不完整")
             else:
-                self.logger.bind(tag=TAG).info("声纹识别功能未启用或配置不完整")
+                self.logger.bind(tag=TAG).info("声纹识别功能未启用")
         except Exception as e:
             self.logger.bind(tag=TAG).warning(f"声纹识别初始化失败: {str(e)}")
 

main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py

@@ -0,0 +1,169 @@
+import os
+import json
+import asyncio
+import tempfile
+import difflib
+from typing import Optional, Tuple, List
+import dashscope
+from config.logger import setup_logging
+from core.providers.asr.base import ASRProviderBase
+from core.providers.asr.dto.dto import InterfaceType
+
+tag = __name__
+logger = setup_logging()
+
+
+class ASRProvider(ASRProviderBase):
+    def __init__(self, config: dict, delete_audio_file: bool):
+        super().__init__()
+        self.interface_type = InterfaceType.STREAM
+        """Qwen3-ASR-Flash ASR初始化"""
+        
+        # 配置参数
+        self.api_key = config.get("api_key")
+        if not self.api_key:
+            raise ValueError("Qwen3-ASR-Flash 需要配置 api_key")
+            
+        self.model_name = config.get("model_name", "qwen3-asr-flash")
+        self.output_dir = config.get("output_dir", "./audio_output")
+        self.delete_audio_file = delete_audio_file
+        
+        # ASR选项配置
+        self.enable_lid = config.get("enable_lid", True)  # 自动语种检测
+        self.enable_itn = config.get("enable_itn", True)  # 逆文本归一化
+        self.language = config.get("language", None)  # 指定语种，默认自动检测
+        self.context = config.get("context", "")  # 上下文信息，用于提高识别准确率
+        
+        # 确保输出目录存在
+        os.makedirs(self.output_dir, exist_ok=True)
+
+    def _prepare_audio_file(self, pcm_data: bytes) -> str:
+        """将PCM数据转换为WAV文件并返回文件路径"""
+        try:
+            import wave
+            
+            # 创建临时WAV文件
+            with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file:
+                temp_path = temp_file.name
+                
+            # 写入WAV格式
+            with wave.open(temp_path, 'wb') as wav_file:
+                wav_file.setnchannels(1)      # 单声道
+                wav_file.setsampwidth(2)      # 16位
+                wav_file.setframerate(16000)  # 16kHz采样率
+                wav_file.writeframes(pcm_data)
+                
+            return temp_path
+            
+        except Exception as e:
+            logger.bind(tag=tag).error(f"音频文件准备失败: {e}")
+            return None
+
+    async def speech_to_text(
+        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """将语音数据转换为文本"""
+        temp_file_path = None
+        file_path = None
+        
+        try:
+            # 解码音频数据
+            if audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data)
+            
+            combined_pcm_data = b"".join(pcm_data)
+            if len(combined_pcm_data) == 0:
+                logger.bind(tag=tag).warning("音频数据为空")
+                return "", None
+            
+            # 准备音频文件
+            temp_file_path = self._prepare_audio_file(combined_pcm_data)
+            if not temp_file_path:
+                return "", None
+            
+            # 保存音频文件（如果需要）
+            if not self.delete_audio_file:
+                file_path = self.save_audio_to_file(pcm_data, session_id)
+            
+            # 构造请求消息
+            messages = [
+                {
+                    "role": "user",
+                    "content": [
+                        {"audio": temp_file_path}
+                    ]
+                }
+            ]
+            
+            # 如果有上下文信息，添加system消息
+            if self.context:
+                messages.insert(0, {
+                    "role": "system", 
+                    "content": [
+                        {"text": self.context}
+                    ]
+                })
+            
+            # 准备ASR选项
+            asr_options = {
+                "enable_lid": self.enable_lid,
+                "enable_itn": self.enable_itn
+            }
+            
+            # 如果指定了语种，添加到选项中
+            if self.language:
+                asr_options["language"] = self.language
+            
+            # 设置API密钥
+            dashscope.api_key = self.api_key
+            
+            # 发送流式请求
+            response = dashscope.MultiModalConversation.call(
+                model=self.model_name,
+                messages=messages,
+                result_format="message",
+                asr_options=asr_options,
+                stream=True
+            )
+            
+            # 处理流式响应
+            full_text = ""
+            last_text = ""  # 用于存储上一个文本片段
+            for chunk in response:
+                try:
+                    text = chunk["output"]["choices"][0]["message"].content[0]["text"]
+                    # 标准化文本片段（去除首尾空格）
+                    normalized_text = text.strip()
+                    # 只有当新文本片段与上一个不同时才处理
+                    if normalized_text != last_text:
+                        # 提取新增的文本部分
+                        # 通过比较当前文本和上一个文本，找到新增的部分
+                        if normalized_text.startswith(last_text):
+                            # 如果当前文本以最后一个文本开头，则新增部分是两者的差集
+                            new_part = normalized_text[len(last_text):]
+                        else:
+                            # 如果不以最后一个文本开头，说明识别结果发生了较大变化，直接使用当前文本
+                            new_part = normalized_text
+                        
+                        # 将新增部分添加到完整文本中
+                        full_text += new_part
+                        last_text = normalized_text
+                    # 这里可以实时处理文本片段，例如通过回调函数
+                except:
+                    pass
+            
+            return full_text, file_path
+                
+        except Exception as e:
+            logger.bind(tag=tag).error(f"语音识别失败: {e}")
+            return "", file_path
+            
+        finally:
+            # 清理临时文件
+            if temp_file_path and os.path.exists(temp_file_path):
+                try:
+                    os.unlink(temp_file_path)
+                except Exception as e:
+                    logger.bind(tag=tag).warning(f"清理临时文件失败: {e}")

main/xiaozhi-server/core/providers/asr/vosk.py

@@ -0,0 +1,114 @@
+import os
+import json
+import time
+from typing import Optional, Tuple, List
+from .base import ASRProviderBase
+from config.logger import setup_logging
+from core.providers.asr.dto.dto import InterfaceType
+import vosk
+
+TAG = __name__
+logger = setup_logging()
+
+class ASRProvider(ASRProviderBase):
+    def __init__(self, config: dict, delete_audio_file: bool = True):
+        super().__init__()
+        self.interface_type = InterfaceType.LOCAL
+        self.model_path = config.get("model_path")
+        self.output_dir = config.get("output_dir", "tmp/")
+        self.delete_audio_file = delete_audio_file
+        
+        # 初始化VOSK模型
+        self.model = None
+        self.recognizer = None
+        self._load_model()
+        
+        # 确保输出目录存在
+        os.makedirs(self.output_dir, exist_ok=True)
+
+    def _load_model(self):
+        """加载VOSK模型"""
+        try:
+            if not os.path.exists(self.model_path):
+                raise FileNotFoundError(f"VOSK模型路径不存在: {self.model_path}")
+                
+            logger.bind(tag=TAG).info(f"正在加载VOSK模型: {self.model_path}")
+            self.model = vosk.Model(self.model_path)
+
+            # 初始化VOSK识别器（采样率必须为16kHz）
+            self.recognizer = vosk.KaldiRecognizer(self.model, 16000)
+
+            logger.bind(tag=TAG).info("VOSK模型加载成功")
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"加载VOSK模型失败: {e}")
+            raise
+
+    async def speech_to_text(
+        self, audio_data: List[bytes], session_id: str, audio_format: str = "opus"
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """将语音数据转换为文本"""
+        file_path = None
+        try:
+            # 检查模型是否加载成功
+            if not self.model:
+                logger.bind(tag=TAG).error("VOSK模型未加载，无法进行识别")
+                return "", None
+
+            # 解码音频（如果原始格式是Opus）
+            if audio_format == "pcm":
+                pcm_data = audio_data
+            else:
+                pcm_data = self.decode_opus(audio_data)
+                
+            if not pcm_data:
+                logger.bind(tag=TAG).warning("解码后的PCM数据为空，无法进行识别")
+                return "", None
+                
+            # 合并PCM数据
+            combined_pcm_data = b"".join(pcm_data)
+            if len(combined_pcm_data) == 0:
+                logger.bind(tag=TAG).warning("合并后的PCM数据为空")
+                return "", None
+
+            # 判断是否保存为WAV文件
+            if not self.delete_audio_file:
+                file_path = self.save_audio_to_file(pcm_data, session_id)
+
+            start_time = time.time()
+            
+            
+            # 进行识别（VOSK推荐每次送入2000字节的数据）
+            chunk_size = 2000
+            text_result = ""
+            
+            for i in range(0, len(combined_pcm_data), chunk_size):
+                chunk = combined_pcm_data[i:i+chunk_size]
+                if self.recognizer.AcceptWaveform(chunk):
+                    result = json.loads(self.recognizer.Result())
+                    text = result.get('text', '')
+                    if text:
+                        text_result += text + " "
+            
+            # 获取最终结果
+            final_result = json.loads(self.recognizer.FinalResult())
+            final_text = final_result.get('text', '')
+            if final_text:
+                text_result += final_text
+            
+            logger.bind(tag=TAG).debug(
+                f"VOSK语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text_result.strip()}"
+            )
+            
+            return text_result.strip(), file_path
+            
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"VOSK语音识别失败: {e}")
+            return "", None
+        finally:
+            # 文件清理逻辑
+            if self.delete_audio_file and file_path and os.path.exists(file_path):
+                try:
+                    os.remove(file_path)
+                    logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_client.py

@@ -167,14 +167,13 @@ class ServerMCPClient:
 
                 # 建立SSEClient
                 elif "url" in self.config:
+                    headers = dict(self.config.get("headers", {}))
+                    # TODO 兼容旧版本
                     if "API_ACCESS_TOKEN" in self.config:
-                        headers = {
-                            "Authorization": f"Bearer {self.config['API_ACCESS_TOKEN']}"
-                        }
-                    else:
-                        headers = {}
+                        headers["Authorization"] = f"Bearer {self.config['API_ACCESS_TOKEN']}"
+                        self.logger.bind(tag=TAG).warning(f"你正在使用旧过时的配置 API_ACCESS_TOKEN ，请在.mcp_server_settings.json中将API_ACCESS_TOKEN直接设置在headers中，例如 'Authorization': 'Bearer API_ACCESS_TOKEN'")
                     sse_r, sse_w = await stack.enter_async_context(
-                        sse_client(self.config["url"], headers=headers)
+                        sse_client(self.config["url"], headers=headers, timeout=self.config.get("timeout", 5), sse_read_timeout=self.config.get("sse_read_timeout", 60 * 5))
                     )
                     read_stream, write_stream = sse_r, sse_w
 

main/xiaozhi-server/core/providers/tts/minimax.py

@@ -1,95 +0,0 @@
-import os
-import uuid
-import json
-import requests
-from datetime import datetime
-from core.providers.tts.base import TTSProviderBase
-from core.utils.util import parse_string_to_list
-
-
-class TTSProvider(TTSProviderBase):
-    def __init__(self, config, delete_audio_file):
-        super().__init__(config, delete_audio_file)
-        self.group_id = config.get("group_id")
-        self.api_key = config.get("api_key")
-        self.model = config.get("model")
-        if config.get("private_voice"):
-            self.voice = config.get("private_voice")
-        else:
-            self.voice = config.get("voice_id")
-
-        default_voice_setting = {
-            "voice_id": "female-shaonv",
-            "speed": 1,
-            "vol": 1,
-            "pitch": 0,
-            "emotion": "happy",
-        }
-        default_pronunciation_dict = {"tone": ["处理/(chu3)(li3)", "危险/dangerous"]}
-        defult_audio_setting = {
-            "sample_rate": 32000,
-            "bitrate": 128000,
-            "format": "mp3",
-            "channel": 1,
-        }
-        self.voice_setting = {
-            **default_voice_setting,
-            **config.get("voice_setting", {}),
-        }
-        self.pronunciation_dict = {
-            **default_pronunciation_dict,
-            **config.get("pronunciation_dict", {}),
-        }
-        self.audio_setting = {**defult_audio_setting, **config.get("audio_setting", {})}
-        self.timber_weights = parse_string_to_list(config.get("timber_weights"))
-
-        if self.voice:
-            self.voice_setting["voice_id"] = self.voice
-
-        self.host = "api.minimax.chat"
-        self.api_url = f"https://{self.host}/v1/t2a_v2?GroupId={self.group_id}"
-        self.header = {
-            "Content-Type": "application/json",
-            "Authorization": f"Bearer {self.api_key}",
-        }
-        self.audio_file_type = defult_audio_setting.get("format", "mp3")
-
-    def generate_filename(self, extension=".mp3"):
-        return os.path.join(
-            self.output_file,
-            f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
-        )
-
-    async def text_to_speak(self, text, output_file):
-        request_json = {
-            "model": self.model,
-            "text": text,
-            "stream": False,
-            "voice_setting": self.voice_setting,
-            "pronunciation_dict": self.pronunciation_dict,
-            "audio_setting": self.audio_setting,
-        }
-
-        if type(self.timber_weights) is list and len(self.timber_weights) > 0:
-            request_json["timber_weights"] = self.timber_weights
-            request_json["voice_setting"]["voice_id"] = ""
-
-        try:
-            resp = requests.post(
-                self.api_url, json.dumps(request_json), headers=self.header
-            )
-            # 检查返回请求数据的status_code是否为0
-            if resp.json()["base_resp"]["status_code"] == 0:
-                data = resp.json()["data"]["audio"]
-                audio_bytes = bytes.fromhex(data)
-                if output_file:
-                    with open(output_file, "wb") as file_to_save:
-                        file_to_save.write(audio_bytes)
-                else:
-                    return audio_bytes
-            else:
-                raise Exception(
-                    f"{__name__} status_code: {resp.status_code} response: {resp.content}"
-                )
-        except Exception as e:
-            raise Exception(f"{__name__} error: {e}")

main/xiaozhi-server/core/providers/tts/minimax_httpstream.py

@@ -1,11 +1,20 @@
 import os
-import uuid
 import json
+import time
+import queue
+import asyncio
+import aiohttp
 import requests
-from datetime import datetime
-from typing import Iterator, Optional, Union
-from core.providers.tts.base import TTSProviderBase
+import traceback
+from config.logger import setup_logging
+from core.utils.tts import MarkdownCleaner
 from core.utils.util import parse_string_to_list
+from core.providers.tts.base import TTSProviderBase
+from core.utils import opus_encoder_utils, textUtils
+from core.providers.tts.dto.dto import SentenceType, ContentType
+
+TAG = __name__
+logger = setup_logging()
 
 
 class TTSProvider(TTSProviderBase):
@@ -28,9 +37,9 @@ class TTSProvider(TTSProviderBase):
         }
         default_pronunciation_dict = {"tone": ["处理/(chu3)(li3)", "危险/dangerous"]}
         defult_audio_setting = {
-            "sample_rate": 32000,
+            "sample_rate": 24000,
             "bitrate": 128000,
-            "format": "mp3",
+            "format": "pcm",
             "channel": 1,
         }
         self.voice_setting = {
@@ -47,66 +56,101 @@ class TTSProvider(TTSProviderBase):
         if self.voice:
             self.voice_setting["voice_id"] = self.voice
 
-        self.host = "api.minimax.chat"
+        self.host = "api.minimaxi.com"  # 备用地址：api-bj.minimaxi.com
         self.api_url = f"https://{self.host}/v1/t2a_v2?GroupId={self.group_id}"
         self.header = {
             "Content-Type": "application/json",
             "Authorization": f"Bearer {self.api_key}",
         }
-        self.audio_file_type = defult_audio_setting.get("format", "mp3")
+        self.audio_file_type = defult_audio_setting.get("format", "pcm")
 
-    def generate_filename(self, extension=".mp3"):
-        return os.path.join(
-            self.output_file,
-            f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
+        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+            sample_rate=24000, channels=1, frame_size_ms=60
         )
 
-    async def text_to_speak(self, text, output_file):
-        """非流式语音合成（保留原有实现）"""
-        request_json = {
-            "model": self.model,
-            "text": text,
-            "stream": False,
-            "voice_setting": self.voice_setting,
-            "pronunciation_dict": self.pronunciation_dict,
-            "audio_setting": self.audio_setting,
-        }
+        # PCM缓冲区
+        self.pcm_buffer = bytearray()
 
-        if type(self.timber_weights) is list and len(self.timber_weights) > 0:
-            request_json["timber_weights"] = self.timber_weights
-            request_json["voice_setting"]["voice_id"] = ""
+    def tts_text_priority_thread(self):
+        """流式文本处理线程"""
+        while not self.conn.stop_event.is_set():
+            try:
+                message = self.tts_text_queue.get(timeout=1)
+                if message.sentence_type == SentenceType.FIRST:
+                    # 初始化参数
+                    self.tts_stop_request = False
+                    self.processed_chars = 0
+                    self.tts_text_buff = []
+                    self.before_stop_play_files.clear()
+                elif ContentType.TEXT == message.content_type:
+                    self.tts_text_buff.append(message.content_detail)
+                    segment_text = self._get_segment_text()
+                    if segment_text:
+                        self.to_tts_single_stream(segment_text)
 
-        try:
-            resp = requests.post(
-                self.api_url, json.dumps(request_json), headers=self.header
-            )
-            if resp.json()["base_resp"]["status_code"] == 0:
-                data = resp.json()["data"]["audio"]
-                audio_bytes = bytes.fromhex(data)
-                if output_file:
-                    with open(output_file, "wb") as file_to_save:
-                        file_to_save.write(audio_bytes)
-                else:
-                    return audio_bytes
+                elif ContentType.FILE == message.content_type:
+                    logger.bind(tag=TAG).info(
+                        f"添加音频文件到待播放列表: {message.content_file}"
+                    )
+                    if message.content_file and os.path.exists(message.content_file):
+                        # 先处理文件音频数据
+                        self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
+                if message.sentence_type == SentenceType.LAST:
+                    # 处理剩余的文本
+                    self._process_remaining_text_stream(True)
+
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logger.bind(tag=TAG).error(
+                    f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
+                )
+
+    def _process_remaining_text_stream(self, is_last=False):
+        """处理剩余的文本并生成语音
+        Returns:
+            bool: 是否成功处理了文本
+        """
+        full_text = "".join(self.tts_text_buff)
+        remaining_text = full_text[self.processed_chars :]
+        if remaining_text:
+            segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
+            if segment_text:
+                self.to_tts_single_stream(segment_text, is_last)
+                self.processed_chars += len(full_text)
             else:
-                raise Exception(
-                    f"{__name__} status_code: {resp.status_code} response: {resp.content}"
+                self._process_before_stop_play_files()
+        else:
+            self._process_before_stop_play_files()
+
+    def to_tts_single_stream(self, text, is_last=False):
+        try:
+            max_repeat_time = 5
+            text = MarkdownCleaner.clean_markdown(text)
+            try:
+                asyncio.run(self.text_to_speak(text, is_last))
+            except Exception as e:
+                logger.bind(tag=TAG).warning(
+                    f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
+                )
+                max_repeat_time -= 1
+
+            if max_repeat_time > 0:
+                logger.bind(tag=TAG).info(
+                    f"语音生成成功: {text}，重试{5 - max_repeat_time}次"
+                )
+            else:
+                logger.bind(tag=TAG).error(
+                    f"语音生成失败: {text}，请检查网络或服务是否正常"
                 )
         except Exception as e:
-            raise Exception(f"{__name__} error: {e}")
+            logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
+        finally:
+            return None
 
-    def text_to_speak_stream(
-        self, 
-        text: str, 
-        chunk_callback: Optional[callable] = None
-    ) -> Iterator[bytes]:
-        """
-        流式语音合成方法
-        :param text: 要合成的文本
-        :param chunk_callback: 可选的回调函数，用于处理每个音频块
-        :return: 生成器，每次产生一个音频数据块(bytes)
-        """
-        request_json = {
+    async def text_to_speak(self, text, is_last):
+        """流式处理TTS音频，每句只推送一次音频列表"""
+        payload = {
             "model": self.model,
             "text": text,
             "stream": True,
@@ -115,116 +159,183 @@ class TTSProvider(TTSProviderBase):
             "audio_setting": self.audio_setting,
         }
 
-        if isinstance(self.timber_weights, list) and len(self.timber_weights) > 0:
-            request_json["timber_weights"] = self.timber_weights
-            request_json["voice_setting"]["voice_id"] = ""
+        if type(self.timber_weights) is list and len(self.timber_weights) > 0:
+            payload["timber_weights"] = self.timber_weights
+            payload["voice_setting"]["voice_id"] = ""
+
+        frame_bytes = int(
+            self.opus_encoder.sample_rate
+            * self.opus_encoder.channels  # 1
+            * self.opus_encoder.frame_size_ms
+            / 1000
+            * 2
+        )  # 16-bit = 2 bytes
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    self.api_url,
+                    headers=self.header,
+                    data=json.dumps(payload),
+                    timeout=10,
+                ) as resp:
+
+                    if resp.status != 200:
+                        logger.bind(tag=TAG).error(
+                            f"TTS请求失败: {resp.status}, {await resp.text()}"
+                        )
+                        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                        return
+
+                    self.pcm_buffer.clear()
+                    self.tts_audio_queue.put((SentenceType.FIRST, [], text))
+
+                    # 处理音频流数据
+                    buffer = b""
+                    async for chunk in resp.content.iter_any():
+                        if not chunk:
+                            continue
+
+                        buffer += chunk
+                        while True:
+                            # 查找数据块分隔符
+                            header_pos = buffer.find(b"data: ")
+                            if header_pos == -1:
+                                break
+
+                            end_pos = buffer.find(b"\n\n", header_pos)
+                            if end_pos == -1:
+                                break
+
+                            # 提取单个完整JSON块
+                            json_str = buffer[header_pos + 6 : end_pos].decode("utf-8")
+                            buffer = buffer[end_pos + 2 :]
+
+                            try:
+                                data = json.loads(json_str)
+                                status = data.get("data", {}).get("status", 1)
+                                audio_hex = data.get("data", {}).get("audio")
+
+                                # 仅处理status=1的有效音频块 忽略status=2的结束汇总块
+                                if status == 1 and audio_hex:
+                                    pcm_data = bytes.fromhex(audio_hex)
+                                    self.pcm_buffer.extend(pcm_data)
+
+                            except json.JSONDecodeError as e:
+                                logger.bind(tag=TAG).error(f"JSON解析失败: {e}")
+                                continue
+
+                        while len(self.pcm_buffer) >= frame_bytes:
+                            frame = bytes(self.pcm_buffer[:frame_bytes])
+                            del self.pcm_buffer[:frame_bytes]
+
+                            self.opus_encoder.encode_pcm_to_opus_stream(
+                                frame, end_of_stream=False, callback=self.handle_opus
+                            )
+
+                    # flush 剩余不足一帧的数据
+                    if self.pcm_buffer:
+                        self.opus_encoder.encode_pcm_to_opus_stream(
+                            bytes(self.pcm_buffer),
+                            end_of_stream=True,
+                            callback=self.handle_opus,
+                        )
+                        self.pcm_buffer.clear()
+
+                    # 如果是最后一段，输出音频获取完毕
+                    if is_last:
+                        self._process_before_stop_play_files()
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            self.tts_audio_queue.put((SentenceType.LAST, [], None))
+
+    async def close(self):
+        """资源清理"""
+        await super().close()
+        if hasattr(self, "opus_encoder"):
+            self.opus_encoder.close()
+
+    def to_tts(self, text: str) -> list:
+        """非流式TTS处理，用于测试及保存音频文件的场景
+        Args:
+            text: 要转换的文本
+        Returns:
+            list: 返回opus编码后的音频数据列表
+        """
+        start_time = time.time()
+        text = MarkdownCleaner.clean_markdown(text)
+
+        payload = {
+            "model": self.model,
+            "text": text,
+            "stream": True,
+            "voice_setting": self.voice_setting,
+            "pronunciation_dict": self.pronunciation_dict,
+            "audio_setting": self.audio_setting,
+        }
+
+        if type(self.timber_weights) is list and len(self.timber_weights) > 0:
+            payload["timber_weights"] = self.timber_weights
+            payload["voice_setting"]["voice_id"] = ""
+
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+        }
 
         try:
             with requests.post(
-                self.api_url,
-                data=json.dumps(request_json),
-                headers=self.header,
-                stream=True
+                self.api_url, data=json.dumps(payload), headers=headers, timeout=5
             ) as response:
-                
-                # 检查HTTP状态码
                 if response.status_code != 200:
-                    raise Exception(
-                        f"HTTP error: {response.status_code}, response: {response.text}"
+                    logger.bind(tag=TAG).error(
+                        f"TTS请求失败: {response.status_code}, {response.text}"
                     )
-                
-                # 处理流式响应
-                for line in response.iter_lines():
-                    if line:  # 过滤空行
-                        # 检查是否为数据行 (SSE格式)
-                        if line.startswith(b'data:'):
-                            try:
-                                data = json.loads(line[5:].strip())  # 去掉"data:"前缀
-                                
-                                # 检查API状态码
-                                if data.get("base_resp", {}).get("status_code", -1) != 0:
-                                    raise Exception(
-                                        f"API error: {data.get('base_resp', {}).get('status_msg')}"
-                                    )
-                                
-                                # 跳过非音频数据块
-                                if "extra_info" in data:
-                                    continue
-                                    
-                                # 提取音频数据
-                                audio_hex = data.get("data", {}).get("audio")
-                                if audio_hex:
-                                    audio_chunk = bytes.fromhex(audio_hex)
-                                    if chunk_callback:
-                                        chunk_callback(audio_chunk)
-                                    yield audio_chunk
-                                    
-                            except json.JSONDecodeError:
-                                # 忽略JSON解析错误（可能是心跳包等）
-                                continue
-                            except Exception as e:
-                                raise e
-                                
-        except Exception as e:
-            raise Exception(f"{__name__} stream error: {e}")
+                    return []
 
-    def save_stream_to_file(
-        self, 
-        text: str, 
-        output_file: Optional[str] = None,
-        progress_callback: Optional[callable] = None
-    ) -> str:
-        """
-        流式合成并保存到文件
-        :param text: 要合成的文本
-        :param output_file: 输出文件路径，如果为None则自动生成
-        :param progress_callback: 可选的回调函数，接收已写入的字节数
-        :return: 保存的文件路径
-        """
-        if not output_file:
-            output_file = self.generate_filename(extension=f".{self.audio_file_type}")
-        
-        os.makedirs(os.path.dirname(output_file), exist_ok=True)
-        
-        total_bytes = 0
-        try:
-            with open(output_file, "wb") as audio_file:
-                for audio_chunk in self.text_to_speak_stream(text):
-                    audio_file.write(audio_chunk)
-                    audio_file.flush()
-                    total_bytes += len(audio_chunk)
-                    if progress_callback:
-                        progress_callback(total_bytes)
-            return output_file
-        except Exception as e:
-            # 清理可能创建的不完整文件
-            if os.path.exists(output_file):
-                os.remove(output_file)
-            raise e
+                logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
+
+                # 使用opus编码器处理PCM数据
+                opus_datas = []
+                full_content = response.content.decode('utf-8')
+                pcm_data = bytearray()
+                for data_block in full_content.split('\n\n'):
+                    if not data_block.startswith('data: '):
+                        continue
+
+                    try:
+                        json_str = data_block[6:]  # 去除'data: '前缀
+                        data = json.loads(json_str)
+                        if data.get('data', {}).get('status') == 1:
+                            audio_hex = data['data']['audio']
+                            pcm_data.extend(bytes.fromhex(audio_hex))
+                    except (json.JSONDecodeError, KeyError) as e:
+                        logger.bind(tag=TAG).warning(f"无效数据块: {e}")
+                        continue
+
+                # 计算每帧的字节数
+                frame_bytes = int(
+                    self.opus_encoder.sample_rate
+                    * self.opus_encoder.channels
+                    * self.opus_encoder.frame_size_ms
+                    / 1000
+                    * 2
+                )
+
+                # 分帧处理合并后的PCM数据
+                for i in range(0, len(pcm_data), frame_bytes):
+                    frame = bytes(pcm_data[i:i+frame_bytes])
+                    if len(frame) < frame_bytes:
+                        frame += b"\x00" * (frame_bytes - len(frame))
+ 
+                    self.opus_encoder.encode_pcm_to_opus_stream(
+                        frame,
+                        end_of_stream=(i + frame_bytes >= len(pcm_data)),
+                        callback=lambda opus: opus_datas.append(opus)
+                    )
+
+                return opus_datas
 
-    def stream_to_audio_player(self, text: str, player_command: list = None):
-        """
-        流式合成并直接播放音频
-        :param text: 要合成的文本
-        :param player_command: 音频播放器命令，默认使用mpv
-        """
-        if player_command is None:
-            player_command = ["mpv", "--no-cache", "--no-terminal", "--", "fd://0"]
-        
-        try:
-            import subprocess
-            player_process = subprocess.Popen(
-                player_command,
-                stdin=subprocess.PIPE,
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-            )
-            
-            for audio_chunk in self.text_to_speak_stream(text):
-                player_process.stdin.write(audio_chunk)
-                player_process.stdin.flush()
-                
-            player_process.stdin.close()
-            player_process.wait()
         except Exception as e:
-            raise Exception(f"Audio player error: {e}")
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            return []

main/xiaozhi-server/core/providers/tts/minimax_webSocket.py

@@ -1,180 +0,0 @@
-import os
-import uuid
-import json
-import asyncio
-import websockets
-import ssl
-from datetime import datetime
-from core.providers.tts.base import TTSProviderBase
-from core.utils.util import parse_string_to_list
-
-
-class TTSProvider(TTSProviderBase):
-    def __init__(self, config, delete_audio_file):
-        super().__init__(config, delete_audio_file)
-        self.group_id = config.get("group_id")
-        self.api_key = config.get("api_key")
-        self.model = config.get("model")
-        
-        # 初始化语音设置
-        default_voice_setting = {
-            "voice_id": "female-shaonv",
-            "speed": 1,
-            "vol": 1,
-            "pitch": 0,
-            "emotion": "happy",
-        }
-        default_pronunciation_dict = {"tone": ["处理/(chu3)(li3)", "危险/dangerous"]}
-        default_audio_setting = {
-            "sample_rate": 32000,
-            "bitrate": 128000,
-            "format": "mp3",
-            "channel": 1,
-        }
-        
-        # 合并配置
-        self.voice_setting = {
-            **default_voice_setting,
-            **config.get("voice_setting", {}),
-        }
-        self.pronunciation_dict = {
-            **default_pronunciation_dict,
-            **config.get("pronunciation_dict", {}),
-        }
-        self.audio_setting = {
-            **default_audio_setting, 
-            **config.get("audio_setting", {})
-        }
-        self.timber_weights = parse_string_to_list(config.get("timber_weights"))
-        
-        # 设置语音ID
-        if config.get("private_voice"):
-            self.voice_setting["voice_id"] = config.get("private_voice")
-        elif config.get("voice_id"):
-            self.voice_setting["voice_id"] = config.get("voice_id")
-        
-        # WebSocket配置
-        self.ws_url = "wss://api.minimaxi.com/ws/v1/t2a_v2"
-        self.headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "GroupId": self.group_id
-        }
-        self.audio_file_type = self.audio_setting.get("format", "mp3")
-
-    def generate_filename(self, extension=".mp3"):
-        """生成唯一的音频文件名"""
-        return os.path.join(
-            self.output_file,
-            f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
-        )
-
-    async def _establish_connection(self):
-        """建立WebSocket连接"""
-        ssl_context = ssl.create_default_context()
-        ssl_context.check_hostname = False
-        ssl_context.verify_mode = ssl.CERT_NONE
-
-        try:
-            ws = await websockets.connect(
-                self.ws_url,
-                additional_headers=self.headers,
-                ssl=ssl_context
-            )
-            connected = json.loads(await ws.recv())
-            if connected.get("event") == "connected_success":
-                print("连接成功")
-                return ws
-            return None
-        except Exception as e:
-            print(f"连接失败: {e}")
-            return None
-
-    async def _start_task(self, websocket):
-        """发送任务开始请求"""
-        start_msg = {
-            "event": "task_start",
-            "model": self.model,
-            "voice_setting": self.voice_setting,
-            "pronunciation_dict": self.pronunciation_dict,
-            "audio_setting": self.audio_setting
-        }
-        
-        if self.timber_weights and len(self.timber_weights) > 0:
-            start_msg["timber_weights"] = self.timber_weights
-            start_msg["voice_setting"]["voice_id"] = ""
-
-        await websocket.send(json.dumps(start_msg))
-        response = json.loads(await websocket.recv())
-        return response.get("event") == "task_started"
-
-    async def _continue_task(self, websocket, text):
-        """发送继续请求并收集音频数据"""
-        await websocket.send(json.dumps({
-            "event": "task_continue",
-            "text": text
-        }))
-
-        audio_chunks = []
-        while True:
-            response = json.loads(await websocket.recv())
-            if "data" in response and "audio" in response["data"]:
-                audio_chunks.append(response["data"]["audio"])
-            if response.get("is_final"):
-                break
-        return "".join(audio_chunks)
-
-    async def _close_connection(self, websocket):
-        """关闭连接"""
-        if websocket:
-            await websocket.send(json.dumps({"event": "task_finish"}))
-            await websocket.close()
-            print("连接已关闭")
-
-    async def text_to_speak(self, text, output_file=None):
-        """主方法：文本转语音"""
-        ws = await self._establish_connection()
-        if not ws:
-            raise Exception("无法建立WebSocket连接")
-
-        try:
-            if not await self._start_task(ws):
-                raise Exception("任务启动失败")
-
-            hex_audio = await self._continue_task(ws, text)
-            audio_bytes = bytes.fromhex(hex_audio)
-
-            # 保存到文件或返回二进制数据
-            if output_file:
-                with open(output_file, "wb") as f:
-                    f.write(audio_bytes)
-                print(f"音频已保存为{output_file}")
-                return output_file
-            else:
-                # 返回音频二进制数据（不播放）
-                return audio_bytes
-
-        finally:
-            await self._close_connection(ws)
-
-
-async def main():
-    """测试用主函数"""
-    # 示例配置
-    config = {
-        "group_id": "YOUR_GROUP_ID",  # 替换为实际的group_id
-        "api_key": "YOUR_API_KEY",    # 替换为实际的api_key
-        "model": "your-model",        # 替换为实际的模型名称
-        "voice_id": "male-qn-qingse",
-        "voice_setting": {
-            "speed": 1.2,
-            "emotion": "happy"
-        }
-    }
-    
-    tts = TTSProvider(config, delete_audio_file=True)
-    output_file = tts.generate_filename()
-    await tts.text_to_speak("这是一个测试文本，用于验证流式语音合成功能", output_file)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

main/xiaozhi-server/core/utils/voiceprint_provider.py

@@ -19,6 +19,8 @@ class VoiceprintProvider:
         self.original_url = config.get("url", "")
         self.speakers = config.get("speakers", [])
         self.speaker_map = self._parse_speakers()
+        # 声纹识别相似度阈值，默认0.4
+        self.similarity_threshold = float(config.get("similarity_threshold", 0.4))
         
         # 解析API地址和密钥
         self.api_url = None
@@ -62,7 +64,7 @@ class VoiceprintProvider:
                     # 进行健康检查，验证服务器是否可用
                     if self._check_server_health():
                         self.enabled = True
-                        logger.bind(tag=TAG).info(f"声纹识别已启用: API={self.api_url}, 说话人={len(self.speaker_ids)}个")
+                        logger.bind(tag=TAG).info(f"声纹识别已启用: API={self.api_url}, 说话人={len(self.speaker_ids)}个, 相似度阈值={self.similarity_threshold}")
                     else:
                         self.enabled = False
                         logger.bind(tag=TAG).warning(f"声纹识别服务器不可用，声纹识别已禁用: {self.api_url}")
@@ -169,12 +171,14 @@ class VoiceprintProvider:
                         
                         logger.bind(tag=TAG).info(f"声纹识别耗时: {total_elapsed_time:.3f}s")
                         
-                        # 置信度检查
-                        if score < 0.5:
-                            logger.bind(tag=TAG).warning(f"声纹识别置信度较低: {score:.3f}")
+                        # 相似度阈值检查
+                        if score < self.similarity_threshold:
+                            logger.bind(tag=TAG).warning(f"声纹识别相似度{score:.3f}低于阈值{self.similarity_threshold}")
+                            return "未知说话人"
                         
                         if speaker_id and speaker_id in self.speaker_map:
                             result_name = self.speaker_map[speaker_id]["name"]
+                            logger.bind(tag=TAG).info(f"声纹识别成功: {result_name} (相似度: {score:.3f})")
                             return result_name
                         else:
                             logger.bind(tag=TAG).warning(f"未识别的说话人ID: {speaker_id}")

main/xiaozhi-server/core/utils/wakeup_word.py

@@ -55,7 +55,7 @@ class WakeupWordsConfig:
             return self._config_cache
 
         try:
-            with open(self.config_file, "a+") as f:
+            with open(self.config_file, "a+", encoding="utf-8") as f:
                 with FileLock(f, timeout=self._lock_timeout):
                     f.seek(0)
                     content = f.read()
@@ -73,7 +73,7 @@ class WakeupWordsConfig:
     def _save_config(self, config: Dict):
         """保存配置到文件，使用文件锁保护"""
         try:
-            with open(self.config_file, "w") as f:
+            with open(self.config_file, "w", encoding="utf-8") as f:
                 with FileLock(f, timeout=self._lock_timeout):
                     yaml.dump(config, f, allow_unicode=True)
                     self._config_cache = config