Merge branch 'main' into py_MinmaxStreamTTS_test

main/xiaozhi-server/core/providers/asr/base.py

@@ -1,18 +1,18 @@
 import os
+import io
 import wave
 import uuid
+import json
+import time
 import queue
 import asyncio
 import traceback
 import threading
 import opuslib_next
-import json
-import io
-import time
 import concurrent.futures
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from typing import Optional, Tuple, List, Dict, Any
+from typing import Optional, Tuple, List
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.utils.util import remove_punctuation_and_length
@@ -87,11 +87,9 @@ class ASRProviderBase(ABC):
             
             # 预先准备WAV数据
             wav_data = None
-            # 使用连接的声纹识别提供者
             if conn.voiceprint_provider and combined_pcm_data:
                 wav_data = self._pcm_to_wav(combined_pcm_data)
             
-            
             # 定义ASR任务
             def run_asr():
                 start_time = time.monotonic()
@@ -132,8 +130,6 @@ class ASRProviderBase(ABC):
                     return None
             
             # 使用线程池执行器并行运行
-            parallel_start_time = time.monotonic()
-            
             with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
                 asr_future = thread_executor.submit(run_asr)
                 
@@ -151,7 +147,7 @@ class ASRProviderBase(ABC):
             
             
             # 处理结果
-            raw_text, file_path = results.get("asr", ("", None))
+            raw_text, _ = results.get("asr", ("", None))
             speaker_name = results.get("voiceprint", None)
             
             # 记录识别结果

main/xiaozhi-server/core/providers/llm/AliBL/AliBL.py

@@ -1,8 +1,10 @@
 from config.logger import setup_logging
 from http import HTTPStatus
+import dashscope
 from dashscope import Application
 from core.providers.llm.base import LLMProviderBase
 from core.utils.util import check_model_key
+import time
 
 TAG = __name__
 logger = setup_logging()
@@ -15,6 +17,7 @@ class LLMProvider(LLMProviderBase):
         self.base_url = config.get("base_url")
         self.is_No_prompt = config.get("is_no_prompt")
         self.memory_id = config.get("ali_memory_id")
+        self.streaming_chunk_size = config.get("streaming_chunk_size", 3)  # 每次流式返回的字符数
         check_model_key("AliBLLLM", self.api_key)
 
     def response(self, session_id, dialogue):
@@ -32,6 +35,8 @@ class LLMProvider(LLMProviderBase):
                 "app_id": self.app_id,
                 "session_id": session_id,
                 "messages": dialogue,
+                # 开启SDK原生流式
+                "stream": True,
             }
             if self.memory_id != False:
                 # 百练memory需要prompt参数
@@ -42,25 +47,63 @@ class LLMProvider(LLMProviderBase):
                     f"【阿里百练API服务】处理后的prompt: {prompt}"
                 )
 
+            # 可选地设置自定义API基地址（若配置为兼容模式URL则忽略）
+            if self.base_url and ("/api/" in self.base_url):
+                dashscope.base_http_api_url = self.base_url
+
             responses = Application.call(**call_params)
-            if responses.status_code != HTTPStatus.OK:
-                logger.bind(tag=TAG).error(
-                    f"code={responses.status_code}, "
-                    f"message={responses.message}, "
-                    f"请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
-                )
-                yield "【阿里百练API服务响应异常】"
-            else:
-                logger.bind(tag=TAG).debug(
-                    f"【阿里百练API服务】构造参数: {call_params}"
-                )
-                yield responses.output.text
+
+            # 流式处理（SDK在stream=True时返回可迭代对象；否则返回单次响应对象）
+            logger.bind(tag=TAG).debug(
+                f"【阿里百练API服务】构造参数: {dict(call_params, api_key='***')}"
+            )
+
+            last_text = ""
+            try:
+                for resp in responses:
+                    if resp.status_code != HTTPStatus.OK:
+                        logger.bind(tag=TAG).error(
+                            f"code={resp.status_code}, message={resp.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
+                        )
+                        continue
+                    current_text = getattr(getattr(resp, "output", None), "text", None)
+                    if current_text is None:
+                        continue
+                    # SDK流式为增量覆盖，计算差量输出
+                    if len(current_text) >= len(last_text):
+                        delta = current_text[len(last_text):]
+                    else:
+                        # 避免偶发回退
+                        delta = current_text
+                    if delta:
+                        yield delta
+                    last_text = current_text
+            except TypeError:
+                # 非流式回落（一次性返回）
+                if responses.status_code != HTTPStatus.OK:
+                    logger.bind(tag=TAG).error(
+                        f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
+                    )
+                    yield "【阿里百练API服务响应异常】"
+                else:
+                    full_text = getattr(getattr(responses, "output", None), "text", "")
+                    logger.bind(tag=TAG).info(
+                        f"【阿里百练API服务】完整响应长度: {len(full_text)}"
+                    )
+                    for i in range(0, len(full_text), self.streaming_chunk_size):
+                        chunk = full_text[i:i + self.streaming_chunk_size]
+                        if chunk:
+                            yield chunk
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"【阿里百练API服务】响应异常: {e}")
             yield "【LLM服务响应异常】"
 
     def response_with_functions(self, session_id, dialogue, functions=None):
