From e8b1156bfdf2d938834417d3868d5603533f87a5 Mon Sep 17 00:00:00 2001 From: MakerZorky <1053714527zhq@gmail.com> Date: Wed, 30 Apr 2025 13:34:00 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=E5=A2=9E=E5=8A=A0DoubaoASR=E7=83=AD?= =?UTF-8?q?=E8=AF=8D=E3=80=81=E6=9B=BF=E6=8D=A2=E8=AF=8D=E9=80=89=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 2 ++ .../core/providers/asr/doubao.py | 18 +++++++++--------- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 21266333..7565179c 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -232,6 +232,8 @@ ASR: appid: 你的火山引擎语音合成服务appid access_token: 你的火山引擎语音合成服务access_token cluster: volcengine_input_common + boosting_table_name: 你的热词文件名称 + correct_table_name: 你的替换词文件名称 output_dir: tmp/ TencentASR: # token申请地址：https://console.cloud.tencent.com/cam/capi diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index 6fba472c..755a19b7 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/main/xiaozhi-server/core/providers/asr/doubao.py @@ -88,6 +88,8 @@ class ASRProvider(ASRProviderBase): self.appid = config.get("appid") self.cluster = config.get("cluster") self.access_token = config.get("access_token") + self.boosting_table_name = config.get("boosting_table_name") + self.correct_table_name = config.get("correct_table_name") self.output_dir = config.get("output_dir") self.delete_audio_file = delete_audio_file @@ -136,7 +138,13 @@ class ASRProvider(ASRProviderBase): "user": { "uid": str(uuid.uuid4()), }, - "request": {"reqid": reqid, "show_utterances": False, "sequence": 1}, + "request": { + "reqid": reqid, + "show_utterances": False, + "sequence": 1, + "boosting_table_name": self.boosting_table_name, + "correct_table_name": self.correct_table_name, + }, "audio": { "format": "raw", "rate": 16000, @@ -232,14 +240,6 @@ class ASRProvider(ASRProviderBase): return pcm_data - @staticmethod - def read_wav_info(data: io.BytesIO = None) -> (int, int, int, int, int): - with io.BytesIO(data) as _f: - wave_fp = wave.open(_f, "rb") - nchannels, sampwidth, framerate, nframes = wave_fp.getparams()[:4] - wave_bytes = wave_fp.readframes(nframes) - return nchannels, sampwidth, framerate, nframes, len(wave_bytes) - @staticmethod def slice_data(data: bytes, chunk_size: int) -> (list, bool): """ From 48f8c6c5b7b09449aa2d2ebb492d012f16877a6c Mon Sep 17 00:00:00 2001 From: goodyhao <865700600@qq.com> Date: Wed, 30 Apr 2025 17:29:27 +0800 Subject: [PATCH 2/4] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0asr,tts=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E4=B8=8A=E6=8A=A5=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AgentChatHistoryController.java | 35 ++++++ .../xiaozhi/modules/agent/dao/AgentDao.java | 15 ++- .../agent/dao/AiAgentChatHistoryDao.java | 16 +++ .../agent/dto/AgentChatHistoryReportDTO.java | 36 ++++++ .../agent/entity/AgentChatHistoryEntity.java | 92 +++++++++++++++ .../service/AgentChatHistoryService.java | 14 +++ .../modules/agent/service/AgentService.java | 10 +- .../biz/AgentChatHistoryBizService.java | 22 ++++ .../impl/AgentChatHistoryBizServiceImpl.java | 83 ++++++++++++++ .../impl/AgentChatHistoryServiceImpl.java | 19 ++++ .../agent/service/impl/AgentServiceImpl.java | 10 +- .../modules/security/config/ShiroConfig.java | 1 + .../resources/db/changelog/202504301634.sql | 18 +++ .../mapper/agent/AiAgentChatHistoryDao.xml | 24 ++++ .../config/manage_api_client.py | 30 +++++ main/xiaozhi-server/core/connection.py | 106 +++++++++++++++++- .../core/handle/asrReportHandle.py | 100 +++++++++++++++++ .../core/handle/receiveAudioHandle.py | 4 + main/xiaozhi-server/core/handle/textHandle.py | 5 + .../core/handle/ttsReportHandle.py | 70 ++++++++++++ main/xiaozhi-server/core/utils/util.py | 2 +- 21 files changed, 706 insertions(+), 6 deletions(-) create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatHistoryDao.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatHistoryService.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/AgentChatHistoryBizService.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatHistoryServiceImpl.java create mode 100644 main/manager-api/src/main/resources/db/changelog/202504301634.sql create mode 100644 main/manager-api/src/main/resources/mapper/agent/AiAgentChatHistoryDao.xml create mode 100644 main/xiaozhi-server/core/handle/asrReportHandle.py create mode 100644 main/xiaozhi-server/core/handle/ttsReportHandle.py diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java new file mode 100644 index 00000000..7abf3d13 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java @@ -0,0 +1,35 @@ +package xiaozhi.modules.agent.controller; + +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.validation.Valid; +import lombok.RequiredArgsConstructor; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import xiaozhi.common.utils.Result; +import xiaozhi.modules.agent.dto.AgentChatHistoryReportDTO; +import xiaozhi.modules.agent.service.biz.AgentChatHistoryBizService; + +@Tag(name = "智能体聊天历史管理") +@RequiredArgsConstructor +@RestController +@RequestMapping("/agent/chat-history") +public class AgentChatHistoryController { + private final AgentChatHistoryBizService agentChatHistoryBizService; + + /** + * 小智服务聊天上报请求 + *

