Merge pull request #2628 from xinnan-tech/py_await

Py await

main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java

@@ -299,7 +299,7 @@ public interface Constant {
     /**
      * 版本号
      */
-    public static final String VERSION = "0.8.8";
+    public static final String VERSION = "0.8.9";
 
     /**
      * 无效固件URL

main/manager-mobile/src/pages/settings/index.vue

@@ -235,7 +235,7 @@ function showAbout() {
     title: t('settings.aboutApp', { appName: import.meta.env.VITE_APP_TITLE }),
     content: t('settings.aboutContent', {
       appName: import.meta.env.VITE_APP_TITLE,
-      version: '0.8.8'
+      version: '0.8.9'
     }),
     showCancel: false,
     confirmText: t('common.confirm'),

main/xiaozhi-server/app.py

@@ -9,6 +9,7 @@ from core.utils.util import get_local_ip, validate_mcp_endpoint
 from core.http_server import SimpleHttpServer
 from core.websocket_server import WebSocketServer
 from core.utils.util import check_ffmpeg_installed
+from core.utils.gc_manager import get_gc_manager
 
 TAG = __name__
 logger = setup_logging()
@@ -63,6 +64,10 @@ async def main():
     # 添加 stdin 监控任务
     stdin_task = asyncio.create_task(monitor_stdin())
 
+    # 启动全局GC管理器（5分钟清理一次）
+    gc_manager = get_gc_manager(interval_seconds=300)
+    await gc_manager.start()
+
     # 启动 WebSocket 服务器
     ws_server = WebSocketServer(config)
     ws_task = asyncio.create_task(ws_server.start())
@@ -122,6 +127,9 @@ async def main():
     except asyncio.CancelledError:
         print("任务被取消，清理资源中...")
     finally:
+        # 停止全局GC管理器
+        await gc_manager.stop()
+
         # 取消所有任务（关键修复点）
         stdin_task.cancel()
         ws_task.cancel()

main/xiaozhi-server/config/config_loader.py

@@ -32,7 +32,16 @@ def load_config():
     custom_config = read_config(custom_config_path)
 
     if custom_config.get("manager-api", {}).get("url"):
-        config = get_config_from_api(custom_config)
+        import asyncio
+        try:
+            loop = asyncio.get_running_loop()
+            # 如果已经在事件循环中，使用异步版本
+            config = asyncio.run_coroutine_threadsafe(
+                get_config_from_api_async(custom_config), loop
+            ).result()
+        except RuntimeError:
+            # 如果不在事件循环中（启动时），创建新的事件循环
+            config = asyncio.run(get_config_from_api_async(custom_config))
     else:
         # 合并配置
         config = merge_configs(default_config, custom_config)
@@ -44,13 +53,13 @@ def load_config():
     return config
 
 
-def get_config_from_api(config):
-    """从Java API获取配置"""
+async def get_config_from_api_async(config):
+    """从Java API获取配置（异步版本）"""
     # 初始化API客户端
     init_service(config)
 
     # 获取服务器配置
-    config_data = get_server_config()
+    config_data = await get_server_config()
     if config_data is None:
         raise Exception("Failed to fetch server config from API")
 
@@ -74,9 +83,9 @@ def get_config_from_api(config):
     return config_data
 
 
-def get_private_config_from_api(config, device_id, client_id):
+async def get_private_config_from_api(config, device_id, client_id):
     """从Java API获取私有配置"""
-    return get_agent_models(device_id, client_id, config["selected_module"])
+    return await get_agent_models(device_id, client_id, config["selected_module"])
 
 
 def ensure_directories(config):

main/xiaozhi-server/config/logger.py

@@ -5,7 +5,7 @@ from config.config_loader import load_config
 from config.settings import check_config_file
 from datetime import datetime
 
-SERVER_VERSION = "0.8.8"
+SERVER_VERSION = "0.8.9"
 _logger_initialized = False
 
 

main/xiaozhi-server/config/manage_api_client.py

@@ -1,5 +1,4 @@
 import os
-import time
 import base64
 from typing import Optional, Dict
 
@@ -20,7 +19,7 @@ class DeviceBindException(Exception):
 
 class ManageApiClient:
     _instance = None
-    _client = None
+    _async_clients = {}  # 为每个事件循环存储独立的客户端
     _secret = None
 
     def __new__(cls, config):
@@ -32,7 +31,7 @@ class ManageApiClient:
 
     @classmethod
     def _init_client(cls, config):
-        """初始化持久化连接池"""
+        """初始化配置（延迟创建客户端）"""
         cls.config = config.get("manager-api")
 
         if not cls.config:
@@ -47,23 +46,40 @@ class ManageApiClient:
         cls._secret = cls.config.get("secret")
         cls.max_retries = cls.config.get("max_retries", 6)  # 最大重试次数
         cls.retry_delay = cls.config.get("retry_delay", 10)  # 初始重试延迟(秒)
-        # NOTE(goody): 2025/4/16 http相关资源统一管理，后续可以增加线程池或者超时
-        # 后续也可以统一配置apiToken之类的走通用的Auth
-        cls._client = httpx.Client(
-            base_url=cls.config.get("url"),
-            headers={
-                "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})",
-                "Accept": "application/json",
-                "Authorization": "Bearer " + cls._secret,
-            },
-            timeout=cls.config.get("timeout", 30),  # 默认超时时间30秒
-        )
+        # 不在这里创建 AsyncClient，延迟到实际使用时创建
+        cls._async_clients = {}
 
     @classmethod
-    def _request(cls, method: str, endpoint: str, **kwargs) -> Dict:
-        """发送单次HTTP请求并处理响应"""
+    async def _ensure_async_client(cls):
+        """确保异步客户端已创建（为每个事件循环创建独立的客户端）"""
+        import asyncio
+        try:
+            loop = asyncio.get_running_loop()
+            loop_id = id(loop)
+
+            # 为每个事件循环创建独立的客户端
+            if loop_id not in cls._async_clients:
+                cls._async_clients[loop_id] = httpx.AsyncClient(
+                    base_url=cls.config.get("url"),
+                    headers={
+                        "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})",
+                        "Accept": "application/json",
+                        "Authorization": "Bearer " + cls._secret,
+                    },
+                    timeout=cls.config.get("timeout", 30),
+                )
+            return cls._async_clients[loop_id]
+        except RuntimeError:
+            # 如果没有运行中的事件循环，创建一个临时的
+            raise Exception("必须在异步上下文中调用")
+
+    @classmethod
+    async def _async_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """发送单次异步HTTP请求并处理响应"""
+        # 确保客户端已创建
+        client = await cls._ensure_async_client()
         endpoint = endpoint.lstrip("/")
-        response = cls._client.request(method, endpoint, **kwargs)
+        response = await client.request(method, endpoint, **kwargs)
         response.raise_for_status()
 
         result = response.json()
@@ -96,22 +112,23 @@ class ManageApiClient:
         return False
 
     @classmethod
-    def _execute_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
-        """带重试机制的请求执行器"""
+    async def _execute_async_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """带重试机制的异步请求执行器"""
+        import asyncio
         retry_count = 0
 
         while retry_count <= cls.max_retries:
             try:
-                # 执行请求
-                return cls._request(method, endpoint, **kwargs)
+                # 执行异步请求
+                return await cls._async_request(method, endpoint, **kwargs)
             except Exception as e:
                 # 判断是否应该重试
                 if retry_count < cls.max_retries and cls._should_retry(e):
                     retry_count += 1
                     print(
-                        f"{method} {endpoint} 请求失败，将在 {cls.retry_delay:.1f} 秒后进行第 {retry_count} 次重试"
+                        f"{method} {endpoint} 异步请求失败，将在 {cls.retry_delay:.1f} 秒后进行第 {retry_count} 次重试"
                     )
-                    time.sleep(cls.retry_delay)
+                    await asyncio.sleep(cls.retry_delay)
                     continue
                 else:
                     # 不重试，直接抛出异常
@@ -119,22 +136,27 @@ class ManageApiClient:
 
     @classmethod
     def safe_close(cls):
-        """安全关闭连接池"""
-        if cls._client:
-            cls._client.close()
-            cls._instance = None
+        """安全关闭所有异步连接池"""
+        import asyncio
+        for client in list(cls._async_clients.values()):
+            try:
+                asyncio.run(client.aclose())
+            except Exception:
+                pass
+        cls._async_clients.clear()
+        cls._instance = None
 
 
-def get_server_config() -> Optional[Dict]:
+async def get_server_config() -> Optional[Dict]:
     """获取服务器基础配置"""
-    return ManageApiClient._instance._execute_request("POST", "/config/server-base")
+    return await ManageApiClient._instance._execute_async_request("POST", "/config/server-base")
 
 
-def get_agent_models(
+async def get_agent_models(
     mac_address: str, client_id: str, selected_module: Dict
 ) -> Optional[Dict]:
     """获取代理模型配置"""
-    return ManageApiClient._instance._execute_request(
+    return await ManageApiClient._instance._execute_async_request(
         "POST",
         "/config/agent-models",
         json={
@@ -145,9 +167,9 @@ def get_agent_models(
     )
 
 
-def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
+async def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
     try:
-        return ManageApiClient._instance._execute_request(
+        return await ManageApiClient._instance._execute_async_request(
             "PUT",
             f"/agent/saveMemory/" + mac_address,
             json={
@@ -159,14 +181,14 @@ def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
         return None
 
 
-def report(
+async def report(
     mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time
 ) -> Optional[Dict]:
-    """带熔断的业务方法示例"""
+    """异步聊天记录上报"""
     if not content or not ManageApiClient._instance:
         return None
     try:
-        return ManageApiClient._instance._execute_request(
+        return await ManageApiClient._instance._execute_async_request(
             "POST",
             f"/agent/chat-history/report",
             json={

main/xiaozhi-server/core/api/vision_handler.py

@@ -96,7 +96,7 @@ class VisionHandler:
             current_config = copy.deepcopy(self.config)
             read_config_from_api = current_config.get("read_config_from_api", False)
             if read_config_from_api:
-                current_config = get_private_config_from_api(
+                current_config = await get_private_config_from_api(
                     current_config,
                     device_id,
                     client_id,

main/xiaozhi-server/core/connection.py

@@ -71,7 +71,7 @@ class ConnectionHandler:
         self.need_bind = False  # 是否需要绑定设备
         self.bind_code = None  # 绑定设备的验证码
         self.last_bind_prompt_time = 0  # 上次播放绑定提示的时间戳(秒)
-        self.bind_prompt_interval = 30  # 绑定提示播放间隔(秒)
+        self.bind_prompt_interval = 60  # 绑定提示播放间隔(秒)
 
         self.read_config_from_api = self.config.get("read_config_from_api", False)
 
@@ -91,7 +91,7 @@ class ConnectionHandler:
         self.client_listen_mode = "auto"
 
         # 线程任务相关
-        self.loop = asyncio.get_event_loop()
+        self.loop = None  # 在 handle_connection 中获取运行中的事件循环
         self.stop_event = threading.Event()
         self.executor = ThreadPoolExecutor(max_workers=5)
 
@@ -166,6 +166,9 @@ class ConnectionHandler:
 
     async def handle_connection(self, ws):
         try:
+            # 获取运行中的事件循环（必须在异步上下文中）
+            self.loop = asyncio.get_running_loop()
+
             # 获取并验证headers
             self.headers = dict(ws.request.headers)
             real_ip = self.headers.get("x-real-ip") or self.headers.get(
@@ -200,10 +203,8 @@ class ConnectionHandler:
             self.welcome_msg = self.config["xiaozhi"]
             self.welcome_msg["session_id"] = self.session_id
 
-            # 获取差异化配置
-            self._initialize_private_config()
-            # 异步初始化
-            self.executor.submit(self._initialize_components)
+            # 在后台初始化配置和组件（完全不阻塞主循环）
+            asyncio.create_task(self._background_initialize())
 
             try:
                 async for message in self.websocket:
@@ -522,21 +523,30 @@ class ConnectionHandler:
         except Exception as e:
             self.logger.bind(tag=TAG).warning(f"声纹识别初始化失败: {str(e)}")
 
-    def _initialize_private_config(self):
-        """如果是从配置文件获取，则进行二次实例化"""
+    async def _background_initialize(self):
+        """在后台初始化配置和组件（完全不阻塞主循环）"""
+        try:
+            # 异步获取差异化配置
+            await self._initialize_private_config_async()
+            # 在线程池中初始化组件
+            self.executor.submit(self._initialize_components)
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"后台初始化失败: {e}")
+
+    async def _initialize_private_config_async(self):
+        """从接口异步获取差异化配置（异步版本，不阻塞主循环）"""
         if not self.read_config_from_api:
             return
-        """从接口获取差异化的配置进行二次实例化，非全量重新实例化"""
         try:
             begin_time = time.time()
-            private_config = get_private_config_from_api(
+            private_config = await get_private_config_from_api(
                 self.config,
                 self.headers.get("device-id"),
                 self.headers.get("client-id", self.headers.get("device-id")),
             )
             private_config["delete_audio"] = bool(self.config.get("delete_audio", True))
             self.logger.bind(tag=TAG).info(
-                f"{time.time() - begin_time} 秒，获取差异化配置成功: {json.dumps(filter_sensitive_info(private_config), ensure_ascii=False)}"
+                f"{time.time() - begin_time} 秒，异步获取差异化配置成功: {json.dumps(filter_sensitive_info(private_config), ensure_ascii=False)}"
             )
         except DeviceNotFoundException as e:
             self.need_bind = True
@@ -547,7 +557,7 @@ class ConnectionHandler:
             private_config = {}
         except Exception as e:
             self.need_bind = True
-            self.logger.bind(tag=TAG).error(f"获取差异化配置失败: {e}")
+            self.logger.bind(tag=TAG).error(f"异步获取差异化配置失败: {e}")
             private_config = {}
 
         init_llm, init_tts, init_memory, init_intent = (
@@ -620,8 +630,12 @@ class ConnectionHandler:
             self.chat_history_conf = int(private_config["chat_history_conf"])
         if private_config.get("mcp_endpoint", None) is not None:
             self.config["mcp_endpoint"] = private_config["mcp_endpoint"]
+
+        # 使用 run_in_executor 在线程池中执行 initialize_modules，避免阻塞主循环
         try:
-            modules = initialize_modules(
+            modules = await self.loop.run_in_executor(
+                None,  # 使用默认线程池
+                initialize_modules,
                 self.logger,
                 private_config,
                 init_vad,
@@ -1034,8 +1048,8 @@ class ConnectionHandler:
     def _process_report(self, type, text, audio_data, report_time):
         """处理上报任务"""
         try:
-            # 执行上报（传入二进制数据）
-            report(self, type, text, audio_data, report_time)
+            # 执行异步上报（在事件循环中运行）
+            asyncio.run(report(self, type, text, audio_data, report_time))
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
         finally:
@@ -1126,10 +1140,6 @@ class ConnectionHandler:
                         f"关闭线程池时出错: {executor_error}"
                     )
                 self.executor = None
-
-            import gc
-            gc.collect()
-
             self.logger.bind(tag=TAG).info("连接资源已释放")
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"关闭连接时出错: {e}")

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

@@ -18,7 +18,7 @@ from config.manage_api_client import report as manage_report
 TAG = __name__
 
 
-def report(conn, type, text, opus_data, report_time):
+async def report(conn, type, text, opus_data, report_time):
     """执行聊天记录上报操作
 
     Args:
@@ -33,8 +33,8 @@ def report(conn, type, text, opus_data, report_time):
             audio_data = opus_to_wav(conn, opus_data)
         else:
             audio_data = None
-        # 执行上报
-        manage_report(
+        # 执行异步上报
+        await manage_report(
             mac_address=conn.device_id,
             session_id=conn.session_id,
             chat_type=type,

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

@@ -4,6 +4,7 @@ import asyncio
 from core.utils import textUtils
 from core.utils.util import audio_to_data
 from core.providers.tts.dto.dto import SentenceType
+from core.utils.audioRateController import AudioRateController
 
 TAG = __name__
 
@@ -30,31 +31,6 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
             await conn.close()
 
 
-def calculate_timestamp_and_sequence(conn, start_time, packet_index, frame_duration=60):
-    """
-    计算音频数据包的时间戳和序列号
-    Args:
-        conn: 连接对象
-        start_time: 起始时间（性能计数器值）
-        packet_index: 数据包索引
-        frame_duration: 帧时长（毫秒），匹配 Opus 编码
-    Returns:
-        tuple: (timestamp, sequence)
-    """
-    # 计算时间戳（使用播放位置计算）
-    timestamp = int((start_time + packet_index * frame_duration / 1000) * 1000) % (
-        2**32
-    )
-
-    # 计算序列号
-    if hasattr(conn, "audio_flow_control"):
-        sequence = conn.audio_flow_control["sequence"]
-    else:
-        sequence = packet_index  # 如果没有流控状态，直接使用索引
-
-    return timestamp, sequence
-
-
 async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     """
     发送带16字节头部的opus数据包给mqtt_gateway
@@ -77,15 +53,20 @@ async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     await conn.websocket.send(complete_packet)
 
 
-# 播放音频
+# 播放音频 - 使用 AudioRateController 进行精确流控
 async def sendAudio(conn, audios, frame_duration=60):
     """
-    发送单个opus包，支持流控
+    发送音频包，使用 AudioRateController 进行精确的流量控制
+
     Args:
         conn: 连接对象
-        opus_packet: 单个opus数据包
-        pre_buffer: 快速发送音频
-        frame_duration: 帧时长（毫秒），匹配 Opus 编码
+        audios: 单个opus包(bytes) 或 opus包列表
+        frame_duration: 帧时长（毫秒），默认60ms
+
+    改进点：
+    1. 使用单一时间基准，避免累积误差
+    2. 每次检查队列时重新计算 elapsed_ms，更精准
+    3. 支持高并发而不产生时间偏差
     """
     if audios is None or len(audios) == 0:
         return
@@ -94,118 +75,133 @@ async def sendAudio(conn, audios, frame_duration=60):
     send_delay = conn.config.get("tts_audio_send_delay", -1) / 1000.0
 
     if isinstance(audios, bytes):
-        # 重置流控状态,第一次读取和会话发生转变时
-        if not hasattr(conn, "audio_flow_control") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id:
-            conn.audio_flow_control = {
-                "last_send_time": 0,
-                "packet_count": 0,
-                "start_time": time.perf_counter(),
-                "sequence": 0,  # 添加序列号
-                "sentence_id": conn.sentence_id,
-            }
+        # 单个 opus 包处理
+        await _sendAudio_single(conn, audios, send_delay, frame_duration)
+    else:
+        # 音频列表处理（如文件型音频）
+        await _sendAudio_list(conn, audios, send_delay, frame_duration)
 
+
+async def _sendAudio_single(conn, opus_packet, send_delay, frame_duration=60):
+    """
+    发送单个 opus 包
+    使用 AudioRateController 进行流控
+    """
+    # 重置流控状态，第一次读取和会话发生转变时
+    if not hasattr(conn, "audio_rate_controller") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id:
+        if hasattr(conn, "audio_rate_controller"):
+            conn.audio_rate_controller.reset()
+        else:
+            conn.audio_rate_controller = AudioRateController(frame_duration)
+            conn.audio_rate_controller.reset()
+
+        conn.audio_flow_control = {
+            "packet_count": 0,
+            "sequence": 0,
+            "sentence_id": conn.sentence_id,
+        }
+
+    if conn.client_abort:
+        return
+
+    conn.last_activity_time = time.time() * 1000
+
+    rate_controller = conn.audio_rate_controller
+    flow_control = conn.audio_flow_control
+    packet_count = flow_control["packet_count"]
+
+    # 预缓冲：前5个包直接发送，不做延迟
+    pre_buffer_count = 5
+
+    if packet_count < pre_buffer_count or send_delay > 0:
+        # 预缓冲阶段或固定延迟模式，直接发送
+        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
+
+        if send_delay > 0 and packet_count >= pre_buffer_count:
+            await asyncio.sleep(send_delay)
+    else:
+        # 使用流控器进行精确的速率控制
+        rate_controller.add_audio(opus_packet)
+
+        async def send_callback(packet):
+            await _do_send_audio(conn, packet, flow_control, frame_duration)
+
+        await rate_controller.check_queue(send_callback)
+
+    # 更新流控状态
+    flow_control["packet_count"] += 1
+    flow_control["sequence"] += 1
+
+
+async def _sendAudio_list(conn, audios, send_delay, frame_duration=60):
+    """
+    发送音频列表（如文件型音频）
+    """
+    if not audios:
+        return
+
+    rate_controller = AudioRateController(frame_duration)
+    rate_controller.reset()
+
+    flow_control = {
+        "packet_count": 0,
+        "sequence": 0,
+    }
+
+    # 预缓冲：前5个包直接发送
+    pre_buffer_frames = min(5, len(audios))
+    for i in range(pre_buffer_frames):
         if conn.client_abort:
             return
+        await _do_send_audio(conn, audios[i], flow_control, frame_duration)
+        conn.client_is_speaking = True
+
+    remaining_audios = audios[pre_buffer_frames:]
+
+    # 处理剩余音频帧
+    for i, opus_packet in enumerate(remaining_audios):
+        if conn.client_abort:
+            break
 
         conn.last_activity_time = time.time() * 1000
 
-        # 预缓冲：前5个包直接发送，不做延迟
-        pre_buffer_count = 5
-        flow_control = conn.audio_flow_control
-        current_time = time.perf_counter()
-
-        if flow_control["packet_count"] < pre_buffer_count:
-            # 预缓冲阶段，直接发送不延迟
-            pass
-        elif send_delay > 0:
-            # 使用固定延迟
+        if send_delay > 0:
+            # 固定延迟模式
             await asyncio.sleep(send_delay)
         else:
-            effective_packet = flow_control["packet_count"] - pre_buffer_count
-            expected_time = flow_control["start_time"] + (
-                effective_packet * frame_duration / 1000
-            )
-            delay = expected_time - current_time
-            if delay > 0:
-                await asyncio.sleep(delay)
-            else:
-                # 纠正误差
-                flow_control["start_time"] += abs(delay)
+            # 使用流控器进行精确延迟
+            rate_controller.add_audio(opus_packet)
 
-        if conn.conn_from_mqtt_gateway:
-            # 计算时间戳和序列号
-            timestamp, sequence = calculate_timestamp_and_sequence(
-                conn,
-                flow_control["start_time"],
-                flow_control["packet_count"],
-                frame_duration,
-            )
-            # 调用通用函数发送带头部的数据包
-            await _send_to_mqtt_gateway(conn, audios, timestamp, sequence)
-        else:
-            # 直接发送opus数据包，不添加头部
-            await conn.websocket.send(audios)
+            async def send_callback(packet):
+                await _do_send_audio(conn, packet, flow_control, frame_duration)
+
+            await rate_controller.check_queue(send_callback)
+            conn.client_is_speaking = True
+            continue
+
+        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
         conn.client_is_speaking = True
 
-        # 更新流控状态
-        flow_control["packet_count"] += 1
-        flow_control["sequence"] += 1
-        flow_control["last_send_time"] = time.perf_counter()
+
+async def _do_send_audio(conn, opus_packet, flow_control, frame_duration=60):
+    """
+    执行实际的音频发送
+    """
+    packet_index = flow_control.get("packet_count", 0)
+    sequence = flow_control.get("sequence", 0)
+
+    if conn.conn_from_mqtt_gateway:
+        # 计算时间戳（基于播放位置）
+        start_time = time.time()
+        timestamp = int(start_time * 1000) % (2**32)
+        await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence)
     else:
-        # 文件型音频走普通播放
-        start_time = time.perf_counter()
-        play_position = 0
+        # 直接发送opus数据包
+        await conn.websocket.send(opus_packet)
 
-        # 执行预缓冲
-        pre_buffer_frames = min(5, len(audios))
-        for i in range(pre_buffer_frames):
-            if conn.conn_from_mqtt_gateway:
-                # 计算时间戳和序列号
-                timestamp, sequence = calculate_timestamp_and_sequence(
-                    conn, start_time, i, frame_duration
-                )
-                # 调用通用函数发送带头部的数据包
-                await _send_to_mqtt_gateway(conn, audios[i], timestamp, sequence)
-            else:
-                # 直接发送预缓冲包，不添加头部
-                await conn.websocket.send(audios[i])
-            conn.client_is_speaking = True
-        remaining_audios = audios[pre_buffer_frames:]
-
-        # 播放剩余音频帧
-        for i, opus_packet in enumerate(remaining_audios):
-            if conn.client_abort:
-                break
-
-            # 重置没有声音的状态
-            conn.last_activity_time = time.time() * 1000
-
-            if send_delay > 0:
-                # 固定延迟模式
-                await asyncio.sleep(send_delay)
-            else:
-                 # 计算预期发送时间
-                expected_time = start_time + (play_position / 1000)
-                current_time = time.perf_counter()
-                delay = expected_time - current_time
-                if delay > 0:
-                    await asyncio.sleep(delay)
-
-            if conn.conn_from_mqtt_gateway:
-                # 计算时间戳和序列号（使用当前的数据包索引确保连续性）
-                packet_index = pre_buffer_frames + i
-                timestamp, sequence = calculate_timestamp_and_sequence(
-                    conn, start_time, packet_index, frame_duration
-                )
-                # 调用通用函数发送带头部的数据包
-                await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence)
-            else:
-                # 直接发送opus数据包，不添加头部
-                await conn.websocket.send(opus_packet)
-
-            conn.client_is_speaking = True
-
-            play_position += frame_duration
+    # 更新流控状态
+    flow_control["packet_count"] = packet_index + 1
+    flow_control["sequence"] = sequence + 1
 
 
 async def send_tts_message(conn, state, text=None):

main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py

@@ -5,6 +5,7 @@ import os
 import yaml
 from config.config_loader import get_project_dir
 from config.manage_api_client import save_mem_local_short
+import asyncio
 from core.utils.util import check_model_key
 
 
@@ -193,7 +194,13 @@ class MemoryProvider(MemoryProviderBase):
                 max_tokens=2000,
                 temperature=0.2,
             )
-            save_mem_local_short(self.role_id, result)
+            # 使用异步版本，需要在事件循环中运行
+            try:
+                loop = asyncio.get_running_loop()
+                loop.create_task(save_mem_local_short(self.role_id, result))
+            except RuntimeError:
+                # 如果没有运行中的事件循环，创建一个新的
+                asyncio.run(save_mem_local_short(self.role_id, result))
         logger.bind(tag=TAG).info(f"Save memory successful - Role: {self.role_id}")
 
         return self.short_memory

main/xiaozhi-server/core/utils/audioRateController.py

@@ -0,0 +1,132 @@
+import time
+import asyncio
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class AudioRateController:
+    """
+    音频速率控制器 - 按照60ms帧时长精确控制音频发送
+    解决高并发下的时间累积误差问题
+
+    关键改进：
+    1. 单一时间基准（start_timestamp 只初始化一次）
+    2. 每次检查队列时重新计算 elapsed_ms，避免累积误差
+    3. 分离"检查时间"和"发送"两个操作
+    4. 支持高并发而不产生延迟
+    """
+
+    def __init__(self, frame_duration=60):
+        """
+        Args:
+            frame_duration: 单个音频帧时长（毫秒），默认60ms
+        """
+        self.frame_duration = frame_duration
+        self.queue = []
+        self.play_position = 0  # 虚拟播放位置（毫秒）
+        self.start_timestamp = None  # 开始时间戳（只读，不修改）
+        self.pending_send_task = None
+        self.logger = logger
+
+    def reset(self):
+        """重置控制器状态"""
+        if self.pending_send_task and not self.pending_send_task.done():
+            self.pending_send_task.cancel()
+            try:
+                # 等待任务取消完成
+                asyncio.get_event_loop().run_until_complete(self.pending_send_task)
+            except asyncio.CancelledError:
+                pass
+
+        self.queue.clear()
+        self.play_position = 0
+        self.start_timestamp = time.time()
+
+    def add_audio(self, opus_packet):
+        """添加音频包到队列"""
+        self.queue.append(("audio", opus_packet))
+
+    def _get_elapsed_ms(self):
+        """获取已经过的时间（毫秒）"""
+        if self.start_timestamp is None:
+            return 0
+        return (time.time() - self.start_timestamp) * 1000
+
+    async def check_queue(self, send_audio_callback):
+        """
+        检查队列并按时发送音频/消息
+
+        Args:
+            send_audio_callback: 发送音频的回调函数 async def(opus_packet)
+        """
+        if self.start_timestamp is None:
+            self.reset()
+
+        while self.queue:
+            item = self.queue[0]
+            item_type = item[0]
+
+            if item_type == "audio":
+                _, opus_packet = item
+
+                # 计算时间差
+                elapsed_ms = self._get_elapsed_ms()
+                output_ms = self.play_position
+
+                if elapsed_ms < output_ms:
+                    # 还不到发送时间，计算等待时长
+                    wait_ms = output_ms - elapsed_ms
+
+                    # 等待后继续检查（允许被中断）
+                    try:
+                        await asyncio.sleep(wait_ms / 1000)
+                    except asyncio.CancelledError:
+                        self.logger.bind(tag=TAG).debug("音频发送任务被取消")
+                        raise
+
+                    # 继续循环检查（时间可能已到）
+                    continue
+
+                # 时间已到，发送音频
+                self.queue.pop(0)
+                self.play_position += self.frame_duration
+
+                try:
+                    await send_audio_callback(opus_packet)
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"发送音频失败: {e}")
+                    raise
+
+
+    async def start_sending(self, send_audio_callback, send_message_callback=None):
+        """
+        启动异步发送任务
+
+        Args:
+            send_audio_callback: 发送音频的回调函数
+            send_message_callback: 发送消息的回调函数
+
+        Returns:
+            asyncio.Task: 发送任务
+        """
+        async def _send_loop():
+            try:
+                while True:
+                    await self.check_queue(send_audio_callback, send_message_callback)
+                    # 如果队列空了，短暂等待后再检查（避免 busy loop）
+                    await asyncio.sleep(0.01)
+            except asyncio.CancelledError:
+                self.logger.bind(tag=TAG).info("音频发送循环已停止")
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"音频发送循环异常: {e}")
+
+        self.pending_send_task = asyncio.create_task(_send_loop())
+        return self.pending_send_task
+
+    def stop_sending(self):
+        """停止发送任务"""
+        if self.pending_send_task and not self.pending_send_task.done():
+            self.pending_send_task.cancel()
+            self.logger.bind(tag=TAG).debug("已取消音频发送任务")

main/xiaozhi-server/core/utils/gc_manager.py

@@ -0,0 +1,122 @@
+"""
+全局GC管理模块
+定期执行垃圾回收，避免频繁触发GC导致的GIL锁问题
+"""
+
+import gc
+import asyncio
+import threading
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class GlobalGCManager:
+    """全局垃圾回收管理器"""
+
+    def __init__(self, interval_seconds=300):
+        """
+        初始化GC管理器
+
+        Args:
+            interval_seconds: GC执行间隔（秒），默认300秒（5分钟）
+        """
+        self.interval_seconds = interval_seconds
+        self._task = None
+        self._stop_event = asyncio.Event()
+        self._lock = threading.Lock()
+
+    async def start(self):
+        """启动定时GC任务"""
+        if self._task is not None:
+            logger.bind(tag=TAG).warning("GC管理器已经在运行")
+            return
+
+        logger.bind(tag=TAG).info(f"启动全局GC管理器，间隔{self.interval_seconds}秒")
+        self._stop_event.clear()
+        self._task = asyncio.create_task(self._gc_loop())
+
+    async def stop(self):
+        """停止定时GC任务"""
+        if self._task is None:
+            return
+
+        logger.bind(tag=TAG).info("停止全局GC管理器")
+        self._stop_event.set()
+
+        if self._task and not self._task.done():
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+
+        self._task = None
+
+    async def _gc_loop(self):
+        """GC循环任务"""
+        try:
+            while not self._stop_event.is_set():
+                # 等待指定间隔
+                try:
+                    await asyncio.wait_for(
+                        self._stop_event.wait(), timeout=self.interval_seconds
+                    )
+                    # 如果stop_event被设置，退出循环
+                    break
+                except asyncio.TimeoutError:
+                    # 超时表示到了执行GC的时间
+                    pass
+
+                # 执行GC
+                await self._run_gc()
+
+        except asyncio.CancelledError:
+            logger.bind(tag=TAG).info("GC循环任务被取消")
+            raise
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"GC循环任务异常: {e}")
+        finally:
+            logger.bind(tag=TAG).info("GC循环任务已退出")
+
+    async def _run_gc(self):
+        """执行垃圾回收"""
+        try:
+            # 在线程池中执行GC，避免阻塞事件循环
+            loop = asyncio.get_running_loop()
+
+            def do_gc():
+                with self._lock:
+                    before = len(gc.get_objects())
+                    collected = gc.collect()
+                    after = len(gc.get_objects())
+                    return before, collected, after
+
+            before, collected, after = await loop.run_in_executor(None, do_gc)
+            logger.bind(tag=TAG).debug(
+                f"全局GC执行完成 - 回收对象: {collected}, "
+                f"对象数量: {before} -> {after}"
+            )
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"执行GC时出错: {e}")
+
+
+# 全局单例
+_gc_manager_instance = None
+
+
+def get_gc_manager(interval_seconds=300):
+    """
+    获取全局GC管理器实例（单例模式）
+
+    Args:
+        interval_seconds: GC执行间隔（秒），默认300秒（5分钟）
+
+    Returns:
+        GlobalGCManager实例
+    """
+    global _gc_manager_instance
+    if _gc_manager_instance is None:
+        _gc_manager_instance = GlobalGCManager(interval_seconds)
+    return _gc_manager_instance

main/xiaozhi-server/core/websocket_server.py

@@ -4,7 +4,7 @@ import json
 import websockets
 from config.logger import setup_logging
 from core.connection import ConnectionHandler
-from config.config_loader import get_config_from_api
+from config.config_loader import get_config_from_api_async
 from core.auth import AuthManager, AuthenticationError
 from core.utils.modules_initialize import initialize_modules
 from core.utils.util import check_vad_update, check_asr_update
@@ -133,8 +133,8 @@ class WebSocketServer:
         """
         try:
             async with self.config_lock:
-                # 重新获取配置
-                new_config = get_config_from_api(self.config)
+                # 重新获取配置（使用异步版本）
+                new_config = await get_config_from_api_async(self.config)
                 if new_config is None:
                     self.logger.bind(tag=TAG).error("获取新配置失败")
                     return False