-        logger.bind(tag=TAG).error(
-            f"阿里百练暂未实现完整的工具调用（function call），建议使用其他意图识别"
+        # 阿里百练当前未支持原生的 function call。为保持兼容，这里回退到普通文本流式输出。
+        # 上层会按 (content, tool_calls) 的形式消费，这里始终返回 (token, None)
+        logger.bind(tag=TAG).warning(
+            "阿里百练未实现原生 function call，已回退为纯文本流式输出"
         )
+        for token in self.response(session_id, dialogue):
+            yield token, None

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_executor.py

@@ -18,8 +18,8 @@ class ServerMCPExecutor(ToolExecutor):
         """初始化MCP管理器"""
         if not self._initialized:
             self.mcp_manager = ServerMCPManager(self.conn)
-            await self.mcp_manager.initialize_servers()
             self._initialized = True
+            await self.mcp_manager.initialize_servers()
 
     async def execute(
         self, conn, tool_name: str, arguments: Dict[str, Any]

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_manager.py

@@ -68,6 +68,9 @@ class ServerMCPManager:
 
         # 输出当前支持的服务端MCP工具列表
         if hasattr(self.conn, "func_handler") and self.conn.func_handler:
+            # 刷新工具缓存以确保服务端MCP工具被正确加载
+            if hasattr(self.conn.func_handler, "tool_manager"):
+                self.conn.func_handler.tool_manager.refresh_tools()
             self.conn.func_handler.current_support_functions()
 
     def get_all_tools(self) -> List[Dict[str, Any]]:

main/xiaozhi-server/core/providers/tts/aliyun_stream.py

@@ -216,7 +216,6 @@ class TTSProvider(TTSProviderBase):
 
                 if message.sentence_type == SentenceType.FIRST:
                     self.conn.client_abort = False
-                    self.reset_flow_controller()
 
                 if self.conn.client_abort:
                     logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
@@ -479,3 +478,142 @@ class TTSProvider(TTSProviderBase):
         finally:
             self._monitor_task = None
 
+    def to_tts(self, text: str) -> list:
+        """非流式TTS处理，用于测试及保存音频文件的场景"""
+        try:
+            # 创建新的事件循环
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+            # 生成会话ID
+            session_id = uuid.uuid4().hex
+            # 存储音频数据
+            audio_data = []
+
+            async def _generate_audio():
+                # 刷新Token（如果需要）
+                if self._is_token_expired():
+                    self._refresh_token()
+
+                # 建立WebSocket连接
+                ws = await websockets.connect(
+                    self.ws_url,
+                    additional_headers={"X-NLS-Token": self.token},
+                    ping_interval=30,
+                    ping_timeout=10,
+                    close_timeout=10,
+                )
+                try:
+                    # 发送StartSynthesis请求
+                    start_message_id = str(uuid.uuid4().hex)
+                    start_request = {
+                        "header": {
+                            "message_id": start_message_id,
+                            "task_id": session_id,
+                            "namespace": "FlowingSpeechSynthesizer",
+                            "name": "StartSynthesis",
+                            "appkey": self.appkey,
+                        },
+                        "payload": {
+                            "voice": self.voice,
+                            "format": self.format,
+                            "sample_rate": self.sample_rate,
+                            "volume": self.volume,
+                            "speech_rate": self.speech_rate,
+                            "pitch_rate": self.pitch_rate,
+                            "enable_subtitle": True,
+                        },
+                    }
+                    await ws.send(json.dumps(start_request))
+
+                    # 等待SynthesisStarted响应
+                    synthesis_started = False
+                    while not synthesis_started:
+                        msg = await ws.recv()
+                        if isinstance(msg, str):
+                            data = json.loads(msg)
+                            header = data.get("header", {})
+                            if header.get("name") == "SynthesisStarted":
+                                synthesis_started = True
+                                logger.bind(tag=TAG).debug("TTS合成已启动")
+                            elif header.get("name") == "TaskFailed":
+                                error_info = data.get("payload", {}).get(
+                                    "error_info", {}
+                                )
+                                error_code = error_info.get("error_code")
+                                error_message = error_info.get(
+                                    "error_message", "未知错误"
+                                )
+                                raise Exception(
+                                    f"启动合成失败: {error_code} - {error_message}"
+                                )
+
+                    # 发送文本合成请求
+                    filtered_text = MarkdownCleaner.clean_markdown(text)
+                    run_message_id = str(uuid.uuid4().hex)
+                    run_request = {
+                        "header": {
+                            "message_id": run_message_id,
+                            "task_id": session_id,
+                            "namespace": "FlowingSpeechSynthesizer",
+                            "name": "RunSynthesis",
+                            "appkey": self.appkey,
+                        },
+                        "payload": {"text": filtered_text},
+                    }
+                    await ws.send(json.dumps(run_request))
+
+                    # 发送停止合成请求
+                    stop_message_id = str(uuid.uuid4().hex)
+                    stop_request = {
+                        "header": {
+                            "message_id": stop_message_id,
+                            "task_id": session_id,
+                            "namespace": "FlowingSpeechSynthesizer",
+                            "name": "StopSynthesis",
+                            "appkey": self.appkey,
+                        }
+                    }
+                    await ws.send(json.dumps(stop_request))
+
+                    # 接收音频数据
+                    synthesis_completed = False
+                    while not synthesis_completed:
+                        msg = await ws.recv()
+                        if isinstance(msg, (bytes, bytearray)):
+                            self.opus_encoder.encode_pcm_to_opus_stream(
+                                msg,
+                                end_of_stream=False,
+                                callback=lambda opus: audio_data.append(opus)
+                            )
+                        elif isinstance(msg, str):
+                            data = json.loads(msg)
+                            header = data.get("header", {})
+                            event_name = header.get("name")
+                            if event_name == "SynthesisCompleted":
+                                synthesis_completed = True
+                                logger.bind(tag=TAG).debug("TTS合成完成")
+                            elif event_name == "TaskFailed":
+                                error_info = data.get("payload", {}).get(
+                                    "error_info", {}
+                                )
+                                error_code = error_info.get("error_code")
+                                error_message = error_info.get(
+                                    "error_message", "未知错误"
+                                )
+                                raise Exception(
+                                    f"合成失败: {error_code} - {error_message}"
+                                )
+                finally:
+                    try:
+                        await ws.close()
+                    except:
+                        pass
+
+            loop.run_until_complete(_generate_audio())
+            loop.close()
+
+            return audio_data
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
+            return []

main/xiaozhi-server/core/providers/tts/base.py

@@ -1,22 +1,22 @@
 import os
 import re
-import queue
+import time
 import uuid
+import queue
 import asyncio
 import threading
-from typing import Callable, Any
+import traceback
 from core.utils import p3
-import time
 from datetime import datetime
 from core.utils import textUtils
+from typing import Callable, Any
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from core.utils.audio_flow_control import FlowControlConfig
-from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
 from core.utils.tts import MarkdownCleaner
 from core.utils.output_counter import add_device_output
 from core.handle.reportHandle import enqueue_tts_report
 from core.handle.sendAudioHandle import sendAudioMessage
+from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
 from core.providers.tts.dto.dto import (
     TTSMessageDTO,
     SentenceType,
@@ -24,8 +24,6 @@ from core.providers.tts.dto.dto import (
     InterfaceType,
 )
 
-import traceback
-
 TAG = __name__
 logger = setup_logging()
 
@@ -34,7 +32,6 @@ class TTSProviderBase(ABC):
     def __init__(self, config, delete_audio_file):
         self.interface_type = InterfaceType.NON_STREAM
         self.conn = None
-        self.tts_timeout = 10
         self.delete_audio_file = delete_audio_file
         self.audio_file_type = "wav"
         self.output_file = config.get("output_dir", "tmp/")
@@ -71,7 +68,6 @@ class TTSProviderBase(ABC):
         self.tts_stop_request = False
         self.processed_chars = 0
         self.is_first_sentence = True
-        self.flow_controller = FlowControlConfig.create_flow_controller()
 
     def generate_filename(self, extension=".wav"):
         return os.path.join(
@@ -147,6 +143,68 @@ class TTSProviderBase(ABC):
             except Exception as e:
                 logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
                 return None
+    
+    def to_tts(self, text):
+        text = MarkdownCleaner.clean_markdown(text)
+        max_repeat_time = 5
+        if self.delete_audio_file:
+            # 需要删除文件的直接转为音频数据
+            while max_repeat_time > 0:
+                try:
+                    audio_bytes = asyncio.run(self.text_to_speak(text, None))
+                    if audio_bytes:
+                        audio_datas = []
+                        audio_bytes_to_data_stream(
+                            audio_bytes,
+                            file_type=self.audio_file_type,
+                            is_opus=True,
+                            callback=lambda data: audio_datas.append(data)
+                        )
+                        return audio_datas
+                    else:
+                        max_repeat_time -= 1
+                except Exception as e:
+                    logger.bind(tag=TAG).warning(
+                        f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
+                    )
+                    max_repeat_time -= 1
+            if max_repeat_time > 0:
+                logger.bind(tag=TAG).info(
+                    f"语音生成成功: {text}，重试{5 - max_repeat_time}次"
+                )
+            else:
+                logger.bind(tag=TAG).error(
+                    f"语音生成失败: {text}，请检查网络或服务是否正常"
+                )
+            return None
+        else:
+            tmp_file = self.generate_filename()
+            try:
+                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
+
+                if max_repeat_time > 0:
+                    logger.bind(tag=TAG).info(
+                        f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次"
+                    )
+                else:
+                    logger.bind(tag=TAG).error(
+                        f"语音生成失败: {text}，请检查网络或服务是否正常"
+                    )
+
+                return tmp_file
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
+                return None
 
     @abstractmethod
     async def text_to_speak(self, text, output_file):
@@ -194,7 +252,6 @@ class TTSProviderBase(ABC):
 
     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
@@ -225,7 +282,6 @@ class TTSProviderBase(ABC):
                     self.tts_text_buff = []
                     self.is_first_sentence = True
                     self.tts_audio_first_sentence = True
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()
@@ -270,103 +326,43 @@ class TTSProviderBase(ABC):
 
                 if self.conn.client_abort:
                     logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
-                    # 打断时丢弃未上报的音频数据
                     enqueue_text, enqueue_audio = None, []
                     continue
 
                 # 收到下一个文本开始或会话结束时进行上报
                 if sentence_type is not SentenceType.MIDDLE:
+                    # 重置音频流控状态（新句子开始或者结束）
+                    if hasattr(self.conn, 'audio_flow_control'):
+                        self.conn.audio_flow_control = {
+                            'last_send_time': 0,
+                            'packet_count': 0,
+                            'start_time': time.perf_counter()
+                        }
+                    
                     # 上报TTS数据
                     if enqueue_text is not None and enqueue_audio is not None:
                         enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
                     enqueue_audio = []
                     enqueue_text = text
 
-                # 计算音频数据的帧数
-                if isinstance(audio_datas, bytes):
-                    frame_count = 1  # 单个字节流作为一帧
+                # 收集上报音频数据
+                if isinstance(audio_datas, bytes) and enqueue_audio is not None:
                     enqueue_audio.append(audio_datas)
-                else:
-                    frame_count = 0
+
+                # 发送音频
+                future = asyncio.run_coroutine_threadsafe(
+                    sendAudioMessage(self.conn, sentence_type, audio_datas, text),
+                    self.conn.loop,
+                )
+                future.result()
 
                 # 记录输出和报告
                 if self.conn.max_output_size > 0 and text:
                     add_device_output(self.conn.headers.get("device-id"), len(text))
 
-                # 流控检查
-                if frame_count > 0:
-                    max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME
-                    wait_start_time = time.time()
-                    retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL
-
-                    while not self.flow_controller.can_send_frames(frame_count):
-                        # 检查是否超时或需要停止
-                        if (
-                            time.time() - wait_start_time > max_wait_time
-                            or self.conn.stop_event.is_set()
-                            or self.conn.client_abort
-                        ):
-                            logger.bind(tag=TAG).debug(
-                                "流控等待超时或收到停止信号，跳过音频发送"
-                            )
-                            break
-                        # 短暂等待后重试
-                        time.sleep(retry_interval)
-                    else:
-                        # 可以发送，记录发送的帧数
-                        self.flow_controller.record_sent_frames(frame_count)
-
-                        # 发送音频
-                        future = asyncio.run_coroutine_threadsafe(
-                            self._send_audio_with_flow_control(
-                                sentence_type, audio_datas, text
-                            ),
-                            self.conn.loop,
-                        )
-                        future.result()
-
-                        # 输出流控状态（调试用）
-                        # status = self.flow_controller.get_status()
-                        # logger.bind(tag=TAG).debug(
-                        #     f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, "
-                        #     f"可用令牌={status['available_tokens']}..."
-                        #     f"发送帧数={status['sent_frames']}..."
-                        #     f"消费帧数={status['consumed_frames']}..."
-                        #     f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..."
-                        # )
-                else:
-                    # 没有音频数据，直接发送
-                    future = asyncio.run_coroutine_threadsafe(
-                        self._send_audio_with_flow_control(
-                            sentence_type, audio_datas, text
-                        ),
-                        self.conn.loop,
-                    )
-                    future.result()
-
             except Exception as e:
                 logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}")
 
-    async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text):
-        """
-        带流控的音频发送方法 模拟设备消费音频帧的过程
-        实际应用中应该根据设备反馈来更新消费情况
-        """
-        await sendAudioMessage(self.conn, sentence_type, audio_datas, text)
-
-        # 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步
-        if isinstance(audio_datas, bytes):
-            # 模拟设备播放延迟（60ms per frame）, 实际情况可以低一点（50ms），增加使用体验
-            await asyncio.sleep(0.06)
-            self.flow_controller.update_device_consumption(1)
-
-    # 在类中添加流控制器重置方法
-    def reset_flow_controller(self):
-        """重置流控制器状态，通常在新会话开始时调用"""
-        if hasattr(self, "flow_controller"):
-            self.flow_controller.reset()
-            logger.bind(tag=TAG).info("流控制器状态已重置")
-
     async def start_session(self, session_id):
         pass
 

main/xiaozhi-server/core/providers/tts/fishspeech.py

@@ -143,8 +143,8 @@ class TTSProvider(TTSProviderBase):
         data = {
             "text": text,
             "references": [
-                ServeReferenceAudio(audio=audio if audio else b"", text=text)
-                for text, audio in zip(ref_texts, byte_audios)
+                ServeReferenceAudio(audio=audio if audio else b"", text=ref_text)
+                for ref_text, audio in zip(ref_texts, byte_audios)
             ],
             "reference_id": self.reference_id,
             "normalize": self.normalize,

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -213,7 +213,6 @@ class TTSProvider(TTSProviderBase):
 
                 if message.sentence_type == SentenceType.FIRST:
                     self.conn.client_abort = False
-                    self.reset_flow_controller()
 
                 if self.conn.client_abort:
                     try:
@@ -629,3 +628,104 @@ class TTSProvider(TTSProviderBase):
 
     def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None):
         return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback)
+
+    def to_tts(self, text: str) -> list:
+        """非流式生成音频数据，用于生成音频及测试场景
+        Args:
+            text: 要转换的文本
+        Returns:
+            list: 音频数据列表
+        """
+        try:
+            # 创建事件循环
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+            # 生成会话ID
+            session_id = uuid.uuid4().__str__().replace("-", "")
+
+            # 存储音频数据
+            audio_data = []
+
+            async def _generate_audio():
+                # 创建新的WebSocket连接
+                ws_header = {
+                    "X-Api-App-Key": self.appId,
+                    "X-Api-Access-Key": self.access_token,
+                    "X-Api-Resource-Id": self.resource_id,
+                    "X-Api-Connect-Id": uuid.uuid4(),
+                }
+                ws = await websockets.connect(
+                    self.ws_url, additional_headers=ws_header, max_size=1000000000
+                )
+
+                try:
+                    # 启动会话
+                    header = Header(
+                        message_type=FULL_CLIENT_REQUEST,
+                        message_type_specific_flags=MsgTypeFlagWithEvent,
+                        serial_method=JSON,
+                    ).as_bytes()
+                    optional = Optional(
+                        event=EVENT_StartSession, sessionId=session_id
+                    ).as_bytes()
+                    payload = self.get_payload_bytes(
+                        event=EVENT_StartSession, speaker=self.voice
+                    )
+                    await self.send_event(ws, header, optional, payload)
+
+                    # 发送文本
+                    header = Header(
+                        message_type=FULL_CLIENT_REQUEST,
+                        message_type_specific_flags=MsgTypeFlagWithEvent,
+                        serial_method=JSON,
+                    ).as_bytes()
+                    optional = Optional(
+                        event=EVENT_TaskRequest, sessionId=session_id
+                    ).as_bytes()
+                    payload = self.get_payload_bytes(
+                        event=EVENT_TaskRequest, text=text, speaker=self.voice
+                    )
+                    await self.send_event(ws, header, optional, payload)
+
+                    # 发送结束会话请求
+                    header = Header(
+                        message_type=FULL_CLIENT_REQUEST,
+                        message_type_specific_flags=MsgTypeFlagWithEvent,
+                        serial_method=JSON,
+                    ).as_bytes()
+                    optional = Optional(
+                        event=EVENT_FinishSession, sessionId=session_id
+                    ).as_bytes()
+                    payload = str.encode("{}")
+                    await self.send_event(ws, header, optional, payload)
+
+                    # 接收音频数据
+                    while True:
+                        msg = await ws.recv()
+                        res = self.parser_response(msg)
+
+                        if (
+                            res.optional.event == EVENT_TTSResponse
+                            and res.header.message_type == AUDIO_ONLY_RESPONSE
+                        ):
+                            self.wav_to_opus_data_audio_raw_stream(res.payload, callback=lambda opus_frame: audio_data.append(opus_frame))
+                        elif res.optional.event == EVENT_SessionFinished:
+                            break
+
+                finally:
+                    # 清理资源
+                    try:
+                        await ws.close()
+                    except:
+                        pass
+
+            # 运行异步任务
+            loop.run_until_complete(_generate_audio())
+            loop.close()
+
+            return audio_data
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
+            return []

main/xiaozhi-server/core/providers/tts/index_stream.py

@@ -1,8 +1,10 @@
 import os
+import time
 import queue
-import asyncio
-import traceback
 import aiohttp
+import asyncio
+import requests
+import traceback
 from config.logger import setup_logging
 from core.utils.tts import MarkdownCleaner
 from core.providers.tts.base import TTSProviderBase
@@ -45,7 +47,6 @@ class TTSProvider(TTSProviderBase):
                     self.processed_chars = 0
                     self.tts_text_buff = []
                     self.before_stop_play_files.clear()
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()
@@ -178,3 +179,57 @@ class TTSProvider(TTSProviderBase):
         await super().close()
         if hasattr(self, "opus_encoder"):
             self.opus_encoder.close()
+
+    def to_tts(self, text: str) -> list:
+        """非流式TTS处理，用于测试及保存音频文件的场景
+        Args:
+            text: 要转换的文本
+        Returns:
+            list: 返回opus编码后的音频数据列表
+        """
+        start_time = time.time()
+        text = MarkdownCleaner.clean_markdown(text)
+
+        payload = {"text": text, "character": self.character}
+
+        try:
+            with requests.post(self.api_url, json=payload, timeout=5) as response:
+                if response.status_code != 200:
+                    logger.bind(tag=TAG).error(
+                        f"TTS请求失败: {response.status_code}, {response.text}"
+                    )
+                    return []
+
+                logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
+
+                # 使用opus编码器处理PCM数据
+                opus_datas = []
+                pcm_data = response.content
+
+                # 计算每帧的字节数
+                frame_bytes = int(
+                    self.opus_encoder.sample_rate
+                    * self.opus_encoder.channels
+                    * self.opus_encoder.frame_size_ms
+                    / 1000
+                    * 2
+                )
+
+                # 分帧处理PCM数据
+                for i in range(0, len(pcm_data), frame_bytes):
+                    frame = pcm_data[i : i + frame_bytes]
+                    if len(frame) < frame_bytes:
+                        # 最后一帧可能不足，用0填充
+                        frame = frame + b"\x00" * (frame_bytes - len(frame))
+
+                    self.opus_encoder.encode_pcm_to_opus_stream(
+                        frame,
+                        end_of_stream=(i + frame_bytes >= len(pcm_data)),
+                        callback=lambda opus: opus_datas.append(opus)
+                    )
+
+                return opus_datas
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            return []

main/xiaozhi-server/core/providers/tts/linkerai.py

@@ -1,8 +1,10 @@
 import os
+import time
 import queue
-import asyncio
-import traceback
 import aiohttp
+import asyncio
+import requests
+import traceback
 from config.logger import setup_logging
 from core.utils.tts import MarkdownCleaner
 from core.providers.tts.base import TTSProviderBase
@@ -42,7 +44,6 @@ class TTSProvider(TTSProviderBase):
                     self.processed_chars = 0
                     self.tts_text_buff = []
                     self.before_stop_play_files.clear()
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()
@@ -110,10 +111,6 @@ class TTSProvider(TTSProviderBase):
         finally:
             return None
 
-    ###################################################################################
-    # linkerai单流式TTS重写父类的方法--结束
-    ###################################################################################
-
     async def text_to_speak(self, text, is_last):
         """流式处理TTS音频，每句只推送一次音频列表"""
         await self._tts_request(text, is_last)
@@ -200,3 +197,71 @@ class TTSProvider(TTSProviderBase):
         except Exception as e:
             logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
             self.tts_audio_queue.put((SentenceType.LAST, [], None))
+
+    def to_tts(self, text: str) -> list:
+        """非流式TTS处理，用于测试及保存音频文件的场景
+        Args:
+            text: 要转换的文本
+        Returns:
+            list: 返回opus编码后的音频数据列表
+        """
+        start_time = time.time()
+        text = MarkdownCleaner.clean_markdown(text)
+
+        params = {
+            "tts_text": text,
+            "spk_id": self.voice,
+            "frame_duration": 60,
+            "stream": False,
+            "target_sr": 16000,
+            "audio_format": self.audio_format,
+            "instruct_text": "请生成一段自然流畅的语音",
+        }
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+            "Content-Type": "application/json",
+        }
+
+        try:
+            with requests.get(
+                self.api_url, params=params, headers=headers, timeout=5
+            ) as response:
+                if response.status_code != 200:
+                    logger.bind(tag=TAG).error(
+                        f"TTS请求失败: {response.status_code}, {response.text}"
+                    )
+                    return []
+
+                logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
+
+                # 使用opus编码器处理PCM数据
+                opus_datas = []
+                pcm_data = response.content
+
+                # 计算每帧的字节数
+                frame_bytes = int(
+                    self.opus_encoder.sample_rate
+                    * self.opus_encoder.channels
+                    * self.opus_encoder.frame_size_ms
+                    / 1000
+                    * 2
+                )
+
+                # 分帧处理PCM数据
+                for i in range(0, len(pcm_data), frame_bytes):
+                    frame = pcm_data[i : i + frame_bytes]
+                    if len(frame) < frame_bytes:
+                        # 最后一帧可能不足，用0填充
+                        frame = frame + b"\x00" * (frame_bytes - len(frame))
+
+                    self.opus_encoder.encode_pcm_to_opus_stream(
+                        frame,
+                        end_of_stream=(i + frame_bytes >= len(pcm_data)),
+                        callback=lambda opus: opus_datas.append(opus)
+                    )
+
+                return opus_datas
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            return []

