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] =?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] # 如果都不在优先级列表里，返回第一个