Merge branch 'py_test_typing' into feature/python-typing

main/xiaozhi-server/core/api/vision_handler.py

@@ -29,7 +29,15 @@ class VisionHandler(BaseHandler):
 
     def _verify_auth_token(self, request) -> Tuple[bool, Optional[str]]:
         """验证认证token"""
+        # 测试模式：允许特定测试令牌或跳过验证
         auth_header = request.headers.get("Authorization", "")
+        client_id = request.headers.get("Client-Id", "")
+
+        # 允许测试客户端跳过认证
+        if client_id == "web_test_client":
+            device_id = request.headers.get("Device-Id", "test_device")
+            return True, device_id
+
         if not auth_header.startswith("Bearer "):
             return False, None
 

main/xiaozhi-server/core/connection.py

@@ -40,8 +40,10 @@ from config.logger import setup_logging, build_module_string, create_connection_
 from config.manage_api_client import DeviceNotFoundException, DeviceBindException
 from core.utils.prompt_manager import PromptManager
 from core.utils.voiceprint_provider import VoiceprintProvider
+from core.utils.util import get_system_error_response
 from core.utils import textUtils
 
+
 TAG = __name__
 
 auto_import_modules("plugins_func.functions")
@@ -85,6 +87,7 @@ class ConnectionHandler:
         self.max_output_size = 0
         self.chat_history_conf = 0
         self.audio_format = "opus"
+        self.sample_rate = 24000  # 默认采样率，从客户端 hello 消息中动态更新
 
         # 客户端状态相关
         self.client_abort = False
@@ -134,7 +137,6 @@ class ConnectionHandler:
         self.current_language_tag = None  # 存储当前ASR识别的语言标签
 
         # llm相关变量
-        self.llm_finish_task = True
         self.dialogue = Dialogue()
 
         # tts相关变量
@@ -206,6 +208,10 @@ class ConnectionHandler:
             self.welcome_msg = self.config["xiaozhi"]
             self.welcome_msg["session_id"] = self.session_id
 
+            # 从配置中读取采样率
+            self.sample_rate = self.welcome_msg["audio_params"]["sample_rate"]
+            self.logger.bind(tag=TAG).info(f"配置输出音频采样率为: {self.sample_rate}")
+
             # 在后台初始化配置和组件（完全不阻塞主循环）
             asyncio.create_task(self._background_initialize())
 
@@ -794,7 +800,6 @@ class ConnectionHandler:
 
         # 为最顶层时新建会话ID和发送FIRST请求
         if depth == 0:
-            self.llm_finish_task = False
             self.sentence_id = str(uuid.uuid4().hex)
             self.dialogue.put(Message(role="user", content=query))
             self.tts.tts_text_queue.put(
@@ -869,46 +874,66 @@ class ConnectionHandler:
         content_arguments = ""
         self.client_abort = False
         emotion_flag = True
-        for response in llm_responses:
-            if self.client_abort:
-                break
-            if self.intent_type == "function_call" and functions is not None:
-                content, tools_call = response
-                if "content" in response:
-                    content = response["content"]
-                    tools_call = None
-                if content is not None and len(content) > 0:
-                    content_arguments += content
+        try:
+            for response in llm_responses:
+                if self.client_abort:
+                    break
+                if self.intent_type == "function_call" and functions is not None:
+                    content, tools_call = response
+                    if "content" in response:
+                        content = response["content"]
+                        tools_call = None
+                    if content is not None and len(content) > 0:
+                        content_arguments += content
 
-                if not tool_call_flag and content_arguments.startswith("<tool_call>"):
-                    # print("content_arguments", content_arguments)
-                    tool_call_flag = True
+                    if not tool_call_flag and content_arguments.startswith("<tool_call>"):
+                        # print("content_arguments", content_arguments)
+                        tool_call_flag = True
 
-                if tools_call is not None and len(tools_call) > 0:
-                    tool_call_flag = True
-                    self._merge_tool_calls(tool_calls_list, tools_call)
-            else:
-                content = response
+                    if tools_call is not None and len(tools_call) > 0:
+                        tool_call_flag = True
+                        self._merge_tool_calls(tool_calls_list, tools_call)
+                else:
+                    content = response
 
-            # 在llm回复中获取情绪表情，一轮对话只在开头获取一次
-            if emotion_flag and content is not None and content.strip():
-                asyncio.run_coroutine_threadsafe(
-                    textUtils.get_emotion(self, content),
-                    self.loop,
-                )
-                emotion_flag = False
-
-            if content is not None and len(content) > 0:
-                if not tool_call_flag:
-                    response_message.append(content)
-                    self.tts.tts_text_queue.put(
-                        TTSMessageDTO(
-                            sentence_id=self.sentence_id,
-                            sentence_type=SentenceType.MIDDLE,
-                            content_type=ContentType.TEXT,
-                            content_detail=content,
-                        )
+                # 在llm回复中获取情绪表情，一轮对话只在开头获取一次
+                if emotion_flag and content is not None and content.strip():
+                    asyncio.run_coroutine_threadsafe(
+                        textUtils.get_emotion(self, content),
+                        self.loop,
                     )
+                    emotion_flag = False
+
+                if content is not None and len(content) > 0:
+                    if not tool_call_flag:
+                        response_message.append(content)
+                        self.tts.tts_text_queue.put(
+                            TTSMessageDTO(
+                                sentence_id=self.sentence_id,
+                                sentence_type=SentenceType.MIDDLE,
+                                content_type=ContentType.TEXT,
+                                content_detail=content,
+                            )
+                        )
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"LLM stream processing error: {e}")
+            self.tts.tts_text_queue.put(
+                TTSMessageDTO(
+                    sentence_id=self.sentence_id,
+                    sentence_type=SentenceType.MIDDLE,
+                    content_type=ContentType.TEXT,
+                    content_detail=get_system_error_response(self.config),
+                )
+            )
+            if depth == 0:
+                self.tts.tts_text_queue.put(
+                    TTSMessageDTO(
+                        sentence_id=self.sentence_id,
+                        sentence_type=SentenceType.LAST,
+                        content_type=ContentType.ACTION,
+                    )
+                )
+            return
         # 处理function call
         if tool_call_flag:
             bHasError = False
@@ -989,7 +1014,6 @@ class ConnectionHandler:
                     content_type=ContentType.ACTION,
                 )
             )
-            self.llm_finish_task = True
             # 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
             self.logger.bind(tag=TAG).debug(
                 lambda: json.dumps(
@@ -1161,6 +1185,8 @@ class ConnectionHandler:
 
             if self.tts:
                 await self.tts.close()
+            if self.asr:
+                await self.asr.close()
 
             # 最后关闭线程池（避免阻塞）
             if self.executor:
@@ -1209,11 +1235,21 @@ class ConnectionHandler:
                 f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}"
             )
 
-    def reset_vad_states(self):
-        self.client_audio_buffer = bytearray()
+    def reset_audio_states(self):
+        """
+        重置所有音频相关状态(VAD + ASR)
+        """
+        # Reset VAD states
+        self.client_audio_buffer.clear()
         self.client_have_voice = False
         self.client_voice_stop = False
-        self.logger.bind(tag=TAG).debug("VAD states reset.")
+        self.client_voice_window.clear()
+        self.last_is_voice = False
+
+        # Clear ASR buffers
+        self.asr_audio.clear()
+
+        self.logger.bind(tag=TAG).debug("All audio states reset.")
 
     def chat_and_close(self, text):
         """Chat with the user and then close the connection"""

main/xiaozhi-server/core/handle/helloHandle.py

@@ -143,7 +143,8 @@ async def wakeupWordsResponse(conn: "ConnectionHandler"):
         # 获取当前音色
         voice = getattr(conn.tts, "voice", "default")
 
-        wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000)
+        # 使用链接的sample_rate
+        wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=conn.sample_rate)
         file_path = wakeup_words_config.generate_file_path(voice)
         with open(file_path, "wb") as f:
             f.write(wav_bytes)

main/xiaozhi-server/core/handle/receiveAudioHandle.py

@@ -21,7 +21,6 @@ async def handleAudioMessage(conn: "ConnectionHandler", audio):
     if hasattr(conn, "just_woken_up") and conn.just_woken_up:
         have_voice = False
         # 设置一个短暂延迟后恢复VAD检测
-        conn.asr_audio.clear()
         if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
             conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
         return

main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py

@@ -29,10 +29,9 @@ class ListenTextMessageHandler(TextMessageHandler):
                 f"客户端拾音模式：{conn.client_listen_mode}"
             )
         if msg_json["state"] == "start":
-            conn.client_have_voice = True
-            conn.client_voice_stop = False
+            # 设备从播放模式切回录音模式,清除所有音频状态和缓冲区
+            conn.reset_audio_states()
         elif msg_json["state"] == "stop":
-            conn.client_have_voice = True
             conn.client_voice_stop = True
             if conn.asr.interface_type == InterfaceType.STREAM:
                 # 流式模式下，发送结束请求
@@ -41,14 +40,13 @@ class ListenTextMessageHandler(TextMessageHandler):
                 # 非流式模式：直接触发ASR识别
                 if len(conn.asr_audio) > 0:
                     asr_audio_task = conn.asr_audio.copy()
-                    conn.asr_audio.clear()
-                    conn.reset_vad_states()
+                    conn.reset_audio_states()
 
                     if len(asr_audio_task) > 0:
                         await conn.asr.handle_voice_stop(conn, asr_audio_task)
         elif msg_json["state"] == "detect":
             conn.client_have_voice = False
-            conn.asr_audio.clear()
+            conn.reset_audio_states()
             if "text" in msg_json:
                 conn.last_activity_time = time.time() * 1000
                 original_text = msg_json["text"]  # 保留原始文本

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

@@ -213,36 +213,24 @@ class ASRProvider(ASRProviderBase):
             return None
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if self._is_token_expired():
             logger.warning("Token已过期，正在自动刷新...")
             self._refresh_token()
 
-        file_path = None
         try:
-            # 解码Opus为PCM
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            combined_pcm_data = b"".join(pcm_data)
-
-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
-
+            if artifacts is None:
+                return "", None
             # 发送请求并获取文本
-            text = await self._send_request(combined_pcm_data)
+            text = await self._send_request(artifacts.pcm_bytes)
 
             if text:
-                return text, file_path
+                return text, artifacts.file_path
 
-            return "", file_path
+            return "", artifacts.file_path
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-            return "", file_path
+            return "", None

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

@@ -129,17 +129,9 @@ class ASRProvider(ASRProviderBase):
     async def open_audio_channels(self, conn):
         await super().open_audio_channels(conn)
 
-    async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
-        # 初始化音频缓存
-        if not hasattr(conn, 'asr_audio_for_voiceprint'):
-            conn.asr_audio_for_voiceprint = []
-        
-        # 存储音频数据
-        if audio:
-            conn.asr_audio_for_voiceprint.append(audio)
-        
-        conn.asr_audio.append(audio)
-        conn.asr_audio = conn.asr_audio[-10:]
+    async def receive_audio(self, conn, audio, audio_have_voice):
+        # 先调用父类方法处理基础逻辑
+        await super().receive_audio(conn, audio, audio_have_voice)
 
         # 只在有声音且没有连接时建立连接（排除正在停止的情况）
         if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -156,7 +148,7 @@ class ASRProvider(ASRProviderBase):
                 await self.asr_ws.send(pcm_frame)
             except Exception as e:
                 logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
-                await self._cleanup(conn)
+                await self._cleanup()
 
     async def _start_recognition(self, conn: "ConnectionHandler"):
         """开始识别会话"""
@@ -208,8 +200,10 @@ class ASRProvider(ASRProviderBase):
         """转发识别结果"""
         try:
             while not conn.stop_event.is_set():
+                # 获取当前连接的音频数据
+                audio_data = conn.asr_audio
                 try:
-                    response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
+                    response = await self.asr_ws.recv()
                     result = json.loads(response)
 
                     header = result.get("header", {})
@@ -261,19 +255,12 @@ class ASRProvider(ASRProviderBase):
 
                                 # 手动模式下，只有在收到stop信号后才触发处理（仅处理一次）
                                 if conn.client_voice_stop:
-                                    audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
-                                    if len(audio_data) > 0:
-                                        logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
-                                        await self.handle_voice_stop(conn, audio_data)
-                                        # 清理音频缓存
-                                        conn.asr_audio.clear()
-                                        conn.reset_vad_states()
+                                    logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
+                                    await self.handle_voice_stop(conn, audio_data)
                                     break
                             else:
                                 # 自动模式下直接覆盖
                                 self.text = text
-                                conn.reset_vad_states()
-                                audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
                                 await self.handle_voice_stop(conn, audio_data)
                                 break
 
@@ -293,11 +280,7 @@ class ASRProvider(ASRProviderBase):
         finally:
             # 清理连接的音频缓存
             await self._cleanup()
-            if conn:
-                if hasattr(conn, 'asr_audio_for_voiceprint'):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, 'asr_audio'):
-                    conn.asr_audio = []
+            conn.reset_audio_states()
 
     async def _send_stop_request(self):
         """发送停止识别请求（不关闭连接）"""
@@ -345,7 +328,7 @@ class ASRProvider(ASRProviderBase):
 
         logger.bind(tag=TAG).debug("ASR会话清理完成")
 
-    async def speech_to_text(self, opus_data, session_id, audio_format):
+    async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
         """获取识别结果"""
         result = self.text
         self.text = ""
@@ -353,7 +336,7 @@ class ASRProvider(ASRProviderBase):
 
     async def close(self):
         """关闭资源"""
-        await self._cleanup(None)
+        await self._cleanup()
         if hasattr(self, 'decoder') and self.decoder is not None:
             try:
                 del self.decoder

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

@@ -55,17 +55,9 @@ class ASRProvider(ASRProviderBase):
     async def open_audio_channels(self, conn):
         await super().open_audio_channels(conn)
 
-    async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
-        # 初始化音频缓存
-        if not hasattr(conn, 'asr_audio_for_voiceprint'):
-            conn.asr_audio_for_voiceprint = []
-
-        # 存储音频数据
-        if audio:
-            conn.asr_audio_for_voiceprint.append(audio)
-
-        conn.asr_audio.append(audio)
-        conn.asr_audio = conn.asr_audio[-10:]
+    async def receive_audio(self, conn, audio, audio_have_voice):
+        # 先调用父类方法处理基础逻辑
+        await super().receive_audio(conn, audio, audio_have_voice)
 
         # 只在有声音且没有连接时建立连接
         if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -170,6 +162,8 @@ class ASRProvider(ASRProviderBase):
         """转发识别结果"""
         try:
             while not conn.stop_event.is_set():
+                # 获取当前连接的音频数据
+                audio_data = conn.asr_audio
                 try:
                     response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
                     result = json.loads(response)
@@ -218,19 +212,12 @@ class ASRProvider(ASRProviderBase):
 
                                 # 手动模式下,只有在收到stop信号后才触发处理
                                 if conn.client_voice_stop:
-                                    audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
-                                    if len(audio_data) > 0:
-                                        logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
-                                        await self.handle_voice_stop(conn, audio_data)
-                                        # 清理音频缓存
-                                        conn.asr_audio.clear()
-                                        conn.reset_vad_states()
+                                    logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
+                                    await self.handle_voice_stop(conn, audio_data)
                                     break
                             else:
                                 # 自动模式下直接覆盖
                                 self.text = text
-                                conn.reset_vad_states()
-                                audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
                                 await self.handle_voice_stop(conn, audio_data)
                                 break
 
@@ -261,11 +248,7 @@ class ASRProvider(ASRProviderBase):
         finally:
             # 清理连接的音频缓存
             await self._cleanup()
-            if conn:
-                if hasattr(conn, 'asr_audio_for_voiceprint'):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, 'asr_audio'):
-                    conn.asr_audio = []
+            conn.reset_audio_states()
 
     async def _send_stop_request(self):
         """发送停止请求(用于手动模式停止录音)"""
@@ -329,7 +312,7 @@ class ASRProvider(ASRProviderBase):
 
         logger.bind(tag=TAG).debug("ASR会话清理完成")
 
-    async def speech_to_text(self, opus_data, session_id, audio_format):
+    async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
         """获取识别结果"""
         result = self.text
         self.text = ""

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

@@ -30,37 +30,26 @@ class ASRProvider(ASRProviderBase):
         os.makedirs(self.output_dir, exist_ok=True)
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if not opus_data:
             logger.bind(tag=TAG).warning("音频数据为空！")
             return None, None
 
-        file_path = None
         try:
             # 检查配置是否已设置
             if not self.app_id or not self.api_key or not self.secret_key:
                 logger.bind(tag=TAG).error("百度语音识别配置未设置，无法进行识别")
-                return None, file_path
+                return None, None
 
-            # 将Opus音频数据解码为PCM
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            combined_pcm_data = b"".join(pcm_data)
-
-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                self.save_audio_to_file(pcm_data, session_id)
+            if artifacts is None:
+                return "", None
 
             start_time = time.time()
             # 识别本地文件
             result = self.client.asr(
-                combined_pcm_data,
+                artifacts.pcm_bytes,
                 "pcm",
                 16000,
                 {
@@ -73,13 +62,13 @@ class ASRProvider(ASRProviderBase):
                     f"百度语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}"
                 )
                 result = result["result"][0]
-                return result, file_path
+                return result, artifacts.file_path
             else:
                 raise Exception(
                     f"百度语音识别失败，错误码: {result['err_no']}，错误信息: {result['err_msg']}"
                 )
-                return None, file_path
+                return None, artifacts.file_path
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"处理音频时发生错误！{e}", exc_info=True)
-            return None, file_path
+            return None, None

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

@@ -5,20 +5,25 @@ import uuid
 import json
 import time
 import queue
+import shutil
 import asyncio
+import tempfile
 import traceback
 import threading
 import opuslib_next
+
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from typing import Optional, Tuple, List, TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from core.connection import ConnectionHandler
+from core.providers.asr.dto.dto import InterfaceType
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.utils.util import remove_punctuation_and_length
 from core.handle.receiveAudioHandle import handleAudioMessage
+from typing import Optional, Tuple, List, NamedTuple, TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -60,18 +65,17 @@ class ASRProviderBase(ABC):
             conn.asr_audio.append(audio)
         else:
             # 自动/实时模式：使用VAD检测
-            have_voice = audio_have_voice
-
             conn.asr_audio.append(audio)
-            if not have_voice and not conn.client_have_voice:
+
+            # 如果没有语音，且之前也没有声音，缓存部分音频
+            if not audio_have_voice and not conn.client_have_voice:
                 conn.asr_audio = conn.asr_audio[-10:]
                 return
 
             # 自动模式下通过VAD检测到语音停止时触发识别
-            if conn.client_voice_stop:
+            if conn.asr.interface_type != InterfaceType.STREAM and conn.client_voice_stop:
                 asr_audio_task = conn.asr_audio.copy()
-                conn.asr_audio.clear()
-                conn.reset_vad_states()
+                conn.reset_audio_states()
 
                 if len(asr_audio_task) > 15:
                     await self.handle_voice_stop(conn, asr_audio_task)
@@ -96,10 +100,14 @@ class ASRProviderBase(ABC):
                 wav_data = self._pcm_to_wav(combined_pcm_data)
 
             # 定义ASR任务
-            asr_task = self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
+            asr_task = self.speech_to_text_wrapper(
+                asr_audio_task, conn.session_id, conn.audio_format
+            )
 
             if conn.voiceprint_provider and wav_data:
-                voiceprint_task = conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
+                voiceprint_task = conn.voiceprint_provider.identify_speaker(
+                    wav_data, conn.session_id
+                )
                 # 并发等待两个结果
                 asr_result, voiceprint_result = await asyncio.gather(
                     asr_task, voiceprint_task, return_exceptions=True
@@ -162,20 +170,20 @@ class ASRProviderBase(ABC):
             if text_len > 0:
                 # 使用自定义模块进行上报
                 await startToChat(conn, enhanced_text)
-                enqueue_asr_report(conn, enhanced_text, asr_audio_task)
-                
+                audio_snapshot = asr_audio_task.copy()
+                enqueue_asr_report(conn, enhanced_text, audio_snapshot)
         except Exception as e:
             logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
             import traceback
+
             logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}")
 
     def _build_enhanced_text(self, text: str, speaker_name: Optional[str]) -> str:
         """构建包含说话人信息的文本（仅用于纯文本ASR）"""
         if speaker_name and speaker_name.strip():
-            return json.dumps({
-                "speaker": speaker_name,
-                "content": text
-            }, ensure_ascii=False)
+            return json.dumps(
+                {"speaker": speaker_name, "content": text}, ensure_ascii=False
+            )
         else:
             return text
 
@@ -184,23 +192,23 @@ class ASRProviderBase(ABC):
         if len(pcm_data) == 0:
             logger.bind(tag=TAG).warning("PCM数据为空，无法转换WAV")
             return b""
-        
+
         # 确保数据长度是偶数（16位音频）
         if len(pcm_data) % 2 != 0:
             pcm_data = pcm_data[:-1]
-        
+
         # 创建WAV文件头
         wav_buffer = io.BytesIO()
         try:
-            with wave.open(wav_buffer, 'wb') as wav_file:
-                wav_file.setnchannels(1)      # 单声道
-                wav_file.setsampwidth(2)      # 16位
+            with wave.open(wav_buffer, "wb") as wav_file:
+                wav_file.setnchannels(1)  # 单声道
+                wav_file.setsampwidth(2)  # 16位
                 wav_file.setframerate(16000)  # 16kHz采样率
                 wav_file.writeframes(pcm_data)
-            
+
             wav_buffer.seek(0)
             wav_data = wav_buffer.read()
-            
+
             return wav_data
         except Exception as e:
             logger.bind(tag=TAG).error(f"WAV转换失败: {e}")
@@ -209,6 +217,44 @@ class ASRProviderBase(ABC):
     def stop_ws_connection(self):
         pass
 
+    async def close(self):
+        pass
+
+    class AudioArtifacts(NamedTuple):
+        pcm_frames: List[bytes]
+        """PCM音频帧列表"""
+        pcm_bytes: bytes
+        """合并后的PCM音频字节数据"""
+        file_path: Optional[str]
+        """WAV文件路径"""
+        temp_path: Optional[str]
+        """临时WAV文件路径"""
+
+    def get_current_artifacts(self) -> Optional["ASRProviderBase.AudioArtifacts"]:
+        return self._current_artifacts
+
+    def requires_file(self) -> bool:
+        """是否需要文件输入"""
+        return False
+
+    def prefers_temp_file(self) -> bool:
+        """是否优先使用临时文件"""
+        return False
+
+    def build_temp_file(self, pcm_bytes: bytes) -> Optional[str]:
+        try:
+            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
+                temp_path = temp_file.name
+            with wave.open(temp_path, "wb") as wav_file:
+                wav_file.setnchannels(1)
+                wav_file.setsampwidth(2)
+                wav_file.setframerate(16000)
+                wav_file.writeframes(pcm_bytes)
+            return temp_path
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"临时音频文件生成失败: {e}")
+            return None
+
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""
         module_name = __name__.split(".")[-1]
@@ -223,11 +269,80 @@ class ASRProviderBase(ABC):
 
         return file_path
 
-    @abstractmethod
-    async def speech_to_text(
+    async def speech_to_text_wrapper(
         self, opus_data: List[bytes], session_id: str, audio_format="opus"
     ) -> Tuple[Optional[str], Optional[str]]:
-        """将语音数据转换为文本"""
+        file_path = None
+        temp_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)
+
+            free_space = shutil.disk_usage(self.output_dir).free
+            if free_space < len(combined_pcm_data) * 2:
+                raise OSError("磁盘空间不足")
+
+            if self.requires_file() and self.prefers_temp_file():
+                temp_path = self.build_temp_file(combined_pcm_data)
+
+            if (hasattr(self, "delete_audio_file") and not self.delete_audio_file) or (
+                self.requires_file() and not self.prefers_temp_file()
+            ):
+                file_path = self.save_audio_to_file(pcm_data, session_id)
+
+            if len(combined_pcm_data) == 0:
+                artifacts = None
+            else:
+                artifacts = ASRProviderBase.AudioArtifacts(
+                    pcm_frames=pcm_data,
+                    pcm_bytes=combined_pcm_data,
+                    file_path=file_path,
+                    temp_path=temp_path,
+                )
+
+            text, _ = await self.speech_to_text(
+                opus_data, session_id, audio_format, artifacts
+            )
+            return text, file_path
+        except OSError as e:
+            logger.bind(tag=TAG).error(f"文件操作错误: {e}")
+            return None, None
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"语音识别失败: {e}")
+            return None, None
+        finally:
+            try:
+                if temp_path and os.path.exists(temp_path):
+                    os.unlink(temp_path)
+                if (
+                    hasattr(self, "delete_audio_file")
+                    and self.delete_audio_file
+                    and file_path
+                    and os.path.exists(file_path)
+                ):
+                    os.remove(file_path)
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"文件清理失败: {e}")
+
+    @abstractmethod
+    async def speech_to_text(
+        self,
+        opus_data: List[bytes],
+        session_id: str,
+        audio_format="opus",
+        artifacts: Optional[AudioArtifacts] = None,
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """将语音数据转换为文本
+
+        :param opus_data: 输入的Opus音频数据
+        :param session_id: 会话ID
+        :param audio_format: 音频格式，默认"opus"
+        :param artifacts: 音频工件，包含PCM数据、文件路径等
+        :return: 识别结果文本和文件路径（如果有）
+        """
         pass
 
     @staticmethod
@@ -238,23 +353,23 @@ class ASRProviderBase(ABC):
             decoder = opuslib_next.Decoder(16000, 1)
             pcm_data = []
             buffer_size = 960  # 每次处理960个采样点 (60ms at 16kHz)
-            
+
             for i, opus_packet in enumerate(opus_data):
                 try:
                     if not opus_packet or len(opus_packet) == 0:
                         continue
-                    
+
                     pcm_frame = decoder.decode(opus_packet, buffer_size)
                     if pcm_frame and len(pcm_frame) > 0:
                         pcm_data.append(pcm_frame)
-                        
+
                 except opuslib_next.OpusError as e:
                     logger.bind(tag=TAG).warning(f"Opus解码错误，跳过数据包 {i}: {e}")
                 except Exception as e:
                     logger.bind(tag=TAG).error(f"音频处理错误，数据包 {i}: {e}")
-            
+
             return pcm_data
-            
+
         except Exception as e:
             logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}")
             return []

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

