From 366bc9e2a2c2e2aa9e392f7cd8ea6ac1052e904b Mon Sep 17 00:00:00 2001 From: ClamJom <3188485441@qq.com> Date: Sat, 14 Feb 2026 19:36:53 +0800 Subject: [PATCH 01/12] fix:knowledge graph bugs --- .../iot/models/KnowledgeGraphicNodeEntity.java | 6 +++--- .../models/KnowledgeGraphicSearchCountEntity.java | 2 +- .../KnowledgeGraphicServiceImpl.java | 14 ++++++++++---- src/main/resources/application.yaml | 3 ++- 4 files changed, 16 insertions(+), 9 deletions(-) diff --git a/src/main/java/top/rslly/iot/models/KnowledgeGraphicNodeEntity.java b/src/main/java/top/rslly/iot/models/KnowledgeGraphicNodeEntity.java index 1e05a5d8..cb3ec329 100644 --- a/src/main/java/top/rslly/iot/models/KnowledgeGraphicNodeEntity.java +++ b/src/main/java/top/rslly/iot/models/KnowledgeGraphicNodeEntity.java @@ -45,15 +45,15 @@ public class KnowledgeGraphicNodeEntity { @Comment("Which user this node belongs to") private int productId; - @Column(name = "hit_times") + @Column(name = "hit_times", columnDefinition = "INT DEFAULT 0") @Comment("Direct hit count since last graph clear") private int hitTimes; - @Column(name = "search_times") + @Column(name = "search_times", columnDefinition = "INT DEFAULT 0") @Comment("Count of how many times this node being searched over") private int searchTimes; - @Column(name = "create_epoch") + @Column(name = "create_epoch", columnDefinition = "INT DEFAULT 0") @Comment("When was this node create, for calculating whether this node should be forgot") private int createEpoch; } diff --git a/src/main/java/top/rslly/iot/models/KnowledgeGraphicSearchCountEntity.java b/src/main/java/top/rslly/iot/models/KnowledgeGraphicSearchCountEntity.java index 47df9cc4..a26f1c63 100644 --- a/src/main/java/top/rslly/iot/models/KnowledgeGraphicSearchCountEntity.java +++ b/src/main/java/top/rslly/iot/models/KnowledgeGraphicSearchCountEntity.java @@ -34,6 +34,6 @@ public class KnowledgeGraphicSearchCountEntity { @Column(name = "product_id") private int productId; - @Column(name = "r_count") + @Column(name = "r_count", columnDefinition = "INT DEFAULT 0") private int count; } diff --git a/src/main/java/top/rslly/iot/services/knowledgeGraphic/KnowledgeGraphicServiceImpl.java b/src/main/java/top/rslly/iot/services/knowledgeGraphic/KnowledgeGraphicServiceImpl.java index 6994b623..dccd7784 100644 --- a/src/main/java/top/rslly/iot/services/knowledgeGraphic/KnowledgeGraphicServiceImpl.java +++ b/src/main/java/top/rslly/iot/services/knowledgeGraphic/KnowledgeGraphicServiceImpl.java @@ -298,8 +298,11 @@ public JsonResult addNode(KnowledgeGraphicNodeEntity node) { @Transactional(rollbackFor = Exception.class) public JsonResult addNode(KnowledgeGraphicNode node) { KnowledgeGraphicNodeEntity nodeDb = knowledgeGraphicNodeRepository.findByName(node.name); - int currentEpoch = - knowledgeGraphicSearchCountRepository.getTopByProductId(node.productId).getCount(); + KnowledgeGraphicSearchCountEntity searchCountEntity = knowledgeGraphicSearchCountRepository.getTopByProductId(node.getProductId()); + int currentEpoch = 0; + if (searchCountEntity != null) { + currentEpoch = searchCountEntity.getCount(); + } if (nodeDb == null) { nodeDb = new KnowledgeGraphicNodeEntity(); nodeDb.setName(node.name); @@ -320,8 +323,11 @@ public JsonResult addNode(KnowledgeGraphicNode node) { public JsonResult addNode(String name, String des, int productId) { KnowledgeGraphicNodeEntity node = knowledgeGraphicNodeRepository.findByNameAndProductId(name, productId); - int currentEpoch = - knowledgeGraphicSearchCountRepository.getTopByProductId(productId).getCount(); + KnowledgeGraphicSearchCountEntity searchCountEntity = knowledgeGraphicSearchCountRepository.getTopByProductId(productId); + int currentEpoch = 0; + if (searchCountEntity != null) { + currentEpoch = searchCountEntity.getCount(); + } String historyDes; boolean needEmbedding = false; if (node == null) { diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 7c0732db..6e41cbe2 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -34,7 +34,8 @@ spring: url: jdbc:mysql://localhost:3306/cwliot1.8?useUnicode=true&characterEncoding=utf8&useSSL=false&allowPublicKeyRetrieval=true driver-class-name: com.mysql.cj.jdbc.Driver # 输入你自己的mysql密码,该密码为随机生成 - password: "XzHvhX4CDaN696oQAXdmlcsrqgWbkxRl" +# password: "XzHvhX4CDaN696oQAXdmlcsrqgWbkxRl" + password: "c.3188485441" type: com.alibaba.druid.pool.DruidDataSource druid: # 下面为连接池的补充设置，应用到上面所有数据源中 From 88bcdf1123a343470868bad4d197d7156ec5650a Mon Sep 17 00:00:00 2001 From: ClamJom <3188485441@qq.com> Date: Fri, 27 Mar 2026 16:53:17 +0800 Subject: [PATCH 02/12] fix:refactored part of XiaoZhiUtil --- .../iot/utility/smartVoice/XiaoZhiUtil.java | 207 +++++++++++++++++- .../utility/smartVoice/XiaoZhiWebsocket.java | 10 +- 2 files changed, 208 insertions(+), 9 deletions(-) diff --git a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java index 2fadd8fb..0a761af7 100644 --- a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java +++ b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java @@ -22,8 +22,10 @@ import cn.hutool.captcha.generator.RandomGenerator; import com.alibaba.fastjson.JSONObject; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import top.rslly.iot.models.AdminConfigEntity; @@ -43,6 +45,8 @@ import jakarta.annotation.PreDestroy; import jakarta.websocket.Session; +import top.rslly.iot.utility.ai.voice.concentus.OpusException; + import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.file.Files; @@ -71,6 +75,8 @@ public class XiaoZhiUtil { private AdminConfigServiceImpl adminConfigService; @Autowired private ProductAsrRepository productAsrRepository; + @Autowired + private RedisTemplate redisTemplate; @Value("${ai.vision-explain-url}") private String visionExplainUrl; @Value("${ai.tts.skip-tool-prefix:true}") @@ -144,6 +150,190 @@ public void destroyMcp(String chatId) { mcpProtocolDeal.destroyMcp(McpWebsocket.DEVICE_SERVER_NAME, chatId); } + /** + * ASR处理，同步方法，解码音频二进制流 + * @param audioList 音频流 + * @param detect 检测文本 + * @return 解码后的数据或Null + * @throws OpusException Opus异常 + */ + public String ASRHandler(List audioList, int productId, String... detect) throws OpusException, IOException { + // 数据太短 + if(audioList.size() <= 20 && detect.length == 0) return "<|2SRT|>"; + // 如果ASR服务没有启动成功，那么应当字节返回空 + if(asrServiceFactory == null) return null; + // 已有检测文本 + if(detect.length != 0) return detect[0]; + var asrService = asrServiceFactory.getService(getProductAsrProvider(productId)); + if(asrService == null) return null; + // 构造解码器 + OpusDecoder decoder = new OpusDecoder(16000, 1); + // 通过字节列表输出流构造字节列表并写入文件，完成WAV->PCM的转换 + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + for(byte[] bytes: audioList){ + if(bytes == null) continue; + byte[] data_packet = new byte[16000]; + int pcm_frame = decoder.decode(bytes, 0, bytes.length, + data_packet, 0, 960, false); + bos.write(data_packet, 0, pcm_frame * 2); + } + // 生成WAV文件 + Path tempFile = Files.createTempFile("audio_", ".wav"); + Files.write(tempFile, bos.toByteArray()); + bos.close(); + // 读取WAV文件并使用ASR服务提取文本返回 + String ret = asrService.getTextRealtime(tempFile.toFile(), 16000, "pcm"); + Files.deleteIfExists(tempFile); + return ret; + } + + /** + * 消息太短或无法解析时的回复 + * @param chatId 聊天的ID + */ + private void sendForUnclearMsg(String chatId){ + XiaoZhiWebsocket.send(chatId, "{\"type\":\"stt\",\"text\":\"" + "没听清楚，说太快了" + "\"}"); + } + + /** + * 发送助手正在思考的信息 + * @param chatId 聊天ID + */ + private void sendWhenThinking(String chatId){ + XiaoZhiWebsocket.send(chatId, "{\"type\": \"tts\", \"state\": \"sentence_start\"," + + " \"text\": \"智能助手思考中" + "\"}"); + JSONObject emotionObject = new JSONObject(); + emotionObject.put("type", "llm"); + emotionObject.put("text", "🤔"); + emotionObject.put("emotion", "thinking"); + XiaoZhiWebsocket.send(chatId, emotionObject.toJSONString()); + } + + private void sendTTSStart(String chatId){ + XiaoZhiWebsocket.send(chatId, "{\"type\":\"tts\",\"state\":\"start\"}"); + } + + /** + * 发送结束信息 + * @param chatId 聊天ID + */ + private void sendEndMsg(String chatId){ + XiaoZhiWebsocket.send(chatId, "{\"type\":\"tts\",\"state\":\"stop\"}"); + } + + private void clearAudioHandlers(String chatId){ + XiaoZhiWebsocket.haveVoice.put(chatId, false); + XiaoZhiWebsocket.isAbort.put(chatId, false); + Router.queueMap.remove(chatId); + this.sendEndMsg(chatId); + } + + @Async("taskExecutor") + public void dealWthAudio2(List audioList, String chatId, int productId, boolean isManual, + String... detect) throws InterruptedException { + if(audioList == null || chatId == null){ + log.error("audioList或chatId为空，audioList: {}, chatId: {}", audioList, chatId); + return; + } + String text = ""; + try { + text = ASRHandler(audioList, productId, detect); + }catch(OpusException e){ + log.error("Opus音频解码错误: {}", e.getMessage()); + return; + }catch(Exception e){ + log.error("ASR提取失败！错误：{}", e.getMessage()); + return; + } + // 如果ASR处理结果为空，通知用户并返回 + if(StringUtils.isEmpty(text)){ + this.sendForUnclearMsg(chatId); + this.sendEndMsg(chatId); + return; + } + // 如果音频太短，通知用户并返回 + if(text.equals("<|2SRT|>")){ + // 保留最后的10帧信息 + int keepFrames = Math.min(10, audioList.size()); // 安全处理边界 + if (audioList.size() > keepFrames) { + audioList.subList(0, audioList.size() - keepFrames).clear(); + } + this.sendForUnclearMsg(chatId); + } + final String tCopy = text; + XiaoZhiWebsocket.voiceContent.computeIfPresent(chatId, (k, v) -> v + tCopy); + XiaoZhiWebsocket.voiceContent.putIfAbsent(chatId, text); + XiaoZhiWebsocket.send(chatId, text); + audioList.clear(); + // 到这里，voiceContent不可能为空 + sendWhenThinking(chatId); + // 下面是TTS的部分 + String voiceContent = XiaoZhiWebsocket.voiceContent.get(chatId); + Map emotionMessage = new HashMap<>(); + emotionMessage.put("chatId", chatId); + var emotionRes = emotionToolAsync.run(voiceContent, emotionMessage); + CompletableFuture res = null; + // 首先获取router的结果 + if(router != null){ + res = CompletableFuture.supplyAsync( + () -> router.response(voiceContent, chatId, productId), + routerExecutor + ); + } + // 结果字符串构造器 + StringBuilder answerSB = new StringBuilder(); + // 用于存储回复句子 + List answerList = new ArrayList<>(); + boolean emotionFlag = false; + if(isManual){ + this.sendTTSStart(chatId); + } + while(res != null && !res.isDone() || + Router.queueMap.containsKey(chatId) && !Router.queueMap.get(chatId).isEmpty()){ + // 表情处理模块 + if(emotionRes != null&& emotionRes.isDone() && !emotionFlag){ + try { + Map emotionResult = emotionRes.get(); + if(emotionResult == null) continue; + JSONObject emotionObject = new JSONObject(); + emotionObject.put("type", "llm"); + emotionObject.put("text", emotionResult.get("emoji")); + emotionObject.put("emotion", emotionResult.get("text")); + XiaoZhiWebsocket.send(chatId, emotionObject.toJSONString()); + }catch(Exception e){ + log.error("处理表情响应出错：{}", e.getMessage()); + } + } + // 处理终止 + if(XiaoZhiWebsocket.isAbort.getOrDefault(chatId, false)){ + XiaoZhiWebsocket.haveVoice.put(chatId, false); + XiaoZhiWebsocket.isAbort.put(chatId, false); + Router.queueMap.remove(chatId); + this.sendEndMsg(chatId); + } + // SSE元素处理 + String element = Router.queueMap.get(chatId).poll(); + if(element == null){ + Thread.sleep(10); + continue; + } + if(element.equals("[DONE]")){ + // 已经抵达最后一帧 + if(!answerSB.isEmpty()){ + // 将之前累计的元素入队并交由转化线程处理， + answerList.add(answerSB.toString()); + // TODO: 异步TTS处理 + } + break; + }else{ + // TODO: 异步TTS处理 + // 判断是否存在标点，如果存在，则从将标点之前的已缓存元素构造字符串并入队， + // 随后清空字符串构造器，将标点之后的部分存入构造器；如果不存在，则将整个元素 + // 加入构造器 + } + } + } + @Async("taskExecutor") public void dealWithAudio(List audioList, String chatId, int productId, boolean isManual, String... detect) @@ -166,7 +356,7 @@ public void dealWithAudio(List audioList, String chatId, int productId, }"""); } - StringBuilder sentences = new StringBuilder(""); + StringBuilder sentences = new StringBuilder(); if (detect.length == 0) { // 安全读取字节数据 ByteArrayOutputStream bos = new ByteArrayOutputStream(); @@ -199,10 +389,10 @@ public void dealWithAudio(List audioList, String chatId, int productId, } else { sentences.append(detect[0]); } - if (sentences.length() > 0) { + if (!sentences.isEmpty()) { if (XiaoZhiWebsocket.voiceContent.containsKey(chatId) && XiaoZhiWebsocket.voiceContent.get(chatId) != null - && XiaoZhiWebsocket.voiceContent.get(chatId).length() > 0) { + && !XiaoZhiWebsocket.voiceContent.get(chatId).isEmpty()) { XiaoZhiWebsocket.voiceContent.put(chatId, XiaoZhiWebsocket.voiceContent.get(chatId) + sentences); } else { @@ -239,7 +429,7 @@ public void dealWithAudio(List audioList, String chatId, int productId, } if (XiaoZhiWebsocket.voiceContent.containsKey(chatId) && XiaoZhiWebsocket.voiceContent.get(chatId) != null - && XiaoZhiWebsocket.voiceContent.get(chatId).length() > 0) { + && !XiaoZhiWebsocket.voiceContent.get(chatId).isEmpty()) { Session session = XiaoZhiWebsocket.clients.get(chatId); if (session != null && session.isOpen()) { if (showThinking) { @@ -285,7 +475,7 @@ public void dealWithAudio(List audioList, String chatId, int productId, } } while ((res != null && !res.isDone()) || Router.queueMap.containsKey(chatId) - && Router.queueMap.get(chatId).size() > 0) { + && !Router.queueMap.get(chatId).isEmpty()) { if (emotionRes != null && emotionRes.isDone() && !emotionFlag) { try { Map emotionResult = emotionRes.get(); @@ -321,7 +511,7 @@ public void dealWithAudio(List audioList, String chatId, int productId, if (Router.queueMap.containsKey(chatId)) { String element = Router.queueMap.get(chatId).poll(); if (element != null && element.equals("[DONE]")) { - if (answerBuilder.length() > 0) { + if (!answerBuilder.isEmpty()) { // 立即发送已累积的内容 JSONObject jsonObject = new JSONObject(); jsonObject.put("type", "tts"); @@ -355,7 +545,8 @@ public void dealWithAudio(List audioList, String chatId, int productId, XiaoZhiWebsocket.isAbort.put(chatId, false); Router.queueMap.remove(chatId); return; - } else if (element != null) { + } + else if (element != null) { element = element.replace("\n", ""); // 查找字符串中的第一个标点位置 int punctuationIndex = -1; @@ -447,7 +638,7 @@ public void dealWithAudio(List audioList, String chatId, int productId, if (res != null) { answer = res.get(); } - if (answer == null || answer.equals("")) { + if (StringUtils.isEmpty(answer)) { answer = "抱歉，我暂时无法理解您的问题。"; } answer = resolveFinalVoiceAnswer(answer, answerBuilder.toString(), queuePlaybackDelivered); diff --git a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiWebsocket.java b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiWebsocket.java index 8d43dd2a..55792f1e 100644 --- a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiWebsocket.java +++ b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiWebsocket.java @@ -611,5 +611,13 @@ private void closeSessionForIdleTimeout(long currentTime) throws IOException { isAbort.put(chatId, false); } - + public static void send(String chatId, String msg){ + Session session = XiaoZhiWebsocket.clients.get(chatId); + if(session == null || !session.isOpen()) return; + try { + session.getBasicRemote().sendText(msg); + }catch(IOException e){ + log.error("发送消息失败！"); + } + } } From 522284765a1d6508aa6d4099f12b6669dad01583 Mon Sep 17 00:00:00 2001 From: ClamJom <3188485441@qq.com> Date: Wed, 1 Apr 2026 17:25:00 +0800 Subject: [PATCH 03/12] feat:refactoring audio handler --- .../top/rslly/iot/config/RedisConfig.java | 1 + .../iot/utility/ai/voice/TTS/EdgeTTs.java | 58 ++++--- .../ai/voice/TTS/MiniMaxTtsService.java | 64 ++++--- .../iot/utility/ai/voice/TTS/Text2audio.java | 4 + .../iot/utility/ai/voice/TTS/TtsService.java | 5 + .../ai/voice/TTS/TtsServiceFactory.java | 35 ++++ .../iot/utility/smartVoice/XiaoZhiUtil.java | 164 +++++++++++++----- 7 files changed, 244 insertions(+), 87 deletions(-) diff --git a/src/main/java/top/rslly/iot/config/RedisConfig.java b/src/main/java/top/rslly/iot/config/RedisConfig.java index 0147714c..e8b0b476 100644 --- a/src/main/java/top/rslly/iot/config/RedisConfig.java +++ b/src/main/java/top/rslly/iot/config/RedisConfig.java @@ -29,6 +29,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.*; import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @Configuration diff --git a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/EdgeTTs.java b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/EdgeTTs.java index abb7bc93..bd11a7c3 100644 --- a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/EdgeTTs.java +++ b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/EdgeTTs.java @@ -30,6 +30,7 @@ import jakarta.websocket.Session; import java.nio.file.Files; import java.nio.file.Paths; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -43,8 +44,27 @@ public class EdgeTTs implements TtsService { @Override public void websocketAudioSync(String text, Float pitch, Float speed, Session session, String chatId, String voice) { - // Only used for WebSocket audio sending. + List audioList = getTextAudio(chatId, text, pitch, speed, voice); final BlockingQueue audioQueue = new LinkedBlockingQueue<>(); + for(byte[] b: audioList){ + audioQueue.offer(b); + } + try{ + AudioUtils.asyncSendAudioQueue(chatId, session, audioQueue); + } catch (Exception e) { + log.error("websocketAudio error for chatId: {}", chatId, e); + } + } + + @Override + public void asyncSynthesizeAndSaveAudio(String text, String chatId) { + throw new UnsupportedOperationException(); + } + + @Override + public List getTextAudio(String chatId, String text, Float pitch, Float speed, String voice) { + // Only used for WebSocket audio sending. + List audioList = new ArrayList<>(); // End-of-stream marker: an empty byte array. final byte[] EOS = new byte[0]; final OpusEncoderUtils encoder = new OpusEncoderUtils(16000, 1, 60); @@ -65,20 +85,20 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se // 获取中文语音 String finalVoiceName = voiceName; Voice voiceObj = TTSVoice.provides().stream() - .filter(v -> v.getShortName().equals(finalVoiceName)) - .collect(Collectors.toList()).get(0); + .filter(v -> v.getShortName().equals(finalVoiceName)) + .collect(Collectors.toList()).get(0); TTS ttsEngine = new TTS(voiceObj, text); // 执行TTS转换获取音频文件 String outputPath = System.getProperty("java.io.tmpdir"); String audioFilePath = ttsEngine.findHeadHook() - .storage(outputPath) - .isRateLimited(true) - .voicePitch(pitchHz + "Hz") - .voiceRate(ratePercent + "%") - .overwrite(false) - .formatMp3() - .trans(); + .storage(outputPath) + .isRateLimited(true) + .voicePitch(pitchHz + "Hz") + .voiceRate(ratePercent + "%") + .overwrite(false) + .formatMp3() + .trans(); // 使用 Paths.get 来正确拼接路径，解决缺少分隔符的问题 fullPath = Paths.get(outputPath, audioFilePath).toString(); @@ -86,17 +106,13 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se // 1. 将MP3转换为PCM (已经设置为16kHz采样率和单声道) byte[] pcmData = AudioUtils.convertMp3ToPcm(fullPath); List packets = encoder.encodePcmToOpus(pcmData, false); - for (byte[] packet : packets) { - audioQueue.offer(packet); - } + audioList.addAll(packets); packets = encoder.encodePcmToOpus(new byte[0], true); - for (byte[] packet : packets) { - audioQueue.offer(packet); - } + audioList.addAll(packets); // Signal end-of-stream by adding an empty array. - audioQueue.offer(EOS); + audioList.add(EOS); - AudioUtils.asyncSendAudioQueue(chatId, session, audioQueue); + return audioList; } catch (Exception e) { log.error("websocketAudio error for chatId: {}", chatId, e); @@ -110,10 +126,6 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se } } } - } - - @Override - public void asyncSynthesizeAndSaveAudio(String text, String chatId) { - throw new UnsupportedOperationException(); + return null; } } diff --git a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/MiniMaxTtsService.java b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/MiniMaxTtsService.java index 5857e7b0..b791156b 100644 --- a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/MiniMaxTtsService.java +++ b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/MiniMaxTtsService.java @@ -37,6 +37,7 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Paths; +import java.util.ArrayList; import java.util.Base64; import java.util.List; import java.util.concurrent.BlockingQueue; @@ -70,8 +71,33 @@ public class MiniMaxTtsService implements TtsService { @Override public void websocketAudioSync(String text, Float pitch, Float speed, Session session, String chatId, String voice) { + List audioList = getTextAudio(chatId, text, pitch, speed, voice); // Only used for WebSocket audio sending. final BlockingQueue audioQueue = new LinkedBlockingQueue<>(); + for(byte[] b : audioList){ + audioQueue.offer(b); + } + try{ + // 异步发送音频队列 + AudioUtils.asyncSendAudioQueue(chatId, session, audioQueue); + } catch (Exception e) { + log.error("MiniMax TTS error for chatId: {}", chatId, e); + } + } + + /** + * 获取文本的Opus音频字节流 + * @param chatId 对话ID + * @param text 文本 + * @param pitch 语调 + * @param speed 语速 + * @param voice 声音类型 + * @return Null或一个转载有Opus字节流的BlockingQueue + */ + @Override + public List getTextAudio(String chatId, String text, Float pitch, Float speed, String voice) { + // Only used for WebSocket audio sending. + List audioList = new ArrayList<>(); // End-of-stream marker: an empty byte array. final byte[] EOS = new byte[0]; final OpusEncoderUtils encoder = new OpusEncoderUtils(16000, 1, 60); @@ -96,8 +122,8 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se int responseCode = connection.getResponseCode(); if (responseCode != 200) { BufferedReader errorReader = - new BufferedReader( - new InputStreamReader(connection.getErrorStream(), StandardCharsets.UTF_8)); + new BufferedReader( + new InputStreamReader(connection.getErrorStream(), StandardCharsets.UTF_8)); StringBuilder errorResponse = new StringBuilder(); String line; while ((line = errorReader.readLine()) != null) { @@ -105,17 +131,17 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se } errorReader.close(); log.error("MiniMax TTS API error for voice '{}': {} - {}", voice, responseCode, - errorResponse); - return; + errorResponse); + return null; } log.info("MiniMax TTS API request successful for voice: {}, text length: {}", voice, - text.length()); + text.length()); // 读取流式响应 ByteArrayOutputStream audioBuffer = new ByteArrayOutputStream(); try (BufferedReader reader = new BufferedReader( - new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) { + new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) { String line; while ((line = reader.readLine()) != null) { if (line.startsWith("data: ")) { @@ -130,9 +156,9 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se // 检查错误状态 JsonNode baseResp = root.path("base_resp"); if (baseResp.has("status_code") - && baseResp.get("status_code").asInt() != 0) { + && baseResp.get("status_code").asInt() != 0) { log.error("MiniMax TTS API returned error: {}", - baseResp.path("status_msg").asText()); + baseResp.path("status_msg").asText()); continue; } @@ -161,12 +187,12 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se byte[] mp3Data = audioBuffer.toByteArray(); if (mp3Data.length == 0) { log.warn( - "MiniMax TTS returned empty audio data for voice: '{}', model: '{}', text length: {}", - voice, model != null && !model.isBlank() ? model : DEFAULT_MODEL, text.length()); - return; + "MiniMax TTS returned empty audio data for voice: '{}', model: '{}', text length: {}", + voice, model != null && !model.isBlank() ? model : DEFAULT_MODEL, text.length()); + return null; } log.info("MiniMax TTS generated {} bytes of audio data for voice: '{}'", mp3Data.length, - voice); + voice); String outputPath = System.getProperty("java.io.tmpdir"); // 将 chatId 中的冒号替换为下划线，避免 Windows 路径非法字符问题 @@ -179,21 +205,16 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se // 编码为 Opus 并发送到队列 List packets = encoder.encodePcmToOpus(pcmData, false); - for (byte[] packet : packets) { - audioQueue.offer(packet); - } + audioList.addAll(packets); // 刷新编码器 packets = encoder.encodePcmToOpus(new byte[0], true); - for (byte[] packet : packets) { - audioQueue.offer(packet); - } + audioList.addAll(packets); // 信号结束 - audioQueue.offer(EOS); + audioList.add(EOS); - // 异步发送音频队列 - AudioUtils.asyncSendAudioQueue(chatId, session, audioQueue); + return audioList; } catch (Exception e) { log.error("MiniMax TTS error for chatId: {}", chatId, e); @@ -208,6 +229,7 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se } } } + return null; } @Override diff --git a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/Text2audio.java b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/Text2audio.java index 99f7e434..f0a638d9 100644 --- a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/Text2audio.java +++ b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/Text2audio.java @@ -210,5 +210,9 @@ public void websocketAudioSync(String text, Float pitch, Float speed, Session se } } + @Override + public List getTextAudio(String chatId, String text, Float pitch, Float speed, String voice) { + return null; + } } diff --git a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsService.java b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsService.java index 11c51e86..8795b282 100644 --- a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsService.java +++ b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsService.java @@ -21,9 +21,14 @@ import jakarta.websocket.Session; +import java.util.List; +import java.util.concurrent.BlockingQueue; + public interface TtsService { void websocketAudioSync(String text, Float pitch, Float speed, Session session, String chatId, String voice); void asyncSynthesizeAndSaveAudio(String text, String chatId); + + List getTextAudio(String chatId, String text, Float pitch, Float speed, String voice); } diff --git a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsServiceFactory.java b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsServiceFactory.java index 4fbfbdf7..7ed38129 100644 --- a/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsServiceFactory.java +++ b/src/main/java/top/rslly/iot/utility/ai/voice/TTS/TtsServiceFactory.java @@ -27,6 +27,9 @@ import jakarta.websocket.Session; +import java.util.List; +import java.util.concurrent.BlockingQueue; + @Component public class TtsServiceFactory { @Autowired @@ -49,6 +52,7 @@ public TtsService getTtsService(String type) { return text2audio; } + @Deprecated public void websocketAudioSync(String text, Session session, String chatId, int productId) { String provider = "dashscope"; // 语音音调 (0.5-2.0) @@ -79,4 +83,35 @@ public void websocketAudioSync(String text, Session session, String chatId, int TtsService ttsService = getTtsService(provider); ttsService.websocketAudioSync(text, pitch, speed, session, chatId, voice); } + + public List getTextAudio(String chatId, String text, int productId){ + String provider = "dashscope"; + // 语音音调 (0.5-2.0) + float pitch = 1.0f; + + // 语音语速 (0.5-2.0) + float speed = 1.0f; + String voice = null; + try { + var roles = productRoleService.findAllByProductId(productId); + if (!roles.isEmpty() && roles.get(0).getVoice() != null) { + voice = roles.get(0).getVoice(); + if (voice.startsWith("edge-")) { + provider = "edge"; + voice = voice.substring(5); + } else if (voice.startsWith("minimax-")) { + provider = "minimax"; + voice = voice.substring(8); + } + } + var voiceDiyEntityList = productVoiceDiyService.findAllByProductId(productId); + if (!voiceDiyEntityList.isEmpty()) { + pitch = Float.parseFloat(voiceDiyEntityList.get(0).getPitch()); + speed = Float.parseFloat(voiceDiyEntityList.get(0).getSpeed()); + } + } catch (Exception ignored) { + } + TtsService ttsService = getTtsService(provider); + return ttsService.getTextAudio(chatId, text, pitch, speed, voice); + } } diff --git a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java index 0a761af7..61c8f377 100644 --- a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java +++ b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java @@ -20,6 +20,7 @@ package top.rslly.iot.utility.smartVoice; import cn.hutool.captcha.generator.RandomGenerator; +import com.alibaba.druid.sql.visitor.functions.Char; import com.alibaba.fastjson.JSONObject; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; @@ -76,7 +77,9 @@ public class XiaoZhiUtil { @Autowired private ProductAsrRepository productAsrRepository; @Autowired - private RedisTemplate redisTemplate; + private RedisTemplate bytesRedisTemplate; + @Autowired + private RedisTemplate redisStateTemplate; // 用于记录TTS段落的处理状态 @Value("${ai.vision-explain-url}") private String visionExplainUrl; @Value("${ai.tts.skip-tool-prefix:true}") @@ -228,46 +231,34 @@ private void clearAudioHandlers(String chatId){ this.sendEndMsg(chatId); } - @Async("taskExecutor") - public void dealWthAudio2(List audioList, String chatId, int productId, boolean isManual, - String... detect) throws InterruptedException { - if(audioList == null || chatId == null){ - log.error("audioList或chatId为空，audioList: {}, chatId: {}", audioList, chatId); - return; + private int getPunctuationPos(String str){ + Set punctuationSet = new HashSet<>(); + String punctuations = "?!:;~.,？！：；~。，"; + for(char c : punctuations.toCharArray()){ + punctuationSet.add(c); } - String text = ""; - try { - text = ASRHandler(audioList, productId, detect); - }catch(OpusException e){ - log.error("Opus音频解码错误: {}", e.getMessage()); - return; - }catch(Exception e){ - log.error("ASR提取失败！错误：{}", e.getMessage()); - return; + for(int i = 0; i < str.length(); i++){ + char c = str.charAt(i); + if(punctuationSet.contains(c)) return i; } - // 如果ASR处理结果为空，通知用户并返回 - if(StringUtils.isEmpty(text)){ - this.sendForUnclearMsg(chatId); - this.sendEndMsg(chatId); - return; - } - // 如果音频太短，通知用户并返回 - if(text.equals("<|2SRT|>")){ - // 保留最后的10帧信息 - int keepFrames = Math.min(10, audioList.size()); // 安全处理边界 - if (audioList.size() > keepFrames) { - audioList.subList(0, audioList.size() - keepFrames).clear(); - } - this.sendForUnclearMsg(chatId); - } - final String tCopy = text; - XiaoZhiWebsocket.voiceContent.computeIfPresent(chatId, (k, v) -> v + tCopy); - XiaoZhiWebsocket.voiceContent.putIfAbsent(chatId, text); - XiaoZhiWebsocket.send(chatId, text); - audioList.clear(); - // 到这里，voiceContent不可能为空 - sendWhenThinking(chatId); - // 下面是TTS的部分 + return -1; + } + + private void asyncTTS(String chatId, String src){ + // 使用Thread.ofVirtual启动这个部分 + // 在Redis中写入当前句子的处理状态为false，表示开始处理 + // 将处理得到的结果转为String并存入Redis，并将当前句子的处理状态置为true以表示完成 + redisStateTemplate.opsForHash().put(chatId, src, false); + } + + /** + * 流式返回的处理方法 + * @param chatId 对话ID + * @param productId 产品ID + * @param isManual 是否对讲机模式 + * @throws InterruptedException 由Thread.sleep抛出的异常 + */ + private void handlerStreamRsp(String chatId, int productId, boolean isManual) throws InterruptedException{ String voiceContent = XiaoZhiWebsocket.voiceContent.get(chatId); Map emotionMessage = new HashMap<>(); emotionMessage.put("chatId", chatId); @@ -283,7 +274,7 @@ public void dealWthAudio2(List audioList, String chatId, int productId, // 结果字符串构造器 StringBuilder answerSB = new StringBuilder(); // 用于存储回复句子 - List answerList = new ArrayList<>(); + Queue answerList = new ArrayDeque<>(); boolean emotionFlag = false; if(isManual){ this.sendTTSStart(chatId); @@ -321,19 +312,106 @@ public void dealWthAudio2(List audioList, String chatId, int productId, // 已经抵达最后一帧 if(!answerSB.isEmpty()){ // 将之前累计的元素入队并交由转化线程处理， - answerList.add(answerSB.toString()); - // TODO: 异步TTS处理 + // 将构造器中缓存的部分交由虚拟线程处理 + String sentence = answerSB.toString(); + // 不允许处理空字符串 + if(StringUtils.isEmpty(sentence)) break; + answerList.add(sentence); + Thread.ofVirtual().start(()->{ + this.asyncTTS(chatId, sentence); + }); } break; }else{ - // TODO: 异步TTS处理 // 判断是否存在标点，如果存在，则从将标点之前的已缓存元素构造字符串并入队， // 随后清空字符串构造器，将标点之后的部分存入构造器；如果不存在，则将整个元素 // 加入构造器 + element = element.replace("\n", ""); + int pIdx = this.getPunctuationPos(element); + if(pIdx != -1){ + String eBefore = element.substring(0, pIdx + 1); + String eAfter = element.substring(pIdx); + answerSB.append(eBefore); + String before = answerSB.toString(); + // 不允许向结果列表中写入空字符串 + if(StringUtils.isEmpty(before)) continue; + answerList.add(before); + // 将标点前的部分交由异步虚拟线程处理 + Thread.ofVirtual().start(()->{ + this.asyncTTS(chatId, before); + }); + answerSB.setLength(0); + answerSB.append(eAfter); + }else + answerSB.append(element); + } + } + String crtS = answerList.poll(); + // 如果结果列表里第一条就是空的字符串，那么就直接退出 + if(StringUtils.isEmpty(crtS)) return; + while(!answerList.isEmpty()){ + if(!redisStateTemplate.opsForHash().hasKey(chatId, crtS)) continue; + Object state = redisStateTemplate.opsForHash().get(chatId, crtS); + if(state == null || !(boolean)state) continue; + // TODO: 发送并重新提取下一条 + } + } + + @Async("taskExecutor") + public void dealWithAudio2(List audioList, String chatId, int productId, boolean isManual, + String... detect) { + if(audioList == null || chatId == null){ + log.error("audioList或chatId为空，audioList: {}, chatId: {}", audioList, chatId); + return; + } + String text = ""; + try { + text = ASRHandler(audioList, productId, detect); + }catch(OpusException e){ + log.error("Opus音频解码错误: {}", e.getMessage()); + return; + }catch(Exception e){ + log.error("ASR提取失败！错误：{}", e.getMessage()); + return; + } + // 如果ASR处理结果为空，通知用户并返回 + if(StringUtils.isEmpty(text)){ + this.sendForUnclearMsg(chatId); + this.sendEndMsg(chatId); + return; + } + // 如果音频太短，通知用户并返回 + if(text.equals("<|2SRT|>")){ + // 保留最后的10帧信息 + int keepFrames = Math.min(10, audioList.size()); // 安全处理边界 + if (audioList.size() > keepFrames) { + audioList.subList(0, audioList.size() - keepFrames).clear(); + } + this.sendForUnclearMsg(chatId); + this.sendEndMsg(chatId); + return; + } + final String tCopy = text; + XiaoZhiWebsocket.voiceContent.computeIfPresent(chatId, (k, v) -> v + tCopy); + XiaoZhiWebsocket.voiceContent.putIfAbsent(chatId, text); + XiaoZhiWebsocket.send(chatId, text); + audioList.clear(); + // 到这里，voiceContent不可能为空 + sendWhenThinking(chatId); + // 下面是TTS的部分 + boolean isStreamRsp = Router.queueMap.containsKey(chatId) && !Router.queueMap.get(chatId).isEmpty(); + if(isStreamRsp){ + try { + this.handlerStreamRsp(chatId, productId, isManual); + return; + }catch(InterruptedException e){ + log.error("音频处理线程出错", e); } } + // 非流式的处理方法 } + @Deprecated @Async("taskExecutor") public void dealWithAudio(List audioList, String chatId, int productId, boolean isManual, String... detect) From 8d689c49ab92cca9e5ff00183006748221ae168b Mon Sep 17 00:00:00 2001 From: ClamJom <3188485441@qq.com> Date: Fri, 3 Apr 2026 14:25:10 +0800 Subject: [PATCH 04/12] feat:refactored audio handler --- .../iot/utility/smartVoice/XiaoZhiUtil.java | 81 ++++++++++++++++--- 1 file changed, 72 insertions(+), 9 deletions(-) diff --git a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java index 61c8f377..880843f9 100644 --- a/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java +++ b/src/main/java/top/rslly/iot/utility/smartVoice/XiaoZhiUtil.java @@ -41,6 +41,7 @@ import top.rslly.iot.utility.ai.tools.EmotionToolAsync; import top.rslly.iot.utility.ai.tools.ToolPrefix; import top.rslly.iot.utility.ai.voice.ASR.AsrServiceFactory; +import top.rslly.iot.utility.ai.voice.AudioUtils; import top.rslly.iot.utility.ai.voice.TTS.TtsServiceFactory; import top.rslly.iot.utility.ai.voice.concentus.OpusDecoder; @@ -50,12 +51,13 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.math.BigInteger; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.security.MessageDigest; import java.util.*; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.*; @Component @Slf4j @@ -244,11 +246,29 @@ private int getPunctuationPos(String str){ return -1; } - private void asyncTTS(String chatId, String src){ + public static String getShortHash(String input, int bytes) { + try { + MessageDigest md = MessageDigest.getInstance("MD5"); // 或 "SHA-1" + byte[] digest = md.digest(input.getBytes(StandardCharsets.UTF_8)); + // 截取前 bytes 个字节 + byte[] truncated = new byte[bytes]; + System.arraycopy(digest, 0, truncated, 0, bytes); + // 转为十六进制字符串 + return new BigInteger(1, truncated).toString(16); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private void asyncTTS(String chatId, String src, int productId){ // 使用Thread.ofVirtual启动这个部分 // 在Redis中写入当前句子的处理状态为false，表示开始处理 // 将处理得到的结果转为String并存入Redis，并将当前句子的处理状态置为true以表示完成 redisStateTemplate.opsForHash().put(chatId, src, false); + List audioBytes = ttsServiceFactory.getTextAudio(chatId, src, productId); + String srcHash = getShortHash(src, 16); + bytesRedisTemplate.opsForList().leftPushAll(srcHash, audioBytes); + redisStateTemplate.opsForHash().put(chatId, src, true); } /** @@ -318,7 +338,7 @@ private void handlerStreamRsp(String chatId, int productId, boolean isManual) th if(StringUtils.isEmpty(sentence)) break; answerList.add(sentence); Thread.ofVirtual().start(()->{ - this.asyncTTS(chatId, sentence); + this.asyncTTS(chatId, sentence, productId); }); } break; @@ -338,7 +358,7 @@ private void handlerStreamRsp(String chatId, int productId, boolean isManual) th answerList.add(before); // 将标点前的部分交由异步虚拟线程处理 Thread.ofVirtual().start(()->{ - this.asyncTTS(chatId, before); + this.asyncTTS(chatId, before, productId); }); answerSB.setLength(0); answerSB.append(eAfter); @@ -353,12 +373,50 @@ private void handlerStreamRsp(String chatId, int productId, boolean isManual) th if(!redisStateTemplate.opsForHash().hasKey(chatId, crtS)) continue; Object state = redisStateTemplate.opsForHash().get(chatId, crtS); if(state == null || !(boolean)state) continue; - // TODO: 发送并重新提取下一条 + BlockingQueue sendQueue = new LinkedBlockingQueue<>(); + String srcHash = getShortHash(crtS, 16); + byte[] bytes = bytesRedisTemplate.opsForList().rightPop(srcHash); + while(bytes != null){ + sendQueue.offer(bytes); + bytes= bytesRedisTemplate.opsForList().rightPop(srcHash); + } + Session session = XiaoZhiWebsocket.clients.get(chatId); + AudioUtils.asyncSendAudioQueue(chatId, session, sendQueue); + } + this.clearAudioHandlers(chatId); + } + + private void handlerSyncRsp(String chatId, int productId) throws IOException, ExecutionException, InterruptedException { + JSONObject emotionObject = new JSONObject(); + emotionObject.put("type", "llm"); + emotionObject.put("text", "😶"); + emotionObject.put("emotion", "neutral"); + + XiaoZhiWebsocket.send(chatId, emotionObject.toJSONString()); + String voiceContent = XiaoZhiWebsocket.voiceContent.get(chatId); + CompletableFuture res = null; + if(router != null){ + res = CompletableFuture.supplyAsync( + () -> router.response(voiceContent, chatId, productId), + routerExecutor + ); + } + String answer = null; + if (res != null) { + answer = res.get(); + } + if (StringUtils.isEmpty(answer)) { + answer = "抱歉，我暂时无法理解您的问题。"; + } + if (answer.length() > 500) + answer = answer.substring(0, 500); + if (!answer.isBlank()) { + splitSentences(answer, chatId, productId); } } @Async("taskExecutor") - public void dealWithAudio2(List audioList, String chatId, int productId, boolean isManual, + public void dealWithAudio(List audioList, String chatId, int productId, boolean isManual, String... detect) { if(audioList == null || chatId == null){ log.error("audioList或chatId为空，audioList: {}, chatId: {}", audioList, chatId); @@ -409,11 +467,16 @@ public void dealWithAudio2(List audioList, String chatId, int productId, } } // 非流式的处理方法 + try { + handlerSyncRsp(chatId, productId); + }catch(Exception e){ + log.error("音频处理出错", e); + } } @Deprecated @Async("taskExecutor") - public void dealWithAudio(List audioList, String chatId, int productId, boolean isManual, + public void dealWithAudioDeprecated(List audioList, String chatId, int productId, boolean isManual, String... detect) throws IOException { Path tempFile = null; From 52e042b6682c8a64d1754bb63a601943f31c60b3 Mon Sep 17 00:00:00 2001 From: ClamJom <3188485441@qq.com> Date: Sun, 12 Apr 2026 18:15:57 +0800 Subject: [PATCH 05/12] fix: audio handler --- pom.xml | 6 + .../top/rslly/iot/config/RedisConfig.java | 85 ++- .../rslly/iot/models/AgentMemoryEntity.java | 2 +- .../rslly/iot/models/ProductRoleEntity.java | 2 +- .../iot/models/ProductRouterSetEntity.java | 2 +- .../iot/utility/smartVoice/XiaoZhiUtil.java | 500 ++++-------------- 6 files changed, 202 insertions(+), 395 deletions(-) diff --git a/pom.xml b/pom.xml index c06d7f0f..8b4bae62 100644 --- a/pom.xml +++ b/pom.xml @@ -108,6 +108,12 @@ druid-spring-boot-starter 1.1.23 + + io.lettuce + lettuce-core + 6.6.0.RELEASE + compile +