diff --git a/main/xiaozhi-server/core/auth.py b/main/xiaozhi-server/core/auth.py index 98bd3235..011f7efb 100755 --- a/main/xiaozhi-server/core/auth.py +++ b/main/xiaozhi-server/core/auth.py @@ -12,8 +12,9 @@ class AuthenticationError(Exception): class AuthMiddleware: """ - 认证中间件（兼容旧版命名） + 认证中间件 用于 WebSocket/MQTT 连接认证 + 集成 AuthManager 的 token 验证逻辑，支持多种认证方式 """ def __init__(self, config: dict): @@ -24,18 +25,34 @@ class AuthMiddleware: config: 配置字典，包含认证相关配置 """ self.config = config - auth_config = config.get("server", {}).get("auth", {}) + server_config = config.get("server", {}) + auth_config = server_config.get("auth", {}) + self.enabled = auth_config.get("enabled", False) self.tokens = auth_config.get("tokens", []) - self.allowed_devices = auth_config.get("allowed_devices", []) + self.allowed_devices = set(auth_config.get("allowed_devices", [])) + + # 获取 auth_key 用于 HMAC token 验证 + self.auth_key = server_config.get("auth_key", "") + expire_seconds = auth_config.get("expire_seconds", None) + + # 创建 AuthManager 实例用于 HMAC token 验证 + if self.auth_key: + self._auth_manager = AuthManager( + secret_key=self.auth_key, + expire_seconds=expire_seconds + ) + else: + self._auth_manager = None - def authenticate(self, device_id: str, token: str = None) -> bool: + def authenticate(self, device_id: str, token: str = None, client_id: str = None) -> bool: """ - 验证设备认证 + 验证设备认证（同步方法） Args: device_id: 设备 ID - token: 认证令牌 + token: 认证令牌（可以是静态 token 或 HMAC token） + client_id: 客户端 ID（用于 HMAC token 验证） Returns: bool: 认证是否通过 @@ -43,18 +60,89 @@ class AuthMiddleware: if not self.enabled: return True - # 检查白名单 - if device_id in self.allowed_devices: + # 1. 检查白名单 + if device_id and device_id in self.allowed_devices: return True - # 检查 token + # 2. 检查静态 token if token: + # 移除 Bearer 前缀（如果有） + if token.startswith("Bearer "): + token = token[7:] + for token_config in self.tokens: if token_config.get("token") == token: return True + # 3. 检查 HMAC token（需要 AuthManager） + if token and self._auth_manager and client_id and device_id: + if self._auth_manager.verify_token(token, client_id, device_id): + return True + return False + async def authenticate_async(self, headers: dict) -> bool: + """ + 从 headers 中提取信息并进行异步认证 + + Args: + headers: HTTP 请求头字典 + + Returns: + bool: 认证是否通过 + + Raises: + AuthenticationError: 认证失败时抛出 + """ + if not self.enabled: + return True + + device_id = headers.get("device-id") + client_id = headers.get("client-id") + authorization = headers.get("authorization", "") + + # 提取 token + token = None + if authorization: + if authorization.startswith("Bearer "): + token = authorization[7:] + else: + token = authorization + + # 执行认证 + if self.authenticate(device_id, token, client_id): + return True + + raise AuthenticationError(f"认证失败: device_id={device_id}") + + def authenticate_websocket(self, websocket) -> bool: + """ + WebSocket 连接认证 + + Args: + websocket: WebSocket 连接对象 + + Returns: + bool: 认证是否通过 + """ + if not self.enabled: + return True + + headers = dict(websocket.request.headers) + device_id = headers.get("device-id") + client_id = headers.get("client-id") + authorization = headers.get("authorization", "") + + # 提取 token + token = None + if authorization: + if authorization.startswith("Bearer "): + token = authorization[7:] + else: + token = authorization + + return self.authenticate(device_id, token, client_id) + class AuthManager: """ diff --git a/main/xiaozhi-server/core/services/connection_service.py b/main/xiaozhi-server/core/services/connection_service.py index e4204cf5..798f527b 100644 --- a/main/xiaozhi-server/core/services/connection_service.py +++ b/main/xiaozhi-server/core/services/connection_service.py @@ -1,4 +1,6 @@ import asyncio +import time +import threading from typing import Dict, Any from urllib.parse import parse_qs, urlparse @@ -9,8 +11,12 @@ from core.components.component_manager import ComponentType from core.pipeline.message_pipeline import MessagePipeline from core.processors.message_router import MessageRouter from config.logger import setup_logging +from config.config_loader import get_private_config_from_api +from config.manage_api_client import DeviceNotFoundException, DeviceBindException +from core.utils.util import check_vad_update, check_asr_update, filter_sensitive_info logger = setup_logging() +TAG = __name__ class ConnectionService: @@ -42,11 +48,11 @@ class ConnectionService: # 设置transport接口 context.transport = transport - # 传入共享 ASR 管理器（如果有） + # 传入共享 ASR 管理器 # 这使得 ASRAdapter 可以使用预加载的模型实例 if '_shared_asr_manager' in self.config: context.shared_asr_manager = self.config['_shared_asr_manager'] - logger.debug("连接使用共享 ASR 实例") + logger.bind(tag=TAG).debug("连接使用共享 ASR 实例") # 兼容性：设置websocket属性（如果transport是WebSocket） if hasattr(transport, '_websocket'): @@ -58,14 +64,23 @@ class ConnectionService: # 创建组件管理器 component_manager = ComponentRegistry.create_component_manager(self.config) + # 设置绑定检查事件 + bind_completed_event = asyncio.Event() + last_bind_prompt_time = 0 + bind_prompt_interval = 60 # 绑定提示播放间隔(秒) + # 启动超时检查任务 timeout_task = None + # 后台初始化任务 + init_task = None try: - logger.info(f"新连接建立: {context.device_id} from {context.client_ip}") + logger.bind(tag=TAG).info(f"新连接建立: {context.device_id} from {context.client_ip}") - # 初始化必要的组件 - await self._initialize_components(context, component_manager) + # 在后台初始化配置和组件（非阻塞） + init_task = asyncio.create_task( + self._background_initialize(context, component_manager, bind_completed_event) + ) # 启动超时检查 timeout_task = asyncio.create_task(self._check_timeout(context, transport)) @@ -73,13 +88,22 @@ class ConnectionService: # 处理消息流 async for message in transport.receive(): try: + # 检查绑定状态 + should_process = await self._check_bind_status( + context, transport, bind_completed_event, + last_bind_prompt_time, bind_prompt_interval + ) + if not should_process: + last_bind_prompt_time = time.time() + continue + await self.message_pipeline.process_message(context, transport, message) except Exception as e: - logger.error(f"处理消息时出错: {e}") + logger.bind(tag=TAG).error(f"处理消息时出错: {e}") # 继续处理其他消息，不中断连接 except Exception as e: - logger.error(f"连接处理出错: {e}") + logger.bind(tag=TAG).error(f"连接处理出错: {e}") finally: # 清理资源 if timeout_task and not timeout_task.done(): @@ -89,38 +113,42 @@ class ConnectionService: except asyncio.CancelledError: pass + if init_task and not init_task.done(): + init_task.cancel() + try: + await init_task + except asyncio.CancelledError: + pass + + # 保存记忆（异步，不阻塞关闭） + await self._save_memory_async(context) + # 清理组件 try: await component_manager.cleanup_all() except Exception as e: - logger.error(f"组件清理失败: {e}") + logger.bind(tag=TAG).error(f"组件清理失败: {e}") # 执行会话清理回调 try: await context.run_cleanup() except Exception as e: - logger.error(f"会话清理失败: {e}") + logger.bind(tag=TAG).error(f"会话清理失败: {e}") # 关闭传输层 try: await transport.close() except Exception as e: - logger.error(f"关闭传输层失败: {e}") + logger.bind(tag=TAG).error(f"关闭传输层失败: {e}") - logger.info(f"连接已关闭: {context.device_id}") + logger.bind(tag=TAG).info(f"连接已关闭: {context.device_id}") async def _extract_device_info(self, context: SessionContext, headers: Dict[str, str]): - """从headers或URL参数中提取设备信息""" - device_id = headers.get("device-id") - client_id = headers.get("client-id") - - # 如果headers中没有device-id，尝试从URL参数中获取 - if not device_id: - # 这里需要从WebSocket请求中获取路径信息 - # 暂时使用占位符，后续在WebSocketServer中传入 - pass - - context.device_id = device_id + """ + 从headers中提取设备信息 + """ + context.device_id = headers.get("device-id") + context.headers["client-id"] = headers.get("client-id") context.client_ip = headers.get("x-real-ip") or headers.get("x-forwarded-for", "unknown") if context.client_ip and "," in context.client_ip: @@ -133,20 +161,177 @@ class ConnectionService: context.component_manager = component_manager # 根据配置确定需要初始化的组件 - required_components = ComponentRegistry.get_required_components(self.config) + required_components = ComponentRegistry.get_required_components(context.config) # 按需初始化组件 for component_type in required_components: component = await component_manager.get_component(component_type, context) if component: - logger.info(f"组件初始化成功: {component_type.value}") + logger.bind(tag=TAG).info(f"组件初始化成功: {component_type.value}") else: - logger.warning(f"组件初始化失败: {component_type.value}") + logger.bind(tag=TAG).warning(f"组件初始化失败: {component_type.value}") except Exception as e: - logger.error(f"组件初始化出错: {e}") + logger.bind(tag=TAG).error(f"组件初始化出错: {e}") raise + async def _background_initialize( + self, + context: SessionContext, + component_manager, + bind_completed_event: asyncio.Event + ): + """在后台初始化配置和组件""" + try: + await self._initialize_private_config(context, bind_completed_event) + await self._initialize_components(context, component_manager) + + except Exception as e: + logger.bind(tag=TAG).error(f"后台初始化失败: {e}") + # 即使初始化失败，也要设置绑定完成事件，避免消息一直被丢弃 + bind_completed_event.set() + + async def _initialize_private_config( + self, + context: SessionContext, + bind_completed_event: asyncio.Event + ): + """从API异步获取差异化配置""" + if not context.read_config_from_api: + context.need_bind = False + bind_completed_event.set() + return + + try: + begin_time = time.time() + + # 在线程池中执行同步API调用 + loop = asyncio.get_event_loop() + private_config = await loop.run_in_executor( + None, + get_private_config_from_api, + context.config, + context.device_id, + context.headers.get("client-id", context.device_id) + ) + + if private_config: + private_config["delete_audio"] = bool(context.config.get("delete_audio", True)) + logger.bind(tag=TAG).info( + f"{time.time() - begin_time:.2f}秒，获取差异化配置成功" + ) + + # 检查是否需要更新 VAD/ASR + init_vad = check_vad_update(context.common_config, private_config) + init_asr = check_asr_update(context.common_config, private_config) + + # 合并私有配置 + context.private_config = private_config + context.config.update(private_config) + + context.need_bind = False + else: + context.need_bind = False + + bind_completed_event.set() + + except DeviceNotFoundException: + logger.bind(tag=TAG).warning(f"设备 {context.device_id} 未找到，需要绑定") + context.need_bind = True + bind_completed_event.set() + except DeviceBindException as e: + logger.bind(tag=TAG).warning(f"设备绑定异常: {e}") + context.need_bind = True + context.bind_code = getattr(e, 'bind_code', None) + bind_completed_event.set() + except Exception as e: + logger.bind(tag=TAG).error(f"获取差异化配置失败: {e}") + context.need_bind = True + bind_completed_event.set() + + async def _check_bind_status( + self, + context: SessionContext, + transport: TransportInterface, + bind_completed_event: asyncio.Event, + last_bind_prompt_time: float, + bind_prompt_interval: int + ) -> bool: + """ + 检查设备绑定状态 + + Returns: + bool: True 表示可以处理消息，False 表示需要丢弃消息 + """ + # 如果还没获取到真实绑定状态，等待一下 + if not bind_completed_event.is_set(): + try: + await asyncio.wait_for(bind_completed_event.wait(), timeout=1) + except asyncio.TimeoutError: + # 超时仍未获取到真实状态，丢弃消息并提示绑定 + await self._prompt_bind_if_needed( + context, transport, last_bind_prompt_time, bind_prompt_interval + ) + return False + + # 检查是否需要绑定 + if context.need_bind: + await self._prompt_bind_if_needed( + context, transport, last_bind_prompt_time, bind_prompt_interval + ) + return False + + return True + + async def _prompt_bind_if_needed( + self, + context: SessionContext, + transport: TransportInterface, + last_prompt_time: float, + prompt_interval: int + ): + """如果需要，播放绑定提示""" + current_time = time.time() + if current_time - last_prompt_time >= prompt_interval: + try: + # 复用现有的绑定提示逻辑 + from core.handle.receiveAudioHandle import check_bind_device + asyncio.create_task(check_bind_device(context)) + except Exception as e: + logger.bind(tag=TAG).error(f"播放绑定提示失败: {e}") + + async def _save_memory_async(self, context: SessionContext): + """异步保存记忆（不阻塞连接关闭）""" + try: + # 获取 memory 组件 + memory = None + if context.component_manager: + memory = context.component_manager.get("memory") + + if memory and context.dialogue and hasattr(memory, 'save_memory'): + # 在线程中异步保存，不等待完成 + def save_memory_task(): + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete( + memory.save_memory(context.dialogue.dialogue, context.session_id) + ) + except Exception as e: + logger.bind(tag=TAG).error(f"保存记忆失败: {e}") + finally: + try: + loop.close() + except Exception: + pass + + # 启动线程保存记忆，不等待完成 + threading.Thread(target=save_memory_task, daemon=True).start() + logger.bind(tag=TAG).debug("记忆保存任务已启动") + + except Exception as e: + logger.bind(tag=TAG).error(f"启动记忆保存失败: {e}") + async def _check_timeout(self, context: SessionContext, transport: TransportInterface): """定期检查连接超时""" timeout_seconds = context.config.get("close_connection_no_voice_time", 120) @@ -157,11 +342,11 @@ class ConnectionService: await asyncio.sleep(check_interval) if context.is_timeout(timeout_seconds): - logger.info(f"连接超时，关闭连接: {context.session_id}") + logger.bind(tag=TAG).info(f"连接超时，关闭连接: {context.session_id}") await transport.close() break except asyncio.CancelledError: pass except Exception as e: - logger.error(f"超时检查出错: {e}") + logger.bind(tag=TAG).error(f"超时检查出错: {e}") diff --git a/main/xiaozhi-server/core/websocket_server_new.py b/main/xiaozhi-server/core/websocket_server_new.py index 238e8ba9..3cca3284 100644 --- a/main/xiaozhi-server/core/websocket_server_new.py +++ b/main/xiaozhi-server/core/websocket_server_new.py @@ -1,13 +1,42 @@ import asyncio +import logging import websockets from typing import Dict, Any from config.logger import setup_logging from core.services.connection_service import ConnectionService from core.transport.websocket_transport import WebSocketTransport -from config.config_loader import get_config_from_api +from config.config_loader import get_config_from_api_async from core.utils.util import check_vad_update, check_asr_update +from core.auth import AuthMiddleware, AuthenticationError + + +class SuppressInvalidHandshakeFilter(logging.Filter): + """过滤掉无效握手错误日志（如HTTPS访问WS端口）""" + + def filter(self, record): + msg = record.getMessage() + suppress_keywords = [ + "opening handshake failed", + "did not receive a valid HTTP request", + "connection closed while reading HTTP request", + "line without CRLF", + ] + return not any(keyword in msg for keyword in suppress_keywords) + + +def _setup_websockets_logger(): + """配置 websockets 相关的所有 logger，过滤无效握手错误""" + filter_instance = SuppressInvalidHandshakeFilter() + for logger_name in ["websockets", "websockets.server", "websockets.client"]: + ws_logger = logging.getLogger(logger_name) + ws_logger.addFilter(filter_instance) + + +_setup_websockets_logger() + logger = setup_logging() +TAG = __name__ class NewWebSocketServer: @@ -26,6 +55,9 @@ class NewWebSocketServer: # 活跃连接管理 self.active_connections = set() + + # 认证中间件 + self.auth_middleware = AuthMiddleware(config) async def start(self): """启动WebSocket服务器""" @@ -33,7 +65,7 @@ class NewWebSocketServer: host = server_config.get("ip", "0.0.0.0") port = int(server_config.get("port", 8000)) - logger.info(f"启动新架构WebSocket服务器: {host}:{port}") + logger.bind(tag=TAG).info(f"启动新架构WebSocket服务器: {host}:{port}") async with websockets.serve( self._handle_connection, @@ -41,13 +73,29 @@ class NewWebSocketServer: port, process_request=self._http_response ): - logger.info("WebSocket服务器启动成功") + logger.bind(tag=TAG).info("WebSocket服务器启动成功") await asyncio.Future() # 保持服务器运行 async def _handle_connection(self, websocket): """处理新连接 - 使用新架构""" # 提取连接头信息 headers = self._extract_headers(websocket) + device_id = headers.get('device-id') + + # 如果没有 device-id，提示并关闭连接 + if not device_id: + await websocket.send("端口正常，如需测试连接，请使用test_page.html") + await websocket.close() + return + + # 连接时认证 + try: + await self._handle_auth(headers) + except AuthenticationError as e: + logger.bind(tag=TAG).warning(f"认证失败: {e}") + await websocket.send("认证失败") + await websocket.close() + return # 创建WebSocket传输层 transport = WebSocketTransport(websocket) @@ -56,15 +104,17 @@ class NewWebSocketServer: self.active_connections.add(transport) try: - logger.info(f"新连接建立: {headers.get('device-id', 'unknown')} from {headers.get('x-real-ip', 'unknown')}") + logger.bind(tag=TAG).info( + f"新连接建立: {device_id} from {headers.get('x-real-ip', 'unknown')}" + ) # 使用ConnectionService处理连接 await self.connection_service.handle_connection(transport, headers) except websockets.exceptions.ConnectionClosed: - logger.info("WebSocket连接正常关闭") + logger.bind(tag=TAG).info("WebSocket连接正常关闭") except Exception as e: - logger.error(f"处理WebSocket连接时出错: {e}", exc_info=True) + logger.bind(tag=TAG).error(f"处理WebSocket连接时出错: {e}", exc_info=True) finally: # 确保从活动连接集合中移除 self.active_connections.discard(transport) @@ -76,32 +126,73 @@ class NewWebSocketServer: elif hasattr(websocket, "state") and websocket.state.name != "CLOSED": await websocket.close() except Exception as close_error: - logger.error(f"强制关闭WebSocket连接时出错: {close_error}") + logger.bind(tag=TAG).error(f"强制关闭WebSocket连接时出错: {close_error}") + + async def _handle_auth(self, headers: Dict[str, str]): + """ + 连接时认证 + + Args: + headers: HTTP 请求头 + + Raises: + AuthenticationError: 认证失败时抛出 + """ + await self.auth_middleware.authenticate_async(headers) def _extract_headers(self, websocket) -> Dict[str, str]: - """从WebSocket请求中提取头信息""" + """ + 从WebSocket请求中提取头信息 + + 支持从以下来源提取信息： + 1. HTTP 请求头 + 2. URL 查询参数（device-id, client-id, authorization） + 3. 路径参数（如 ?from=mqtt_gateway） + """ headers = {} - # 提取请求头 - if hasattr(websocket, 'request_headers'): + # 1. 提取 HTTP 请求头 + if hasattr(websocket, 'request') and hasattr(websocket.request, 'headers'): + for name, value in websocket.request.headers.items(): + headers[name.lower()] = value + elif hasattr(websocket, 'request_headers'): for name, value in websocket.request_headers.items(): headers[name.lower()] = value - # 提取路径参数（如果有的话） - if hasattr(websocket, 'path'): - # 可以从路径中提取device-id等参数 - # 例如: /ws?device-id=xxx&client-id=yyy + # 2. 提取路径参数（如果有的话） + request_path = None + if hasattr(websocket, 'request') and hasattr(websocket.request, 'path'): + request_path = websocket.request.path + elif hasattr(websocket, 'path'): + request_path = websocket.path + + if request_path: from urllib.parse import urlparse, parse_qs - parsed = urlparse(websocket.path) + parsed = urlparse(request_path) query_params = parse_qs(parsed.query) + # 处理关键参数：device-id, client-id, authorization + key_params = ['device-id', 'client-id', 'authorization'] + for key in key_params: + if key in query_params and query_params[key]: + # URL 参数优先级低于 header + if key not in headers or not headers[key]: + headers[key] = query_params[key][0] + + # 处理其他参数 for key, values in query_params.items(): - if values: + if values and key not in headers: headers[key] = values[0] + + # 检查是否来自 MQTT 网关 + if request_path.endswith("?from=mqtt_gateway") or "from=mqtt_gateway" in request_path: + headers['from_mqtt_gateway'] = 'true' - # 提取远程地址 + # 3. 提取远程地址 if hasattr(websocket, 'remote_address'): - headers['x-real-ip'] = websocket.remote_address[0] + # 如果 headers 中没有 x-real-ip，使用 remote_address + if 'x-real-ip' not in headers: + headers['x-real-ip'] = websocket.remote_address[0] return headers @@ -124,53 +215,84 @@ class NewWebSocketServer: """ try: async with self.config_lock: - logger.info("开始更新服务器配置") + logger.bind(tag=TAG).info("开始更新服务器配置") - # 重新获取配置 - new_config = get_config_from_api(self.config) + # 异步获取新配置 + new_config = await get_config_from_api_async(self.config) if new_config is None: - logger.error("获取新配置失败") + logger.bind(tag=TAG).error("获取新配置失败") return False - logger.info("获取新配置成功") + logger.bind(tag=TAG).info("获取新配置成功") - # 检查配置变化 - config_changed = self._check_config_changes(self.config, new_config) + # 检查 VAD 和 ASR 类型是否需要更新 + update_vad = check_vad_update(self.config, new_config) + update_asr = check_asr_update(self.config, new_config) + logger.bind(tag=TAG).info( + f"检查VAD和ASR类型是否需要更新: VAD={update_vad}, ASR={update_asr}" + ) + + # 检查配置是否有重大变化 + changed_configs = self._get_changed_configs(self.config, new_config) + + # 保存旧配置引用 + old_config = self.config # 更新配置 self.config = new_config - # 重新创建连接服务（如果配置有重大变化） - if config_changed: - logger.info("配置有重大变化，重新创建连接服务") + # 根据变化类型进行更新 + if changed_configs: + logger.bind(tag=TAG).info(f"配置项变化: {', '.join(changed_configs)}") + + # 重新创建连接服务，使用新配置 + # 注意：已建立的连接会继续使用旧配置，只有新连接使用新配置 self.connection_service = ConnectionService(new_config) + + # 如果 ASR 配置变化且使用共享 ASR 管理器，需要特殊处理 + if update_asr and '_shared_asr_manager' in old_config: + logger.bind(tag=TAG).warning( + "ASR 配置已变化，但共享 ASR 管理器需要重启服务才能更新" + ) + else: + # 即使没有重大变化，也更新 ConnectionService 的配置引用 + self.connection_service.config = new_config - logger.info("配置更新任务执行完毕") + # 更新认证中间件 + self.auth_middleware = AuthMiddleware(new_config) + + logger.bind(tag=TAG).info("配置更新任务执行完毕") return True except Exception as e: - logger.error(f"更新服务器配置失败: {str(e)}", exc_info=True) + logger.bind(tag=TAG).error(f"更新服务器配置失败: {str(e)}", exc_info=True) return False - def _check_config_changes(self, old_config: Dict[str, Any], new_config: Dict[str, Any]) -> bool: - """检查配置是否有重大变化""" - # 检查关键配置项是否变化 + def _get_changed_configs(self, old_config: Dict[str, Any], new_config: Dict[str, Any]) -> list: + """ + 获取变化的配置项列表 + + Returns: + list: 变化的配置项名称列表 + """ + changed = [] key_configs = [ "selected_module", - "vad", - "asr", - "llm", - "tts", - "memory", - "intent" + "VAD", + "ASR", + "LLM", + "TTS", + "Memory", + "Intent" ] for key in key_configs: - if old_config.get(key) != new_config.get(key): - logger.info(f"配置项 {key} 发生变化") - return True + old_value = old_config.get(key) + new_value = new_config.get(key) + if old_value != new_value: + changed.append(key) - return False + return changed def get_active_connections_count(self) -> int: """获取活跃连接数""" diff --git a/main/xiaozhi-server/core/xiaozhi_server_facade.py b/main/xiaozhi-server/core/xiaozhi_server_facade.py index 8579e251..fd8ba966 100644 --- a/main/xiaozhi-server/core/xiaozhi_server_facade.py +++ b/main/xiaozhi-server/core/xiaozhi_server_facade.py @@ -226,7 +226,7 @@ class XiaozhiServerFacade: if self.multi_protocol_server: await self.multi_protocol_server.stop() - # 关闭共享 ASR 管理器（优雅停机） + # 关闭共享 ASR 管理器 if self.shared_asr_manager: logger.bind(tag=TAG).info("正在关闭共享 ASR 管理器...") await self.shared_asr_manager.shutdown()