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

This commit is contained in:
Sakura-RanChen
2026-02-02 11:29:47 +08:00
parent 9682f615cf
commit f5b4bae927
8 changed files with 53 additions and 125 deletions
+15 -3
View File
@@ -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"""
@@ -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
@@ -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"] # 保留原始文本
@@ -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):
"""发送停止识别请求(不关闭连接)"""
@@ -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):
"""发送停止请求(用于手动模式停止录音)"""
+11 -7
View File
@@ -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]
@@ -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 = []
@@ -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 = []