diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java index b3c036ee..abac385b 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java +++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java @@ -227,7 +227,7 @@ public interface Constant { /** * 版本号 */ - public static final String VERSION = "0.4.4"; + public static final String VERSION = "0.5.1"; /** * 无效固件URL diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 662f731e..38dc10fa 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -453,14 +453,14 @@ TTS: volume_ratio: 1.0 pitch_ratio: 1.0 #火山tts，支持双向流式tts - HuoshanTTS: - type: huoshan + HuoshanDoubleStreamTTS: + type: huoshan_double_stream # 如果是机智云 wss://bytedance.gizwitsapi.com/api/v3/tts/bidirection # 机智云不需要天填 appid ws_url: wss://openspeech.bytedance.com/api/v3/tts/bidirection appid: 你的火山引擎语音合成服务appid access_token: 你的火山引擎语音合成服务access_token - # 产看https://www.volcengine.com/docs/6561/1329505，volc.service_type.10029，volc.megatts.default + # 资源信息ID，https://www.volcengine.com/docs/6561/1329505，volc.service_type.10029，大模型语音合成及混音 resource_id: volc.service_type.10029 speaker: zh_female_meilinvyou_moon_bigtts CosyVoiceSiliconflow: diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py index 5588e00f..5bbda5a1 100644 --- a/main/xiaozhi-server/config/logger.py +++ b/main/xiaozhi-server/config/logger.py @@ -4,7 +4,7 @@ from loguru import logger from config.config_loader import load_config from config.settings import check_config_file -SERVER_VERSION = "0.4.4" +SERVER_VERSION = "0.5.1" def get_module_abbreviation(module_name, module_dict): diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 80e3b18f..93532734 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -8,25 +8,24 @@ import time import queue import asyncio import traceback - import threading 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.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType +from core.providers.tts.default import DefaultTTS from core.handle.textHandle import handleTextMessage from core.utils.util import ( - get_string_no_punctuation_or_emoji, extract_json_from_string, initialize_modules, check_vad_update, check_asr_update, filter_sensitive_info, + initialize_tts, ) -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 +34,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__ @@ -53,7 +52,6 @@ class ConnectionHandler: _vad, _asr, _llm, - _tts, _memory, _intent, server=None, @@ -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,10 @@ class ConnectionHandler: # 依赖的组件 self.vad = None self.asr = None + self.tts = None self._asr = _asr self._vad = _vad self.llm = _llm - self.tts = _tts self.memory = _memory self.intent = _intent @@ -125,13 +115,11 @@ class ConnectionHandler: self.asr_server_receive = True # llm相关变量 - self.llm_finish_task = False + self.llm_finish_task = True self.dialogue = Dialogue() # tts相关变量 - self.tts_first_text_index = -1 - self.tts_last_text_index = -1 - self.tts_duration = 0 + self.sentence_id = None # iot相关变量 self.iot_descriptors = {} @@ -203,15 +191,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 +311,11 @@ class ConnectionHandler: self.vad = self._vad if self.asr is None: self.asr = self._asr + if self.tts is None: + self.tts = self._initialize_tts() + # 使用事件循环运行异步方法 + asyncio.run_coroutine_threadsafe(self.tts.open_audio_channels(self), self.loop) + """加载记忆""" self._initialize_memory() """加载意图识别""" @@ -352,6 +336,17 @@ class ConnectionHandler: self.report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") + def _initialize_tts(self): + """初始化TTS""" + tts = None + if not self.need_bind: + tts = initialize_tts(self.config) + + if tts is None: + tts = DefaultTTS(self.config, delete_audio_file=True) + + return tts + def _initialize_private_config(self): """如果是从配置文件获取，则进行二次实例化""" if not self.read_config_from_api: @@ -512,80 +507,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 - + def chat(self, query, tool_call=False): + self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") 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""" 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 = [] try: - start_time = time.time() - # 使用带记忆的对话 memory_str = None if self.memory is not None: @@ -594,88 +529,79 @@ class ConnectionHandler: ) memory_str = future.result() - # self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}") + uuid_str = str(uuid.uuid4()).replace("-", "") + self.sentence_id = uuid_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 - self.llm_finish_task = False - text_index = 0 - # 处理流式响应 tool_call_flag = False function_name = None function_id = None function_arguments = "" content_arguments = "" - uuid_str = str(uuid.uuid4()).replace("-", "") - self.u_id = uuid_str + text_index = 0 + 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) - if self.client_abort: break - end_time = time.time() - - 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="", + sentence_id=self.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, ) ) - 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, + sentence_id=self.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=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="" - ) - ) - # 处理function call if tool_call_flag: bHasError = False @@ -718,14 +644,21 @@ class ConnectionHandler: result = self.func_handler.handle_llm_function_call( self, function_call_data ) - self._handle_function_result(result, function_call_data, text_index + 1) + self._handle_function_result(result, function_call_data) # 存储对话内容 if len(response_message) > 0: self.dialogue.put( Message(role="assistant", content="".join(response_message)) ) - + if text_index > 0: + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) + ) self.llm_finish_task = True self.logger.bind(tag=TAG).debug( json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False) @@ -775,11 +708,10 @@ class ConnectionHandler: return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="") - def _handle_function_result(self, result, function_call_data, text_index): + def _handle_function_result(self, result, function_call_data): if result.action == Action.RESPONSE: # 直接回复前端 text = result.response - self.recode_first_last_text(text, text_index) - self.tts.tts_one_sentence(self, text) + self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) self.dialogue.put(Message(role="assistant", content=text)) elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复 text = result.result @@ -813,35 +745,13 @@ 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 + self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) 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): """聊天记录上报工作线程""" @@ -872,14 +782,6 @@ class ConnectionHandler: def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") self.asr_server_receive = True - self.tts_last_text_index = -1 - self.tts_first_text_index = -1 - - def recode_first_last_text(self, text, text_index=0): - if self.tts_first_text_index == -1: - self.logger.bind(tag=TAG).info(f"大模型说出第一句话: {text}") - self.tts_first_text_index = text_index - self.tts_last_text_index = text_index async def close(self, ws=None): """资源清理方法""" @@ -905,7 +807,6 @@ class ConnectionHandler: await ws.close() elif self.websocket: await self.websocket.close() - await self.tts.close() # 最后关闭线程池（避免阻塞） if self.executor: @@ -917,11 +818,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_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}" ) # 使用非阻塞方式清空队列 - for q in [self.tts_queue, self.audio_play_queue]: + for q in [self.tts.tts_text_queue, self.tts.tts_audio_queue]: if not q: continue while True: @@ -931,7 +832,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_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_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..8a0b2ab9 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)) @@ -43,9 +44,6 @@ async def checkWakeupWords(conn, text): _, filtered_text = remove_punctuation_and_length(text) if filtered_text in conn.config.get("wakeup_words"): await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - conn.llm_finish_task = True file = getWakeupWordFile(WAKEUP_CONFIG["file_name"]) if file is None: @@ -55,7 +53,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.tts_audio_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..f3dd5cc5 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -2,7 +2,9 @@ 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.providers.tts.dto.dto import ContentType from core.utils.dialogue import Message from plugins_func.register import Action from loguru import logger @@ -15,10 +17,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 +110,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 +132,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 +143,5 @@ 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) - conn.dialogue.put(Message(role="assistant", content=text)) +def speak_txt(conn, text): + conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 9c693751..14e1bc5a 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -5,6 +5,7 @@ from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit from core.handle.reportHandle import enqueue_asr_report +from core.handle.sendAudioHandle import SentenceType from core.utils.util import audio_to_data TAG = __name__ @@ -39,7 +40,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 +79,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): @@ -112,12 +111,9 @@ async def no_voice_close_connect(conn): async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - 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.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) conn.close_after_chat = True @@ -132,14 +128,11 @@ async def check_bind_device(conn): text = f"请登录控制面板，输入{conn.bind_code}，绑定设备。" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 6 - conn.llm_finish_task = True # 播放提示音 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.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 @@ -147,16 +140,14 @@ 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.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None)) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) else: text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - 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.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 5b73ed18..5694c618 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,70 +1,110 @@ -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.providers.tts.dto.dto import SentenceType +from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion +from loguru import logger 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, sentenceType, audios, text): # 发送句子开始消息 - 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, + } + ) + ) + pre_buffer = False + if conn.tts.tts_audio_first_sentence and text is not None: + conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") + conn.tts.tts_audio_first_sentence = False + pre_buffer = True + await send_tts_message(conn, "sentence_start", text) + + await sendAudio(conn, audios, pre_buffer) + + await send_tts_message(conn, "sentence_end", text) + + # 发送结束消息（如果是最后一个文本） + if conn.llm_finish_task and sentenceType == SentenceType.LAST: + await send_tts_message(conn, "stop", None) + if conn.close_after_chat: + await conn.close() + + +# 播放音频 +async def sendAudio(conn, audios, pre_buffer=True): + if audios is None or len(audios) == 0: + return # 流控参数优化 - 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 +113,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 +136,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/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index fee62364..74fb4091 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -155,12 +155,6 @@ class ASRProvider(ASRProviderBase): # 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}", - ) - def _construct_request_url(self) -> str: """构造请求URL，包含参数""" request = f"{self.base_url}?appkey={self.app_key}" @@ -239,7 +233,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..e7f31d18 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun.py @@ -1,4 +1,3 @@ -import os import uuid import json import hmac @@ -8,16 +7,9 @@ import requests from datetime import datetime from core.providers.tts.base import TTSProviderBase -from pydub import AudioSegment - 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: @@ -155,15 +147,9 @@ class TTSProvider(TTSProviderBase): # 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}", - ) - - 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, @@ -177,7 +163,7 @@ class TTSProvider(TTSProviderBase): "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 @@ -188,31 +174,13 @@ class TTSProvider(TTSProviderBase): 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: + 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 - 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..b68744fe 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -1,24 +1,20 @@ -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 os import queue +import uuid +import asyncio +import threading +from core.utils import p3 +from datetime import datetime +from core.utils import textUtils +from abc import ABC, abstractmethod +from config.logger import setup_logging +from core.utils.util import audio_to_data +from core.utils.tts import MarkdownCleaner +from core.handle.sendAudioHandle import sendAudioMessage +from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType -from core.providers.tts.dto.dto import MsgType, TTSMessageDTO, SentenceType + +import traceback TAG = __name__ logger = setup_logging() @@ -26,274 +22,86 @@ 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_audio_first_sentence = True self.tts_text_buff = [] self.punctuations = ( "。", - "？", - "！", - "；", - "：", ".", + "？", "?", + "！", "!", + "；", ";", - ":", - " ", + "：", + ) + self.first_sentence_punctuations = ( + "，", + "~", + "、", ",", - "，", + "。", + ".", + "？", + "?", + "！", + "!", + "；", + ";", + "：", ) - self.tts_request = False self.tts_stop_request = False self.processed_chars = 0 - self.stream = False - self.last_to_opus_raw = b"" + self.is_first_sentence = True - # 启动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 + def generate_filename(self, extension=".wav"): + return os.path.join( + self.output_file, + f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - tts_priority.start() - async def close(self): - self.stop_event + 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 - 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: + 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 - 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() - @abstractmethod - def generate_filename(self): + async def text_to_speak(self, text, output_file): pass - @abstractmethod - async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False): - pass - - async def text_to_speak_stream(self, text, queue: queue.Queue, text_index=0): - raise Exception("该TTS还没有实现stream模式") - def audio_to_pcm_data(self, audio_file_path): """音频文件转换为PCM编码""" return audio_to_data(audio_file_path, is_opus=False) @@ -302,16 +110,212 @@ 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 tts_one_sentence( + self, + conn, + content_type, + content_detail=None, + content_file=None, + sentence_id=None, + ): + """发送一句话""" + if not sentence_id: + if conn.sentence_id: + sentence_id = conn.sentence_id + else: + sentence_id = str(uuid.uuid4()).replace("-", "") + conn.sentence_id = sentence_id + self.tts_text_queue.put( + TTSMessageDTO( + sentence_id=sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) + self.tts_text_queue.put( + TTSMessageDTO( + sentence_id=sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=content_type, + content_detail=content_detail, + content_file=content_file, + ) + ) + self.tts_text_queue.put( + TTSMessageDTO( + sentence_id=sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) + ) - 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 + async def open_audio_channels(self, conn): + self.conn = conn + self.tts_timeout = conn.config.get("tts_timeout", 10) + # tts 消化线程 + self.tts_priority_thread = threading.Thread( + target=self.tts_text_priority_thread, daemon=True + ) + self.tts_priority_thread.start() + + # 音频播放 消化线程 + self.audio_play_priority_thread = threading.Thread( + target=self._audio_play_priority_thread, daemon=True + ) + self.audio_play_priority_thread.start() + + # 这里默认是非流式的处理方式 + # 流式处理方式请在子类中重写 + def tts_text_priority_thread(self): + while not self.conn.stop_event.is_set(): + try: + message = self.tts_text_queue.get(timeout=1) + if message.sentence_type == SentenceType.FIRST: + # 初始化参数 + self.tts_stop_request = False + self.processed_chars = 0 + self.tts_text_buff = [] + self.is_first_sentence = True + self.tts_audio_first_sentence = True + elif ContentType.TEXT == message.content_type: + self.tts_text_buff.append(message.content_detail) + segment_text = self._get_segment_text() + if segment_text: + tts_file = self.to_tts(segment_text) + if tts_file: + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, segment_text) + ) + elif ContentType.FILE == message.content_type: + self._process_remaining_text() + tts_file = message.content_file + if tts_file and os.path.exists(tts_file): + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, message.content_detail) + ) + + if message.sentence_type == SentenceType.LAST: + self._process_remaining_text() + self.tts_audio_queue.put( + (message.sentence_type, [], message.content_detail) + ) + + except queue.Empty: + continue + except Exception as e: + logger.bind(tag=TAG).error( + f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) + + def _audio_play_priority_thread(self): + while not self.conn.stop_event.is_set(): + text = None + try: + try: + sentence_type, audio_datas, text = self.tts_audio_queue.get( + timeout=1 + ) + except queue.Empty: + if self.conn.stop_event.is_set(): + break + continue + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self.conn.loop, + ) + future.result() + except Exception as e: + logger.bind(tag=TAG).error( + f"audio_play_priority priority_thread: {text} {e}" + ) + + async def start_session(self, session_id): + pass + + async def finish_session(self, session_id): + pass + + async def close(self): + """资源清理方法""" + if hasattr(self, "ws") and self.ws: + await self.ws.close() + + def _get_segment_text(self): + # 合并当前全部文本并处理未分割部分 + full_text = "".join(self.tts_text_buff) + current_text = full_text[self.processed_chars :] # 从未处理的位置开始 + last_punct_pos = -1 + + # 根据是否是第一句话选择不同的标点符号集合 + punctuations_to_use = ( + self.first_sentence_punctuations + if self.is_first_sentence + else self.punctuations + ) + + for punct in punctuations_to_use: + 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) # 更新已处理字符位置 + + # 如果是第一句话，在找到第一个逗号后，将标志设置为False + if self.is_first_sentence: + self.is_first_sentence = False + + return segment_text + elif self.tts_stop_request and current_text: + segment_text = current_text + self.is_first_sentence = True # 重置标志 + return segment_text + else: + return None + + def _process_audio_file(self, tts_file): + """处理音频文件并转换为指定格式 + + Args: + tts_file: 音频文件路径 + content_detail: 内容详情 + + Returns: + tuple: (sentence_type, audio_datas, content_detail) + """ + audio_datas = [] + 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) + return audio_datas + + def _process_remaining_text(self): + """处理剩余的文本并生成语音 + + Returns: + bool: 是否成功处理了文本 + """ + full_text = "".join(self.tts_text_buff) + remaining_text = full_text[self.processed_chars :] + if remaining_text: + segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) + if segment_text: + tts_file = self.to_tts(segment_text) + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, audio_datas, segment_text) + ) + self.processed_chars += len(full_text) + return True + return False diff --git a/main/xiaozhi-server/core/providers/tts/cozecn.py b/main/xiaozhi-server/core/providers/tts/cozecn.py index 61221e85..c3019b67 100644 --- a/main/xiaozhi-server/core/providers/tts/cozecn.py +++ b/main/xiaozhi-server/core/providers/tts/cozecn.py @@ -1,14 +1,5 @@ -import os -import uuid -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,19 +11,12 @@ 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" self.api_url = f"https://{self.host}/v1/audio/speech" - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +27,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}") 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..7818d288 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/default.py @@ -0,0 +1,23 @@ +import os +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..78058c31 100644 --- a/main/xiaozhi-server/core/providers/tts/doubao.py +++ b/main/xiaozhi-server/core/providers/tts/doubao.py @@ -1,13 +1,7 @@ -import os import uuid 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 @@ -45,14 +39,7 @@ class TTSProvider(TTSProviderBase): self.header = {"Authorization": f"{self.authorization}{self.access_token}"} check_model_key("TTS", self.access_token) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +70,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 +78,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 index 706e7d35..a57a27de 100644 --- a/main/xiaozhi-server/core/providers/tts/dto/dto.py +++ b/main/xiaozhi-server/core/providers/tts/dto/dto.py @@ -1,42 +1,36 @@ 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" +from typing import Union, Optional class SentenceType(Enum): - # 句子开始 - SENTENCE_START = "SENTENCE_START" - # 句子结束 - SENTENCE_END = "SENTENCE_END" + # 说话阶段 + FIRST = "FIRST" # 首句话 + MIDDLE = "MIDDLE" # 说话中 + LAST = "LAST" # 最后一句 + + +class ContentType(Enum): + # 内容类型 + TEXT = "TEXT" # 文本内容 + FILE = "FILE" # 文件内容 + ACTION = "ACTION" # 动作内容 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 + def __init__( + self, + sentence_id: str, + # 说话阶段 + sentence_type: SentenceType, + # 内容类型 + content_type: ContentType, + # 内容详情，一般是需要转换的文本或者音频的歌词 + content_detail: Optional[str] = None, + # 如果内容类型为文件，则需要传入文件路径 + content_file: Optional[str] = None, + ): + self.sentence_id = sentence_id 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})" + self.content_type = content_type + self.content_detail = content_detail + self.content_file = content_file 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..23b20e76 100644 --- a/main/xiaozhi-server/core/providers/tts/fishspeech.py +++ b/main/xiaozhi-server/core/providers/tts/fishspeech.py @@ -1,25 +1,11 @@ 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 @@ -144,127 +130,51 @@ class TTSProvider(TTSProviderBase): self.seed = int(config.get("seed")) if config.get("seed") else None self.api_url = config.get("api_url", "http://127.0.0.1:8080/v1/tts") - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", + 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] + + 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, + } + + pydantic_data = ServeTTSRequest(**data) + + 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", + }, ) - 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 + if response.status_code == 200: + audio_content = response.content - 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, - } + with open(output_file, "wb") as audio_file: + audio_file.write(audio_content) - # 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) - 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, - ) - - else: - print("请求失败:", response.status_code, response.text) - except Exception as e: - logger.bind(tag=TAG).error("tts发生错误") - traceback.print_exc() - raise e + 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..e76b3f22 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,7 @@ -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() @@ -71,14 +66,7 @@ class TTSProvider(TTSProviderBase): config.get("aux_ref_audio_paths") ) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +91,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..6b66efc3 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,7 @@ -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() @@ -38,14 +33,7 @@ class TTSProvider(TTSProviderBase): self.inp_refs = parse_string_to_list(config.get("inp_refs")) self.if_sr = str(config.get("if_sr", False)).lower() in ("true", "1", "yes") - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +52,9 @@ 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_double_stream.py similarity index 76% rename from main/xiaozhi-server/core/providers/tts/huoshan.py rename to main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index 734917cf..c26eec3a 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -1,21 +1,14 @@ 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 core.utils import opus_encoder_utils +import queue 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 +from core.providers.tts.dto.dto import SentenceType, ContentType TAG = __name__ logger = setup_logging() @@ -152,13 +145,21 @@ class TTSProvider(TTSProviderBase): 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 = "" + # 合成文字语音后，播放的音频文件列表 + self.tts_audio_files = [] + self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( + sample_rate=16000, channels=1, frame_size_ms=60 + ) - async def open_audio_channels(self): - await super().open_audio_channels() + ################################################################################### + # 火山双流式TTS重写父类的方法--开始 + ################################################################################### + + async def open_audio_channels(self, conn): + await super().open_audio_channels(conn) ws_header = { "X-Api-App-Key": self.appId, "X-Api-Access-Key": self.access_token, @@ -169,16 +170,96 @@ class TTSProvider(TTSProviderBase): self.ws_url, additional_headers=ws_header, max_size=1000000000 ) tts_priority = threading.Thread( - target=self._start_monitor_tts_response_thread(), daemon=True + 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}", + def tts_text_priority_thread(self): + while not self.conn.stop_event.is_set(): + try: + message = self.tts_text_queue.get(timeout=1) + logger.bind(tag=TAG).debug( + f"TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name}" + ) + if message.sentence_type == SentenceType.FIRST: + # 初始化参数 + future = asyncio.run_coroutine_threadsafe( + self.start_session(self.conn.sentence_id), loop=self.conn.loop + ) + future.result() + self.tts_audio_first_sentence = True + elif ContentType.TEXT == message.content_type: + if message.content_detail: + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(message.content_detail, None), + loop=self.conn.loop, + ) + future.result() + elif ContentType.FILE == message.content_type: + pass + + if message.sentence_type == SentenceType.LAST: + future = asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), loop=self.conn.loop + ) + future.result() + + except queue.Empty: + continue + except Exception as e: + logger.bind(tag=TAG).error( + f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) + + async def text_to_speak(self, text, _): + # 发送文本 + await self.send_text(self.speaker, text, self.conn.sentence_id) + return + + ################################################################################### + # 火山双流式TTS重写父类的方法--结束 + ################################################################################### + def _start_monitor_tts_response_thread(self): + # 初始化链接 + asyncio.run_coroutine_threadsafe( + self._start_monitor_tts_response(), loop=self.conn.loop ) + async def _start_monitor_tts_response(self): + while not self.conn.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_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((SentenceType.FIRST, [], self.tts_text)) + elif ( + res.optional.event == EVENT_TTSResponse + and res.header.message_type == AUDIO_ONLY_RESPONSE + ): + logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") + opus_datas = self.wav_to_opus_data_audio_raw(res.payload) + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～{len(opus_datas)}" + ) + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, None)) + elif res.optional.event == EVENT_TTSSentenceEnd: + logger.bind(tag=TAG).debug(f"句子结束～～{self.tts_text}") + elif res.optional.event == EVENT_SessionFinished: + logger.bind(tag=TAG).debug(f"会话结束～～") + self.tts_audio_queue.put((SentenceType.LAST, [], None)) + 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 + async def send_event( self, header: bytes, optional: bytes | None = None, payload: bytes = None ): @@ -282,8 +363,8 @@ class TTSProvider(TTSProviderBase): 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__}") + logger.bind(tag=TAG).debug(f"===>{tag_msg} header:{res.header.__dict__}") + logger.bind(tag=TAG).debug(f"===>{tag_msg} optional:{res.optional.__dict__}") def get_payload_bytes( self, @@ -324,7 +405,6 @@ class TTSProvider(TTSProviderBase): 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, @@ -333,9 +413,10 @@ class TTSProvider(TTSProviderBase): optional = Optional(event=EVENT_StartSession, sessionId=session_id).as_bytes() payload = self.get_payload_bytes(event=EVENT_StartSession, speaker=self.speaker) await self.send_event(header, optional, payload) + logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") async def finish_session(self, session_id): - self.stop_event_response.set() + logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, @@ -353,101 +434,12 @@ class TTSProvider(TTSProviderBase): 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 + 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 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..bdbc9174 100644 --- a/main/xiaozhi-server/core/providers/tts/openai.py +++ b/main/xiaozhi-server/core/providers/tts/openai.py @@ -1,13 +1,10 @@ -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): @@ -29,14 +26,7 @@ class TTSProvider(TTSProviderBase): self.output_file = config.get("output_dir", "tmp/") check_model_key("TTS", self.api_key) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +40,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..8f7fd641 100644 --- a/main/xiaozhi-server/core/providers/tts/siliconflow.py +++ b/main/xiaozhi-server/core/providers/tts/siliconflow.py @@ -1,12 +1,5 @@ -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): @@ -26,14 +19,7 @@ class TTSProvider(TTSProviderBase): self.host = "api.siliconflow.cn" self.api_url = f"https://{self.host}/v1/audio/speech" - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - 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 +30,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}") diff --git a/main/xiaozhi-server/core/providers/tts/tencent.py b/main/xiaozhi-server/core/providers/tts/tencent.py index f2501e2f..7a55328a 100644 --- a/main/xiaozhi-server/core/providers/tts/tencent.py +++ b/main/xiaozhi-server/core/providers/tts/tencent.py @@ -1,6 +1,5 @@ import hashlib import hmac -import os import time import uuid import json @@ -121,12 +120,6 @@ class TTSProvider(TTSProviderBase): msg = msg.encode("utf-8") return hmac.new(key, msg, hashlib.sha256).digest() - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): # 构建请求体 request_json = { diff --git a/main/xiaozhi-server/core/providers/tts/ttson.py b/main/xiaozhi-server/core/providers/tts/ttson.py index cc458574..b3c7797f 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请求失败") diff --git a/main/xiaozhi-server/core/opus/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py similarity index 99% rename from main/xiaozhi-server/core/opus/opus_encoder_utils.py rename to main/xiaozhi-server/core/utils/opus_encoder_utils.py index d9f6de2f..23d26c39 100644 --- a/main/xiaozhi-server/core/opus/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -3,7 +3,6 @@ Opus编码工具类 将PCM音频数据编码为Opus格式 """ -import array import logging import traceback diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 0b7e46a1..8f5ded4b 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -269,16 +269,7 @@ def initialize_modules( # 初始化TTS模块 if init_tts: select_tts_module = config["selected_module"]["TTS"] - tts_type = ( - select_tts_module - if "type" not in config["TTS"][select_tts_module] - else config["TTS"][select_tts_module]["type"] - ) - modules["tts"] = tts.create_instance( - tts_type, - config["TTS"][select_tts_module], - str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), - ) + modules["tts"] = initialize_tts(config) logger.bind(tag=TAG).info(f"初始化组件: tts成功 {select_tts_module}") # 初始化LLM模块 @@ -355,6 +346,21 @@ def initialize_modules( return modules +def initialize_tts(config): + select_tts_module = config["selected_module"]["TTS"] + tts_type = ( + select_tts_module + if "type" not in config["TTS"][select_tts_module] + else config["TTS"][select_tts_module]["type"] + ) + new_tts = tts.create_instance( + tts_type, + config["TTS"][select_tts_module], + str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), + ) + return new_tts + + def analyze_emotion(text): """ 分析文本情感并返回对应的emoji名称（支持中英文） @@ -882,7 +888,10 @@ def audio_to_data(audio_file_path, is_opus=True): # 获取原始PCM数据（16位小端） raw_data = audio.raw_data + return pcm_to_data(raw_data, is_opus), duration + +def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) @@ -910,7 +919,7 @@ def audio_to_data(audio_file_path, is_opus=True): datas.append(frame_data) - return datas, duration + return datas def check_vad_update(before_config, new_config): diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 3f9e2fd7..e60f0c87 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -19,16 +19,16 @@ class WebSocketServer: "VAD" in self.config["selected_module"], "ASR" in self.config["selected_module"], "LLM" in self.config["selected_module"], - "TTS" in self.config["selected_module"], + False, "Memory" in self.config["selected_module"], "Intent" in self.config["selected_module"], ) self._vad = modules["vad"] if "vad" in modules else None self._asr = modules["asr"] if "asr" in modules else None - self._tts = modules["tts"] if "tts" in modules else None 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 + self.active_connections = set() async def start(self): @@ -49,7 +49,6 @@ class WebSocketServer: self._vad, self._asr, self._llm, - self._tts, self._memory, self._intent, self, # 传入server实例 @@ -98,7 +97,7 @@ class WebSocketServer: update_vad, update_asr, "LLM" in new_config["selected_module"], - "TTS" in new_config["selected_module"], + False, "Memory" in new_config["selected_module"], "Intent" in new_config["selected_module"], ) @@ -108,8 +107,6 @@ class WebSocketServer: self._vad = modules["vad"] if "asr" in modules: self._asr = modules["asr"] - if "tts" in modules: - self._tts = modules["tts"] if "llm" in modules: self._llm = modules["llm"] if "intent" in modules: 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..6d184b6b 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -7,12 +7,11 @@ 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 from core.utils.dialogue import Message +from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType TAG = __name__ @@ -38,7 +37,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 +121,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,29 +218,34 @@ 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) - - 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( + conn.tts.tts_text_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, + sentence_id=conn.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, ) ) - conn.tts.tts_audio_queue.put( + conn.tts.tts_text_queue.put( TTSMessageDTO( - u_id=conn.u_id, - msg_type=MsgType.STOP_TTS_RESPONSE, - content=[], - tts_finish_text="", - sentence_type=None, + sentence_id=conn.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=text, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.FILE, + content_file=music_path, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, ) )