Merge branch 'py_test_tts' into openai-asr

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

@@ -1,6 +1,6 @@
 -- 添加百度ASR模型配置
 delete from `ai_model_config` where `id` = 'ASR_BaiduASR';
-INSERT INTO `ai_model_config` VALUES ('ASR_BaiduASR', 'ASR', 'BaiduASR', '百度语音识别', 0, 1, '{\"type\": \"baidu\", \"app_id\": \"\", \"api_key\": \"\", \"secret_key\": \"\", \"dev_pid\": 1537, \"output_dir\": \"tmp/\"}', NULL, NULL, 7, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_model_config` VALUES ('ASR_BaiduASR', 'ASR', 'BaiduASR', '百度语音识别', 0, 1, '{\"type\": \"baidu\", \"app_id\": \"\", \"api_key\": \"\", \"secret_key\": \"\", \"dev_pid\": 1537, \"output_dir\": \"tmp/\"}', NULL, NULL, 8, NULL, NULL, NULL, NULL);
 
 
 -- 添加百度ASR供应器

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

@@ -0,0 +1,26 @@
+-- 添加阿里云流式ASR供应器
+delete from `ai_model_provider` where id = 'SYSTEM_ASR_AliyunStreamASR';
+INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
+('SYSTEM_ASR_AliyunStreamASR', 'ASR', 'aliyun_stream', '阿里云语音识别(流式)', '[{"key":"appkey","label":"应用AppKey","type":"string"},{"key":"token","label":"临时Token","type":"string"},{"key":"access_key_id","label":"AccessKey ID","type":"string"},{"key":"access_key_secret","label":"AccessKey Secret","type":"string"},{"key":"host","label":"服务地址","type":"string"},{"key":"max_sentence_silence","label":"断句检测时间","type":"number"},{"key":"output_dir","label":"输出目录","type":"string"}]', 6, 1, NOW(), 1, NOW());
+
+-- 添加阿里云流式ASR模型配置
+delete from `ai_model_config` where id = 'ASR_AliyunStreamASR';
+INSERT INTO `ai_model_config` VALUES ('ASR_AliyunStreamASR', 'ASR', 'AliyunStreamASR', '阿里云语音识别(流式)', 0, 1, '{\"type\": \"aliyun_stream\", \"appkey\": \"\", \"token\": \"\", \"access_key_id\": \"\", \"access_key_secret\": \"\", \"host\": \"nls-gateway-cn-shanghai.aliyuncs.com\", \"max_sentence_silence\": 800, \"output_dir\": \"tmp/\"}', NULL, NULL, 7, NULL, NULL, NULL, NULL);
+
+-- 更新阿里云流式ASR配置说明
+UPDATE `ai_model_config` SET 
+`doc_link` = 'https://nls-portal.console.aliyun.com/',
+`remark` = '阿里云流式ASR配置说明：
+1. 阿里云ASR和阿里云(流式)ASR的区别是：阿里云ASR是一次性识别，阿里云(流式)ASR是实时流式识别
+2. 流式ASR具有更低的延迟和更好的实时性，适合语音交互场景
+3. 需要在阿里云智能语音交互控制台创建应用并获取认证信息
+4. 支持中文实时语音识别，支持标点符号预测和逆文本规范化
+5. 需要网络连接，输出文件保存在tmp/目录
+申请步骤：
+1. 访问 https://nls-portal.console.aliyun.com/ 开通智能语音交互服务
+2. 访问 https://nls-portal.console.aliyun.com/applist 创建项目并获取appkey
+3. 访问 https://nls-portal.console.aliyun.com/overview 获取临时token（或配置access_key_id和access_key_secret自动获取）
+4. 如需动态token管理，建议配置access_key_id和access_key_secret
+5. max_sentence_silence参数控制断句检测时间（毫秒），默认800ms
+如需了解更多参数配置，请参考：https://help.aliyun.com/zh/isi/developer-reference/real-time-speech-recognition
+' WHERE `id` = 'ASR_AliyunStreamASR';

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

@@ -0,0 +1,56 @@
+-- 添加阿里云流式TTS供应器
+delete from `ai_model_provider` where id = 'SYSTEM_TTS_AliyunStreamTTS';
+INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
+('SYSTEM_TTS_AliyunStreamTTS', 'TTS', 'aliyun_stream', '阿里云语音合成(流式)', '[{"key":"appkey","label":"应用AppKey","type":"string"},{"key":"token","label":"临时Token","type":"string"},{"key":"access_key_id","label":"AccessKey ID","type":"string"},{"key":"access_key_secret","label":"AccessKey Secret","type":"string"},{"key":"host","label":"服务地址","type":"string"},{"key":"voice","label":"默认音色","type":"string"},{"key":"format","label":"音频格式","type":"string"},{"key":"sample_rate","label":"采样率","type":"number"},{"key":"volume","label":"音量","type":"number"},{"key":"speech_rate","label":"语速","type":"number"},{"key":"pitch_rate","label":"音调","type":"number"},{"key":"output_dir","label":"输出目录","type":"string"}]', 15, 1, NOW(), 1, NOW());
+
+-- 添加阿里云流式TTS模型配置
+delete from `ai_model_config` where id = 'TTS_AliyunStreamTTS';
+INSERT INTO `ai_model_config` VALUES ('TTS_AliyunStreamTTS', 'TTS', 'AliyunStreamTTS', '阿里云语音合成(流式)', 0, 1, '{\"type\": \"aliyun_stream\", \"appkey\": \"\", \"token\": \"\", \"access_key_id\": \"\", \"access_key_secret\": \"\", \"host\": \"nls-gateway-cn-beijing.aliyuncs.com\", \"voice\": \"longxiaochun\", \"format\": \"pcm\", \"sample_rate\": 16000, \"volume\": 50, \"speech_rate\": 0, \"pitch_rate\": 0, \"output_dir\": \"tmp/\"}', NULL, NULL, 18, NULL, NULL, NULL, NULL);
+
+-- 更新阿里云流式TTS配置说明
+UPDATE `ai_model_config` SET 
+`doc_link` = 'https://nls-portal.console.aliyun.com/',
+`remark` = '阿里云流式TTS配置说明：
+1. 阿里云TTS和阿里云(流式)TTS的区别是：阿里云TTS是一次性合成，阿里云(流式)TTS是实时流式合成
+2. 流式TTS具有更低的延迟和更好的实时性，适合语音交互场景
+3. 需要在阿里云智能语音交互控制台创建应用并获取认证信息
+4. 支持CosyVoice大模型音色，音质更加自然
+5. 支持实时调节音量、语速、音调等参数
+申请步骤：
+1. 访问 https://nls-portal.console.aliyun.com/ 开通智能语音交互服务
+2. 访问 https://nls-portal.console.aliyun.com/applist 创建项目并获取appkey
+3. 访问 https://nls-portal.console.aliyun.com/overview 获取临时token（或配置access_key_id和access_key_secret自动获取）
+4. 如需动态token管理，建议配置access_key_id和access_key_secret
+5. 可选择北京、上海等不同地域的服务器以优化延迟
+6. voice参数支持CosyVoice大模型音色，如longxiaochun、longyueyue等
+如需了解更多参数配置，请参考：https://help.aliyun.com/zh/isi/developer-reference/real-time-speech-synthesis
+' WHERE `id` = 'TTS_AliyunStreamTTS';
+
+-- 添加阿里云流式TTS音色
+delete from `ai_tts_voice` where tts_model_id = 'TTS_AliyunStreamTTS';
+-- 温柔女声系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0001', 'TTS_AliyunStreamTTS', '龙小淳-温柔姐姐', 'longxiaochun', '中文及中英文混合', NULL, NULL, NULL, NULL, 1, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0002', 'TTS_AliyunStreamTTS', '龙小夏-温柔女声', 'longxiaoxia', '中文及中英文混合', NULL, NULL, NULL, NULL, 2, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0003', 'TTS_AliyunStreamTTS', '龙玫-温柔女声', 'longmei', '中文及中英文混合', NULL, NULL, NULL, NULL, 3, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0004', 'TTS_AliyunStreamTTS', '龙瑰-温柔女声', 'longgui', '中文及中英文混合', NULL, NULL, NULL, NULL, 4, NULL, NULL, NULL, NULL);
+-- 御姐女声系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0005', 'TTS_AliyunStreamTTS', '龙玉-御姐女声', 'longyu', '中文及中英文混合', NULL, NULL, NULL, NULL, 5, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0006', 'TTS_AliyunStreamTTS', '龙娇-御姐女声', 'longjiao', '中文及中英文混合', NULL, NULL, NULL, NULL, 6, NULL, NULL, NULL, NULL);
+-- 男声系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0007', 'TTS_AliyunStreamTTS', '龙臣-译制片男声', 'longchen', '中文及中英文混合', NULL, NULL, NULL, NULL, 7, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0008', 'TTS_AliyunStreamTTS', '龙修-青年男声', 'longxiu', '中文及中英文混合', NULL, NULL, NULL, NULL, 8, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0009', 'TTS_AliyunStreamTTS', '龙橙-阳光男声', 'longcheng', '中文及中英文混合', NULL, NULL, NULL, NULL, 9, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0010', 'TTS_AliyunStreamTTS', '龙哲-成熟男声', 'longzhe', '中文及中英文混合', NULL, NULL, NULL, NULL, 10, NULL, NULL, NULL, NULL);
+-- 专业播报系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0011', 'TTS_AliyunStreamTTS', 'Bella2.0-新闻女声', 'loongbella', '中文及中英文混合', NULL, NULL, NULL, NULL, 11, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0012', 'TTS_AliyunStreamTTS', 'Stella2.0-飒爽女声', 'loongstella', '中文及中英文混合', NULL, NULL, NULL, NULL, 12, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0013', 'TTS_AliyunStreamTTS', '龙书-新闻男声', 'longshu', '中文及中英文混合', NULL, NULL, NULL, NULL, 13, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0014', 'TTS_AliyunStreamTTS', '龙婧-严肃女声', 'longjing', '中文及中英文混合', NULL, NULL, NULL, NULL, 14, NULL, NULL, NULL, NULL);
+-- 特色音色系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0015', 'TTS_AliyunStreamTTS', '龙奇-活泼童声', 'longqi', '中文及中英文混合', NULL, NULL, NULL, NULL, 15, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0016', 'TTS_AliyunStreamTTS', '龙华-活泼女童', 'longhua', '中文及中英文混合', NULL, NULL, NULL, NULL, 16, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0017', 'TTS_AliyunStreamTTS', '龙无-无厘头男声', 'longwu', '中文及中英文混合', NULL, NULL, NULL, NULL, 17, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0018', 'TTS_AliyunStreamTTS', '龙大锤-幽默男声', 'longdachui', '中文及中英文混合', NULL, NULL, NULL, NULL, 18, NULL, NULL, NULL, NULL);
+-- 粤语系列
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0019', 'TTS_AliyunStreamTTS', '龙嘉怡-粤语女声', 'longjiayi', '粤语及粤英混合', NULL, NULL, NULL, NULL, 19, NULL, NULL, NULL, NULL);
+INSERT INTO `ai_tts_voice` VALUES ('TTS_AliyunStreamTTS_0020', 'TTS_AliyunStreamTTS', '龙桃-粤语女声', 'longtao', '粤语及粤英混合', NULL, NULL, NULL, NULL, 20, NULL, NULL, NULL, NULL);

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

@@ -101,12 +101,12 @@ databaseChangeLog:
             encoding: utf8
             path: classpath:db/changelog/202505022134.sql
   - changeSet:
-      id: 202505081146
+      id: 202505081147
       author: hrz
       changes:
         - sqlFile:
             encoding: utf8
-            path: classpath:db/changelog/202505081146.sql
+            path: classpath:db/changelog/202505081147.sql
   - changeSet:
       id: 202505091555
       author: whosmyqueen
@@ -246,4 +246,18 @@ databaseChangeLog:
       changes:
         - sqlFile:
             encoding: utf8
-            path: classpath:db/changelog/202507101201.sql            
+            path: classpath:db/changelog/202507101201.sql            
+  - changeSet:
+      id: 202507071130
+      author: cgd
+      changes:
+          - sqlFile:
+              encoding: utf8
+              path: classpath:db/changelog/202507071130.sql
+  - changeSet:
+      id: 202507071530
+      author: cgd
+      changes:
+          - sqlFile:
+              encoding: utf8
+              path: classpath:db/changelog/202507071530.sql

main/xiaozhi-server/config.yaml

@@ -301,9 +301,13 @@ ASR:
     output_dir: tmp/
   AliyunASR:
     # 阿里云智能语音交互服务，需要先在阿里云平台开通服务，然后获取验证信息
+    # HTTP POST请求，一次性处理完整音频
     # 平台地址：https://nls-portal.console.aliyun.com/
     # appkey地址：https://nls-portal.console.aliyun.com/applist
     # token地址：https://nls-portal.console.aliyun.com/overview
+    # AliyunASR和AliyunStreamASR的区别是：AliyunASR是批量处理场景，AliyunStreamASR是实时交互场景
+    # 一般来说非流式ASR更便宜（0.004元/秒，¥0.24/分钟）
+    # 但是AliyunStreamASR实时性更好（0.005元/秒，¥0.3/分钟）
     # 定义ASR API类型
     type: aliyun
     appkey: 你的阿里云智能语音交互服务项目Appkey
@@ -311,6 +315,26 @@ ASR:
     access_key_id: 你的阿里云账号access_key_id
     access_key_secret: 你的阿里云账号access_key_secret
     output_dir: tmp/
+  AliyunStreamASR:
+    # 阿里云智能语音交互服务 - 实时流式语音识别
+    # WebSocket连接，实时处理音频流
+    # 平台地址：https://nls-portal.console.aliyun.com/
+    # appkey地址：https://nls-portal.console.aliyun.com/applist
+    # token地址：https://nls-portal.console.aliyun.com/overview
+    # AliyunASR和AliyunStreamASR的区别是：AliyunASR是批量处理场景，AliyunStreamASR是实时交互场景
+    # 一般来说非流式ASR更便宜（0.004元/秒，¥0.24/分钟）
+    # 但是AliyunStreamASR实时性更好（0.005元/秒，¥0.3/分钟）
+    # 定义ASR API类型
+    type: aliyun_stream
+    appkey: 你的阿里云智能语音交互服务项目Appkey
+    token: 你的阿里云智能语音交互服务AccessToken，临时的24小时，要长期用下方的access_key_id，access_key_secret
+    access_key_id: 你的阿里云账号access_key_id
+    access_key_secret: 你的阿里云账号access_key_secret
+    # 服务器地域选择，可选择距离更近的服务器以减少延迟，如nls-gateway-cn-hangzhou.aliyuncs.com(杭州)等
+    host: nls-gateway-cn-shanghai.aliyuncs.com
+    # 断句检测时间(毫秒)，控制静音多长时间后进行断句，默认800毫秒
+    max_sentence_silence: 800
+    output_dir: tmp/
   BaiduASR:
     # 获取AppID、API Key、Secret Key：https://console.bce.baidu.com/ai-engine/old/#/ai/speech/app/list
     # 查看资源额度：https://console.bce.baidu.com/ai-engine/old/#/ai/speech/overview/resource/list
@@ -326,6 +350,7 @@ VAD:
   SileroVAD:
     type: silero
     threshold: 0.5
+    threshold_low: 0.3
     model_dir: models/snakers4_silero-vad
     min_silence_duration_ms: 200  # 如果说话停顿比较长，可以把这个值设置大一些
 
@@ -671,6 +696,30 @@ TTS:
     # pitch_rate: 0
     # 添加 302.ai TTS 配置
     # token申请地址：https://dash.302.ai/
+  AliyunStreamTTS:
+    # 阿里云CosyVoice大模型流式文本语音合成
+    # 采用FlowingSpeechSynthesizer接口，支持更低延迟和更自然的语音质量
+    # 流式文本语音合成仅提供商用版，不支持试用，详情请参见试用版和商用版。要使用该功能，请开通商用版。
+    # 支持龙系列专用音色：longxiaochun、longyu、longchen等
+    # 平台地址：https://nls-portal.console.aliyun.com/
+    # appkey地址：https://nls-portal.console.aliyun.com/applist
+    # token地址：https://nls-portal.console.aliyun.com/overview
+    # 使用三阶段流式交互：StartSynthesis -> RunSynthesis -> StopSynthesis
+    type: aliyun_stream
+    output_dir: tmp/
+    appkey: 你的阿里云智能语音交互服务项目Appkey
+    token: 你的阿里云智能语音交互服务AccessToken，临时的24小时，要长期用下方的access_key_id，access_key_secret
+    voice: longxiaochun 
+    access_key_id: 你的阿里云账号access_key_id
+    access_key_secret: 你的阿里云账号access_key_secret
+    # 服务器地域选择，可选择距离更近的服务器以减少延迟，如nls-gateway-cn-hangzhou.aliyuncs.com(杭州)等
+    host: nls-gateway-cn-shanghai.aliyuncs.com
+    # 以下可不用设置，使用默认设置
+    # format: pcm  # 音频格式：pcm、wav、mp3
+    # sample_rate: 16000  # 采样率：8000、16000、24000
+    # volume: 50  # 音量：0-100
+    # speech_rate: 0  # 语速：-500到500
+    # pitch_rate: 0  # 语调：-500到500
   TencentTTS:
     # 腾讯云智能语音交互服务，需要先在腾讯云平台开通服务
     # appid、secret_id、secret_key申请地址：https://console.cloud.tencent.com/cam/capi

main/xiaozhi-server/core/connection.py

@@ -17,6 +17,7 @@ from core.utils.util import (
     filter_sensitive_info,
 )
 from typing import Dict, Any
+from collections import deque
 from core.utils.modules_initialize import (
     initialize_modules,
     initialize_tts,
@@ -112,6 +113,8 @@ class ConnectionHandler:
         self.client_have_voice = False
         self.last_activity_time = 0.0  # 统一的活动时间戳（毫秒）
         self.client_voice_stop = False
+        self.client_voice_window = deque(maxlen=5)
+        self.last_is_voice = False
 
         # asr相关变量
         # 因为实际部署时可能会用到公共的本地ASR，不能把变量暴露给公共ASR
@@ -608,13 +611,24 @@ class ConnectionHandler:
         # 更新系统prompt至上下文
         self.dialogue.update_system_message(self.prompt)
 
-    def chat(self, query, tool_call=False):
+    def chat(self, query, tool_call=False, depth=0):
         self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
         self.llm_finish_task = False
 
         if not tool_call:
             self.dialogue.put(Message(role="user", content=query))
 
+        # 为最顶层时新建会话ID和发送FIRST请求
+        if depth == 0:
+            self.sentence_id = str(uuid.uuid4().hex)
+            self.tts.tts_text_queue.put(
+                TTSMessageDTO(
+                    sentence_id=self.sentence_id,
+                    sentence_type=SentenceType.FIRST,
+                    content_type=ContentType.ACTION,
+                )
+            )
+
         # Define intent functions
         functions = None
         if self.intent_type == "function_call" and hasattr(self, "func_handler"):
@@ -630,8 +644,6 @@ class ConnectionHandler:
                 )
                 memory_str = future.result()
 
-            self.sentence_id = str(uuid.uuid4().hex)
-
             if self.intent_type == "function_call" and functions is not None:
                 # 使用支持functions的streaming接口
                 llm_responses = self.llm.response_with_functions(
@@ -654,7 +666,6 @@ class ConnectionHandler:
         function_id = None
         function_arguments = ""
         content_arguments = ""
-        text_index = 0
         self.client_abort = False
         for response in llm_responses:
             if self.client_abort:
@@ -684,14 +695,6 @@ class ConnectionHandler:
             if content is not None and len(content) > 0:
                 if not tool_call_flag:
                     response_message.append(content)
-                    if text_index == 0:
-                        self.tts.tts_text_queue.put(
-                            TTSMessageDTO(
-                                sentence_id=self.sentence_id,
-                                sentence_type=SentenceType.FIRST,
-                                content_type=ContentType.ACTION,
-                            )
-                        )
                     self.tts.tts_text_queue.put(
                         TTSMessageDTO(
                             sentence_id=self.sentence_id,
@@ -700,7 +703,6 @@ class ConnectionHandler:
                             content_detail=content,
                         )
                     )
-                    text_index += 1
         # 处理function call
         if tool_call_flag:
             bHasError = False
@@ -725,6 +727,11 @@ class ConnectionHandler:
                         f"function call error: {content_arguments}"
                     )
             if not bHasError:
+                # 如需要大模型先处理一轮，添加相关处理后的日志情况
+                if len(response_message) > 0:
+                    self.dialogue.put(
+                        Message(role="assistant", content="".join(response_message))
+                    )
                 response_message.clear()
                 self.logger.bind(tag=TAG).debug(
                     f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}"
@@ -742,14 +749,14 @@ class ConnectionHandler:
                     ),
                     self.loop,
                 ).result()
-                self._handle_function_result(result, function_call_data)
+                self._handle_function_result(result, function_call_data, depth=depth)
 
         # 存储对话内容
         if len(response_message) > 0:
             self.dialogue.put(
                 Message(role="assistant", content="".join(response_message))
             )
-        if text_index > 0:
+        if depth == 0:
             self.tts.tts_text_queue.put(
                 TTSMessageDTO(
                     sentence_id=self.sentence_id,
@@ -764,7 +771,7 @@ class ConnectionHandler:
 
         return True
 
-    def _handle_function_result(self, result, function_call_data):
+    def _handle_function_result(self, result, function_call_data, depth):
         if result.action == Action.RESPONSE:  # 直接回复前端
             text = result.response
             self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
@@ -801,7 +808,7 @@ class ConnectionHandler:
                         content=text,
                     )
                 )
-                self.chat(text, tool_call=True)
+                self.chat(text, tool_call=True, depth=depth + 1)
         elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
             text = result.response if result.response else result.result
             self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
@@ -914,6 +921,9 @@ class ConnectionHandler:
             except Exception as ws_error:
                 self.logger.bind(tag=TAG).error(f"关闭WebSocket连接时出错: {ws_error}")
 
+            if self.tts:
+                await self.tts.close()
+
             # 最后关闭线程池（避免阻塞）
             if self.executor:
                 try:

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

@@ -6,9 +6,8 @@ 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 core.providers.tools.device_mcp import call_mcp_tool
 from plugins_func.register import Action, ActionResponse
-from loguru import logger
+from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
 
 TAG = __name__
 
@@ -29,6 +28,8 @@ async def handle_user_intent(conn, text):
     intent_result = await analyze_intent_with_llm(conn, text)
     if not intent_result:
         return False
+    # 会话开始时生成sentence_id
+    conn.sentence_id = str(uuid.uuid4().hex)
     # 处理各种意图
     return await process_intent_result(conn, intent_result, text)
 
@@ -153,5 +154,19 @@ async def process_intent_result(conn, intent_result, original_text):
 
 
 def speak_txt(conn, text):
+    conn.tts.tts_text_queue.put(
+        TTSMessageDTO(
+            sentence_id=conn.sentence_id,
+            sentence_type=SentenceType.FIRST,
+            content_type=ContentType.ACTION,
+        )
+    )
     conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text)
+    conn.tts.tts_text_queue.put(
+        TTSMessageDTO(
+            sentence_id=conn.sentence_id,
+            sentence_type=SentenceType.LAST,
+            content_type=ContentType.ACTION,
+        )
+    )
     conn.dialogue.put(Message(role="assistant", content=text))

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

@@ -0,0 +1,286 @@
+import json
+import time
+import uuid
+import hmac
+import base64
+import hashlib
+import asyncio
+import requests
+import websockets
+import opuslib_next
+import random
+from typing import Optional, Tuple, List
+from urllib import parse
+from datetime import datetime
+from config.logger import setup_logging
+from core.providers.asr.base import ASRProviderBase
+from core.providers.asr.dto.dto import InterfaceType
+
+TAG = __name__
+logger = setup_logging()
+
+
+class AccessToken:
+    @staticmethod
+    def _encode_text(text):
+        encoded_text = parse.quote_plus(text)
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
+
+    @staticmethod
+    def _encode_dict(dic):
+        keys = dic.keys()
+        dic_sorted = [(key, dic[key]) for key in sorted(keys)]
+        encoded_text = parse.urlencode(dic_sorted)
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
+
+    @staticmethod
+    def create_token(access_key_id, access_key_secret):
+        parameters = {
+            "AccessKeyId": access_key_id,
+            "Action": "CreateToken",
+            "Format": "JSON",
+            "RegionId": "cn-shanghai",
+            "SignatureMethod": "HMAC-SHA1",
+            "SignatureNonce": str(uuid.uuid1()),
+            "SignatureVersion": "1.0",
+            "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            "Version": "2019-02-28",
+        }
+        query_string = AccessToken._encode_dict(parameters)
+        string_to_sign = (
+            "GET" + "&" + AccessToken._encode_text("/") + "&" + AccessToken._encode_text(query_string)
+        )
+        secreted_string = hmac.new(
+            bytes(access_key_secret + "&", encoding="utf-8"),
+            bytes(string_to_sign, encoding="utf-8"),
+            hashlib.sha1,
+        ).digest()
+        signature = base64.b64encode(secreted_string)
+        signature = AccessToken._encode_text(signature)
+        full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (signature, query_string)
+        response = requests.get(full_url)
+        if response.ok:
+            root_obj = response.json()
+            if "Token" in root_obj:
+                return root_obj["Token"]["Id"], root_obj["Token"]["ExpireTime"]
+        return None, None
+
+
+class ASRProvider(ASRProviderBase):
+    def __init__(self, config, delete_audio_file):
+        super().__init__()
+        self.interface_type = InterfaceType.STREAM
+        self.config = config
+        self.text = ""
+        self.decoder = opuslib_next.Decoder(16000, 1)
+        self.asr_ws = None
+        self.forward_task = None
+        self.is_processing = False
+        self.server_ready = False  # 服务器准备状态
+
+        # 基础配置
+        self.access_key_id = config.get("access_key_id")
+        self.access_key_secret = config.get("access_key_secret")
+        self.appkey = config.get("appkey")
+        self.token = config.get("token")
+        self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com")
+        self.ws_url = f"wss://{self.host}/ws/v1"
+        self.max_sentence_silence = config.get("max_sentence_silence")
+        self.output_dir = config.get("output_dir", "./audio_output")
+        self.delete_audio_file = delete_audio_file
+        self.expire_time = None
+
+        # Token管理
+        if self.access_key_id and self.access_key_secret:
+            self._refresh_token()
+        elif not self.token:
+            raise ValueError("必须提供access_key_id+access_key_secret或者直接提供token")
+
+    def _refresh_token(self):
+        """刷新Token"""
+        self.token, expire_time_str = AccessToken.create_token(self.access_key_id, self.access_key_secret)
+        if not self.token:
+            raise ValueError("无法获取有效的访问Token")
+        
+        try:
+            expire_str = str(expire_time_str).strip()
+            if expire_str.isdigit():
+                expire_time = datetime.fromtimestamp(int(expire_str))
+            else:
+                expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ")
+            self.expire_time = expire_time.timestamp() - 60
+        except:
+            self.expire_time = None
+
+    def _is_token_expired(self):
+        """检查Token是否过期"""
+        return self.expire_time and time.time() > self.expire_time
+
+    async def open_audio_channels(self, conn):
+        await super().open_audio_channels(conn)
+
+    async def receive_audio(self, conn, audio, audio_have_voice):
+        conn.asr_audio.append(audio)
+        conn.asr_audio = conn.asr_audio[-10:]
+
+        # 只在有声音且没有连接时建立连接
+        if audio_have_voice and not self.is_processing:
+            try:
+                await self._start_recognition(conn)
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"开始识别失败: {str(e)}")
+                await self._cleanup()
+                return
+
+        if self.asr_ws and self.is_processing and self.server_ready:
+            try:
+                pcm_frame = self.decoder.decode(audio, 960)
+                await self.asr_ws.send(pcm_frame)
+            except Exception as e:
+                logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
+                await self._cleanup()
+
+    async def _start_recognition(self, conn):
+        """开始识别会话"""
+        if self._is_token_expired():
+            self._refresh_token()
+        
+        # 建立连接
+        headers = {"X-NLS-Token": self.token}
+        self.asr_ws = await websockets.connect(
+            self.ws_url,
+            additional_headers=headers,
+            max_size=1000000000,
+            ping_interval=None,
+            ping_timeout=None,
+            close_timeout=5,
+        )
+        
+        self.is_processing = True
+        self.server_ready = False  # 重置服务器准备状态
+        self.forward_task = asyncio.create_task(self._forward_results(conn))
+        
+        # 发送开始请求
+        start_request = {
+            "header": {
+                "namespace": "SpeechTranscriber",
+                "name": "StartTranscription",
+                "status": 20000000,
+                "message_id": ''.join(random.choices('0123456789abcdef', k=32)),
+                "task_id": ''.join(random.choices('0123456789abcdef', k=32)),
+                "status_text": "Gateway:SUCCESS:Success.",
+                "appkey": self.appkey
+            },
+            "payload": {
+                "format": "pcm",
+                "sample_rate": 16000,
+                "enable_intermediate_result": True,
+                "enable_punctuation_prediction": True,
+                "enable_inverse_text_normalization": True,
+                "max_sentence_silence": self.max_sentence_silence,
+                "enable_voice_detection": False,
+            }
+        }
+        await self.asr_ws.send(json.dumps(start_request, ensure_ascii=False))
+        logger.bind(tag=TAG).info("已发送开始请求，等待服务器准备...")
+
+    async def _forward_results(self, conn):
+        """转发识别结果"""
+        try:
+            while self.asr_ws and not conn.stop_event.is_set():
+                try:
+                    response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
+                    result = json.loads(response)
+                    
+                    header = result.get("header", {})
+                    payload = result.get("payload", {})
+                    message_name = header.get("name", "")
+                    status = header.get("status", 0)
+                    
+                    if status != 20000000:
+                        if status in [40000004, 40010004]:  # 连接超时或客户端断开
+                            logger.bind(tag=TAG).warning(f"连接问题，状态码: {status}")
+                            break
+                        elif status in [40270002, 40270003]:  # 音频问题
+                            logger.bind(tag=TAG).warning(f"音频处理问题，状态码: {status}")
+                            continue
+                        else:
+                            logger.bind(tag=TAG).error(f"识别错误，状态码: {status}, 消息: {header.get('status_text', '')}")
+                            continue
+                    
+                    # 收到TranscriptionStarted表示服务器准备好接收音频数据
+                    if message_name == "TranscriptionStarted":
+                        self.server_ready = True
+                        logger.bind(tag=TAG).info("服务器已准备，开始发送缓存音频...")
+                        
+                        # 发送缓存音频
+                        if conn.asr_audio:
+                            for cached_audio in conn.asr_audio[-10:]:
+                                try:
+                                    pcm_frame = self.decoder.decode(cached_audio, 960)
+                                    await self.asr_ws.send(pcm_frame)
+                                except Exception as e:
+                                    logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
+                                    break
+                        continue
+                    
+                    if message_name == "TranscriptionResultChanged":
+                        # 中间结果
+                        text = payload.get("result", "")
+                        if text:
+                            self.text = text
+                    elif message_name == "SentenceEnd":
+                        # 最终结果
+                        text = payload.get("result", "")
+                        if text:
+                            self.text = text
+                            conn.reset_vad_states()
+                            await self.handle_voice_stop(conn, None)
+                            break
+                    elif message_name == "TranscriptionCompleted":
+                        # 识别完成
+                        self.is_processing = False
+                        break
+                        
+                except asyncio.TimeoutError:
+                    continue
+                except websockets.exceptions.ConnectionClosed:
+                    break
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"处理结果失败: {str(e)}")
+                    break
+                    
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"结果转发失败: {str(e)}")
+        finally:
+            await self._cleanup()
+
+    async def _cleanup(self):
+        """清理资源"""
+        self.is_processing = False
+        self.server_ready = False  # 重置服务器准备状态
+        
+        if self.forward_task and not self.forward_task.done():
+            self.forward_task.cancel()
+            try:
+                await asyncio.wait_for(self.forward_task, timeout=1.0)
+            except:
+                pass
+            self.forward_task = None
+        
+        if self.asr_ws:
+            try:
+                await asyncio.wait_for(self.asr_ws.close(), timeout=2.0)
+            except:
+                pass
+            self.asr_ws = None
+
+    async def speech_to_text(self, opus_data, session_id, audio_format):
+        """获取识别结果"""
+        result = self.text
+        self.text = ""
+        return result, None
+
+    async def close(self):
+        """关闭资源"""
+        await self._cleanup()

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

@@ -0,0 +1,604 @@
+import uuid
+import json
+import hmac
+import hashlib
+import base64
+import time
+import queue
+import asyncio
+import traceback
+from asyncio import Task
+import websockets
+import os
+from datetime import datetime
+from urllib import parse
+from core.providers.tts.base import TTSProviderBase
+from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
+from core.utils.tts import MarkdownCleaner
+from core.utils import opus_encoder_utils, textUtils
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class AccessToken:
+    @staticmethod
+    def _encode_text(text):
+        encoded_text = parse.quote_plus(text)
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
+
+    @staticmethod
+    def _encode_dict(dic):
+        keys = dic.keys()
+        dic_sorted = [(key, dic[key]) for key in sorted(keys)]
+        encoded_text = parse.urlencode(dic_sorted)
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
+
+    @staticmethod
+    def create_token(access_key_id, access_key_secret):
+        parameters = {
+            "AccessKeyId": access_key_id,
+            "Action": "CreateToken",
+            "Format": "JSON",
+            "RegionId": "cn-shanghai",  # 使用上海地域进行Token获取
+            "SignatureMethod": "HMAC-SHA1",
+            "SignatureNonce": str(uuid.uuid1()),
+            "SignatureVersion": "1.0",
+            "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            "Version": "2019-02-28",
+        }
+
+        query_string = AccessToken._encode_dict(parameters)
+        string_to_sign = (
+            "GET"
+            + "&"
+            + AccessToken._encode_text("/")
+            + "&"
+            + AccessToken._encode_text(query_string)
+        )
+
+        secreted_string = hmac.new(
+            bytes(access_key_secret + "&", encoding="utf-8"),
+            bytes(string_to_sign, encoding="utf-8"),
+            hashlib.sha1,
+        ).digest()
+        signature = base64.b64encode(secreted_string)
+        signature = AccessToken._encode_text(signature)
+
+        full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (
+            signature,
+            query_string,
+        )
+
+        import requests
+        response = requests.get(full_url)
+        if response.ok:
+            root_obj = response.json()
+            key = "Token"
+            if key in root_obj:
+                token = root_obj[key]["Id"]
+                expire_time = root_obj[key]["ExpireTime"]
+                return token, expire_time
+        return None, None
+
+
+class TTSProvider(TTSProviderBase):
+    def __init__(self, config, delete_audio_file):
+        super().__init__(config, delete_audio_file)
+
+        # 设置为流式接口类型
+        self.interface_type = InterfaceType.DUAL_STREAM
+
+        # 基础配置
+        self.access_key_id = config.get("access_key_id")
+        self.access_key_secret = config.get("access_key_secret")
+        self.appkey = config.get("appkey")
+        self.format = config.get("format", "pcm")
+        self.audio_file_type = config.get("format", "pcm")
+
+        # 采样率配置
+        sample_rate = config.get("sample_rate", "16000")
+        self.sample_rate = int(sample_rate) if sample_rate else 16000
+
+        # 音色配置 - CosyVoice大模型音色
+        if config.get("private_voice"):
+            self.voice = config.get("private_voice")
+        else:
+            self.voice = config.get("voice", "longxiaochun")  # CosyVoice默认音色
+
+        # 音频参数配置
+        volume = config.get("volume", "50")
+        self.volume = int(volume) if volume else 50
+
+        speech_rate = config.get("speech_rate", "0")
+        self.speech_rate = int(speech_rate) if speech_rate else 0
+
+        pitch_rate = config.get("pitch_rate", "0")
+        self.pitch_rate = int(pitch_rate) if pitch_rate else 0
+
+        # WebSocket配置
+        self.host = config.get("host", "nls-gateway-cn-beijing.aliyuncs.com")
+        self.ws_url = f"wss://{self.host}/ws/v1"
+        self.ws = None
+        self._monitor_task = None
+        self.last_active_time = None
+
+        # 专属tts设置
+        self.message_id = ""
+        self.tts_text = ''
+        self.text_buffer = []
+
+        # 创建Opus编码器
+        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+            sample_rate=16000, channels=1, frame_size_ms=60
+        )
+
+        # Token管理
+        if self.access_key_id and self.access_key_secret:
+            self._refresh_token()
+        else:
+            self.token = config.get("token")
+            self.expire_time = None
+
+    def _refresh_token(self):
+        """刷新Token并记录过期时间"""
+        if self.access_key_id and self.access_key_secret:
+            self.token, expire_time_str = AccessToken.create_token(
+                self.access_key_id, self.access_key_secret
+            )
+            if not expire_time_str:
+                raise ValueError("无法获取有效的Token过期时间")
+
+            expire_str = str(expire_time_str).strip()
+
+            try:
+                if expire_str.isdigit():
+                    expire_time = datetime.fromtimestamp(int(expire_str))
+                else:
+                    expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ")
+                self.expire_time = expire_time.timestamp() - 60
+            except Exception as e:
+                raise ValueError(f"无效的过期时间格式: {expire_str}") from e
+        else:
+            self.expire_time = None
+
+        if not self.token:
+            raise ValueError("无法获取有效的访问Token")
+
+    def _is_token_expired(self):
+        """检查Token是否过期"""
+        if not self.expire_time:
+            return False
+        return time.time() > self.expire_time
+
+    async def _ensure_connection(self):
+        """确保WebSocket连接可用"""
+        try:
+            if self._is_token_expired():
+                logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...")
+                self._refresh_token()
+            current_time = time.time()
+            if self.ws and current_time - self.last_active_time < 10:
+                # 10秒内才可以复用链接进行连续对话
+                logger.bind(tag=TAG).info(f"使用已有链接...")
+                return self.ws
+            logger.bind(tag=TAG).info("开始建立新连接...")
+
+            self.ws = await websockets.connect(
+                self.ws_url,
+                additional_headers={"X-NLS-Token": self.token},
+                ping_interval=30,
+                ping_timeout=10,
+                close_timeout=10,
+            )
+            logger.bind(tag=TAG).info("WebSocket连接建立成功")
+            self.last_active_time = time.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):
+        """流式文本处理线程"""
+        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:
+                    logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
+                    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}")
+
+                        # aliyunStream独有的参数生成
+                        self.message_id = str(uuid.uuid4().hex)
+                        self.text_buffer = []
+
+                        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}"
+                            )
+                            self.text_buffer.append(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):
+                        # 先处理文件音频数据
+                        file_audio = self._process_audio_file(message.content_file)
+                        self.before_stop_play_files.append(
+                            (file_audio, message.content_detail)
+                        )
+
+                if message.sentence_type == SentenceType.LAST:
+                    try:
+                        logger.bind(tag=TAG).info("开始结束TTS会话...")
+                        self.tts_text = textUtils.get_string_no_punctuation_or_emoji(
+                            ''.join(self.text_buffer).replace('\n', '')
+                        )
+                        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()}"
+                )
+
+    async def text_to_speak(self, text, _):
+        try:
+            if self.ws is None:
+                logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本")
+                return
+            filtered_text = MarkdownCleaner.clean_markdown(text)
+            run_request = {
+                "header": {
+                    "message_id": self.message_id,
+                    "task_id": self.conn.sentence_id,
+                    "namespace": "FlowingSpeechSynthesizer",
+                    "name": "RunSynthesis",
+                    "appkey": self.appkey,
+                },
+                "payload": {
+                    "text": filtered_text
+                }
+            }
+            await self.ws.send(json.dumps(run_request))
+            self.last_active_time = time.time()
+            return
+
+        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):
+        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())
+
+            start_request = {
+                "header": {
+                    "message_id": self.message_id,
+                    "task_id": self.conn.sentence_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 self.ws.send(json.dumps(start_request))
+            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):
+        logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
+        try:
+            if self.ws:
+                stop_request = {
+                    "header": {
+                        "message_id": self.message_id,
+                        "task_id": self.conn.sentence_id,
+                        "namespace": "FlowingSpeechSynthesizer",
+                        "name": "StopSynthesis",
+                        "appkey": self.appkey,
+                    }
+                }
+                await self.ws.send(json.dumps(stop_request))
+                logger.bind(tag=TAG).info("会话结束请求已发送")
+                self.last_active_time = time.time()
+                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
+
+        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响应"""
+        opus_datas_cache = []
+        is_first_sentence = True
+        first_sentence_segment_count = 0  # 添加计数器
+        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):  # 文本控制消息
+                        try:
+                            data = json.loads(msg)
+                            header = data.get("header", {})
+                            event_name = header.get("name")
+                            if event_name == "SynthesisStarted":
+                                logger.bind(tag=TAG).debug("TTS合成已启动")
+                            elif event_name == "SentenceBegin":
+                                logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}")
+                                opus_datas_cache = []
+                                self.tts_audio_queue.put((SentenceType.FIRST, [], self.tts_text))
+                            elif event_name == "SentenceEnd":
+                                logger.bind(tag=TAG).info(f"句子语音生成成功： {self.tts_text}")
+                                if not is_first_sentence or first_sentence_segment_count > 10:
+                                    # 发送缓存的数据
+                                    self.tts_audio_queue.put(
+                                        (SentenceType.MIDDLE, opus_datas_cache, None)
+                                    )
+                                # 第一句话结束后，将标志设置为False
+                                is_first_sentence = False
+                            elif event_name == "SynthesisCompleted":
+                                logger.bind(tag=TAG).debug(f"会话结束～～")
+                                self._process_before_stop_play_files()
+                                session_finished = True
+                                self.reuse_judgment = time.time()
+                                self.tts_text = ''
+                                break
+                        except json.JSONDecodeError:
+                            logger.bind(tag=TAG).warning("收到无效的JSON消息")
+                    # 二进制消息（音频数据）
+                    elif isinstance(msg, (bytes, bytearray)):
+                        logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
+                        opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False)
+                        logger.bind(tag=TAG).debug(
+                            f"推送数据到队列里面帧数～～{len(opus_datas)}"
+                        )
+                        if is_first_sentence:
+                            first_sentence_segment_count += 1
+                            if first_sentence_segment_count <= 6:
+                                self.tts_audio_queue.put(
+                                    (SentenceType.MIDDLE, opus_datas, None)
+                                )
+                            else:
+                                opus_datas_cache.extend(opus_datas)
+                        else:
+                            # 后续句子缓存
+                            opus_datas_cache.extend(opus_datas)
+
+                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:
+        """非流式TTS处理，用于测试及保存音频文件的场景"""
+        try:
+            # 创建新的事件循环
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+            # 生成会话ID
+            session_id = uuid.uuid4().hex
+            message_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_request = {
+                        "header": {
+                            "message_id": 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))
+
+                    # 发送文本合成请求
+                    filtered_text = MarkdownCleaner.clean_markdown(text)
+                    run_request = {
+                        "header": {
+                            "message_id": 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_request = {
+                        "header": {
+                            "message_id": message_id,
+                            "task_id": session_id,
+                            "namespace": "FlowingSpeechSynthesizer",
+                            "name": "StopSynthesis",
+                            "appkey": self.appkey,
+                        }
+                    }
+                    await ws.send(json.dumps(stop_request))
+
+                    # 接收音频数据
+                    while True:
+                        msg = await ws.recv()
+                        if isinstance(msg, (bytes, bytearray)):
+                            # 编码为Opus并收集
+                            opus_frames = self.opus_encoder.encode_pcm_to_opus(msg, False)
+                            audio_data.extend(opus_frames)
+                        elif isinstance(msg, str):
+                            data = json.loads(msg)
+                            header = data.get("header", {})
+                            if header.get("name") == "SynthesisCompleted":
+                                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/base.py

@@ -159,15 +159,8 @@ class TTSProviderBase(ABC):
             if conn.sentence_id:
                 sentence_id = conn.sentence_id
             else:
-                sentence_id = str(uuid.uuid4()).replace("-", "")
+                sentence_id = str(uuid.uuid4().hex)
                 conn.sentence_id = sentence_id
-        self.tts_text_queue.put(
-            TTSMessageDTO(
-                sentence_id=sentence_id,
-                sentence_type=SentenceType.FIRST,
-                content_type=ContentType.ACTION,
-            )
-        )
         # 对于单句的文本，进行分段处理
         segments = re.split(r"([。！？!?；;\n])", content_detail)
         for seg in segments:
@@ -180,13 +173,6 @@ class TTSProviderBase(ABC):
                     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
@@ -209,6 +195,8 @@ class TTSProviderBase(ABC):
         while not self.conn.stop_event.is_set():
             try:
                 message = self.tts_text_queue.get(timeout=1)
+                if message.sentence_type == SentenceType.FIRST:
+                    self.conn.client_abort = False
                 if self.conn.client_abort:
                     logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
                     continue
@@ -362,10 +350,8 @@ class TTSProviderBase(ABC):
         return audio_datas
 
     def _process_before_stop_play_files(self):
-        for tts_file, text in self.before_stop_play_files:
+        for audio_datas, text in self.before_stop_play_files:
-            if tts_file and os.path.exists(tts_file):
+            self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text))
-                audio_datas = self._process_audio_file(tts_file)
-                self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text))
         self.before_stop_play_files.clear()
         self.tts_audio_queue.put((SentenceType.LAST, [], None))
 

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

@@ -10,7 +10,6 @@ from config.logger import setup_logging
 from core.utils import opus_encoder_utils
 from core.utils.util import check_model_key
 from core.providers.tts.base import TTSProviderBase
-from core.handle.abortHandle import handleAbortMessage
 from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
 from asyncio import Task
 
@@ -175,6 +174,9 @@ class TTSProvider(TTSProviderBase):
     async def _ensure_connection(self):
         """建立新的WebSocket连接"""
         try:
+            if self.ws:
+                logger.bind(tag=TAG).info(f"使用已有链接...")
+                return self.ws
             logger.bind(tag=TAG).info("开始建立新连接...")
             ws_header = {
                 "X-Api-App-Key": self.appId,
@@ -200,6 +202,10 @@ class TTSProvider(TTSProviderBase):
                 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:
                     logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
                     continue
@@ -244,9 +250,12 @@ class TTSProvider(TTSProviderBase):
                     logger.bind(tag=TAG).info(
                         f"添加音频文件到待播放列表: {message.content_file}"
                     )
-                    self.before_stop_play_files.append(
+                    if message.content_file and os.path.exists(message.content_file):
-                        (message.content_file, message.content_detail)
+                        # 先处理文件音频数据
-                    )
+                        file_audio = self._process_audio_file(message.content_file)
+                        self.before_stop_play_files.append(
+                            (file_audio, message.content_detail)
+                        )
 
                 if message.sentence_type == SentenceType.LAST:
                     try:
@@ -295,25 +304,15 @@ class TTSProvider(TTSProviderBase):
     async def start_session(self, session_id):
         logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
         try:
-            task = self._monitor_task
+            # 会话开始时检测上个会话的监听状态
             if (
-                task is not None
+                self._monitor_task is not None
-                and isinstance(task, Task)
+                and isinstance(self._monitor_task, Task)
-                and not task.done()
+                and not self._monitor_task.done()
             ):
-                logger.bind(tag=TAG).info("等待上一个监听任务结束...")
+                logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...")
-                if self.ws is not None:
+                await self.close()
-                    logger.bind(tag=TAG).info("强制关闭上一个WebSocket连接以唤醒监听任务...")
+
-                    try:
-                        await self.ws.close()
-                    except Exception as e:
-                        logger.bind(tag=TAG).warning(f"关闭上一个ws异常: {e}")
-                    self.ws = None
-                try:
-                    await asyncio.wait_for(task, timeout=8)
-                except Exception as e:
-                    logger.bind(tag=TAG).warning(f"等待监听任务异常: {e}")
-                self._monitor_task = None
             # 建立新连接
             await self._ensure_connection()
 
@@ -336,19 +335,7 @@ class TTSProvider(TTSProviderBase):
         except Exception as e:
             logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
             # 确保清理资源
-            if hasattr(self, "_monitor_task"):
+            await self.close()
-                try:
-                    self._monitor_task.cancel()
-                    await self._monitor_task
-                except:
-                    pass
-                self._monitor_task = None
-            if self.ws:
-                try:
-                    await self.ws.close()
-                except:
-                    pass
-                self.ws = None
             raise
 
     async def finish_session(self, session_id):
@@ -368,7 +355,7 @@ class TTSProvider(TTSProviderBase):
                 logger.bind(tag=TAG).info("会话结束请求已发送")
 
                 # 等待监听任务完成
-                if hasattr(self, "_monitor_task"):
+                if self._monitor_task:
                     try:
                         await self._monitor_task
                     except Exception as e:
@@ -378,28 +365,25 @@ class TTSProvider(TTSProviderBase):
                     finally:
                         self._monitor_task = None
 
-                # 关闭连接
-                await self.close()
         except Exception as e:
             logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
             # 确保清理资源
-            if hasattr(self, "_monitor_task"):
+            await self.close()
-                try:
-                    self._monitor_task.cancel()
-                    await self._monitor_task
-                except:
-                    pass
-                self._monitor_task = None
-            if self.ws:
-                try:
-                    await self.ws.close()
-                except:
-                    pass
-                self.ws = None
             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
+
         if self.ws:
             try:
                 await self.ws.close()
@@ -413,6 +397,7 @@ class TTSProvider(TTSProviderBase):
         is_first_sentence = True
         first_sentence_segment_count = 0  # 添加计数器
         try:
+            session_finished = False  # 标记会话是否正常结束
             while not self.conn.stop_event.is_set():
                 try:
                     # 确保 `recv()` 运行在同一个 event loop
@@ -450,10 +435,10 @@ class TTSProvider(TTSProviderBase):
                                     (SentenceType.MIDDLE, opus_datas, None)
                                 )
                             else:
-                                opus_datas_cache = opus_datas_cache + opus_datas
+                                opus_datas_cache.extend(opus_datas)
                         else:
                             # 后续句子缓存
-                            opus_datas_cache = opus_datas_cache + opus_datas
+                            opus_datas_cache.extend(opus_datas)
                     elif res.optional.event == EVENT_TTSSentenceEnd:
                         logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
                         if not is_first_sentence or first_sentence_segment_count > 10:
@@ -466,6 +451,7 @@ class TTSProvider(TTSProviderBase):
                     elif res.optional.event == EVENT_SessionFinished:
                         logger.bind(tag=TAG).debug(f"会话结束～～")
                         self._process_before_stop_play_files()
+                        session_finished = True
                         break
                 except websockets.ConnectionClosed:
                     logger.bind(tag=TAG).warning("WebSocket连接已关闭")
@@ -476,15 +462,15 @@ class TTSProvider(TTSProviderBase):
                     )
                     traceback.print_exc()
                     break
-        finally:
+            # 仅在连接异常时才关闭
-            # 确保清理资源
+            if not session_finished and self.ws:
-            if self.ws:
                 try:
                     await self.ws.close()
                 except:
                     pass
                 self.ws = None
-            # 监听任务退出时清理引用
+        # 监听任务退出时清理引用
+        finally:
             self._monitor_task = None
 
     async def send_event(

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

@@ -1,3 +1,4 @@
+import os
 import queue
 import asyncio
 import traceback
@@ -63,9 +64,12 @@ class TTSProvider(TTSProviderBase):
                     logger.bind(tag=TAG).info(
                         f"添加音频文件到待播放列表: {message.content_file}"
                     )
-                    self.before_stop_play_files.append(
+                    if message.content_file and os.path.exists(message.content_file):
-                        (message.content_file, message.content_detail)
+                        # 先处理文件音频数据
-                    )
+                        file_audio = self._process_audio_file(message.content_file)
+                        self.before_stop_play_files.append(
+                            (file_audio, message.content_detail)
+                        )
 
                 if message.sentence_type == SentenceType.LAST:
                     # 处理剩余的文本

main/xiaozhi-server/core/providers/vad/silero.py

@@ -23,22 +23,24 @@ class VADProvider(VADProviderBase):
 
         # 处理空字符串的情况
         threshold = config.get("threshold", "0.5")
+        threshold_low = config.get("threshold_low", "0.2")
         min_silence_duration_ms = config.get("min_silence_duration_ms", "1000")
 
         self.vad_threshold = float(threshold) if threshold else 0.5
+        self.vad_threshold_low = float(threshold_low) if threshold_low else 0.2
+
         self.silence_threshold_ms = (
             int(min_silence_duration_ms) if min_silence_duration_ms else 1000
         )
 
+        # 至少要多少帧才算有语音
+        self.frame_window_threshold = 3
+
     def is_vad(self, conn, opus_packet):
         try:
             pcm_frame = self.decoder.decode(opus_packet, 960)
             conn.client_audio_buffer.extend(pcm_frame)  # 将新数据加入缓冲区
 
-            # 确保帧计数器存在
-            if not hasattr(conn, "client_voice_frame_count"):
-                conn.client_voice_frame_count = 0
-
             # 处理缓冲区中的完整帧（每次处理512采样点）
             client_have_voice = False
             while len(conn.client_audio_buffer) >= 512 * 2:
@@ -54,15 +56,21 @@ class VADProvider(VADProviderBase):
                 # 检测语音活动
                 with torch.no_grad():
                     speech_prob = self.model(audio_tensor, 16000).item()
-                is_voice = speech_prob >= self.vad_threshold
 
-                if is_voice:
+                # 双阈值判断
-                    conn.client_voice_frame_count += 1
+                if speech_prob >= self.vad_threshold:
+                    is_voice = True
+                elif speech_prob <= self.vad_threshold_low:
+                    is_voice = False
                 else:
-                    conn.client_voice_frame_count = 0
+                    is_voice = conn.last_is_voice
 
-                # 只有连续4帧检测到语音才认为有语音
+                # 声音没低于最低值则延续前一个状态，判断为有声音
-                client_have_voice = conn.client_voice_frame_count >= 4
+                conn.last_is_voice = is_voice
+
+                # 更新滑动窗口
+                conn.client_voice_window.append(is_voice)
+                client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold)
 
                 # 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话
                 if conn.client_have_voice and not client_have_voice:

main/xiaozhi-server/plugins_func/functions/play_music.py

@@ -1,13 +1,10 @@
-from config.logger import setup_logging
 import os
 import re
 import time
 import random
-import asyncio
 import difflib
 import traceback
 from pathlib import Path
-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
@@ -51,8 +48,8 @@ def play_music(conn, song_name: str):
             )
 
         # 提交异步任务
-        future = asyncio.run_coroutine_threadsafe(
+        task = conn.loop.create_task(
-            handle_music_command(conn, music_intent), conn.loop
+            handle_music_command(conn, music_intent)  # 封装异步逻辑
         )
 
         # 非阻塞回调处理
@@ -63,7 +60,7 @@ def play_music(conn, song_name: str):
             except Exception as e:
                 conn.logger.bind(tag=TAG).error(f"播放失败: {e}")
 
-        future.add_done_callback(handle_done)
+        task.add_done_callback(handle_done)
 
         return ActionResponse(
             action=Action.NONE, result="指令已接收", response="正在为您播放音乐"
@@ -178,13 +175,13 @@ def _get_random_play_prompt(song_name):
     # 移除文件扩展名
     clean_name = os.path.splitext(song_name)[0]
     prompts = [
-        f"正在为您播放，{clean_name}",
+        f"正在为您播放，《{clean_name}》",
-        f"请欣赏歌曲，{clean_name}",
+        f"请欣赏歌曲，《{clean_name}》",
-        f"即将为您播放，{clean_name}",
+        f"即将为您播放，《{clean_name}》",
-        f"为您带来，{clean_name}",
+        f"现在为您带来，《{clean_name}》",
-        f"让我们聆听，{clean_name}",
+        f"让我们一起聆听，《{clean_name}》",
-        f"接下来请欣赏，{clean_name}",
+        f"接下来请欣赏，《{clean_name}》",
-        f"为您献上，{clean_name}",
+        f"此刻为您献上，《{clean_name}》",
     ]
     # 直接使用random.choice，不设置seed
     return random.choice(prompts)
@@ -218,13 +215,14 @@ async def play_local_music(conn, specific_file=None):
         await send_stt_message(conn, text)
         conn.dialogue.put(Message(role="assistant", content=text))
 
-        conn.tts.tts_text_queue.put(
+        if conn.intent_type == "intent_llm":
-            TTSMessageDTO(
+            conn.tts.tts_text_queue.put(
-                sentence_id=conn.sentence_id,
+                TTSMessageDTO(
-                sentence_type=SentenceType.FIRST,
+                    sentence_id=conn.sentence_id,
-                content_type=ContentType.ACTION,
+                    sentence_type=SentenceType.FIRST,
+                    content_type=ContentType.ACTION,
+                )
             )
-        )
         conn.tts.tts_text_queue.put(
             TTSMessageDTO(
                 sentence_id=conn.sentence_id,
@@ -241,13 +239,14 @@ async def play_local_music(conn, specific_file=None):
                 content_file=music_path,
             )
         )
-        conn.tts.tts_text_queue.put(
+        if conn.intent_type == "intent_llm":
-            TTSMessageDTO(
+            conn.tts.tts_text_queue.put(
-                sentence_id=conn.sentence_id,
+                TTSMessageDTO(
-                sentence_type=SentenceType.LAST,
+                    sentence_id=conn.sentence_id,
-                content_type=ContentType.ACTION,
+                    sentence_type=SentenceType.LAST,
+                    content_type=ContentType.ACTION,
+                )
             )
-        )
 
     except Exception as e:
         conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")