From 7f864eb84d72deefdfca9a09c881bc15da7db00a Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 28 May 2025 11:23:39 +0800 Subject: [PATCH 1/6] =?UTF-8?q?update:=E4=BF=AE=E5=A4=8Dweb=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E5=B7=A5=E5=85=B7Maximum=20call=20stack=20size=20exce?= =?UTF-8?q?eded=20=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/test/test_page.html | 74 ++++++++++++++----------- 1 file changed, 41 insertions(+), 33 deletions(-) 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(); From d7f0e88801b20677a93fc43594c47913bf4417cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=8D=8E=E4=BE=A8?= Date: Wed, 28 May 2025 15:47:49 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E7=81=AB=E5=B1=B1=E5=8F=8C=E5=90=91?= =?UTF-8?q?=E6=B5=81=E5=BC=8Ftts=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/huoshan_double_stream.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) 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..1e76f538 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -236,6 +236,7 @@ class TTSProvider(TTSProviderBase): ) async def _start_monitor_tts_response(self): + opus_datas_cache = [] while not self.conn.stop_event.is_set(): try: msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop @@ -245,8 +246,9 @@ class TTSProvider(TTSProviderBase): 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 +258,10 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).debug( f"推送数据到队列里面帧数～～{len(opus_datas)}" ) - self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, None)) + 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}") + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas_cache, self.tts_text)) elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") for tts_file, text in self.before_stop_play_files: From b225e8afd5013fef3542caf4dbbe91953367e209 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 28 May 2025 16:19:21 +0800 Subject: [PATCH 3/6] =?UTF-8?q?update:=E4=BC=98=E5=8C=96=E7=AC=AC=E4=B8=80?= =?UTF-8?q?=E5=8F=A5=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/tts/huoshan_double_stream.py | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) 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 1e76f538..2d635207 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -237,9 +237,12 @@ 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:") @@ -258,10 +261,23 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).debug( f"推送数据到队列里面帧数～～{len(opus_datas)}" ) - opus_datas_cache = opus_datas_cache + opus_datas + 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).info(f"句子语音生成成功：{self.tts_text}") - self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas_cache, 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: @@ -272,6 +288,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 # 连接关闭时退出监听 From fd17465d94106190083302918c0ec7149c050ad1 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 28 May 2025 18:07:58 +0800 Subject: [PATCH 4/6] =?UTF-8?q?update:=E5=A2=9E=E5=BC=BATTS=E6=B5=81?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=85=B3=E9=97=AD=E5=92=8C=E5=BC=80=E5=90=AF?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/tts/huoshan_double_stream.py | 79 +++++++++++++++---- 1 file changed, 62 insertions(+), 17 deletions(-) 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 2d635207..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重写父类的方法--开始 ################################################################################### @@ -444,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): # 关闭之前的对话 From cd930edd06c3e3a9742ad7a4cfa737fcd827139d Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 28 May 2025 18:22:31 +0800 Subject: [PATCH 5/6] =?UTF-8?q?update:=E4=BC=98=E5=8C=96=E8=B5=84=E6=BA=90?= =?UTF-8?q?=E9=87=8A=E6=94=BE=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 59 ++++++++++++++++---------- 1 file changed, 37 insertions(+), 22 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 11ae6836..f66c42e0 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -805,6 +805,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 @@ -835,35 +838,47 @@ 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() + # 等待上报队列处理完成 + if hasattr(self, "report_queue"): + try: + # 添加毒丸对象 + self.report_queue.put(None) + # 等待队列处理完成 + self.report_queue.join() + except Exception as e: + self.logger.bind(tag=TAG).error(f"等待上报队列处理完成时出错: {e}") - # 清空任务队列 - 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): """清空所有任务队列""" From 2586843654ee5bf962df7f6ed2ae0b871977a25d Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 28 May 2025 21:34:21 +0800 Subject: [PATCH 6/6] =?UTF-8?q?update:=E4=BF=AE=E5=A4=8D=E9=80=80=E5=87=BA?= =?UTF-8?q?=E5=8D=A1=E5=A3=B3=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index f66c42e0..04c352bf 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -597,7 +597,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"] @@ -852,16 +852,6 @@ class ConnectionHandler: if self.stop_event: self.stop_event.set() - # 等待上报队列处理完成 - if hasattr(self, "report_queue"): - try: - # 添加毒丸对象 - self.report_queue.put(None) - # 等待队列处理完成 - self.report_queue.join() - except Exception as e: - self.logger.bind(tag=TAG).error(f"等待上报队列处理完成时出错: {e}") - # 清空任务队列 self.clear_queues() @@ -888,7 +878,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: