update: TTS复用链接,VAD双阈值判断

This commit is contained in:
Sakura-RanChen
2025-07-03 09:16:06 +08:00
parent 20adbadc02
commit 69cac9d40a
4 changed files with 52 additions and 19 deletions
+1
View File
@@ -325,6 +325,7 @@ VAD:
SileroVAD: SileroVAD:
type: silero type: silero
threshold: 0.5 threshold: 0.5
threshold_low: 0.3
model_dir: models/snakers4_silero-vad model_dir: models/snakers4_silero-vad
min_silence_duration_ms: 200 # 如果说话停顿比较长,可以把这个值设置大一些 min_silence_duration_ms: 200 # 如果说话停顿比较长,可以把这个值设置大一些
+6
View File
@@ -17,6 +17,7 @@ from core.utils.util import (
filter_sensitive_info, filter_sensitive_info,
) )
from typing import Dict, Any from typing import Dict, Any
from collections import deque
from core.utils.modules_initialize import ( from core.utils.modules_initialize import (
initialize_modules, initialize_modules,
initialize_tts, initialize_tts,
@@ -112,6 +113,8 @@ class ConnectionHandler:
self.client_have_voice = False self.client_have_voice = False
self.last_activity_time = 0.0 # 统一的活动时间戳(毫秒) self.last_activity_time = 0.0 # 统一的活动时间戳(毫秒)
self.client_voice_stop = False self.client_voice_stop = False
self.client_voice_window = deque(maxlen=5)
self.last_is_voice = False
# asr相关变量 # asr相关变量
# 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR # 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR
@@ -858,6 +861,9 @@ class ConnectionHandler:
elif self.websocket: elif self.websocket:
await self.websocket.close() await self.websocket.close()
if self.tts:
await self.tts.close()
# 最后关闭线程池(避免阻塞) # 最后关闭线程池(避免阻塞)
if self.executor: if self.executor:
self.executor.shutdown(wait=False) self.executor.shutdown(wait=False)
@@ -1,4 +1,3 @@
import os
import uuid import uuid
import json import json
import queue import queue
@@ -10,7 +9,6 @@ from config.logger import setup_logging
from core.utils import opus_encoder_utils from core.utils import opus_encoder_utils
from core.utils.util import check_model_key from core.utils.util import check_model_key
from core.providers.tts.base import TTSProviderBase from core.providers.tts.base import TTSProviderBase
from core.handle.abortHandle import handleAbortMessage
from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
from asyncio import Task from asyncio import Task
@@ -175,6 +173,9 @@ class TTSProvider(TTSProviderBase):
async def _ensure_connection(self): async def _ensure_connection(self):
"""建立新的WebSocket连接""" """建立新的WebSocket连接"""
try: try:
if self.ws:
logger.bind(tag=TAG).info(f"使用已有链接...")
return self.ws
logger.bind(tag=TAG).info("开始建立新连接...") logger.bind(tag=TAG).info("开始建立新连接...")
ws_header = { ws_header = {
"X-Api-App-Key": self.appId, "X-Api-App-Key": self.appId,
@@ -368,9 +369,17 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).info("会话结束请求已发送") logger.bind(tag=TAG).info("会话结束请求已发送")
# 等待监听任务完成 # 等待监听任务完成
if hasattr(self, "_monitor_task"): if hasattr(self, "_monitor_task") and self._monitor_task is not None:
try: try:
await self._monitor_task # 设置4秒超时等待监听任务结束
await asyncio.wait_for(self._monitor_task, timeout=4)
except asyncio.TimeoutError:
logger.bind(tag=TAG).warning("等待监听任务超时,强制取消")
self._monitor_task.cancel()
try:
await self._monitor_task
except asyncio.CancelledError:
pass
except Exception as e: except Exception as e:
logger.bind(tag=TAG).error( logger.bind(tag=TAG).error(
f"等待监听任务完成时发生错误: {str(e)}" f"等待监听任务完成时发生错误: {str(e)}"
@@ -378,8 +387,6 @@ class TTSProvider(TTSProviderBase):
finally: finally:
self._monitor_task = None self._monitor_task = None
# 关闭连接
await self.close()
except Exception as e: except Exception as e:
logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
# 确保清理资源 # 确保清理资源
@@ -400,6 +407,15 @@ class TTSProvider(TTSProviderBase):
async def close(self): async def close(self):
"""资源清理方法""" """资源清理方法"""
# 取消监听任务
if self._monitor_task:
try:
self._monitor_task.cancel()
await self._monitor_task
except asyncio.CancelledError:
pass
self._monitor_task = None
if self.ws: if self.ws:
try: try:
await self.ws.close() await self.ws.close()
@@ -413,6 +429,7 @@ class TTSProvider(TTSProviderBase):
is_first_sentence = True is_first_sentence = True
first_sentence_segment_count = 0 # 添加计数器 first_sentence_segment_count = 0 # 添加计数器
try: try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set(): while not self.conn.stop_event.is_set():
try: try:
# 确保 `recv()` 运行在同一个 event loop # 确保 `recv()` 运行在同一个 event loop
@@ -466,6 +483,7 @@ class TTSProvider(TTSProviderBase):
elif res.optional.event == EVENT_SessionFinished: elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束~~") logger.bind(tag=TAG).debug(f"会话结束~~")
self._process_before_stop_play_files() self._process_before_stop_play_files()
session_finished = True
break break
except websockets.ConnectionClosed: except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭") logger.bind(tag=TAG).warning("WebSocket连接已关闭")
@@ -476,15 +494,15 @@ class TTSProvider(TTSProviderBase):
) )
traceback.print_exc() traceback.print_exc()
break break
finally: # 仅在连接异常时才关闭
# 确保清理资源 if not session_finished and self.ws:
if self.ws:
try: try:
await self.ws.close() await self.ws.close()
except: except:
pass pass
self.ws = None self.ws = None
# 监听任务退出时清理引用 # 监听任务退出时清理引用
finally:
self._monitor_task = None self._monitor_task = None
async def send_event( async def send_event(
@@ -23,22 +23,24 @@ class VADProvider(VADProviderBase):
# 处理空字符串的情况 # 处理空字符串的情况
threshold = config.get("threshold", "0.5") threshold = config.get("threshold", "0.5")
threshold_low = config.get("threshold_low", "0.2")
min_silence_duration_ms = config.get("min_silence_duration_ms", "1000") min_silence_duration_ms = config.get("min_silence_duration_ms", "1000")
self.vad_threshold = float(threshold) if threshold else 0.5 self.vad_threshold = float(threshold) if threshold else 0.5
self.vad_threshold_low = float(threshold_low) if threshold_low else 0.2
self.silence_threshold_ms = ( self.silence_threshold_ms = (
int(min_silence_duration_ms) if min_silence_duration_ms else 1000 int(min_silence_duration_ms) if min_silence_duration_ms else 1000
) )
# 至少要多少帧才算有语音
self.frame_window_threshold = 3
def is_vad(self, conn, opus_packet): def is_vad(self, conn, opus_packet):
try: try:
pcm_frame = self.decoder.decode(opus_packet, 960) pcm_frame = self.decoder.decode(opus_packet, 960)
conn.client_audio_buffer.extend(pcm_frame) # 将新数据加入缓冲区 conn.client_audio_buffer.extend(pcm_frame) # 将新数据加入缓冲区
# 确保帧计数器存在
if not hasattr(conn, "client_voice_frame_count"):
conn.client_voice_frame_count = 0
# 处理缓冲区中的完整帧(每次处理512采样点) # 处理缓冲区中的完整帧(每次处理512采样点)
client_have_voice = False client_have_voice = False
while len(conn.client_audio_buffer) >= 512 * 2: while len(conn.client_audio_buffer) >= 512 * 2:
@@ -54,15 +56,21 @@ class VADProvider(VADProviderBase):
# 检测语音活动 # 检测语音活动
with torch.no_grad(): with torch.no_grad():
speech_prob = self.model(audio_tensor, 16000).item() speech_prob = self.model(audio_tensor, 16000).item()
is_voice = speech_prob >= self.vad_threshold
if is_voice: # 双阈值判断
conn.client_voice_frame_count += 1 if speech_prob >= self.vad_threshold:
is_voice = True
elif speech_prob <= self.vad_threshold_low:
is_voice = False
else: else:
conn.client_voice_frame_count = 0 is_voice = conn.last_is_voice
# 只有连续4帧检测到语音才认为有 # 声音没低于最低值则延续前一个状态,判断为有
client_have_voice = conn.client_voice_frame_count >= 4 conn.last_is_voice = is_voice
# 更新滑动窗口
conn.client_voice_window.append(is_voice)
client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold)
# 如果之前有声音,但本次没有声音,且与上次有声音的时间差已经超过了静默阈值,则认为已经说完一句话 # 如果之前有声音,但本次没有声音,且与上次有声音的时间差已经超过了静默阈值,则认为已经说完一句话
if conn.client_have_voice and not client_have_voice: if conn.client_have_voice and not client_have_voice: