diff --git a/.gitignore b/.gitignore index f4efaeaa..9469baac 100644 --- a/.gitignore +++ b/.gitignore @@ -107,6 +107,7 @@ celerybeat.pid *.sage.py # Environments +.env .venv env/ venv/ @@ -145,24 +146,31 @@ tmp .history .DS_Store main/xiaozhi-server/data -main/xiaozhi-server/config/assets/wakeup_words.* + main/manager-web/node_modules .config.yaml .secrets.yaml .private_config.yaml +.env.development # model files main/xiaozhi-server/models/SenseVoiceSmall/model.pt main/xiaozhi-server/models/sherpa-onnx* +/main/xiaozhi-server/audio_ref/ +/audio_ref/ +/asr-models/iic/SenseVoiceSmall/ +/main/xiaozhi-server/asr-models/iic/SenseVoiceSmall/ +/models/SenseVoiceSmall/model.pt my_wakeup_words.mp3 !main/xiaozhi-server/config/assets/bind_code.wav +!main/xiaozhi-server/config/assets/wakeup_words.wav !main/xiaozhi-server/config/assets/bind_not_found.wav !main/xiaozhi-server/config/assets/bind_code/*.wav !main/xiaozhi-server/config/assets/max_output_size.wav main/manager-api/.vscode - # Ignore webpack cache directory main/manager-web/.webpack_cache/ main/xiaozhi-server/mysql uploadfile +*.json .vscode diff --git a/main/manager-api/pom.xml b/main/manager-api/pom.xml index 883295bf..bc4030c0 100644 --- a/main/manager-api/pom.xml +++ b/main/manager-api/pom.xml @@ -258,4 +258,4 @@ - \ No newline at end of file + diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java index b3c036ee..abac385b 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java +++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java @@ -227,7 +227,7 @@ public interface Constant { /** * 版本号 */ - public static final String VERSION = "0.4.4"; + public static final String VERSION = "0.5.1"; /** * 无效固件URL diff --git a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java index e8eaf462..01f1ccf2 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java +++ b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java @@ -4,7 +4,6 @@ import java.util.List; import java.util.Objects; import org.apache.shiro.authz.UnauthorizedException; -import org.springframework.context.support.DefaultMessageSourceResolvable; import org.springframework.dao.DuplicateKeyException; import org.springframework.validation.ObjectError; import org.springframework.web.bind.MethodArgumentNotValidException; diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java index bdaed950..ae58783f 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java @@ -28,4 +28,6 @@ public class AgentChatHistoryReportDTO { private String content; @Schema(description = "base64编码的opus音频数据", example = "") private String audioBase64; + @Schema(description = "上报时间，十位时间戳，空时默认使用当前时间", example = "1745657732") + private Long reportTime; } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java index e284b69f..04be9f36 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java @@ -27,4 +27,4 @@ public interface AgentChatAudioService extends IService { * @return 音频数据 */ byte[] getAudio(String audioId); -} \ No newline at end of file +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java index bf6c188b..087c1b9d 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java @@ -47,7 +47,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic public Boolean report(AgentChatHistoryReportDTO report) { String macAddress = report.getMacAddress(); Byte chatType = report.getChatType(); - log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType); + Long reportTimeMillis = null != report.getReportTime() ? report.getReportTime() * 1000 : System.currentTimeMillis(); + log.info("小智设备聊天上报请求: macAddress={}, type={} reportTime={}", macAddress, chatType, reportTimeMillis); // 根据设备MAC地址查询对应的默认智能体，判断是否需要上报 AgentEntity agentEntity = agentService.getDefaultAgentByMacAddress(macAddress); @@ -59,10 +60,10 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic String agentId = agentEntity.getId(); if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT.getCode())) { - saveChatText(report, agentId, macAddress, null); + saveChatText(report, agentId, macAddress, null, reportTimeMillis); } else if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT_AUDIO.getCode())) { String audioId = saveChatAudio(report); - saveChatText(report, agentId, macAddress, audioId); + saveChatText(report, agentId, macAddress, audioId, reportTimeMillis); } // 更新设备最后对话时间 @@ -92,8 +93,7 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic /** * 组装上报数据 */ - private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId) { - + private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId, Long reportTime) { // 构建聊天记录实体 AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder() .macAddress(macAddress) @@ -102,6 +102,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic .chatType(report.getChatType()) .content(report.getContent()) .audioId(audioId) + .createdAt(new Date(reportTime)) + // NOTE(haotian): 2025/5/26 updateAt可以不设置，重点是createAt，而且这样可以看到上报延迟 .build(); // 保存数据 diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java index cf804678..efcae485 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java @@ -31,4 +31,4 @@ public class AgentChatAudioServiceImpl extends ServiceImpl map = (Map) model.getConfigJson(); if ("intent_llm".equals(map.get("type"))) { intentLLMModelId = (String) map.get("llm"); - if (intentLLMModelId != null && intentLLMModelId.equals(llmModelId)) { + if (StringUtils.isNotBlank(intentLLMModelId) && intentLLMModelId.equals(llmModelId)) { intentLLMModelId = null; } } @@ -281,10 +282,31 @@ public class ConfigServiceImpl implements ConfigService { } } } + if ("Memory".equals(modelTypes[i])) { + Map map = (Map) model.getConfigJson(); + if ("mem_local_short".equals(map.get("type"))) { + memLocalShortLLMModelId = (String) map.get("llm"); + if (StringUtils.isNotBlank(memLocalShortLLMModelId) + && memLocalShortLLMModelId.equals(llmModelId)) { + memLocalShortLLMModelId = null; + } + } + } // 如果是LLM类型，且intentLLMModelId不为空，则添加附加模型 - if ("LLM".equals(modelTypes[i]) && intentLLMModelId != null) { - ModelConfigEntity intentLLM = modelConfigService.getModelById(intentLLMModelId, isCache); - typeConfig.put(intentLLM.getId(), intentLLM.getConfigJson()); + if ("LLM".equals(modelTypes[i])) { + if (StringUtils.isNotBlank(intentLLMModelId)) { + if (!typeConfig.containsKey(intentLLMModelId)) { + ModelConfigEntity intentLLM = modelConfigService.getModelById(intentLLMModelId, isCache); + typeConfig.put(intentLLM.getId(), intentLLM.getConfigJson()); + } + } + if (StringUtils.isNotBlank(memLocalShortLLMModelId)) { + if (!typeConfig.containsKey(memLocalShortLLMModelId)) { + ModelConfigEntity memLocalShortLLM = modelConfigService + .getModelById(memLocalShortLLMModelId, isCache); + typeConfig.put(memLocalShortLLM.getId(), memLocalShortLLM.getConfigJson()); + } + } } } result.put(modelTypes[i], typeConfig); diff --git a/main/manager-api/src/main/resources/application-dev.yml b/main/manager-api/src/main/resources/application-dev.yml index b49b31ca..1c86f288 100644 --- a/main/manager-api/src/main/resources/application-dev.yml +++ b/main/manager-api/src/main/resources/application-dev.yml @@ -48,4 +48,4 @@ spring: max-active: 8 # 连接池最大连接数（使用负值表示没有限制） max-idle: 8 # 连接池中的最大空闲连接 min-idle: 0 # 连接池中的最小空闲连接 - shutdown-timeout: 100ms # 客户端优雅关闭的等待时间 \ No newline at end of file + shutdown-timeout: 100ms # 客户端优雅关闭的等待时间 diff --git a/main/manager-api/src/main/resources/db/changelog/202505271414.sql b/main/manager-api/src/main/resources/db/changelog/202505271414.sql new file mode 100644 index 00000000..588cd13c --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202505271414.sql @@ -0,0 +1,51 @@ +-- 本地短期记忆配置可以设置独立的LLM +update `ai_model_provider` set fields = '[{"key":"llm","label":"LLM模型","type":"string"}]' where id = 'SYSTEM_Memory_mem_local_short'; +update `ai_model_config` set config_json = '{\"type\": \"mem_local_short\", \"llm\": \"LLM_ChatGLMLLM\"}' where id = 'Memory_mem_local_short'; + +-- 增加火山双流式TTS供应器和模型配置 +delete from `ai_model_provider` where id = 'SYSTEM_TTS_HSDSTTS'; +INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES +('SYSTEM_TTS_HSDSTTS', 'TTS', 'huoshan_double_stream', '火山双流式语音合成', '[{"key":"ws_url","label":"WebSocket地址","type":"string"},{"key":"appid","label":"应用ID","type":"string"},{"key":"access_token","label":"访问令牌","type":"string"},{"key":"resource_id","label":"资源ID","type":"string"},{"key":"speaker","label":"默认音色","type":"string"}]', 13, 1, NOW(), 1, NOW()); + +delete from `ai_model_config` where id = 'TTS_HuoshanDoubleStreamTTS'; +INSERT INTO `ai_model_config` VALUES ('TTS_HuoshanDoubleStreamTTS', 'TTS', 'HuoshanDoubleStreamTTS', '火山双流式语音合成', 0, 1, '{\"type\": \"huoshan_double_stream\", \"ws_url\": \"wss://openspeech.bytedance.com/api/v3/tts/bidirection\", \"appid\": \"你的火山引擎语音合成服务appid\", \"access_token\": \"你的火山引擎语音合成服务access_token\", \"resource_id\": \"volc.service_type.10029\", \"speaker\": \"zh_female_wanwanxiaohe_moon_bigtts\"}', NULL, NULL, 16, NULL, NULL, NULL, NULL); + +-- 火山双流式TT模型配置说明文档 +UPDATE `ai_model_config` SET +`doc_link` = 'https://console.volcengine.com/speech/service/10007', +`remark` = '火山引擎语音合成服务配置说明： +1. 访问 https://www.volcengine.com/ 注册并开通火山引擎账号 +2. 访问 https://console.volcengine.com/speech/service/10007 开通语音合成大模型，购买音色 +3. 在页面底部获取appid和access_token +5. 资源ID固定为：volc.service_type.10029（大模型语音合成及混音） +6. 填入配置文件中' WHERE `id` = 'TTS_HuoshanDoubleStreamTTS'; + + +-- 添加火山双流式TTS音色 +delete from `ai_tts_voice` where tts_model_id = 'TTS_HuoshanDoubleStreamTTS'; +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0001', 'TTS_HuoshanDoubleStreamTTS', '爽快思思/Skye', 'zh_female_shuangkuaisisi_moon_bigtts', '中文、英文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Skye.mp3', NULL, 1, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0002', 'TTS_HuoshanDoubleStreamTTS', '温暖阿虎/Alvin', 'zh_male_wennuanahu_moon_bigtts', '中文、英文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Alvin.mp3', NULL, 2, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0003', 'TTS_HuoshanDoubleStreamTTS', '少年梓辛/Brayan', 'zh_male_shaonianzixin_moon_bigtts', '中文、英文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Brayan.mp3', NULL, 3, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0004', 'TTS_HuoshanDoubleStreamTTS', '邻家女孩', 'zh_female_linjianvhai_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E9%82%BB%E5%AE%B6%E5%A5%B3%E5%AD%A9.mp3', NULL, 4, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0005', 'TTS_HuoshanDoubleStreamTTS', '渊博小叔', 'zh_male_yuanboxiaoshu_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%B8%8A%E5%8D%9A%E5%B0%8F%E5%8F%94.mp3', NULL, 5, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0006', 'TTS_HuoshanDoubleStreamTTS', '阳光青年', 'zh_male_yangguangqingnian_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E9%98%B3%E5%85%89%E9%9D%92%E5%B9%B4.mp3', NULL, 6, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0007', 'TTS_HuoshanDoubleStreamTTS', '京腔侃爷/Harmony', 'zh_male_jingqiangkanye_moon_bigtts', '中文、英文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Harmony.mp3', NULL, 7, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0008', 'TTS_HuoshanDoubleStreamTTS', '湾湾小何', 'zh_female_wanwanxiaohe_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%B9%BE%E6%B9%BE%E5%B0%8F%E4%BD%95.mp3', NULL, 8, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0009', 'TTS_HuoshanDoubleStreamTTS', '湾区大叔', 'zh_female_wanqudashu_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%B9%BE%E5%8C%BA%E5%A4%A7%E5%8F%94.mp3', NULL, 9, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0010', 'TTS_HuoshanDoubleStreamTTS', '呆萌川妹', 'zh_female_daimengchuanmei_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%91%86%E8%90%8C%E5%B7%9D%E5%A6%B9.mp3', NULL, 10, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0011', 'TTS_HuoshanDoubleStreamTTS', '广州德哥', 'zh_male_guozhoudege_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%B9%BF%E5%B7%9E%E5%BE%B7%E5%93%A5.mp3', NULL, 11, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0012', 'TTS_HuoshanDoubleStreamTTS', '北京小爷', 'zh_male_beijingxiaoye_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%8C%97%E4%BA%AC%E5%B0%8F%E7%88%B7.mp3', NULL, 12, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0013', 'TTS_HuoshanDoubleStreamTTS', '浩宇小哥', 'zh_male_haoyuxiaoge_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%B5%A9%E5%AE%87%E5%B0%8F%E5%93%A5.mp3', NULL, 13, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0014', 'TTS_HuoshanDoubleStreamTTS', '广西远舟', 'zh_male_guangxiyuanzhou_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%B9%BF%E8%A5%BF%E8%BF%9C%E8%88%9F.mp3', NULL, 14, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0015', 'TTS_HuoshanDoubleStreamTTS', '妹坨洁儿', 'zh_female_meituojieer_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%A6%B9%E5%9D%A8%E6%B4%81%E5%84%BF.mp3', NULL, 15, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0016', 'TTS_HuoshanDoubleStreamTTS', '豫州子轩', 'zh_male_yuzhouzixuan_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E8%B1%AB%E5%B7%9E%E5%AD%90%E8%BD%A9.mp3', NULL, 16, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0017', 'TTS_HuoshanDoubleStreamTTS', '高冷御姐', 'zh_female_gaolengyujie_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E9%AB%98%E5%86%B7%E5%BE%A1%E5%A7%90.mp3', NULL, 17, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0018', 'TTS_HuoshanDoubleStreamTTS', '傲娇霸总', 'zh_male_aojiaobazong_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E5%82%B2%E5%A8%87%E9%9C%B8%E6%80%BB.mp3', NULL, 18, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0019', 'TTS_HuoshanDoubleStreamTTS', '魅力女友', 'zh_female_meilinvyou_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E9%AD%85%E5%8A%9B%E5%A5%B3%E5%8F%8B.mp3', NULL, 19, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0020', 'TTS_HuoshanDoubleStreamTTS', '深夜播客', 'zh_male_shenyeboke_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%B7%B1%E5%A4%9C%E6%92%AD%E5%AE%A2.mp3', NULL, 20, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0021', 'TTS_HuoshanDoubleStreamTTS', '柔美女友', 'zh_female_sajiaonvyou_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%9F%94%E7%BE%8E%E5%A5%B3%E5%8F%8B.mp3', NULL, 21, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0022', 'TTS_HuoshanDoubleStreamTTS', '撒娇学妹', 'zh_female_yuanqinvyou_moon_bigtts', '中文', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%92%92%E5%A8%87%E5%AD%A6%E5%A6%B9.mp3', NULL, 22, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0023', 'TTS_HuoshanDoubleStreamTTS', 'かずね（和音）', 'multi_male_jingqiangkanye_moon_bigtts', '日语、西语', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Javier.wav', NULL, 23, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0024', 'TTS_HuoshanDoubleStreamTTS', 'はるこ（晴子）', 'multi_female_shuangkuaisisi_moon_bigtts', '日语、西语', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Esmeralda.mp3', NULL, 24, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0025', 'TTS_HuoshanDoubleStreamTTS', 'あけみ（朱美）', 'multi_female_gaolengyujie_moon_bigtts', '日语', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/%E6%9C%B1%E7%BE%8E.mp3', NULL, 25, NULL, NULL, NULL, NULL); +INSERT INTO `ai_tts_voice` VALUES ('TTS_HuoshanDoubleStreamTTS_0026', 'TTS_HuoshanDoubleStreamTTS', 'ひろし（広志）', 'multi_male_wanqudashu_moon_bigtts', '日语、西语', 'https://lf3-static.bytednsdoc.com/obj/eden-cn/lm_hz_ihsph/ljhwZthlaukjlkulzlp/portal/bigtts/Roberto.wav', NULL, 26, NULL, NULL, NULL, NULL); diff --git a/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml b/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml index 9d3808da..55b7cd4c 100755 --- a/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml +++ b/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml @@ -162,4 +162,11 @@ databaseChangeLog: changes: - sqlFile: encoding: utf8 - path: classpath:db/changelog/202505151451.sql \ No newline at end of file + path: classpath:db/changelog/202505151451.sql + - changeSet: + id: 202505271414 + author: hrz + changes: + - sqlFile: + encoding: utf8 + path: classpath:db/changelog/202505271414.sql \ No newline at end of file diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 44845df8..25c8e017 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -85,6 +85,7 @@ module_test: # 唤醒词，用于识别唤醒词还是讲话内容 wakeup_words: - "你好小智" + - "嘿你好呀" - "你好小志" - "小爱同学" - "你好小鑫" @@ -218,8 +219,12 @@ Memory: # 不想使用记忆功能，可以使用nomem type: nomem mem_local_short: - # 本地记忆功能，通过selected_module的llm总结，数据保存在本地，不会上传到服务器 + # 本地记忆功能，通过selected_module的llm总结，数据保存在本地服务器，不会上传到外部服务器 type: mem_local_short + # 配备记忆存储独立的思考模型 + # 如果这里不填，则会默认使用selected_module.LLM的模型作为意图识别的思考模型 + # 如果你的不想使用selected_module.LLM记忆存储，这里最好使用独立的LLM作为意图识别，例如使用免费的ChatGLMLLM + llm: ChatGLMLLM ASR: FunASR: @@ -325,7 +330,7 @@ LLM: # 定义LLM API类型 type: openai # 先开通服务，打开以下网址，开通的服务搜索Doubao-1.5-pro，开通它 - # 开通改地址：https://console.volcengine.com/ark/region:ark+cn-beijing/openManagement?LLM=%7B%7D&OpenTokenDrawer=false + # 开通地址：https://console.volcengine.com/ark/region:ark+cn-beijing/openManagement?LLM=%7B%7D&OpenTokenDrawer=false # 免费额度500000token # 开通后，进入这里获取密钥：https://console.volcengine.com/ark/region:ark+cn-beijing/apiKey?apikey=%7B%7D base_url: https://ark.cn-beijing.volces.com/api/v3 @@ -452,6 +457,19 @@ TTS: speed_ratio: 1.0 volume_ratio: 1.0 pitch_ratio: 1.0 + #火山tts，支持双向流式tts + HuoshanDoubleStreamTTS: + type: huoshan_double_stream + # 访问 https://console.volcengine.com/speech/service/10007 开通语音合成大模型，购买音色 + # 在页面底部获取appid和access_token + # 资源ID固定为：volc.service_type.10029（大模型语音合成及混音） + # 如果是机智云，把接口地址换成wss://bytedance.gizwitsapi.com/api/v3/tts/bidirection + # 机智云不需要天填 appid + ws_url: wss://openspeech.bytedance.com/api/v3/tts/bidirection + appid: 你的火山引擎语音合成服务appid + access_token: 你的火山引擎语音合成服务access_token + resource_id: volc.service_type.10029 + speaker: zh_female_wanwanxiaohe_moon_bigtts CosyVoiceSiliconflow: type: siliconflow # 硅基流动TTS diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py index 44b3aab4..e5fec630 100644 --- a/main/xiaozhi-server/config/logger.py +++ b/main/xiaozhi-server/config/logger.py @@ -4,7 +4,7 @@ from loguru import logger from config.config_loader import load_config from config.settings import check_config_file -SERVER_VERSION = "0.4.4" +SERVER_VERSION = "0.5.1" _logger_initialized = False diff --git a/main/xiaozhi-server/config/manage_api_client.py b/main/xiaozhi-server/config/manage_api_client.py index 11e0ecbe..cd3f86c4 100644 --- a/main/xiaozhi-server/config/manage_api_client.py +++ b/main/xiaozhi-server/config/manage_api_client.py @@ -160,7 +160,7 @@ def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]: def report( - mac_address: str, session_id: str, chat_type: int, content: str, audio + mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time ) -> Optional[Dict]: """带熔断的业务方法示例""" if not content or not ManageApiClient._instance: @@ -174,6 +174,7 @@ def report( "sessionId": session_id, "chatType": chat_type, "content": content, + "reportTime": report_time, "audioBase64": ( base64.b64encode(audio).decode("utf-8") if audio else None ), diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 3666e264..11ae6836 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -1,42 +1,40 @@ import os +import sys import copy import json -import subprocess -import sys import uuid import time import queue import asyncio -import traceback - import threading +import traceback +import subprocess import websockets -from typing import Dict, Any -from plugins_func.loadplugins import auto_import_modules -from config.logger import setup_logging, build_module_string, update_module_string -from config.config_loader import get_project_dir -from core.utils import p3 -from core.utils.dialogue import Message, Dialogue -from core.handle.textHandle import handleTextMessage from core.utils.util import ( - get_string_no_punctuation_or_emoji, extract_json_from_string, initialize_modules, check_vad_update, check_asr_update, filter_sensitive_info, + initialize_tts, ) -from concurrent.futures import ThreadPoolExecutor, TimeoutError -from core.handle.sendAudioHandle import sendAudioMessage -from core.handle.receiveAudioHandle import handleAudioMessage +from typing import Dict, Any +from core.mcp.manager import MCPManager +from core.handle.reportHandle import report +from core.providers.tts.default import DefaultTTS +from concurrent.futures import ThreadPoolExecutor +from core.utils.dialogue import Message, Dialogue +from core.handle.textHandle import handleTextMessage from core.handle.functionHandler import FunctionHandler +from plugins_func.loadplugins import auto_import_modules from plugins_func.register import Action, ActionResponse from core.auth import AuthMiddleware, AuthenticationError -from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api +from core.handle.receiveAudioHandle import handleAudioMessage +from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType +from config.logger import setup_logging, build_module_string, update_module_string from config.manage_api_client import DeviceNotFoundException, DeviceBindException -from core.utils.output_counter import add_device_output -from core.handle.reportHandle import enqueue_tts_report, report + TAG = __name__ @@ -54,7 +52,6 @@ class ConnectionHandler: _vad, _asr, _llm, - _tts, _memory, _intent, server=None, @@ -87,24 +84,22 @@ class ConnectionHandler: # 线程任务相关 self.loop = asyncio.get_event_loop() self.stop_event = threading.Event() - self.tts_queue = queue.Queue() - self.audio_play_queue = queue.Queue() - self.executor = ThreadPoolExecutor(max_workers=10) + self.executor = ThreadPoolExecutor(max_workers=5) - # 上报线程 + # 添加上报线程池 self.report_queue = queue.Queue() self.report_thread = None - # TODO(haotian): 2025/5/12 可以通过修改此处，调节asr的上报和tts的上报 + # 未来可以通过修改此处，调节asr的上报和tts的上报，目前默认都开启 self.report_asr_enable = self.read_config_from_api self.report_tts_enable = self.read_config_from_api # 依赖的组件 self.vad = None self.asr = None + self.tts = None self._asr = _asr self._vad = _vad self.llm = _llm - self.tts = _tts self.memory = _memory self.intent = _intent @@ -120,12 +115,11 @@ class ConnectionHandler: self.asr_server_receive = True # llm相关变量 - self.llm_finish_task = False + self.llm_finish_task = True self.dialogue = Dialogue() # tts相关变量 - self.tts_first_text_index = -1 - self.tts_last_text_index = -1 + self.sentence_id = None # iot相关变量 self.iot_descriptors = {} @@ -196,17 +190,6 @@ class ConnectionHandler: self._initialize_private_config() # 异步初始化 self.executor.submit(self._initialize_components) - # tts 消化线程 - self.tts_priority_thread = threading.Thread( - target=self._tts_priority_thread, daemon=True - ) - self.tts_priority_thread.start() - - # 音频播放 消化线程 - self.audio_play_priority_thread = threading.Thread( - target=self._audio_play_priority_thread, daemon=True - ) - self.audio_play_priority_thread.start() try: async for message in self.websocket: @@ -315,29 +298,39 @@ class ConnectionHandler: ) def _initialize_components(self): - self.selected_module_str = build_module_string( - self.config.get("selected_module", {}) - ) - update_module_string(self.selected_module_str) - """初始化组件""" - if self.config.get("prompt") is not None: - self.prompt = self.config["prompt"] - self.change_system_prompt(self.prompt) - self.logger.bind(tag=TAG).info( - f"初始化组件: prompt成功 {self.prompt[:50]}..." + try: + self.selected_module_str = build_module_string( + self.config.get("selected_module", {}) + ) + update_module_string(self.selected_module_str) + """初始化组件""" + if self.config.get("prompt") is not None: + self.prompt = self.config["prompt"] + self.change_system_prompt(self.prompt) + self.logger.bind(tag=TAG).info( + f"初始化组件: prompt成功 {self.prompt[:50]}..." + ) + + """初始化本地组件""" + if self.vad is None: + self.vad = self._vad + if self.asr is None: + self.asr = self._asr + if self.tts is None: + self.tts = self._initialize_tts() + # 使用事件循环运行异步方法 + asyncio.run_coroutine_threadsafe( + self.tts.open_audio_channels(self), self.loop ) - """初始化本地组件""" - if self.vad is None: - self.vad = self._vad - if self.asr is None: - self.asr = self._asr - """加载记忆""" - self._initialize_memory() - """加载意图识别""" - self._initialize_intent() - """初始化上报线程""" - self._init_report_threads() + """加载记忆""" + self._initialize_memory() + """加载意图识别""" + self._initialize_intent() + """初始化上报线程""" + self._init_report_threads() + except Exception as e: + self.logger.bind(tag=TAG).error(f"实例化组件失败: {e}") def _init_report_threads(self): """初始化ASR和TTS上报线程""" @@ -352,6 +345,17 @@ class ConnectionHandler: self.report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") + def _initialize_tts(self): + """初始化TTS""" + tts = None + if not self.need_bind: + tts = initialize_tts(self.config) + + if tts is None: + tts = DefaultTTS(self.config, delete_audio_file=True) + + return tts + def _initialize_private_config(self): """如果是从配置文件获取，则进行二次实例化""" if not self.read_config_from_api: @@ -458,6 +462,37 @@ class ConnectionHandler: save_to_file=not self.read_config_from_api, ) + # 获取记忆总结配置 + memory_config = self.config["Memory"] + memory_type = self.config["Memory"][self.config["selected_module"]["Memory"]][ + "type" + ] + # 如果使用 nomen，直接返回 + if memory_type == "nomem": + return + # 使用 mem_local_short 模式 + elif memory_type == "mem_local_short": + memory_llm_name = memory_config[self.config["selected_module"]["Memory"]][ + "llm" + ] + if memory_llm_name and memory_llm_name in self.config["LLM"]: + # 如果配置了专用LLM，则创建独立的LLM实例 + from core.utils import llm as llm_utils + + memory_llm_config = self.config["LLM"][memory_llm_name] + memory_llm_type = memory_llm_config.get("type", memory_llm_name) + memory_llm = llm_utils.create_instance( + memory_llm_type, memory_llm_config + ) + self.logger.bind(tag=TAG).info( + f"为记忆总结创建了专用LLM: {memory_llm_name}, 类型: {memory_llm_type}" + ) + self.memory.set_llm(memory_llm) + else: + # 否则使用主LLM + self.memory.set_llm(self.llm) + self.logger.bind(tag=TAG).info("使用主LLM作为意图识别模型") + def _initialize_intent(self): self.intent_type = self.config["Intent"][ self.config["selected_module"]["Intent"] @@ -513,7 +548,8 @@ class ConnectionHandler: self.dialogue.update_system_message(self.prompt) def chat(self, query, tool_call=False): - self.logger.bind(tag=TAG).debug(f"Chat: {query}") + 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)) @@ -523,7 +559,6 @@ class ConnectionHandler: if self.intent_type == "function_call" and hasattr(self, "func_handler"): functions = self.func_handler.get_functions() response_message = [] - processed_chars = 0 # 跟踪已处理的字符位置 try: # 使用带记忆的对话 @@ -534,6 +569,9 @@ class ConnectionHandler: ) memory_str = future.result() + uuid_str = str(uuid.uuid4()).replace("-", "") + self.sentence_id = uuid_str + if functions is not None: # 使用支持functions的streaming接口 llm_responses = self.llm.response_with_functions( @@ -550,15 +588,13 @@ class ConnectionHandler: self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}") return None - self.llm_finish_task = False - text_index = 0 - # 处理流式响应 tool_call_flag = False function_name = None function_id = None function_arguments = "" content_arguments = "" + text_index = 0 for response in llm_responses: if self.intent_type == "function_call": @@ -586,47 +622,26 @@ class ConnectionHandler: if content is not None and len(content) > 0: if not tool_call_flag: response_message.append(content) - if self.client_abort: break - end_time = time.time() - # self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}") - - # 处理文本分段和TTS逻辑 - # 合并当前全部文本并处理未分割部分 - full_text = "".join(response_message) - current_text = full_text[processed_chars:] # 从未处理的位置开始 - - # 查找最后一个有效标点 - punctuations = ("。", ".", "？", "?", "！", "!", "；", ";", "：") - last_punct_pos = -1 - number_flag = True - for punct in punctuations: - pos = current_text.rfind(punct) - prev_char = current_text[pos - 1] if pos - 1 >= 0 else "" - # 如果.前面是数字统一判断为小数 - if prev_char.isdigit() and punct == ".": - number_flag = False - if pos > last_punct_pos and number_flag: - last_punct_pos = pos - - # 找到分割点则处理 - if last_punct_pos != -1: - segment_text_raw = current_text[: last_punct_pos + 1] - segment_text = get_string_no_punctuation_or_emoji( - segment_text_raw - ) - if segment_text: - text_index += 1 - self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit( - self.speak_and_play, None, segment_text, text_index + 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_queue.put((future, text_index)) - # 更新已处理字符位置 - processed_chars += len(segment_text_raw) - + ) + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=content, + ) + ) + text_index += 1 # 处理function call if tool_call_flag: bHasError = False @@ -669,27 +684,21 @@ class ConnectionHandler: result = self.func_handler.handle_llm_function_call( self, function_call_data ) - self._handle_function_result(result, function_call_data, text_index + 1) - - # 处理最后剩余的文本 - full_text = "".join(response_message) - remaining_text = full_text[processed_chars:] - if remaining_text: - segment_text = get_string_no_punctuation_or_emoji(remaining_text) - if segment_text: - text_index += 1 - self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit( - self.speak_and_play, None, segment_text, text_index - ) - self.tts_queue.put((future, text_index)) + self._handle_function_result(result, function_call_data) # 存储对话内容 if len(response_message) > 0: self.dialogue.put( Message(role="assistant", content="".join(response_message)) ) - + if text_index > 0: + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=self.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) + ) self.llm_finish_task = True self.logger.bind(tag=TAG).debug( json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False) @@ -739,12 +748,10 @@ class ConnectionHandler: return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="") - def _handle_function_result(self, result, function_call_data, text_index): + def _handle_function_result(self, result, function_call_data): if result.action == Action.RESPONSE: # 直接回复前端 text = result.response - self.recode_first_last_text(text, text_index) - future = self.executor.submit(self.speak_and_play, None, text, text_index) - self.tts_queue.put((future, text_index)) + self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) self.dialogue.put(Message(role="assistant", content=text)) elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复 text = result.result @@ -781,109 +788,11 @@ class ConnectionHandler: self.chat(text, tool_call=True) elif result.action == Action.NOTFOUND or result.action == Action.ERROR: text = result.result - self.recode_first_last_text(text, text_index) - future = self.executor.submit(self.speak_and_play, None, text, text_index) - self.tts_queue.put((future, text_index)) + self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) self.dialogue.put(Message(role="assistant", content=text)) else: pass - def _tts_priority_thread(self): - while not self.stop_event.is_set(): - text = None - try: - try: - item = self.tts_queue.get(timeout=1) - if item is None: - continue - future, text_index = item # 解包获取 Future 和 text_index - except queue.Empty: - if self.stop_event.is_set(): - break - continue - if future is None: - continue - text = None - audio_datas, tts_file = [], None - try: - self.logger.bind(tag=TAG).debug("正在处理TTS任务...") - tts_timeout = int(self.config.get("tts_timeout", 10)) - tts_file, text, _ = future.result(timeout=tts_timeout) - if tts_file is None: - self.logger.bind(tag=TAG).error( - f"TTS出错： file is empty: {text_index}: {text}" - ) - else: - self.logger.bind(tag=TAG).debug( - f"TTS生成：文件路径: {tts_file}" - ) - if os.path.exists(tts_file): - if tts_file.endswith(".p3"): - audio_datas, _ = p3.decode_opus_from_file(tts_file) - elif self.audio_format == "pcm": - audio_datas, _ = self.tts.audio_to_pcm_data(tts_file) - else: - audio_datas, _ = self.tts.audio_to_opus_data(tts_file) - # 在这里上报TTS数据 - enqueue_tts_report( - self, tts_file if text is None else text, audio_datas - ) - else: - self.logger.bind(tag=TAG).error( - f"TTS出错：文件不存在{tts_file}" - ) - except TimeoutError: - self.logger.bind(tag=TAG).error("TTS超时") - except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS出错: {e}") - if not self.client_abort: - # 如果没有中途打断就发送语音 - self.audio_play_queue.put((audio_datas, text, text_index)) - if ( - self.tts.delete_audio_file - and tts_file is not None - and os.path.exists(tts_file) - and tts_file.startswith(self.tts.output_file) - ): - os.remove(tts_file) - except Exception as e: - self.logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}") - self.clearSpeakStatus() - asyncio.run_coroutine_threadsafe( - self.websocket.send( - json.dumps( - { - "type": "tts", - "state": "stop", - "session_id": self.session_id, - } - ) - ), - self.loop, - ) - self.logger.bind(tag=TAG).error( - f"tts_priority priority_thread: {text} {e}" - ) - - def _audio_play_priority_thread(self): - while not self.stop_event.is_set(): - text = None - try: - try: - audio_datas, text, text_index = self.audio_play_queue.get(timeout=1) - except queue.Empty: - if self.stop_event.is_set(): - break - continue - future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self, audio_datas, text, text_index), self.loop - ) - future.result() - except Exception as e: - self.logger.bind(tag=TAG).error( - f"audio_play_priority priority_thread: {text} {e}" - ) - def _report_worker(self): """聊天记录上报工作线程""" while not self.stop_event.is_set(): @@ -893,16 +802,15 @@ class ConnectionHandler: if item is None: # 检测毒丸对象 break - type, text, audio_data = item + type, text, audio_data, report_time = item try: - # 执行上报（传入二进制数据） - report(self, type, text, audio_data) + # 提交任务到线程池 + self.executor.submit( + self._process_report, type, text, audio_data, report_time + ) except Exception as e: self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}") - finally: - # 标记任务完成 - self.report_queue.task_done() except queue.Empty: continue except Exception as e: @@ -910,33 +818,20 @@ class ConnectionHandler: self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") - def speak_and_play(self, file_path, content, text_index=0): - if file_path is not None: - self.logger.bind(tag=TAG).info(f"无需tts转换: 从文件播放，{file_path}") - return file_path, content, text_index - if content is None or len(content) <= 0: - self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{content}") - return None, content, text_index - tts_file = self.tts.to_tts(content) - if tts_file is None: - self.logger.bind(tag=TAG).error(f"tts转换失败，{content}") - return None, content, text_index - self.logger.bind(tag=TAG).debug(f"TTS 文件生成完毕: {tts_file}") - if self.max_output_size > 0: - add_device_output(self.headers.get("device-id"), len(content)) - return tts_file, content, text_index + def _process_report(self, type, text, audio_data, report_time): + """处理上报任务""" + try: + # 执行上报（传入二进制数据） + report(self, type, text, audio_data, report_time) + except Exception as e: + self.logger.bind(tag=TAG).error(f"上报处理异常: {e}") + finally: + # 标记任务完成 + self.report_queue.task_done() def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") self.asr_server_receive = True - self.tts_last_text_index = -1 - self.tts_first_text_index = -1 - - def recode_first_last_text(self, text, text_index=0): - if self.tts_first_text_index == -1: - self.logger.bind(tag=TAG).info(f"大模型说出第一句话: {text}") - self.tts_first_text_index = text_index - self.tts_last_text_index = text_index async def close(self, ws=None): """资源清理方法""" @@ -972,23 +867,24 @@ class ConnectionHandler: def clear_queues(self): """清空所有任务队列""" - self.logger.bind(tag=TAG).debug( - f"开始清理: TTS队列大小={self.tts_queue.qsize()}, 音频队列大小={self.audio_play_queue.qsize()}" - ) + if self.tts: + self.logger.bind(tag=TAG).debug( + f"开始清理: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}" + ) - # 使用非阻塞方式清空队列 - for q in [self.tts_queue, self.audio_play_queue]: - if not q: - continue - while True: - try: - q.get_nowait() - except queue.Empty: - break + # 使用非阻塞方式清空队列 + for q in [self.tts.tts_text_queue, self.tts.tts_audio_queue]: + if not q: + continue + while True: + try: + q.get_nowait() + except queue.Empty: + break - self.logger.bind(tag=TAG).debug( - f"清理结束: TTS队列大小={self.tts_queue.qsize()}, 音频队列大小={self.audio_play_queue.qsize()}" - ) + self.logger.bind(tag=TAG).debug( + f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}" + ) def reset_vad_states(self): self.client_audio_buffer = bytearray() diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index e29ef450..6fd40401 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,11 +1,13 @@ +import os +import time import json -from core.handle.sendAudioHandle import send_stt_message -from core.utils.util import remove_punctuation_and_length +import random import shutil import asyncio -import os -import random -import time +from core.handle.sendAudioHandle import send_stt_message +from core.utils.util import remove_punctuation_and_length +from core.providers.tts.dto.dto import ContentType, InterfaceType + TAG = __name__ @@ -26,7 +28,8 @@ async def handleHelloMessage(conn, msg_json): format = audio_params.get("format") conn.logger.bind(tag=TAG).info(f"客户端音频格式: {format}") conn.audio_format = format - conn.asr.set_audio_format(format) + if conn.asr is not None: + conn.asr.set_audio_format(format) conn.welcome_msg["audio_params"] = audio_params await conn.websocket.send(json.dumps(conn.welcome_msg)) @@ -36,6 +39,10 @@ async def checkWakeupWords(conn, text): enable_wakeup_words_response_cache = conn.config[ "enable_wakeup_words_response_cache" ] + """是否用的是非流式tts""" + if conn.tts and conn.tts.interface_type != InterfaceType.NON_STREAM: + return False + """是否开启唤醒词加速""" if not enable_wakeup_words_response_cache: return False @@ -43,19 +50,17 @@ async def checkWakeupWords(conn, text): _, filtered_text = remove_punctuation_and_length(text) if filtered_text in conn.config.get("wakeup_words"): await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - conn.llm_finish_task = True file = getWakeupWordFile(WAKEUP_CONFIG["file_name"]) if file is None: asyncio.create_task(wakeupWordsResponse(conn)) return False - opus_packets, _ = conn.tts.audio_to_opus_data(file) text_hello = WAKEUP_CONFIG["text"] if not text_hello: text_hello = text - conn.audio_play_queue.put((opus_packets, text_hello, 0)) + conn.tts.tts_one_sentence( + conn, ContentType.FILE, content_file=file, content_detail=text_hello + ) if time.time() - WAKEUP_CONFIG["create_time"] > WAKEUP_CONFIG["refresh_time"]: asyncio.create_task(wakeupWordsResponse(conn)) return True diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index e61b1265..4c876550 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -4,6 +4,7 @@ import uuid from core.handle.sendAudioHandle import send_stt_message 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 plugins_func.register import Action from loguru import logger @@ -143,11 +144,6 @@ async def process_intent_result(conn, intent_result, original_text): def speak_txt(conn, text): - text_index = ( - conn.tts_last_text_index + 1 if hasattr(conn, "tts_last_text_index") else 0 - ) - conn.recode_first_last_text(text, text_index) - future = conn.executor.submit(conn.speak_and_play, None, text, text_index) - conn.llm_finish_task = True - conn.tts_queue.put((future, text_index)) + conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text) conn.dialogue.put(Message(role="assistant", content=text)) + diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 2584ab83..18482a04 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -5,6 +5,7 @@ from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit from core.handle.reportHandle import enqueue_asr_report +from core.handle.sendAudioHandle import SentenceType from core.utils.util import audio_to_data TAG = __name__ @@ -46,9 +47,8 @@ async def handleAudioMessage(conn, audio): text_len, _ = remove_punctuation_and_length(raw_text) if text_len > 0: # 使用自定义模块进行上报 - enqueue_asr_report(conn, raw_text, copy.deepcopy(conn.asr_audio)) - await startToChat(conn, raw_text) + enqueue_asr_report(conn, raw_text, copy.deepcopy(conn.asr_audio)) else: conn.asr_server_receive = True conn.asr_audio.clear() @@ -110,12 +110,9 @@ async def no_voice_close_connect(conn): async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - conn.llm_finish_task = True file_path = "config/assets/max_output_size.wav" opus_packets, _ = audio_to_data(file_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) conn.close_after_chat = True @@ -130,14 +127,11 @@ async def check_bind_device(conn): text = f"请登录控制面板，输入{conn.bind_code}，绑定设备。" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 6 - conn.llm_finish_task = True # 播放提示音 music_path = "config/assets/bind_code.wav" opus_packets, _ = audio_to_data(music_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 @@ -145,16 +139,14 @@ async def check_bind_device(conn): digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" num_packets, _ = audio_to_data(num_path) - conn.audio_play_queue.put((num_packets, None, i + 1)) + conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None)) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) else: text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) - conn.tts_first_text_index = 0 - conn.tts_last_text_index = 0 - conn.llm_finish_task = True music_path = "config/assets/bind_not_found.wav" opus_packets, _ = audio_to_data(music_path) - conn.audio_play_queue.put((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) diff --git a/main/xiaozhi-server/core/handle/reportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py index bb1ea066..b1214fd5 100644 --- a/main/xiaozhi-server/core/handle/reportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -8,6 +8,7 @@ TTS上报功能已集成到ConnectionHandler类中。 具体实现请参考core/connection.py中的相关代码。 """ +import time import opuslib_next @@ -16,7 +17,7 @@ from config.manage_api_client import report as manage_report TAG = __name__ -def report(conn, type, text, opus_data): +def report(conn, type, text, opus_data, report_time): """执行聊天记录上报操作 Args: @@ -24,6 +25,7 @@ def report(conn, type, text, opus_data): type: 上报类型，1为用户，2为智能体 text: 合成文本 opus_data: opus音频数据 + report_time: 上报时间 """ try: if opus_data: @@ -37,6 +39,7 @@ def report(conn, type, text, opus_data): chat_type=type, content=text, audio=audio_data, + report_time=report_time, ) except Exception as e: conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}") @@ -104,12 +107,12 @@ def enqueue_tts_report(conn, text, opus_data): try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 if conn.chat_history_conf == 2: - conn.report_queue.put((2, text, opus_data)) + conn.report_queue.put((2, text, opus_data, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) else: - conn.report_queue.put((2, text, None)) + conn.report_queue.put((2, text, None, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 不上报音频" ) @@ -132,12 +135,12 @@ def enqueue_asr_report(conn, text, opus_data): try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 if conn.chat_history_conf == 2: - conn.report_queue.put((1, text, opus_data)) + conn.report_queue.put((1, text, opus_data, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) else: - conn.report_queue.put((1, text, None)) + conn.report_queue.put((1, text, None, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频" ) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 89f16426..5694c618 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,7 +1,9 @@ import json import asyncio import time +from core.providers.tts.dto.dto import SentenceType from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion +from loguru import logger TAG = __name__ @@ -30,7 +32,7 @@ emoji_map = { } -async def sendAudioMessage(conn, audios, text, text_index=0): +async def sendAudioMessage(conn, sentenceType, audios, text): # 发送句子开始消息 if text is not None: emotion = analyze_emotion(text) @@ -45,18 +47,20 @@ async def sendAudioMessage(conn, audios, text, text_index=0): } ) ) - - if text_index == conn.tts_first_text_index: + pre_buffer = False + if conn.tts.tts_audio_first_sentence and text is not None: conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") + conn.tts.tts_audio_first_sentence = False + pre_buffer = True + await send_tts_message(conn, "sentence_start", text) - is_first_audio = text_index == conn.tts_first_text_index - await sendAudio(conn, audios, pre_buffer=is_first_audio) + await sendAudio(conn, audios, pre_buffer) await send_tts_message(conn, "sentence_end", text) # 发送结束消息（如果是最后一个文本） - if conn.llm_finish_task and text_index == conn.tts_last_text_index: + if conn.llm_finish_task and sentenceType == SentenceType.LAST: await send_tts_message(conn, "stop", None) if conn.close_after_chat: await conn.close() @@ -64,6 +68,8 @@ async def sendAudioMessage(conn, audios, text, text_index=0): # 播放音频 async def sendAudio(conn, audios, pre_buffer=True): + if audios is None or len(audios) == 0: + return # 流控参数优化 frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码 start_time = time.perf_counter() diff --git a/main/xiaozhi-server/core/ota_server.py b/main/xiaozhi-server/core/ota_server.py index 383725d9..89e36869 100644 --- a/main/xiaozhi-server/core/ota_server.py +++ b/main/xiaozhi-server/core/ota_server.py @@ -98,7 +98,6 @@ class SimpleOtaServer: content_type="application/json", ) except Exception as e: - self.logger.bind(tag=TAG).error(f"OTA请求异常: {e}") return_json = {"success": False, "message": "request error."} response = web.Response( text=json.dumps(return_json, separators=(",", ":")), diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index 6606168c..74fb4091 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -155,12 +155,6 @@ class ASRProvider(ASRProviderBase): # f"剩余 {remaining:.2f}秒") return time.time() > self.expire_time - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - def _construct_request_url(self) -> str: """构造请求URL，包含参数""" request = f"{self.base_url}?appkey={self.app_key}" diff --git a/main/xiaozhi-server/core/providers/llm/base.py b/main/xiaozhi-server/core/providers/llm/base.py index 97d0d8e7..77786140 100644 --- a/main/xiaozhi-server/core/providers/llm/base.py +++ b/main/xiaozhi-server/core/providers/llm/base.py @@ -10,7 +10,7 @@ class LLMProviderBase(ABC): """LLM response generator""" pass - def response_no_stream(self, system_prompt, user_prompt): + def response_no_stream(self, system_prompt, user_prompt, **kwargs): try: # 构造对话格式 dialogue = [ @@ -18,7 +18,7 @@ class LLMProviderBase(ABC): {"role": "user", "content": user_prompt} ] result = "" - for part in self.response("", dialogue): + for part in self.response("", dialogue, **kwargs): result += part return result @@ -30,7 +30,7 @@ class LLMProviderBase(ABC): """ Default implementation for function calling (streaming) This should be overridden by providers that support function calls - + Returns: generator that yields either text tokens or a special function call token """ # For providers that don't support functions, just return regular response diff --git a/main/xiaozhi-server/core/providers/llm/coze/coze.py b/main/xiaozhi-server/core/providers/llm/coze/coze.py index 60a04667..19002ac8 100644 --- a/main/xiaozhi-server/core/providers/llm/coze/coze.py +++ b/main/xiaozhi-server/core/providers/llm/coze/coze.py @@ -25,7 +25,7 @@ class LLMProvider(LLMProviderBase): self.session_conversation_map = {} # 存储session_id和conversation_id的映射 check_model_key("CozeLLM", self.personal_access_token) - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): coze_api_token = self.personal_access_token coze_api_base = COZE_CN_BASE_URL diff --git a/main/xiaozhi-server/core/providers/llm/dify/dify.py b/main/xiaozhi-server/core/providers/llm/dify/dify.py index 52ca9853..8b01261f 100644 --- a/main/xiaozhi-server/core/providers/llm/dify/dify.py +++ b/main/xiaozhi-server/core/providers/llm/dify/dify.py @@ -17,7 +17,7 @@ class LLMProvider(LLMProviderBase): self.session_conversation_map = {} # 存储session_id和conversation_id的映射 check_model_key("DifyLLM", self.api_key) - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: # 取最后一条用户消息 last_msg = next(m for m in reversed(dialogue) if m["role"] == "user") diff --git a/main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py b/main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py index b2e9d6b7..a5581541 100644 --- a/main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py +++ b/main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py @@ -16,7 +16,7 @@ class LLMProvider(LLMProviderBase): self.variables = config.get("variables", {}) check_model_key("FastGPTLLM", self.api_key) - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: # 取最后一条用户消息 last_msg = next(m for m in reversed(dialogue) if m["role"] == "user") diff --git a/main/xiaozhi-server/core/providers/llm/gemini/gemini.py b/main/xiaozhi-server/core/providers/llm/gemini/gemini.py index d1cf238e..3369aa2d 100644 --- a/main/xiaozhi-server/core/providers/llm/gemini/gemini.py +++ b/main/xiaozhi-server/core/providers/llm/gemini/gemini.py @@ -112,7 +112,7 @@ class LLMProvider(LLMProviderBase): ] # Gemini文档提到，无需维护session-id，直接用dialogue拼接而成 - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): yield from self._generate(dialogue, None) def response_with_functions(self, session_id, dialogue, functions=None): diff --git a/main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py b/main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py index c436fc9e..f203e5f7 100644 --- a/main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py +++ b/main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py @@ -14,7 +14,7 @@ class LLMProvider(LLMProviderBase): self.base_url = config.get("base_url", config.get("url")) # 默认使用 base_url self.api_url = f"{self.base_url}/api/conversation/process" # 拼接完整的 API URL - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: # home assistant语音助手自带意图，无需使用xiaozhi ai自带的，只需要把用户说的话传递给home assistant即可 diff --git a/main/xiaozhi-server/core/providers/llm/ollama/ollama.py b/main/xiaozhi-server/core/providers/llm/ollama/ollama.py index aad6354f..c973cacd 100644 --- a/main/xiaozhi-server/core/providers/llm/ollama/ollama.py +++ b/main/xiaozhi-server/core/providers/llm/ollama/ollama.py @@ -18,13 +18,13 @@ class LLMProvider(LLMProviderBase): self.client = OpenAI( base_url=self.base_url, - api_key="ollama" # Ollama doesn't need an API key but OpenAI client requires one + api_key="ollama", # Ollama doesn't need an API key but OpenAI client requires one ) # 检查是否是qwen3模型 self.is_qwen3 = self.model_name and self.model_name.lower().startswith("qwen3") - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令 if self.is_qwen3: @@ -35,7 +35,9 @@ class LLMProvider(LLMProviderBase): for i in range(len(dialogue_copy) - 1, -1, -1): if dialogue_copy[i]["role"] == "user": # 在用户消息前添加/no_think指令 - dialogue_copy[i]["content"] = "/no_think " + dialogue_copy[i]["content"] + dialogue_copy[i]["content"] = ( + "/no_think " + dialogue_copy[i]["content"] + ) logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令") break @@ -43,9 +45,7 @@ class LLMProvider(LLMProviderBase): dialogue = dialogue_copy responses = self.client.chat.completions.create( - model=self.model_name, - messages=dialogue, - stream=True + model=self.model_name, messages=dialogue, stream=True ) is_active = True # 用于处理跨chunk的标签 @@ -53,29 +53,33 @@ class LLMProvider(LLMProviderBase): for chunk in responses: try: - delta = chunk.choices[0].delta if getattr(chunk, 'choices', None) else None - content = delta.content if hasattr(delta, 'content') else '' + delta = ( + chunk.choices[0].delta + if getattr(chunk, "choices", None) + else None + ) + content = delta.content if hasattr(delta, "content") else "" if content: # 将内容添加到缓冲区 buffer += content # 处理缓冲区中的标签 - while '' in buffer and '' in buffer: + while "" in buffer and "" in buffer: # 找到完整的标签并移除 - pre = buffer.split('', 1)[0] - post = buffer.split('', 1)[1] + pre = buffer.split("", 1)[0] + post = buffer.split("", 1)[1] buffer = pre + post # 处理只有开始标签的情况 - if '' in buffer: + if "" in buffer: is_active = False - buffer = buffer.split('', 1)[0] + buffer = buffer.split("", 1)[0] # 处理只有结束标签的情况 - if '' in buffer: + if "" in buffer: is_active = True - buffer = buffer.split('', 1)[1] + buffer = buffer.split("", 1)[1] # 如果当前处于活动状态且缓冲区有内容，则输出 if is_active and buffer: @@ -100,7 +104,9 @@ class LLMProvider(LLMProviderBase): for i in range(len(dialogue_copy) - 1, -1, -1): if dialogue_copy[i]["role"] == "user": # 在用户消息前添加/no_think指令 - dialogue_copy[i]["content"] = "/no_think " + dialogue_copy[i]["content"] + dialogue_copy[i]["content"] = ( + "/no_think " + dialogue_copy[i]["content"] + ) logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令") break @@ -119,9 +125,15 @@ class LLMProvider(LLMProviderBase): for chunk in stream: try: - delta = chunk.choices[0].delta if getattr(chunk, 'choices', None) else None - content = delta.content if hasattr(delta, 'content') else None - tool_calls = delta.tool_calls if hasattr(delta, 'tool_calls') else None + delta = ( + chunk.choices[0].delta + if getattr(chunk, "choices", None) + else None + ) + content = delta.content if hasattr(delta, "content") else None + tool_calls = ( + delta.tool_calls if hasattr(delta, "tool_calls") else None + ) # 如果是工具调用，直接传递 if tool_calls: @@ -134,21 +146,21 @@ class LLMProvider(LLMProviderBase): buffer += content # 处理缓冲区中的标签 - while '' in buffer and '' in buffer: + while "" in buffer and "" in buffer: # 找到完整的标签并移除 - pre = buffer.split('', 1)[0] - post = buffer.split('', 1)[1] + pre = buffer.split("", 1)[0] + post = buffer.split("", 1)[1] buffer = pre + post # 处理只有开始标签的情况 - if '' in buffer: + if "" in buffer: is_active = False - buffer = buffer.split('', 1)[0] + buffer = buffer.split("", 1)[0] # 处理只有结束标签的情况 - if '' in buffer: + if "" in buffer: is_active = True - buffer = buffer.split('', 1)[1] + buffer = buffer.split("", 1)[1] # 如果当前处于活动状态且缓冲区有内容，则输出 if is_active and buffer: diff --git a/main/xiaozhi-server/core/providers/llm/openai/openai.py b/main/xiaozhi-server/core/providers/llm/openai/openai.py index a20067af..bc0e7f21 100644 --- a/main/xiaozhi-server/core/providers/llm/openai/openai.py +++ b/main/xiaozhi-server/core/providers/llm/openai/openai.py @@ -16,26 +16,37 @@ class LLMProvider(LLMProviderBase): self.base_url = config.get("base_url") else: self.base_url = config.get("url") - max_tokens = config.get("max_tokens") - if max_tokens is None or max_tokens == "": - max_tokens = 500 - try: - max_tokens = int(max_tokens) - except (ValueError, TypeError): - max_tokens = 500 - self.max_tokens = max_tokens + param_defaults = { + "max_tokens": (500, int), + "temperature": (0.7, lambda x: round(float(x), 1)), + "top_p": (1.0, lambda x: round(float(x), 1)), + "frequency_penalty": (0, lambda x: round(float(x), 1)) + } + + for param, (default, converter) in param_defaults.items(): + value = config.get(param) + try: + setattr(self, param, converter(value) if value not in (None, "") else default) + except (ValueError, TypeError): + setattr(self, param, default) + + logger.debug( + f"意图识别参数初始化: {self.temperature}, {self.max_tokens}, {self.top_p}, {self.frequency_penalty}") check_model_key("LLM", self.api_key) self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url) - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: responses = self.client.chat.completions.create( model=self.model_name, messages=dialogue, stream=True, - max_tokens=self.max_tokens, + max_tokens=kwargs.get("max_tokens", self.max_tokens), + temperature=kwargs.get("temperature", self.temperature), + top_p=kwargs.get("top_p", self.top_p), + frequency_penalty=kwargs.get("frequency_penalty", self.frequency_penalty), ) is_active = True diff --git a/main/xiaozhi-server/core/providers/llm/xinference/xinference.py b/main/xiaozhi-server/core/providers/llm/xinference/xinference.py index b90b0418..a4e9a5e6 100644 --- a/main/xiaozhi-server/core/providers/llm/xinference/xinference.py +++ b/main/xiaozhi-server/core/providers/llm/xinference/xinference.py @@ -15,39 +15,45 @@ class LLMProvider(LLMProviderBase): # 如果没有v1，增加v1 if not self.base_url.endswith("/v1"): self.base_url = f"{self.base_url}/v1" - - logger.bind(tag=TAG).info(f"Initializing Xinference LLM provider with model: {self.model_name}, base_url: {self.base_url}") + + logger.bind(tag=TAG).info( + f"Initializing Xinference LLM provider with model: {self.model_name}, base_url: {self.base_url}" + ) try: self.client = OpenAI( base_url=self.base_url, - api_key="xinference" # Xinference has a similar setup to Ollama where it doesn't need an actual key + api_key="xinference", # Xinference has a similar setup to Ollama where it doesn't need an actual key ) logger.bind(tag=TAG).info("Xinference client initialized successfully") except Exception as e: logger.bind(tag=TAG).error(f"Error initializing Xinference client: {e}") raise - def response(self, session_id, dialogue): + def response(self, session_id, dialogue, **kwargs): try: - logger.bind(tag=TAG).debug(f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}") - responses = self.client.chat.completions.create( - model=self.model_name, - messages=dialogue, - stream=True + logger.bind(tag=TAG).debug( + f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}" ) - is_active=True + responses = self.client.chat.completions.create( + model=self.model_name, messages=dialogue, stream=True + ) + is_active = True for chunk in responses: try: - delta = chunk.choices[0].delta if getattr(chunk, 'choices', None) else None - content = delta.content if hasattr(delta, 'content') else '' + delta = ( + chunk.choices[0].delta + if getattr(chunk, "choices", None) + else None + ) + content = delta.content if hasattr(delta, "content") else "" if content: - if '' in content: + if "" in content: is_active = False - content = content.split('')[0] - if '' in content: + content = content.split("")[0] + if "" in content: is_active = True - content = content.split('')[-1] + content = content.split("")[-1] if is_active: yield content except Exception as e: @@ -59,10 +65,14 @@ class LLMProvider(LLMProviderBase): def response_with_functions(self, session_id, dialogue, functions=None): try: - logger.bind(tag=TAG).debug(f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}") + logger.bind(tag=TAG).debug( + f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}" + ) if functions: - logger.bind(tag=TAG).debug(f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}") - + logger.bind(tag=TAG).debug( + f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}" + ) + stream = self.client.chat.completions.create( model=self.model_name, messages=dialogue, @@ -74,7 +84,7 @@ class LLMProvider(LLMProviderBase): delta = chunk.choices[0].delta content = delta.content tool_calls = delta.tool_calls - + if content: yield content, tool_calls elif tool_calls: @@ -82,4 +92,7 @@ class LLMProvider(LLMProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"Error in Xinference function call: {e}") - yield {"type": "content", "content": f"【Xinference服务响应异常: {str(e)}】"} + yield { + "type": "content", + "content": f"【Xinference服务响应异常: {str(e)}】", + } diff --git a/main/xiaozhi-server/core/providers/memory/base.py b/main/xiaozhi-server/core/providers/memory/base.py index 19dcabb8..f404f15e 100644 --- a/main/xiaozhi-server/core/providers/memory/base.py +++ b/main/xiaozhi-server/core/providers/memory/base.py @@ -9,7 +9,13 @@ class MemoryProviderBase(ABC): def __init__(self, config): self.config = config self.role_id = None - self.llm = None + + def set_llm(self, llm): + self.llm = llm + # 获取模型名称和类型信息 + model_name = getattr(llm, "model_name", str(llm.__class__.__name__)) + # 记录更详细的日志 + logger.bind(tag=TAG).info(f"记忆总结设置LLM: {model_name}") @abstractmethod async def save_memory(self, msgs): diff --git a/main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py b/main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py index a916e6c9..9f855cc6 100644 --- a/main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py +++ b/main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py @@ -107,7 +107,7 @@ TAG = __name__ class MemoryProvider(MemoryProviderBase): def __init__(self, config, summary_memory): super().__init__(config) - self.short_momery = "" + self.short_memory = "" self.save_to_file = True self.memory_path = get_project_dir() + "data/.memory.yaml" self.load_memory(summary_memory) @@ -122,7 +122,7 @@ class MemoryProvider(MemoryProviderBase): def load_memory(self, summary_memory): # api获取到总结记忆后直接返回 if summary_memory or not self.save_to_file: - self.short_momery = summary_memory + self.short_memory = summary_memory return all_memory = {} @@ -130,18 +130,21 @@ class MemoryProvider(MemoryProviderBase): with open(self.memory_path, "r", encoding="utf-8") as f: all_memory = yaml.safe_load(f) or {} if self.role_id in all_memory: - self.short_momery = all_memory[self.role_id] + self.short_memory = all_memory[self.role_id] def save_memory_to_file(self): all_memory = {} if os.path.exists(self.memory_path): with open(self.memory_path, "r", encoding="utf-8") as f: all_memory = yaml.safe_load(f) or {} - all_memory[self.role_id] = self.short_momery + all_memory[self.role_id] = self.short_memory with open(self.memory_path, "w", encoding="utf-8") as f: yaml.dump(all_memory, f, allow_unicode=True) async def save_memory(self, msgs): + # 打印使用的模型信息 + model_info = getattr(self.llm, "model_name", str(self.llm.__class__.__name__)) + logger.bind(tag=TAG).debug(f"使用记忆保存模型: {model_info}") if self.llm is None: logger.bind(tag=TAG).error("LLM is not set for memory provider") return None @@ -155,31 +158,39 @@ class MemoryProvider(MemoryProviderBase): msgStr += f"User: {msg.content}\n" elif msg.role == "assistant": msgStr += f"Assistant: {msg.content}\n" - if self.short_momery and len(self.short_momery) > 0: + if self.short_memory and len(self.short_memory) > 0: msgStr += "历史记忆：\n" - msgStr += self.short_momery + msgStr += self.short_memory # 当前时间 time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) msgStr += f"当前时间：{time_str}" if self.save_to_file: - result = self.llm.response_no_stream(short_term_memory_prompt, msgStr) + result = self.llm.response_no_stream( + short_term_memory_prompt, + msgStr, + max_tokens=2000, + temperature=0.2, + ) json_str = extract_json_data(result) try: json.loads(json_str) # 检查json格式是否正确 - self.short_momery = json_str + self.short_memory = json_str self.save_memory_to_file() except Exception as e: print("Error:", e) else: result = self.llm.response_no_stream( - short_term_memory_prompt_only_content, msgStr + short_term_memory_prompt_only_content, + msgStr, + max_tokens=2000, + temperature=0.2, ) save_mem_local_short(self.role_id, result) logger.bind(tag=TAG).info(f"Save memory successful - Role: {self.role_id}") - return self.short_momery + return self.short_memory async def query_memory(self, query: str) -> str: - return self.short_momery + return self.short_memory diff --git a/main/xiaozhi-server/core/providers/tts/aliyun.py b/main/xiaozhi-server/core/providers/tts/aliyun.py index 61ab4364..1fda8d77 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun.py @@ -1,4 +1,3 @@ -import os import uuid import json import hmac @@ -8,61 +7,74 @@ import requests from datetime import datetime from core.providers.tts.base import TTSProviderBase -import http.client -import urllib.parse import time import uuid from urllib import parse + + class AccessToken: @staticmethod def _encode_text(text): encoded_text = parse.quote_plus(text) - return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~') - + 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', '~') - + 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'} + 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) # print('规范化的请求字符串: %s' % query_string) # 构造待签名字符串 - string_to_sign = 'GET' + '&' + AccessToken._encode_text('/') + '&' + AccessToken._encode_text(query_string) + string_to_sign = ( + "GET" + + "&" + + AccessToken._encode_text("/") + + "&" + + AccessToken._encode_text(query_string) + ) # print('待签名的字符串: %s' % string_to_sign) # 计算签名 - secreted_string = hmac.new(bytes(access_key_secret + '&', encoding='utf-8'), - bytes(string_to_sign, encoding='utf-8'), - hashlib.sha1).digest() + 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) # print('签名: %s' % signature) # 进行URL编码 signature = AccessToken._encode_text(signature) # print('URL编码后的签名: %s' % signature) # 调用服务 - full_url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s' % (signature, query_string) + full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % ( + signature, + query_string, + ) # print('url: %s' % full_url) # 提交HTTP GET请求 response = requests.get(full_url) if response.ok: root_obj = response.json() - key = 'Token' + key = "Token" if key in root_obj: - token = root_obj[key]['Id'] - expire_time = root_obj[key]['ExpireTime'] + token = root_obj[key]["Id"] + expire_time = root_obj[key]["ExpireTime"] return token, expire_time # print(response.text) return None, None @@ -70,26 +82,36 @@ class AccessToken: class TTSProvider(TTSProviderBase): - def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) - + # 新增空值判断逻辑 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", "wav") - self.sample_rate = config.get("sample_rate", 16000) - self.voice = config.get("voice", "xiaoyun") - self.volume = config.get("volume", 50) - self.speech_rate = config.get("speech_rate", 0) - self.pitch_rate = config.get("pitch_rate", 0) + + sample_rate = config.get("sample_rate", "16000") + self.sample_rate = int(sample_rate) if sample_rate else 16000 + + if config.get("private_voice"): + self.voice = config.get("private_voice") + else: + self.voice = config.get("voice", "xiaoyun") + + 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 + self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com") self.api_url = f"https://{self.host}/stream/v1/tts" - self.header = { - "Content-Type": "application/json" - } + self.header = {"Content-Type": "application/json"} if self.access_key_id and self.access_key_secret: # 使用密钥对生成临时token @@ -99,35 +121,30 @@ class TTSProvider(TTSProviderBase): 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 + self.access_key_id, self.access_key_secret ) if not expire_time_str: 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" - ) + 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") @@ -142,8 +159,6 @@ class TTSProvider(TTSProviderBase): # f"过期时间 {datetime.fromtimestamp(self.expire_time)} | " # f"剩余 {remaining:.2f}秒") return time.time() > self.expire_time - def generate_filename(self, extension=".wav"): - return os.path.join(self.output_file, f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}") async def text_to_speak(self, text, output_file): if self._is_token_expired(): @@ -158,21 +173,27 @@ class TTSProvider(TTSProviderBase): "voice": self.voice, "volume": self.volume, "speech_rate": self.speech_rate, - "pitch_rate": self.pitch_rate + "pitch_rate": self.pitch_rate, } # print(self.api_url, json.dumps(request_json, ensure_ascii=False)) try: - resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header) + resp = requests.post( + self.api_url, json.dumps(request_json), headers=self.header + ) if resp.status_code == 401: # Token过期特殊处理 self._refresh_token() - resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header) + resp = requests.post( + self.api_url, json.dumps(request_json), headers=self.header + ) # 检查返回请求数据的mime类型是否是audio/***，是则保存到指定路径下；返回的是binary格式的 - if resp.headers['Content-Type'].startswith('audio/'): - with open(output_file, 'wb') as f: + if resp.headers["Content-Type"].startswith("audio/"): + with open(output_file, "wb") as f: f.write(resp.content) return output_file else: - raise Exception(f"{__name__} status_code: {resp.status_code} response: {resp.content}") + raise Exception( + f"{__name__} status_code: {resp.status_code} response: {resp.content}" + ) except Exception as e: raise Exception(f"{__name__} error: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index f2632c02..0b17eef5 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -1,9 +1,27 @@ -import asyncio -from config.logger import setup_logging import os +import queue +import uuid +import asyncio +import threading +from core.utils import p3 +from datetime import datetime +from core.utils import textUtils from abc import ABC, abstractmethod -from core.utils.tts import MarkdownCleaner +from config.logger import setup_logging from core.utils.util import audio_to_data +from core.utils.tts import MarkdownCleaner +from core.utils.output_counter import add_device_output +from core.handle.reportHandle import enqueue_tts_report +from core.handle.sendAudioHandle import sendAudioMessage +from core.providers.tts.dto.dto import ( + TTSMessageDTO, + SentenceType, + ContentType, + InterfaceType, +) + + +import traceback TAG = __name__ logger = setup_logging() @@ -11,12 +29,51 @@ logger = setup_logging() class TTSProviderBase(ABC): def __init__(self, config, delete_audio_file): + self.interface_type = InterfaceType.NON_STREAM + self.conn = None + self.tts_timeout = 10 self.delete_audio_file = delete_audio_file - self.output_file = config.get("output_dir") + self.output_file = config.get("output_dir", "tmp/") + self.tts_text_queue = queue.Queue() + self.tts_audio_queue = queue.Queue() + self.tts_audio_first_sentence = True - @abstractmethod - def generate_filename(self): - pass + self.tts_text_buff = [] + self.punctuations = ( + "。", + ".", + "？", + "?", + "！", + "!", + "；", + ";", + "：", + ) + self.first_sentence_punctuations = ( + "，", + "~", + "、", + ",", + "。", + ".", + "？", + "?", + "！", + "!", + "；", + ";", + "：", + ) + self.tts_stop_request = False + self.processed_chars = 0 + self.is_first_sentence = True + + def generate_filename(self, extension=".wav"): + return os.path.join( + self.output_file, + f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", + ) def to_tts(self, text): tmp_file = self.generate_filename() @@ -60,3 +117,224 @@ class TTSProviderBase(ABC): def audio_to_opus_data(self, audio_file_path): """音频文件转换为Opus编码""" return audio_to_data(audio_file_path, is_opus=True) + + def tts_one_sentence( + self, + conn, + content_type, + content_detail=None, + content_file=None, + sentence_id=None, + ): + """发送一句话""" + if not sentence_id: + if conn.sentence_id: + sentence_id = conn.sentence_id + else: + sentence_id = str(uuid.uuid4()).replace("-", "") + conn.sentence_id = sentence_id + self.tts_text_queue.put( + TTSMessageDTO( + sentence_id=sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) + self.tts_text_queue.put( + TTSMessageDTO( + sentence_id=sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=content_type, + content_detail=content_detail, + 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 + self.tts_timeout = conn.config.get("tts_timeout", 10) + # tts 消化线程 + self.tts_priority_thread = threading.Thread( + target=self.tts_text_priority_thread, daemon=True + ) + self.tts_priority_thread.start() + + # 音频播放 消化线程 + self.audio_play_priority_thread = threading.Thread( + target=self._audio_play_priority_thread, daemon=True + ) + self.audio_play_priority_thread.start() + + # 这里默认是非流式的处理方式 + # 流式处理方式请在子类中重写 + def tts_text_priority_thread(self): + while not self.conn.stop_event.is_set(): + try: + message = self.tts_text_queue.get(timeout=1) + if message.sentence_type == SentenceType.FIRST: + # 初始化参数 + self.tts_stop_request = False + self.processed_chars = 0 + self.tts_text_buff = [] + self.is_first_sentence = True + self.tts_audio_first_sentence = True + elif ContentType.TEXT == message.content_type: + self.tts_text_buff.append(message.content_detail) + segment_text = self._get_segment_text() + if segment_text: + tts_file = self.to_tts(segment_text) + if tts_file: + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, segment_text) + ) + elif ContentType.FILE == message.content_type: + self._process_remaining_text() + tts_file = message.content_file + if tts_file and os.path.exists(tts_file): + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, message.content_detail) + ) + + if message.sentence_type == SentenceType.LAST: + self._process_remaining_text() + self.tts_audio_queue.put( + (message.sentence_type, [], message.content_detail) + ) + + except queue.Empty: + continue + except Exception as e: + logger.bind(tag=TAG).error( + f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) + + def _audio_play_priority_thread(self): + while not self.conn.stop_event.is_set(): + text = None + try: + try: + sentence_type, audio_datas, text = self.tts_audio_queue.get( + timeout=1 + ) + except queue.Empty: + if self.conn.stop_event.is_set(): + break + continue + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self.conn.loop, + ) + future.result() + if self.conn.max_output_size > 0 and text: + add_device_output(self.conn.headers.get("device-id"), len(text)) + enqueue_tts_report(self.conn, text, audio_datas) + except Exception as e: + logger.bind(tag=TAG).error( + f"audio_play_priority priority_thread: {text} {e}" + ) + + async def start_session(self, session_id): + pass + + async def finish_session(self, session_id): + pass + + async def close(self): + """资源清理方法""" + if hasattr(self, "ws") and self.ws: + await self.ws.close() + + def _get_segment_text(self): + # 合并当前全部文本并处理未分割部分 + full_text = "".join(self.tts_text_buff) + current_text = full_text[self.processed_chars :] # 从未处理的位置开始 + last_punct_pos = -1 + + # 根据是否是第一句话选择不同的标点符号集合 + punctuations_to_use = ( + self.first_sentence_punctuations + if self.is_first_sentence + else self.punctuations + ) + + for punct in punctuations_to_use: + pos = current_text.rfind(punct) + if (pos != -1 and last_punct_pos == -1) or ( + pos != -1 and pos < last_punct_pos + ): + last_punct_pos = pos + + if last_punct_pos != -1: + segment_text_raw = current_text[: last_punct_pos + 1] + segment_text = textUtils.get_string_no_punctuation_or_emoji( + segment_text_raw + ) + self.processed_chars += len(segment_text_raw) # 更新已处理字符位置 + + # 如果是第一句话，在找到第一个逗号后，将标志设置为False + if self.is_first_sentence: + self.is_first_sentence = False + + return segment_text + elif self.tts_stop_request and current_text: + segment_text = current_text + self.is_first_sentence = True # 重置标志 + return segment_text + else: + return None + + def _process_audio_file(self, tts_file): + """处理音频文件并转换为指定格式 + + Args: + tts_file: 音频文件路径 + content_detail: 内容详情 + + Returns: + tuple: (sentence_type, audio_datas, content_detail) + """ + audio_datas = [] + if tts_file.endswith(".p3"): + audio_datas, _ = p3.decode_opus_from_file(tts_file) + elif self.conn.audio_format == "pcm": + audio_datas, _ = self.audio_to_pcm_data(tts_file) + else: + audio_datas, _ = self.audio_to_opus_data(tts_file) + + if ( + self.delete_audio_file + and tts_file is not None + and os.path.exists(tts_file) + and tts_file.startswith(self.output_file) + ): + os.remove(tts_file) + return audio_datas + + def _process_remaining_text(self): + """处理剩余的文本并生成语音 + + Returns: + bool: 是否成功处理了文本 + """ + full_text = "".join(self.tts_text_buff) + remaining_text = full_text[self.processed_chars :] + if remaining_text: + segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) + if segment_text: + tts_file = self.to_tts(segment_text) + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, audio_datas, segment_text) + ) + self.processed_chars += len(full_text) + return True + return False diff --git a/main/xiaozhi-server/core/providers/tts/cozecn.py b/main/xiaozhi-server/core/providers/tts/cozecn.py index 56314f4f..c3019b67 100644 --- a/main/xiaozhi-server/core/providers/tts/cozecn.py +++ b/main/xiaozhi-server/core/providers/tts/cozecn.py @@ -1,9 +1,4 @@ -import os -import uuid -import json -import base64 import requests -from datetime import datetime from core.providers.tts.base import TTSProviderBase @@ -21,12 +16,6 @@ class TTSProvider(TTSProviderBase): self.host = "api.coze.cn" self.api_url = f"https://{self.host}/v1/audio/speech" - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): request_json = { "model": self.model, @@ -47,4 +36,4 @@ class TTSProvider(TTSProviderBase): file_to_save = open(output_file, "wb") file_to_save.write(data) except Exception as e: - raise Exception(f"{__name__} error: {e}") \ No newline at end of file + raise Exception(f"{__name__} error: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/default.py b/main/xiaozhi-server/core/providers/tts/default.py new file mode 100644 index 00000000..7818d288 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/default.py @@ -0,0 +1,23 @@ +import os +from config.logger import setup_logging +from core.providers.tts.base import TTSProviderBase + +TAG = __name__ +logger = setup_logging() + + +class DefaultTTS(TTSProviderBase): + def __init__(self, config, delete_audio_file=True): + super().__init__(config, delete_audio_file) + self.output_dir = config.get("output_dir", "output") + if not os.path.exists(self.output_dir): + os.makedirs(self.output_dir) + + def generate_filename(self): + """生成唯一的音频文件名""" + import uuid + + return os.path.join(self.output_dir, f"{uuid.uuid4()}.wav") + + async def text_to_speak(self, text, output_file): + logger.bind(tag=TAG).error(f"无法实例化 TTS 服务，请检查配置") diff --git a/main/xiaozhi-server/core/providers/tts/doubao.py b/main/xiaozhi-server/core/providers/tts/doubao.py index 3a0fdde3..78058c31 100644 --- a/main/xiaozhi-server/core/providers/tts/doubao.py +++ b/main/xiaozhi-server/core/providers/tts/doubao.py @@ -1,9 +1,7 @@ -import os import uuid import json import base64 import requests -from datetime import datetime from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase from config.logger import setup_logging @@ -41,12 +39,6 @@ class TTSProvider(TTSProviderBase): self.header = {"Authorization": f"{self.authorization}{self.access_token}"} check_model_key("TTS", self.access_token) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): request_json = { "app": { diff --git a/main/xiaozhi-server/core/providers/tts/dto/dto.py b/main/xiaozhi-server/core/providers/tts/dto/dto.py new file mode 100644 index 00000000..b2317340 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/dto/dto.py @@ -0,0 +1,43 @@ +from enum import Enum +from typing import Union, Optional + + +class SentenceType(Enum): + # 说话阶段 + FIRST = "FIRST" # 首句话 + MIDDLE = "MIDDLE" # 说话中 + LAST = "LAST" # 最后一句 + + +class ContentType(Enum): + # 内容类型 + TEXT = "TEXT" # 文本内容 + FILE = "FILE" # 文件内容 + ACTION = "ACTION" # 动作内容 + + +class InterfaceType(Enum): + # 接口类型 + DUAL_STREAM = "DUAL_STREAM" # 双流式 + SINGLE_STREAM = "SINGLE_STREAM" # 单流式 + NON_STREAM = "NON_STREAM" # 非流式 + + +class TTSMessageDTO: + def __init__( + self, + sentence_id: str, + # 说话阶段 + sentence_type: SentenceType, + # 内容类型 + content_type: ContentType, + # 内容详情，一般是需要转换的文本或者音频的歌词 + content_detail: Optional[str] = None, + # 如果内容类型为文件，则需要传入文件路径 + content_file: Optional[str] = None, + ): + self.sentence_id = sentence_id + self.sentence_type = sentence_type + self.content_type = content_type + self.content_detail = content_detail + self.content_file = content_file diff --git a/main/xiaozhi-server/core/providers/tts/fishspeech.py b/main/xiaozhi-server/core/providers/tts/fishspeech.py index 71f1ac88..23b20e76 100644 --- a/main/xiaozhi-server/core/providers/tts/fishspeech.py +++ b/main/xiaozhi-server/core/providers/tts/fishspeech.py @@ -1,12 +1,9 @@ import base64 -import os -import uuid import requests import ormsgpack from pathlib import Path from pydantic import BaseModel, Field, conint, model_validator from typing_extensions import Annotated -from datetime import datetime from typing import Literal from core.utils.util import check_model_key, parse_string_to_list from core.providers.tts.base import TTSProviderBase @@ -133,12 +130,6 @@ class TTSProvider(TTSProviderBase): self.seed = int(config.get("seed")) if config.get("seed") else None self.api_url = config.get("api_url", "http://127.0.0.1:8080/v1/tts") - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): # Prepare reference data byte_audios = [audio_to_bytes(ref_audio) for ref_audio in self.reference_audio] diff --git a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py index 2d58679d..e76b3f22 100644 --- a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py +++ b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v2.py @@ -1,10 +1,5 @@ -import os -import uuid -import json -import base64 import requests from config.logger import setup_logging -from datetime import datetime from core.providers.tts.base import TTSProviderBase from core.utils.util import parse_string_to_list @@ -71,12 +66,6 @@ class TTSProvider(TTSProviderBase): config.get("aux_ref_audio_paths") ) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): request_json = { "text": text, diff --git a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py index d4da23ed..6b66efc3 100644 --- a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py +++ b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py @@ -1,8 +1,5 @@ -import os -import uuid import requests from config.logger import setup_logging -from datetime import datetime from core.providers.tts.base import TTSProviderBase from core.utils.util import parse_string_to_list @@ -36,12 +33,6 @@ class TTSProvider(TTSProviderBase): self.inp_refs = parse_string_to_list(config.get("inp_refs")) self.if_sr = str(config.get("if_sr", False)).lower() in ("true", "1", "yes") - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): request_params = { "refer_wav_path": self.refer_wav_path, @@ -67,4 +58,3 @@ class TTSProvider(TTSProviderBase): error_msg = f"GPT_SoVITS_V3 TTS请求失败: {resp.status_code} - {resp.text}" logger.bind(tag=TAG).error(error_msg) raise Exception(error_msg) - diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py new file mode 100644 index 00000000..095437a9 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -0,0 +1,462 @@ +import os +import uuid +import json +import queue +import asyncio +import threading +import traceback +import websockets +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.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType + +TAG = __name__ +logger = setup_logging() + +PROTOCOL_VERSION = 0b0001 +DEFAULT_HEADER_SIZE = 0b0001 + +# Message Type: +FULL_CLIENT_REQUEST = 0b0001 +AUDIO_ONLY_RESPONSE = 0b1011 +FULL_SERVER_RESPONSE = 0b1001 +ERROR_INFORMATION = 0b1111 + +# Message Type Specific Flags +MsgTypeFlagNoSeq = 0b0000 # Non-terminal packet with no sequence +MsgTypeFlagPositiveSeq = 0b1 # Non-terminal packet with sequence > 0 +MsgTypeFlagLastNoSeq = 0b10 # last packet with no sequence +MsgTypeFlagNegativeSeq = 0b11 # Payload contains event number (int32) +MsgTypeFlagWithEvent = 0b100 +# Message Serialization +NO_SERIALIZATION = 0b0000 +JSON = 0b0001 +# Message Compression +COMPRESSION_NO = 0b0000 +COMPRESSION_GZIP = 0b0001 + +EVENT_NONE = 0 +EVENT_Start_Connection = 1 + +EVENT_FinishConnection = 2 + +EVENT_ConnectionStarted = 50 # 成功建连 + +EVENT_ConnectionFailed = 51 # 建连失败（可能是无法通过权限认证） + +EVENT_ConnectionFinished = 52 # 连接结束 + +# 上行Session事件 +EVENT_StartSession = 100 + +EVENT_FinishSession = 102 +# 下行Session事件 +EVENT_SessionStarted = 150 +EVENT_SessionFinished = 152 + +EVENT_SessionFailed = 153 + +# 上行通用事件 +EVENT_TaskRequest = 200 + +# 下行TTS事件 +EVENT_TTSSentenceStart = 350 + +EVENT_TTSSentenceEnd = 351 + +EVENT_TTSResponse = 352 + + +class Header: + def __init__( + self, + protocol_version=PROTOCOL_VERSION, + header_size=DEFAULT_HEADER_SIZE, + message_type: int = 0, + message_type_specific_flags: int = 0, + serial_method: int = NO_SERIALIZATION, + compression_type: int = COMPRESSION_NO, + reserved_data=0, + ): + self.header_size = header_size + self.protocol_version = protocol_version + self.message_type = message_type + self.message_type_specific_flags = message_type_specific_flags + self.serial_method = serial_method + self.compression_type = compression_type + self.reserved_data = reserved_data + + def as_bytes(self) -> bytes: + return bytes( + [ + (self.protocol_version << 4) | self.header_size, + (self.message_type << 4) | self.message_type_specific_flags, + (self.serial_method << 4) | self.compression_type, + self.reserved_data, + ] + ) + + +class Optional: + def __init__( + self, event: int = EVENT_NONE, sessionId: str = None, sequence: int = None + ): + self.event = event + self.sessionId = sessionId + self.errorCode: int = 0 + self.connectionId: str | None = None + self.response_meta_json: str | None = None + self.sequence = sequence + + # 转成 byte 序列 + def as_bytes(self) -> bytes: + option_bytes = bytearray() + if self.event != EVENT_NONE: + option_bytes.extend(self.event.to_bytes(4, "big", signed=True)) + if self.sessionId is not None: + session_id_bytes = str.encode(self.sessionId) + size = len(session_id_bytes).to_bytes(4, "big", signed=True) + option_bytes.extend(size) + option_bytes.extend(session_id_bytes) + if self.sequence is not None: + option_bytes.extend(self.sequence.to_bytes(4, "big", signed=True)) + return option_bytes + + +class Response: + def __init__(self, header: Header, optional: Optional): + self.optional = optional + self.header = header + self.payload: bytes | None = None + + def __str__(self): + return super().__str__() + + +class TTSProvider(TTSProviderBase): + def __init__(self, config, delete_audio_file): + super().__init__(config, delete_audio_file) + self.interface_type = InterfaceType.DUAL_STREAM + self.appId = config.get("appid") + self.access_token = config.get("access_token") + self.cluster = config.get("cluster") + self.resource_id = config.get("resource_id") + if config.get("private_voice"): + self.speaker = config.get("private_voice") + else: + self.speaker = config.get("speaker") + self.voice = config.get("voice") + self.ws_url = config.get("ws_url") + self.authorization = config.get("authorization") + self.header = {"Authorization": f"{self.authorization}{self.access_token}"} + self.enable_two_way = True + self.start_connection_flag = False + self.tts_text = "" + # 合成文字语音后，播放的音频文件列表 + self.before_stop_play_files = [] + self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( + sample_rate=16000, channels=1, frame_size_ms=60 + ) + check_model_key("TTS", self.access_token) + + ################################################################################### + # 火山双流式TTS重写父类的方法--开始 + ################################################################################### + + async def open_audio_channels(self, conn): + await super().open_audio_channels(conn) + ws_header = { + "X-Api-App-Key": self.appId, + "X-Api-Access-Key": self.access_token, + "X-Api-Resource-Id": self.resource_id, + "X-Api-Connect-Id": uuid.uuid4(), + } + self.ws = await websockets.connect( + self.ws_url, additional_headers=ws_header, max_size=1000000000 + ) + tts_priority = threading.Thread( + target=self._start_monitor_tts_response_thread, daemon=True + ) + tts_priority.start() + + 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}" + ) + if message.sentence_type == SentenceType.FIRST: + # 初始化参数 + future = asyncio.run_coroutine_threadsafe( + self.start_session(self.conn.sentence_id), loop=self.conn.loop + ) + future.result() + self.tts_audio_first_sentence = True + self.before_stop_play_files.clear() + elif ContentType.TEXT == message.content_type: + if message.content_detail: + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(message.content_detail, None), + loop=self.conn.loop, + ) + future.result() + elif ContentType.FILE == message.content_type: + self.before_stop_play_files.append( + (message.content_file, message.content_detail) + ) + + if message.sentence_type == SentenceType.LAST: + future = asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), loop=self.conn.loop + ) + future.result() + + 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, _): + # 发送文本 + await self.send_text(self.speaker, text, self.conn.sentence_id) + return + + ################################################################################### + # 火山双流式TTS重写父类的方法--结束 + ################################################################################### + def _start_monitor_tts_response_thread(self): + # 初始化链接 + asyncio.run_coroutine_threadsafe( + self._start_monitor_tts_response(), loop=self.conn.loop + ) + + async def _start_monitor_tts_response(self): + while not self.conn.stop_event.is_set(): + try: + msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop + res = self.parser_response(msg) + self.print_response(res, "send_text res:") + + if res.optional.event == EVENT_TTSSentenceStart: + json_data = json.loads(res.payload.decode("utf-8")) + self.tts_text = json_data.get("text", "") + logger.bind(tag=TAG).info(f"语音生成成功: {self.tts_text}") + self.tts_audio_queue.put((SentenceType.FIRST, [], self.tts_text)) + elif ( + res.optional.event == EVENT_TTSResponse + and res.header.message_type == AUDIO_ONLY_RESPONSE + ): + logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") + opus_datas = self.wav_to_opus_data_audio_raw(res.payload) + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～{len(opus_datas)}" + ) + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, None)) + elif res.optional.event == EVENT_TTSSentenceEnd: + logger.bind(tag=TAG).debug(f"句子结束～～{self.tts_text}") + elif res.optional.event == EVENT_SessionFinished: + logger.bind(tag=TAG).debug(f"会话结束～～") + for tts_file, text in self.before_stop_play_files: + if tts_file and os.path.exists(tts_file): + 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)) + continue + except websockets.ConnectionClosed: + break # 连接关闭时退出监听 + except Exception as e: + logger.bind(tag=TAG).error(f"Error in _start_monitor_tts_response: {e}") + traceback.print_exc() + continue + + async def send_event( + self, header: bytes, optional: bytes | None = None, payload: bytes = None + ): + full_client_request = bytearray(header) + if optional is not None: + full_client_request.extend(optional) + if payload is not None: + payload_size = len(payload).to_bytes(4, "big", signed=True) + full_client_request.extend(payload_size) + full_client_request.extend(payload) + await self.ws.send(full_client_request) + + async def send_text(self, speaker: str, text: str, session_id): + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional(event=EVENT_TaskRequest, sessionId=session_id).as_bytes() + payload = self.get_payload_bytes( + event=EVENT_TaskRequest, text=text, speaker=speaker + ) + return await self.send_event(header, optional, payload) + + # 读取 res 数组某段 字符串内容 + def read_res_content(self, res: bytes, offset: int): + content_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) + offset += 4 + content = str(res[offset : offset + content_size]) + offset += content_size + return content, offset + + # 读取 payload + def read_res_payload(self, res: bytes, offset: int): + payload_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) + offset += 4 + payload = res[offset : offset + payload_size] + offset += payload_size + return payload, offset + + def parser_response(self, res) -> Response: + if isinstance(res, str): + raise RuntimeError(res) + response = Response(Header(), Optional()) + # 解析结果 + # header + header = response.header + num = 0b00001111 + header.protocol_version = res[0] >> 4 & num + header.header_size = res[0] & 0x0F + header.message_type = (res[1] >> 4) & num + header.message_type_specific_flags = res[1] & 0x0F + header.serialization_method = res[2] >> num + header.message_compression = res[2] & 0x0F + header.reserved = res[3] + # + offset = 4 + optional = response.optional + if header.message_type == FULL_SERVER_RESPONSE or AUDIO_ONLY_RESPONSE: + # read event + if header.message_type_specific_flags == MsgTypeFlagWithEvent: + optional.event = int.from_bytes(res[offset:8], "big", signed=True) + offset += 4 + if optional.event == EVENT_NONE: + return response + # read connectionId + elif optional.event == EVENT_ConnectionStarted: + optional.connectionId, offset = self.read_res_content(res, offset) + elif optional.event == EVENT_ConnectionFailed: + optional.response_meta_json, offset = self.read_res_content( + res, offset + ) + elif ( + optional.event == EVENT_SessionStarted + or optional.event == EVENT_SessionFailed + or optional.event == EVENT_SessionFinished + ): + optional.sessionId, offset = self.read_res_content(res, offset) + optional.response_meta_json, offset = self.read_res_content( + res, offset + ) + else: + optional.sessionId, offset = self.read_res_content(res, offset) + response.payload, offset = self.read_res_payload(res, offset) + + elif header.message_type == ERROR_INFORMATION: + optional.errorCode = int.from_bytes( + res[offset : offset + 4], "big", signed=True + ) + offset += 4 + response.payload, offset = self.read_res_payload(res, offset) + return response + + async def start_connection(self): + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + ).as_bytes() + optional = Optional(event=EVENT_Start_Connection).as_bytes() + payload = str.encode("{}") + return await self.send_event(header, optional, payload) + + def print_response(self, res, tag_msg: str): + logger.bind(tag=TAG).debug(f"===>{tag_msg} header:{res.header.__dict__}") + logger.bind(tag=TAG).debug(f"===>{tag_msg} optional:{res.optional.__dict__}") + + def get_payload_bytes( + self, + uid="1234", + event=EVENT_NONE, + text="", + speaker="", + audio_format="pcm", + audio_sample_rate=16000, + ): + return str.encode( + json.dumps( + { + "user": {"uid": uid}, + "event": event, + "namespace": "BidirectionalTTS", + "req_params": { + "text": text, + "speaker": speaker, + "audio_params": { + "format": audio_format, + "sample_rate": audio_sample_rate, + }, + }, + } + ) + ) + + async def finish_connection(self): + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional(event=EVENT_FinishConnection).as_bytes() + payload = str.encode("{}") + await self.send_event(header, optional, payload) + return + + async def start_session(self, session_id): + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional(event=EVENT_StartSession, sessionId=session_id).as_bytes() + payload = self.get_payload_bytes(event=EVENT_StartSession, speaker=self.speaker) + await self.send_event(header, optional, payload) + logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") + + async def finish_session(self, session_id): + logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional(event=EVENT_FinishSession, sessionId=session_id).as_bytes() + payload = str.encode("{}") + await self.send_event(header, optional, payload) + return + + async def reset(self): + # 关闭之前的对话 + if self.start_connection_flag: + await self.finish_connection() + self.start_connection_flag = False + await self.start_connection() + self.start_connection_flag = True + + async def close(self): + """资源清理方法""" + await self.finish_connection() + await self.ws.close() + + def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False): + opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end) + return opus_datas diff --git a/main/xiaozhi-server/core/providers/tts/openai.py b/main/xiaozhi-server/core/providers/tts/openai.py index a7c4e30d..bdbc9174 100644 --- a/main/xiaozhi-server/core/providers/tts/openai.py +++ b/main/xiaozhi-server/core/providers/tts/openai.py @@ -1,7 +1,4 @@ -import os -import uuid import requests -from datetime import datetime from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase from config.logger import setup_logging @@ -29,12 +26,6 @@ class TTSProvider(TTSProviderBase): self.output_file = config.get("output_dir", "tmp/") check_model_key("TTS", self.api_key) - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): headers = { "Authorization": f"Bearer {self.api_key}", diff --git a/main/xiaozhi-server/core/providers/tts/siliconflow.py b/main/xiaozhi-server/core/providers/tts/siliconflow.py index 9e30f721..8f7fd641 100644 --- a/main/xiaozhi-server/core/providers/tts/siliconflow.py +++ b/main/xiaozhi-server/core/providers/tts/siliconflow.py @@ -1,7 +1,4 @@ -import os -import uuid import requests -from datetime import datetime from core.providers.tts.base import TTSProviderBase @@ -22,12 +19,6 @@ class TTSProvider(TTSProviderBase): self.host = "api.siliconflow.cn" self.api_url = f"https://{self.host}/v1/audio/speech" - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): request_json = { "model": self.model, @@ -47,4 +38,4 @@ class TTSProvider(TTSProviderBase): file_to_save = open(output_file, "wb") file_to_save.write(data) except Exception as e: - raise Exception(f"{__name__} error: {e}") \ No newline at end of file + raise Exception(f"{__name__} error: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/tencent.py b/main/xiaozhi-server/core/providers/tts/tencent.py index f2501e2f..7a55328a 100644 --- a/main/xiaozhi-server/core/providers/tts/tencent.py +++ b/main/xiaozhi-server/core/providers/tts/tencent.py @@ -1,6 +1,5 @@ import hashlib import hmac -import os import time import uuid import json @@ -121,12 +120,6 @@ class TTSProvider(TTSProviderBase): msg = msg.encode("utf-8") return hmac.new(key, msg, hashlib.sha256).digest() - def generate_filename(self, extension=".wav"): - return os.path.join( - self.output_file, - f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", - ) - async def text_to_speak(self, text, output_file): # 构建请求体 request_json = { diff --git a/main/xiaozhi-server/core/providers/tts/ttson.py b/main/xiaozhi-server/core/providers/tts/ttson.py index f3a0fbdc..b3c7797f 100644 --- a/main/xiaozhi-server/core/providers/tts/ttson.py +++ b/main/xiaozhi-server/core/providers/tts/ttson.py @@ -82,4 +82,4 @@ class TTSProvider(TTSProviderBase): except Exception as e: print("error:", e) - raise Exception(f"{__name__}: TTS请求失败") \ No newline at end of file + raise Exception(f"{__name__}: TTS请求失败") diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py new file mode 100644 index 00000000..23d26c39 --- /dev/null +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -0,0 +1,136 @@ +""" +Opus编码工具类 +将PCM音频数据编码为Opus格式 +""" + +import logging +import traceback + +import numpy as np +from typing import List, Optional +from opuslib_next import Encoder +from opuslib_next import constants + + +class OpusEncoderUtils: + """PCM到Opus的编码器""" + + def __init__(self, sample_rate: int, channels: int, frame_size_ms: int): + """ + 初始化Opus编码器 + + Args: + sample_rate: 采样率 (Hz) + channels: 通道数 (1=单声道, 2=立体声) + frame_size_ms: 帧大小 (毫秒) + """ + self.sample_rate = sample_rate + self.channels = channels + self.frame_size_ms = frame_size_ms + # 计算每帧样本数 = 采样率 * 帧大小(毫秒) / 1000 + self.frame_size = (sample_rate * frame_size_ms) // 1000 + # 总帧大小 = 每帧样本数 * 通道数 + self.total_frame_size = self.frame_size * channels + + # 比特率和复杂度设置 + self.bitrate = 24000 # bps + self.complexity = 10 # 最高质量 + + # 缓冲区初始化为空 + self.buffer = np.array([], dtype=np.int16) + + try: + # 创建Opus编码器 + self.encoder = Encoder( + sample_rate, channels, constants.APPLICATION_AUDIO # 音频优化模式 + ) + self.encoder.bitrate = self.bitrate + self.encoder.complexity = self.complexity + self.encoder.signal = constants.SIGNAL_VOICE # 语音信号优化 + except Exception as e: + logging.error(f"初始化Opus编码器失败: {e}") + raise RuntimeError("初始化失败") from e + + def reset_state(self): + """重置编码器状态""" + self.encoder.reset_state() + self.buffer = np.array([], dtype=np.int16) + + def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: + """ + 将PCM数据编码为Opus格式 + + Args: + pcm_data: PCM字节数据 + end_of_stream: 是否为流的结束 + + Returns: + Opus数据包列表 + """ + # 将字节数据转换为short数组 + new_samples = self._convert_bytes_to_shorts(pcm_data) + + # 校验PCM数据 + self._validate_pcm_data(new_samples) + + # 将新数据追加到缓冲区 + self.buffer = np.append(self.buffer, new_samples) + + opus_packets = [] + offset = 0 + + # 处理所有完整帧 + while offset <= len(self.buffer) - self.total_frame_size: + frame = self.buffer[offset : offset + self.total_frame_size] + output = self._encode(frame) + if output: + opus_packets.append(output) + offset += self.total_frame_size + + # 保留未处理的样本 + self.buffer = self.buffer[offset:] + + # 流结束时处理剩余数据 + if end_of_stream and len(self.buffer) > 0: + # 创建最后一帧并用0填充 + last_frame = np.zeros(self.total_frame_size, dtype=np.int16) + last_frame[: len(self.buffer)] = self.buffer + + output = self._encode(last_frame) + if output: + opus_packets.append(output) + self.buffer = np.array([], dtype=np.int16) + + return opus_packets + + def _encode(self, frame: np.ndarray) -> Optional[bytes]: + """编码一帧音频数据""" + try: + # 将numpy数组转换为bytes + frame_bytes = frame.tobytes() + # opuslib要求输入字节数必须是channels*2的倍数 + encoded = self.encoder.encode(frame_bytes, self.frame_size) + return encoded + except Exception as e: + logging.error(f"Opus编码失败: {e}") + traceback.print_exc() + return None + + def _convert_bytes_to_shorts(self, bytes_data: bytes) -> np.ndarray: + """将字节数组转换为short数组 (16位PCM)""" + # 假设输入是小端字节序的16位PCM + return np.frombuffer(bytes_data, dtype=np.int16) + + def _validate_pcm_data(self, pcm_shorts: np.ndarray) -> None: + """验证PCM数据是否有效""" + # 16位PCM数据范围是 -32768 到 32767 + if np.any((pcm_shorts < -32768) | (pcm_shorts > 32767)): + invalid_samples = pcm_shorts[(pcm_shorts < -32768) | (pcm_shorts > 32767)] + logging.warning(f"发现无效PCM样本: {invalid_samples[:5]}...") + # 在实际应用中可以选择裁剪而不是抛出异常 + # np.clip(pcm_shorts, -32768, 32767, out=pcm_shorts) + + def close(self): + """关闭编码器并释放资源""" + # opuslib没有明确的关闭方法，Python的垃圾回收会处理 + pass diff --git a/main/xiaozhi-server/core/utils/textUtils.py b/main/xiaozhi-server/core/utils/textUtils.py new file mode 100644 index 00000000..7fadd3ff --- /dev/null +++ b/main/xiaozhi-server/core/utils/textUtils.py @@ -0,0 +1,34 @@ +def get_string_no_punctuation_or_emoji(s): + """去除字符串首尾的空格、标点符号和表情符号""" + chars = list(s) + # 处理开头的字符 + start = 0 + while start < len(chars) and is_punctuation_or_emoji(chars[start]): + start += 1 + # 处理结尾的字符 + end = len(chars) - 1 + while end >= start and is_punctuation_or_emoji(chars[end]): + end -= 1 + return ''.join(chars[start:end + 1]) + +def is_punctuation_or_emoji(char): + """检查字符是否为空格、指定标点或表情符号""" + # 定义需要去除的中英文标点（包括全角/半角） + punctuation_set = { + '，', ',', # 中文逗号 + 英文逗号 + '。', '.', # 中文句号 + 英文句号 + '！', '!', # 中文感叹号 + 英文感叹号 + '-', '－', # 英文连字符 + 中文全角横线 + '、' # 中文顿号 + } + if char.isspace() or char in punctuation_set: + return True + # 检查表情符号（保留原有逻辑） + code_point = ord(char) + emoji_ranges = [ + (0x1F600, 0x1F64F), (0x1F300, 0x1F5FF), + (0x1F680, 0x1F6FF), (0x1F900, 0x1F9FF), + (0x1FA70, 0x1FAFF), (0x2600, 0x26FF), + (0x2700, 0x27BF) + ] + return any(start <= code_point <= end for start, end in emoji_ranges) \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 0b7e46a1..8f5ded4b 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -269,16 +269,7 @@ def initialize_modules( # 初始化TTS模块 if init_tts: select_tts_module = config["selected_module"]["TTS"] - tts_type = ( - select_tts_module - if "type" not in config["TTS"][select_tts_module] - else config["TTS"][select_tts_module]["type"] - ) - modules["tts"] = tts.create_instance( - tts_type, - config["TTS"][select_tts_module], - str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), - ) + modules["tts"] = initialize_tts(config) logger.bind(tag=TAG).info(f"初始化组件: tts成功 {select_tts_module}") # 初始化LLM模块 @@ -355,6 +346,21 @@ def initialize_modules( return modules +def initialize_tts(config): + select_tts_module = config["selected_module"]["TTS"] + tts_type = ( + select_tts_module + if "type" not in config["TTS"][select_tts_module] + else config["TTS"][select_tts_module]["type"] + ) + new_tts = tts.create_instance( + tts_type, + config["TTS"][select_tts_module], + str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), + ) + return new_tts + + def analyze_emotion(text): """ 分析文本情感并返回对应的emoji名称（支持中英文） @@ -882,7 +888,10 @@ def audio_to_data(audio_file_path, is_opus=True): # 获取原始PCM数据（16位小端） raw_data = audio.raw_data + return pcm_to_data(raw_data, is_opus), duration + +def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) @@ -910,7 +919,7 @@ def audio_to_data(audio_file_path, is_opus=True): datas.append(frame_data) - return datas, duration + return datas def check_vad_update(before_config, new_config): diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 3f9e2fd7..e60f0c87 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -19,16 +19,16 @@ class WebSocketServer: "VAD" in self.config["selected_module"], "ASR" in self.config["selected_module"], "LLM" in self.config["selected_module"], - "TTS" in self.config["selected_module"], + False, "Memory" in self.config["selected_module"], "Intent" in self.config["selected_module"], ) self._vad = modules["vad"] if "vad" in modules else None self._asr = modules["asr"] if "asr" in modules else None - self._tts = modules["tts"] if "tts" in modules else None self._llm = modules["llm"] if "llm" in modules else None self._intent = modules["intent"] if "intent" in modules else None self._memory = modules["memory"] if "memory" in modules else None + self.active_connections = set() async def start(self): @@ -49,7 +49,6 @@ class WebSocketServer: self._vad, self._asr, self._llm, - self._tts, self._memory, self._intent, self, # 传入server实例 @@ -98,7 +97,7 @@ class WebSocketServer: update_vad, update_asr, "LLM" in new_config["selected_module"], - "TTS" in new_config["selected_module"], + False, "Memory" in new_config["selected_module"], "Intent" in new_config["selected_module"], ) @@ -108,8 +107,6 @@ class WebSocketServer: self._vad = modules["vad"] if "asr" in modules: self._asr = modules["asr"] - if "tts" in modules: - self._tts = modules["tts"] if "llm" in modules: self._llm = modules["llm"] if "intent" in modules: diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 76465641..6d184b6b 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -11,6 +11,7 @@ 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 +from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType TAG = __name__ @@ -217,14 +218,36 @@ async def play_local_music(conn, specific_file=None): await send_stt_message(conn, text) conn.dialogue.put(Message(role="assistant", content=text)) - conn.recode_first_last_text(text, 0) - future = conn.executor.submit(conn.speak_and_play, None, text, 0) - conn.tts_queue.put((future, 0)) - - conn.recode_first_last_text(text, 1) - future = conn.executor.submit(conn.speak_and_play, music_path, None, 1) - conn.tts_queue.put((future, 1)) - conn.llm_finish_task = True + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.FIRST, + content_type=ContentType.ACTION, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=text, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.FILE, + content_file=music_path, + ) + ) + conn.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=conn.sentence_id, + sentence_type=SentenceType.LAST, + content_type=ContentType.ACTION, + ) + ) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")