update:优化对话数据上传

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)

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)
 
         # 清空任务队列

main/xiaozhi-server/core/handle/asrReportHandle.py

@@ -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}")

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:

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":

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}")