diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 375bd117..e0eb90bc 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -594,13 +594,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"): @@ -616,8 +627,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( @@ -640,7 +649,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: @@ -670,14 +678,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, @@ -686,7 +686,6 @@ class ConnectionHandler: content_detail=content, ) ) - text_index += 1 # 处理function call if tool_call_flag: bHasError = False @@ -711,6 +710,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}" @@ -728,14 +732,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, @@ -750,7 +754,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) @@ -787,7 +791,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) diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 2d0baca4..a398195b 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) @@ -158,5 +159,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 d196b8f9..c89b3247 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 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 6f8fcca5..49643b85 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -202,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 @@ -300,24 +304,23 @@ 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 + logger.bind(tag=TAG).info("取消上一个监听任务...") + self._monitor_task.cancel() try: - await asyncio.wait_for(task, timeout=8) + # 等待任务取消完成 + await asyncio.wait_for(self._monitor_task, timeout=3) + except asyncio.CancelledError: + logger.bind(tag=TAG).info("监听任务已成功取消") + except asyncio.TimeoutError: + logger.bind(tag=TAG).warning("取消监听任务超时，可能仍在运行") except Exception as e: - logger.bind(tag=TAG).warning(f"等待监听任务异常: {e}") + logger.bind(tag=TAG).warning(f"取消监听任务异常: {e}") self._monitor_task = None # 建立新连接 await self._ensure_connection() @@ -341,19 +344,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): @@ -379,11 +370,14 @@ class TTSProvider(TTSProviderBase): 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 + if not self._monitor_task.done(): + self._monitor_task.cancel() + try: + await self._monitor_task + except asyncio.CancelledError: + pass + except Exception as e: + logger.bind(tag=TAG).warning(f"监听任务取消时发生错误: {e}") except Exception as e: logger.bind(tag=TAG).error( f"等待监听任务完成时发生错误: {str(e)}" @@ -394,31 +388,21 @@ 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 close(self): """资源清理方法""" # 取消监听任务 - if self._monitor_task: + if self._monitor_task and not self._monitor_task.done(): try: self._monitor_task.cancel() await self._monitor_task except asyncio.CancelledError: pass - self._monitor_task = None + except Exception as e: + logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}") + self._monitor_task = None if self.ws: try: diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index a6eb5822..6451e526 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="正在为您播放音乐" @@ -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)}")