diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 41ac96b7..e6bec412 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -326,6 +326,7 @@ VAD: SileroVAD: type: silero threshold: 0.5 + threshold_low: 0.3 model_dir: models/snakers4_silero-vad min_silence_duration_ms: 200 # 如果说话停顿比较长，可以把这个值设置大一些 diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index f12ffc5e..7b45ad2d 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -17,6 +17,7 @@ from core.utils.util import ( filter_sensitive_info, ) from typing import Dict, Any +from collections import deque from core.utils.modules_initialize import ( initialize_modules, initialize_tts, @@ -112,6 +113,8 @@ class ConnectionHandler: self.client_have_voice = False self.last_activity_time = 0.0 # 统一的活动时间戳（毫秒） self.client_voice_stop = False + self.client_voice_window = deque(maxlen=5) + self.last_is_voice = False # asr相关变量 # 因为实际部署时可能会用到公共的本地ASR，不能把变量暴露给公共ASR @@ -608,13 +611,24 @@ class ConnectionHandler: # 更新系统prompt至上下文 self.dialogue.update_system_message(self.prompt) - def chat(self, query, tool_call=False): + def chat(self, query, tool_call=False, depth=0): self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") self.llm_finish_task = False if not tool_call: self.dialogue.put(Message(role="user", content=query)) + # 为最顶层时新建会话ID和发送FIRST请求 + if depth == 0: + self.sentence_id = str(uuid.uuid4().hex) + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) + # Define intent functions functions = None if self.intent_type == "function_call" and hasattr(self, "func_handler"): @@ -630,8 +644,6 @@ class ConnectionHandler: ) memory_str = future.result() - self.sentence_id = str(uuid.uuid4().hex) - if self.intent_type == "function_call" and functions is not None: # 使用支持functions的streaming接口 llm_responses = self.llm.response_with_functions( @@ -654,7 +666,6 @@ class ConnectionHandler: function_id = None function_arguments = "" content_arguments = "" - text_index = 0 self.client_abort = False for response in llm_responses: if self.client_abort: @@ -684,14 +695,6 @@ class ConnectionHandler: if content is not None and len(content) > 0: if not tool_call_flag: response_message.append(content) - if text_index == 0: - self.tts.tts_text_queue.put( - TTSMessageDTO( - sentence_id=self.sentence_id, - sentence_type=SentenceType.FIRST, - content_type=ContentType.ACTION, - ) - ) self.tts.tts_text_queue.put( TTSMessageDTO( sentence_id=self.sentence_id, @@ -700,7 +703,6 @@ class ConnectionHandler: content_detail=content, ) ) - text_index += 1 # 处理function call if tool_call_flag: bHasError = False @@ -725,6 +727,11 @@ class ConnectionHandler: f"function call error: {content_arguments}" ) if not bHasError: + # 如需要大模型先处理一轮，添加相关处理后的日志情况 + if len(response_message) > 0: + self.dialogue.put( + Message(role="assistant", content="".join(response_message)) + ) response_message.clear() self.logger.bind(tag=TAG).debug( f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}" @@ -742,14 +749,14 @@ class ConnectionHandler: ), self.loop, ).result() - self._handle_function_result(result, function_call_data) + self._handle_function_result(result, function_call_data, depth=depth) # 存储对话内容 if len(response_message) > 0: self.dialogue.put( Message(role="assistant", content="".join(response_message)) ) - if text_index > 0: + if depth == 0: self.tts.tts_text_queue.put( TTSMessageDTO( sentence_id=self.sentence_id, @@ -764,7 +771,7 @@ class ConnectionHandler: return True - def _handle_function_result(self, result, function_call_data): + def _handle_function_result(self, result, function_call_data, depth): if result.action == Action.RESPONSE: # 直接回复前端 text = result.response self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) @@ -801,7 +808,7 @@ class ConnectionHandler: content=text, ) ) - self.chat(text, tool_call=True) + self.chat(text, tool_call=True, depth=depth + 1) elif result.action == Action.NOTFOUND or result.action == Action.ERROR: text = result.response if result.response else result.result self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) @@ -914,6 +921,9 @@ class ConnectionHandler: except Exception as ws_error: self.logger.bind(tag=TAG).error(f"关闭WebSocket连接时出错: {ws_error}") + if self.tts: + await self.tts.close() + # 最后关闭线程池（避免阻塞） if self.executor: try: diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index f9f8e3ab..ccff094c 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -6,9 +6,8 @@ from core.handle.helloHandle import checkWakeupWords from core.utils.util import remove_punctuation_and_length from core.providers.tts.dto.dto import ContentType from core.utils.dialogue import Message -from core.providers.tools.device_mcp import call_mcp_tool from plugins_func.register import Action, ActionResponse -from loguru import logger +from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType TAG = __name__ @@ -29,6 +28,8 @@ async def handle_user_intent(conn, text): intent_result = await analyze_intent_with_llm(conn, text) if not intent_result: return False + # 会话开始时生成sentence_id + conn.sentence_id = str(uuid.uuid4().hex) # 处理各种意图 return await process_intent_result(conn, intent_result, text) @@ -153,5 +154,19 @@ async def process_intent_result(conn, intent_result, original_text): def speak_txt(conn, text): + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) + ) conn.dialogue.put(Message(role="assistant", content=text)) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 223cb807..9127ec12 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -161,13 +161,6 @@ class TTSProviderBase(ABC): else: sentence_id = str(uuid.uuid4()).replace("-", "") conn.sentence_id = sentence_id - self.tts_text_queue.put( - TTSMessageDTO( - sentence_id=sentence_id, - sentence_type=SentenceType.FIRST, - content_type=ContentType.ACTION, - ) - ) # 对于单句的文本，进行分段处理 segments = re.split(r"([。！？!?；;\n])", content_detail) for seg in segments: @@ -180,13 +173,6 @@ class TTSProviderBase(ABC): content_file=content_file, ) ) - self.tts_text_queue.put( - TTSMessageDTO( - sentence_id=sentence_id, - sentence_type=SentenceType.LAST, - content_type=ContentType.ACTION, - ) - ) async def open_audio_channels(self, conn): self.conn = conn @@ -209,6 +195,8 @@ class TTSProviderBase(ABC): while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) + if message.sentence_type == SentenceType.FIRST: + self.conn.client_abort = False if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue @@ -362,10 +350,8 @@ class TTSProviderBase(ABC): return audio_datas def _process_before_stop_play_files(self): - for tts_file, text in self.before_stop_play_files: - if tts_file and os.path.exists(tts_file): - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text)) + for audio_datas, text in self.before_stop_play_files: + self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text)) self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index 1ebaf15a..0eb7c72d 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -10,7 +10,6 @@ from config.logger import setup_logging from core.utils import opus_encoder_utils from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase -from core.handle.abortHandle import handleAbortMessage from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType from asyncio import Task @@ -175,6 +174,9 @@ class TTSProvider(TTSProviderBase): async def _ensure_connection(self): """建立新的WebSocket连接""" try: + if self.ws: + logger.bind(tag=TAG).info(f"使用已有链接...") + return self.ws logger.bind(tag=TAG).info("开始建立新连接...") ws_header = { "X-Api-App-Key": self.appId, @@ -200,6 +202,10 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).debug( f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}" ) + + if message.sentence_type == SentenceType.FIRST: + self.conn.client_abort = False + if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue @@ -244,9 +250,12 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).info( f"添加音频文件到待播放列表: {message.content_file}" ) - self.before_stop_play_files.append( - (message.content_file, message.content_detail) - ) + if message.content_file and os.path.exists(message.content_file): + # 先处理文件音频数据 + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) if message.sentence_type == SentenceType.LAST: try: @@ -295,25 +304,15 @@ class TTSProvider(TTSProviderBase): async def start_session(self, session_id): logger.bind(tag=TAG).info(f"开始会话～～{session_id}") try: - task = self._monitor_task + # 会话开始时检测上个会话的监听状态 if ( - task is not None - and isinstance(task, Task) - and not task.done() + self._monitor_task is not None + and isinstance(self._monitor_task, Task) + and not self._monitor_task.done() ): - logger.bind(tag=TAG).info("等待上一个监听任务结束...") - if self.ws is not None: - logger.bind(tag=TAG).info("强制关闭上一个WebSocket连接以唤醒监听任务...") - try: - await self.ws.close() - except Exception as e: - logger.bind(tag=TAG).warning(f"关闭上一个ws异常: {e}") - self.ws = None - try: - await asyncio.wait_for(task, timeout=8) - except Exception as e: - logger.bind(tag=TAG).warning(f"等待监听任务异常: {e}") - self._monitor_task = None + logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...") + await self.close() + # 建立新连接 await self._ensure_connection() @@ -336,19 +335,7 @@ class TTSProvider(TTSProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}") # 确保清理资源 - if hasattr(self, "_monitor_task"): - try: - self._monitor_task.cancel() - await self._monitor_task - except: - pass - self._monitor_task = None - if self.ws: - try: - await self.ws.close() - except: - pass - self.ws = None + await self.close() raise async def finish_session(self, session_id): @@ -368,7 +355,7 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).info("会话结束请求已发送") # 等待监听任务完成 - if hasattr(self, "_monitor_task"): + if self._monitor_task: try: await self._monitor_task except Exception as e: @@ -378,28 +365,25 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - # 关闭连接 - await self.close() except Exception as e: logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") # 确保清理资源 - if hasattr(self, "_monitor_task"): - try: - self._monitor_task.cancel() - await self._monitor_task - except: - pass - self._monitor_task = None - if self.ws: - try: - await self.ws.close() - except: - pass - self.ws = None + await self.close() raise async def close(self): """资源清理方法""" + # 取消监听任务 + if self._monitor_task: + try: + self._monitor_task.cancel() + await self._monitor_task + except asyncio.CancelledError: + pass + except Exception as e: + logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}") + self._monitor_task = None + if self.ws: try: await self.ws.close() @@ -413,6 +397,7 @@ class TTSProvider(TTSProviderBase): is_first_sentence = True first_sentence_segment_count = 0 # 添加计数器 try: + session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): try: # 确保 `recv()` 运行在同一个 event loop @@ -466,6 +451,7 @@ class TTSProvider(TTSProviderBase): elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() + session_finished = True break except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") @@ -476,15 +462,15 @@ class TTSProvider(TTSProviderBase): ) traceback.print_exc() break - finally: - # 确保清理资源 - if self.ws: + # 仅在连接异常时才关闭 + if not session_finished and self.ws: try: await self.ws.close() except: pass self.ws = None # 监听任务退出时清理引用 + finally: self._monitor_task = None async def send_event( diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index ee108540..1046b972 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,3 +1,4 @@ +import os import queue import asyncio import traceback @@ -63,9 +64,12 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).info( f"添加音频文件到待播放列表: {message.content_file}" ) - self.before_stop_play_files.append( - (message.content_file, message.content_detail) - ) + if message.content_file and os.path.exists(message.content_file): + # 先处理文件音频数据 + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) if message.sentence_type == SentenceType.LAST: # 处理剩余的文本 diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 2c45ba8c..b516d8fb 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -23,22 +23,24 @@ class VADProvider(VADProviderBase): # 处理空字符串的情况 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") 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 = ( int(min_silence_duration_ms) if min_silence_duration_ms else 1000 ) + # 至少要多少帧才算有语音 + self.frame_window_threshold = 3 + def is_vad(self, conn, opus_packet): try: pcm_frame = self.decoder.decode(opus_packet, 960) conn.client_audio_buffer.extend(pcm_frame) # 将新数据加入缓冲区 - # 确保帧计数器存在 - if not hasattr(conn, "client_voice_frame_count"): - conn.client_voice_frame_count = 0 - # 处理缓冲区中的完整帧（每次处理512采样点） client_have_voice = False while len(conn.client_audio_buffer) >= 512 * 2: @@ -54,15 +56,21 @@ class VADProvider(VADProviderBase): # 检测语音活动 with torch.no_grad(): 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: - 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: diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 6d184b6b..2cbc4018 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -1,13 +1,10 @@ -from config.logger import setup_logging import os import re import time import random -import asyncio import difflib import traceback from pathlib import Path -from core.utils import p3 from core.handle.sendAudioHandle import send_stt_message from plugins_func.register import register_function, ToolType, ActionResponse, Action from core.utils.dialogue import Message @@ -51,8 +48,8 @@ def play_music(conn, song_name: str): ) # 提交异步任务 - future = asyncio.run_coroutine_threadsafe( - handle_music_command(conn, music_intent), conn.loop + task = conn.loop.create_task( + handle_music_command(conn, music_intent) # 封装异步逻辑 ) # 非阻塞回调处理 @@ -63,7 +60,7 @@ def play_music(conn, song_name: str): except Exception as e: conn.logger.bind(tag=TAG).error(f"播放失败: {e}") - future.add_done_callback(handle_done) + task.add_done_callback(handle_done) return ActionResponse( action=Action.NONE, result="指令已接收", response="正在为您播放音乐" @@ -178,13 +175,13 @@ def _get_random_play_prompt(song_name): # 移除文件扩展名 clean_name = os.path.splitext(song_name)[0] prompts = [ - f"正在为您播放，{clean_name}", - f"请欣赏歌曲，{clean_name}", - f"即将为您播放，{clean_name}", - f"为您带来，{clean_name}", - f"让我们聆听，{clean_name}", - f"接下来请欣赏，{clean_name}", - f"为您献上，{clean_name}", + f"正在为您播放，《{clean_name}》", + f"请欣赏歌曲，《{clean_name}》", + f"即将为您播放，《{clean_name}》", + f"现在为您带来，《{clean_name}》", + f"让我们一起聆听，《{clean_name}》", + f"接下来请欣赏，《{clean_name}》", + f"此刻为您献上，《{clean_name}》", ] # 直接使用random.choice，不设置seed return random.choice(prompts) @@ -218,13 +215,14 @@ async def play_local_music(conn, specific_file=None): await send_stt_message(conn, text) conn.dialogue.put(Message(role="assistant", content=text)) - conn.tts.tts_text_queue.put( - TTSMessageDTO( - sentence_id=conn.sentence_id, - sentence_type=SentenceType.FIRST, - content_type=ContentType.ACTION, + if conn.intent_type == "intent_llm": + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) ) - ) conn.tts.tts_text_queue.put( TTSMessageDTO( sentence_id=conn.sentence_id, @@ -241,13 +239,14 @@ async def play_local_music(conn, specific_file=None): content_file=music_path, ) ) - conn.tts.tts_text_queue.put( - TTSMessageDTO( - sentence_id=conn.sentence_id, - sentence_type=SentenceType.LAST, - content_type=ContentType.ACTION, + if conn.intent_type == "intent_llm": + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) ) - ) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")