From d619d7ac7677493c86f06251445cf2d57b4f7970 Mon Sep 17 00:00:00 2001 From: GoodyHao <865700600@qq.com> Date: Mon, 12 May 2025 17:55:54 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E4=B8=8A?= =?UTF-8?q?=E6=8A=A5=E5=91=BD=E5=90=8D=E5=90=AB=E4=B9=89=EF=BC=8C=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E4=B8=8D=E5=90=8C=E4=B8=8A=E6=8A=A5=E5=BC=80=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 40 ++++++++++--------- .../core/handle/receiveAudioHandle.py | 4 +- .../{ttsReportHandle.py => reportHandle.py} | 37 +++++++++++++---- main/xiaozhi-server/core/handle/textHandle.py | 6 +-- 4 files changed, 55 insertions(+), 32 deletions(-) rename main/xiaozhi-server/core/handle/{ttsReportHandle.py => reportHandle.py} (73%) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 2a85c0ad..0b285fc6 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -33,7 +33,7 @@ from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api from config.manage_api_client import DeviceNotFoundException, DeviceBindException from core.utils.output_counter import add_device_output -from core.handle.ttsReportHandle import enqueue_tts_report, report_tts +from core.handle.reportHandle import enqueue_tts_report, report TAG = __name__ @@ -88,8 +88,11 @@ class ConnectionHandler: self.executor = ThreadPoolExecutor(max_workers=10) # 上报线程 - self.tts_report_queue = queue.Queue() - self.tts_report_thread = None + self.report_queue = queue.Queue() + self.report_thread = None + # TODO(haotian): 2025/5/12 可以通过修改此处，调节asr的上报和tts的上报 + self.report_asr_enable = self.read_config_from_api + self.report_tts_enable = self.read_config_from_api # 依赖的组件 self.vad = None @@ -306,11 +309,11 @@ class ConnectionHandler: """初始化ASR和TTS上报线程""" if not self.read_config_from_api or self.need_bind: return - if self.tts_report_thread is None or not self.tts_report_thread.is_alive(): - self.tts_report_thread = threading.Thread( - target=self._tts_report_worker, daemon=True + if self.report_thread is None or not self.report_thread.is_alive(): + self.report_thread = threading.Thread( + target=self._report_worker, daemon=True ) - self.tts_report_thread.start() + self.report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") def _initialize_private_config(self): @@ -860,8 +863,8 @@ class ConnectionHandler: audio_datas, _ = self.tts.audio_to_pcm_data(tts_file) else: audio_datas, _ = self.tts.audio_to_opus_data(tts_file) - # 在这里上报TTS数据（使用文件路径） - enqueue_tts_report(self, 2, text, audio_datas) + # 在这里上报TTS数据 + enqueue_tts_report(self, text, audio_datas) else: self.logger.bind(tag=TAG).error( f"TTS出错：文件不存在{tts_file}" @@ -917,13 +920,12 @@ class ConnectionHandler: f"audio_play_priority priority_thread: {text} {e}" ) - def _tts_report_worker(self): - """TTS上报工作线程""" - + def _report_worker(self): + """聊天记录上报工作线程""" while not self.stop_event.is_set(): try: # 从队列获取数据，设置超时以便定期检查停止事件 - item = self.tts_report_queue.get(timeout=1) + item = self.report_queue.get(timeout=1) if item is None: # 检测毒丸对象 break @@ -931,18 +933,18 @@ class ConnectionHandler: try: # 执行上报（传入二进制数据） - report_tts(self, type, text, audio_data) + report(self, type, text, audio_data) except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS上报线程异常: {e}") + self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}") finally: # 标记任务完成 - self.tts_report_queue.task_done() + self.report_queue.task_done() except queue.Empty: continue except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS上报工作线程异常: {e}") + self.logger.bind(tag=TAG).error(f"聊天记录上报工作线程异常: {e}") - self.logger.bind(tag=TAG).info("TTS上报线程已退出") + self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") def speak_and_play(self, text, text_index=0): if text is None or len(text) <= 0: @@ -990,7 +992,7 @@ class ConnectionHandler: self.executor = None # 添加毒丸对象到上报队列确保线程退出 - self.tts_report_queue.put(None) + self.report_queue.put(None) # 清空任务队列 self.clear_queues() diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 352d0656..96df9f23 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -4,7 +4,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -from core.handle.ttsReportHandle import enqueue_tts_report +from core.handle.reportHandle import enqueue_asr_report from core.utils.util import audio_to_data TAG = __name__ @@ -44,7 +44,7 @@ async def handleAudioMessage(conn, audio): text_len, _ = remove_punctuation_and_length(text) if text_len > 0: # 使用自定义模块进行上报 - enqueue_tts_report(conn, 1, text, copy.deepcopy(conn.asr_audio)) + enqueue_asr_report(conn, text, copy.deepcopy(conn.asr_audio)) await startToChat(conn, text) else: diff --git a/main/xiaozhi-server/core/handle/ttsReportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py similarity index 73% rename from main/xiaozhi-server/core/handle/ttsReportHandle.py rename to main/xiaozhi-server/core/handle/reportHandle.py index ae8928ce..f8287fdc 100644 --- a/main/xiaozhi-server/core/handle/ttsReportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -11,13 +11,13 @@ TTS上报功能已集成到ConnectionHandler类中。 import opuslib_next -from config.manage_api_client import report +from config.manage_api_client import report as manage_report TAG = __name__ -def report_tts(conn, type, text, opus_data): - """执行TTS上报操作 +def report(conn, type, text, opus_data): + """执行聊天记录上报操作 Args: conn: 连接对象 @@ -31,7 +31,7 @@ def report_tts(conn, type, text, opus_data): else: audio_data = None # 执行上报 - report( + manage_report( mac_address=conn.device_id, session_id=conn.session_id, chat_type=type, @@ -39,7 +39,7 @@ def report_tts(conn, type, text, opus_data): audio=audio_data, ) except Exception as e: - conn.logger.bind(tag=TAG).error(f"TTS上报失败: {e}") + conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}") def opus_to_wav(conn, opus_data): @@ -89,8 +89,8 @@ def opus_to_wav(conn, opus_data): return bytes(wav_header) + pcm_data_bytes -def enqueue_tts_report(conn, type, text, opus_data): - if not conn.read_config_from_api or conn.need_bind: +def enqueue_tts_report(conn, text, opus_data): + if not conn.read_config_from_api or conn.need_bind or not conn.report_tts_enable: return """将TTS数据加入上报队列 @@ -101,10 +101,31 @@ def enqueue_tts_report(conn, type, text, opus_data): """ try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 - conn.tts_report_queue.put((type, text, opus_data)) + conn.report_queue.put((2, text, opus_data)) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) except Exception as e: conn.logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}") + + +def enqueue_asr_report(conn, text, opus_data): + if not conn.read_config_from_api or conn.need_bind or not conn.report_asr_enable: + return + """将ASR数据加入上报队列 + + Args: + conn: 连接对象 + text: 合成文本 + opus_data: opus音频数据 + """ + try: + # 使用连接对象的队列，传入文本和二进制数据而非文件路径 + conn.report_queue.put((1, text, opus_data)) + + conn.logger.bind(tag=TAG).debug( + f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " + ) + except Exception as e: + conn.logger.bind(tag=TAG).error(f"加入ASR上报队列失败: {text}, {e}") diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index a3acd3f0..79874f1c 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -5,7 +5,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.iotHandle import handleIotDescriptors, handleIotStatus -from core.handle.ttsReportHandle import enqueue_tts_report +from core.handle.reportHandle import enqueue_asr_report import asyncio TAG = __name__ @@ -56,11 +56,11 @@ async def handleTextMessage(conn, message): await send_tts_message(conn, "stop", None) elif is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_tts_report(conn, 1, "嘿，你好呀", []) + enqueue_asr_report(conn, "嘿，你好呀", []) await startToChat(conn, "嘿，你好呀") else: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_tts_report(conn, 1, text, []) + enqueue_asr_report(conn, text, []) # 否则需要LLM对文字内容进行答复 await startToChat(conn, text) elif msg_json["type"] == "iot": From 0b11953f6f93b54aacf5bb425155b66701eea34d Mon Sep 17 00:00:00 2001 From: goodyhao <865700600@qq.com> Date: Mon, 12 May 2025 17:55:54 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E4=B8=8A?= =?UTF-8?q?=E6=8A=A5=E5=91=BD=E5=90=8D=E5=90=AB=E4=B9=89=EF=BC=8C=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E4=B8=8D=E5=90=8C=E4=B8=8A=E6=8A=A5=E5=BC=80=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 40 ++++++++-------- .../core/handle/receiveAudioHandle.py | 4 +- .../{ttsReportHandle.py => reportHandle.py} | 46 +++++++++++++++---- main/xiaozhi-server/core/handle/textHandle.py | 6 +-- 4 files changed, 63 insertions(+), 33 deletions(-) rename main/xiaozhi-server/core/handle/{ttsReportHandle.py => reportHandle.py} (69%) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 5f2286db..63b79736 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -33,7 +33,7 @@ from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api from config.manage_api_client import DeviceNotFoundException, DeviceBindException from core.utils.output_counter import add_device_output -from core.handle.ttsReportHandle import enqueue_tts_report, report_tts +from core.handle.reportHandle import enqueue_tts_report, report TAG = __name__ @@ -89,8 +89,11 @@ class ConnectionHandler: self.executor = ThreadPoolExecutor(max_workers=10) # 上报线程 - self.tts_report_queue = queue.Queue() - self.tts_report_thread = None + self.report_queue = queue.Queue() + self.report_thread = None + # TODO(haotian): 2025/5/12 可以通过修改此处，调节asr的上报和tts的上报 + self.report_asr_enable = self.read_config_from_api + self.report_tts_enable = self.read_config_from_api # 依赖的组件 self.vad = None @@ -317,11 +320,11 @@ class ConnectionHandler: return if self.chat_history_conf == 0: return - if self.tts_report_thread is None or not self.tts_report_thread.is_alive(): - self.tts_report_thread = threading.Thread( - target=self._tts_report_worker, daemon=True + if self.report_thread is None or not self.report_thread.is_alive(): + self.report_thread = threading.Thread( + target=self._report_worker, daemon=True ) - self.tts_report_thread.start() + self.report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") def _initialize_private_config(self): @@ -872,8 +875,8 @@ class ConnectionHandler: audio_datas, _ = self.tts.audio_to_pcm_data(tts_file) else: audio_datas, _ = self.tts.audio_to_opus_data(tts_file) - # 在这里上报TTS数据（使用文件路径） - enqueue_tts_report(self, 2, text, audio_datas) + # 在这里上报TTS数据 + enqueue_tts_report(self, text, audio_datas) else: self.logger.bind(tag=TAG).error( f"TTS出错：文件不存在{tts_file}" @@ -929,13 +932,12 @@ class ConnectionHandler: f"audio_play_priority priority_thread: {text} {e}" ) - def _tts_report_worker(self): - """TTS上报工作线程""" - + def _report_worker(self): + """聊天记录上报工作线程""" while not self.stop_event.is_set(): try: # 从队列获取数据，设置超时以便定期检查停止事件 - item = self.tts_report_queue.get(timeout=1) + item = self.report_queue.get(timeout=1) if item is None: # 检测毒丸对象 break @@ -943,18 +945,18 @@ class ConnectionHandler: try: # 执行上报（传入二进制数据） - report_tts(self, type, text, audio_data) + report(self, type, text, audio_data) except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS上报线程异常: {e}") + self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}") finally: # 标记任务完成 - self.tts_report_queue.task_done() + self.report_queue.task_done() except queue.Empty: continue except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS上报工作线程异常: {e}") + self.logger.bind(tag=TAG).error(f"聊天记录上报工作线程异常: {e}") - self.logger.bind(tag=TAG).info("TTS上报线程已退出") + self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") def speak_and_play(self, text, text_index=0): if text is None or len(text) <= 0: @@ -1002,7 +1004,7 @@ class ConnectionHandler: self.executor = None # 添加毒丸对象到上报队列确保线程退出 - self.tts_report_queue.put(None) + self.report_queue.put(None) # 清空任务队列 self.clear_queues() diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 352d0656..96df9f23 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -4,7 +4,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -from core.handle.ttsReportHandle import enqueue_tts_report +from core.handle.reportHandle import enqueue_asr_report from core.utils.util import audio_to_data TAG = __name__ @@ -44,7 +44,7 @@ async def handleAudioMessage(conn, audio): text_len, _ = remove_punctuation_and_length(text) if text_len > 0: # 使用自定义模块进行上报 - enqueue_tts_report(conn, 1, text, copy.deepcopy(conn.asr_audio)) + enqueue_asr_report(conn, text, copy.deepcopy(conn.asr_audio)) await startToChat(conn, text) else: diff --git a/main/xiaozhi-server/core/handle/ttsReportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py similarity index 69% rename from main/xiaozhi-server/core/handle/ttsReportHandle.py rename to main/xiaozhi-server/core/handle/reportHandle.py index a6605e29..bb1ea066 100644 --- a/main/xiaozhi-server/core/handle/ttsReportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -11,13 +11,13 @@ TTS上报功能已集成到ConnectionHandler类中。 import opuslib_next -from config.manage_api_client import report +from config.manage_api_client import report as manage_report TAG = __name__ -def report_tts(conn, type, text, opus_data): - """执行TTS上报操作 +def report(conn, type, text, opus_data): + """执行聊天记录上报操作 Args: conn: 连接对象 @@ -31,7 +31,7 @@ def report_tts(conn, type, text, opus_data): else: audio_data = None # 执行上报 - report( + manage_report( mac_address=conn.device_id, session_id=conn.session_id, chat_type=type, @@ -39,7 +39,7 @@ def report_tts(conn, type, text, opus_data): audio=audio_data, ) except Exception as e: - conn.logger.bind(tag=TAG).error(f"TTS上报失败: {e}") + conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}") def opus_to_wav(conn, opus_data): @@ -89,8 +89,8 @@ def opus_to_wav(conn, opus_data): return bytes(wav_header) + pcm_data_bytes -def enqueue_tts_report(conn, type, text, opus_data): - if not conn.read_config_from_api or conn.need_bind: +def enqueue_tts_report(conn, text, opus_data): + if not conn.read_config_from_api or conn.need_bind or not conn.report_tts_enable: return if conn.chat_history_conf == 0: return @@ -104,14 +104,42 @@ def enqueue_tts_report(conn, type, text, opus_data): try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 if conn.chat_history_conf == 2: - conn.tts_report_queue.put((type, text, opus_data)) + conn.report_queue.put((2, text, opus_data)) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) else: - conn.tts_report_queue.put((type, text, None)) + conn.report_queue.put((2, text, None)) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 不上报音频" ) except Exception as e: conn.logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}") + + +def enqueue_asr_report(conn, text, opus_data): + if not conn.read_config_from_api or conn.need_bind or not conn.report_asr_enable: + return + if conn.chat_history_conf == 0: + return + """将ASR数据加入上报队列 + + Args: + conn: 连接对象 + text: 合成文本 + opus_data: opus音频数据 + """ + try: + # 使用连接对象的队列，传入文本和二进制数据而非文件路径 + if conn.chat_history_conf == 2: + conn.report_queue.put((1, text, opus_data)) + conn.logger.bind(tag=TAG).debug( + f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " + ) + else: + conn.report_queue.put((1, text, None)) + conn.logger.bind(tag=TAG).debug( + f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频" + ) + except Exception as e: + conn.logger.bind(tag=TAG).error(f"加入ASR上报队列失败: {text}, {e}") diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index a3acd3f0..79874f1c 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -5,7 +5,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.iotHandle import handleIotDescriptors, handleIotStatus -from core.handle.ttsReportHandle import enqueue_tts_report +from core.handle.reportHandle import enqueue_asr_report import asyncio TAG = __name__ @@ -56,11 +56,11 @@ async def handleTextMessage(conn, message): await send_tts_message(conn, "stop", None) elif is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_tts_report(conn, 1, "嘿，你好呀", []) + enqueue_asr_report(conn, "嘿，你好呀", []) await startToChat(conn, "嘿，你好呀") else: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_tts_report(conn, 1, text, []) + enqueue_asr_report(conn, text, []) # 否则需要LLM对文字内容进行答复 await startToChat(conn, text) elif msg_json["type"] == "iot":