@@ -232,24 +232,13 @@ class ASRProvider(ASRProviderBase):
             yield data[offset:data_len], True
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
 
-        file_path = None
         try:
-            # 合并所有opus数据包
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            combined_pcm_data = b"".join(pcm_data)
-
-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
+            if artifacts is None:
+                return "", None
 
             # 直接使用PCM数据
             # 计算分段大小 (单声道, 16bit, 16kHz采样率)
@@ -258,14 +247,14 @@ class ASRProvider(ASRProviderBase):
 
             # 语音识别
             start_time = time.time()
-            text = await self._send_request(combined_pcm_data, segment_size)
+            text = await self._send_request(artifacts.pcm_bytes, segment_size)
             if text:
                 logger.bind(tag=TAG).debug(
                     f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
                 )
-                return text, file_path
-            return "", file_path
+                return text, artifacts.file_path
+            return "", artifacts.file_path
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-            return "", file_path
+            return "", None

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

@@ -66,22 +66,9 @@ class ASRProvider(ASRProviderBase):
         await super().open_audio_channels(conn)
 
     async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
-        conn.asr_audio.append(audio)
-        conn.asr_audio = conn.asr_audio[-10:]
-        # 存储音频数据
-        if not hasattr(conn, "asr_audio_for_voiceprint"):
-            conn.asr_audio_for_voiceprint = []
-        conn.asr_audio_for_voiceprint.append(audio)
-
-        # 当没有音频数据时处理完整语音片段
-        if (
-            conn.client_listen_mode != "manual"
-            and not audio
-            and len(conn.asr_audio_for_voiceprint) > 0
-        ):
-            await self.handle_voice_stop(conn, conn.asr_audio_for_voiceprint)
-            conn.asr_audio_for_voiceprint = []
-
+        # 先调用父类方法处理基础逻辑
+        await super().receive_audio(conn, audio, audio_have_voice)
+        
         # 如果本次有声音，且之前没有建立连接
         if audio_have_voice and self.asr_ws is None and not self.is_processing:
             try:
@@ -174,7 +161,7 @@ class ASRProvider(ASRProviderBase):
         try:
             while self.asr_ws and not conn.stop_event.is_set():
                 # 获取当前连接的音频数据
-                audio_data = getattr(conn, "asr_audio_for_voiceprint", [])
+                audio_data = conn.asr_audio
                 try:
                     response = await self.asr_ws.recv()
                     result = self.parse_response(response)
@@ -200,7 +187,6 @@ class ASRProvider(ASRProviderBase):
                             ):
                                 logger.bind(tag=TAG).error(f"识别文本：空")
                                 self.text = ""
-                                conn.reset_vad_states()
                                 if len(audio_data) > 15:  # 确保有足够音频数据
                                     await self.handle_voice_stop(conn, audio_data)
                                 break
@@ -211,18 +197,9 @@ class ASRProvider(ASRProviderBase):
                                 if self.enable_multilingual:
                                     continue
 
-                                if (
-                                    conn.client_listen_mode == "manual"
-                                    and conn.client_voice_stop
-                                    and len(audio_data) > 0
-                                ):
-                                    logger.bind(tag=TAG).debug(
-                                        "消息结束收到停止信号，触发处理"
-                                    )
+                                if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 15:
+                                    logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理")
                                     await self.handle_voice_stop(conn, audio_data)
-                                    # 清理音频缓存
-                                    conn.asr_audio.clear()
-                                    conn.reset_vad_states()
                                     break
 
                             for utterance in utterances:
@@ -240,24 +217,13 @@ class ASRProvider(ASRProviderBase):
                                             self.text = current_text
 
                                         # 在接收消息中途时收到停止信号
-                                        if (
-                                            conn.client_voice_stop
-                                            and len(audio_data) > 0
-                                        ):
-                                            logger.bind(tag=TAG).debug(
-                                                "消息中途收到停止信号，触发处理"
-                                            )
-                                            await self.handle_voice_stop(
-                                                conn, audio_data
-                                            )
-                                            # 清理音频缓存
-                                            conn.asr_audio.clear()
-                                            conn.reset_vad_states()
+                                        if conn.client_voice_stop and len(audio_data) > 0:
+                                            logger.bind(tag=TAG).debug("消息中途收到停止信号，触发处理")
+                                            await self.handle_voice_stop(conn, audio_data)
                                         break
                                     else:
                                         # 自动模式下直接覆盖
                                         self.text = current_text
-                                        conn.reset_vad_states()
                                         if len(audio_data) > 15:  # 确保有足够音频数据
                                             await self.handle_voice_stop(
                                                 conn, audio_data
@@ -288,11 +254,8 @@ class ASRProvider(ASRProviderBase):
                 await self.asr_ws.close()
                 self.asr_ws = None
             self.is_processing = False
-            if conn:
-                if hasattr(conn, "asr_audio_for_voiceprint"):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, "asr_audio"):
-                    conn.asr_audio = []
+            # 重置所有音频相关状态
+            conn.reset_audio_states()
 
     def stop_ws_connection(self):
         if self.asr_ws:
@@ -436,7 +399,7 @@ class ASRProvider(ASRProviderBase):
             logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}")
             raise
 
-    async def speech_to_text(self, opus_data, session_id, audio_format):
+    async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
         result = self.text
         self.text = ""  # 清空text
         return result, None
@@ -463,11 +426,3 @@ class ASRProvider(ASRProviderBase):
                 logger.bind(tag=TAG).debug("Doubao decoder resources released")
             except Exception as e:
                 logger.bind(tag=TAG).debug(f"释放Doubao decoder资源时出错: {e}")
-
-        # 清理所有连接的音频缓冲区
-        if hasattr(self, "_connections"):
-            for conn in self._connections.values():
-                if hasattr(conn, "asr_audio_for_voiceprint"):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, "asr_audio"):
-                    conn.asr_audio = []

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

@@ -64,39 +64,21 @@ class ASRProvider(ASRProviderBase):
             )
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """语音转文本主处理逻辑"""
-        file_path = None
         retry_count = 0
-
+        
         while retry_count < MAX_RETRIES:
             try:
-                # 合并所有opus数据包
-                if audio_format == "pcm":
-                    pcm_data = opus_data
-                else:
-                    pcm_data = self.decode_opus(opus_data)
-
-                combined_pcm_data = b"".join(pcm_data)
-
-                # 检查磁盘空间
-                if not self.delete_audio_file:
-                    free_space = shutil.disk_usage(self.output_dir).free
-                    if free_space < len(combined_pcm_data) * 2:  # 预留2倍空间
-                        raise OSError("磁盘空间不足")
-
-                # 判断是否保存为WAV文件
-                if self.delete_audio_file:
-                    pass
-                else:
-                    file_path = self.save_audio_to_file(pcm_data, session_id)
+                if artifacts is None:
+                    return "", None
 
                 # 语音识别 - 使用线程池避免阻塞事件循环
                 start_time = time.time()
                 result = await asyncio.to_thread(
                     self.model.generate,
-                    input=combined_pcm_data,
+                    input=artifacts.pcm_bytes,
                     cache={},
                     language="auto",
                     use_itn=True,
@@ -107,7 +89,7 @@ class ASRProvider(ASRProviderBase):
                     f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text['content']}"
                 )
 
-                return text, file_path
+                return text, artifacts.file_path
 
             except OSError as e:
                 retry_count += 1
@@ -115,7 +97,7 @@ class ASRProvider(ASRProviderBase):
                     logger.bind(tag=TAG).error(
                         f"语音识别失败（已重试{retry_count}次）: {e}", exc_info=True
                     )
-                    return "", file_path
+                    return "", None
                 logger.bind(tag=TAG).warning(
                     f"语音识别失败，正在重试（{retry_count}/{MAX_RETRIES}）: {e}"
                 )
@@ -123,15 +105,4 @@ class ASRProvider(ASRProviderBase):
 
             except Exception as e:
                 logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-                return "", file_path
-
-            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}"
-                        )
+                return "", None

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

@@ -101,7 +101,7 @@ class ASRProvider(ASRProviderBase):
         logger.bind(tag=TAG).debug(f"Sent end message: {end_message}")
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """
         Convert speech data to text using FunASR.
@@ -109,18 +109,9 @@ class ASRProvider(ASRProviderBase):
         :param session_id: Unique session identifier.
         :return: Tuple containing recognized text and optional timestamp.
         """
-        file_path = None
-        if audio_format == "pcm":
-            pcm_data = opus_data
-        else:
-            pcm_data = self.decode_opus(opus_data)
-        combined_pcm_data = b"".join(pcm_data)
-
-        # 判断是否保存为WAV文件
-        if self.delete_audio_file:
-            pass
-        else:
-            file_path = self.save_audio_to_file(pcm_data, session_id)
+        
+        if artifacts is None:
+            return "", None
         auth_header = {"Authorization": "Bearer; {}".format(self.api_key)}
         async with websockets.connect(
             self.uri,
@@ -132,7 +123,7 @@ class ASRProvider(ASRProviderBase):
             try:
                 # Use asyncio to handle WebSocket communication
                 send_task = asyncio.create_task(
-                    self._send_data(ws, combined_pcm_data, session_id)
+                    self._send_data(ws, artifacts.pcm_bytes, session_id)
                 )
                 receive_task = asyncio.create_task(self._receive_responses(ws))
 
@@ -161,14 +152,14 @@ class ASRProvider(ASRProviderBase):
                 result = lang_tag_filter(result)
                 return (
                     result,
-                    file_path,
+                    artifacts.file_path,
                 )  # Return the recognized text and timestamp (if any)
 
             except websockets.exceptions.ConnectionClosed as e:
                 logger.bind(tag=TAG).error(f"WebSocket connection closed: {e}")
-                return "", file_path
+                return "", artifacts.file_path
             except Exception as e:
                 logger.bind(tag=TAG).error(
                     f"Error during speech-to-text conversion: {e}", exc_info=True
                 )
-                return "", file_path
+                return "", artifacts.file_path

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

@@ -21,20 +21,16 @@ class ASRProvider(ASRProviderBase):
 
         os.makedirs(self.output_dir, exist_ok=True)
 
-    async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus") -> Tuple[Optional[str], Optional[str]]:
+    def requires_file(self) -> bool:
+        return True
+
+    async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None) -> Tuple[Optional[str], Optional[str]]:
         file_path = None
         try:
