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 diff --git a/main/xiaozhi-server/core/providers/tts/alibl_stream.py b/main/xiaozhi-server/core/providers/tts/alibl_stream.py index b276c855..8eb0f708 100644 --- a/main/xiaozhi-server/core/providers/tts/alibl_stream.py +++ b/main/xiaozhi-server/core/providers/tts/alibl_stream.py @@ -8,6 +8,7 @@ import traceback import websockets from asyncio import Task +from typing import Callable, Any from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -392,6 +393,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 0e17a229..cf06e7ff 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -13,6 +13,7 @@ import websockets from asyncio import Task from urllib import parse from datetime import datetime +from typing import Callable, Any from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -471,6 +472,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 a24d09b0..152aa33f 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -698,6 +698,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 6ecefc86..be53284e 100644 --- a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py @@ -11,6 +11,7 @@ import traceback import websockets from asyncio import Task +from typing import Callable, Any from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from urllib.parse import urlencode, urlparse @@ -481,9 +482,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": {