Merge pull request #1704 from xinnan-tech/py_websoket_link

update:优化超时资源优化
This commit is contained in:
欣南科技
2025-06-27 23:50:25 +08:00
committed by GitHub
4 changed files with 27 additions and 26 deletions
+19 -15
View File
@@ -110,10 +110,8 @@ class ConnectionHandler:
# vad相关变量 # vad相关变量
self.client_audio_buffer = bytearray() self.client_audio_buffer = bytearray()
self.client_have_voice = False self.client_have_voice = False
self.client_have_voice_last_time = 0.0 self.last_activity_time = 0.0 # 统一的活动时间戳(毫秒)
self.client_no_voice_last_time = 0.0
self.client_voice_stop = False self.client_voice_stop = False
self.client_voice_frame_count = 0
# asr相关变量 # asr相关变量
# 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR # 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR
@@ -143,10 +141,10 @@ class ConnectionHandler:
self.load_function_plugin = False self.load_function_plugin = False
self.intent_type = "nointent" self.intent_type = "nointent"
self.timeout_task = None
self.timeout_seconds = ( self.timeout_seconds = (
int(self.config.get("close_connection_no_voice_time", 120)) + 60 int(self.config.get("close_connection_no_voice_time", 120)) + 60
) # 在原来第一道关闭的基础上加60秒,进行二道关闭 ) # 在原来第一道关闭的基础上加60秒,进行二道关闭
self.timeout_task = None
# {"mcp":true} 表示启用MCP功能 # {"mcp":true} 表示启用MCP功能
self.features = None self.features = None
@@ -187,6 +185,9 @@ class ConnectionHandler:
self.websocket = ws self.websocket = ws
self.device_id = self.headers.get("device-id", None) self.device_id = self.headers.get("device-id", None)
# 初始化活动时间戳
self.last_activity_time = time.time() * 1000
# 启动超时检查任务 # 启动超时检查任务
self.timeout_task = asyncio.create_task(self._check_timeout()) self.timeout_task = asyncio.create_task(self._check_timeout())
@@ -243,12 +244,8 @@ class ConnectionHandler:
async def _route_message(self, message): async def _route_message(self, message):
"""消息路由""" """消息路由"""
# 重置超时计时器
if self.timeout_task:
self.timeout_task.cancel()
self.timeout_task = asyncio.create_task(self._check_timeout())
if isinstance(message, str): if isinstance(message, str):
self.last_activity_time = time.time() * 1000
await handleTextMessage(self, message) await handleTextMessage(self, message)
elif isinstance(message, bytes): elif isinstance(message, bytes):
if self.vad is None: if self.vad is None:
@@ -898,7 +895,6 @@ class ConnectionHandler:
def reset_vad_states(self): def reset_vad_states(self):
self.client_audio_buffer = bytearray() self.client_audio_buffer = bytearray()
self.client_have_voice = False self.client_have_voice = False
self.client_have_voice_last_time = 0
self.client_voice_stop = False self.client_voice_stop = False
self.logger.bind(tag=TAG).debug("VAD states reset.") self.logger.bind(tag=TAG).debug("VAD states reset.")
@@ -917,10 +913,18 @@ class ConnectionHandler:
"""检查连接超时""" """检查连接超时"""
try: try:
while not self.stop_event.is_set(): while not self.stop_event.is_set():
await asyncio.sleep(self.timeout_seconds) # 检查是否超时(只有在时间戳已初始化的情况下)
if not self.stop_event.is_set(): if self.last_activity_time > 0.0:
self.logger.bind(tag=TAG).info("连接超时,准备关闭") current_time = time.time() * 1000
await self.close(self.websocket) if (
break current_time - self.last_activity_time
> self.timeout_seconds * 1000
):
if not self.stop_event.is_set():
self.logger.bind(tag=TAG).info("连接超时,准备关闭")
await self.close(self.websocket)
break
# 每10秒检查一次,避免过于频繁
await asyncio.sleep(10)
except Exception as e: except Exception as e:
self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}") self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")
@@ -66,12 +66,11 @@ async def startToChat(conn, text):
async def no_voice_close_connect(conn, have_voice): async def no_voice_close_connect(conn, have_voice):
if have_voice: if have_voice:
conn.client_no_voice_last_time = 0.0 conn.last_activity_time = time.time() * 1000
return return
if conn.client_no_voice_last_time == 0.0: # 只有在已经初始化过时间戳的情况下才进行超时检查
conn.client_no_voice_last_time = time.time() * 1000 if conn.last_activity_time > 0.0:
else: no_voice_time = time.time() * 1000 - conn.last_activity_time
no_voice_time = time.time() * 1000 - conn.client_no_voice_last_time
close_connection_no_voice_time = int( close_connection_no_voice_time = int(
conn.config.get("close_connection_no_voice_time", 120) conn.config.get("close_connection_no_voice_time", 120)
) )
@@ -93,7 +93,7 @@ async def sendAudio(conn, audios, pre_buffer=True):
break break
# 重置没有声音的状态 # 重置没有声音的状态
conn.client_no_voice_last_time = 0.0 conn.last_activity_time = time.time() * 1000
# 计算预期发送时间 # 计算预期发送时间
expected_time = start_time + (play_position / 1000) expected_time = start_time + (play_position / 1000)
@@ -35,7 +35,7 @@ class VADProvider(VADProviderBase):
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"): if not hasattr(conn, "client_voice_frame_count"):
conn.client_voice_frame_count = 0 conn.client_voice_frame_count = 0
@@ -66,14 +66,12 @@ class VADProvider(VADProviderBase):
# 如果之前有声音,但本次没有声音,且与上次有声音的时间差已经超过了静默阈值,则认为已经说完一句话 # 如果之前有声音,但本次没有声音,且与上次有声音的时间差已经超过了静默阈值,则认为已经说完一句话
if conn.client_have_voice and not client_have_voice: if conn.client_have_voice and not client_have_voice:
stop_duration = ( stop_duration = time.time() * 1000 - conn.last_activity_time
time.time() * 1000 - conn.client_have_voice_last_time
)
if stop_duration >= self.silence_threshold_ms: if stop_duration >= self.silence_threshold_ms:
conn.client_voice_stop = True conn.client_voice_stop = True
if client_have_voice: if client_have_voice:
conn.client_have_voice = True conn.client_have_voice = True
conn.client_have_voice_last_time = time.time() * 1000 conn.last_activity_time = time.time() * 1000
return client_have_voice return client_have_voice
except opuslib_next.OpusError as e: except opuslib_next.OpusError as e: