From 24526ad206a9fc45825c1be346703b9bcbb527e0 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Mon, 26 May 2025 02:20:38 +0800 Subject: [PATCH] =?UTF-8?q?update:=E6=97=A7=E9=9D=9E=E6=B5=81=E5=A4=B1?= =?UTF-8?q?=E5=85=BC=E5=AE=B9=E6=94=B9=E9=80=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 142 +++------ .../xiaozhi-server/core/handle/helloHandle.py | 7 +- .../core/handle/intentHandler.py | 10 +- .../core/handle/receiveAudioHandle.py | 17 +- .../core/handle/sendAudioHandle.py | 14 +- .../xiaozhi-server/core/providers/tts/base.py | 293 +++++++++++------- .../core/providers/tts/dto/dto.py | 36 +++ .../core/providers/tts/huoshan.py | 36 ++- main/xiaozhi-server/core/utils/textUtils.py | 34 ++ .../plugins_func/functions/play_music.py | 39 ++- 10 files changed, 355 insertions(+), 273 deletions(-) create mode 100644 main/xiaozhi-server/core/providers/tts/dto/dto.py create mode 100644 main/xiaozhi-server/core/utils/textUtils.py diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index e240ae37..66a8ed54 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -8,16 +8,15 @@ 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.utils.dialogue import Message, Dialogue +from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType 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, @@ -29,7 +28,6 @@ from core.handle.receiveAudioHandle import handleAudioMessage from core.handle.functionHandler import FunctionHandler from plugins_func.register import Action, ActionResponse from core.auth import AuthMiddleware, AuthenticationError -from core.providers.tts.base import TTSImplementationType from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api from config.manage_api_client import DeviceNotFoundException, DeviceBindException @@ -117,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_session_id = None + self.sentence_id = None # iot相关变量 self.iot_descriptors = {} @@ -501,6 +497,7 @@ class ConnectionHandler: self.dialogue.update_system_message(self.prompt) def chat(self, query, tool_call=False): + self.llm_finish_task = False self.logger.bind(tag=TAG).debug(f"Chat: {query}") if not tool_call: @@ -511,7 +508,6 @@ class ConnectionHandler: if self.intent_type == "function_call" and hasattr(self, "func_handler"): functions = self.func_handler.get_functions() response_message = [] - processed_chars = 0 # 跟踪已处理的字符位置 try: # 使用带记忆的对话 @@ -523,7 +519,7 @@ class ConnectionHandler: memory_str = future.result() uuid_str = str(uuid.uuid4()).replace("-", "") - self.tts_session_id = uuid_str + self.sentence_id = uuid_str if functions is not None: # 使用支持functions的streaming接口 @@ -541,15 +537,13 @@ class ConnectionHandler: 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 = "" + text_index = 0 for response in llm_responses: if self.intent_type == "function_call": @@ -577,54 +571,26 @@ class ConnectionHandler: if content is not None and len(content) > 0: if not tool_call_flag: response_message.append(content) - if self.client_abort: break - # 处理文本分段和TTS逻辑 - # 合并当前全部文本并处理未分割部分 - full_text = "".join(response_message) - current_text = full_text[processed_chars:] # 从未处理的位置开始 - - # 查找最后一个有效标点 - punctuations = ( - "。", - ".", - "？", - "?", - "！", - "!", - "；", - ";", - "：", - ) - last_punct_pos = -1 - number_flag = True - for punct in punctuations: - pos = current_text.rfind(punct) - prev_char = current_text[pos - 1] if pos - 1 >= 0 else "" - # 如果.前面是数字统一判断为小数 - if prev_char.isdigit() and punct == ".": - number_flag = False - if pos > last_punct_pos and number_flag: - last_punct_pos = pos - - # 找到分割点则处理 - if last_punct_pos != -1: - segment_text_raw = current_text[: last_punct_pos + 1] - segment_text = get_string_no_punctuation_or_emoji( - segment_text_raw - ) - if segment_text: - text_index += 1 - self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit( - self.speak_and_play, None, segment_text, text_index + if text_index == 0: + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, ) - self.tts.tts_queue.put((future, text_index)) - # 更新已处理字符位置 - processed_chars += len(segment_text_raw) - + ) + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=content, + ) + ) + text_index += 1 # 处理function call if tool_call_flag: bHasError = False @@ -667,27 +633,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) - - # 处理最后剩余的文本 - full_text = "".join(response_message) - remaining_text = full_text[processed_chars:] - if remaining_text: - segment_text = get_string_no_punctuation_or_emoji(remaining_text) - if segment_text: - text_index += 1 - self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit( - self.speak_and_play, None, segment_text, text_index - ) - self.tts.tts_queue.put((future, text_index)) + 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) @@ -737,12 +697,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) - future = self.executor.submit(self.speak_and_play, None, text, text_index) - self.tts.tts_queue.put((future, text_index)) + 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 @@ -779,9 +737,7 @@ class ConnectionHandler: 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) - future = self.executor.submit(self.speak_and_play, None, text, text_index) - self.tts.tts_queue.put((future, text_index)) + self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) self.dialogue.put(Message(role="assistant", content=text)) else: pass @@ -812,33 +768,9 @@ class ConnectionHandler: self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") - def speak_and_play(self, file_path, content, text_index=0): - if file_path is not None: - self.logger.bind(tag=TAG).info(f"无需tts转换: 从文件播放，{file_path}") - return file_path, content, text_index - if content is None or len(content) <= 0: - self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{content}") - return None, content, text_index - tts_file = self.tts.to_tts(content, text_index) - if tts_file is None: - self.logger.bind(tag=TAG).error(f"tts转换失败，{content}") - return None, content, text_index - self.logger.bind(tag=TAG).debug(f"TTS 文件生成完毕: {tts_file}") - if self.max_output_size > 0: - add_device_output(self.headers.get("device-id"), len(content)) - return tts_file, content, text_index - def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") self.asr_server_receive = True - 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): """资源清理方法""" @@ -875,11 +807,11 @@ class ConnectionHandler: def clear_queues(self): """清空所有任务队列""" self.logger.bind(tag=TAG).debug( - f"开始清理: TTS队列大小={self.tts.tts_queue.qsize()}, 音频队列大小={self.tts.audio_play_queue.qsize()}" + f"开始清理: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}" ) # 使用非阻塞方式清空队列 - for q in [self.tts.tts_queue, self.tts.audio_play_queue]: + for q in [self.tts.tts_text_queue, self.tts.tts_audio_queue]: if not q: continue while True: @@ -889,7 +821,7 @@ class ConnectionHandler: break self.logger.bind(tag=TAG).debug( - f"清理结束: TTS队列大小={self.tts.tts_queue.qsize()}, 音频队列大小={self.tts.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 100dde80..8a0b2ab9 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -44,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: @@ -56,7 +53,7 @@ async def checkWakeupWords(conn, text): text_hello = WAKEUP_CONFIG["text"] if not text_hello: text_hello = text - conn.tts.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 @@ -91,7 +88,7 @@ async def wakeupWordsResponse(conn): result = conn.llm.response_no_stream(conn.config["prompt"], wakeup_word) if result is None or result == "": return - tts_file = await asyncio.to_thread(conn.tts.to_tts, result, 0) + tts_file = await asyncio.to_thread(conn.tts.to_tts, result) if tts_file is not None and os.path.exists(tts_file): file_type = os.path.splitext(tts_file)[1] diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index e61b1265..f3dd5cc5 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -4,6 +4,7 @@ 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 @@ -143,11 +144,4 @@ async def process_intent_result(conn, intent_result, original_text): def speak_txt(conn, text): - text_index = ( - conn.tts_last_text_index + 1 if hasattr(conn, "tts_last_text_index") else 0 - ) - conn.recode_first_last_text(text, text_index) - future = conn.executor.submit(conn.speak_and_play, None, text, text_index) - conn.llm_finish_task = True - conn.tts_queue.put((future, text_index)) - conn.dialogue.put(Message(role="assistant", content=text)) + 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 91bb2bb4..49ba3d07 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -110,12 +110,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.tts.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((opus_packets, text, 0)) conn.close_after_chat = True @@ -130,14 +127,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.tts.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((opus_packets, text, 0)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 @@ -145,16 +139,13 @@ 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.tts.audio_play_queue.put((num_packets, None, i + 1)) + conn.tts.tts_audio_queue.put((num_packets, None, i + 1)) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue 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.tts.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((opus_packets, text, 0)) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index d400be63..f3561efd 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,6 +1,7 @@ import json import asyncio import time +from core.providers.tts.dto.dto import SentenceType from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion TAG = __name__ @@ -30,7 +31,7 @@ emoji_map = { } -async def sendAudioMessage(conn, audios, text, text_index=0): +async def sendAudioMessage(conn, sentenceType, audios, text): # 发送句子开始消息 if text is not None: emotion = analyze_emotion(text) @@ -46,25 +47,24 @@ async def sendAudioMessage(conn, audios, text, text_index=0): ) ) - if text_index == conn.tts_first_text_index: - conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") await send_tts_message(conn, "sentence_start", text) - is_first_audio = text_index == conn.tts_first_text_index - await sendAudio(conn, audios, pre_buffer=is_first_audio) + await sendAudio(conn, audios, True) await send_tts_message(conn, "sentence_end", text) # 发送结束消息（如果是最后一个文本） - if conn.llm_finish_task and text_index == conn.tts_last_text_index: + if conn.llm_finish_task and sentenceType == SentenceType.LAST: await send_tts_message(conn, "stop", None) - await conn.tts.finish_session(conn.tts_session_id) + await conn.tts.finish_session(conn.sentence_id) 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 # 流控参数优化 frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码 start_time = time.perf_counter() diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index a0956325..44c2aacb 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -2,12 +2,12 @@ import asyncio from config.logger import setup_logging import queue import os -import json +import uuid import threading -from enum import Enum from core.utils import p3 +from core.utils import textUtils from core.handle.sendAudioHandle import sendAudioMessage -from core.handle.reportHandle import enqueue_tts_report +from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType from abc import ABC, abstractmethod from core.utils.tts import MarkdownCleaner from core.utils.util import audio_to_data @@ -16,30 +16,39 @@ TAG = __name__ logger = setup_logging() -class TTSImplementationType(Enum): - """TTS实现类型枚举""" - - NON_STREAMING = "non_streaming" # 非流式实现 - SINGLE_STREAMING = "single_streaming" # 单流式实现 - DOUBLE_STREAMING = "double_streaming" # 双流式实现 - - class TTSProviderBase(ABC): def __init__(self, config, delete_audio_file): self.conn = None self.tts_timeout = 10 self.delete_audio_file = delete_audio_file self.output_file = config.get("output_dir") - self.tts_queue = queue.Queue() - self.audio_play_queue = queue.Queue() - # 添加实现类型属性，默认为非流式 - self.interface_type = TTSImplementationType.NON_STREAMING + self.tts_text_queue = queue.Queue() + self.tts_audio_queue = queue.Queue() + self.stop_event = threading.Event() + self.loop = asyncio.get_event_loop() + + self.tts_text_buff = [] + self.punctuations = ( + "。", + ".", + "？", + "?", + "！", + "!", + "；", + ";", + "：", + ) + self.first_sentence_punctuations = ("，", ",") + self.tts_stop_request = False + self.processed_chars = 0 + self.is_first_sentence = True @abstractmethod def generate_filename(self): pass - def to_tts(self, text, index): + def to_tts(self, text): tmp_file = self.generate_filename() try: max_repeat_time = 5 @@ -82,12 +91,51 @@ class TTSProviderBase(ABC): """音频文件转换为Opus编码""" return audio_to_data(audio_file_path, is_opus=True) + 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, + ) + ) + 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_priority_thread, daemon=True + target=self._tts_text_priority_thread, daemon=True ) self.tts_priority_thread.start() @@ -97,113 +145,60 @@ class TTSProviderBase(ABC): ) self.audio_play_priority_thread.start() - def _tts_priority_thread(self): - while not self.conn.stop_event.is_set(): - text = None + # 这里默认是非流式的处理方式 + # 流式处理方式请在子类中重写 + def _tts_text_priority_thread(self): + while not self.stop_event.is_set(): try: - try: - item = self.tts_queue.get(timeout=1) - if item is None: - continue - future, text_index = item # 解包获取 Future 和 text_index - except queue.Empty: - if self.conn.stop_event.is_set(): - break - continue - if future is None: - continue - text = None - audio_datas, tts_file = [], None - try: - logger.bind(tag=TAG).debug("正在处理TTS任务...") - tts_file, text, _ = future.result(timeout=self.tts_timeout) - - # 如果tts_file返回流式标识，则不继续处理 - if tts_file is None: - logger.bind(tag=TAG).error( - f"TTS出错： file is empty: {text_index}: {text}" + message = self.tts_text_queue.get() + if message.sentence_type == SentenceType.FIRST: + # 初始化参数 + self.tts_stop_request = False + self.processed_chars = 0 + self.tts_text_buff = [] + self.is_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) + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, segment_text) ) - elif ( - tts_file == TTSImplementationType.SINGLE_STREAMING.value - or tts_file == TTSImplementationType.DOUBLE_STREAMING.value - ): - logger.bind(tag=TAG).debug(f"TTS生成：流式标识: {tts_file}") - continue - else: - logger.bind(tag=TAG).debug(f"TTS生成：文件路径: {tts_file}") - if os.path.exists(tts_file): - if tts_file.endswith(".p3"): - audio_datas, _ = p3.decode_opus_from_file(tts_file) - elif self.conn.audio_format == "pcm": - audio_datas, _ = self.audio_to_pcm_data(tts_file) - else: - audio_datas, _ = self.audio_to_opus_data(tts_file) - # 在这里上报TTS数据 - enqueue_tts_report( - self.conn, - tts_file if text is None else text, - audio_datas, - ) - else: - logger.bind(tag=TAG).error(f"TTS出错：文件不存在{tts_file}") - except TimeoutError: - logger.bind(tag=TAG).error("TTS超时") - except Exception as e: - import traceback + elif ContentType.FILE == message.content_type: + self._process_remaining_text() - error_info = { - "error_type": type(e).__name__, - "error_message": str(e), - "stack_trace": traceback.format_exc(), - "text_index": text_index, - "text": text, - "tts_file": tts_file, - "audio_format": getattr(self.conn, "audio_format", None), - "interface_type": self.interface_type.value, - } - logger.bind(tag=TAG).error( - f"TTS处理出错: {json.dumps(error_info, ensure_ascii=False)}" + tts_file = message.content_file + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, message.content_detail) ) - if not self.conn.client_abort: - # 如果没有中途打断就发送语音 - self.audio_play_queue.put((audio_datas, text, text_index)) - if ( - self.delete_audio_file - and tts_file is not None - and os.path.exists(tts_file) - and tts_file.startswith(self.output_file) - ): - os.remove(tts_file) + if message.sentence_type == SentenceType.LAST: + self._process_remaining_text() + + self.tts_audio_queue.put( + (message.sentence_type, [], message.content_detail) + ) + except Exception as e: - logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}") - self.conn.clearSpeakStatus() - asyncio.run_coroutine_threadsafe( - self.conn.websocket.send( - json.dumps( - { - "type": "tts", - "state": "stop", - "session_id": self.conn.session_id, - } - ) - ), - self.conn.loop, - ) - logger.bind(tag=TAG).error(f"tts_priority priority_thread: {text} {e}") + logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}") def _audio_play_priority_thread(self): - while not self.conn.stop_event.is_set(): + while not self.stop_event.is_set(): text = None try: try: - audio_datas, text, text_index = self.audio_play_queue.get(timeout=1) + sentence_type, audio_datas, text = self.tts_audio_queue.get( + timeout=1 + ) except queue.Empty: - if self.conn.stop_event.is_set(): + if self.stop_event.is_set(): break continue future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self.conn, audio_datas, text, text_index), - self.conn.loop, + sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self.loop, ) future.result() except Exception as e: @@ -221,3 +216,81 @@ class TTSProviderBase(ABC): """资源清理方法""" 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/dto/dto.py b/main/xiaozhi-server/core/providers/tts/dto/dto.py new file mode 100644 index 00000000..a57a27de --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/dto/dto.py @@ -0,0 +1,36 @@ +from enum import Enum +from typing import Union, Optional + + +class SentenceType(Enum): + # 说话阶段 + FIRST = "FIRST" # 首句话 + MIDDLE = "MIDDLE" # 说话中 + LAST = "LAST" # 最后一句 + + +class ContentType(Enum): + # 内容类型 + TEXT = "TEXT" # 文本内容 + FILE = "FILE" # 文件内容 + ACTION = "ACTION" # 动作内容 + + +class TTSMessageDTO: + 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_type = content_type + self.content_detail = content_detail + self.content_file = content_file diff --git a/main/xiaozhi-server/core/providers/tts/huoshan.py b/main/xiaozhi-server/core/providers/tts/huoshan.py index bc7d05c8..a5bcfd76 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan.py @@ -7,7 +7,7 @@ import json import websockets from datetime import datetime from config.logger import setup_logging -from core.providers.tts.base import TTSProviderBase, TTSImplementationType +from core.providers.tts.base import TTSProviderBase from core.utils.util import pcm_to_data TAG = __name__ @@ -149,7 +149,6 @@ class TTSProvider(TTSProviderBase): self.enable_two_way = True self.start_connection_flag = False self.tts_text = "" - self.interface_type = TTSImplementationType.DOUBLE_STREAMING async def open_audio_channels(self, conn): await super().open_audio_channels(conn) @@ -173,17 +172,21 @@ class TTSProvider(TTSProviderBase): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - def to_tts(self, text, index): - if index == self.conn.tts_first_text_index: - future = asyncio.run_coroutine_threadsafe( - self.start_session(self.conn.tts_session_id), loop=self.conn.loop - ) - future.result() + def to_tts(self, text): future = asyncio.run_coroutine_threadsafe( - self.text_to_speak(text, None), loop=self.conn.loop + self.start_session(self.conn.sentence_id), loop=self.loop + ) + future.result() + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(text, None), loop=self.loop + ) + future.result() + future = asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), loop=self.loop ) future.result() return self.interface_type.value + async def send_event( self, header: bytes, optional: bytes | None = None, payload: bytes = None ): @@ -341,6 +344,7 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).info(f"会话开始～～{session_id}") async def finish_session(self, session_id): + logger.bind(tag=TAG).info(f"会话结束～～{session_id}") self.stop_event_response.set() header = Header( message_type=FULL_CLIENT_REQUEST, @@ -369,19 +373,18 @@ class TTSProvider(TTSProviderBase): async def text_to_speak(self, text, _): # 发送文本 - await self.send_text(self.speaker, text, self.conn.tts_session_id) + await self.send_text(self.speaker, text, self.conn.sentence_id) logger.bind(tag=TAG).info(f"发送文本～～{text}") return def _start_monitor_tts_response_thread(self): # 初始化链接 asyncio.run_coroutine_threadsafe( - self._start_monitor_tts_response(), loop=self.conn.loop + self._start_monitor_tts_response(), loop=self.loop ) async def _start_monitor_tts_response(self): - chunk_total = b"" - while not self.conn.stop_event.is_set(): + while not self.stop_event.is_set(): try: msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop res = self.parser_response(msg) @@ -396,20 +399,19 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).info( f"推送数据到队列里面帧数～～{len(opus_datas)}" ) - self.audio_play_queue.put( + self.tts_audio_queue.put( (opus_datas, None, self.conn.tts_last_text_index - 1) ) 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.audio_play_queue.put( + self.tts_audio_queue.put( ([], self.tts_text, self.conn.tts_first_text_index) ) - elif res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子结束～～{self.tts_text}") - self.audio_play_queue.put( + self.tts_audio_queue.put( ([], self.tts_text, self.conn.tts_last_text_index) ) elif res.optional.event == EVENT_SessionFinished: diff --git a/main/xiaozhi-server/core/utils/textUtils.py b/main/xiaozhi-server/core/utils/textUtils.py new file mode 100644 index 00000000..7fadd3ff --- /dev/null +++ b/main/xiaozhi-server/core/utils/textUtils.py @@ -0,0 +1,34 @@ +def get_string_no_punctuation_or_emoji(s): + """去除字符串首尾的空格、标点符号和表情符号""" + chars = list(s) + # 处理开头的字符 + start = 0 + while start < len(chars) and is_punctuation_or_emoji(chars[start]): + start += 1 + # 处理结尾的字符 + end = len(chars) - 1 + while end >= start and is_punctuation_or_emoji(chars[end]): + end -= 1 + return ''.join(chars[start:end + 1]) + +def is_punctuation_or_emoji(char): + """检查字符是否为空格、指定标点或表情符号""" + # 定义需要去除的中英文标点（包括全角/半角） + punctuation_set = { + '，', ',', # 中文逗号 + 英文逗号 + '。', '.', # 中文句号 + 英文句号 + '！', '!', # 中文感叹号 + 英文感叹号 + '-', '－', # 英文连字符 + 中文全角横线 + '、' # 中文顿号 + } + if char.isspace() or char in punctuation_set: + return True + # 检查表情符号（保留原有逻辑） + code_point = ord(char) + emoji_ranges = [ + (0x1F600, 0x1F64F), (0x1F300, 0x1F5FF), + (0x1F680, 0x1F6FF), (0x1F900, 0x1F9FF), + (0x1FA70, 0x1FAFF), (0x2600, 0x26FF), + (0x2700, 0x27BF) + ] + return any(start <= code_point <= end for start, end in emoji_ranges) \ No newline at end of file diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 0446e4b2..6d184b6b 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -11,6 +11,7 @@ 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__ @@ -217,14 +218,36 @@ async def play_local_music(conn, specific_file=None): await send_stt_message(conn, text) conn.dialogue.put(Message(role="assistant", content=text)) - conn.recode_first_last_text(text, 0) - future = conn.executor.submit(conn.speak_and_play, None, text, 0) - conn.tts.tts_queue.put((future, 0)) - - conn.recode_first_last_text(text, 1) - future = conn.executor.submit(conn.speak_and_play, music_path, None, 1) - conn.tts.tts_queue.put((future, 1)) - conn.llm_finish_task = True + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + 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, + ) + ) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")