update:旧非流失兼容改造

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.sentence_id = None
-        self.tts_last_text_index = -1
-        self.tts_session_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逻辑
+                    if text_index == 0:
-                    # 合并当前全部文本并处理未分割部分
+                        self.tts.tts_text_queue.put(
-                    full_text = "".join(response_message)
+                            TTSMessageDTO(
-                    current_text = full_text[processed_chars:]  # 从未处理的位置开始
+                                sentence_id=self.sentence_id,
-
+                                sentence_type=SentenceType.FIRST,
-                    # 查找最后一个有效标点
+                                content_type=ContentType.ACTION,
-                    punctuations = (
-                        "。",
-                        ".",
-                        "？",
-                        "?",
-                        "！",
-                        "!",
-                        "；",
-                        ";",
-                        "：",
-                    )
-                    last_punct_pos = -1
-                    number_flag = True
-                    for punct in punctuations:
-                        pos = current_text.rfind(punct)
-                        prev_char = current_text[pos - 1] if pos - 1 >= 0 else ""
-                        # 如果.前面是数字统一判断为小数
-                        if prev_char.isdigit() and punct == ".":
-                            number_flag = False
-                        if pos > last_punct_pos and number_flag:
-                            last_punct_pos = pos
-
-                    # 找到分割点则处理
-                    if last_punct_pos != -1:
-                        segment_text_raw = current_text[: last_punct_pos + 1]
-                        segment_text = get_string_no_punctuation_or_emoji(
-                            segment_text_raw
-                        )
-                        if segment_text:
-                            text_index += 1
-                            self.recode_first_last_text(segment_text, text_index)
-                            future = self.executor.submit(
-                                self.speak_and_play, None, segment_text, text_index
                             )
-                            self.tts.tts_queue.put((future, text_index))
+                        )
-                            # 更新已处理字符位置
+                    self.tts.tts_text_queue.put(
-                            processed_chars += len(segment_text_raw)
+                        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)
+                self._handle_function_result(result, function_call_data)
-
-        # 处理最后剩余的文本
-        full_text = "".join(response_message)
-        remaining_text = full_text[processed_chars:]
-        if remaining_text:
-            segment_text = get_string_no_punctuation_or_emoji(remaining_text)
-            if segment_text:
-                text_index += 1
-                self.recode_first_last_text(segment_text, text_index)
-                future = self.executor.submit(
-                    self.speak_and_play, None, segment_text, text_index
-                )
-                self.tts.tts_queue.put((future, text_index))
 
         # 存储对话内容
         if len(response_message) > 0:
             self.dialogue.put(
                 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)
+            self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
-            future = self.executor.submit(self.speak_and_play, None, text, text_index)
-            self.tts.tts_queue.put((future, text_index))
             self.dialogue.put(Message(role="assistant", content=text))
         elif result.action == Action.REQLLM:  # 调用函数后再请求llm生成回复
             text = result.result
@@ -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)
+            self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
-            future = self.executor.submit(self.speak_and_play, None, text, text_index)
-            self.tts.tts_queue.put((future, text_index))
             self.dialogue.put(Message(role="assistant", content=text))
         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):

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]

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.tts_one_sentence(conn, ContentType.TEXT, content_detail=text)
-        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))

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))

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, True)
-    await sendAudio(conn, audios, pre_buffer=is_first_audio)
 
     await send_tts_message(conn, "sentence_end", text)
 
     # 发送结束消息（如果是最后一个文本）
-    if conn.llm_finish_task and text_index == conn.tts_last_text_index:
+    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()

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.tts_text_queue = queue.Queue()
-        self.audio_play_queue = queue.Queue()
+        self.tts_audio_queue = queue.Queue()
-        # 添加实现类型属性，默认为非流式
+        self.stop_event = threading.Event()
-        self.interface_type = TTSImplementationType.NON_STREAMING
+        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:
+                message = self.tts_text_queue.get()
-                    item = self.tts_queue.get(timeout=1)
+                if message.sentence_type == SentenceType.FIRST:
-                    if item is None:
+                    # 初始化参数
-                        continue
+                    self.tts_stop_request = False
-                    future, text_index = item  # 解包获取 Future 和 text_index
+                    self.processed_chars = 0
-                except queue.Empty:
+                    self.tts_text_buff = []
-                    if self.conn.stop_event.is_set():
+                    self.is_first_sentence = True
-                        break
+                elif ContentType.TEXT == message.content_type:
-                    continue
+                    self.tts_text_buff.append(message.content_detail)
-                if future is None:
+                    segment_text = self._get_segment_text()
-                    continue
+                    if segment_text:
-                text = None
+                        tts_file = self.to_tts(segment_text)
-                audio_datas, tts_file = [], None
+                        audio_datas = self._process_audio_file(tts_file)
-                try:
+                        self.tts_audio_queue.put(
-                    logger.bind(tag=TAG).debug("正在处理TTS任务...")
+                            (message.sentence_type, audio_datas, segment_text)
-                    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}"
                         )
-                    elif (
+                elif ContentType.FILE == message.content_type:
-                        tts_file == TTSImplementationType.SINGLE_STREAMING.value
+                    self._process_remaining_text()
-                        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
 
-                    error_info = {
+                    tts_file = message.content_file
-                        "error_type": type(e).__name__,
+                    audio_datas = self._process_audio_file(tts_file)
-                        "error_message": str(e),
+                    self.tts_audio_queue.put(
-                        "stack_trace": traceback.format_exc(),
+                        (message.sentence_type, audio_datas, message.content_detail)
-                        "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)}"
                     )
-                if not self.conn.client_abort:
+                if message.sentence_type == SentenceType.LAST:
-                    # 如果没有中途打断就发送语音
+                    self._process_remaining_text()
-                    self.audio_play_queue.put((audio_datas, text, text_index))
+
-                if (
+                    self.tts_audio_queue.put(
-                    self.delete_audio_file
+                        (message.sentence_type, [], message.content_detail)
-                    and tts_file is not None
+                    )
-                    and os.path.exists(tts_file)
+
-                    and tts_file.startswith(self.output_file)
-                ):
-                    os.remove(tts_file)
             except Exception as e:
-                logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}")
+                logger.bind(tag=TAG).error(f"Failed to process TTS text: {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}")
 
     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),
+                    sendAudioMessage(self.conn, sentence_type, audio_datas, text),
-                    self.conn.loop,
+                    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

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

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):
+    def to_tts(self, text):
-        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()
         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.stop_event.is_set():
-        while not self.conn.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:

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)

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)
+        conn.tts.tts_text_queue.put(
-        future = conn.executor.submit(conn.speak_and_play, None, text, 0)
+            TTSMessageDTO(
-        conn.tts.tts_queue.put((future, 0))
+                sentence_id=conn.sentence_id,
-
+                sentence_type=SentenceType.FIRST,
-        conn.recode_first_last_text(text, 1)
+                content_type=ContentType.ACTION,
-        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.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)}")