diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 80e3b18f..42271f91 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -14,7 +14,6 @@ import websockets from typing import Dict, Any from plugins_func.loadplugins import auto_import_modules from config.logger import setup_logging -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType from core.utils.dialogue import Message, Dialogue from core.handle.textHandle import handleTextMessage from core.utils.util import ( @@ -25,8 +24,7 @@ from core.utils.util import ( check_asr_update, filter_sensitive_info, ) -from concurrent.futures import ThreadPoolExecutor, TimeoutError -from core.handle.sendAudioHandle import sendAudioMessage +from concurrent.futures import ThreadPoolExecutor from core.handle.receiveAudioHandle import handleAudioMessage from core.handle.functionHandler import FunctionHandler from plugins_func.register import Action, ActionResponse @@ -35,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.reportHandle import enqueue_tts_report, report +from core.handle.reportHandle import report TAG = __name__ @@ -69,8 +67,6 @@ class ConnectionHandler: self.bind_code = None self.read_config_from_api = self.config.get("read_config_from_api", False) - self.tts_stream = self.config.get("TTS_SET", {}).get("TTS_STREAM", False) - self.websocket = None self.headers = None self.device_id = None @@ -78,7 +74,6 @@ class ConnectionHandler: self.client_ip_info = {} self.prompt = None self.welcome_msg = None - self.u_id = None self.max_output_size = 0 self.chat_history_conf = 0 @@ -89,12 +84,7 @@ class ConnectionHandler: # 线程任务相关 self.loop = asyncio.get_event_loop() self.stop_event = threading.Event() - self.tts_queue = queue.Queue() - self.tts_queue_stream = queue.Queue() - self.audio_play_queue = queue.Queue() - max_workers = self.config.get("TTS_SET", {}).get("MAX_WORKERS", 10) - self.executor = ThreadPoolExecutor(max_workers=max_workers) - self.start_tts_request_flag = False + self.executor = ThreadPoolExecutor(max_workers=10) # 上报线程 self.report_queue = queue.Queue() @@ -106,10 +96,11 @@ class ConnectionHandler: # 依赖的组件 self.vad = None self.asr = None + self.tts = None self._asr = _asr self._vad = _vad + self._tts = _tts self.llm = _llm - self.tts = _tts self.memory = _memory self.intent = _intent @@ -131,7 +122,6 @@ class ConnectionHandler: # tts相关变量 self.tts_first_text_index = -1 self.tts_last_text_index = -1 - self.tts_duration = 0 # iot相关变量 self.iot_descriptors = {} @@ -203,15 +193,6 @@ class ConnectionHandler: # 异步初始化 self.executor.submit(self._initialize_components) - # 音频播放 消化线程 - self.audio_play_priority_thread = threading.Thread( - target=self._audio_play_priority_thread, daemon=True - ) - self.audio_play_priority_thread.start() - - # 打开音频通道 - await self.tts.open_audio_channels() - try: async for message in self.websocket: await self._route_message(message) @@ -332,6 +313,9 @@ class ConnectionHandler: self.vad = self._vad if self.asr is None: self.asr = self._asr + if self.tts is None: + self.tts = self._tts + self.tts.startSession(self) """加载记忆""" self._initialize_memory() """加载意图识别""" @@ -512,80 +496,20 @@ class ConnectionHandler: # 更新系统prompt至上下文 self.dialogue.update_system_message(self.prompt) - def chat(self, query): - - self.dialogue.put(Message(role="user", content=query)) - - response_message = [] - try: - # 使用带记忆的对话 - memory_str = None - if self.memory is not None: - future = asyncio.run_coroutine_threadsafe( - self.memory.query_memory(query), self.loop - ) - memory_str = future.result() - - self.logger.bind(tag=TAG).debug(f"记忆内容: {memory_str}") - llm_responses = self.llm.response( - self.session_id, self.dialogue.get_llm_dialogue_with_memory(memory_str) - ) - except Exception as e: - self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}") - return None - - self.llm_finish_task = False - text_index = 0 - uuid_str = str(uuid.uuid4()).replace("-", "") - self.u_id = uuid_str - for content in llm_responses: - response_message.append(content) - if self.client_abort: - break - if text_index == 0: - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content="" - ) - ) - self.start_tts_request_flag = True - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=content - ) - ) - text_index += 1 - if self.start_tts_request_flag: - self.start_tts_request_flag = False - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content="" - ) - ) - - self.llm_finish_task = True - self.dialogue.put(Message(role="assistant", content="".join(response_message))) - self.logger.bind(tag=TAG).debug( - json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False) - ) - return True - - def chat_with_function_calling(self, query, tool_call=False): - self.logger.bind(tag=TAG).debug(f"Chat with function calling start: {query}") - """Chat with function calling for intent detection using streaming""" + def chat(self, query, tool_call=False): + self.logger.bind(tag=TAG).debug(f"Chat: {query}") if not tool_call: self.dialogue.put(Message(role="user", content=query)) # Define intent functions functions = None - if hasattr(self, "func_handler"): + if self.intent_type == "function_call" and hasattr(self, "func_handler"): functions = self.func_handler.get_functions() response_message = [] + processed_chars = 0 # 跟踪已处理的字符位置 try: - start_time = time.time() - # 使用带记忆的对话 memory_str = None if self.memory is not None: @@ -594,14 +518,18 @@ class ConnectionHandler: ) memory_str = future.result() - # self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}") - - # 使用支持functions的streaming接口 - llm_responses = self.llm.response_with_functions( - self.session_id, - self.dialogue.get_llm_dialogue_with_memory(memory_str), - functions=functions, - ) + if functions is not None: + # 使用支持functions的streaming接口 + llm_responses = self.llm.response_with_functions( + self.session_id, + self.dialogue.get_llm_dialogue_with_memory(memory_str), + functions=functions, + ) + else: + llm_responses = self.llm.response( + self.session_id, + self.dialogue.get_llm_dialogue_with_memory(memory_str), + ) except Exception as e: self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}") return None @@ -615,30 +543,30 @@ class ConnectionHandler: function_id = None function_arguments = "" content_arguments = "" - uuid_str = str(uuid.uuid4()).replace("-", "") - self.u_id = uuid_str + for response in llm_responses: - content, tools_call = response + if self.intent_type == "function_call": + content, tools_call = response + if "content" in response: + content = response["content"] + tools_call = None + if content is not None and len(content) > 0: + content_arguments += content - if "content" in response: - content = response["content"] - tools_call = None - if content is not None and len(content) > 0: - content_arguments += content - - if not tool_call_flag and content_arguments.startswith(""): - # print("content_arguments", content_arguments) - tool_call_flag = True - - if tools_call is not None: - tool_call_flag = True - if tools_call[0].id is not None: - function_id = tools_call[0].id - if tools_call[0].function.name is not None: - function_name = tools_call[0].function.name - if tools_call[0].function.arguments is not None: - function_arguments += tools_call[0].function.arguments + if not tool_call_flag and content_arguments.startswith(""): + # print("content_arguments", content_arguments) + tool_call_flag = True + if tools_call is not None: + tool_call_flag = True + if tools_call[0].id is not None: + function_id = tools_call[0].id + if tools_call[0].function.name is not None: + function_name = tools_call[0].function.name + if tools_call[0].function.arguments is not None: + function_arguments += tools_call[0].function.arguments + else: + content = response if content is not None and len(content) > 0: if not tool_call_flag: response_message.append(content) @@ -647,34 +575,41 @@ class ConnectionHandler: break end_time = time.time() + # self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}") - self.logger.bind(tag=TAG).debug( - f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}" - ) - if text_index == 0: - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, - msg_type=MsgType.START_TTS_REQUEST, - content="", + # 处理文本分段和TTS逻辑 + # 合并当前全部文本并处理未分割部分 + full_text = "".join(response_message) + current_text = full_text[processed_chars:] # 从未处理的位置开始 + + # 查找最后一个有效标点 + punctuations = ("。", ".", "？", "?", "！", "!", "；", ";", "：") + last_punct_pos = -1 + number_flag = True + for punct in punctuations: + pos = current_text.rfind(punct) + prev_char = current_text[pos - 1] if pos - 1 >= 0 else "" + # 如果.前面是数字统一判断为小数 + if prev_char.isdigit() and punct == ".": + number_flag = False + if pos > last_punct_pos and number_flag: + last_punct_pos = pos + + # 找到分割点则处理 + if last_punct_pos != -1: + segment_text_raw = current_text[: last_punct_pos + 1] + segment_text = get_string_no_punctuation_or_emoji( + segment_text_raw + ) + if segment_text: + text_index += 1 + self.recode_first_last_text(segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, None, segment_text, text_index ) - ) - self.start_tts_request_flag = True - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, - msg_type=MsgType.TTS_TEXT_REQUEST, - content=content, - ) - ) - text_index += 1 - if self.start_tts_request_flag: - self.start_tts_request_flag = False - self.tts.tts_text_queue.put( - TTSMessageDTO( - u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content="" - ) - ) + self.tts.tts_queue.put((future, text_index)) + # 更新已处理字符位置 + processed_chars += len(segment_text_raw) # 处理function call if tool_call_flag: @@ -720,6 +655,19 @@ class ConnectionHandler: ) self._handle_function_result(result, function_call_data, text_index + 1) + # 处理最后剩余的文本 + full_text = "".join(response_message) + remaining_text = full_text[processed_chars:] + if remaining_text: + segment_text = get_string_no_punctuation_or_emoji(remaining_text) + if segment_text: + text_index += 1 + self.recode_first_last_text(segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, None, segment_text, text_index + ) + self.tts.tts_queue.put((future, text_index)) + # 存储对话内容 if len(response_message) > 0: self.dialogue.put( @@ -779,7 +727,8 @@ class ConnectionHandler: if result.action == Action.RESPONSE: # 直接回复前端 text = result.response self.recode_first_last_text(text, text_index) - self.tts.tts_one_sentence(self, text) + future = self.executor.submit(self.speak_and_play, None, text, text_index) + self.tts.tts_queue.put((future, text_index)) self.dialogue.put(Message(role="assistant", content=text)) elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复 text = result.result @@ -813,35 +762,15 @@ class ConnectionHandler: content=text, ) ) - self.chat_with_function_calling(text, tool_call=True) + self.chat(text, tool_call=True) elif result.action == Action.NOTFOUND or result.action == Action.ERROR: text = result.result self.recode_first_last_text(text, text_index) - self.tts.tts_one_sentence(self, text) - self.dialogue.put(Message(role="assistant", content=text)) - elif result.action == Action.NONE: - # 啥也不干 - text = result.result + future = self.executor.submit(self.speak_and_play, None, text, text_index) + self.tts.tts_queue.put((future, text_index)) self.dialogue.put(Message(role="assistant", content=text)) else: - text = result.result - self.recode_first_last_text(text, text_index) - self.tts.tts_one_sentence(self, text) - self.dialogue.put(Message(role="assistant", content=text)) - - def _audio_play_priority_thread(self): - while not self.stop_event.is_set(): - text = None - try: - ttsMessageDTO = self.tts.tts_audio_queue.get() - future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self, ttsMessageDTO), self.loop - ) - future.result() - except Exception as e: - self.logger.bind(tag=TAG).error( - f"audio_play_priority priority_thread: {text} {e}" - ) + pass def _report_worker(self): """聊天记录上报工作线程""" @@ -869,6 +798,22 @@ class ConnectionHandler: self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") + def speak_and_play(self, file_path, content, text_index=0): + if file_path is not None: + self.logger.bind(tag=TAG).info(f"无需tts转换: 从文件播放，{file_path}") + return file_path, content, text_index + if content is None or len(content) <= 0: + self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{content}") + return None, content, text_index + tts_file = self.tts.to_tts(content) + if tts_file is None: + self.logger.bind(tag=TAG).error(f"tts转换失败，{content}") + return None, content, text_index + self.logger.bind(tag=TAG).debug(f"TTS 文件生成完毕: {tts_file}") + if self.max_output_size > 0: + add_device_output(self.headers.get("device-id"), len(content)) + return tts_file, content, text_index + def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") self.asr_server_receive = True @@ -905,7 +850,6 @@ class ConnectionHandler: await ws.close() elif self.websocket: await self.websocket.close() - await self.tts.close() # 最后关闭线程池（避免阻塞） if self.executor: @@ -917,11 +861,11 @@ class ConnectionHandler: def clear_queues(self): """清空所有任务队列""" self.logger.bind(tag=TAG).debug( - f"开始清理: TTS队列大小={self.tts_queue.qsize()}, 音频队列大小={self.audio_play_queue.qsize()}" + f"开始清理: TTS队列大小={self.tts.tts_queue.qsize()}, 音频队列大小={self.tts.audio_play_queue.qsize()}" ) # 使用非阻塞方式清空队列 - for q in [self.tts_queue, self.audio_play_queue]: + for q in [self.tts.tts_queue, self.tts.audio_play_queue]: if not q: continue while True: @@ -931,7 +875,7 @@ class ConnectionHandler: break self.logger.bind(tag=TAG).debug( - f"清理结束: TTS队列大小={self.tts_queue.qsize()}, 音频队列大小={self.audio_play_queue.qsize()}" + f"清理结束: TTS队列大小={self.tts.tts_queue.qsize()}, 音频队列大小={self.tts.audio_play_queue.qsize()}" ) def reset_vad_states(self): diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index e29ef450..f81a0229 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -26,7 +26,8 @@ async def handleHelloMessage(conn, msg_json): format = audio_params.get("format") conn.logger.bind(tag=TAG).info(f"客户端音频格式: {format}") conn.audio_format = format - conn.asr.set_audio_format(format) + if conn.asr is not None: + conn.asr.set_audio_format(format) conn.welcome_msg["audio_params"] = audio_params await conn.websocket.send(json.dumps(conn.welcome_msg)) @@ -55,7 +56,7 @@ async def checkWakeupWords(conn, text): text_hello = WAKEUP_CONFIG["text"] if not text_hello: text_hello = text - conn.audio_play_queue.put((opus_packets, text_hello, 0)) + conn.tts.audio_play_queue.put((opus_packets, text_hello, 0)) if time.time() - WAKEUP_CONFIG["create_time"] > WAKEUP_CONFIG["refresh_time"]: asyncio.create_task(wakeupWordsResponse(conn)) return True diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index c1b0f1ee..e61b1265 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -2,6 +2,7 @@ from config.logger import setup_logging import json import uuid from core.handle.sendAudioHandle import send_stt_message +from core.handle.helloHandle import checkWakeupWords from core.utils.util import remove_punctuation_and_length from core.utils.dialogue import Message from plugins_func.register import Action @@ -15,10 +16,9 @@ async def handle_user_intent(conn, text): filtered_text = remove_punctuation_and_length(text)[1] if await check_direct_exit(conn, filtered_text): return True - # 4月4日因流式改造暂时关闭唤醒词加速功能 # 检查是否是唤醒词 - # if await checkWakeupWords(conn, filtered_text): - # return True + if await checkWakeupWords(conn, filtered_text): + return True if conn.intent_type == "function_call": # 使用支持function calling的聊天方法,不再进行意图分析 @@ -109,21 +109,21 @@ async def process_intent_result(conn, intent_result, original_text): if result.action == Action.RESPONSE: # 直接回复前端 text = result.response if text is not None: - speak_and_play(conn, text) + speak_txt(conn, text) elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复 text = result.result conn.dialogue.put(Message(role="tool", content=text)) llm_result = conn.intent.replyResult(text, original_text) if llm_result is None: llm_result = text - speak_and_play(conn, llm_result) + speak_txt(conn, llm_result) elif ( result.action == Action.NOTFOUND or result.action == Action.ERROR ): text = result.result if text is not None: - speak_and_play(conn, text) + speak_txt(conn, text) elif function_name != "play_music": # For backward compatibility with original code # 获取当前最新的文本索引 @@ -131,7 +131,7 @@ async def process_intent_result(conn, intent_result, original_text): if text is None: text = result.result if text is not None: - speak_and_play(conn, text) + speak_txt(conn, text) # 将函数执行放在线程池中 conn.executor.submit(process_function_call) @@ -142,6 +142,12 @@ async def process_intent_result(conn, intent_result, original_text): return False -def speak_and_play(conn, text): - conn.tts.tts_one_sentence(conn, text) +def speak_txt(conn, text): + text_index = ( + conn.tts_last_text_index + 1 if hasattr(conn, "tts_last_text_index") else 0 + ) + conn.recode_first_last_text(text, text_index) + future = conn.executor.submit(conn.speak_and_play, None, text, text_index) + conn.llm_finish_task = True + conn.tts_queue.put((future, text_index)) conn.dialogue.put(Message(role="assistant", content=text)) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 9c693751..91bb2bb4 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -39,7 +39,9 @@ async def handleAudioMessage(conn, audio): if len(conn.asr_audio) < 15: conn.asr_server_receive = True else: - raw_text, _ = await conn.asr.speech_to_text(conn.asr_audio, conn.session_id) # 确保ASR模块返回原始文本 + raw_text, _ = await conn.asr.speech_to_text( + conn.asr_audio, conn.session_id + ) # 确保ASR模块返回原始文本 conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}") text_len, _ = remove_punctuation_and_length(raw_text) if text_len > 0: @@ -76,11 +78,7 @@ async def startToChat(conn, text): # 意图未被处理，继续常规聊天流程 await send_stt_message(conn, text) - if conn.intent_type == "function_call": - # 使用支持function calling的聊天方法 - conn.executor.submit(conn.chat_with_function_calling, text) - else: - conn.executor.submit(conn.chat, text) + conn.executor.submit(conn.chat, text) async def no_voice_close_connect(conn): @@ -117,7 +115,7 @@ async def max_out_size(conn): conn.llm_finish_task = True file_path = "config/assets/max_output_size.wav" opus_packets, _ = audio_to_data(file_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.audio_play_queue.put((opus_packets, text, 0)) conn.close_after_chat = True @@ -139,7 +137,7 @@ async def check_bind_device(conn): # 播放提示音 music_path = "config/assets/bind_code.wav" opus_packets, _ = audio_to_data(music_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.audio_play_queue.put((opus_packets, text, 0)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 @@ -147,7 +145,7 @@ async def check_bind_device(conn): digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" num_packets, _ = audio_to_data(num_path) - conn.audio_play_queue.put((num_packets, None, i + 1)) + conn.tts.audio_play_queue.put((num_packets, None, i + 1)) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue @@ -159,4 +157,4 @@ async def check_bind_device(conn): conn.llm_finish_task = True music_path = "config/assets/bind_not_found.wav" opus_packets, _ = audio_to_data(music_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.audio_play_queue.put((opus_packets, text, 0)) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 5b73ed18..89f16426 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,70 +1,104 @@ -import traceback - -from config.logger import setup_logging import json import asyncio import time - -from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, MsgType -from core.utils.util import ( - remove_punctuation_and_length, - get_string_no_punctuation_or_emoji, -) +from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion TAG = __name__ -logger = setup_logging() + +emoji_map = { + "neutral": "😶", + "happy": "🙂", + "laughing": "😆", + "funny": "😂", + "sad": "😔", + "angry": "😠", + "crying": "😭", + "loving": "😍", + "embarrassed": "😳", + "surprised": "😲", + "shocked": "😱", + "thinking": "🤔", + "winking": "😉", + "cool": "😎", + "relaxed": "😌", + "delicious": "🤤", + "kissy": "😘", + "confident": "😏", + "sleepy": "😴", + "silly": "😜", + "confused": "🙄", +} -async def sendAudioMessage(conn, ttsMessageDTO: TTSMessageDTO): - if ttsMessageDTO.u_id != conn.u_id: - logger.bind(tag=TAG).info( - f"msg id:{ttsMessageDTO.u_id},不是当前对话，当前对话id：{conn.u_id}" - ) - return +async def sendAudioMessage(conn, audios, text, text_index=0): # 发送句子开始消息 - if SentenceType.SENTENCE_START == ttsMessageDTO.sentence_type: - logger.bind(tag=TAG).info(f"发送第一段语音: {ttsMessageDTO.tts_finish_text}") - await send_tts_message(conn, "sentence_start", ttsMessageDTO.tts_finish_text) + if text is not None: + emotion = analyze_emotion(text) + emoji = emoji_map.get(emotion, "🙂") # 默认使用笑脸 + await conn.websocket.send( + json.dumps( + { + "type": "llm", + "text": emoji, + "emotion": emotion, + "session_id": conn.session_id, + } + ) + ) + if text_index == conn.tts_first_text_index: + conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") + await send_tts_message(conn, "sentence_start", text) + + is_first_audio = text_index == conn.tts_first_text_index + await sendAudio(conn, audios, pre_buffer=is_first_audio) + + await send_tts_message(conn, "sentence_end", text) + + # 发送结束消息（如果是最后一个文本） + if conn.llm_finish_task and text_index == conn.tts_last_text_index: + await send_tts_message(conn, "stop", None) + if conn.close_after_chat: + await conn.close() + + +# 播放音频 +async def sendAudio(conn, audios, pre_buffer=True): # 流控参数优化 - original_frame_duration = 60 # 原始帧时长（毫秒） - adjusted_frame_duration = int(original_frame_duration * 1) # 缩短20% - total_frames = len(ttsMessageDTO.content) # 获取总帧数 - compensation = ( - total_frames * (original_frame_duration - adjusted_frame_duration) / 1000 - ) # 补偿时间（秒） - + frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码 start_time = time.perf_counter() - play_position = 0 # 已播放时长（毫秒） + play_position = 0 + last_reset_time = time.perf_counter() # 记录最后的重置时间 - for opus_packet in ttsMessageDTO.content: + # 仅当第一句话时执行预缓冲 + if pre_buffer: + pre_buffer_frames = min(3, len(audios)) + for i in range(pre_buffer_frames): + await conn.websocket.send(audios[i]) + remaining_audios = audios[pre_buffer_frames:] + else: + remaining_audios = audios + + # 播放剩余音频帧 + for opus_packet in remaining_audios: if conn.client_abort: return - # 计算带加速因子的预期时间 + # 每分钟重置一次计时器 + if time.perf_counter() - last_reset_time > 60: + await conn.reset_timeout() + last_reset_time = time.perf_counter() + + # 计算预期发送时间 expected_time = start_time + (play_position / 1000) current_time = time.perf_counter() - - # 流控等待（使用加速后的帧时长） delay = expected_time - current_time if delay > 0: await asyncio.sleep(delay) await conn.websocket.send(opus_packet) - play_position += adjusted_frame_duration # 使用调整后的帧时长 - # 补偿因加速损失的时长 - # if compensation > 0: - # await asyncio.sleep(compensation) - if SentenceType.SENTENCE_END == ttsMessageDTO.sentence_type: - logger.bind(tag=TAG).info(f"发送最后一段语音: {ttsMessageDTO.tts_finish_text}") - await send_tts_message(conn, "sentence_end", ttsMessageDTO.tts_finish_text) - - # 发送结束消息（如果是最后一个文本） - if conn.llm_finish_task and MsgType.STOP_TTS_RESPONSE == ttsMessageDTO.msg_type: - await send_tts_message(conn, "stop", None) - if conn.close_after_chat: - await conn.close() + play_position += frame_duration async def send_tts_message(conn, state, text=None): @@ -73,10 +107,22 @@ async def send_tts_message(conn, state, text=None): if text is not None: message["text"] = text - await conn.websocket.send(json.dumps(message)) + # TTS播放结束 if state == "stop": + # 播放提示音 + tts_notify = conn.config.get("enable_stop_tts_notify", False) + if tts_notify: + stop_tts_notify_voice = conn.config.get( + "stop_tts_notify_voice", "config/assets/tts_notify.mp3" + ) + audios, _ = conn.tts.audio_to_opus_data(stop_tts_notify_voice) + await sendAudio(conn, audios) + # 清除服务端讲话状态 conn.clearSpeakStatus() + # 发送消息到客户端 + await conn.websocket.send(json.dumps(message)) + async def send_stt_message(conn, text): """发送 STT 状态消息""" @@ -84,14 +130,4 @@ async def send_stt_message(conn, text): await conn.websocket.send( json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id}) ) - await conn.websocket.send( - json.dumps( - { - "type": "llm", - "text": "😊", - "emotion": "happy", - "session_id": conn.session_id, - } - ) - ) await send_tts_message(conn, "start") diff --git a/main/xiaozhi-server/core/opus/opus_encoder_utils.py b/main/xiaozhi-server/core/opus/opus_encoder_utils.py deleted file mode 100644 index d9f6de2f..00000000 --- a/main/xiaozhi-server/core/opus/opus_encoder_utils.py +++ /dev/null @@ -1,137 +0,0 @@ -""" -Opus编码工具类 -将PCM音频数据编码为Opus格式 -""" - -import array -import logging -import traceback - -import numpy as np -from typing import List, Optional -from opuslib_next import Encoder -from opuslib_next import constants - - -class OpusEncoderUtils: - """PCM到Opus的编码器""" - - def __init__(self, sample_rate: int, channels: int, frame_size_ms: int): - """ - 初始化Opus编码器 - - Args: - sample_rate: 采样率 (Hz) - channels: 通道数 (1=单声道, 2=立体声) - frame_size_ms: 帧大小 (毫秒) - """ - self.sample_rate = sample_rate - self.channels = channels - self.frame_size_ms = frame_size_ms - # 计算每帧样本数 = 采样率 * 帧大小(毫秒) / 1000 - self.frame_size = (sample_rate * frame_size_ms) // 1000 - # 总帧大小 = 每帧样本数 * 通道数 - self.total_frame_size = self.frame_size * channels - - # 比特率和复杂度设置 - self.bitrate = 24000 # bps - self.complexity = 10 # 最高质量 - - # 缓冲区初始化为空 - self.buffer = np.array([], dtype=np.int16) - - try: - # 创建Opus编码器 - self.encoder = Encoder( - sample_rate, channels, constants.APPLICATION_AUDIO # 音频优化模式 - ) - self.encoder.bitrate = self.bitrate - self.encoder.complexity = self.complexity - self.encoder.signal = constants.SIGNAL_VOICE # 语音信号优化 - except Exception as e: - logging.error(f"初始化Opus编码器失败: {e}") - raise RuntimeError("初始化失败") from e - - def reset_state(self): - """重置编码器状态""" - self.encoder.reset_state() - self.buffer = np.array([], dtype=np.int16) - - def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: - """ - 将PCM数据编码为Opus格式 - - Args: - pcm_data: PCM字节数据 - end_of_stream: 是否为流的结束 - - Returns: - Opus数据包列表 - """ - # 将字节数据转换为short数组 - new_samples = self._convert_bytes_to_shorts(pcm_data) - - # 校验PCM数据 - self._validate_pcm_data(new_samples) - - # 将新数据追加到缓冲区 - self.buffer = np.append(self.buffer, new_samples) - - opus_packets = [] - offset = 0 - - # 处理所有完整帧 - while offset <= len(self.buffer) - self.total_frame_size: - frame = self.buffer[offset : offset + self.total_frame_size] - output = self._encode(frame) - if output: - opus_packets.append(output) - offset += self.total_frame_size - - # 保留未处理的样本 - self.buffer = self.buffer[offset:] - - # 流结束时处理剩余数据 - if end_of_stream and len(self.buffer) > 0: - # 创建最后一帧并用0填充 - last_frame = np.zeros(self.total_frame_size, dtype=np.int16) - last_frame[: len(self.buffer)] = self.buffer - - output = self._encode(last_frame) - if output: - opus_packets.append(output) - self.buffer = np.array([], dtype=np.int16) - - return opus_packets - - def _encode(self, frame: np.ndarray) -> Optional[bytes]: - """编码一帧音频数据""" - try: - # 将numpy数组转换为bytes - frame_bytes = frame.tobytes() - # opuslib要求输入字节数必须是channels*2的倍数 - encoded = self.encoder.encode(frame_bytes, self.frame_size) - return encoded - except Exception as e: - logging.error(f"Opus编码失败: {e}") - traceback.print_exc() - return None - - def _convert_bytes_to_shorts(self, bytes_data: bytes) -> np.ndarray: - """将字节数组转换为short数组 (16位PCM)""" - # 假设输入是小端字节序的16位PCM - return np.frombuffer(bytes_data, dtype=np.int16) - - def _validate_pcm_data(self, pcm_shorts: np.ndarray) -> None: - """验证PCM数据是否有效""" - # 16位PCM数据范围是 -32768 到 32767 - if np.any((pcm_shorts < -32768) | (pcm_shorts > 32767)): - invalid_samples = pcm_shorts[(pcm_shorts < -32768) | (pcm_shorts > 32767)] - logging.warning(f"发现无效PCM样本: {invalid_samples[:5]}...") - # 在实际应用中可以选择裁剪而不是抛出异常 - # np.clip(pcm_shorts, -32768, 32767, out=pcm_shorts) - - def close(self): - """关闭编码器并释放资源""" - # opuslib没有明确的关闭方法，Python的垃圾回收会处理 - pass diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index fee62364..6606168c 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -239,7 +239,7 @@ class ASRProvider(ASRProviderBase): ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if self._is_token_expired(): - logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...") + logger.warning("Token已过期，正在自动刷新...") self._refresh_token() file_path = None diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 227a906d..9d974180 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -30,15 +30,25 @@ class ASRProviderBase(ABC): @staticmethod def decode_opus(opus_data: List[bytes]) -> bytes: """将Opus音频数据解码为PCM数据""" + try: + decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道 + pcm_data = [] + buffer_size = 960 # 每次处理960个采样点 - decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道 - pcm_data = [] + for opus_packet in opus_data: + try: + # 使用较小的缓冲区大小进行处理 + pcm_frame = decoder.decode(opus_packet, buffer_size) + if pcm_frame: + pcm_data.append(pcm_frame) + except opuslib_next.OpusError as e: + logger.bind(tag=TAG).warning(f"Opus解码错误，跳过当前数据包: {e}") + continue + except Exception as e: + logger.bind(tag=TAG).error(f"音频处理错误: {e}", exc_info=True) + continue - for opus_packet in opus_data: - try: - pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms - pcm_data.append(pcm_frame) - except opuslib_next.OpusError as e: - logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True) - - return pcm_data + return pcm_data + except Exception as e: + logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}", exc_info=True) + return [] diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index ec3df17f..c8446574 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/main/xiaozhi-server/core/providers/asr/fun_local.py @@ -9,10 +9,14 @@ import uuid from core.providers.asr.base import ASRProviderBase from funasr import AutoModel from funasr.utils.postprocess_utils import rich_transcription_postprocess +import shutil TAG = __name__ logger = setup_logging() +MAX_RETRIES = 2 +RETRY_DELAY = 1 # 重试延迟（秒） + # 捕获标准输出 class CaptureOutput: @@ -68,46 +72,69 @@ class ASRProvider(ASRProviderBase): ) -> Tuple[Optional[str], Optional[str]]: """语音转文本主处理逻辑""" file_path = None - try: - # 合并所有opus数据包 - if self.audio_format == "pcm": - pcm_data = opus_data - else: - pcm_data = self.decode_opus(opus_data) + retry_count = 0 - combined_pcm_data = b"".join(pcm_data) + while retry_count < MAX_RETRIES: + try: + # 合并所有opus数据包 + if self.audio_format == "pcm": + pcm_data = opus_data + else: + pcm_data = self.decode_opus(opus_data) - # 判断是否保存为WAV文件 - if self.delete_audio_file: - pass - else: - file_path = self.save_audio_to_file(pcm_data, session_id) + combined_pcm_data = b"".join(pcm_data) - # 语音识别 - start_time = time.time() - result = self.model.generate( - input=combined_pcm_data, - cache={}, - language="auto", - use_itn=True, - batch_size_s=60, - ) - text = rich_transcription_postprocess(result[0]["text"]) - logger.bind(tag=TAG).debug( - f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}" - ) + # 检查磁盘空间 + if not self.delete_audio_file: + free_space = shutil.disk_usage(self.output_dir).free + if free_space < len(combined_pcm_data) * 2: # 预留2倍空间 + raise OSError("磁盘空间不足") - return text, file_path + # 判断是否保存为WAV文件 + if self.delete_audio_file: + pass + else: + file_path = self.save_audio_to_file(pcm_data, session_id) - except Exception as e: - logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True) - return "", file_path + # 语音识别 + start_time = time.time() + result = self.model.generate( + input=combined_pcm_data, + cache={}, + language="auto", + use_itn=True, + batch_size_s=60, + ) + text = rich_transcription_postprocess(result[0]["text"]) + logger.bind(tag=TAG).debug( + f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}" + ) - # finally: - # # 文件清理逻辑 - # if self.delete_audio_file and file_path and os.path.exists(file_path): - # try: - # os.remove(file_path) - # logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}") - # except Exception as e: - # logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}") + return text, file_path + + except OSError as e: + retry_count += 1 + if retry_count >= MAX_RETRIES: + logger.bind(tag=TAG).error( + f"语音识别失败（已重试{retry_count}次）: {e}", exc_info=True + ) + return "", file_path + logger.bind(tag=TAG).warning( + f"语音识别失败，正在重试（{retry_count}/{MAX_RETRIES}）: {e}" + ) + time.sleep(RETRY_DELAY) + + except Exception as e: + logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True) + return "", file_path + + finally: + # 文件清理逻辑 + if self.delete_audio_file and file_path and os.path.exists(file_path): + try: + os.remove(file_path) + logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}") + except Exception as e: + logger.bind(tag=TAG).error( + f"文件删除失败: {file_path} | 错误: {e}" + ) diff --git a/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py b/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py index ee657420..51696bb7 100644 --- a/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py +++ b/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py @@ -72,6 +72,10 @@ class IntentProvider(IntentProviderBase): '返回: {"function_call": {"name": "get_time"}}\n' "```\n" "```\n" + "用户: 当前电池电量是多少？\n" + '返回: {"function_call": {"name": "get_battery_level", "arguments": {"response_success": "当前电池电量为{value}%", "response_failure": "无法获取Battery的当前电量百分比"}}}\n' + "```\n" + "```\n" "用户: 我想结束对话\n" '返回: {"function_call": {"name": "handle_exit_intent", "arguments": {"say_goodbye": "goodbye"}}}\n' "```\n" @@ -224,7 +228,8 @@ class IntentProvider(IntentProviderBase): if function_name == "continue_chat": # 保留非工具相关的消息 clean_history = [ - msg for msg in conn.dialogue.dialogue + msg + for msg in conn.dialogue.dialogue if msg.role not in ["tool", "function"] ] conn.dialogue.dialogue = clean_history diff --git a/main/xiaozhi-server/core/providers/tts/aliyun.py b/main/xiaozhi-server/core/providers/tts/aliyun.py index a71151d2..61ab4364 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun.py @@ -8,81 +8,61 @@ import requests from datetime import datetime from core.providers.tts.base import TTSProviderBase -from pydub import AudioSegment - +import http.client +import urllib.parse import time import uuid from urllib import parse -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType -from config.logger import setup_logging - -TAG = __name__ -logger = setup_logging() - - class AccessToken: @staticmethod def _encode_text(text): encoded_text = parse.quote_plus(text) - return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") - + return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~') + @staticmethod def _encode_dict(dic): keys = dic.keys() dic_sorted = [(key, dic[key]) for key in sorted(keys)] encoded_text = parse.urlencode(dic_sorted) - return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") - + return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~') + @staticmethod def create_token(access_key_id, access_key_secret): - parameters = { - "AccessKeyId": access_key_id, - "Action": "CreateToken", - "Format": "JSON", - "RegionId": "cn-shanghai", - "SignatureMethod": "HMAC-SHA1", - "SignatureNonce": str(uuid.uuid1()), - "SignatureVersion": "1.0", - "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "Version": "2019-02-28", - } + parameters = {'AccessKeyId': access_key_id, + 'Action': 'CreateToken', + 'Format': 'JSON', + 'RegionId': 'cn-shanghai', + 'SignatureMethod': 'HMAC-SHA1', + 'SignatureNonce': str(uuid.uuid1()), + 'SignatureVersion': '1.0', + 'Timestamp': time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + 'Version': '2019-02-28'} # 构造规范化的请求字符串 query_string = AccessToken._encode_dict(parameters) # print('规范化的请求字符串: %s' % query_string) # 构造待签名字符串 - string_to_sign = ( - "GET" - + "&" - + AccessToken._encode_text("/") - + "&" - + AccessToken._encode_text(query_string) - ) + string_to_sign = 'GET' + '&' + AccessToken._encode_text('/') + '&' + AccessToken._encode_text(query_string) # print('待签名的字符串: %s' % string_to_sign) # 计算签名 - secreted_string = hmac.new( - bytes(access_key_secret + "&", encoding="utf-8"), - bytes(string_to_sign, encoding="utf-8"), - hashlib.sha1, - ).digest() + secreted_string = hmac.new(bytes(access_key_secret + '&', encoding='utf-8'), + bytes(string_to_sign, encoding='utf-8'), + hashlib.sha1).digest() signature = base64.b64encode(secreted_string) # print('签名: %s' % signature) # 进行URL编码 signature = AccessToken._encode_text(signature) # print('URL编码后的签名: %s' % signature) # 调用服务 - full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % ( - signature, - query_string, - ) + full_url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s' % (signature, query_string) # print('url: %s' % full_url) # 提交HTTP GET请求 response = requests.get(full_url) if response.ok: root_obj = response.json() - key = "Token" + key = 'Token' if key in root_obj: - token = root_obj[key]["Id"] - expire_time = root_obj[key]["ExpireTime"] + token = root_obj[key]['Id'] + expire_time = root_obj[key]['ExpireTime'] return token, expire_time # print(response.text) return None, None @@ -90,9 +70,10 @@ class AccessToken: class TTSProvider(TTSProviderBase): + def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) - + # 新增空值判断逻辑 self.access_key_id = config.get("access_key_id") self.access_key_secret = config.get("access_key_secret") @@ -106,7 +87,9 @@ class TTSProvider(TTSProviderBase): self.pitch_rate = config.get("pitch_rate", 0) self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com") self.api_url = f"https://{self.host}/stream/v1/tts" - self.header = {"Content-Type": "application/json"} + self.header = { + "Content-Type": "application/json" + } if self.access_key_id and self.access_key_secret: # 使用密钥对生成临时token @@ -116,30 +99,35 @@ class TTSProvider(TTSProviderBase): self.token = config.get("token") self.expire_time = None + def _refresh_token(self): """刷新Token并记录过期时间""" if self.access_key_id and self.access_key_secret: self.token, expire_time_str = AccessToken.create_token( - self.access_key_id, self.access_key_secret + self.access_key_id, + self.access_key_secret ) if not expire_time_str: raise ValueError("无法获取有效的Token过期时间") try: - # 统一转换为字符串处理 + #统一转换为字符串处理 expire_str = str(expire_time_str).strip() if expire_str.isdigit(): expire_time = datetime.fromtimestamp(int(expire_str)) else: - expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ") + expire_time = datetime.strptime( + expire_str, + "%Y-%m-%dT%H:%M:%SZ" + ) self.expire_time = expire_time.timestamp() - 60 except Exception as e: raise ValueError(f"无效的过期时间格式: {expire_str}") from e - + else: self.expire_time = None - + if not self.token: raise ValueError("无法获取有效的访问Token") @@ -154,16 +142,12 @@ class TTSProvider(TTSProviderBase): # f"过期时间 {datetime.fromtimestamp(self.expire_time)} | " # f"剩余 {remaining:.2f}秒") return time.time() > self.expire_time - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) + return os.path.join(self.output_file, f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}") - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): + async def text_to_speak(self, text, output_file): if self._is_token_expired(): - logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...") + logger.warning("Token已过期，正在自动刷新...") self._refresh_token() request_json = { "appkey": self.appkey, @@ -174,45 +158,21 @@ class TTSProvider(TTSProviderBase): "voice": self.voice, "volume": self.volume, "speech_rate": self.speech_rate, - "pitch_rate": self.pitch_rate, + "pitch_rate": self.pitch_rate } - print(self.api_url, json.dumps(request_json, ensure_ascii=False)) + # print(self.api_url, json.dumps(request_json, ensure_ascii=False)) try: - resp = requests.post( - self.api_url, json.dumps(request_json), headers=self.header - ) + resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header) if resp.status_code == 401: # Token过期特殊处理 self._refresh_token() - resp = requests.post( - self.api_url, json.dumps(request_json), headers=self.header - ) + resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header) # 检查返回请求数据的mime类型是否是audio/***，是则保存到指定路径下；返回的是binary格式的 - tmp_file = self.generate_filename() - if resp.headers["Content-Type"].startswith("audio/"): - with open(tmp_file, "wb") as f: + if resp.headers['Content-Type'].startswith('audio/'): + with open(output_file, 'wb') as f: f.write(resp.content) + return output_file else: - raise Exception( - f"{__name__} status_code: {resp.status_code} response: {resp.content}" - ) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass - + raise Exception(f"{__name__} status_code: {resp.status_code} response: {resp.content}") except Exception as e: raise Exception(f"{__name__} error: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 481f1bc5..9ae8e436 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -1,24 +1,15 @@ import asyncio -import gc -import io -import threading -import traceback -import uuid -from concurrent.futures import ThreadPoolExecutor - -import torch -import torchaudio - from config.logger import setup_logging -import numpy as np -from pydub import AudioSegment -from abc import ABC, abstractmethod -from core.utils import textUtils -from core.utils.util import audio_to_data -from core.opus import opus_encoder_utils import queue - -from core.providers.tts.dto.dto import MsgType, TTSMessageDTO, SentenceType +import os +import json +import threading +from core.utils import p3 +from core.handle.sendAudioHandle import sendAudioMessage +from core.handle.reportHandle import enqueue_tts_report +from abc import ABC, abstractmethod +from core.utils.tts import MarkdownCleaner +from core.utils.util import audio_to_data TAG = __name__ logger = setup_logging() @@ -26,273 +17,51 @@ logger = setup_logging() class TTSProviderBase(ABC): def __init__(self, config, delete_audio_file): - self.config = config + self.conn = None + self.tts_timeout = 10 self.delete_audio_file = delete_audio_file self.output_file = config.get("output_dir") - self.tts_text_queue = queue.Queue() - self.tts_audio_queue = queue.Queue() - self.enable_two_way = False - self.stop_event = threading.Event() - self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( - sample_rate=16000, channels=1, frame_size_ms=60 - ) - - self.tts_text_buff = [] - self.punctuations = ( - "。", - "？", - "！", - "；", - "：", - ".", - "?", - "!", - ";", - ":", - " ", - ",", - "，", - ) - self.tts_request = False - self.tts_stop_request = False - self.processed_chars = 0 - self.stream = False - self.last_to_opus_raw = b"" - - # 启动tts_text_queue监听线程 - # 线程任务相关 - self.loop = asyncio.get_event_loop() - self.process_tasks_loop = asyncio.get_event_loop() - self.max_workers = self.config.get("TTS_SET", {}).get("MAX_WORKERS", 3) - self.active_tasks = set() # 追踪当前运行的任务 - self.executor = ThreadPoolExecutor(max_workers=self.max_workers) - - async def open_audio_channels(self): - # 启动tts_text_queue监听线程 - tts_priority = threading.Thread( - target=self._tts_text_priority_thread, daemon=True - ) - tts_priority.start() - - async def close(self): - self.stop_event - - def _get_segment_text(self): - # 合并当前全部文本并处理未分割部分 - full_text = "".join(self.tts_text_buff) - current_text = full_text[self.processed_chars :] # 从未处理的位置开始 - last_punct_pos = -1 - for punct in self.punctuations: - pos = current_text.rfind(punct) - if (pos != -1 and last_punct_pos == -1) or ( - pos != -1 and pos < last_punct_pos - ): - last_punct_pos = pos - if last_punct_pos != -1: - segment_text_raw = current_text[: last_punct_pos + 1] - segment_text = textUtils.get_string_no_punctuation_or_emoji( - segment_text_raw - ) - self.processed_chars += len(segment_text_raw) # 更新已处理字符位置 - return segment_text - elif self.tts_stop_request and current_text: - segment_text = current_text - return segment_text - else: - return None - - async def process_generator(self, generator): - async for tts_data in generator: - self.tts_audio_queue.put(tts_data) - - def _tts_text_priority_thread(self): - logger.bind(tag=TAG).info("开始监听tts文本") - if self.enable_two_way: - self._enable_two_way_tts() - else: - self._no_enable_two_way_tts() - - async def start_session(self, session_id): - pass - - async def finish_session(self, session_id): - pass - - def tts_one_sentence(self, conn, text, u_id=None): - if not u_id: - if conn.u_id: - u_id = conn.u_id - else: - u_id = str(uuid.uuid4()).replace("-", "") - conn.u_id = u_id - self.tts_text_queue.put( - TTSMessageDTO(u_id=u_id, msg_type=MsgType.START_TTS_REQUEST, content="") - ) - self.tts_text_queue.put( - TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_REQUEST, content=text) - ) - self.tts_text_queue.put( - TTSMessageDTO(u_id=u_id, msg_type=MsgType.STOP_TTS_REQUEST, content="") - ) - - def _enable_two_way_tts(self): - while not self.stop_event.is_set(): - try: - ttsMessageDTO = self.tts_text_queue.get() - msg_type = ttsMessageDTO.msg_type - if msg_type == MsgType.START_TTS_REQUEST: - # 开始传输tts文本 - self.tts_request = True - self.tts_stop_request = False - self.u_id = ttsMessageDTO.u_id - # 开启session - future = asyncio.run_coroutine_threadsafe( - self.start_session(ttsMessageDTO.u_id), loop=self.loop - ) - future.result() - # await self.start_session(ttsMessageDTO.u_id) - elif self.tts_request and msg_type == MsgType.TTS_TEXT_REQUEST: - future = asyncio.run_coroutine_threadsafe( - self.text_to_speak( - u_id=ttsMessageDTO.u_id, text=ttsMessageDTO.content - ), - loop=self.loop, - ) - future.result() - elif msg_type == MsgType.STOP_TTS_REQUEST: - self.tts_request = False - self.tts_stop_request = True - future = asyncio.run_coroutine_threadsafe( - self.finish_session(ttsMessageDTO.u_id), loop=self.loop - ) - future.result() - - except Exception as e: - logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}") - # 报错了。要关闭说话 - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.STOP_TTS_RESPONSE, - content=[], - tts_finish_text="", - sentence_type=None, - ) - ) - traceback.print_exc() - - def _no_enable_two_way_tts(self): - # 为这个线程创建一个新的事件循环 - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - while not self.stop_event.is_set(): - try: - ttsMessageDTO = self.tts_text_queue.get() - msg_type = ttsMessageDTO.msg_type - if not self.enable_two_way: - if msg_type == MsgType.START_TTS_REQUEST: - # 开始传输tts文本 - self.tts_request = True - self.tts_stop_request = False - self.processed_chars = 0 - self.tts_text_buff = [] - elif self.tts_request and msg_type == MsgType.TTS_TEXT_REQUEST: - self.tts_text_buff.append(ttsMessageDTO.content) - elif msg_type == MsgType.STOP_TTS_REQUEST: - # 结束传输tts文本,处理最尾巴的数据 - self.tts_request = False - self.tts_stop_request = True - segment_text = self._get_segment_text() - if segment_text: - # 修改部分：创建协程对象 - # 修改部分：创建协程对象 - tts_generator = self.text_to_speak( - ttsMessageDTO.u_id, - segment_text, - True if msg_type == MsgType.STOP_TTS_REQUEST else False, - ( - True - if msg_type == MsgType.START_TTS_REQUEST - else False - ), - ) - future = asyncio.run_coroutine_threadsafe( - self.process_generator(tts_generator), self.loop - ) - self.active_tasks.add(future) - if self.active_tasks: - - async def wrap_future(future): - return await asyncio.wrap_future(future) - - wrapped_tasks = [ - wrap_future(task) for task in self.active_tasks - ] - done, _ = loop.run_until_complete( - asyncio.wait(wrapped_tasks) - ) - self.active_tasks -= done - - # 发送合成结束 - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=ttsMessageDTO.u_id, - msg_type=MsgType.STOP_TTS_RESPONSE, - content=[], - tts_finish_text="", - sentence_type=SentenceType.SENTENCE_END, - ) - ) - - segment_text = self._get_segment_text() - if segment_text: - # 确保这里得到的是协程对象 - tts_generator = self.text_to_speak( - ttsMessageDTO.u_id, - segment_text, - msg_type == MsgType.STOP_TTS_REQUEST, - msg_type == MsgType.START_TTS_REQUEST, - ) - # 提交协程到事件循环 - tts_generator_future = asyncio.run_coroutine_threadsafe( - self.process_generator(tts_generator), loop - ) - self.active_tasks.add(tts_generator_future) - if len(self.active_tasks) >= self.max_workers: - # 等待所有任务完成 - try: - - async def wrap_future(future): - return await asyncio.wrap_future(future) - - wrapped_tasks = [ - wrap_future(task) for task in self.active_tasks - ] - done, _ = loop.run_until_complete( - asyncio.wait(wrapped_tasks) - ) - self.active_tasks -= done - except Exception as e: - logger.bind(tag=TAG).error( - f"Failed to process TTS text: {e}" - ) - traceback.print_exc() - else: - pass - except Exception as e: - logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}") - traceback.print_exc() + self.tts_queue = queue.Queue() + self.audio_play_queue = queue.Queue() @abstractmethod def generate_filename(self): pass - @abstractmethod - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - pass + def to_tts(self, text): + tmp_file = self.generate_filename() + try: + max_repeat_time = 5 + text = MarkdownCleaner.clean_markdown(text) + while not os.path.exists(tmp_file) and max_repeat_time > 0: + try: + asyncio.run(self.text_to_speak(text, tmp_file)) + except Exception as e: + logger.bind(tag=TAG).warning( + f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}" + ) + # 未执行成功，删除文件 + if os.path.exists(tmp_file): + os.remove(tmp_file) + max_repeat_time -= 1 - async def text_to_speak_stream(self, text, queue: queue.Queue, text_index=0): - raise Exception("该TTS还没有实现stream模式") + if max_repeat_time > 0: + logger.bind(tag=TAG).info( + f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次" + ) + else: + logger.bind(tag=TAG).error( + f"语音生成失败: {text}，请检查网络或服务是否正常" + ) + + return tmp_file + except Exception as e: + logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") + return None + + @abstractmethod + async def text_to_speak(self, text, output_file): + pass def audio_to_pcm_data(self, audio_file_path): """音频文件转换为PCM编码""" @@ -302,16 +71,109 @@ class TTSProviderBase(ABC): """音频文件转换为Opus编码""" return audio_to_data(audio_file_path, is_opus=True) - def get_audio_from_tts(self, data_bytes, src_rate, to_rate=16000): - tts_speech = torch.from_numpy( - np.array(np.frombuffer(data_bytes, dtype=np.int16)) - ).unsqueeze(dim=0) - with io.BytesIO() as bf: - torchaudio.save(bf, tts_speech, src_rate, format="wav") - audio = AudioSegment.from_file(bf, format="wav") - audio = audio.set_channels(1).set_frame_rate(to_rate) - return audio + def startSession(self, conn): + self.conn = conn + self.tts_timeout = conn.config.get("tts_timeout", 10) + # tts 消化线程 + self.tts_priority_thread = threading.Thread( + target=self._tts_priority_thread, daemon=True + ) + self.tts_priority_thread.start() - def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False): - opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end) - return opus_datas + # 音频播放 消化线程 + self.audio_play_priority_thread = threading.Thread( + target=self._audio_play_priority_thread, daemon=True + ) + self.audio_play_priority_thread.start() + + def _tts_priority_thread(self): + while not self.conn.stop_event.is_set(): + text = None + try: + try: + item = self.tts_queue.get(timeout=1) + if item is None: + continue + future, text_index = item # 解包获取 Future 和 text_index + except queue.Empty: + if self.conn.stop_event.is_set(): + break + continue + if future is None: + continue + text = None + audio_datas, tts_file = [], None + try: + logger.bind(tag=TAG).debug("正在处理TTS任务...") + tts_file, text, _ = future.result(timeout=self.tts_timeout) + if tts_file is None: + logger.bind(tag=TAG).error( + f"TTS出错： file is empty: {text_index}: {text}" + ) + else: + logger.bind(tag=TAG).debug(f"TTS生成：文件路径: {tts_file}") + if os.path.exists(tts_file): + if tts_file.endswith(".p3"): + audio_datas, _ = p3.decode_opus_from_file(tts_file) + elif self.conn.audio_format == "pcm": + audio_datas, _ = self.audio_to_pcm_data(tts_file) + else: + audio_datas, _ = self.audio_to_opus_data(tts_file) + # 在这里上报TTS数据 + enqueue_tts_report( + self.conn, + tts_file if text is None else text, + audio_datas, + ) + else: + logger.bind(tag=TAG).error(f"TTS出错：文件不存在{tts_file}") + except TimeoutError: + logger.bind(tag=TAG).error("TTS超时") + except Exception as e: + logger.bind(tag=TAG).error(f"TTS出错: {e}") + if not self.conn.client_abort: + # 如果没有中途打断就发送语音 + self.audio_play_queue.put((audio_datas, text, text_index)) + if ( + self.delete_audio_file + and tts_file is not None + and os.path.exists(tts_file) + and tts_file.startswith(self.output_file) + ): + os.remove(tts_file) + except Exception as e: + logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}") + self.conn.clearSpeakStatus() + asyncio.run_coroutine_threadsafe( + self.conn.websocket.send( + json.dumps( + { + "type": "tts", + "state": "stop", + "session_id": self.session_id, + } + ) + ), + self.conn.loop, + ) + logger.bind(tag=TAG).error(f"tts_priority priority_thread: {text} {e}") + + def _audio_play_priority_thread(self): + while not self.conn.stop_event.is_set(): + text = None + try: + try: + audio_datas, text, text_index = self.audio_play_queue.get(timeout=1) + except queue.Empty: + if self.conn.stop_event.is_set(): + break + continue + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self.conn, audio_datas, text, text_index), + self.conn.loop, + ) + future.result() + except Exception as e: + logger.bind(tag=TAG).error( + f"audio_play_priority priority_thread: {text} {e}" + ) diff --git a/main/xiaozhi-server/core/providers/tts/cozecn.py b/main/xiaozhi-server/core/providers/tts/cozecn.py index 61221e85..56314f4f 100644 --- a/main/xiaozhi-server/core/providers/tts/cozecn.py +++ b/main/xiaozhi-server/core/providers/tts/cozecn.py @@ -4,11 +4,7 @@ import json import base64 import requests from datetime import datetime - -from pydub import AudioSegment - from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType class TTSProvider(TTSProviderBase): @@ -20,7 +16,6 @@ class TTSProvider(TTSProviderBase): self.voice = config.get("private_voice") else: self.voice = config.get("voice") - self.response_format = config.get("response_format") self.host = "api.coze.cn" @@ -32,7 +27,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): + async def text_to_speak(self, text, output_file): request_json = { "model": self.model, "input": text, @@ -43,27 +38,13 @@ class TTSProvider(TTSProviderBase): "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", } - response = requests.request( - "POST", self.api_url, json=request_json, headers=headers - ) - data = response.content - tmp_file = self.generate_filename() - file_to_save = open(tmp_file, "wb") - file_to_save.write(data) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 + try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass + response = requests.request( + "POST", self.api_url, json=request_json, headers=headers + ) + data = response.content + file_to_save = open(output_file, "wb") + file_to_save.write(data) + except Exception as e: + raise Exception(f"{__name__} error: {e}") \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/custom.py b/main/xiaozhi-server/core/providers/tts/custom.py index 3ddbeeeb..a5300a3e 100644 --- a/main/xiaozhi-server/core/providers/tts/custom.py +++ b/main/xiaozhi-server/core/providers/tts/custom.py @@ -2,17 +2,13 @@ import os import json import uuid import requests -from pydub import AudioSegment - from config.logger import setup_logging from datetime import datetime from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType TAG = __name__ logger = setup_logging() - class TTSProvider(TTSProviderBase): def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) @@ -33,14 +29,10 @@ class TTSProvider(TTSProviderBase): raise TypeError("Custom TTS配置参数出错, 请参考配置说明") def generate_filename(self): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}.{self.format}", - ) + return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}.{self.format}") - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): + async def text_to_speak(self, text, output_file): request_params = {} - tmp_file = self.generate_filename() for k, v in self.params.items(): if isinstance(v, str) and "{prompt_text}" in v: v = v.replace("{prompt_text}", text) @@ -51,26 +43,9 @@ class TTSProvider(TTSProviderBase): else: resp = requests.get(self.url, params=request_params, headers=self.headers) if resp.status_code == 200: - with open(tmp_file, "wb") as file: + with open(output_file, "wb") as file: file.write(resp.content) else: - logger.bind(tag=TAG).error( - f"Custom TTS请求失败: {resp.status_code} - {resp.text}" - ) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format=self.format) - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass + error_msg = f"Custom TTS请求失败: {resp.status_code} - {resp.text}" + logger.bind(tag=TAG).error(error_msg) + raise Exception(error_msg) # 抛出异常，让调用方捕获 diff --git a/main/xiaozhi-server/core/providers/tts/default.py b/main/xiaozhi-server/core/providers/tts/default.py new file mode 100644 index 00000000..15798a08 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/default.py @@ -0,0 +1,24 @@ +import os +import asyncio +from config.logger import setup_logging +from core.providers.tts.base import TTSProviderBase + +TAG = __name__ +logger = setup_logging() + + +class DefaultTTS(TTSProviderBase): + def __init__(self, config, delete_audio_file=True): + super().__init__(config, delete_audio_file) + self.output_dir = config.get("output_dir", "output") + if not os.path.exists(self.output_dir): + os.makedirs(self.output_dir) + + def generate_filename(self): + """生成唯一的音频文件名""" + import uuid + + return os.path.join(self.output_dir, f"{uuid.uuid4()}.wav") + + async def text_to_speak(self, text, output_file): + logger.bind(tag=TAG).error(f"无法实例化 TTS 服务，请检查配置") diff --git a/main/xiaozhi-server/core/providers/tts/doubao.py b/main/xiaozhi-server/core/providers/tts/doubao.py index 34ec5771..3a0fdde3 100644 --- a/main/xiaozhi-server/core/providers/tts/doubao.py +++ b/main/xiaozhi-server/core/providers/tts/doubao.py @@ -4,10 +4,6 @@ import json import base64 import requests from datetime import datetime - -from pydub import AudioSegment - -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase from config.logger import setup_logging @@ -51,8 +47,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): request_json = { "app": { "appid": f"{self.appid}", @@ -83,7 +78,7 @@ class TTSProvider(TTSProviderBase): ) if "data" in resp.json(): data = resp.json()["data"] - file_to_save = open(tmp_file, "wb") + file_to_save = open(output_file, "wb") file_to_save.write(base64.b64decode(data)) else: raise Exception( @@ -91,15 +86,3 @@ class TTSProvider(TTSProviderBase): ) except Exception as e: raise Exception(f"{__name__} error: {e}") - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_RESPONSE, content=opus_datas, tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass diff --git a/main/xiaozhi-server/core/providers/tts/dto/__init__.py b/main/xiaozhi-server/core/providers/tts/dto/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/main/xiaozhi-server/core/providers/tts/dto/dto.py b/main/xiaozhi-server/core/providers/tts/dto/dto.py deleted file mode 100644 index 706e7d35..00000000 --- a/main/xiaozhi-server/core/providers/tts/dto/dto.py +++ /dev/null @@ -1,42 +0,0 @@ -from enum import Enum -from typing import Union - - -class MsgType(Enum): - # 请求类型 - START_TTS_REQUEST = "START_TTS_REQUEST" - TTS_TEXT_REQUEST = "TTS_TEXT_REQUEST" - STOP_TTS_REQUEST = "STOP_TTS_REQUEST" - - # 返回类型 - START_TTS_RESPONSE = "START_TTS_RESPONSE" - TTS_TEXT_RESPONSE = "TTS_TEXT_RESPONSE" - STOP_TTS_RESPONSE = "STOP_TTS_RESPONSE" - - -class SentenceType(Enum): - # 句子开始 - SENTENCE_START = "SENTENCE_START" - # 句子结束 - SENTENCE_END = "SENTENCE_END" - - -class TTSMessageDTO: - def __init__(self, u_id: str, msg_type: MsgType, content: Union[str, bytes], tts_finish_text=None, - sentence_type: SentenceType = None, duration=0): - if not isinstance(msg_type, MsgType): - raise ValueError("msg_type must be an instance of MsgType Enum") - if not isinstance(content, (str, list, bytes)): - raise ValueError("content must be of type str or bytes") - - # 唯一id，每个合成到合成结束，使用同一个id - self.u_id = u_id - self.msg_type = msg_type - self.sentence_type = sentence_type - self.content = content - self.tts_finish_text = tts_finish_text - self.duration = duration - - def __repr__(self): - content_preview = self.content if isinstance(self.content, str) else "" - return f"MessageDTO(msg_type={self.msg_type}, content={content_preview})" diff --git a/main/xiaozhi-server/core/providers/tts/edge.py b/main/xiaozhi-server/core/providers/tts/edge.py index b623c77c..3d9b2547 100644 --- a/main/xiaozhi-server/core/providers/tts/edge.py +++ b/main/xiaozhi-server/core/providers/tts/edge.py @@ -1,17 +1,8 @@ -import io import os import uuid import edge_tts from datetime import datetime - -from pydub import AudioSegment - -from config.logger import setup_logging from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType - -TAG = __name__ -logger = setup_logging() class TTSProvider(TTSProviderBase): @@ -28,37 +19,19 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): + async def text_to_speak(self, text, output_file): try: - communicate = edge_tts.Communicate( - text, voice=self.voice - ) # Use your preferred voice - tmp_file = self.generate_filename() - await communicate.save(tmp_file) - - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="mp3") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 + communicate = edge_tts.Communicate(text, voice=self.voice) + # 确保目录存在并创建空文件 + os.makedirs(os.path.dirname(output_file), exist_ok=True) + with open(output_file, "wb") as f: pass + + # 流式写入音频数据 + with open(output_file, "ab") as f: # 改为追加模式避免覆盖 + async for chunk in communicate.stream(): + if chunk["type"] == "audio": # 只处理音频数据块 + f.write(chunk["data"]) except Exception as e: - logger.bind(tag=TAG).error(f"TTSProvider text_to_speak error: {e}") - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=[], - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) + error_msg = f"Edge TTS请求失败: {e}" + raise Exception(error_msg) # 抛出异常，让调用方捕获 \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/fishspeech.py b/main/xiaozhi-server/core/providers/tts/fishspeech.py index 2049193e..71f1ac88 100644 --- a/main/xiaozhi-server/core/providers/tts/fishspeech.py +++ b/main/xiaozhi-server/core/providers/tts/fishspeech.py @@ -1,25 +1,14 @@ import base64 import os -import traceback import uuid -import queue -import io - -import numpy as np import requests import ormsgpack from pathlib import Path - -import torch -import torchaudio from pydantic import BaseModel, Field, conint, model_validator -from pydub import AudioSegment from typing_extensions import Annotated from datetime import datetime from typing import Literal - -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType -from core.utils.util import check_model_key +from core.utils.util import check_model_key, parse_string_to_list from core.providers.tts.base import TTSProviderBase from config.logger import setup_logging @@ -150,121 +139,51 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - def _get_audio_from_tts(self, data_bytes): - tts_speech = torch.from_numpy( - np.array(np.frombuffer(data_bytes, dtype=np.int16)) - ).unsqueeze(dim=0) - with io.BytesIO() as bf: - torchaudio.save(bf, tts_speech, 44100, format="wav") - audio = AudioSegment.from_file(bf, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - return audio + async def text_to_speak(self, text, output_file): + # Prepare reference data + byte_audios = [audio_to_bytes(ref_audio) for ref_audio in self.reference_audio] + ref_texts = [read_ref_text(ref_text) for ref_text in self.reference_text] - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - try: - data = { - "text": text, - "reference_id": self.reference_id, - "normalize": self.normalize, - "format": self.format, - "max_new_tokens": self.max_new_tokens, - "chunk_length": self.chunk_length, - "top_p": self.top_p, - "repetition_penalty": self.repetition_penalty, - "temperature": self.temperature, - "streaming": self.streaming, - "use_memory_cache": self.use_memory_cache, - "seed": self.seed, - } + data = { + "text": text, + "references": [ + ServeReferenceAudio(audio=audio if audio else b"", text=text) + for text, audio in zip(ref_texts, byte_audios) + ], + "reference_id": self.reference_id, + "normalize": self.normalize, + "format": self.format, + "max_new_tokens": self.max_new_tokens, + "chunk_length": self.chunk_length, + "top_p": self.top_p, + "repetition_penalty": self.repetition_penalty, + "temperature": self.temperature, + "streaming": self.streaming, + "use_memory_cache": self.use_memory_cache, + "seed": self.seed, + } - # Prepare reference data - if self.reference_audio and self.reference_text: - byte_audios = [ - audio_to_bytes(ref_audio) for ref_audio in self.reference_audio - ] - ref_texts = [ - read_ref_text(ref_text) for ref_text in self.reference_text - ] - data["references"] = ( - [ - ServeReferenceAudio(audio=audio if audio else b"", text=text) - for text, audio in zip(ref_texts, byte_audios) - ], - ) - data["reference_id"] = None + pydantic_data = ServeTTSRequest(**data) - pydantic_data = ServeTTSRequest(**data) - audio_buff = None - chunk_total = b"" - last_raw = b"" - audio_raw = b"" - print("请求tts") - with requests.post( - self.api_url, - data=ormsgpack.packb( - pydantic_data, option=ormsgpack.OPT_SERIALIZE_PYDANTIC - ), - headers={ - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/msgpack", - }, - ) as response: - if response.status_code == 200: - index = 0 - for chunk in response.iter_content(): - # 拼接当前块和上一块数据 - chunk_total += chunk - # 最后一个是静音，说明是一个完整的音频 - if ( - len(chunk_total) % 2 == 0 - and chunk_total[-2:] == b"\x00\x00" - ): - audio = self._get_audio_from_tts(chunk_total) - audio_raw = audio_raw + audio.raw_data - # 长度凑够2贞开始发送，60ms*2=120ms - if len(audio_raw) >= 3840: - opus_datas = self.wav_to_opus_data_audio_raw(audio_raw) - if index == 0: - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - else: - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=None, - ) - audio_raw = b"" - chunk_total = b"" - if len(chunk_total) > 0: - audio = self._get_audio_from_tts(chunk_total) - audio_raw = audio_raw + audio.raw_data - opus_datas = self.wav_to_opus_data_audio_raw(audio_raw) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_END, - ) - else: - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=[], - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_END, - ) + response = requests.post( + self.api_url, + data=ormsgpack.packb( + pydantic_data, option=ormsgpack.OPT_SERIALIZE_PYDANTIC + ), + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/msgpack", + }, + ) - else: - print("请求失败:", response.status_code, response.text) - except Exception as e: - logger.bind(tag=TAG).error("tts发生错误") - traceback.print_exc() - raise e + if response.status_code == 200: + audio_content = response.content + + with open(output_file, "wb") as audio_file: + audio_file.write(audio_content) + + else: + error_msg = f"Request failed with status code {response.status_code}" + print(error_msg) + print(response.json()) + raise Exception(error_msg) diff --git a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py index 5afad8b6..2d58679d 100644 --- a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py +++ b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py @@ -1,12 +1,12 @@ import os import uuid +import json +import base64 import requests -from pydub import AudioSegment -from core.utils.util import parse_string_to_list from config.logger import setup_logging from datetime import datetime from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType +from core.utils.util import parse_string_to_list TAG = __name__ logger = setup_logging() @@ -77,8 +77,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): request_json = { "text": text, "text_lang": self.text_lang, @@ -103,26 +102,9 @@ class TTSProvider(TTSProviderBase): resp = requests.post(self.url, json=request_json) if resp.status_code == 200: - with open(tmp_file, "wb") as file: + with open(output_file, "wb") as file: file.write(resp.content) else: - logger.bind(tag=TAG).error( - f"GPT_SoVITS_V2 TTS请求失败: {resp.status_code} - {resp.text}" - ) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass + error_msg = f"GPT_SoVITS_V2 TTS请求失败: {resp.status_code} - {resp.text}" + logger.bind(tag=TAG).error(error_msg) + raise Exception(error_msg) diff --git a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py index 2391bbb3..d4da23ed 100644 --- a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py +++ b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py @@ -1,12 +1,10 @@ import os import uuid import requests -from pydub import AudioSegment -from core.utils.util import parse_string_to_list from config.logger import setup_logging from datetime import datetime from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType +from core.utils.util import parse_string_to_list TAG = __name__ logger = setup_logging() @@ -44,8 +42,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): request_params = { "refer_wav_path": self.refer_wav_path, "prompt_text": self.prompt_text, @@ -64,26 +61,10 @@ class TTSProvider(TTSProviderBase): resp = requests.get(self.url, params=request_params) if resp.status_code == 200: - with open(tmp_file, "wb") as file: + with open(output_file, "wb") as file: file.write(resp.content) else: - logger.bind(tag=TAG).error( - f"GPT_SoVITS_V3 TTS请求失败: {resp.status_code} - {resp.text}" - ) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass + error_msg = f"GPT_SoVITS_V3 TTS请求失败: {resp.status_code} - {resp.text}" + logger.bind(tag=TAG).error(error_msg) + raise Exception(error_msg) + diff --git a/main/xiaozhi-server/core/providers/tts/huoshan.py b/main/xiaozhi-server/core/providers/tts/huoshan.py deleted file mode 100644 index 734917cf..00000000 --- a/main/xiaozhi-server/core/providers/tts/huoshan.py +++ /dev/null @@ -1,453 +0,0 @@ -import asyncio -import io -import os -import subprocess -import threading -import traceback -import uuid -import json -import base64 -import requests -from datetime import datetime - -import websockets - -from config.logger import setup_logging -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType -from core.utils.util import check_model_key -from core.providers.tts.base import TTSProviderBase - -TAG = __name__ -logger = setup_logging() - -PROTOCOL_VERSION = 0b0001 -DEFAULT_HEADER_SIZE = 0b0001 - -# Message Type: -FULL_CLIENT_REQUEST = 0b0001 -AUDIO_ONLY_RESPONSE = 0b1011 -FULL_SERVER_RESPONSE = 0b1001 -ERROR_INFORMATION = 0b1111 - -# Message Type Specific Flags -MsgTypeFlagNoSeq = 0b0000 # Non-terminal packet with no sequence -MsgTypeFlagPositiveSeq = 0b1 # Non-terminal packet with sequence > 0 -MsgTypeFlagLastNoSeq = 0b10 # last packet with no sequence -MsgTypeFlagNegativeSeq = 0b11 # Payload contains event number (int32) -MsgTypeFlagWithEvent = 0b100 -# Message Serialization -NO_SERIALIZATION = 0b0000 -JSON = 0b0001 -# Message Compression -COMPRESSION_NO = 0b0000 -COMPRESSION_GZIP = 0b0001 - -EVENT_NONE = 0 -EVENT_Start_Connection = 1 - -EVENT_FinishConnection = 2 - -EVENT_ConnectionStarted = 50 # 成功建连 - -EVENT_ConnectionFailed = 51 # 建连失败（可能是无法通过权限认证） - -EVENT_ConnectionFinished = 52 # 连接结束 - -# 上行Session事件 -EVENT_StartSession = 100 - -EVENT_FinishSession = 102 -# 下行Session事件 -EVENT_SessionStarted = 150 -EVENT_SessionFinished = 152 - -EVENT_SessionFailed = 153 - -# 上行通用事件 -EVENT_TaskRequest = 200 - -# 下行TTS事件 -EVENT_TTSSentenceStart = 350 - -EVENT_TTSSentenceEnd = 351 - -EVENT_TTSResponse = 352 - - -class Header: - def __init__( - self, - protocol_version=PROTOCOL_VERSION, - header_size=DEFAULT_HEADER_SIZE, - message_type: int = 0, - message_type_specific_flags: int = 0, - serial_method: int = NO_SERIALIZATION, - compression_type: int = COMPRESSION_NO, - reserved_data=0, - ): - self.header_size = header_size - self.protocol_version = protocol_version - self.message_type = message_type - self.message_type_specific_flags = message_type_specific_flags - self.serial_method = serial_method - self.compression_type = compression_type - self.reserved_data = reserved_data - - def as_bytes(self) -> bytes: - return bytes( - [ - (self.protocol_version << 4) | self.header_size, - (self.message_type << 4) | self.message_type_specific_flags, - (self.serial_method << 4) | self.compression_type, - self.reserved_data, - ] - ) - - -class Optional: - def __init__( - self, event: int = EVENT_NONE, sessionId: str = None, sequence: int = None - ): - self.event = event - self.sessionId = sessionId - self.errorCode: int = 0 - self.connectionId: str | None = None - self.response_meta_json: str | None = None - self.sequence = sequence - - # 转成 byte 序列 - def as_bytes(self) -> bytes: - option_bytes = bytearray() - if self.event != EVENT_NONE: - option_bytes.extend(self.event.to_bytes(4, "big", signed=True)) - if self.sessionId is not None: - session_id_bytes = str.encode(self.sessionId) - size = len(session_id_bytes).to_bytes(4, "big", signed=True) - option_bytes.extend(size) - option_bytes.extend(session_id_bytes) - if self.sequence is not None: - option_bytes.extend(self.sequence.to_bytes(4, "big", signed=True)) - return option_bytes - - -class Response: - def __init__(self, header: Header, optional: Optional): - self.optional = optional - self.header = header - self.payload: bytes | None = None - - def __str__(self): - return super().__str__() - - -class TTSProvider(TTSProviderBase): - def __init__(self, config, delete_audio_file): - super().__init__(config, delete_audio_file) - self.appId = config.get("appid") - self.access_token = config.get("access_token") - self.cluster = config.get("cluster") - self.resource_id = config.get("resource_id") - self.voice = config.get("voice") - self.ws_url = config.get("ws_url") - self.authorization = config.get("authorization") - self.speaker = config.get("speaker") - self.header = {"Authorization": f"{self.authorization}{self.access_token}"} - self.stop_event_response = threading.Event() - self.enable_two_way = True - self.start_connection_flag = False - self.tts_text = "" - - async def open_audio_channels(self): - await super().open_audio_channels() - ws_header = { - "X-Api-App-Key": self.appId, - "X-Api-Access-Key": self.access_token, - "X-Api-Resource-Id": self.resource_id, - "X-Api-Connect-Id": uuid.uuid4(), - } - self.ws = await websockets.connect( - self.ws_url, additional_headers=ws_header, max_size=1000000000 - ) - tts_priority = threading.Thread( - target=self._start_monitor_tts_response_thread(), daemon=True - ) - tts_priority.start() - - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - - async def send_event( - self, header: bytes, optional: bytes | None = None, payload: bytes = None - ): - full_client_request = bytearray(header) - if optional is not None: - full_client_request.extend(optional) - if payload is not None: - payload_size = len(payload).to_bytes(4, "big", signed=True) - full_client_request.extend(payload_size) - full_client_request.extend(payload) - await self.ws.send(full_client_request) - - async def send_text(self, speaker: str, text: str, session_id): - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional(event=EVENT_TaskRequest, sessionId=session_id).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_TaskRequest, text=text, speaker=speaker - ) - return await self.send_event(header, optional, payload) - - # 读取 res 数组某段 字符串内容 - def read_res_content(self, res: bytes, offset: int): - content_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) - offset += 4 - content = str(res[offset : offset + content_size]) - offset += content_size - return content, offset - - # 读取 payload - def read_res_payload(self, res: bytes, offset: int): - payload_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) - offset += 4 - payload = res[offset : offset + payload_size] - offset += payload_size - return payload, offset - - def parser_response(self, res) -> Response: - if isinstance(res, str): - raise RuntimeError(res) - response = Response(Header(), Optional()) - # 解析结果 - # header - header = response.header - num = 0b00001111 - header.protocol_version = res[0] >> 4 & num - header.header_size = res[0] & 0x0F - header.message_type = (res[1] >> 4) & num - header.message_type_specific_flags = res[1] & 0x0F - header.serialization_method = res[2] >> num - header.message_compression = res[2] & 0x0F - header.reserved = res[3] - # - offset = 4 - optional = response.optional - if header.message_type == FULL_SERVER_RESPONSE or AUDIO_ONLY_RESPONSE: - # read event - if header.message_type_specific_flags == MsgTypeFlagWithEvent: - optional.event = int.from_bytes(res[offset:8], "big", signed=True) - offset += 4 - if optional.event == EVENT_NONE: - return response - # read connectionId - elif optional.event == EVENT_ConnectionStarted: - optional.connectionId, offset = self.read_res_content(res, offset) - elif optional.event == EVENT_ConnectionFailed: - optional.response_meta_json, offset = self.read_res_content( - res, offset - ) - elif ( - optional.event == EVENT_SessionStarted - or optional.event == EVENT_SessionFailed - or optional.event == EVENT_SessionFinished - ): - optional.sessionId, offset = self.read_res_content(res, offset) - optional.response_meta_json, offset = self.read_res_content( - res, offset - ) - else: - optional.sessionId, offset = self.read_res_content(res, offset) - response.payload, offset = self.read_res_payload(res, offset) - - elif header.message_type == ERROR_INFORMATION: - optional.errorCode = int.from_bytes( - res[offset : offset + 4], "big", signed=True - ) - offset += 4 - response.payload, offset = self.read_res_payload(res, offset) - return response - - async def start_connection(self): - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - ).as_bytes() - optional = Optional(event=EVENT_Start_Connection).as_bytes() - payload = str.encode("{}") - return await self.send_event(header, optional, payload) - - def print_response(self, res, tag_msg: str): - logger.bind(tag=TAG).info(f"===>{tag_msg} header:{res.header.__dict__}") - logger.bind(tag=TAG).info(f"===>{tag_msg} optional:{res.optional.__dict__}") - - def get_payload_bytes( - self, - uid="1234", - event=EVENT_NONE, - text="", - speaker="", - audio_format="pcm", - audio_sample_rate=16000, - ): - return str.encode( - json.dumps( - { - "user": {"uid": uid}, - "event": event, - "namespace": "BidirectionalTTS", - "req_params": { - "text": text, - "speaker": speaker, - "audio_params": { - "format": audio_format, - "sample_rate": audio_sample_rate, - }, - }, - } - ) - ) - - async def finish_connection(self): - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional(event=EVENT_FinishConnection).as_bytes() - payload = str.encode("{}") - await self.send_event(header, optional, payload) - return - - async def start_session(self, session_id): - self.stop_event_response.clear() - 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): - self.stop_event_response.set() - 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 def reset(self): - # 关闭之前的对话 - if self.start_connection_flag: - await self.finish_connection() - self.start_connection_flag = False - await self.start_connection() - self.start_connection_flag = True - await super().reset() - - async def close(self): - super().close() - """资源清理方法""" - await self.finish_connection() - await self.ws.close() - - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - # 发送文本 - await self.send_text(self.speaker, text, u_id) - return - - def _start_monitor_tts_response_thread(self): - # 初始化链接 - asyncio.run_coroutine_threadsafe( - self._start_monitor_tts_response(), loop=self.loop - ) - - async def _start_monitor_tts_response(self): - chunk_total = b"" - while not self.stop_event.is_set(): - try: - msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop - res = self.parser_response(msg) - self.print_response(res, "send_text res:") - - if ( - res.optional.event == EVENT_TTSResponse - and res.header.message_type == AUDIO_ONLY_RESPONSE - ): - logger.bind(tag=TAG).info(f"推送数据到队列里面～～") - opus_datas = self.wav_to_opus_data_audio_raw(res.payload) - logger.bind(tag=TAG).info(f"推送数据到队列里面帧数～～{len(opus_datas)}") - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text="", - sentence_type=None, - duration=0, - ) - ) - elif 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}") - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=[], - tts_finish_text=self.tts_text, - sentence_type=SentenceType.SENTENCE_START, - ) - ) - elif res.optional.event == EVENT_TTSSentenceEnd: - logger.bind(tag=TAG).info(f"句子结束～～{self.tts_text}") - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=[], - tts_finish_text=self.tts_text, - sentence_type=SentenceType.SENTENCE_END, - ) - ) - elif res.optional.event == EVENT_SessionFinished: - logger.bind(tag=TAG).info(f"会话结束～～,最后一句补零") - opus_datas = self.wav_to_opus_data_audio_raw(b"", is_end=True) - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text="", - sentence_type=None, - duration=0, - ) - ) - self.tts_audio_queue.put( - TTSMessageDTO( - u_id=self.u_id, - msg_type=MsgType.STOP_TTS_RESPONSE, - content=[], - tts_finish_text=self.tts_text, - sentence_type=SentenceType.SENTENCE_END, - ) - ) - else: - continue - except websockets.ConnectionClosed: - break # 连接关闭时退出监听 - except Exception as e: - logger.bind(tag=TAG).error(f"Error in _start_monitor_tts_response: {e}") - traceback.print_exc() - continue diff --git a/main/xiaozhi-server/core/providers/tts/minimax.py b/main/xiaozhi-server/core/providers/tts/minimax.py index 06d87da2..dd406b64 100644 --- a/main/xiaozhi-server/core/providers/tts/minimax.py +++ b/main/xiaozhi-server/core/providers/tts/minimax.py @@ -2,11 +2,9 @@ import os import uuid import json import requests -from core.utils.util import parse_string_to_list from datetime import datetime -from pydub import AudioSegment from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType +from core.utils.util import parse_string_to_list class TTSProvider(TTSProviderBase): @@ -61,8 +59,7 @@ class TTSProvider(TTSProviderBase): f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): request_json = { "model": self.model, "text": text, @@ -83,7 +80,7 @@ class TTSProvider(TTSProviderBase): # 检查返回请求数据的status_code是否为0 if resp.json()["base_resp"]["status_code"] == 0: data = resp.json()["data"]["audio"] - file_to_save = open(tmp_file, "wb") + file_to_save = open(output_file, "wb") file_to_save.write(bytes.fromhex(data)) else: raise Exception( @@ -91,20 +88,3 @@ class TTSProvider(TTSProviderBase): ) except Exception as e: raise Exception(f"{__name__} error: {e}") - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="mp3") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass diff --git a/main/xiaozhi-server/core/providers/tts/openai.py b/main/xiaozhi-server/core/providers/tts/openai.py index 64203508..a7c4e30d 100644 --- a/main/xiaozhi-server/core/providers/tts/openai.py +++ b/main/xiaozhi-server/core/providers/tts/openai.py @@ -2,12 +2,12 @@ import os import uuid import requests from datetime import datetime - -from pydub import AudioSegment - -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase +from config.logger import setup_logging + +TAG = __name__ +logger = setup_logging() class TTSProvider(TTSProviderBase): @@ -35,8 +35,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", @@ -50,26 +49,9 @@ class TTSProvider(TTSProviderBase): } response = requests.post(self.api_url, json=data, headers=headers) if response.status_code == 200: - with open(tmp_file, "wb") as audio_file: + with open(output_file, "wb") as audio_file: audio_file.write(response.content) else: raise Exception( f"OpenAI TTS请求失败: {response.status_code} - {response.text}" ) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass diff --git a/main/xiaozhi-server/core/providers/tts/siliconflow.py b/main/xiaozhi-server/core/providers/tts/siliconflow.py index 72f59e66..9e30f721 100644 --- a/main/xiaozhi-server/core/providers/tts/siliconflow.py +++ b/main/xiaozhi-server/core/providers/tts/siliconflow.py @@ -2,11 +2,7 @@ import os import uuid import requests from datetime import datetime - -from pydub import AudioSegment - from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType class TTSProvider(TTSProviderBase): @@ -32,8 +28,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): request_json = { "model": self.model, "input": text, @@ -44,26 +39,12 @@ class TTSProvider(TTSProviderBase): "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", } - response = requests.request( - "POST", self.api_url, json=request_json, headers=headers - ) - data = response.content - file_to_save = open(tmp_file, "wb") - file_to_save.write(data) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="wav") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass + response = requests.request( + "POST", self.api_url, json=request_json, headers=headers + ) + data = response.content + file_to_save = open(output_file, "wb") + file_to_save.write(data) + except Exception as e: + raise Exception(f"{__name__} error: {e}") \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/ttson.py b/main/xiaozhi-server/core/providers/tts/ttson.py index cc458574..f3a0fbdc 100644 --- a/main/xiaozhi-server/core/providers/tts/ttson.py +++ b/main/xiaozhi-server/core/providers/tts/ttson.py @@ -4,11 +4,11 @@ import json import requests import shutil from datetime import datetime - -from pydub import AudioSegment - from core.providers.tts.base import TTSProviderBase -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType +from config.logger import setup_logging + +TAG = __name__ +logger = setup_logging() class TTSProvider(TTSProviderBase): @@ -39,8 +39,7 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - tmp_file = self.generate_filename() + async def text_to_speak(self, text, output_file): url = f"{self.url}{self.token}" result = "firefly" payload = json.dumps( @@ -59,7 +58,8 @@ class TTSProvider(TTSProviderBase): resp = requests.request("POST", url, data=payload) if resp.status_code != 200: - return + logger.bind(tag=TAG).error(f"TTSON 请求失败: {resp.text}") + raise Exception(f"{__name__}: TTS请求失败") resp_json = resp.json() try: result = ( @@ -71,29 +71,15 @@ class TTSProvider(TTSProviderBase): + "&voice_audio_path=" + resp_json["voice_path"] ) + + audio_content = requests.get(result) + with open(output_file, "wb") as f: + f.write(audio_content.content) + return True + voice_path = resp_json.get("voice_path") + des_path = output_file + shutil.move(voice_path, des_path) + except Exception as e: print("error:", e) - - audio_content = requests.get(result) - with open(tmp_file, "wb") as f: - f.write(audio_content.content) - # 使用 pydub 读取临时文件 - audio = AudioSegment.from_file(tmp_file, format="mp3") - audio = audio.set_channels(1).set_frame_rate(16000) - opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data) - yield TTSMessageDTO( - u_id=u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_datas, - tts_finish_text=text, - sentence_type=SentenceType.SENTENCE_START, - ) - # 用完后删除临时文件 - try: - os.remove(tmp_file) - except FileNotFoundError: - # 若文件不存在，忽略该异常 - pass - voice_path = resp_json.get("voice_path") - des_path = tmp_file - shutil.move(voice_path, des_path) + raise Exception(f"{__name__}: TTS请求失败") \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/textUtils.py b/main/xiaozhi-server/core/utils/textUtils.py deleted file mode 100644 index 7fadd3ff..00000000 --- a/main/xiaozhi-server/core/utils/textUtils.py +++ /dev/null @@ -1,34 +0,0 @@ -def get_string_no_punctuation_or_emoji(s): - """去除字符串首尾的空格、标点符号和表情符号""" - chars = list(s) - # 处理开头的字符 - start = 0 - while start < len(chars) and is_punctuation_or_emoji(chars[start]): - start += 1 - # 处理结尾的字符 - end = len(chars) - 1 - while end >= start and is_punctuation_or_emoji(chars[end]): - end -= 1 - return ''.join(chars[start:end + 1]) - -def is_punctuation_or_emoji(char): - """检查字符是否为空格、指定标点或表情符号""" - # 定义需要去除的中英文标点（包括全角/半角） - punctuation_set = { - '，', ',', # 中文逗号 + 英文逗号 - '。', '.', # 中文句号 + 英文句号 - '！', '!', # 中文感叹号 + 英文感叹号 - '-', '－', # 英文连字符 + 中文全角横线 - '、' # 中文顿号 - } - if char.isspace() or char in punctuation_set: - return True - # 检查表情符号（保留原有逻辑） - code_point = ord(char) - emoji_ranges = [ - (0x1F600, 0x1F64F), (0x1F300, 0x1F5FF), - (0x1F680, 0x1F6FF), (0x1F900, 0x1F9FF), - (0x1FA70, 0x1FAFF), (0x2600, 0x26FF), - (0x2700, 0x27BF) - ] - return any(start <= code_point <= end for start, end in emoji_ranges) \ No newline at end of file diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 3f9e2fd7..578e1a01 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -2,6 +2,7 @@ import asyncio import websockets from config.logger import setup_logging from core.connection import ConnectionHandler +from core.providers.tts.default import DefaultTTS from core.utils.util import initialize_modules, check_vad_update, check_asr_update from config.config_loader import get_config_from_api @@ -29,6 +30,10 @@ class WebSocketServer: self._llm = modules["llm"] if "llm" in modules else None self._intent = modules["intent"] if "intent" in modules else None self._memory = modules["memory"] if "memory" in modules else None + + if self._tts is None: + self._tts = DefaultTTS(self.config, delete_audio_file=True) + self.active_connections = set() async def start(self): diff --git a/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py b/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py index fa357ff5..f89f30b0 100644 --- a/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py +++ b/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py @@ -15,15 +15,26 @@ async def _get_device_status(conn, device_name, device_type, property_name): return status -async def _set_device_property(conn, device_name, device_type, method_name, property_name, new_value=None, action=None, step=10): +async def _set_device_property( + conn, + device_name, + device_type, + method_name, + property_name, + new_value=None, + action=None, + step=10, +): """设置设备属性""" - current_value = await _get_device_status(conn, device_name, device_type, property_name) + current_value = await _get_device_status( + conn, device_name, device_type, property_name + ) - if action == 'raise': + if action == "raise": current_value += step - elif action == 'lower': + elif action == "lower": current_value -= step - elif action == 'set': + elif action == "set": if new_value is None: raise Exception(f"缺少{property_name}参数") current_value = new_value @@ -37,8 +48,7 @@ async def _set_device_property(conn, device_name, device_type, method_name, prop def _handle_device_action(conn, func, success_message, error_message, *args, **kwargs): """处理设备操作的通用函数""" - future = asyncio.run_coroutine_threadsafe( - func(conn, *args, **kwargs), conn.loop) + future = asyncio.run_coroutine_threadsafe(func(conn, *args, **kwargs), conn.loop) try: result = future.result() logger.bind(tag=TAG).info(f"{success_message}: {result}") @@ -75,26 +85,41 @@ handle_device_function_desc = { "device_type": { "type": "string", "description": "设备类型，**严格限定为Speaker（音量）或Screen（亮度）**，其他设备类型禁止调用此函数", - "enum": ["Speaker", "Screen"] - + "enum": ["Speaker", "Screen"], }, "action": { "type": "string", - "description": "动作名称，可选值：get(获取),set(设置),raise(提高),lower(降低)" + "description": "动作名称，可选值：get(获取),set(设置),raise(提高),lower(降低)", }, "value": { "type": "integer", - "description": "值大小，可选值：0-100之间的整数" - } + "description": "值大小，可选值：0-100之间的整数", + }, }, - "required": ["device_type", "action"] - } - } + "required": ["device_type", "action"], + }, + }, } -@register_function('handle_speaker_volume_or_screen_brightness', handle_device_function_desc, ToolType.IOT_CTL) -def handle_speaker_volume_or_screen_brightness(conn, device_type: str, action: str, value: int = None): +@register_function( + "handle_speaker_volume_or_screen_brightness", + handle_device_function_desc, + ToolType.IOT_CTL, +) +def handle_speaker_volume_or_screen_brightness( + conn, device_type: str, action: str, value: int = None +): + # 检查value是否为中文值 + if ( + value is not None + and isinstance(value, str) + and any("\u4e00" <= char <= "\u9fff" for char in str(value)) + ): + raise Exception( + f"请直接告诉我要将{'音量' if device_type=='Speaker' else '亮度'}调整成多少" + ) + if device_type == "Speaker": method_name, property_name, device_name = "SetVolume", "volume", "音量" elif device_type == "Screen": @@ -108,13 +133,25 @@ def handle_speaker_volume_or_screen_brightness(conn, device_type: str, action: s if action == "get": # get return _handle_device_action( - conn, _get_device_status, f"当前{device_name}", f"获取{device_name}失败", - device_name=device_name, device_type=device_type, property_name=property_name, + conn, + _get_device_status, + f"当前{device_name}", + f"获取{device_name}失败", + device_name=device_name, + device_type=device_type, + property_name=property_name, ) else: # set, raise, lower return _handle_device_action( - conn, _set_device_property, f"{device_name}已调整到", f"{device_name}调整失败", - device_name=device_name, device_type=device_type, method_name=method_name, - property_name=property_name, new_value=value, action=action + conn, + _set_device_property, + f"{device_name}已调整到", + f"{device_name}调整失败", + device_name=device_name, + device_type=device_type, + method_name=method_name, + property_name=property_name, + new_value=value, + action=action, ) diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 4c180ce3..0446e4b2 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -7,8 +7,6 @@ import asyncio import difflib import traceback from pathlib import Path - -from core.providers.tts.dto.dto import TTSMessageDTO, MsgType from core.utils import p3 from core.handle.sendAudioHandle import send_stt_message from plugins_func.register import register_function, ToolType, ActionResponse, Action @@ -38,7 +36,7 @@ play_music_function_desc = { @register_function("play_music", play_music_function_desc, ToolType.SYSTEM_CTL) -def play_music(conn, song_name=None): +def play_music(conn, song_name: str): try: music_intent = ( f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐" @@ -122,8 +120,8 @@ def get_music_files(music_dir, music_ext): def initialize_music_handler(conn): global MUSIC_CACHE if MUSIC_CACHE == {}: - if "music" in conn.config: - MUSIC_CACHE["music_config"] = conn.config["music"] + if "play_music" in conn.config["plugins"]: + MUSIC_CACHE["music_config"] = conn.config["plugins"]["play_music"] MUSIC_CACHE["music_dir"] = os.path.abspath( MUSIC_CACHE["music_config"].get("music_dir", "./music") # 默认路径修改 ) @@ -219,31 +217,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_one_sentence(conn, text) + conn.recode_first_last_text(text, 0) + future = conn.executor.submit(conn.speak_and_play, None, text, 0) + conn.tts.tts_queue.put((future, 0)) - if music_path.endswith(".p3"): - opus_packets, _ = p3.decode_opus_from_file(music_path) - else: - opus_packets, _ = conn.tts.audio_to_opus_data(music_path) - conn.tts.tts_audio_queue.put( - TTSMessageDTO( - u_id=conn.u_id, - msg_type=MsgType.TTS_TEXT_RESPONSE, - content=opus_packets, - tts_finish_text="", - sentence_type=None, - duration=0, - ) - ) - conn.tts.tts_audio_queue.put( - TTSMessageDTO( - u_id=conn.u_id, - msg_type=MsgType.STOP_TTS_RESPONSE, - content=[], - tts_finish_text="", - sentence_type=None, - ) - ) + conn.recode_first_last_text(text, 1) + future = conn.executor.submit(conn.speak_and_play, music_path, None, 1) + conn.tts.tts_queue.put((future, 1)) + conn.llm_finish_task = True except Exception as e: conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")