-            start_time = time.time()
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            file_path = self.save_audio_to_file(pcm_data, session_id)
-
-            logger.bind(tag=TAG).debug(
-                f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}"
-            )
-
+            if artifacts is None:
+                return "", None
+            file_path = artifacts.file_path
+                
             logger.bind(tag=TAG).info(f"file path: {file_path}")
             headers = {
                 "Authorization": f"Bearer {self.api_key}",
@@ -71,12 +67,4 @@ class ASRProvider(ASRProviderBase):
         except Exception as e:
             logger.bind(tag=TAG).error(f"语音识别失败: {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/asr/qwen3_asr_flash.py

@@ -1,5 +1,4 @@
 import os
-import tempfile
 from typing import Optional, Tuple, List
 import dashscope
 from config.logger import setup_logging
@@ -35,56 +34,25 @@ class ASRProvider(ASRProviderBase):
         # 确保输出目录存在
         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
+    def prefers_temp_file(self) -> bool:
+        return True
+
+    def requires_file(self) -> bool:
+        return True
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> 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("音频数据为空")
+            if artifacts is None:
                 return "", None
-            
-            # 准备音频文件
-            temp_file_path = self._prepare_audio_file(combined_pcm_data)
+            temp_file_path = artifacts.temp_path
+            file_path = artifacts.file_path
             if not temp_file_path:
-                return "", None
-            
-            # 保存音频文件（如果需要）
-            if not self.delete_audio_file:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
-            
+                return "", file_path
             # 构造请求消息
             messages = [
                 {
@@ -141,11 +109,3 @@ class ASRProvider(ASRProviderBase):
         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/sherpa_onnx_local.py

@@ -120,24 +120,19 @@ class ASRProvider(ASRProviderBase):
             samples_float32 = samples_float32 / 32768
             return samples_float32, f.getframerate()
 
+    def requires_file(self) -> bool:
+        return True
+
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """语音转文本主处理逻辑"""
         file_path = None
         try:
-            # 保存音频文件
-            start_time = time.time()
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            file_path = self.save_audio_to_file(pcm_data, session_id)
-            logger.bind(tag=TAG).debug(
-                f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}"
-            )
+            if artifacts is None:
+                return "", None
+            file_path = artifacts.file_path
 
-            # 语音识别
             start_time = time.time()
             s = self.model.create_stream()
             samples, sample_rate = self.read_wave(file_path)
@@ -153,11 +148,3 @@ class ASRProvider(ASRProviderBase):
         except Exception as e:
             logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
             return "", file_path
-        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/asr/tencent.py

@@ -32,35 +32,24 @@ class ASRProvider(ASRProviderBase):
         os.makedirs(self.output_dir, exist_ok=True)
 
     async def speech_to_text(
-        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if not opus_data:
             logger.bind(tag=TAG).warning("音频数据为空！")
             return None, None
 
-        file_path = None
         try:
             # 检查配置是否已设置
             if not self.secret_id or not self.secret_key:
                 logger.bind(tag=TAG).error("腾讯云语音识别配置未设置，无法进行识别")
-                return None, file_path
+                return None, None
 
-            # 将Opus音频数据解码为PCM
-            if audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
-            combined_pcm_data = b"".join(pcm_data)
-
-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                self.save_audio_to_file(pcm_data, session_id)
+            if artifacts is None:
+                return "", None
 
             # 将音频数据转换为Base64编码
-            base64_audio = base64.b64encode(combined_pcm_data).decode("utf-8")
+            base64_audio = base64.b64encode(artifacts.pcm_bytes).decode("utf-8")
 
             # 构建请求体
             request_body = self._build_request_body(base64_audio)
@@ -77,11 +66,11 @@ class ASRProvider(ASRProviderBase):
                     f"腾讯云语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}"
                 )
 
-            return result, file_path
+            return result, artifacts.file_path
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"处理音频时发生错误！{e}", exc_info=True)
-            return None, file_path
+            return None, None
 
     def _build_request_body(self, base64_audio: str) -> str:
         """构建请求体"""

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

@@ -44,36 +44,21 @@ class ASRProvider(ASRProviderBase):
             raise
 
     async def speech_to_text(
-        self, audio_data: List[bytes], session_id: str, audio_format: str = "opus"
+        self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
     ) -> 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数据为空，无法进行识别")
+            
+            if artifacts is None:
                 return "", None
-                
-            # 合并PCM数据
-            combined_pcm_data = b"".join(pcm_data)
-            if len(combined_pcm_data) == 0:
+            if not artifacts.pcm_bytes:
                 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()
             
             
@@ -81,8 +66,8 @@ class ASRProvider(ASRProviderBase):
             chunk_size = 2000
             text_result = ""
             
-            for i in range(0, len(combined_pcm_data), chunk_size):
-                chunk = combined_pcm_data[i:i+chunk_size]
+            for i in range(0, len(artifacts.pcm_bytes), chunk_size):
+                chunk = artifacts.pcm_bytes[i:i+chunk_size]
                 if self.recognizer.AcceptWaveform(chunk):
                     result = json.loads(self.recognizer.Result())
                     text = result.get('text', '')
@@ -99,16 +84,8 @@ class ASRProvider(ASRProviderBase):
                 f"VOSK语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text_result.strip()}"
             )
             
-            return text_result.strip(), file_path
+            return text_result.strip(), artifacts.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/asr/xunfei_stream.py

@@ -104,11 +104,6 @@ class ASRProvider(ASRProviderBase):
         # 先调用父类方法处理基础逻辑
         await super().receive_audio(conn, audio, audio_have_voice)
 
-        # 存储音频数据用于声纹识别
-        if not hasattr(conn, "asr_audio_for_voiceprint"):
-            conn.asr_audio_for_voiceprint = []
-        conn.asr_audio_for_voiceprint.append(audio)
-
         # 如果本次有声音，且之前没有建立连接
         if audio_have_voice and self.asr_ws is None and not self.is_processing:
             try:
@@ -234,14 +229,8 @@ class ASRProvider(ASRProviderBase):
                                     self.text += w
 
                     if status == 2:
-                        if conn.client_listen_mode == "manual":
-                            audio_data = getattr(conn, "asr_audio_for_voiceprint", [])
-                            if len(audio_data) > 0:
-                                logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
-                                await self.handle_voice_stop(conn, audio_data)
-                                # 清理音频缓存
-                                conn.asr_audio.clear()
-                        conn.reset_vad_states()
+                        logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
+                        await self.handle_voice_stop(conn, conn.asr_audio)
                         break
 
                 except asyncio.TimeoutError:
@@ -265,13 +254,7 @@ class ASRProvider(ASRProviderBase):
         finally:
             # 清理连接资源
             await self._cleanup()
-
-            # 清理连接的音频缓存
-            if conn:
-                if hasattr(conn, "asr_audio_for_voiceprint"):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, "asr_audio"):
-                    conn.asr_audio = []
+            conn.reset_audio_states()
 
     async def handle_voice_stop(
         self, conn: "ConnectionHandler", asr_audio_task: List[bytes]
@@ -339,7 +322,7 @@ class ASRProvider(ASRProviderBase):
 
         logger.bind(tag=TAG).debug("ASR会话清理完成")
 
-    async def speech_to_text(self, opus_data, session_id, audio_format):
+    async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
         """获取识别结果"""
         result = self.text
         self.text = ""
@@ -368,10 +351,3 @@ class ASRProvider(ASRProviderBase):
             except Exception as e:
                 logger.bind(tag=TAG).debug(f"释放Xunfei decoder资源时出错: {e}")
 
-        # 清理所有连接的音频缓冲区
-        if hasattr(self, "_connections"):
-            for conn in self._connections.values():
-                if hasattr(conn, "asr_audio_for_voiceprint"):
-                    conn.asr_audio_for_voiceprint = []
-                if hasattr(conn, "asr_audio"):
-                    conn.asr_audio = []

main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py

@@ -5,11 +5,14 @@ if TYPE_CHECKING:
 from ..base import IntentProviderBase
 from plugins_func.functions.play_music import initialize_music_handler
 from config.logger import setup_logging
+from core.utils.util import get_system_error_response
 import re
 import json
 import hashlib
 import time
 
+
+
 TAG = __name__
 logger = setup_logging()
 
@@ -118,12 +121,16 @@ class IntentProvider(IntentProviderBase):
         return prompt
 
     def replyResult(self, text: str, original_text: str):
-        llm_result = self.llm.response_no_stream(
-            system_prompt=text,
-            user_prompt="请根据以上内容，像人类一样说话的口吻回复用户，要求简洁，请直接返回结果。用户现在说："
-            + original_text,
-        )
-        return llm_result
+        try:
+            llm_result = self.llm.response_no_stream(
+                system_prompt=text,
+                user_prompt="请根据以上内容，像人类一样说话的口吻回复用户，要求简洁，请直接返回结果。用户现在说："
+                + original_text,
+            )
+            return llm_result
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Error in generating reply result: {e}")
+            return get_system_error_response(self.config)
 
     async def detect_intent(
         self, conn: "ConnectionHandler", dialogue_history: List[Dict], text: str
@@ -199,9 +206,13 @@ class IntentProvider(IntentProviderBase):
         llm_start_time = time.time()
         logger.bind(tag=TAG).debug(f"开始LLM意图识别调用, 模型: {model_info}")
 
-        intent = self.llm.response_no_stream(
-            system_prompt=prompt_music, user_prompt=user_prompt
-        )
+        try:
+            intent = self.llm.response_no_stream(
+                system_prompt=prompt_music, user_prompt=user_prompt
+            )
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Error in intent detection LLM call: {e}")
+            return '{"function_call": {"name": "continue_chat"}}'
 
         # 记录LLM调用完成时间
         llm_time = time.time() - llm_start_time

main/xiaozhi-server/core/providers/llm/AliBL/AliBL.py

@@ -21,83 +21,78 @@ class LLMProvider(LLMProviderBase):
         check_model_key("AliBLLLM", self.api_key)
 
     def response(self, session_id, dialogue):
-        try:
-            # 处理dialogue
-            if self.is_No_prompt:
-                dialogue.pop(0)
-                logger.bind(tag=TAG).debug(
-                    f"【阿里百练API服务】处理后的dialogue: {dialogue}"
-                )
-
-            # 构造调用参数
-            call_params = {
-                "api_key": self.api_key,
-                "app_id": self.app_id,
-                "session_id": session_id,
-                "messages": dialogue,
-                # 开启SDK原生流式
-                "stream": True,
-            }
-            if self.memory_id != False:
-                # 百练memory需要prompt参数
-                prompt = dialogue[-1].get("content")
-                call_params["memory_id"] = self.memory_id
-                call_params["prompt"] = prompt
-                logger.bind(tag=TAG).debug(
-                    f"【阿里百练API服务】处理后的prompt: {prompt}"
-                )
-
-            # 可选地设置自定义API基地址（若配置为兼容模式URL则忽略）
-            if self.base_url and ("/api/" in self.base_url):
-                dashscope.base_http_api_url = self.base_url
-
-            responses = Application.call(**call_params)
-
-            # 流式处理（SDK在stream=True时返回可迭代对象；否则返回单次响应对象）
+        # 处理dialogue
+        if self.is_No_prompt:
+            dialogue.pop(0)
             logger.bind(tag=TAG).debug(
-                f"【阿里百练API服务】构造参数: {dict(call_params, api_key='***')}"
+                f"【阿里百练API服务】处理后的dialogue: {dialogue}"
             )
 
-            last_text = ""
-            try:
-                for resp in responses:
-                    if resp.status_code != HTTPStatus.OK:
-                        logger.bind(tag=TAG).error(
-                            f"code={resp.status_code}, message={resp.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
-                        )
-                        continue
-                    current_text = getattr(getattr(resp, "output", None), "text", None)
-                    if current_text is None:
-                        continue
-                    # SDK流式为增量覆盖，计算差量输出
-                    if len(current_text) >= len(last_text):
-                        delta = current_text[len(last_text):]
-                    else:
-                        # 避免偶发回退
-                        delta = current_text
-                    if delta:
-                        yield delta
-                    last_text = current_text
-            except TypeError:
-                # 非流式回落（一次性返回）
-                if responses.status_code != HTTPStatus.OK:
-                    logger.bind(tag=TAG).error(
-                        f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
-                    )
-                    yield "【阿里百练API服务响应异常】"
-                else:
-                    full_text = getattr(getattr(responses, "output", None), "text", "")
-                    logger.bind(tag=TAG).info(
-                        f"【阿里百练API服务】完整响应长度: {len(full_text)}"
-                    )
-                    for i in range(0, len(full_text), self.streaming_chunk_size):
-                        chunk = full_text[i:i + self.streaming_chunk_size]
-                        if chunk:
-                            yield chunk
+        # 构造调用参数
+        call_params = {
+            "api_key": self.api_key,
+            "app_id": self.app_id,
+            "session_id": session_id,
+            "messages": dialogue,
+            # 开启SDK原生流式
+            "stream": True,
+        }
+        if self.memory_id != False:
+            # 百练memory需要prompt参数
+            prompt = dialogue[-1].get("content")
+            call_params["memory_id"] = self.memory_id
+            call_params["prompt"] = prompt
+            logger.bind(tag=TAG).debug(
+                f"【阿里百练API服务】处理后的prompt: {prompt}"
+            )
 
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"【阿里百练API服务】响应异常: {e}")
-            yield "【LLM服务响应异常】"
+        # 可选地设置自定义API基地址（若配置为兼容模式URL则忽略）
+        if self.base_url and ("/api/" in self.base_url):
+            dashscope.base_http_api_url = self.base_url
+
+        responses = Application.call(**call_params)
+
+        # 流式处理（SDK在stream=True时返回可迭代对象；否则返回单次响应对象）
+        logger.bind(tag=TAG).debug(
+            f"【阿里百练API服务】构造参数: {dict(call_params, api_key='***')}"
+        )
+
+        last_text = ""
+        try:
+            for resp in responses:
+                if resp.status_code != HTTPStatus.OK:
+                    logger.bind(tag=TAG).error(
+                        f"code={resp.status_code}, message={resp.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
+                    )
+                    continue
+                current_text = getattr(getattr(resp, "output", None), "text", None)
+                if current_text is None:
+                    continue
+                # SDK流式为增量覆盖，计算差量输出
+                if len(current_text) >= len(last_text):
+                    delta = current_text[len(last_text):]
+                else:
+                    # 避免偶发回退
+                    delta = current_text
+                if delta:
+                    yield delta
+                last_text = current_text
+        except TypeError:
+            # 非流式回落（一次性返回）
+            if responses.status_code != HTTPStatus.OK:
+                logger.bind(tag=TAG).error(
+                    f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
+                )
+                yield "【阿里百练API服务响应异常】"
+            else:
+                full_text = getattr(getattr(responses, "output", None), "text", "")
+                logger.bind(tag=TAG).info(
+                    f"【阿里百练API服务】完整响应长度: {len(full_text)}"
+                )
+                for i in range(0, len(full_text), self.streaming_chunk_size):
+                    chunk = full_text[i:i + self.streaming_chunk_size]
+                    if chunk:
+                        yield chunk
 
     def response_with_functions(self, session_id, dialogue, functions=None):
         # 阿里百练当前未支持原生的 function call。为保持兼容，这里回退到普通文本流式输出。

main/xiaozhi-server/core/providers/llm/base.py

@@ -11,20 +11,15 @@ class LLMProviderBase(ABC):
         pass
 
     def response_no_stream(self, system_prompt, user_prompt, **kwargs):
-        try:
-            # 构造对话格式
-            dialogue = [
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": user_prompt}
-            ]
-            result = ""
-            for part in self.response("", dialogue, **kwargs):
-                result += part
-            return result
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}")
-            return "【LLM服务响应异常】"
+        # 构造对话格式
+        dialogue = [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_prompt}
+        ]
+        result = ""
+        for part in self.response("", dialogue, **kwargs):
+            result += part
+        return result
     
     def response_with_functions(self, session_id, dialogue, functions=None):
         """

main/xiaozhi-server/core/providers/llm/dify/dify.py

@@ -20,76 +20,71 @@ class LLMProvider(LLMProviderBase):
             logger.bind(tag=TAG).error(model_key_msg)
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            # 取最后一条用户消息
-            last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
-            conversation_id = self.session_conversation_map.get(session_id)
+        # 取最后一条用户消息
+        last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
+        conversation_id = self.session_conversation_map.get(session_id)
 
-            # 发起流式请求
+        # 发起流式请求
+        if self.mode == "chat-messages":
+            request_json = {
+                "query": last_msg["content"],
+                "response_mode": "streaming",
+                "user": session_id,
+                "inputs": {},
+                "conversation_id": conversation_id,
+            }
+        elif self.mode == "workflows/run":
+            request_json = {
+                "inputs": {"query": last_msg["content"]},
+                "response_mode": "streaming",
+                "user": session_id,
+            }
+        elif self.mode == "completion-messages":
+            request_json = {
+                "inputs": {"query": last_msg["content"]},
+                "response_mode": "streaming",
+                "user": session_id,
+            }
+
+        with requests.post(
+            f"{self.base_url}/{self.mode}",
+            headers={"Authorization": f"Bearer {self.api_key}"},
+            json=request_json,
+            stream=True,
+        ) as r:
             if self.mode == "chat-messages":
-                request_json = {
-                    "query": last_msg["content"],
-                    "response_mode": "streaming",
-                    "user": session_id,
-                    "inputs": {},
-                    "conversation_id": conversation_id,
-                }
+                for line in r.iter_lines():
+                    if line.startswith(b"data: "):
+                        event = json.loads(line[6:])
+                        # 如果没有找到conversation_id，则获取此次conversation_id
+                        if not conversation_id:
+                            conversation_id = event.get("conversation_id")
+                            self.session_conversation_map[session_id] = (
+                                conversation_id  # 更新映射
+                            )
+                        # 过滤 message_replace 事件，此事件会全量推一次
+                        if event.get("event") != "message_replace" and event.get(
+                            "answer"
+                        ):
+                            yield event["answer"]
             elif self.mode == "workflows/run":
-                request_json = {
-                    "inputs": {"query": last_msg["content"]},
-                    "response_mode": "streaming",
-                    "user": session_id,
-                }
+                for line in r.iter_lines():
+                    if line.startswith(b"data: "):
+                        event = json.loads(line[6:])
+                        if event.get("event") == "workflow_finished":
+                            if event["data"]["status"] == "succeeded":
+                                yield event["data"]["outputs"]["answer"]
+                            else:
+                                yield "【服务响应异常】"
             elif self.mode == "completion-messages":
-                request_json = {
-                    "inputs": {"query": last_msg["content"]},
-                    "response_mode": "streaming",
-                    "user": session_id,
-                }
-
-            with requests.post(
-                f"{self.base_url}/{self.mode}",
-                headers={"Authorization": f"Bearer {self.api_key}"},
-                json=request_json,
-                stream=True,
-            ) as r:
-                if self.mode == "chat-messages":
-                    for line in r.iter_lines():
-                        if line.startswith(b"data: "):
-                            event = json.loads(line[6:])
-                            # 如果没有找到conversation_id，则获取此次conversation_id
-                            if not conversation_id:
-                                conversation_id = event.get("conversation_id")
-                                self.session_conversation_map[session_id] = (
-                                    conversation_id  # 更新映射
-                                )
-                            # 过滤 message_replace 事件，此事件会全量推一次
-                            if event.get("event") != "message_replace" and event.get(
-                                "answer"
-                            ):
-                                yield event["answer"]
-                elif self.mode == "workflows/run":
-                    for line in r.iter_lines():
-                        if line.startswith(b"data: "):
-                            event = json.loads(line[6:])
-                            if event.get("event") == "workflow_finished":
-                                if event["data"]["status"] == "succeeded":
-                                    yield event["data"]["outputs"]["answer"]
-                                else:
-                                    yield "【服务响应异常】"
-                elif self.mode == "completion-messages":
-                    for line in r.iter_lines():
-                        if line.startswith(b"data: "):
-                            event = json.loads(line[6:])
-                            # 过滤 message_replace 事件，此事件会全量推一次
-                            if event.get("event") != "message_replace" and event.get(
-                                "answer"
-                            ):
-                                yield event["answer"]
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in response generation: {e}")
-            yield "【服务响应异常】"
+                for line in r.iter_lines():
+                    if line.startswith(b"data: "):
+                        event = json.loads(line[6:])
+                        # 过滤 message_replace 事件，此事件会全量推一次
+                        if event.get("event") != "message_replace" and event.get(
+                            "answer"
+                        ):
+                            yield event["answer"]
 
     def response_with_functions(self, session_id, dialogue, functions=None):
         if len(dialogue) == 2 and functions is not None and len(functions) > 0:

main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py

@@ -19,53 +19,48 @@ class LLMProvider(LLMProviderBase):
             logger.bind(tag=TAG).error(model_key_msg)
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            # 取最后一条用户消息
-            last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
+        # 取最后一条用户消息
+        last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
 
-            # 发起流式请求
-            with requests.post(
-                f"{self.base_url}/chat/completions",
-                headers={"Authorization": f"Bearer {self.api_key}"},
-                json={
-                    "stream": True,
-                    "chatId": session_id,
-                    "detail": self.detail,
-                    "variables": self.variables,
-                    "messages": [{"role": "user", "content": last_msg["content"]}],
-                },
-                stream=True,
-            ) as r:
-                for line in r.iter_lines():
-                    if line:
-                        try:
-                            if line.startswith(b"data: "):
-                                if line[6:].decode("utf-8") == "[DONE]":
-                                    break
+        # 发起流式请求
+        with requests.post(
+            f"{self.base_url}/chat/completions",
+            headers={"Authorization": f"Bearer {self.api_key}"},
+            json={
+                "stream": True,
+                "chatId": session_id,
+                "detail": self.detail,
+                "variables": self.variables,
+                "messages": [{"role": "user", "content": last_msg["content"]}],
+            },
+            stream=True,
+        ) as r:
+            for line in r.iter_lines():
+                if line:
+                    try:
+                        if line.startswith(b"data: "):
+                            if line[6:].decode("utf-8") == "[DONE]":
+                                break
 
-                                data = json.loads(line[6:])
-                                if "choices" in data and len(data["choices"]) > 0:
-                                    delta = data["choices"][0].get("delta", {})
-                                    if (
-                                        delta
-                                        and "content" in delta
-                                        and delta["content"] is not None
-                                    ):
-                                        content = delta["content"]
-                                        if "<think>" in content:
-                                            continue
-                                        if "</think>" in content:
-                                            continue
-                                        yield content
+                            data = json.loads(line[6:])
+                            if "choices" in data and len(data["choices"]) > 0:
+                                delta = data["choices"][0].get("delta", {})
+                                if (
+                                    delta
+                                    and "content" in delta
+                                    and delta["content"] is not None
+                                ):
+                                    content = delta["content"]
+                                    if "<think>" in content:
+                                        continue
+                                    if "</think>" in content:
+                                        continue
+                                    yield content
 
-                        except json.JSONDecodeError as e:
-                            continue
-                        except Exception as e:
-                            continue
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in response generation: {e}")
-            yield "【服务响应异常】"
+                    except json.JSONDecodeError as e:
+                        continue
+                    except Exception as e:
+                        continue
 
     def response_with_functions(self, session_id, dialogue, functions=None):
         logger.bind(tag=TAG).error(

main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py

@@ -15,55 +15,49 @@ class LLMProvider(LLMProviderBase):
         self.api_url = f"{self.base_url}/api/conversation/process"  # 拼接完整的 API URL
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            # home assistant语音助手自带意图，无需使用xiaozhi ai自带的，只需要把用户说的话传递给home assistant即可
+        # home assistant语音助手自带意图，无需使用xiaozhi ai自带的，只需要把用户说的话传递给home assistant即可
 
-            # 提取最后一个 role 为 'user' 的 content
-            input_text = None
-            if isinstance(dialogue, list):  # 确保 dialogue 是一个列表
-                # 逆序遍历，找到最后一个 role 为 'user' 的消息
-                for message in reversed(dialogue):
-                    if message.get("role") == "user":  # 找到 role 为 'user' 的消息
-                        input_text = message.get("content", "")
-                        break  # 找到后立即退出循环
+        # 提取最后一个 role 为 'user' 的 content
+        input_text = None
+        if isinstance(dialogue, list):  # 确保 dialogue 是一个列表
+            # 逆序遍历，找到最后一个 role 为 'user' 的消息
+            for message in reversed(dialogue):
+                if message.get("role") == "user":  # 找到 role 为 'user' 的消息
+                    input_text = message.get("content", "")
+                    break  # 找到后立即退出循环
 
-            # 构造请求数据
-            payload = {
-                "text": input_text,
-                "agent_id": self.agent_id,
-                "conversation_id": session_id,  # 使用 session_id 作为 conversation_id
-            }
-            # 设置请求头
-            headers = {
-                "Authorization": f"Bearer {self.api_key}",
-                "Content-Type": "application/json",
-            }
+        # 构造请求数据
+        payload = {
+            "text": input_text,
+            "agent_id": self.agent_id,
+            "conversation_id": session_id,  # 使用 session_id 作为 conversation_id
+        }
+        # 设置请求头
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
 
-            # 发起 POST 请求
-            response = requests.post(self.api_url, json=payload, headers=headers)
+        # 发起 POST 请求
+        response = requests.post(self.api_url, json=payload, headers=headers)
 
-            # 检查请求是否成功
-            response.raise_for_status()
+        # 检查请求是否成功
+        response.raise_for_status()
 
-            # 解析返回数据
-            data = response.json()
-            speech = (
-                data.get("response", {})
-                .get("speech", {})
-                .get("plain", {})
-                .get("speech", "")
-            )
+        # 解析返回数据
+        data = response.json()
+        speech = (
+            data.get("response", {})
+            .get("speech", {})
+            .get("plain", {})
+            .get("speech", "")
+        )
 
-            # 返回生成的内容
-            if speech:
-                yield speech
-            else:
-                logger.bind(tag=TAG).warning("API 返回数据中没有 speech 内容")
-
-        except RequestException as e:
-            logger.bind(tag=TAG).error(f"HTTP 请求错误: {e}")
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"生成响应时出错: {e}")
+        # 返回生成的内容
+        if speech:
+            yield speech
+        else:
+            logger.bind(tag=TAG).warning("API 返回数据中没有 speech 内容")
 
     def response_with_functions(self, session_id, dialogue, functions=None):
         logger.bind(tag=TAG).error(

main/xiaozhi-server/core/providers/llm/ollama/ollama.py

@@ -25,151 +25,141 @@ class LLMProvider(LLMProviderBase):
         self.is_qwen3 = self.model_name and self.model_name.lower().startswith("qwen3")
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
-            if self.is_qwen3:
-                # 复制对话列表，避免修改原始对话
-                dialogue_copy = dialogue.copy()
+        # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
+        if self.is_qwen3:
+            # 复制对话列表，避免修改原始对话
+            dialogue_copy = dialogue.copy()
 
-                # 找到最后一条用户消息
-                for i in range(len(dialogue_copy) - 1, -1, -1):
-                    if dialogue_copy[i]["role"] == "user":
-                        # 在用户消息前添加/no_think指令
-                        dialogue_copy[i]["content"] = (
-                            "/no_think " + dialogue_copy[i]["content"]
-                        )
-                        logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
-                        break
-
-                # 使用修改后的对话
-                dialogue = dialogue_copy
-
-            responses = self.client.chat.completions.create(
-                model=self.model_name, messages=dialogue, stream=True
-            )
-            is_active = True
-            # 用于处理跨chunk的标签
-            buffer = ""
-
-            for chunk in responses:
-                try:
-                    delta = (
-                        chunk.choices[0].delta
-                        if getattr(chunk, "choices", None)
-                        else None
+            # 找到最后一条用户消息
+            for i in range(len(dialogue_copy) - 1, -1, -1):
+                if dialogue_copy[i]["role"] == "user":
+                    # 在用户消息前添加/no_think指令
+                    dialogue_copy[i]["content"] = (
+                        "/no_think " + dialogue_copy[i]["content"]
                     )
-                    content = delta.content if hasattr(delta, "content") else ""
+                    logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
+                    break
 
-                    if content:
-                        # 将内容添加到缓冲区
-                        buffer += content
+            # 使用修改后的对话
+            dialogue = dialogue_copy
 
-                        # 处理缓冲区中的标签
-                        while "<think>" in buffer and "</think>" in buffer:
-                            # 找到完整的<think></think>标签并移除
-                            pre = buffer.split("<think>", 1)[0]
-                            post = buffer.split("</think>", 1)[1]
-                            buffer = pre + post
+        responses = self.client.chat.completions.create(
+            model=self.model_name, messages=dialogue, stream=True
+        )
+        is_active = True
+        # 用于处理跨chunk的标签
+        buffer = ""
 
-                        # 处理只有开始标签的情况
-                        if "<think>" in buffer:
-                            is_active = False
-                            buffer = buffer.split("<think>", 1)[0]
+        for chunk in responses:
+            try:
+                delta = (
+                    chunk.choices[0].delta
+                    if getattr(chunk, "choices", None)
+                    else None
+                )
+                content = delta.content if hasattr(delta, "content") else ""
 
-                        # 处理只有结束标签的情况
-                        if "</think>" in buffer:
-                            is_active = True
-                            buffer = buffer.split("</think>", 1)[1]
+                if content:
+                    # 将内容添加到缓冲区
+                    buffer += content
 
-                        # 如果当前处于活动状态且缓冲区有内容，则输出
-                        if is_active and buffer:
-                            yield buffer
-                            buffer = ""  # 清空缓冲区
+                    # 处理缓冲区中的标签
+                    while "<think>" in buffer and "</think>" in buffer:
+                        # 找到完整的<think></think>标签并移除
+                        pre = buffer.split("<think>", 1)[0]
+                        post = buffer.split("</think>", 1)[1]
+                        buffer = pre + post
 
-                except Exception as e:
-                    logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
+                    # 处理只有开始标签的情况
+                    if "<think>" in buffer:
+                        is_active = False
+                        buffer = buffer.split("<think>", 1)[0]
 
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}")
-            yield "【Ollama服务响应异常】"
+                    # 处理只有结束标签的情况
+                    if "</think>" in buffer:
+                        is_active = True
+                        buffer = buffer.split("</think>", 1)[1]
+
+                    # 如果当前处于活动状态且缓冲区有内容，则输出
+                    if is_active and buffer:
+                        yield buffer
+                        buffer = ""  # 清空缓冲区
+
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
 
     def response_with_functions(self, session_id, dialogue, functions=None):
-        try:
-            # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
-            if self.is_qwen3:
-                # 复制对话列表，避免修改原始对话
-                dialogue_copy = dialogue.copy()
+        # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
+        if self.is_qwen3:
+            # 复制对话列表，避免修改原始对话
+            dialogue_copy = dialogue.copy()
 
-                # 找到最后一条用户消息
-                for i in range(len(dialogue_copy) - 1, -1, -1):
-                    if dialogue_copy[i]["role"] == "user":
-                        # 在用户消息前添加/no_think指令
-                        dialogue_copy[i]["content"] = (
-                            "/no_think " + dialogue_copy[i]["content"]
-                        )
-                        logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
-                        break
-
-                # 使用修改后的对话
-                dialogue = dialogue_copy
-
-            stream = self.client.chat.completions.create(
-                model=self.model_name,
-                messages=dialogue,
-                stream=True,
-                tools=functions,
-            )
-
-            is_active = True
-            buffer = ""
-
-            for chunk in stream:
-                try:
-                    delta = (
-                        chunk.choices[0].delta
-                        if getattr(chunk, "choices", None)
-                        else None
-                    )
-                    content = delta.content if hasattr(delta, "content") else None
-                    tool_calls = (
-                        delta.tool_calls if hasattr(delta, "tool_calls") else None
+            # 找到最后一条用户消息
+            for i in range(len(dialogue_copy) - 1, -1, -1):
+                if dialogue_copy[i]["role"] == "user":
+                    # 在用户消息前添加/no_think指令
+                    dialogue_copy[i]["content"] = (
+                        "/no_think " + dialogue_copy[i]["content"]
                     )
+                    logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
+                    break
 
-                    # 如果是工具调用，直接传递
-                    if tool_calls:
-                        yield None, tool_calls
-                        continue
+            # 使用修改后的对话
+            dialogue = dialogue_copy
 
-                    # 处理文本内容
-                    if content:
-                        # 将内容添加到缓冲区
-                        buffer += content
+        stream = self.client.chat.completions.create(
+            model=self.model_name,
+            messages=dialogue,
+            stream=True,
+            tools=functions,
+        )
 
-                        # 处理缓冲区中的标签
-                        while "<think>" in buffer and "</think>" in buffer:
-                            # 找到完整的<think></think>标签并移除
-                            pre = buffer.split("<think>", 1)[0]
-                            post = buffer.split("</think>", 1)[1]
-                            buffer = pre + post
+        is_active = True
+        buffer = ""
 
-                        # 处理只有开始标签的情况
-                        if "<think>" in buffer:
-                            is_active = False
-                            buffer = buffer.split("<think>", 1)[0]
+        for chunk in stream:
+            try:
+                delta = (
+                    chunk.choices[0].delta
+                    if getattr(chunk, "choices", None)
+                    else None
+                )
+                content = delta.content if hasattr(delta, "content") else None
+                tool_calls = (
+                    delta.tool_calls if hasattr(delta, "tool_calls") else None
+                )
 
-                        # 处理只有结束标签的情况
-                        if "</think>" in buffer:
-                            is_active = True
-                            buffer = buffer.split("</think>", 1)[1]
-
-                        # 如果当前处于活动状态且缓冲区有内容，则输出
-                        if is_active and buffer:
-                            yield buffer, None
-                            buffer = ""  # 清空缓冲区
-                except Exception as e:
-                    logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
+                # 如果是工具调用，直接传递
+                if tool_calls:
+                    yield None, tool_calls
                     continue
 
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in Ollama function call: {e}")
-            yield f"【Ollama服务响应异常: {str(e)}】", None
+                # 处理文本内容
+                if content:
+                    # 将内容添加到缓冲区
+                    buffer += content
+
+                    # 处理缓冲区中的标签
+                    while "<think>" in buffer and "</think>" in buffer:
+                        # 找到完整的<think></think>标签并移除
+                        pre = buffer.split("<think>", 1)[0]
+                        post = buffer.split("</think>", 1)[1]
+                        buffer = pre + post
+
+                    # 处理只有开始标签的情况
+                    if "<think>" in buffer:
+                        is_active = False
+                        buffer = buffer.split("<think>", 1)[0]
+
+                    # 处理只有结束标签的情况
+                    if "</think>" in buffer:
+                        is_active = True
+                        buffer = buffer.split("</think>", 1)[1]
+
+                    # 如果当前处于活动状态且缓冲区有内容，则输出
+                    if is_active and buffer:
+                        yield buffer, None
+                        buffer = ""  # 清空缓冲区
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
+                continue

main/xiaozhi-server/core/providers/llm/openai/openai.py

@@ -56,87 +56,78 @@ class LLMProvider(LLMProviderBase):
         return dialogue
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            dialogue = self.normalize_dialogue(dialogue)
+        dialogue = self.normalize_dialogue(dialogue)
 
-            request_params = {
-                "model": self.model_name,
-                "messages": dialogue,
-                "stream": True,
-            }
+        request_params = {
+            "model": self.model_name,
+            "messages": dialogue,
+            "stream": True,
+        }
 
-            # 添加可选参数,只有当参数不为None时才添加
-            optional_params = {
-                "max_tokens": kwargs.get("max_tokens", self.max_tokens),
-                "temperature": kwargs.get("temperature", self.temperature),
-                "top_p": kwargs.get("top_p", self.top_p),
-                "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
-            }
+        # 添加可选参数,只有当参数不为None时才添加
+        optional_params = {
+            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
+            "temperature": kwargs.get("temperature", self.temperature),
+            "top_p": kwargs.get("top_p", self.top_p),
+            "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
+        }
 
-            for key, value in optional_params.items():
-                if value is not None:
-                    request_params[key] = value
+        for key, value in optional_params.items():
+            if value is not None:
+                request_params[key] = value
 
-            responses = self.client.chat.completions.create(**request_params)
+        responses = self.client.chat.completions.create(**request_params)
 
-            is_active = True
-            for chunk in responses:
-                try:
-                    delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None
-                    content = getattr(delta, "content", "") if delta else ""
-                except IndexError:
-                    content = ""
-                if content:
-                    if "<think>" in content:
-                        is_active = False
-                        content = content.split("<think>")[0]
-                    if "</think>" in content:
-                        is_active = True
-                        content = content.split("</think>")[-1]
-                    if is_active:
-                        yield content
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in response generation: {e}")
+        is_active = True
+        for chunk in responses:
+            try:
+                delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None
+                content = getattr(delta, "content", "") if delta else ""
+            except IndexError:
+                content = ""
+            if content:
+                if "<think>" in content:
+                    is_active = False
+                    content = content.split("<think>")[0]
+                if "</think>" in content:
+                    is_active = True
+                    content = content.split("</think>")[-1]
+                if is_active:
+                    yield content
 
     def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
-        try:
-            dialogue = self.normalize_dialogue(dialogue)
+        dialogue = self.normalize_dialogue(dialogue)
 
-            request_params = {
-                "model": self.model_name,
-                "messages": dialogue,
-                "stream": True,
-                "tools": functions,
-            }
+        request_params = {
+            "model": self.model_name,
+            "messages": dialogue,
+            "stream": True,
+            "tools": functions,
+        }
 
-            optional_params = {
-                "max_tokens": kwargs.get("max_tokens", self.max_tokens),
-                "temperature": kwargs.get("temperature", self.temperature),
-                "top_p": kwargs.get("top_p", self.top_p),
-                "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
-            }
+        optional_params = {
+            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
+            "temperature": kwargs.get("temperature", self.temperature),
+            "top_p": kwargs.get("top_p", self.top_p),
+            "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
+        }
 
-            for key, value in optional_params.items():
-                if value is not None:
-                    request_params[key] = value
+        for key, value in optional_params.items():
+            if value is not None:
+                request_params[key] = value
 
-            stream = self.client.chat.completions.create(**request_params)
+        stream = self.client.chat.completions.create(**request_params)
 
-            for chunk in stream:
-                if getattr(chunk, "choices", None):
-                    delta = chunk.choices[0].delta
-                    content = getattr(delta, "content", "")
-                    tool_calls = getattr(delta, "tool_calls", None)
-                    yield content, tool_calls
-                elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
-                    usage_info = getattr(chunk, "usage", None)
-                    logger.bind(tag=TAG).info(
-                        f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
-                        f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
-                        f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
-                    )
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in function call streaming: {e}")
-            yield f"【OpenAI服务响应异常: {e}】", None
+        for chunk in stream:
+            if getattr(chunk, "choices", None):
+                delta = chunk.choices[0].delta
+                content = getattr(delta, "content", "")
+                tool_calls = getattr(delta, "tool_calls", None)
+                yield content, tool_calls
+            elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
+                usage_info = getattr(chunk, "usage", None)
+                logger.bind(tag=TAG).info(
+                    f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
+                    f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
+                    f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
+                )

main/xiaozhi-server/core/providers/llm/xinference/xinference.py

@@ -31,68 +31,55 @@ class LLMProvider(LLMProviderBase):
             raise
 
     def response(self, session_id, dialogue, **kwargs):
-        try:
-            logger.bind(tag=TAG).debug(
-                f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
-            )
-            responses = self.client.chat.completions.create(
-                model=self.model_name, messages=dialogue, stream=True
-            )
-            is_active = True
-            for chunk in responses:
-                try:
-                    delta = (
-                        chunk.choices[0].delta
-                        if getattr(chunk, "choices", None)
-                        else None
-                    )
-                    content = delta.content if hasattr(delta, "content") else ""
-                    if content:
-                        if "<think>" in content:
-                            is_active = False
-                            content = content.split("<think>")[0]
-                        if "</think>" in content:
-                            is_active = True
-                            content = content.split("</think>")[-1]
-                        if is_active:
-                            yield content
-                except Exception as e:
-                    logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in Xinference response generation: {e}")
-            yield "【Xinference服务响应异常】"
+        logger.bind(tag=TAG).debug(
+            f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
+        )
+        responses = self.client.chat.completions.create(
+            model=self.model_name, messages=dialogue, stream=True
+        )
+        is_active = True
+        for chunk in responses:
+            try:
+                delta = (
+                    chunk.choices[0].delta
+                    if getattr(chunk, "choices", None)
+                    else None
+                )
+                content = delta.content if hasattr(delta, "content") else ""
+                if content:
+                    if "<think>" in content:
+                        is_active = False
+                        content = content.split("<think>")[0]
+                    if "</think>" in content:
+                        is_active = True
+                        content = content.split("</think>")[-1]
+                    if is_active:
+                        yield content
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
 
     def response_with_functions(self, session_id, dialogue, functions=None):
-        try:
+        logger.bind(tag=TAG).debug(
+            f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
+        )
+        if functions:
             logger.bind(tag=TAG).debug(
-                f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
-            )
-            if functions:
-                logger.bind(tag=TAG).debug(
-                    f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}"
-                )
-
-            stream = self.client.chat.completions.create(
-                model=self.model_name,
-                messages=dialogue,
-                stream=True,
-                tools=functions,
+                f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}"
             )
 
-            for chunk in stream:
-                delta = chunk.choices[0].delta
-                content = delta.content
-                tool_calls = delta.tool_calls
+        stream = self.client.chat.completions.create(
+            model=self.model_name,
+            messages=dialogue,
+            stream=True,
+            tools=functions,
+        )
 
-                if content:
-                    yield content, tool_calls
-                elif tool_calls:
-                    yield None, tool_calls
+        for chunk in stream:
+            delta = chunk.choices[0].delta
+            content = delta.content
+            tool_calls = delta.tool_calls
 
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Error in Xinference function call: {e}")
-            yield {
-                "type": "content",
-                "content": f"【Xinference服务响应异常: {str(e)}】",
-            }
+            if content:
+                yield content, tool_calls
+            elif tool_calls:
+                yield None, tool_calls

main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py

@@ -162,19 +162,19 @@ class MemoryProvider(MemoryProviderBase):
         msgStr += f"当前时间：{time_str}"
 
         if self.save_to_file:
-            result = self.llm.response_no_stream(
-                short_term_memory_prompt,
-                msgStr,
-                max_tokens=2000,
-                temperature=0.2,
-            )
-            json_str = extract_json_data(result)
             try:
+                result = self.llm.response_no_stream(
+                    short_term_memory_prompt,
+                    msgStr,
+                    max_tokens=2000,
+                    temperature=0.2,
+                )
+                json_str = extract_json_data(result)
                 json.loads(json_str)  # 检查json格式是否正确
                 self.short_memory = json_str
                 self.save_memory_to_file()
             except Exception as e:
-                print("Error:", e)
+                logger.bind(tag=TAG).error(f"Error in saving memory: {e}")
         else:
             # 当save_to_file为False时，调用Java端的聊天记录总结接口
             summary_id = session_id if session_id else self.role_id

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

@@ -41,8 +41,6 @@ class TTSProvider(TTSProviderBase):
 
         # 音频参数配置
         self.format = config.get("format", "pcm")
-        sample_rate = config.get("sample_rate", "24000")
-        self.sample_rate = int(sample_rate) if sample_rate else 24000
 
         volume = config.get("volume", "50")
         self.volume = int(volume) if volume else 50
@@ -60,11 +58,6 @@ class TTSProvider(TTSProviderBase):
             "X-DashScope-DataInspection": "enable",
         }
 
-        # 创建Opus编码器
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=self.sample_rate, channels=1, frame_size_ms=60
-        )
-
     async def _ensure_connection(self):
         """确保WebSocket连接可用，支持60秒内连接复用"""
         try:
@@ -245,7 +238,7 @@ class TTSProvider(TTSProviderBase):
                         "text_type": "PlainText",
                         "voice": self.voice,
                         "format": self.format,
-                        "sample_rate": self.sample_rate,
+                        "sample_rate": self.conn.sample_rate,
                         "volume": self.volume,
                         "rate": self.rate,
                         "pitch": self.pitch,
@@ -429,7 +422,7 @@ class TTSProvider(TTSProviderBase):
                                 "text_type": "PlainText",
                                 "voice": self.voice,
                                 "format": self.format,
-                                "sample_rate": self.sample_rate,
+                                "sample_rate": self.conn.sample_rate,
                                 "volume": self.volume,
                                 "rate": self.rate,
                                 "pitch": self.pitch,

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

@@ -95,8 +95,6 @@ class TTSProvider(TTSProviderBase):
         self.appkey = config.get("appkey")
         self.format = config.get("format", "wav")
         self.audio_file_type = config.get("format", "wav")
-        sample_rate = config.get("sample_rate", "16000")
-        self.sample_rate = int(sample_rate) if sample_rate else 16000
 
         if config.get("private_voice"):
             self.voice = config.get("private_voice")
@@ -172,7 +170,7 @@ class TTSProvider(TTSProviderBase):
             "token": self.token,
             "text": text,
             "format": self.format,
-            "sample_rate": self.sample_rate,
+            "sample_rate": self.conn.sample_rate,
             "voice": self.voice,
             "volume": self.volume,
             "speech_rate": self.speech_rate,

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

@@ -99,10 +99,6 @@ class TTSProvider(TTSProviderBase):
         self.format = config.get("format", "pcm")
         self.audio_file_type = config.get("format", "pcm")
 
-        # 采样率配置
-        sample_rate = config.get("sample_rate", "16000")
-        self.sample_rate = int(sample_rate) if sample_rate else 16000
-
         # 音色配置 - CosyVoice大模型音色
         if config.get("private_voice"):
             self.voice = config.get("private_voice")
@@ -134,11 +130,6 @@ class TTSProvider(TTSProviderBase):
         # 专属tts设置
         self.task_id = uuid.uuid4().hex
 
-        # 创建Opus编码器
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=16000, channels=1, frame_size_ms=60
-        )
-
         # Token管理
         if self.access_key_id and self.access_key_secret:
             self._refresh_token()
@@ -344,7 +335,7 @@ class TTSProvider(TTSProviderBase):
                 "payload": {
                     "voice": self.voice,
                     "format": self.format,
-                    "sample_rate": self.sample_rate,
+                    "sample_rate": self.conn.sample_rate,
                     "volume": self.volume,
                     "speech_rate": self.speech_rate,
                     "pitch_rate": self.pitch_rate,
@@ -508,7 +499,7 @@ class TTSProvider(TTSProviderBase):
                         "payload": {
                             "voice": self.voice,
                             "format": self.format,
-                            "sample_rate": self.sample_rate,
+                            "sample_rate": self.conn.sample_rate,
                             "volume": self.volume,
                             "speech_rate": self.speech_rate,
                             "pitch_rate": self.pitch_rate,

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

@@ -1,17 +1,18 @@
 import os
 import re
-import time
 import uuid
 import queue
 import asyncio
 import threading
 import traceback
+
 from core.utils import p3
 from datetime import datetime
 from core.utils import textUtils
 from typing import Callable, Any
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
+from core.utils import opus_encoder_utils
 from core.utils.tts import MarkdownCleaner
 from core.utils.output_counter import add_device_output
 from core.handle.reportHandle import enqueue_tts_report
@@ -97,6 +98,8 @@ class TTSProviderBase(ABC):
                             file_type=self.audio_file_type,
                             is_opus=True,
                             callback=opus_handler,
+                            sample_rate=self.conn.sample_rate,
+                            opus_encoder=self.opus_encoder,
                         )
                         break
                     else:
@@ -138,7 +141,7 @@ class TTSProviderBase(ABC):
                     logger.bind(tag=TAG).error(
                         f"语音生成失败: {text}，请检查网络或服务是否正常"
                     )
-                    self.tts_audio_queue.put((SentenceType.FIRST, None, text))
+                self.tts_audio_queue.put((SentenceType.FIRST, None, text))
                 self._process_audio_file_stream(tmp_file, callback=opus_handler)
             except Exception as e:
                 logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
@@ -158,7 +161,8 @@ class TTSProviderBase(ABC):
                             audio_bytes,
                             file_type=self.audio_file_type,
                             is_opus=True,
-                            callback=lambda data: audio_datas.append(data)
+                            callback=lambda data: audio_datas.append(data),
+                            sample_rate=self.conn.sample_rate,
                         )
                         return audio_datas
                     else:
@@ -214,13 +218,13 @@ class TTSProviderBase(ABC):
         self, audio_file_path, callback: Callable[[Any], Any] = None
     ):
         """音频文件转换为PCM编码"""
-        return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback)
+        return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=None)
 
     def audio_to_opus_data_stream(
         self, audio_file_path, callback: Callable[[Any], Any] = None
     ):
         """音频文件转换为Opus编码"""
-        return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback)
+        return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=self.opus_encoder)
 
     def tts_one_sentence(
         self,
@@ -252,6 +256,13 @@ class TTSProviderBase(ABC):
 
     async def open_audio_channels(self, conn):
         self.conn = conn
+
+        # 根据conn的sample_rate创建编码器，如果子类已经创建则不覆盖（IndexTTS接口返回为24kHZ-待重采样处理）
+        if not hasattr(self, 'opus_encoder') or self.opus_encoder is None:
+            self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+                sample_rate=conn.sample_rate, channels=1, frame_size_ms=60
+            )
+
         # tts 消化线程
         self.tts_priority_thread = threading.Thread(
             target=self.tts_text_priority_thread, daemon=True

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

@@ -154,16 +154,29 @@ class TTSProvider(TTSProviderBase):
             self.voice = config.get("private_voice")
         else:
             self.voice = config.get("speaker")
-        speech_rate = config.get("speech_rate", "0")
-        loudness_rate = config.get("loudness_rate", "0")
-        pitch = config.get("pitch", "0")
-        self.speech_rate = int(speech_rate) if speech_rate else 0
-        self.loudness_rate = int(loudness_rate) if loudness_rate else 0
-        self.pitch = int(pitch) if pitch else 0
-        # 多情感音色参数
-        self.emotion = config.get("emotion", "neutral")  
-        emotion_scale = config.get("emotion_scale", "4")
-        self.emotion_scale = int(emotion_scale) if emotion_scale else 4
+
+        # 默认 audio_params 配置
+        default_audio_params = {
+            "speech_rate": 0,
+            "loudness_rate": 0
+        }
+
+        # 默认 additions 配置
+        default_additions = {
+            "aigc_metadata": {},
+            "cache_config": {},
+            "post_process": {
+                "pitch": 0
+            }
+        }
+
+        # 默认 mix_speaker 配置
+        default_mix_speaker = {}
+
+        # 合并用户配置
+        self.audio_params = {**default_audio_params, **config.get("audio_params", {})}
+        self.additions = {**default_additions, **config.get("additions", {})}
+        self.mix_speaker = {**default_mix_speaker, **config.get("mix_speaker", {})}
 
         self.ws_url = config.get("ws_url")
         self.authorization = config.get("authorization")
@@ -171,9 +184,7 @@ class TTSProvider(TTSProviderBase):
         enable_ws_reuse_value = config.get("enable_ws_reuse", True)
         self.enable_ws_reuse = False if str(enable_ws_reuse_value).lower() == 'false' else True
         self.tts_text = ""
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=16000, channels=1, frame_size_ms=60
-        )
+
         model_key_msg = check_model_key("TTS", self.access_token)
         if model_key_msg:
             logger.bind(tag=TAG).error(model_key_msg)
@@ -181,6 +192,8 @@ class TTSProvider(TTSProviderBase):
     async def open_audio_channels(self, conn):
         try:
             await super().open_audio_channels(conn)
+            # 更新 audio_params 中的采样率为实际的 conn.sample_rate
+            self.audio_params["sample_rate"] = conn.sample_rate
         except Exception as e:
             logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}")
             self.ws = None
@@ -646,20 +659,18 @@ class TTSProvider(TTSProviderBase):
         text="",
         speaker="",
         audio_format="pcm",
-        audio_sample_rate=16000,
     ):
-        audio_params = {
-            "format": audio_format,
-            "sample_rate": audio_sample_rate,
-            "speech_rate": self.speech_rate,
-            "loudness_rate": self.loudness_rate
+        # 构建 req_params
+        req_params = {
+            "text": text,
+            "speaker": speaker,
+            "audio_params": {**self.audio_params, "format": audio_format},
+            "additions": json.dumps(self.additions)
         }
-
-        # 如果是多情感音色,添加情感参数
-        if '_emo_' in self.voice:
-            if self.emotion:
-                audio_params["emotion"] = self.emotion
-                audio_params["emotion_scale"] = self.emotion_scale
+        
+        # 如果有 mix_speaker 配置，添加到 req_params
+        if self.mix_speaker:
+            req_params["mix_speaker"] = self.mix_speaker
 
         return str.encode(
             json.dumps(
@@ -667,17 +678,7 @@ class TTSProvider(TTSProviderBase):
                     "user": {"uid": uid},
                     "event": event,
                     "namespace": "BidirectionalTTS",
-                    "req_params": {
-                        "text": text,
-                        "speaker": speaker,
-                        "audio_params": audio_params,
-                        "additions": json.dumps({
-                            "post_process": {
-                                "pitch": self.pitch
-                            }
-                        })
-                    },
-
+                    "req_params": req_params
                 }
             )
         )

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

@@ -174,6 +174,20 @@ class TTSProvider(TTSProviderBase):
             logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
             self.tts_audio_queue.put((SentenceType.LAST, [], None))
 
+    def audio_to_pcm_data_stream(
+        self, audio_file_path, callback=None
+    ):
+        """音频文件转换为PCM编码，使用24kHz采样率"""
+        from core.utils.util import audio_to_data_stream
+        return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback, sample_rate=24000, opus_encoder=None)
+
+    def audio_to_opus_data_stream(
+        self, audio_file_path, callback=None
+    ):
+        """音频文件转换为Opus编码，使用24kHz采样率和自己的编码器"""
+        from core.utils.util import audio_to_data_stream
+        return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback, sample_rate=24000, opus_encoder=self.opus_encoder)
+
     async def close(self):
         """资源清理"""
         await super().close()

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

@@ -25,11 +25,6 @@ class TTSProvider(TTSProviderBase):
         self.audio_format = "pcm"
         self.before_stop_play_files = []
 
-        # 创建Opus编码器
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=16000, channels=1, frame_size_ms=60
-        )
-
         # PCM缓冲区
         self.pcm_buffer = bytearray()
 
@@ -127,7 +122,7 @@ class TTSProvider(TTSProviderBase):
             "spk_id": self.voice,
             "frame_durition": 60,
             "stream": "true",
-            "target_sr": 16000,
+            "target_sr": self.conn.sample_rate,
             "audio_format": "pcm",
             "instruct_text": "请生成一段自然流畅的语音",
         }
@@ -136,7 +131,7 @@ class TTSProvider(TTSProviderBase):
             "Content-Type": "application/json",
         }
 
-        # 一帧 PCM 所需字节数：60 ms × 16 kHz × 1 ch × 2 B = 1 920
+        # 一帧 PCM 所需字节数：60 ms × sample_rate × 1 ch × 2 B
         frame_bytes = int(
             self.opus_encoder.sample_rate
             * self.opus_encoder.channels  # 1
@@ -213,7 +208,7 @@ class TTSProvider(TTSProviderBase):
             "spk_id": self.voice,
             "frame_duration": 60,
             "stream": False,
-            "target_sr": 16000,
+            "target_sr": self.conn.sample_rate,
             "audio_format": self.audio_format,
             "instruct_text": "请生成一段自然流畅的语音",
         }

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

@@ -64,13 +64,17 @@ class TTSProvider(TTSProviderBase):
         }
         self.audio_file_type = defult_audio_setting.get("format", "pcm")
 
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=24000, channels=1, frame_size_ms=60
-        )
-
         # PCM缓冲区
         self.pcm_buffer = bytearray()
 
+    async def open_audio_channels(self, conn):
+        """初始化音频通道,并根据conn.sample_rate更新配置"""
+        # 调用父类方法
+        await super().open_audio_channels(conn)
+
+        # 更新audio_setting中的采样率为实际的conn.sample_rate
+        self.audio_setting["sample_rate"] = conn.sample_rate
+
     def tts_text_priority_thread(self):
         """流式文本处理线程"""
         while not self.conn.stop_event.is_set():
@@ -212,6 +216,18 @@ class TTSProvider(TTSProviderBase):
 
                             try:
                                 data = json.loads(json_str)
+
+                                # 检查业务层错误
+                                base_resp = data.get("base_resp", {})
+                                status_code = base_resp.get("status_code", 0)
+                                if status_code != 0:
+                                    status_msg = base_resp.get("status_msg", "未知错误")
+                                    logger.bind(tag=TAG).error(
+                                        f"TTS请求失败, 错误码:{status_code}, 错误消息:{status_msg}"
+                                    )
+                                    self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                                    return
+
                                 status = data.get("data", {}).get("status", 1)
                                 audio_hex = data.get("data", {}).get("audio")
 

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

@@ -25,10 +25,7 @@ class TTSProvider(TTSProviderBase):
             self.spk_id = int(config.get("private_voice"))
         else:
             self.spk_id = int(config.get("spk_id", "0"))
-            
-        sample_rate = config.get("sample_rate", 24000)
-        self.sample_rate = float(sample_rate) if sample_rate else 24000
-        
+
         speed = config.get("speed", 1.0)
         self.speed = float(speed) if speed else 1.0
         

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

@@ -13,7 +13,6 @@ class TTSProvider(TTSProviderBase):
             self.voice = config.get("voice")
         self.response_format = config.get("response_format", "mp3")
         self.audio_file_type = config.get("response_format", "mp3")
-        self.sample_rate = config.get("sample_rate")
         self.speed = float(config.get("speed", 1.0))
         self.gain = config.get("gain")
 

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

@@ -91,9 +91,6 @@ class TTSProvider(TTSProviderBase):
         # 音频编码配置
         self.format = config.get("format", "raw")
 
-        sample_rate = config.get("sample_rate", "24000")
-        self.sample_rate = int(sample_rate) if sample_rate else 24000
-
         # 口语化配置
         self.oral_level = config.get("oral_level", "mid")
 
@@ -113,11 +110,6 @@ class TTSProvider(TTSProviderBase):
         # 序列号管理
         self.text_seq = 0
 
-        # 创建Opus编码器
-        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
-            sample_rate=self.sample_rate, channels=1, frame_size_ms=60
-        )
-
         # 验证必需参数
         if not all([self.app_id, self.api_key, self.api_secret]):
             raise ValueError("讯飞TTS需要配置app_id、api_key和api_secret")
@@ -507,7 +499,7 @@ class TTSProvider(TTSProviderBase):
                     "rhy": 0,
                     "audio": {
                         "encoding": self.format,
-                        "sample_rate": self.sample_rate,
+                        "sample_rate": self.conn.sample_rate,
                         "channels": 1,
                         "bit_depth": 16,
                         "frame_size": 0

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

@@ -227,7 +227,7 @@ def extract_json_from_string(input_string):
 
 
 def audio_to_data_stream(
-    audio_file_path, is_opus=True, callback: Callable[[Any], Any] = None
+    audio_file_path, is_opus=True, callback: Callable[[Any], Any] = None, sample_rate=16000, opus_encoder=None
 ) -> None:
     # 获取文件后缀名
     file_type = os.path.splitext(audio_file_path)[1]
@@ -238,12 +238,12 @@ def audio_to_data_stream(
         audio_file_path, format=file_type, parameters=["-nostdin"]
     )
 
-    # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
-    audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
+    # 转换为单声道/指定采样率/16位小端编码（确保与编码器匹配）
+    audio = audio.set_channels(1).set_frame_rate(sample_rate).set_sample_width(2)
 
     # 获取原始PCM数据（16位小端）
     raw_data = audio.raw_data
-    pcm_to_data_stream(raw_data, is_opus, callback)
+    pcm_to_data_stream(raw_data, is_opus, callback, sample_rate, opus_encoder)
 
 
 async def audio_to_data(
@@ -325,7 +325,7 @@ async def audio_to_data(
 
 
 def audio_bytes_to_data_stream(
-    audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]
+    audio_bytes, file_type, is_opus, callback: Callable[[Any], Any], sample_rate=16000, opus_encoder=None
 ) -> None:
     """
     直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3
