Merge branch 'py_AliBL_TTS' of https://github.com/xinnan-tech/xiaozhi-esp32-server

main/manager-api/src/main/resources/db/changelog/202509161701.sql

@@ -0,0 +1,54 @@
+-- 添加阿里百炼流式TTS 供应器
+delete from `ai_model_provider` where id = 'SYSTEM_TTS_AliBLStreamTTS';
+INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
+('SYSTEM_TTS_AliBLStreamTTS', 'TTS', 'alibl_stream', '阿里百炼流式语音合成', '[{"key":"api_key","label":"API密钥","type":"string"},{"key":"output_dir","label":"输出目录","type":"string"},{"key":"model","label":"模型","type":"string"},{"key":"voice","label":"音色","type":"string"},{"key":"format","label":"音频格式","type":"string"},{"key":"sample_rate","label":"采样率","type":"number"},{"key": "volume", "type": "number", "label": "音量"},{"key": "rate", "type": "number", "label": "语速"},{"key": "pitch", "type": "number", "label": "音调"}]', 19, 1, NOW(), 1, NOW());
+
+-- 添加阿里百炼流式TTS模型配置
+delete from `ai_model_config` where id = 'TTS_AliBLStreamTTS';
+INSERT INTO `ai_model_config` VALUES ('TTS_AliBLStreamTTS', 'TTS', 'AliBLStreamTTS', '阿里百炼流式语音合成', 0, 1, '{\"type\": \"alibl_stream\", \"appkey\": \"\", \"output_dir\": \"tmp/\", \"model\": \"cosyvoice-v2\", \"voice\": \"longcheng_v2\", \"format\": \"pcm\", \"sample_rate\": 24000, \"volume\": 50, \"rate\": 1, \"pitch\": 1}', NULL, NULL, 22, NULL, NULL, NULL, NULL);
+
+-- 更新阿里百炼流式TTS配置说明
+UPDATE `ai_model_config` SET 
+`doc_link` = 'https://bailian.console.aliyun.com/?apiKey=1#/api-key',
+`remark` = '阿里百炼流式TTS说明：
+1. 访问 https://bailian.console.aliyun.com/?apiKey=1#/api-key 创建项目并获取appkey
+2. 支持实时流式合成，具有较低的延迟
+3. 支持多种音色设置和音频参数调整
+4. 支持CosyVoice-V3大模型音色，价格实惠(0.4元/万字符)
+5. 支持实时调节音量、语速、音调等参数
+6. 如果需要使用CosyVoice-V3模型和一些限制类型的音色，需要联系阿里百炼客服申请
+' WHERE `id` = 'TTS_AliBLStreamTTS';
+
+-- 添加阿里百炼流式TTS音色
+delete from `ai_tts_voice` where tts_model_id = 'TTS_AliBLStreamTTS';
+
+-- 语音助手
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0001', 'TTS_AliBLStreamTTS', '龙小淳-知性积极女', 'longxiaochun_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 1, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0002', 'TTS_AliBLStreamTTS', '龙小夏-沉稳权威女', 'longxiaoxia_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 2, NULL, NULL, NULL, NULL);
+
+-- 直播带货
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0003', 'TTS_AliBLStreamTTS', '龙安燃-活泼质感女', 'longanran', '中文及中英文混合', NULL, NULL, NULL, NULL, 3, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0004', 'TTS_AliBLStreamTTS', '龙安宣-经典直播女', 'longanxuan', '中文及中英文混合', NULL, NULL, NULL, NULL, 4, NULL, NULL, NULL, NULL);
+
+-- 社交陪伴
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0005', 'TTS_AliBLStreamTTS', '龙寒-温暖痴情男', 'longhan_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 5, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0006', 'TTS_AliBLStreamTTS', '龙颜-温暖春风女', 'longyan_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 6, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0007', 'TTS_AliBLStreamTTS', '龙菲菲-甜美矫情女', 'longfeifei_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 7, NULL, NULL, NULL, NULL);
+
+-- 方言
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0008', 'TTS_AliBLStreamTTS', '龙老铁-东北直率男', 'longlaotie_v2', '中文(东北)及中英文混合', NULL, NULL, NULL, NULL, 8, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0009', 'TTS_AliBLStreamTTS', '龙嘉怡-知性粤语女', 'longjiayi_v2', '中文(粤语)及中英文混合', NULL, NULL, NULL, NULL, 9, NULL, NULL, NULL, NULL);
+
+-- 童声
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0010', 'TTS_AliBLStreamTTS', '龙杰力豆-阳光顽皮男', 'longjielidou_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 10, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0011', 'TTS_AliBLStreamTTS', '龙铃-稚气呆板女', 'longling_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 11, NULL, NULL, NULL, NULL);
+
+-- 诗歌朗诵
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0012', 'TTS_AliBLStreamTTS', '李白-古代诗仙男', 'libai_v2', '中文及中英文混合', NULL, NULL, NULL, NULL, 12, NULL, NULL, NULL, NULL);
+
+-- 出海营销
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0013', 'TTS_AliBLStreamTTS', 'loongeva-知性英文女', 'loongeva_v2', '英式英文', NULL, NULL, NULL, NULL, 13, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0014', 'TTS_AliBLStreamTTS', 'loongbrian-沉稳英文男', 'loongbrian_v2', '英式英文', NULL, NULL, NULL, NULL, 14, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0015', 'TTS_AliBLStreamTTS', 'loongkyong-韩语女', 'loongkyong_v2', '韩语', NULL, NULL, NULL, NULL, 15, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0016', 'TTS_AliBLStreamTTS', 'loongtomoka-日语女', 'loongtomoka_v2', '日语', NULL, NULL, NULL, NULL, 16, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliBLStreamTTS_0017', 'TTS_AliBLStreamTTS', 'loongtomoya-日语男', 'loongtomoya_v2', '日语', NULL, NULL, NULL, NULL, 17, NULL, NULL, NULL, NULL);