+ * 小智服务聊天上报请求，包含Base64编码的音频数据和相关信息。 + * + * @param request 包含上传文件及相关信息的请求对象 + */ + @Operation(summary = "小智服务聊天上报请求") + @PostMapping("/report") + public Result uploadFile(@Valid @RequestBody AgentChatHistoryReportDTO request) { + Boolean result = agentChatHistoryBizService.report(request); + return new Result().ok(result); + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AgentDao.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AgentDao.java index 8c80d6cc..3b4be452 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AgentDao.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AgentDao.java @@ -3,6 +3,7 @@ package xiaozhi.modules.agent.dao; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; import xiaozhi.common.dao.BaseDao; import xiaozhi.modules.agent.entity.AgentEntity; @@ -15,4 +16,16 @@ public interface AgentDao extends BaseDao { * @return 设备数量 */ Integer getDeviceCountByAgentId(@Param("agentId") String agentId); -} \ No newline at end of file + + /** + * 根据设备MAC地址查询对应设备的默认智能体信息 + * + * @param macAddress 设备MAC地址 + * @return 默认智能体信息 + */ + @Select(" SELECT a.* FROM ai_device d " + + " LEFT JOIN ai_agent a ON d.agent_id = a.id " + + " WHERE d.mac_address = #{macAddress} " + + " ORDER BY d.id DESC LIMIT 1") + AgentEntity getDefaultAgentByMacAddress(@Param("macAddress") String macAddress); +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatHistoryDao.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatHistoryDao.java new file mode 100644 index 00000000..9c07be1d --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatHistoryDao.java @@ -0,0 +1,16 @@ +package xiaozhi.modules.agent.dao; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import org.apache.ibatis.annotations.Mapper; +import xiaozhi.modules.agent.entity.AgentChatHistoryEntity; + +/** + * {@link AgentChatHistoryEntity} 智能体聊天历史记录Dao对象 + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +@Mapper +public interface AiAgentChatHistoryDao extends BaseMapper { +} 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 new file mode 100644 index 00000000..0f79dda6 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java @@ -0,0 +1,36 @@ +package xiaozhi.modules.agent.dto; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +/** + * 小智设备聊天上报请求 + * + * @author Haotian + * @version 1.0, 2025/5/8 + */ +@Data +@Schema(description = "小智设备聊天上报请求") +public class AgentChatHistoryReportDTO { + @Schema(description = "MAC地址", example = "00:11:22:33:44:55") + @NotBlank + private String macAddress; + @Schema(description = "会话ID", example = "79578c31-f1fb-426a-900e-1e934215f05a") + @NotBlank + private String sessionId; + @Schema(description = "排序值（与session_id对应）", example = "1745566378") + @NotNull + private Long sort; + @Schema(description = "消息类型: 1-用户, 2-智能体", example = "1") + @NotNull + private Byte chatType; + @Schema(description = "聊天内容", example = "你好呀") + @NotBlank + private String content; + @Schema(description = "文件数据（Base64编码）", example = "") + private String fileBase64; + @Schema(description = "文件扩展名（如wav、mp3等）", example = "wav") + private String fileExtension; +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java new file mode 100644 index 00000000..3e486f43 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java @@ -0,0 +1,92 @@ +package xiaozhi.modules.agent.entity; + +import java.util.Date; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * 智能体聊天记录表 + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +@TableName(value = "ai_agent_chat_history") +public class AgentChatHistoryEntity { + /** + * 主键ID + */ + @TableId(type = IdType.AUTO) + private Long id; + + /** + * MAC地址 + */ + @TableField(value = "mac_address") + private String macAddress; + + /** + * 智能体id + */ + @TableField(value = "agent_id") + private String agentId; + + /** + * 会话ID + */ + @TableField(value = "session_id") + private String sessionId; + + /** + * 排序值（与session_id对应），使用时间戳，方便排序 + */ + @TableField(value = "sort") + private Long sort; + + /** + * 消息类型: 1-用户, 2-智能体 + */ + @TableField(value = "chat_type") + private Byte chatType; + + /** + * 聊天内容 + */ + @TableField(value = "content") + private String content; + + /** + * 音频base64数据 + */ + @TableField(value = "audio") + private String audio; + + /** + * 音频URL + */ + @TableField(value = "audio_url") + private String audioUrl; + + /** + * 创建时间 + */ + @TableField(value = "created_at") + private Date createdAt; + + /** + * 更新时间 + */ + @TableField(value = "updated_at") + private Date updatedAt; +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatHistoryService.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatHistoryService.java new file mode 100644 index 00000000..e852cae3 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatHistoryService.java @@ -0,0 +1,14 @@ +package xiaozhi.modules.agent.service; + +import com.baomidou.mybatisplus.extension.service.IService; +import xiaozhi.modules.agent.entity.AgentChatHistoryEntity; + +/** + * 智能体聊天记录表处理service + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +public interface AgentChatHistoryService extends IService { +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentService.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentService.java index 8d0b2959..48355a4b 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentService.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentService.java @@ -42,4 +42,12 @@ public interface AgentService extends BaseService { * @return 设备数量 */ Integer getDeviceCountByAgentId(String agentId); -} \ No newline at end of file + + /** + * 根据设备MAC地址查询对应设备的默认智能体信息 + * + * @param macAddress 设备MAC地址 + * @return 默认智能体信息，不存在时返回null + */ + AgentEntity getDefaultAgentByMacAddress(String macAddress); +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/AgentChatHistoryBizService.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/AgentChatHistoryBizService.java new file mode 100644 index 00000000..191b231b --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/AgentChatHistoryBizService.java @@ -0,0 +1,22 @@ +package xiaozhi.modules.agent.service.biz; + +import xiaozhi.modules.agent.dto.AgentChatHistoryReportDTO; + +/** + * 智能体聊天历史业务逻辑层 + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +public interface AgentChatHistoryBizService { + + /** + * 聊天上报方法 + * + * @param agentChatHistoryReportDTO 包含聊天上报所需信息的输入对象 + * 例如：设备MAC地址、文件类型、内容等 + * @return 上传结果，true表示成功，false表示失败 + */ + Boolean report(AgentChatHistoryReportDTO agentChatHistoryReportDTO); +} 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 new file mode 100644 index 00000000..ae99455d --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java @@ -0,0 +1,83 @@ +package xiaozhi.modules.agent.service.biz.impl; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import xiaozhi.modules.agent.dto.AgentChatHistoryReportDTO; +import xiaozhi.modules.agent.entity.AgentEntity; +import xiaozhi.modules.agent.entity.AgentChatHistoryEntity; +import xiaozhi.modules.agent.service.AgentChatHistoryService; +import xiaozhi.modules.agent.service.AgentService; +import xiaozhi.modules.agent.service.biz.AgentChatHistoryBizService; + +import javax.annotation.Nullable; + +/** + * {@link AgentChatHistoryBizService} impl + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +@Service +@Slf4j +@RequiredArgsConstructor +public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizService { + private final AgentService agentService; + private final AgentChatHistoryService agentChatHistoryService; + + /** + * 处理聊天记录上报，包括文件上传和相关信息记录 + * + * @param report 包含聊天上报所需信息的输入对象 + * @return 上传结果，true表示成功，false表示失败 + */ + @Override + @Transactional(rollbackFor = Exception.class) + public Boolean report(AgentChatHistoryReportDTO report) { + final String macAddress = report.getMacAddress(); + final Byte chatType = report.getChatType(); + log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType); + + // 1. 上传音频文件 + final String uploadUrl = this.upload(report); + + // 2. 组装上报数据 + // 2.1 根据设备MAC地址查询对应的默认智能体，判断是否需要上报 + AgentEntity agentEntity = agentService.getDefaultAgentByMacAddress(macAddress); + if (agentEntity == null) { + return false; + } + final String agentId = agentEntity.getId(); + log.info("设备 {} 对应智能体 {} 上报", macAddress, agentEntity.getId()); + + // 2.2 构建聊天记录实体 + final AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder() + .macAddress(macAddress) + .agentId(agentId) + .sessionId(report.getSessionId()) + .sort(report.getSort()) + .chatType(report.getChatType()) + .content(report.getContent()) + .audio(report.getFileBase64()) + .audioUrl(uploadUrl) + .build(); + + // 3. 保存数据 + agentChatHistoryService.save(entity); + return Boolean.TRUE; + } + + /** + * 上传文件 + * + * @param report 上报文件数据 + * @return 上传文件url + */ + @Nullable + private String upload(AgentChatHistoryReportDTO report) { + // TODO(haotian): 2025/4/30 根据需要自定义完成上传生成url即可 + return null; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatHistoryServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatHistoryServiceImpl.java new file mode 100644 index 00000000..95293a3f --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatHistoryServiceImpl.java @@ -0,0 +1,19 @@ +package xiaozhi.modules.agent.service.impl; + +import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import org.springframework.stereotype.Service; +import xiaozhi.modules.agent.dao.AiAgentChatHistoryDao; +import xiaozhi.modules.agent.entity.AgentChatHistoryEntity; +import xiaozhi.modules.agent.service.AgentChatHistoryService; + +/** + * 智能体聊天记录表处理service {@link AgentChatHistoryService} impl + * + * @author Goody + * @version 1.0, 2025/4/30 + * @since 1.0.0 + */ +@Service +public class AgentChatHistoryServiceImpl extends ServiceImpl implements AgentChatHistoryService { + +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentServiceImpl.java index ff284ef0..4c529882 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentServiceImpl.java @@ -130,4 +130,12 @@ public class AgentServiceImpl extends BaseServiceImpl imp return deviceCount != null ? deviceCount : 0; } -} \ No newline at end of file + + @Override + public AgentEntity getDefaultAgentByMacAddress(String macAddress) { + if (StringUtils.isEmpty(macAddress)) { + return null; + } + return agentDao.getDefaultAgentByMacAddress(macAddress); + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/security/config/ShiroConfig.java b/main/manager-api/src/main/java/xiaozhi/modules/security/config/ShiroConfig.java index 62e192b7..b462b2c4 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/security/config/ShiroConfig.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/security/config/ShiroConfig.java @@ -84,6 +84,7 @@ public class ShiroConfig { filterMap.put("/user/register", "anon"); // 将config路径使用server服务过滤器 filterMap.put("/config/**", "server"); + filterMap.put("/agent/chat-history/report", "server"); filterMap.put("/**", "oauth2"); shiroFilter.setFilterChainDefinitionMap(filterMap); diff --git a/main/manager-api/src/main/resources/db/changelog/202504301634.sql b/main/manager-api/src/main/resources/db/changelog/202504301634.sql new file mode 100644 index 00000000..6a86a417 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202504301634.sql @@ -0,0 +1,18 @@ +-- 初始化智能体聊天记录 +DROP TABLE IF EXISTS ai_agent_chat_history; +CREATE TABLE ai_agent_chat_history +( + id BIGINT AUTO_INCREMENT COMMENT '主键ID' + PRIMARY KEY, + mac_address VARCHAR(50) COMMENT 'MAC地址', + agent_id BIGINT DEFAULT 0 COMMENT '智能体id', + session_id VARCHAR(50) COMMENT '会话ID', + sort BIGINT COMMENT '排序值（与session_id对应），使用时间戳，方便排序', + chat_type TINYINT(3) COMMENT '消息类型: 1-用户, 2-智能体', + content VARCHAR(1024) COMMENT '聊天内容', + audio text COMMENT '音频base64数据', + audio_url VARCHAR(256) COMMENT '音频URL', + created_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL COMMENT '创建时间', + updated_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间', + INDEX idx_mac_session (mac_address, sort) +) COMMENT '智能体聊天记录表'; diff --git a/main/manager-api/src/main/resources/mapper/agent/AiAgentChatHistoryDao.xml b/main/manager-api/src/main/resources/mapper/agent/AiAgentChatHistoryDao.xml new file mode 100644 index 00000000..2d87d8cf --- /dev/null +++ b/main/manager-api/src/main/resources/mapper/agent/AiAgentChatHistoryDao.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + + + + id, mac_address, agent_id, session_id, sort, chat_type, content, audio, audio_url, + created_at, updated_at + + diff --git a/main/xiaozhi-server/config/manage_api_client.py b/main/xiaozhi-server/config/manage_api_client.py index 770cbf56..ec8fd30a 100644 --- a/main/xiaozhi-server/config/manage_api_client.py +++ b/main/xiaozhi-server/config/manage_api_client.py @@ -1,3 +1,4 @@ +import base64 import os import time from typing import Optional, Dict @@ -145,6 +146,35 @@ def get_agent_models( }, ) +async def report(mac_address: str, + session_id: str, + sort: int, + chat_type: int, + content: str, + audio, + file_extension: str = "wav", + need_report: bool = None, + report_type: int = None, + reported: bool = None) -> Optional[Dict]: + """带熔断的业务方法示例""" + if not content or not ManageApiClient._instance: + return None + return await ManageApiClient._instance._execute_request( + "POST", + f"/agent/chat-history/report", + json = { + "macAddress": mac_address, + "sessionId": session_id, + "sort": sort, + "chatType": chat_type, + "content": content, + "fileBase64": base64.b64encode(audio).decode('utf-8'), + "fileExtension": file_extension, + "needReport": need_report, + "reportType": report_type, + "reported": reported + } + ) def init_service(config): ManageApiClient(config) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index fbd3569e..1d1caf04 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -30,6 +30,7 @@ from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api from config.manage_api_client import DeviceNotFoundException, DeviceBindException from core.utils.output_counter import add_device_output +from core.handle.ttsReportHandle import enqueue_tts_report TAG = __name__ @@ -54,6 +55,7 @@ class ConnectionHandler: self.websocket = None self.headers = None + self.device_id = None self.client_ip = None self.client_ip_info = {} self.session_id = None @@ -72,6 +74,13 @@ class ConnectionHandler: self.audio_play_queue = queue.Queue() self.executor = ThreadPoolExecutor(max_workers=10) + # 上报线程标志 + self.session_open_time = time.time() + self.tts_report_queue = queue.Queue() + self.asr_report_queue = queue.Queue() + self.asr_report_thread = None + self.tts_report_thread = None + # 依赖的组件 self.vad = _vad self.asr = _asr @@ -153,6 +162,7 @@ class ConnectionHandler: # 认证通过,继续处理 self.websocket = ws + self.device_id = self.headers.get("device-id", None) self.session_id = str(uuid.uuid4()) # 启动超时检查任务 @@ -301,6 +311,26 @@ class ConnectionHandler: self._initialize_memory() """加载意图识别""" self._initialize_intent() + """初始化上报线程""" + self._init_report_threads() + + def _init_report_threads(self): + """初始化ASR和TTS上报线程""" + if self.asr_report_thread is None or not self.asr_report_thread.is_alive(): + self.asr_report_thread = threading.Thread( + target=self._asr_report_worker, + daemon=True + ) + self.asr_report_thread.start() + self.logger.bind(tag=TAG).info("ASR上报线程已启动") + + if self.tts_report_thread is None or not self.tts_report_thread.is_alive(): + self.tts_report_thread = threading.Thread( + target=self._tts_report_worker, + daemon=True + ) + self.tts_report_thread.start() + self.logger.bind(tag=TAG).info("TTS上报线程已启动") def _initialize_private_config(self): read_config_from_api = self.config.get("read_config_from_api", False) @@ -427,8 +457,7 @@ class ConnectionHandler: def _initialize_memory(self): """初始化记忆模块""" - device_id = self.headers.get("device-id", None) - self.memory.init_memory(device_id, self.llm) + self.memory.init_memory(self.device_id, self.llm) def _initialize_intent(self): if ( @@ -862,6 +891,9 @@ class ConnectionHandler: f"TTS生成：文件路径: {tts_file}" ) if os.path.exists(tts_file): + # 在这里上报TTS数据（使用文件路径） + enqueue_tts_report(self, text, tts_file) + opus_datas, duration = self.tts.audio_to_opus_data(tts_file) else: self.logger.bind(tag=TAG).error( @@ -918,6 +950,72 @@ class ConnectionHandler: f"audio_play_priority priority_thread: {text} {e}" ) + def _asr_report_worker(self): + """ASR上报工作线程""" + # 提前导入避免循环引用问题 + from core.handle.asrReportHandle import report_asr + + while not self.stop_event.is_set(): + try: + # 从队列获取数据，设置超时以便定期检查停止事件 + item = self.asr_report_queue.get(timeout=1) + if item is None: # 检测毒丸对象 + break + + text, file_path = item + + try: + # 执行上报（传入文件路径） + await_result = report_asr(self, text, file_path) + + # 使用asyncio.run_coroutine_threadsafe执行异步操作 + future = asyncio.run_coroutine_threadsafe(await_result, self.loop) + future.result() + except Exception as e: + self.logger.bind(tag=TAG).error(f"ASR上报线程异常: {e}") + finally: + # 标记任务完成 + self.asr_report_queue.task_done() + except queue.Empty: + continue + except Exception as e: + self.logger.bind(tag=TAG).error(f"ASR上报工作线程异常: {e}") + + self.logger.bind(tag=TAG).info("ASR上报线程已退出") + + def _tts_report_worker(self): + """TTS上报工作线程""" + # 提前导入避免循环引用问题 + from core.handle.ttsReportHandle import report_tts + + while not self.stop_event.is_set(): + try: + # 从队列获取数据，设置超时以便定期检查停止事件 + item = self.tts_report_queue.get(timeout=1) + if item is None: # 检测毒丸对象 + break + + text, audio_data = item + + try: + # 执行上报（传入二进制数据） + await_result = report_tts(self, text, audio_data) + + # 使用asyncio.run_coroutine_threadsafe执行异步操作 + future = asyncio.run_coroutine_threadsafe(await_result, self.loop) + future.result() + except Exception as e: + self.logger.bind(tag=TAG).error(f"TTS上报线程异常: {e}") + finally: + # 标记任务完成 + self.tts_report_queue.task_done() + except queue.Empty: + continue + except Exception as e: + self.logger.bind(tag=TAG).error(f"TTS上报工作线程异常: {e}") + + self.logger.bind(tag=TAG).info("TTS上报线程已退出") + def speak_and_play(self, text, text_index=0): if text is None or len(text) <= 0: self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{text}") @@ -963,6 +1061,10 @@ class ConnectionHandler: self.executor.shutdown(wait=False, cancel_futures=True) self.executor = None + # 添加毒丸对象到上报队列确保线程退出 + self.asr_report_queue.put(None) + self.tts_report_queue.put(None) + # 清空任务队列 self.clear_queues() diff --git a/main/xiaozhi-server/core/handle/asrReportHandle.py b/main/xiaozhi-server/core/handle/asrReportHandle.py new file mode 100644 index 00000000..70601804 --- /dev/null +++ b/main/xiaozhi-server/core/handle/asrReportHandle.py @@ -0,0 +1,100 @@ +""" +ASR上报功能已集成到ConnectionHandler类中。 + +上报功能包括： +1. 每个连接对象拥有自己的上报队列和处理线程 +2. 上报线程的生命周期与连接对象绑定 +3. 使用ConnectionHandler.enqueue_asr_report方法进行上报 + +具体实现请参考core/connection.py中的相关代码。 +""" + +import os +from config.logger import setup_logging +from config.manage_api_client import report + +TAG = __name__ +logger = setup_logging() + +async def report_asr(conn, text, file_path): + """执行ASR上报操作 + + Args: + conn: 连接对象 + text: 识别文本 + file_path: 音频文件路径（可以为None或空字符串，表示纯文本上报） + """ + audio_data = None + try: + # 处理无音频的纯文本上报 + if not file_path or not os.path.exists(file_path): + # 纯文本上报时使用空音频数据 + result = await report( + mac_address=conn.device_id, + session_id=conn.session_id, + sort=int(conn.session_open_time), + chat_type=1, # ASR类型为1 + content=text, + audio=b'', # 空音频数据 + file_extension="wav" + ) + logger.bind(tag=TAG).info(f"纯文本上报成功: {conn.device_id}, {conn.session_id}") + else: + # 读取文件为二进制数据 + with open(file_path, 'rb') as f: + audio_data = f.read() + + # 正常ASR上报（带音频） + result = await report( + mac_address=conn.device_id, + session_id=conn.session_id, + sort=int(conn.session_open_time), + chat_type=1, # ASR类型为1 + content=text, + audio=audio_data, + file_extension="wav" + ) + logger.bind(tag=TAG).info(f"ASR上报成功: {conn.device_id}, {conn.session_id}，文件: {file_path}") + + return result + except Exception as e: + logger.bind(tag=TAG).error(f"ASR上报失败: {e}") + return None + finally: + # 清理资源 + if file_path and os.path.exists(file_path): + try: + os.remove(file_path) + logger.bind(tag=TAG).debug(f"ASR上报后删除文件: {file_path}") + except Exception as e: + logger.bind(tag=TAG).error(f"ASR上报后删除文件失败: {e}") + + # 手动清理audio_data + if audio_data: + del audio_data + +def enqueue_asr_report(conn, text, audio): + """将ASR数据加入上报队列 + + Args: + conn: 连接对象 + text: 识别文本 + audio: 音频数据（可以为空列表，表示纯文本上报） + """ + try: + if not audio or len(audio) == 0: + # 纯文本上报，不需要保存文件 + file_path = None + else: + # 保存音频数据到文件 + file_path = conn.asr.save_audio_to_file(audio, conn.session_id) + + # 使用连接对象的队列，传入文件路径 + conn.asr_report_queue.put((text, file_path)) + + if not audio or len(audio) == 0: + logger.bind(tag=TAG).info(f"纯文本数据已加入上报队列: {conn.device_id}, {text[:20] if text else ''}...") + else: + logger.bind(tag=TAG).info(f"ASR数据已加入上报队列: {conn.device_id}, 文件: {file_path}") + except Exception as e: + logger.bind(tag=TAG).error(f"加入ASR上报队列失败: {e}") diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 485544f7..2d8b0eef 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -4,6 +4,7 @@ from core.utils.util import remove_punctuation_and_length 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.asrReportHandle import enqueue_asr_report TAG = __name__ logger = setup_logging() @@ -40,6 +41,9 @@ async def handleAudioMessage(conn, audio): logger.bind(tag=TAG).info(f"识别文本: {text}") text_len, _ = remove_punctuation_and_length(text) if text_len > 0: + # 使用自定义模块进行上报 + enqueue_asr_report(conn, text, conn.asr_audio) + await startToChat(conn, text) else: conn.asr_server_receive = True diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index a26f5bd9..fa81eb76 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -6,6 +6,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.iotHandle import handleIotDescriptors, handleIotStatus +from core.handle.asrReportHandle import enqueue_asr_report import asyncio TAG = __name__ @@ -54,8 +55,12 @@ async def handleTextMessage(conn, message): await send_stt_message(conn, text) await send_tts_message(conn, "stop", None) elif is_wakeup_words: + # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） + enqueue_asr_report(conn, "嘿，你好呀", []) await startToChat(conn, "嘿，你好呀") else: + # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） + enqueue_asr_report(conn, text, []) # 否则需要LLM对文字内容进行答复 await startToChat(conn, text) elif msg_json["type"] == "iot": diff --git a/main/xiaozhi-server/core/handle/ttsReportHandle.py b/main/xiaozhi-server/core/handle/ttsReportHandle.py new file mode 100644 index 00000000..ed3a6a76 --- /dev/null +++ b/main/xiaozhi-server/core/handle/ttsReportHandle.py @@ -0,0 +1,70 @@ +""" +TTS上报功能已集成到ConnectionHandler类中。 + +上报功能包括： +1. 每个连接对象拥有自己的上报队列和处理线程 +2. 上报线程的生命周期与连接对象绑定 +3. 使用ConnectionHandler.enqueue_tts_report方法进行上报 + +具体实现请参考core/connection.py中的相关代码。 +""" + +import os +from config.logger import setup_logging +from config.manage_api_client import report + +TAG = __name__ +logger = setup_logging() + +async def report_tts(conn, text, audio_data): + """执行TTS上报操作 + + Args: + conn: 连接对象 + text: 合成文本 + audio_data: 音频二进制数据 + """ + try: + # 执行上报 + result = await report( + mac_address=conn.device_id, + session_id=conn.session_id, + sort=int(conn.session_open_time), + chat_type=2, # TTS类型为2 + content=text, + audio=audio_data, + file_extension="wav" + ) + logger.bind(tag=TAG).info(f"TTS上报成功: {conn.device_id}, {conn.session_id}, 数据大小: {len(audio_data)} 字节") + return result + except Exception as e: + logger.bind(tag=TAG).error(f"TTS上报失败: {e}") + return None + finally: + # 手动清理audio_data引用，帮助垃圾回收 + del audio_data + +def enqueue_tts_report(conn, text, file_path): + """将TTS数据加入上报队列 + + Args: + conn: 连接对象 + text: 合成文本 + file_path: TTS音频文件路径 + """ + try: + # 检查文件是否存在 + if not file_path or not os.path.exists(file_path): + logger.bind(tag=TAG).error(f"加入TTS上报队列失败: 文件不存在 {file_path}") + return + + # 立即读取文件为二进制数据，因为外部会删除文件 + with open(file_path, 'rb') as f: + audio_data = f.read() + + # 使用连接对象的队列，传入文本和二进制数据而非文件路径 + conn.tts_report_queue.put((text, audio_data)) + + logger.bind(tag=TAG).info(f"TTS数据已加入上报队列: {conn.device_id}, 文件大小: {len(audio_data)} 字节") + except Exception as e: + logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {e}, 文件: {file_path}") diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 1cc9742b..a1956cee 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -484,4 +484,4 @@ def analyze_emotion(text): if emotion in top_emotions: return emotion - return top_emotions[0] # 如果都不在优先级列表里，返回第一个 \ No newline at end of file + return top_emotions[0] # 如果都不在优先级列表里，返回第一个 From 54692859fde8782962d84b87a25a1a91aad3a8a6 Mon Sep 17 00:00:00 2001 From: MakerZorky <152245790+MakerZorky@users.noreply.github.com> Date: Wed, 30 Apr 2025 19:23:44 +0800 Subject: [PATCH 3/4] =?UTF-8?q?Update:config.yaml=E7=83=AD=E8=AF=8D?= =?UTF-8?q?=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 7565179c..6948aebd 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -232,8 +232,9 @@ ASR: appid: 你的火山引擎语音合成服务appid access_token: 你的火山引擎语音合成服务access_token cluster: volcengine_input_common - boosting_table_name: 你的热词文件名称 - correct_table_name: 你的替换词文件名称 + # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738 + boosting_table_name: （选填）你的热词文件名称 + correct_table_name: （选填）你的替换词文件名称 output_dir: tmp/ TencentASR: # token申请地址：https://console.cloud.tencent.com/cam/capi From f0e353cc1a9b0a6394942ebc77add67ae74f96ef Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Fri, 2 May 2025 00:07:36 +0800 Subject: [PATCH 4/4] =?UTF-8?q?update:=E4=BC=98=E5=8C=96=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E4=B8=8A=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + .../common/service/impl/BaseServiceImpl.java | 2 - .../common/service/impl/CrudServiceImpl.java | 1 - .../AgentChatHistoryController.java | 9 +- .../agent/dao/AiAgentChatAudioDao.java | 18 +++ .../agent/dto/AgentChatHistoryReportDTO.java | 11 +- .../agent/entity/AgentChatAudioEntity.java | 29 +++++ .../agent/entity/AgentChatHistoryEntity.java | 51 ++++---- .../agent/service/AgentChatAudioService.java | 22 ++++ .../impl/AgentChatHistoryBizServiceImpl.java | 50 ++++---- .../impl/AgentChatAudioServiceImpl.java | 28 +++++ .../config/controller/ConfigController.java | 4 +- .../service/impl/ConfigServiceImpl.java | 1 - .../resources/db/changelog/202504301634.sql | 18 --- .../resources/db/changelog/202505012207.sql | 24 ++++ .../db/changelog/db.changelog-master.yaml | 9 +- .../config/manage_api_client.py | 66 ++++++----- main/xiaozhi-server/core/connection.py | 112 ++++++------------ .../core/handle/asrReportHandle.py | 100 ---------------- .../core/handle/receiveAudioHandle.py | 5 +- main/xiaozhi-server/core/handle/textHandle.py | 23 ++-- .../core/handle/ttsReportHandle.py | 51 ++++---- 22 files changed, 295 insertions(+), 340 deletions(-) create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatAudioDao.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatAudioEntity.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java delete mode 100644 main/manager-api/src/main/resources/db/changelog/202504301634.sql create mode 100644 main/manager-api/src/main/resources/db/changelog/202505012207.sql delete mode 100644 main/xiaozhi-server/core/handle/asrReportHandle.py diff --git a/.gitignore b/.gitignore index e1ee3d0c..f4efaeaa 100644 --- a/.gitignore +++ b/.gitignore @@ -165,3 +165,4 @@ main/manager-api/.vscode main/manager-web/.webpack_cache/ main/xiaozhi-server/mysql uploadfile +.vscode diff --git a/main/manager-api/src/main/java/xiaozhi/common/service/impl/BaseServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/common/service/impl/BaseServiceImpl.java index 8ebb4cd1..9d2ba5d5 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/service/impl/BaseServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/common/service/impl/BaseServiceImpl.java @@ -128,12 +128,10 @@ public abstract class BaseServiceImpl, T> implements Bas return SqlHelper.retBool(result); } - @SuppressWarnings("unchecked") protected Class currentMapperClass() { return (Class) ReflectionKit.getSuperClassGenericType(this.getClass(), BaseServiceImpl.class, 0); } - @SuppressWarnings("unchecked") @Override public Class currentModelClass() { return (Class) ReflectionKit.getSuperClassGenericType(this.getClass(), BaseServiceImpl.class, 1); diff --git a/main/manager-api/src/main/java/xiaozhi/common/service/impl/CrudServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/common/service/impl/CrudServiceImpl.java index 524a8cb2..9a1bf39e 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/service/impl/CrudServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/common/service/impl/CrudServiceImpl.java @@ -24,7 +24,6 @@ import xiaozhi.common.utils.ConvertUtils; public abstract class CrudServiceImpl, T, D> extends BaseServiceImpl implements CrudService { - @SuppressWarnings("unchecked") protected Class currentDtoClass() { return (Class) ReflectionKit.getSuperClassGenericType(getClass(), CrudServiceImpl.class, 2); } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java index 7abf3d13..9d6542cc 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentChatHistoryController.java @@ -1,13 +1,14 @@ package xiaozhi.modules.agent.controller; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; -import org.springframework.web.bind.annotation.PostMapping; -import org.springframework.web.bind.annotation.RequestBody; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; import xiaozhi.common.utils.Result; import xiaozhi.modules.agent.dto.AgentChatHistoryReportDTO; import xiaozhi.modules.agent.service.biz.AgentChatHistoryBizService; diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatAudioDao.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatAudioDao.java new file mode 100644 index 00000000..355b25ac --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dao/AiAgentChatAudioDao.java @@ -0,0 +1,18 @@ +package xiaozhi.modules.agent.dao; + +import org.apache.ibatis.annotations.Mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; + +import xiaozhi.modules.agent.entity.AgentChatAudioEntity; + +/** + * {@link AgentChatAudioEntity} 智能体聊天音频数据Dao对象 + * + * @author Goody + * @version 1.0, 2025/5/8 + * @since 1.0.0 + */ +@Mapper +public interface AiAgentChatAudioDao extends BaseMapper { +} \ No newline at end of file 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 0f79dda6..8227260d 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 @@ -20,17 +20,12 @@ public class AgentChatHistoryReportDTO { @Schema(description = "会话ID", example = "79578c31-f1fb-426a-900e-1e934215f05a") @NotBlank private String sessionId; - @Schema(description = "排序值（与session_id对应）", example = "1745566378") - @NotNull - private Long sort; @Schema(description = "消息类型: 1-用户, 2-智能体", example = "1") @NotNull private Byte chatType; @Schema(description = "聊天内容", example = "你好呀") @NotBlank private String content; - @Schema(description = "文件数据（Base64编码）", example = "") - private String fileBase64; - @Schema(description = "文件扩展名（如wav、mp3等）", example = "wav") - private String fileExtension; -} + @Schema(description = "文件数据（opus编码）", example = "") + private String opusDataBase64; +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatAudioEntity.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatAudioEntity.java new file mode 100644 index 00000000..ac30d9d4 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatAudioEntity.java @@ -0,0 +1,29 @@ +package xiaozhi.modules.agent.entity; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; + +import lombok.Data; + +/** + * 智能体聊天音频数据表 + * + * @author Goody + * @version 1.0, 2025/5/8 + * @since 1.0.0 + */ +@Data +@TableName("ai_agent_chat_audio") +public class AgentChatAudioEntity { + /** + * 主键ID + */ + @TableId(type = IdType.ASSIGN_UUID) + private String id; + + /** + * 音频opus数据 + */ + private byte[] audio; +} \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java index 3e486f43..96e82885 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/entity/AgentChatHistoryEntity.java @@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableField; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; + import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -25,68 +26,62 @@ import lombok.NoArgsConstructor; @TableName(value = "ai_agent_chat_history") public class AgentChatHistoryEntity { /** - * 主键ID - */ + * 主键ID + */ @TableId(type = IdType.AUTO) private Long id; /** - * MAC地址 - */ + * MAC地址 + */ @TableField(value = "mac_address") private String macAddress; /** - * 智能体id - */ + * 智能体id + */ @TableField(value = "agent_id") private String agentId; /** - * 会话ID - */ + * 会话ID + */ @TableField(value = "session_id") private String sessionId; /** - * 排序值（与session_id对应），使用时间戳，方便排序 - */ - @TableField(value = "sort") - private Long sort; - - /** - * 消息类型: 1-用户, 2-智能体 - */ + * 消息类型: 1-用户, 2-智能体 + */ @TableField(value = "chat_type") private Byte chatType; /** - * 聊天内容 - */ + * 聊天内容 + */ @TableField(value = "content") private String content; /** - * 音频base64数据 - */ - @TableField(value = "audio") - private String audio; + * 音频base64数据 + */ + @TableField(value = "audio_id") + private String audioId; /** - * 音频URL - */ + * 音频URL + */ @TableField(value = "audio_url") private String audioUrl; /** - * 创建时间 - */ + * 创建时间 + */ @TableField(value = "created_at") private Date createdAt; /** - * 更新时间 - */ + * 更新时间 + */ @TableField(value = "updated_at") private Date updatedAt; } 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 new file mode 100644 index 00000000..fdd0e1f6 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java @@ -0,0 +1,22 @@ +package xiaozhi.modules.agent.service; + +import com.baomidou.mybatisplus.extension.service.IService; + +import xiaozhi.modules.agent.entity.AgentChatAudioEntity; + +/** + * 智能体聊天音频数据表处理service + * + * @author Goody + * @version 1.0, 2025/5/8 + * @since 1.0.0 + */ +public interface AgentChatAudioService extends IService { + /** + * 保存音频数据 + * + * @param audioData 音频数据 + * @return 音频ID + */ + String saveAudio(byte[] audioData); +} \ 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 ae99455d..edd96256 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 @@ -1,18 +1,18 @@ package xiaozhi.modules.agent.service.biz.impl; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; import xiaozhi.modules.agent.dto.AgentChatHistoryReportDTO; -import xiaozhi.modules.agent.entity.AgentEntity; import xiaozhi.modules.agent.entity.AgentChatHistoryEntity; +import xiaozhi.modules.agent.entity.AgentEntity; +import xiaozhi.modules.agent.service.AgentChatAudioService; import xiaozhi.modules.agent.service.AgentChatHistoryService; import xiaozhi.modules.agent.service.AgentService; import xiaozhi.modules.agent.service.biz.AgentChatHistoryBizService; -import javax.annotation.Nullable; - /** * {@link AgentChatHistoryBizService} impl * @@ -26,6 +26,7 @@ import javax.annotation.Nullable; public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizService { private final AgentService agentService; private final AgentChatHistoryService agentChatHistoryService; + private final AgentChatAudioService agentChatAudioService; /** * 处理聊天记录上报，包括文件上传和相关信息记录 @@ -36,12 +37,23 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic @Override @Transactional(rollbackFor = Exception.class) public Boolean report(AgentChatHistoryReportDTO report) { - final String macAddress = report.getMacAddress(); - final Byte chatType = report.getChatType(); + String macAddress = report.getMacAddress(); + Byte chatType = report.getChatType(); log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType); - // 1. 上传音频文件 - final String uploadUrl = this.upload(report); + // 1. base64解码report.getOpusDataBase64(),存入ai_agent_chat_audio表 + String audioId = null; + if (report.getOpusDataBase64() != null && !report.getOpusDataBase64().isEmpty()) { + try { + // TODO: 需要考虑保留什么格式的音频数据，比如是opus还是wave + // byte[] audioData = Base64.getDecoder().decode(report.getOpusDataBase64()); + // audioId = agentChatAudioService.saveAudio(audioData); + // log.info("音频数据保存成功，audioId={}", audioId); + } catch (Exception e) { + log.error("音频数据保存失败", e); + return false; + } + } // 2. 组装上报数据 // 2.1 根据设备MAC地址查询对应的默认智能体，判断是否需要上报 @@ -49,35 +61,21 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic if (agentEntity == null) { return false; } - final String agentId = agentEntity.getId(); + String agentId = agentEntity.getId(); log.info("设备 {} 对应智能体 {} 上报", macAddress, agentEntity.getId()); // 2.2 构建聊天记录实体 - final AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder() + AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder() .macAddress(macAddress) .agentId(agentId) .sessionId(report.getSessionId()) - .sort(report.getSort()) .chatType(report.getChatType()) .content(report.getContent()) - .audio(report.getFileBase64()) - .audioUrl(uploadUrl) + .audioId(audioId) .build(); // 3. 保存数据 agentChatHistoryService.save(entity); return Boolean.TRUE; } - - /** - * 上传文件 - * - * @param report 上报文件数据 - * @return 上传文件url - */ - @Nullable - private String upload(AgentChatHistoryReportDTO report) { - // TODO(haotian): 2025/4/30 根据需要自定义完成上传生成url即可 - return null; - } } 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 new file mode 100644 index 00000000..c222abe4 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java @@ -0,0 +1,28 @@ +package xiaozhi.modules.agent.service.impl; + +import org.springframework.stereotype.Service; + +import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; + +import xiaozhi.modules.agent.dao.AiAgentChatAudioDao; +import xiaozhi.modules.agent.entity.AgentChatAudioEntity; +import xiaozhi.modules.agent.service.AgentChatAudioService; + +/** + * 智能体聊天音频数据表处理service {@link AgentChatAudioService} impl + * + * @author Goody + * @version 1.0, 2025/5/8 + * @since 1.0.0 + */ +@Service +public class AgentChatAudioServiceImpl extends ServiceImpl + implements AgentChatAudioService { + @Override + public String saveAudio(byte[] audioData) { + AgentChatAudioEntity entity = new AgentChatAudioEntity(); + entity.setAudio(audioData); + save(entity); + return entity.getId(); + } +} \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/modules/config/controller/ConfigController.java b/main/manager-api/src/main/java/xiaozhi/modules/config/controller/ConfigController.java index 13106869..f83efb3a 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/config/controller/ConfigController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/config/controller/ConfigController.java @@ -1,6 +1,5 @@ package xiaozhi.modules.config.controller; -import jakarta.validation.Valid; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -8,12 +7,12 @@ import org.springframework.web.bind.annotation.RestController; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.validation.Valid; import lombok.AllArgsConstructor; import xiaozhi.common.utils.Result; import xiaozhi.common.validator.ValidatorUtils; import xiaozhi.modules.config.dto.AgentModelsDTO; import xiaozhi.modules.config.service.ConfigService; -import xiaozhi.modules.sys.service.SysParamsService; /** * xiaozhi-server 配置获取 @@ -26,7 +25,6 @@ import xiaozhi.modules.sys.service.SysParamsService; @AllArgsConstructor public class ConfigController { private final ConfigService configService; - private final SysParamsService sysParamsService; @PostMapping("server-base") @Operation(summary = "获取配置") diff --git a/main/manager-api/src/main/java/xiaozhi/modules/config/service/impl/ConfigServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/config/service/impl/ConfigServiceImpl.java index e4914029..f0ed4bf1 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/config/service/impl/ConfigServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/config/service/impl/ConfigServiceImpl.java @@ -153,7 +153,6 @@ public class ConfigServiceImpl implements ConfigService { * @param paramsList 系统参数列表 * @return 配置信息 */ - @SuppressWarnings("unchecked") private Object buildConfig(Map config) { // 查询所有系统参数 diff --git a/main/manager-api/src/main/resources/db/changelog/202504301634.sql b/main/manager-api/src/main/resources/db/changelog/202504301634.sql deleted file mode 100644 index 6a86a417..00000000 --- a/main/manager-api/src/main/resources/db/changelog/202504301634.sql +++ /dev/null @@ -1,18 +0,0 @@ --- 初始化智能体聊天记录 -DROP TABLE IF EXISTS ai_agent_chat_history; -CREATE TABLE ai_agent_chat_history -( - id BIGINT AUTO_INCREMENT COMMENT '主键ID' - PRIMARY KEY, - mac_address VARCHAR(50) COMMENT 'MAC地址', - agent_id BIGINT DEFAULT 0 COMMENT '智能体id', - session_id VARCHAR(50) COMMENT '会话ID', - sort BIGINT COMMENT '排序值（与session_id对应），使用时间戳，方便排序', - chat_type TINYINT(3) COMMENT '消息类型: 1-用户, 2-智能体', - content VARCHAR(1024) COMMENT '聊天内容', - audio text COMMENT '音频base64数据', - audio_url VARCHAR(256) COMMENT '音频URL', - created_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL COMMENT '创建时间', - updated_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间', - INDEX idx_mac_session (mac_address, sort) -) COMMENT '智能体聊天记录表'; diff --git a/main/manager-api/src/main/resources/db/changelog/202505012207.sql b/main/manager-api/src/main/resources/db/changelog/202505012207.sql new file mode 100644 index 00000000..5003edc7 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202505012207.sql @@ -0,0 +1,24 @@ +-- 初始化智能体聊天记录 +DROP TABLE IF EXISTS ai_chat_history; +DROP TABLE IF EXISTS ai_agent_chat_history; +CREATE TABLE ai_agent_chat_history +( + id BIGINT AUTO_INCREMENT COMMENT '主键ID' PRIMARY KEY, + mac_address VARCHAR(50) COMMENT 'MAC地址', + agent_id VARCHAR(32) COMMENT '智能体id', + session_id VARCHAR(50) COMMENT '会话ID', + chat_type TINYINT(3) COMMENT '消息类型: 1-用户, 2-智能体', + content VARCHAR(1024) COMMENT '聊天内容', + audio_id VARCHAR(32) COMMENT '音频ID', + created_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL COMMENT '创建时间', + updated_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) NOT NULL ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间', + INDEX idx_ai_agent_chat_history_mac (mac_address), + INDEX idx_ai_agent_chat_history_agent_id (agent_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '智能体聊天记录表'; + +DROP TABLE IF EXISTS ai_agent_chat_audio; +CREATE TABLE ai_agent_chat_audio +( + id VARCHAR(32) COMMENT '主键ID' PRIMARY KEY, + audio LONGBLOB COMMENT '音频opus数据' +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '智能体聊天音频数据表'; \ No newline at end of file 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 ffe96a9b..aaef45bc 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 @@ -92,4 +92,11 @@ databaseChangeLog: changes: - sqlFile: encoding: utf8 - path: classpath:db/changelog/202504301341.sql \ No newline at end of file + path: classpath:db/changelog/202504301341.sql + - changeSet: + id: 202505012207 + author: Goody + changes: + - sqlFile: + encoding: utf8 + path: classpath:db/changelog/202505012207.sql \ No newline at end of file diff --git a/main/xiaozhi-server/config/manage_api_client.py b/main/xiaozhi-server/config/manage_api_client.py index ec8fd30a..7de4c8fe 100644 --- a/main/xiaozhi-server/config/manage_api_client.py +++ b/main/xiaozhi-server/config/manage_api_client.py @@ -1,6 +1,6 @@ -import base64 import os import time +import base64 from typing import Optional, Dict import httpx @@ -54,7 +54,7 @@ class ManageApiClient: headers={ "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})", "Accept": "application/json", - "Authorization": "Bearer " + cls._secret + "Authorization": "Bearer " + cls._secret, }, timeout=cls.config.get("timeout", 30), # 默认超时时间30秒 ) @@ -127,9 +127,7 @@ class ManageApiClient: def get_server_config() -> Optional[Dict]: """获取服务器基础配置""" - return ManageApiClient._instance._execute_request( - "POST", "/config/server-base" - ) + return ManageApiClient._instance._execute_request("POST", "/config/server-base") def get_agent_models( @@ -146,35 +144,41 @@ def get_agent_models( }, ) -async def report(mac_address: str, - session_id: str, - sort: int, - chat_type: int, - content: str, - audio, - file_extension: str = "wav", - need_report: bool = None, - report_type: int = None, - reported: bool = None) -> Optional[Dict]: + +def report( + mac_address: str, session_id: str, chat_type: int, content: str, opus_data +) -> Optional[Dict]: """带熔断的业务方法示例""" if not content or not ManageApiClient._instance: return None - return await ManageApiClient._instance._execute_request( - "POST", - f"/agent/chat-history/report", - json = { - "macAddress": mac_address, - "sessionId": session_id, - "sort": sort, - "chatType": chat_type, - "content": content, - "fileBase64": base64.b64encode(audio).decode('utf-8'), - "fileExtension": file_extension, - "needReport": need_report, - "reportType": report_type, - "reported": reported - } - ) + try: + # 处理opus_data为列表的情况 + if isinstance(opus_data, list): + # 将列表中的所有bytes数据合并 + combined_data = b"".join(opus_data) + else: + combined_data = opus_data + + # 将二进制数据转换为Base64编码的字符串 + opus_data_base64 = ( + base64.b64encode(combined_data).decode("utf-8") if combined_data else None + ) + + return ManageApiClient._instance._execute_request( + "POST", + f"/agent/chat-history/report", + json={ + "macAddress": mac_address, + "sessionId": session_id, + "chatType": chat_type, + "content": content, + "opusDataBase64": opus_data_base64, + }, + ) + except Exception as e: + print(f"TTS上报失败: {e}") + return None + def init_service(config): ManageApiClient(config) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 87052940..f08a61ff 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -17,7 +17,6 @@ from core.handle.textHandle import handleTextMessage from core.utils.util import ( get_string_no_punctuation_or_emoji, extract_json_from_string, - get_ip_info, initialize_modules, ) from concurrent.futures import ThreadPoolExecutor, TimeoutError @@ -30,7 +29,7 @@ from core.mcp.manager import MCPManager from config.config_loader import get_private_config_from_api from config.manage_api_client import DeviceNotFoundException, DeviceBindException from core.utils.output_counter import add_device_output -from core.handle.ttsReportHandle import enqueue_tts_report +from core.handle.ttsReportHandle import enqueue_tts_report, report_tts TAG = __name__ @@ -43,7 +42,15 @@ class TTSException(RuntimeError): class ConnectionHandler: def __init__( - self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _memory, _intent, server=None + self, + config: Dict[str, Any], + _vad, + _asr, + _llm, + _tts, + _memory, + _intent, + server=None, ): self.config = config self.server = server @@ -52,6 +59,7 @@ class ConnectionHandler: self.need_bind = False self.bind_code = None + self.read_config_from_api = self.config.get("read_config_from_api", False) self.websocket = None self.headers = None @@ -74,11 +82,8 @@ class ConnectionHandler: self.audio_play_queue = queue.Queue() self.executor = ThreadPoolExecutor(max_workers=10) - # 上报线程标志 - self.session_open_time = time.time() + # 上报线程 self.tts_report_queue = queue.Queue() - self.asr_report_queue = queue.Queue() - self.asr_report_thread = None self.tts_report_thread = None # 依赖的组件 @@ -275,19 +280,27 @@ class ConnectionHandler: ) except Exception as e: self.logger.bind(tag=TAG).error(f"模块初始化失败: {str(e)}") - await self.websocket.send(json.dumps({ - "type": "config_update_response", - "status": "error", - "message": f"模块初始化失败: {str(e)}" - })) + await self.websocket.send( + json.dumps( + { + "type": "config_update_response", + "status": "error", + "message": f"模块初始化失败: {str(e)}", + } + ) + ) return # 返回成功响应 - await self.websocket.send(json.dumps({ - "type": "config_update_response", - "status": "success", - "message": f"已更新配置: {', '.join(updated_modules)}" - })) + await self.websocket.send( + json.dumps( + { + "type": "config_update_response", + "status": "success", + "message": f"已更新配置: {', '.join(updated_modules)}", + } + ) + ) def _initialize_components(self, private_config): """初始化组件""" @@ -305,26 +318,18 @@ class ConnectionHandler: def _init_report_threads(self): """初始化ASR和TTS上报线程""" - if self.asr_report_thread is None or not self.asr_report_thread.is_alive(): - self.asr_report_thread = threading.Thread( - target=self._asr_report_worker, - daemon=True - ) - self.asr_report_thread.start() - self.logger.bind(tag=TAG).info("ASR上报线程已启动") - + if not self.read_config_from_api: + return if self.tts_report_thread is None or not self.tts_report_thread.is_alive(): self.tts_report_thread = threading.Thread( - target=self._tts_report_worker, - daemon=True + target=self._tts_report_worker, daemon=True ) self.tts_report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") def _initialize_private_config(self): - read_config_from_api = self.config.get("read_config_from_api", False) """如果是从配置文件获取，则进行二次实例化""" - if not read_config_from_api: + if not self.read_config_from_api: return """从接口获取差异化的配置进行二次实例化，非全量重新实例化""" try: @@ -880,10 +885,9 @@ class ConnectionHandler: f"TTS生成：文件路径: {tts_file}" ) if os.path.exists(tts_file): + opus_datas, _ = self.tts.audio_to_opus_data(tts_file) # 在这里上报TTS数据（使用文件路径） - enqueue_tts_report(self, text, tts_file) - - opus_datas, duration = self.tts.audio_to_opus_data(tts_file) + enqueue_tts_report(self, 2, text, opus_datas) else: self.logger.bind(tag=TAG).error( f"TTS出错：文件不存在{tts_file}" @@ -939,43 +943,8 @@ class ConnectionHandler: f"audio_play_priority priority_thread: {text} {e}" ) - def _asr_report_worker(self): - """ASR上报工作线程""" - # 提前导入避免循环引用问题 - from core.handle.asrReportHandle import report_asr - - while not self.stop_event.is_set(): - try: - # 从队列获取数据，设置超时以便定期检查停止事件 - item = self.asr_report_queue.get(timeout=1) - if item is None: # 检测毒丸对象 - break - - text, file_path = item - - try: - # 执行上报（传入文件路径） - await_result = report_asr(self, text, file_path) - - # 使用asyncio.run_coroutine_threadsafe执行异步操作 - future = asyncio.run_coroutine_threadsafe(await_result, self.loop) - future.result() - except Exception as e: - self.logger.bind(tag=TAG).error(f"ASR上报线程异常: {e}") - finally: - # 标记任务完成 - self.asr_report_queue.task_done() - except queue.Empty: - continue - except Exception as e: - self.logger.bind(tag=TAG).error(f"ASR上报工作线程异常: {e}") - - self.logger.bind(tag=TAG).info("ASR上报线程已退出") - def _tts_report_worker(self): """TTS上报工作线程""" - # 提前导入避免循环引用问题 - from core.handle.ttsReportHandle import report_tts while not self.stop_event.is_set(): try: @@ -984,15 +953,11 @@ class ConnectionHandler: if item is None: # 检测毒丸对象 break - text, audio_data = item + type, text, audio_data = item try: # 执行上报（传入二进制数据） - await_result = report_tts(self, text, audio_data) - - # 使用asyncio.run_coroutine_threadsafe执行异步操作 - future = asyncio.run_coroutine_threadsafe(await_result, self.loop) - future.result() + report_tts(self, type, text, audio_data) except Exception as e: self.logger.bind(tag=TAG).error(f"TTS上报线程异常: {e}") finally: @@ -1051,7 +1016,6 @@ class ConnectionHandler: self.executor = None # 添加毒丸对象到上报队列确保线程退出 - self.asr_report_queue.put(None) self.tts_report_queue.put(None) # 清空任务队列 diff --git a/main/xiaozhi-server/core/handle/asrReportHandle.py b/main/xiaozhi-server/core/handle/asrReportHandle.py deleted file mode 100644 index 70601804..00000000 --- a/main/xiaozhi-server/core/handle/asrReportHandle.py +++ /dev/null @@ -1,100 +0,0 @@ -""" -ASR上报功能已集成到ConnectionHandler类中。 - -上报功能包括： -1. 每个连接对象拥有自己的上报队列和处理线程 -2. 上报线程的生命周期与连接对象绑定 -3. 使用ConnectionHandler.enqueue_asr_report方法进行上报 - -具体实现请参考core/connection.py中的相关代码。 -""" - -import os -from config.logger import setup_logging -from config.manage_api_client import report - -TAG = __name__ -logger = setup_logging() - -async def report_asr(conn, text, file_path): - """执行ASR上报操作 - - Args: - conn: 连接对象 - text: 识别文本 - file_path: 音频文件路径（可以为None或空字符串，表示纯文本上报） - """ - audio_data = None - try: - # 处理无音频的纯文本上报 - if not file_path or not os.path.exists(file_path): - # 纯文本上报时使用空音频数据 - result = await report( - mac_address=conn.device_id, - session_id=conn.session_id, - sort=int(conn.session_open_time), - chat_type=1, # ASR类型为1 - content=text, - audio=b'', # 空音频数据 - file_extension="wav" - ) - logger.bind(tag=TAG).info(f"纯文本上报成功: {conn.device_id}, {conn.session_id}") - else: - # 读取文件为二进制数据 - with open(file_path, 'rb') as f: - audio_data = f.read() - - # 正常ASR上报（带音频） - result = await report( - mac_address=conn.device_id, - session_id=conn.session_id, - sort=int(conn.session_open_time), - chat_type=1, # ASR类型为1 - content=text, - audio=audio_data, - file_extension="wav" - ) - logger.bind(tag=TAG).info(f"ASR上报成功: {conn.device_id}, {conn.session_id}，文件: {file_path}") - - return result - except Exception as e: - logger.bind(tag=TAG).error(f"ASR上报失败: {e}") - return None - finally: - # 清理资源 - if file_path and os.path.exists(file_path): - try: - os.remove(file_path) - logger.bind(tag=TAG).debug(f"ASR上报后删除文件: {file_path}") - except Exception as e: - logger.bind(tag=TAG).error(f"ASR上报后删除文件失败: {e}") - - # 手动清理audio_data - if audio_data: - del audio_data - -def enqueue_asr_report(conn, text, audio): - """将ASR数据加入上报队列 - - Args: - conn: 连接对象 - text: 识别文本 - audio: 音频数据（可以为空列表，表示纯文本上报） - """ - try: - if not audio or len(audio) == 0: - # 纯文本上报，不需要保存文件 - file_path = None - else: - # 保存音频数据到文件 - file_path = conn.asr.save_audio_to_file(audio, conn.session_id) - - # 使用连接对象的队列，传入文件路径 - conn.asr_report_queue.put((text, file_path)) - - if not audio or len(audio) == 0: - logger.bind(tag=TAG).info(f"纯文本数据已加入上报队列: {conn.device_id}, {text[:20] if text else ''}...") - else: - logger.bind(tag=TAG).info(f"ASR数据已加入上报队列: {conn.device_id}, 文件: {file_path}") - except Exception as e: - logger.bind(tag=TAG).error(f"加入ASR上报队列失败: {e}") diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 2d8b0eef..a346fb61 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,10 +1,11 @@ from config.logger import setup_logging import time +import copy from core.utils.util import remove_punctuation_and_length 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.asrReportHandle import enqueue_asr_report +from core.handle.ttsReportHandle import enqueue_tts_report TAG = __name__ logger = setup_logging() @@ -42,7 +43,7 @@ async def handleAudioMessage(conn, audio): text_len, _ = remove_punctuation_and_length(text) if text_len > 0: # 使用自定义模块进行上报 - enqueue_asr_report(conn, text, conn.asr_audio) + enqueue_tts_report(conn, 1, text, copy.deepcopy(conn.asr_audio)) await startToChat(conn, text) else: diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 08c1c40e..d350f78c 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -6,7 +6,7 @@ from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.iotHandle import handleIotDescriptors, handleIotStatus -from core.handle.asrReportHandle import enqueue_asr_report +from core.handle.ttsReportHandle import enqueue_tts_report import asyncio TAG = __name__ @@ -56,11 +56,11 @@ async def handleTextMessage(conn, message): await send_tts_message(conn, "stop", None) elif is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, "嘿，你好呀", []) + enqueue_tts_report(conn, 1, "嘿，你好呀", []) await startToChat(conn, "嘿，你好呀") else: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, text, []) + enqueue_tts_report(conn, 1, text, []) # 否则需要LLM对文字内容进行答复 await startToChat(conn, text) elif msg_json["type"] == "iot": @@ -70,19 +70,22 @@ async def handleTextMessage(conn, message): asyncio.create_task(handleIotStatus(conn, msg_json["states"])) elif msg_json["type"] == "server": # 如果配置是从API读取的，则需要验证secret - read_config_from_api = conn.config.get("read_config_from_api", False) - if not read_config_from_api: + if not conn.read_config_from_api: return # 获取post请求的secret post_secret = msg_json.get("content", {}).get("secret", "") secret = conn.config["manager-api"].get("secret", "") # 如果secret不匹配，则返回 if post_secret != secret: - await conn.websocket.send(json.dumps({ - "type": "config_update_response", - "status": "error", - "message": "服务器密钥验证失败" - })) + await conn.websocket.send( + json.dumps( + { + "type": "config_update_response", + "status": "error", + "message": "服务器密钥验证失败", + } + ) + ) return # 动态更新配置 if msg_json["action"] == "update_config": diff --git a/main/xiaozhi-server/core/handle/ttsReportHandle.py b/main/xiaozhi-server/core/handle/ttsReportHandle.py index ed3a6a76..3aec8a32 100644 --- a/main/xiaozhi-server/core/handle/ttsReportHandle.py +++ b/main/xiaozhi-server/core/handle/ttsReportHandle.py @@ -9,62 +9,51 @@ TTS上报功能已集成到ConnectionHandler类中。 具体实现请参考core/connection.py中的相关代码。 """ -import os from config.logger import setup_logging from config.manage_api_client import report TAG = __name__ logger = setup_logging() -async def report_tts(conn, text, audio_data): + +def report_tts(conn, type, text, opus_data): """执行TTS上报操作 - + Args: conn: 连接对象 + type: 上报类型，1为用户，2为智能体 text: 合成文本 - audio_data: 音频二进制数据 + opus_data: opus音频数据 """ try: # 执行上报 - result = await report( + report( mac_address=conn.device_id, session_id=conn.session_id, - sort=int(conn.session_open_time), - chat_type=2, # TTS类型为2 + chat_type=type, content=text, - audio=audio_data, - file_extension="wav" + opus_data=opus_data, ) - logger.bind(tag=TAG).info(f"TTS上报成功: {conn.device_id}, {conn.session_id}, 数据大小: {len(audio_data)} 字节") - return result except Exception as e: logger.bind(tag=TAG).error(f"TTS上报失败: {e}") - return None - finally: - # 手动清理audio_data引用，帮助垃圾回收 - del audio_data -def enqueue_tts_report(conn, text, file_path): + +def enqueue_tts_report(conn, type, text, opus_data): + if not conn.read_config_from_api: + return """将TTS数据加入上报队列 - + Args: conn: 连接对象 text: 合成文本 - file_path: TTS音频文件路径 + opus_data: opus音频数据 """ try: - # 检查文件是否存在 - if not file_path or not os.path.exists(file_path): - logger.bind(tag=TAG).error(f"加入TTS上报队列失败: 文件不存在 {file_path}") - return - - # 立即读取文件为二进制数据，因为外部会删除文件 - with open(file_path, 'rb') as f: - audio_data = f.read() - # 使用连接对象的队列，传入文本和二进制数据而非文件路径 - conn.tts_report_queue.put((text, audio_data)) - - logger.bind(tag=TAG).info(f"TTS数据已加入上报队列: {conn.device_id}, 文件大小: {len(audio_data)} 字节") + conn.tts_report_queue.put((type, text, opus_data)) + + logger.bind(tag=TAG).info( + f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " + ) except Exception as e: - logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {e}, 文件: {file_path}") + logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}")