diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 6193ca99..1ea88e9b 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -606,7 +606,7 @@ class ConnectionHandler: text_index = 0 for response in llm_responses: - if self.intent_type == "function_call": + if functions is not None: content, tools_call = response if "content" in response: content = response["content"] @@ -822,6 +822,9 @@ class ConnectionHandler: type, text, audio_data, report_time = item try: + # 检查线程池状态 + if self.executor is None: + continue # 提交任务到线程池 self.executor.submit( self._process_report, type, text, audio_data, report_time @@ -852,35 +855,37 @@ class ConnectionHandler: async def close(self, ws=None): """资源清理方法""" + try: + # 取消超时任务 + if self.timeout_task: + self.timeout_task.cancel() + self.timeout_task = None - # 取消超时任务 - if self.timeout_task: - self.timeout_task.cancel() - self.timeout_task = None + # 清理MCP资源 + if hasattr(self, "mcp_manager") and self.mcp_manager: + await self.mcp_manager.cleanup_all() - # 清理MCP资源 - if hasattr(self, "mcp_manager") and self.mcp_manager: - await self.mcp_manager.cleanup_all() + # 触发停止事件 + if self.stop_event: + self.stop_event.set() - # 触发停止事件 - if self.stop_event: - self.stop_event.set() + # 清空任务队列 + self.clear_queues() - # 清空任务队列 - self.clear_queues() + # 关闭WebSocket连接 + if ws: + await ws.close() + elif self.websocket: + await self.websocket.close() - # 关闭WebSocket连接 - if ws: - await ws.close() - elif self.websocket: - await self.websocket.close() + # 最后关闭线程池（避免阻塞） + if self.executor: + self.executor.shutdown(wait=False) + self.executor = None - # 最后关闭线程池（避免阻塞） - if self.executor: - self.executor.shutdown(wait=False) - self.executor = None - - self.logger.bind(tag=TAG).info("连接资源已释放") + self.logger.bind(tag=TAG).info("连接资源已释放") + except Exception as e: + self.logger.bind(tag=TAG).error(f"关闭连接时出错: {e}") def clear_queues(self): """清空所有任务队列""" @@ -890,7 +895,11 @@ class ConnectionHandler: ) # 使用非阻塞方式清空队列 - for q in [self.tts.tts_text_queue, self.tts.tts_audio_queue]: + for q in [ + self.tts.tts_text_queue, + self.tts.tts_audio_queue, + self.report_queue, + ]: if not q: continue while True: 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 095437a9..25548958 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -161,6 +161,12 @@ class TTSProvider(TTSProviderBase): ) check_model_key("TTS", self.access_token) + # 添加会话状态控制 + self._session_lock = asyncio.Lock() # 会话操作的并发锁 + self._current_session_id = None # 当前会话ID + self._session_started = False # 会话是否已开始 + self._session_finished = False # 会话是否已结束 + ################################################################################### # 火山双流式TTS重写父类的方法--开始 ################################################################################### @@ -236,17 +242,22 @@ class TTSProvider(TTSProviderBase): ) async def _start_monitor_tts_response(self): + opus_datas_cache = [] + # 添加标志来区分是否是第一句话 + is_first_sentence = True while not self.conn.stop_event.is_set(): try: - msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop + # 确保 `recv()` 运行在同一个 event loop + msg = await self.ws.recv() res = self.parser_response(msg) self.print_response(res, "send_text res:") if res.optional.event == EVENT_TTSSentenceStart: json_data = json.loads(res.payload.decode("utf-8")) self.tts_text = json_data.get("text", "") - logger.bind(tag=TAG).info(f"语音生成成功: {self.tts_text}") + logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}") self.tts_audio_queue.put((SentenceType.FIRST, [], self.tts_text)) + opus_datas_cache = [] elif ( res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE @@ -256,9 +267,23 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).debug( f"推送数据到队列里面帧数～～{len(opus_datas)}" ) - self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, None)) + if is_first_sentence: + # 第一句话直接发送 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas, self.tts_text) + ) + else: + # 后续句子缓存 + opus_datas_cache = opus_datas_cache + opus_datas elif res.optional.event == EVENT_TTSSentenceEnd: - logger.bind(tag=TAG).debug(f"句子结束～～{self.tts_text}") + logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") + if not is_first_sentence: + # 只有非第一句话才发送缓存的数据 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, self.tts_text) + ) + # 第一句话结束后，将标志设置为False + is_first_sentence = False elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") for tts_file, text in self.before_stop_play_files: @@ -269,6 +294,9 @@ class TTSProvider(TTSProviderBase): ) self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) + + opus_datas_cache = [] + is_first_sentence = True continue except websockets.ConnectionClosed: break # 连接关闭时退出监听 @@ -422,27 +450,66 @@ class TTSProvider(TTSProviderBase): return async def start_session(self, session_id): - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional(event=EVENT_StartSession, sessionId=session_id).as_bytes() - payload = self.get_payload_bytes(event=EVENT_StartSession, speaker=self.speaker) - await self.send_event(header, optional, payload) logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") + async with self._session_lock: + # 如果已有会话未结束，先关闭它 + if self._session_started and not self._session_finished: + logger.bind(tag=TAG).warning( + f"发现未关闭的会话 {self._current_session_id}，正在关闭..." + ) + await self.finish_session(self._current_session_id) + + # 重置会话状态 + self._current_session_id = session_id + self._session_started = True + self._session_finished = False + + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_StartSession, sessionId=session_id + ).as_bytes() + payload = self.get_payload_bytes( + event=EVENT_StartSession, speaker=self.speaker + ) + await self.send_event(header, optional, payload) async def finish_session(self, session_id): logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional(event=EVENT_FinishSession, sessionId=session_id).as_bytes() - payload = str.encode("{}") - await self.send_event(header, optional, payload) - return + async with self._session_lock: + # 检查会话状态 + if not self._session_started: + logger.bind(tag=TAG).warning(f"尝试关闭未开始的会话 {session_id}") + return + + if self._session_finished: + logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭") + return + + if self._current_session_id != session_id: + logger.bind(tag=TAG).warning( + f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}" + ) + return + + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_FinishSession, sessionId=session_id + ).as_bytes() + payload = str.encode("{}") + await self.send_event(header, optional, payload) + + # 更新会话状态 + self._session_finished = True + self._session_started = False + self._current_session_id = None async def reset(self): # 关闭之前的对话 diff --git a/main/xiaozhi-server/test/test_page.html b/main/xiaozhi-server/test/test_page.html index cd55d5fc..fbad025d 100644 --- a/main/xiaozhi-server/test/test_page.html +++ b/main/xiaozhi-server/test/test_page.html @@ -803,7 +803,10 @@ if (frameData && frameData.length > 0) { // 转换为Float32 const floatData = convertInt16ToFloat32(frameData); - decodedSamples.push(...floatData); + // 使用循环替代展开运算符 + for (let i = 0; i < floatData.length; i++) { + decodedSamples.push(floatData[i]); + } } } catch (error) { log("Opus解码失败: " + error.message, 'error'); @@ -811,8 +814,10 @@ } if (decodedSamples.length > 0) { - // 添加到解码队列 - this.queue.push(...decodedSamples); + // 使用循环替代展开运算符 + for (let i = 0; i < decodedSamples.length; i++) { + this.queue.push(decodedSamples[i]); + } this.totalSamples += decodedSamples.length; // 如果累积了至少0.2秒的音频，开始播放 @@ -868,36 +873,39 @@ this.source = null; this.playing = false; - // 如果队列中还有数据或者缓冲区有新数据，继续播放 - if (this.queue.length > 0) { - setTimeout(() => this.startPlaying(), 10); - } else if (audioBufferQueue.length > 0) { - // 缓冲区有新数据，进行解码 - const frames = [...audioBufferQueue]; - audioBufferQueue = []; - this.decodeOpusFrames(frames); - } else if (this.endOfStream) { - // 流已结束且没有更多数据 - log("音频播放完成", 'info'); - isAudioPlaying = false; - this.endOfStream = false; - streamingContext = null; - } else { - // 等待更多数据 - setTimeout(() => { - // 如果仍然没有新数据，但有更多的包到达 - if (this.queue.length === 0 && audioBufferQueue.length > 0) { - const frames = [...audioBufferQueue]; - audioBufferQueue = []; - this.decodeOpusFrames(frames); - } else if (this.queue.length === 0 && audioBufferQueue.length === 0) { - // 真的没有更多数据了 - log("音频播放完成 (超时)", 'info'); - isAudioPlaying = false; - streamingContext = null; - } - }, 500); // 500ms超时 - } + // 使用setTimeout避免递归调用 + setTimeout(() => { + // 如果队列中还有数据，继续播放 + if (this.queue.length > 0) { + this.startPlaying(); + } else if (audioBufferQueue.length > 0) { + // 缓冲区有新数据，进行解码 + const frames = [...audioBufferQueue]; + audioBufferQueue = []; + this.decodeOpusFrames(frames); + } else if (this.endOfStream) { + // 流已结束且没有更多数据 + log("音频播放完成", 'info'); + isAudioPlaying = false; + this.endOfStream = false; + streamingContext = null; + } else { + // 等待更多数据 + setTimeout(() => { + // 如果仍然没有新数据，但有更多的包到达 + if (this.queue.length === 0 && audioBufferQueue.length > 0) { + const frames = [...audioBufferQueue]; + audioBufferQueue = []; + this.decodeOpusFrames(frames); + } else if (this.queue.length === 0 && audioBufferQueue.length === 0) { + // 真的没有更多数据了 + log("音频播放完成 (超时)", 'info'); + isAudioPlaying = false; + streamingContext = null; + } + }, 500); // 500ms超时 + } + }, 10); // 10ms延迟，避免立即递归 }; this.source.start();