Merge pull request #2929 from xinnan-tech/py_fix_asr

fix: 模式切换时音频残余，优化相关状态管理

main/xiaozhi-server/core/connection.py

@@ -1166,6 +1166,8 @@ class ConnectionHandler:
 
             if self.tts:
                 await self.tts.close()
+            if self.asr:
+                await self.asr.close()
 
             # 最后关闭线程池（避免阻塞）
             if self.executor:
@@ -1214,11 +1216,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/receiveAudioHandle.py

@@ -17,7 +17,6 @@ async def handleAudioMessage(conn, 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

@@ -26,10 +26,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:
                 # 流式模式下，发送结束请求
@@ -38,14 +37,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_stream.py

@@ -126,16 +126,8 @@ class ASRProvider(ASRProviderBase):
         await super().open_audio_channels(conn)
 
     async def receive_audio(self, conn, 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:]
+        # 先调用父类方法处理基础逻辑
+        await super().receive_audio(conn, audio, audio_have_voice)
 
         # 只在有声音且没有连接时建立连接（排除正在停止的情况）
         if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -204,6 +196,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)
@@ -257,19 +251,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
 
@@ -289,11 +276,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):
         """发送停止识别请求（不关闭连接）"""

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

@@ -52,16 +52,8 @@ class ASRProvider(ASRProviderBase):
         await super().open_audio_channels(conn)
 
     async def receive_audio(self, conn, 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:]
+        # 先调用父类方法处理基础逻辑
+        await super().receive_audio(conn, audio, audio_have_voice)
 
         # 只在有声音且没有连接时建立连接
         if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -166,6 +158,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)
@@ -214,19 +208,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
 
@@ -257,11 +244,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):
         """发送停止请求(用于手动模式停止录音)"""

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

@@ -12,6 +12,7 @@ import opuslib_next
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
+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
@@ -57,18 +58,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)
@@ -159,7 +159,8 @@ 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}")
@@ -206,6 +207,9 @@ class ASRProviderBase(ABC):
     def stop_ws_connection(self):
         pass
 
+    async def close(self):
+        pass
+
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""
         module_name = __name__.split(".")[-1]

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

@@ -60,17 +60,8 @@ class ASRProvider(ASRProviderBase):
         await super().open_audio_channels(conn)
 
     async def receive_audio(self, conn, 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:
@@ -164,7 +155,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)
@@ -189,7 +180,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
@@ -200,12 +190,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:
+                                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:
@@ -226,14 +213,10 @@ class ASRProvider(ASRProviderBase):
                                         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()
                                         break
                                     else:
                                         # 自动模式下直接覆盖
                                         self.text = current_text
-                                        conn.reset_vad_states()
                                         if len(audio_data) > 15:  # 确保有足够音频数据
                                             await self.handle_voice_stop(conn, audio_data)
                                     break
@@ -262,11 +245,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:
@@ -435,11 +415,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/xunfei_stream.py

@@ -101,11 +101,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:
@@ -232,13 +227,8 @@ class ASRProvider(ASRProviderBase):
 
                     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:
@@ -262,13 +252,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, asr_audio_task: List[bytes]):
         """处理语音停止，发送最后一帧并处理识别结果"""
@@ -363,10 +347,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 = []