@@ -338,18 +338,30 @@ def audio_bytes_to_data_stream(
         audio = AudioSegment.from_file(
             BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"]
         )
-        audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
+        audio = audio.set_channels(1).set_frame_rate(sample_rate).set_sample_width(2)
         raw_data = audio.raw_data
-        pcm_to_data_stream(raw_data, is_opus, callback)
+        pcm_to_data_stream(raw_data, is_opus, callback, sample_rate, opus_encoder)
 
 
-def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None):
-    # 初始化Opus编码器
-    encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO)
+def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None, sample_rate=16000, opus_encoder=None):
+    """
+    将PCM数据流式编码为Opus或直接输出PCM
+
+    Args:
+        raw_data: PCM原始数据
+        is_opus: 是否编码为Opus
+        callback: 回调函数
+        sample_rate: 采样率
+        opus_encoder: OpusEncoderUtils对象(推荐提供以保持编码器状态连续)
+    """
+    using_temp_encoder = False
+    if is_opus and opus_encoder is None:
+        encoder = opuslib_next.Encoder(sample_rate, 1, opuslib_next.APPLICATION_AUDIO)
+        using_temp_encoder = True
 
     # 编码参数
     frame_duration = 60  # 60ms per frame
-    frame_size = int(16000 * frame_duration / 1000)  # 960 samples/frame
+    frame_size = int(sample_rate * frame_duration / 1000)  # samples/frame
 
     # 按帧处理所有音频数据（包括最后一帧可能补零）
     for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
