From cc081bc9ebe7ac3b317b81e89bd6846423d8209e Mon Sep 17 00:00:00 2001 From: DaGou12138 <991623169@qq.com> Date: Thu, 26 Feb 2026 17:46:20 +0800 Subject: [PATCH 1/2] =?UTF-8?q?#2939=20=E4=BF=AE=E5=A4=8DgetDeviceTools()?= =?UTF-8?q?=E4=B8=AD=E6=9C=AA=E8=80=83=E8=99=91=E5=88=86=E9=A1=B5=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/DeviceServiceImpl.java | 105 +++++++++++++----- 1 file changed, 75 insertions(+), 30 deletions(-) diff --git a/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java index 57e7247e..adabbd95 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java @@ -4,6 +4,7 @@ import java.nio.charset.StandardCharsets; import java.security.InvalidKeyException; import java.security.NoSuchAlgorithmException; import java.time.Instant; +import java.util.ArrayList; import java.util.Base64; import java.util.Date; import java.util.HashMap; @@ -38,6 +39,8 @@ import cn.hutool.crypto.digest.DigestUtil; import cn.hutool.http.ContentType; import cn.hutool.http.Header; import cn.hutool.http.HttpRequest; +import cn.hutool.json.JSONArray; +import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import jakarta.servlet.http.HttpServletRequest; import lombok.AllArgsConstructor; @@ -700,40 +703,82 @@ public class DeviceServiceImpl extends BaseServiceImpl // 构建完整的URL String url = StrUtil.format("http://{}/api/commands/{}", mqttGatewayUrl, clientId); - // 构建请求体 - Map payload = MapUtil - .builder(new HashMap()) - .put("jsonrpc", "2.0") - .put("id", 2) - .put("method", "tools/list") - .put("params", MapUtil.builder(new HashMap()) - .put("withUserTools", true) - .build()) - .build(); + // 存储所有工具列表 + List allTools = new ArrayList<>(); + String cursor = null; - Map requestBody = MapUtil - .builder(new HashMap()) - .put("type", "mcp") - .put("payload", payload) - .build(); - - // 发送请求 - String resultMessage = HttpRequest.post(url) - .header(Header.CONTENT_TYPE, ContentType.JSON.getValue()) - .header(Header.AUTHORIZATION, "Bearer " + generateBearerToken()) - .body(JSONUtil.toJsonStr(requestBody)) - .timeout(10000) // 超时，毫秒 - .execute().body(); - - // 解析响应 - if (StringUtils.isNotBlank(resultMessage)) { - cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(resultMessage); - if (jsonObject.getBool("success", false)) { - return jsonObject.get("data"); + // 循环获取分页数据 + while (true) { + // 构建params + Map paramsMap = MapUtil.builder(new HashMap()) + .put("withUserTools", true) + .build(); + // 如果有cursor，添加到请求参数中 + if (StringUtils.isNotBlank(cursor)) { + paramsMap.put("cursor", cursor); } + + // 构建请求体 + Map payload = MapUtil + .builder(new HashMap()) + .put("jsonrpc", "2.0") + .put("id", 2) + .put("method", "tools/list") + .put("params", paramsMap) + .build(); + + Map requestBody = MapUtil + .builder(new HashMap()) + .put("type", "mcp") + .put("payload", payload) + .build(); + + // 发送请求 + String resultMessage = HttpRequest.post(url) + .header(Header.CONTENT_TYPE, ContentType.JSON.getValue()) + .header(Header.AUTHORIZATION, "Bearer " + generateBearerToken()) + .body(JSONUtil.toJsonStr(requestBody)) + .timeout(10000) // 超时，毫秒 + .execute().body(); + + // 解析响应 + if (StringUtils.isBlank(resultMessage)) { + break; + } + + JSONObject jsonObject = JSONUtil.parseObj(resultMessage); + if (!jsonObject.getBool("success", false)) { + break; + } + + JSONObject data = jsonObject.getJSONObject("data"); + if (data == null) { + break; + } + + // 获取当前页的工具列表 + JSONArray tools = data.getJSONArray("tools"); + if (tools != null && !tools.isEmpty()) { + allTools.addAll(tools); + } + + // 获取下一页的cursor + String nextCursor = data.getStr("nextCursor"); + if (StringUtils.isBlank(nextCursor)) { + // 没有下一页了 + break; + } + cursor = nextCursor; } - return null; + // 构建返回结果 + if (allTools.isEmpty()) { + return null; + } + + Map resultData = new HashMap<>(); + resultData.put("tools", allTools); + return resultData; } @Override From a4583e273a96545fb43158b64018554ab3dfc7e7 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Sun, 1 Mar 2026 12:59:45 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix=EF=BC=9A=E9=87=8D=E5=86=99=20audio=5Fto?= =?UTF-8?q?=5Fopus=5Fdata=5Fstream=20=E6=96=B9=E6=B3=95=EF=BC=8C=E8=AE=A9?= =?UTF-8?q?=E9=9F=B3=E9=A2=91=E6=96=87=E4=BB=B6=E5=A4=84=E7=90=86=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E7=8B=AC=E7=AB=8B=E7=BC=96=E7=A0=81=E5=99=A8=EF=BC=8C?= =?UTF-8?q?=E9=81=BF=E5=85=8D=E4=B8=8E=20TTS=20=E6=B5=81=E7=BC=96=E7=A0=81?= =?UTF-8?q?=E5=99=A8=E5=86=B2=E7=AA=81=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/alibl_stream.py | 19 ++++++++++++++ .../core/providers/tts/aliyun_stream.py | 19 ++++++++++++++ .../providers/tts/huoshan_double_stream.py | 14 +++++++++++ .../core/providers/tts/xunfei_stream.py | 25 ++++++++++++++++--- 4 files changed, 74 insertions(+), 3 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/alibl_stream.py b/main/xiaozhi-server/core/providers/tts/alibl_stream.py index c636d0cb..8974124c 100644 --- a/main/xiaozhi-server/core/providers/tts/alibl_stream.py +++ b/main/xiaozhi-server/core/providers/tts/alibl_stream.py @@ -6,6 +6,7 @@ import queue import asyncio import traceback import websockets +from typing import Callable, Any from asyncio import Task from config.logger import setup_logging from core.utils import opus_encoder_utils @@ -383,6 +384,24 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None + def audio_to_opus_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): + """重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。 + 双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码， + 同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder， + 共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。 + """ + from core.utils.util import audio_to_data_stream + + return audio_to_data_stream( + audio_file_path, + is_opus=True, + callback=callback, + sample_rate=self.conn.sample_rate, + opus_encoder=None, + ) + def to_tts(self, text: str) -> list: """非流式生成音频数据，用于生成音频及测试场景""" try: diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 6026eafa..0a09c65e 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -8,6 +8,7 @@ import time import queue import asyncio import traceback +from typing import Callable, Any from asyncio import Task import websockets import os @@ -463,6 +464,24 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None + def audio_to_opus_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): + """重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。 + 双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码， + 同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder， + 共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。 + """ + from core.utils.util import audio_to_data_stream + + return audio_to_data_stream( + audio_file_path, + is_opus=True, + callback=callback, + sample_rate=self.conn.sample_rate, + opus_encoder=None, + ) + def to_tts(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景""" try: diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index b1ba1bed..eefee383 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -683,6 +683,20 @@ class TTSProvider(TTSProviderBase): ) ) + def audio_to_opus_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): + """重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。 + 双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码， + 同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder， + 共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。 + """ + from core.utils.util import audio_to_data_stream + return audio_to_data_stream( + audio_file_path, is_opus=True, callback=callback, + sample_rate=self.conn.sample_rate, opus_encoder=None + ) + def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) diff --git a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py index e7d4d4e9..e08c5617 100644 --- a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py @@ -9,6 +9,7 @@ import hashlib import asyncio import traceback import websockets +from typing import Callable, Any from asyncio import Task from config.logger import setup_logging from core.utils import opus_encoder_utils @@ -472,9 +473,27 @@ class TTSProvider(TTSProviderBase): return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] - - def _build_base_request(self, status,text=" "): + return [] + + def audio_to_opus_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): + """重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。 + 双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码， + 同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder， + 共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。 + """ + from core.utils.util import audio_to_data_stream + + return audio_to_data_stream( + audio_file_path, + is_opus=True, + callback=callback, + sample_rate=self.conn.sample_rate, + opus_encoder=None, + ) + + def _build_base_request(self, status, text=" "): """构建基础请求结构""" return { "header": {