main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -345,3 +345,10 @@ databaseChangeLog:
         - sqlFile:
             encoding: utf8
             path: classpath:db/changelog/202509080922.sql
+  - changeSet:
+      id: 202509161701
+      author: RanChen
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202509161701.sql

main/xiaozhi-server/config.yaml

@@ -922,3 +922,18 @@ TTS:
     # 默认音色，如需其他音色可到项目assets文件夹下注册
     voice: "jay_klee"
     output_dir: tmp/
+  AliBLTTS:
+    # 阿里百炼CosyVoice大模型流式文本语音合成
+    # 可在这里找到你的 api_key https://bailian.console.aliyun.com/?apiKey=1#/api-key
+    # cosyvoice-v3和部分音色需要申请开通
+    type: alibl_stream
+    api_key: 你的api_key
+    model: "cosyvoice-v2"
+    voice: "longcheng_v2"
+    output_dir: tmp/
+    # 以下可不用设置，使用默认设置
+    # format: pcm  # 音频格式：pcm、wav、mp3、opus
+    # sample_rate: 24000  # 采样率：16000, 24000, 48000
+    # volume: 50  # 音量：0-100
+    # rate: 1  # 语速：0.5~2
+    # pitch: 1  # 语调：0.5~2

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -114,6 +114,9 @@ async def sendAudio(conn, audios, frame_duration=60):
         delay = expected_time - current_time
         if delay > 0:
             await asyncio.sleep(delay)
+        else:
+            # 纠正误差
+            flow_control["start_time"] += abs(delay)
 
         if conn.conn_from_mqtt_gateway:
             # 计算时间戳和序列号

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

@@ -0,0 +1,522 @@
+import os
+import uuid
+import json
+import time
+import queue
+import asyncio
+import traceback
+import websockets
+from asyncio import Task
+from config.logger import setup_logging
+from core.utils import opus_encoder_utils
+from core.utils.tts import MarkdownCleaner
+from core.providers.tts.base import TTSProviderBase
+from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
+
+TAG = __name__
+logger = setup_logging()
+
+
+class TTSProvider(TTSProviderBase):
+    def __init__(self, config, delete_audio_file):
+        super().__init__(config, delete_audio_file)
+
+        self.interface_type = InterfaceType.DUAL_STREAM
+        # 基础配置
+        self.api_key = config.get("api_key")
+        if not self.api_key:
+            raise ValueError("api_key is required for CosyVoice TTS")
+
+        # WebSocket配置
+        self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
+        self.ws = None
+        self._monitor_task = None
+        self.last_active_time = None
+
+        # 模型和音色配置
+        self.model = config.get("model", "cosyvoice-v2")
+        self.voice = config.get("voice", "longxiaochun_v2")  # 默认音色
+        if config.get("private_voice"):
+            self.voice = config.get("private_voice")
+
+        # 音频参数配置
+        self.format = config.get("format", "pcm")
+        sample_rate = config.get("sample_rate", "24000")
+        self.sample_rate = int(sample_rate) if sample_rate else 24000
+
+        volume = config.get("volume", "50")
+        self.volume = int(volume) if volume else 50
+
+        rate = config.get("rate", "1.0")
+        self.rate = float(rate) if rate else 1.0
+
+        pitch = config.get("pitch", "1.0")
+        self.pitch = float(pitch) if pitch else 1.0
+
+        self.header = {
+            "Authorization": f"Bearer {self.api_key}",
+            # "user-agent": "your_platform_info", // 可选
+            # "X-DashScope-WorkSpace": workspace, // 可选，阿里云百炼业务空间ID
+            "X-DashScope-DataInspection": "enable",
+        }
+
+        # 创建Opus编码器
+        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+            sample_rate=self.sample_rate, channels=1, frame_size_ms=60
+        )
+
+    async def _ensure_connection(self):
+        """确保WebSocket连接可用，支持60秒内连接复用"""
+        try:
+            current_time = time.time()
+            if self.ws and current_time - self.last_active_time < 60:
+                # 一分钟内才可以复用链接进行连续对话
+                logger.bind(tag=TAG).info(f"使用已有链接...")
+                return self.ws
+            logger.bind(tag=TAG).info("开始建立新连接...")
+
+            self.ws = await websockets.connect(
+                self.ws_url,
+                additional_headers=self.header,
+                ping_interval=30,
+                ping_timeout=10,
+                close_timeout=10,
+            )
+
+            logger.bind(tag=TAG).info("WebSocket连接建立成功")
+            self.last_active_time = current_time
+            return self.ws
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
+            self.ws = None
+            self.last_active_time = None
+            raise
+
+    def tts_text_priority_thread(self):
+        """流式TTS文本处理线程"""
+        while not self.conn.stop_event.is_set():
+            try:
+                message = self.tts_text_queue.get(timeout=1)
+                logger.bind(tag=TAG).debug(
+                    f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
+                )
+
+                if message.sentence_type == SentenceType.FIRST:
+                    self.conn.client_abort = False
+
+                if self.conn.client_abort:
+                    try:
+                        logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
+                        continue
+                    except Exception as e:
+                        logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}")
+                        continue
+
+                if message.sentence_type == SentenceType.FIRST:
+                    # 初始化会话
+                    try:
+                        if not getattr(self.conn, "sentence_id", None): 
+                            self.conn.sentence_id = uuid.uuid4().hex
+                            logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}")
+
+                        logger.bind(tag=TAG).info("开始启动TTS会话...")
+                        future = asyncio.run_coroutine_threadsafe(
+                            self.start_session(self.conn.sentence_id),
+                            loop=self.conn.loop,
+                        )
+                        future.result()
+                        self.before_stop_play_files.clear()
+                        logger.bind(tag=TAG).info("TTS会话启动成功")
+                    except Exception as e:
+                        logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
+                        continue
+
+                elif ContentType.TEXT == message.content_type:
+                    if message.content_detail:
+                        try:
+                            logger.bind(tag=TAG).debug(
+                                f"开始发送TTS文本: {message.content_detail}"
+                            )
+                            future = asyncio.run_coroutine_threadsafe(
+                                self.text_to_speak(message.content_detail, None),
+                                loop=self.conn.loop,
+                            )
+                            future.result()
+                            logger.bind(tag=TAG).debug("TTS文本发送成功")
+                        except Exception as e:
+                            logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
+                            continue
+
+                elif ContentType.FILE == message.content_type:
+                    logger.bind(tag=TAG).info(
+                        f"添加音频文件到待播放列表: {message.content_file}"
+                    )
+                    if message.content_file and os.path.exists(message.content_file):
+                        # 先处理文件音频数据
+                        self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
+
+                if message.sentence_type == SentenceType.LAST:
+                    try:
+                        logger.bind(tag=TAG).info("开始结束TTS会话...")
+                        future = asyncio.run_coroutine_threadsafe(
+                            self.finish_session(self.conn.sentence_id),
+                            loop=self.conn.loop,
+                        )
+                        future.result()
+                    except Exception as e:
+                        logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
+                        continue
+
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logger.bind(tag=TAG).error(
+                    f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
+                )
+                continue
+
+    async def text_to_speak(self, text, _):
+        """发送文本到TTS服务进行合成"""
+        try:
+            if self.ws is None:
+                logger.bind(tag=TAG).warning("WebSocket连接不存在，终止发送文本")
+                return
+
+            # 过滤Markdown
+            filtered_text = MarkdownCleaner.clean_markdown(text)
+
+            # 发送continue-task消息
+            continue_task_message = {
+                "header": {
+                    "action": "continue-task",
+                    "task_id": self.conn.sentence_id,
+                    "streaming": "duplex",
+                },
+                "payload": {"input": {"text": filtered_text}},
+            }
+
+            await self.ws.send(json.dumps(continue_task_message))
+            self.last_active_time = time.time()
+            logger.bind(tag=TAG).debug(f"已发送文本: {filtered_text}")
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
+            if self.ws:
+                try:
+                    await self.ws.close()
+                except:
+                    pass
+                self.ws = None
+            raise
+
+    async def start_session(self, session_id):
+        """启动TTS会话"""
+        logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
+        try:
+            # 检查并清理上一个会话的监听任务
+            if (
+                self._monitor_task is not None
+                and isinstance(self._monitor_task, Task)
+                and not self._monitor_task.done()
+            ):
+                logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务...")
+                await self.close()
+
+            # 确保连接可用
+            await self._ensure_connection()
+
+            # 启动监听任务
+            self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
+
+            # 发送run-task消息启动会话
+            run_task_message = {
+                "header": {
+                    "action": "run-task",
+                    "task_id": session_id,
+                    "streaming": "duplex",
+                },
+                "payload": {
+                    "task_group": "audio",
+                    "task": "tts",
+                    "function": "SpeechSynthesizer",
+                    "model": self.model,
+                    "parameters": {
+                        "text_type": "PlainText",
+                        "voice": self.voice,
+                        "format": self.format,
+                        "sample_rate": self.sample_rate,
+                        "volume": self.volume,
+                        "rate": self.rate,
+                        "pitch": self.pitch,
+                    },
+                    "input": {}
+                },
+            }
+
+            await self.ws.send(json.dumps(run_task_message))
+            self.last_active_time = time.time()
+            logger.bind(tag=TAG).info("会话启动请求已发送")
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
+            await self.close()
+            raise
+
+    async def finish_session(self, session_id):
+        """结束TTS会话"""
+        logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
+        try:
+            if self.ws and session_id:
+                # 发送finish-task消息
+                finish_task_message = {
+                    "header": {
+                        "action": "finish-task",
+                        "task_id": session_id,
+                        "streaming": "duplex",
+                    },
+                    "payload": {
+                        "input": {}
+                    }
+                }
+
+                await self.ws.send(json.dumps(finish_task_message))
+                self.last_active_time = time.time()
+                logger.bind(tag=TAG).info("会话结束请求已发送")
+                # 等待监听任务完成
+                if self._monitor_task:
+                    try:
+                        await self._monitor_task
+                    except Exception as e:
+                        logger.bind(tag=TAG).error(
+                            f"等待监听任务完成时发生错误: {str(e)}"
+                        )
+                    finally:
+                        self._monitor_task = None
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
+            await self.close()
+            raise
+
+    async def close(self):
+        """清理资源"""
+        # 取消监听任务
+        if self._monitor_task:
+            try:
+                self._monitor_task.cancel()
+                await self._monitor_task
+            except asyncio.CancelledError:
+                pass
+            except Exception as e:
+                logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}")
+            self._monitor_task = None
+
+        # 关闭WebSocket连接
+        if self.ws:
+            try:
+                await self.ws.close()
+            except:
+                pass
+            self.ws = None
+            self.last_active_time = None
+
+    async def _start_monitor_tts_response(self):
+        """监听TTS响应"""
+        try:
+            session_finished = False
+            while not self.conn.stop_event.is_set():
+                try:
+                    msg = await self.ws.recv()
+                    self.last_active_time = time.time()
+
+                    # 检查客户端是否中止
+                    if self.conn.client_abort:
+                        logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应")
+                        break
+
+                    if isinstance(msg, str):  # JSON控制消息
+                        try:
+                            data = json.loads(msg)
+                            event = data["header"].get("event")
+
+                            if event == "task-started":
+                                logger.bind(tag=TAG).debug("TTS任务启动成功~")
+                                self.tts_audio_queue.put((SentenceType.FIRST, [], None))
+                            elif event == "result-generated":
+                                # 发送缓存的数据
+                                if self.conn.tts_MessageText:
+                                    logger.bind(tag=TAG).info(
+                                        f"句子语音生成成功： {self.conn.tts_MessageText}"
+                                    )
+                                    self.tts_audio_queue.put(
+                                        (SentenceType.FIRST, [], self.conn.tts_MessageText)
+                                    )
+                                    self.conn.tts_MessageText = None
+                            elif event == "task-finished":
+                                logger.bind(tag=TAG).debug("TTS任务完成~")
+                                self._process_before_stop_play_files()
+                                session_finished = True
+                                break
+                            elif event == "task-failed":
+                                error_code = data["header"].get("error_code", "unknown")
+                                error_message = data["header"].get("error_message", "未知错误")
+                                logger.bind(tag=TAG).error(
+                                    f"TTS任务失败: {error_code} - {error_message}"
+                                )
+                                break
+                        except json.JSONDecodeError:
+                            logger.bind(tag=TAG).warning("收到无效的JSON消息")
+                    elif isinstance(msg, (bytes, bytearray)):
+                        self.opus_encoder.encode_pcm_to_opus_stream(
+                            msg, False, callback=self.handle_opus
+                        )
+                except websockets.ConnectionClosed:
+                    logger.bind(tag=TAG).warning("WebSocket连接已关闭")
+                    break
+                except Exception as e:
+                    logger.bind(tag=TAG).error(
+                        f"处理TTS响应时出错: {e}\n{traceback.format_exc()}"
+                    )
+                    break
+
+            # 仅在连接异常且非正常结束时才关闭连接
+            if not session_finished and self.ws:
+                try:
+                    await self.ws.close()
+                except:
+                    pass
+                self.ws = None
+        # 监听任务退出时清理引用
+        finally:
+            self._monitor_task = None
+
+    def to_tts(self, text: str) -> list:
+        """非流式生成音频数据，用于生成音频及测试场景"""
+        try:
+            # 创建事件循环
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+            # 生成会话ID
+            session_id = uuid.uuid4().hex
+            # 存储音频数据
+            audio_data = []
+
+            async def _generate_audio():
+                ws = await websockets.connect(
+                    self.ws_url,
+                    additional_headers=self.header,
+                    ping_interval=30,
+                    ping_timeout=10,
+                    close_timeout=10,
+                    max_size=10 * 1024 * 1024,
+                )
+
+                try:
+                    # 发送run-task消息启动会话
+                    run_task_message = {
+                        "header": {
+                            "action": "run-task",
+                            "task_id": session_id,
+                            "streaming": "duplex",
+                        },
+                        "payload": {
+                            "task_group": "audio",
+                            "task": "tts",
+                            "function": "SpeechSynthesizer",
+                            "model": self.model,
+                            "parameters": {
+                                "text_type": "PlainText",
+                                "voice": self.voice,
+                                "format": self.format,
+                                "sample_rate": self.sample_rate,
+                                "volume": self.volume,
+                                "rate": self.rate,
+                                "pitch": self.pitch,
+                            },
+                            "input": {}
+                        },
+                    }
+                    await ws.send(json.dumps(run_task_message))
+
+                    # 等待任务启动
+                    task_started = False
+                    while not task_started:
+                        msg = await ws.recv()
+                        if isinstance(msg, str):
+                            data = json.loads(msg)
+                            header = data.get("header", {})
+                            if header.get("event") == "task-started":
+                                task_started = True
+                                logger.bind(tag=TAG).debug("TTS任务已启动")
+                            elif header.get("event") == "task-failed":
+                                error_code = header.get("error_code", "unknown")
+                                error_message = header.get("error_message", "未知错误")
+                                raise Exception(
+                                    f"启动任务失败: {error_code} - {error_message}"
+                                )
+
+                    # 发送文本
+                    filtered_text = MarkdownCleaner.clean_markdown(text)
+                    # 发送continue-task消息
+                    continue_task_message = {
+                        "header": {
+                            "action": "continue-task",
+                            "task_id": session_id,
+                            "streaming": "duplex",
+                        },
+                        "payload": {"input": {"text": filtered_text}},
+                    }
+                    await ws.send(json.dumps(continue_task_message))
+
+                    # 发送finish-task消息
+                    finish_task_message = {
+                        "header": {
+                            "action": "finish-task",
+                            "task_id": session_id,
+                            "streaming": "duplex",
+                        },
+                        "payload": {
+                            "input": {}
+                        }
+                    }
+                    await ws.send(json.dumps(finish_task_message))
+
+                    # 接收音频数据
+                    task_finished = False
+                    while not task_finished:
+                        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", {})
+                            if header.get("event") == "task-finished":
+                                task_finished = True
+                                logger.bind(tag=TAG).debug("TTS任务完成")
+                            elif header.get("event") == "task-failed":
+                                error_code = header.get("error_code", "unknown")
+                                error_message = header.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/aliyun_stream.py

@@ -457,7 +457,6 @@ class TTSProvider(TTSProviderBase):
                             logger.bind(tag=TAG).warning("收到无效的JSON消息")
                     # 二进制消息（音频数据）
                     elif isinstance(msg, (bytes, bytearray)):
-                        logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
                         self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus)
                 except websockets.ConnectionClosed:
                     logger.bind(tag=TAG).warning("WebSocket连接已关闭")

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

@@ -331,15 +331,6 @@ class TTSProviderBase(ABC):
 
                 # 收到下一个文本开始或会话结束时进行上报
                 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(),
-                            'sequence': 0  # 添加序列号
-                        }
-                    
                     # 上报TTS数据
                     if enqueue_text is not None and enqueue_audio is not None:
                         enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)

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

@@ -449,7 +449,6 @@ class TTSProvider(TTSProviderBase):
                         res.optional.event == EVENT_TTSResponse
                         and res.header.message_type == AUDIO_ONLY_RESPONSE
                     ):
-                        logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
                         self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
                     elif res.optional.event == EVENT_TTSSentenceEnd:
                         logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")