@@ -361,12 +373,17 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] =
             chunk += b"\x00" * (frame_size * 2 - len(chunk))
 
         if is_opus:
-            # 转换为numpy数组处理
-            np_frame = np.frombuffer(chunk, dtype=np.int16)
-            # 编码Opus数据
-            frame_data = encoder.encode(np_frame.tobytes(), frame_size)
-            callback(frame_data)
+            if using_temp_encoder:
+                # 使用临时编码器（仅用于独立音频场景）
+                np_frame = np.frombuffer(chunk, dtype=np.int16)
+                frame_data = encoder.encode(np_frame.tobytes(), frame_size)
+                callback(frame_data)
+            else:
+                # 使用外部编码器（TTS流式场景,保持状态连续）
+                is_last = (i + frame_size * 2 >= len(raw_data))
+                opus_encoder.encode_pcm_to_opus_stream(chunk, end_of_stream=is_last, callback=callback)
         else:
+            # PCM模式,直接输出
             frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
             callback(frame_data)
 
@@ -573,3 +590,14 @@ def validate_mcp_endpoint(mcp_endpoint: str) -> bool:
         return False
 
     return True
+
+def get_system_error_response(config: dict) -> str:
+    """获取系统错误时的回复
+
+    Args:
+        config: 配置字典
+
+    Returns:
+        str: 系统错误时的回复
+    """
+    return config.get("system_error_response", "主人，小智现在有点忙，我们稍后再试吧。")