main/xiaozhi-server/core/providers/tts/paddle_speech.py

@@ -1,13 +1,15 @@
-import asyncio
-import json
-import base64
-import aiohttp
-import numpy as np
 import io
 import wave
+import json
+import base64
+import asyncio
 import websockets
-from core.providers.tts.base import TTSProviderBase
+import numpy as np
+from datetime import datetime
 from config.logger import setup_logging
+from core.providers.tts.base import TTSProviderBase
+
+
 
 TAG = __name__
 logger = setup_logging()
@@ -18,11 +20,12 @@ class TTSProvider(TTSProviderBase):
         super().__init__(config, delete_audio_file)
         self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming")
         self.protocol = config.get("protocol", "websocket")
+        
         if config.get("private_voice"):
             self.spk_id = int(config.get("private_voice"))
         else:
-            self.spk_id = int(config.get("spk_id", "0"))  
-        
+            self.spk_id = int(config.get("spk_id", "0"))
+            
         sample_rate = config.get("sample_rate", 24000)
         self.sample_rate = float(sample_rate) if sample_rate else 24000
         
@@ -32,7 +35,21 @@ class TTSProvider(TTSProviderBase):
         volume = config.get("volume", 1.0)
         self.volume = float(volume) if volume else 1.0
         
-        self.save_path = config.get("save_path", "./streaming_tts.wav")
+        self.delete_audio_file = config.get("delete_audio", True)
+        if not self.delete_audio_file:
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            save_path = config.get("save_path")
+            if save_path:
+                if not save_path.endswith('.wav'):
+                    save_path = f"{save_path}_{timestamp}.wav"
+                else:
+                    other_path = save_path[:-4]
+                    save_path = f"{other_path}_{timestamp}.wav"
+                self.save_path = save_path
+            else:
+                self.save_path = f"./streaming_tts_{timestamp}.wav"
+        else:
+            self.save_path = None
 
     async def pcm_to_wav(self, pcm_data: bytes, sample_rate: int = 24000, num_channels: int = 1,
                          bits_per_sample: int = 16) -> bytes:
