diff --git a/main/manager-api/pom.xml b/main/manager-api/pom.xml index 883295bf..bc4030c0 100644 --- a/main/manager-api/pom.xml +++ b/main/manager-api/pom.xml @@ -258,4 +258,4 @@ - \ No newline at end of file + diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java index bdaed950..ae58783f 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java @@ -28,4 +28,6 @@ public class AgentChatHistoryReportDTO { private String content; @Schema(description = "base64编码的opus音频数据", example = "") private String audioBase64; + @Schema(description = "上报时间，十位时间戳，空时默认使用当前时间", example = "1745657732") + private Long reportTime; } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java index e284b69f..04be9f36 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java @@ -27,4 +27,4 @@ public interface AgentChatAudioService extends IService { * @return 音频数据 */ byte[] getAudio(String audioId); -} \ No newline at end of file +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java index bf6c188b..087c1b9d 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java @@ -47,7 +47,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic public Boolean report(AgentChatHistoryReportDTO report) { String macAddress = report.getMacAddress(); Byte chatType = report.getChatType(); - log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType); + Long reportTimeMillis = null != report.getReportTime() ? report.getReportTime() * 1000 : System.currentTimeMillis(); + log.info("小智设备聊天上报请求: macAddress={}, type={} reportTime={}", macAddress, chatType, reportTimeMillis); // 根据设备MAC地址查询对应的默认智能体，判断是否需要上报 AgentEntity agentEntity = agentService.getDefaultAgentByMacAddress(macAddress); @@ -59,10 +60,10 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic String agentId = agentEntity.getId(); if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT.getCode())) { - saveChatText(report, agentId, macAddress, null); + saveChatText(report, agentId, macAddress, null, reportTimeMillis); } else if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT_AUDIO.getCode())) { String audioId = saveChatAudio(report); - saveChatText(report, agentId, macAddress, audioId); + saveChatText(report, agentId, macAddress, audioId, reportTimeMillis); } // 更新设备最后对话时间 @@ -92,8 +93,7 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic /** * 组装上报数据 */ - private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId) { - + private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId, Long reportTime) { // 构建聊天记录实体 AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder() .macAddress(macAddress) @@ -102,6 +102,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic .chatType(report.getChatType()) .content(report.getContent()) .audioId(audioId) + .createdAt(new Date(reportTime)) + // NOTE(haotian): 2025/5/26 updateAt可以不设置，重点是createAt，而且这样可以看到上报延迟 .build(); // 保存数据 diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java index cf804678..efcae485 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java @@ -31,4 +31,4 @@ public class AgentChatAudioServiceImpl extends ServiceImpl Optional[Dict]: def report( - mac_address: str, session_id: str, chat_type: int, content: str, audio + mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time ) -> Optional[Dict]: """带熔断的业务方法示例""" if not content or not ManageApiClient._instance: @@ -174,6 +174,7 @@ def report( "sessionId": session_id, "chatType": chat_type, "content": content, + "reportTime": report_time, "audioBase64": ( base64.b64encode(audio).decode("utf-8") if audio else None ), diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 3d9d8e3c..9742ebc3 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -890,11 +890,11 @@ class ConnectionHandler: if item is None: # 检测毒丸对象 break - type, text, audio_data = item + type, text, audio_data, report_time = item try: # 提交任务到线程池 - self.report_thread_pool.submit(self._process_report, type, text, audio_data) + self.report_thread_pool.submit(self._process_report, type, text, audio_data, report_time) except Exception as e: self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}") except queue.Empty: @@ -904,11 +904,11 @@ class ConnectionHandler: self.logger.bind(tag=TAG).info("聊天记录上报线程已退出") - def _process_report(self, type, text, audio_data): + def _process_report(self, type, text, audio_data, report_time): """处理上报任务""" try: # 执行上报（传入二进制数据） - report(self, type, text, audio_data) + report(self, type, text, audio_data, report_time) except Exception as e: self.logger.bind(tag=TAG).error(f"上报处理异常: {e}") finally: @@ -973,6 +973,12 @@ class ConnectionHandler: self.executor.shutdown(wait=False) self.executor = None + # 关闭上报线程池 + if self.report_thread_pool: + self.report_thread_pool.shutdown(wait=False) + self.report_thread_pool = None + self.logger.bind(tag=TAG).info("上报线程池已关闭") + self.logger.bind(tag=TAG).info("连接资源已释放") def clear_queues(self): diff --git a/main/xiaozhi-server/core/handle/reportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py index bb1ea066..b1214fd5 100644 --- a/main/xiaozhi-server/core/handle/reportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -8,6 +8,7 @@ TTS上报功能已集成到ConnectionHandler类中。 具体实现请参考core/connection.py中的相关代码。 """ +import time import opuslib_next @@ -16,7 +17,7 @@ from config.manage_api_client import report as manage_report TAG = __name__ -def report(conn, type, text, opus_data): +def report(conn, type, text, opus_data, report_time): """执行聊天记录上报操作 Args: @@ -24,6 +25,7 @@ def report(conn, type, text, opus_data): type: 上报类型，1为用户，2为智能体 text: 合成文本 opus_data: opus音频数据 + report_time: 上报时间 """ try: if opus_data: @@ -37,6 +39,7 @@ def report(conn, type, text, opus_data): chat_type=type, content=text, audio=audio_data, + report_time=report_time, ) except Exception as e: conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}") @@ -104,12 +107,12 @@ def enqueue_tts_report(conn, text, opus_data): try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 if conn.chat_history_conf == 2: - conn.report_queue.put((2, text, opus_data)) + conn.report_queue.put((2, text, opus_data, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) else: - conn.report_queue.put((2, text, None)) + conn.report_queue.put((2, text, None, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"TTS数据已加入上报队列: {conn.device_id}, 不上报音频" ) @@ -132,12 +135,12 @@ def enqueue_asr_report(conn, text, opus_data): try: # 使用连接对象的队列，传入文本和二进制数据而非文件路径 if conn.chat_history_conf == 2: - conn.report_queue.put((1, text, opus_data)) + conn.report_queue.put((1, text, opus_data, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} " ) else: - conn.report_queue.put((1, text, None)) + conn.report_queue.put((1, text, None, int(time.time()))) conn.logger.bind(tag=TAG).debug( f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频" )