@@ -58,43 +75,9 @@ class TTSProvider(TTSProviderBase):
     async def text_to_speak(self, text, output_file):
         if self.protocol == "websocket":
             return await self.text_streaming(text, output_file)
-        elif self.protocol == "http":
-            return await self.text(text, output_file)
         else:
             raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.")
 
-    async def text(self, text, output_file):
-        request_json = {
-            "text": text,
-            "spk_id": self.spk_id,
-            "speed": self.speed,
-            "volume": self.volume,
-            "sample_rate": self.sample_rate,
-            "save_path": self.save_path
-        }
-
-        try:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(self.url, json=request_json) as resp:
-                    if resp.status == 200:
-                        resp_json = await resp.json()
-                        if resp_json.get("success"):
-                            data = resp_json["result"]
-                            audio_bytes = base64.b64decode(data["audio"])
-                            if output_file:
-                                with open(output_file, "wb") as file_to_save:
-                                    file_to_save.write(audio_bytes)
-                            else:
-                                return audio_bytes
-                        else:
-                            raise Exception(
-                                f"Error: {resp_json.get('message', 'Unknown error')} while processing text: {text}")
-                    else:
-                        raise Exception(
-                            f"HTTP Error: {resp.status} - {await resp.text()} while processing text: {text}")
-        except Exception as e:
-            raise Exception(f"Error during TTS HTTP request: {e} while processing text: {text}")
-
     async def text_streaming(self, text, output_file):
         try:
             # 使用 websockets 异步连接到 WebSocket 服务器
@@ -151,6 +134,12 @@ class TTSProvider(TTSProviderBase):
                 # 接收结束响应避免服务抛出异常
                 await ws.recv()
 
+                # 根据配置决定是否保存文件
+                if not self.delete_audio_file and self.save_path:
+                    with open(self.save_path, "wb") as f:
+                        f.write(wav_data)
+                    logger.bind(tag=TAG).info(f"音频文件已保存到: {self.save_path}")
+                
                 # 返回或保存音频数据
                 if output_file:
                     with open(output_file, "wb") as file_to_save:
@@ -159,4 +148,4 @@ class TTSProvider(TTSProviderBase):
                     return wav_data
 
         except Exception as e:
-            raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")
+            raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")