小智终端 5种 场景模式 OK

wangzhibo

2026-09-11 659c470eba542383987aad227b7f918100f131fa

 xiaozhi-esp32-server/main/xiaozhi-server/core/connection.py

@@ -45,6 +45,7 @@
from core.utils.voiceprint_provider import VoiceprintProvider
from core.utils.util import get_system_error_response
from core.utils import textUtils
from core.websocket_handshake import is_client_hello, is_priority_listen_detect


TAG = __name__
@@ -122,6 +123,7 @@
        self.loop = None  # 在 handle_connection 中获取运行中的事件循环
        self.stop_event = threading.Event()
        self.executor = ThreadPoolExecutor(max_workers=5)
        self._tts_channel_lock = threading.Lock()

        # 添加上报线程池
        self.report_queue = queue.Queue()
@@ -209,6 +211,18 @@
        self.tool_allowlist = []
        self.original_system_prompt = None
        self.awaiting_voice_app_choice = False
        self.shadowing_index = 0
        self.shadowing_prompt = ""
        self.shadowing_clips = []
        self.shadowing_awaiting_user = False
        self.shadowing_clip_gen = 0
        self.shadowing_timeout_task = None
        self.shadowing_assess_index = 0
        self.last_asr_language = None
        self.task_session = None
        self.last_asr_audio_snapshot = None
        self.shadowing_listen_pcm = []
        self.shadowing_audio_saved_index = -1

    async def handle_connection(self, ws: websockets.ServerConnection):
        try:
@@ -232,6 +246,8 @@

            # 认证通过,继续处理
            self.websocket = ws
            if self.server is not None:
                await self.server.register_device_handler(self.device_id, self)

            # 检查是否来自MQTT连接
            request_path = ws.request.path
@@ -249,7 +265,7 @@
            # 启动AEC缓存清理任务
            self._aec_cache_cleanup_task = asyncio.create_task(self._check_aec_cache_expiry())

            self.welcome_msg = self.config["xiaozhi"]
            self.welcome_msg = copy.deepcopy(self.config.get("xiaozhi") or {})
            self.welcome_msg["session_id"] = self.session_id

            # 从配置中读取采样率
@@ -354,6 +370,12 @@

    async def _route_message(self, message):
        """消息路由"""
        # hello 必须立刻应答。绑定状态还在拉配置时如果丢掉 hello，
        # 设备会卡在 handshake/hello 等待，表现为连不上语音服务。
        if is_client_hello(message) or is_priority_listen_detect(message):
            await handleTextMessage(self, message)
            return

        # 检查是否已经获取到真实的绑定状态
        if not self.bind_completed_event.is_set():
            # 还没有获取到真实状态，等待直到获取到真实状态或超时
@@ -610,14 +632,45 @@
                )
            )

    def _initialize_components(self):
        try:
    def _tts_config_ready(self) -> bool:
        selected = (self.config.get("selected_module") or {}).get("TTS")
        providers = self.config.get("TTS") or {}
        return bool(selected) and selected in providers

    async def _open_tts_channels(self):
        """Open TTS worker threads. Must run on the connection event loop after TTS config exists."""
        if self.tts is None and not self._tts_config_ready():
            self.logger.bind(tag=TAG).info("TTS配置尚未从智控台加载，延后打开通道")
            return
        with self._tts_channel_lock:
            if self.tts is None:
                self.tts = self._initialize_tts()
            # 打开语音合成通道
            asyncio.run_coroutine_threadsafe(
                self.tts.open_audio_channels(self), self.loop
            )
            tts = self.tts
        if tts is None:
            self.logger.bind(tag=TAG).warning("TTS实例不可用，无法打开通道")
            return
        await tts.open_audio_channels(self)
        play = getattr(tts, "audio_play_priority_thread", None)
        text_th = getattr(tts, "tts_priority_thread", None)
        if play and play.is_alive() and text_th and text_th.is_alive():
            self.logger.bind(tag=TAG).info("TTS通道已打开")
        else:
            self.logger.bind(tag=TAG).error("TTS通道打开后工作线程未运行")

    def _ensure_tts_channels(self):
        """Start TTS worker threads once; used after private-config init."""
        try:
            if self.loop is None:
                return
            fut = asyncio.run_coroutine_threadsafe(self._open_tts_channels(), self.loop)
            fut.result(timeout=15)
        except Exception as e:
            self.logger.bind(tag=TAG).error(f"打开TTS通道失败: {e}")

    def _initialize_components(self):
        try:
            if self.tts is not None:
                self._ensure_tts_channels()
            if self.need_bind:
                self.bind_completed_event.set()
                return
@@ -755,8 +808,11 @@
    def _initialize_tts(self):
        """初始化TTS"""
        tts = None
        if not self.need_bind:
        if not self.need_bind and self._tts_config_ready():
            tts = initialize_tts(self.config)
        elif not self._tts_config_ready():
            self.logger.bind(tag=TAG).info("跳过TTS初始化：配置中没有可用的TTS模块")
            return None

        if tts is None:
            tts = DefaultTTS(self.config, delete_audio_file=True)
@@ -954,6 +1010,11 @@
            modules = {}
        if modules.get("tts", None) is not None:
            self.tts = modules["tts"]
        if self.tts is not None:
            try:
                await self._open_tts_channels()
            except Exception as e:
                self.logger.bind(tag=TAG).error(f"打开TTS通道失败: {e}")
        if modules.get("vad", None) is not None:
            self.vad = modules["vad"]
        if modules.get("asr", None) is not None:
@@ -1534,11 +1595,11 @@

        self.logger.bind(tag=TAG).info("聊天记录上报线程已退出")

    def _process_report(self, type, text, audio_data, report_time):
    def _process_report(self, type, text, audio_data, report_time, task_instance_id=None, step_index=None):
        """处理上报任务"""
        try:
            # 执行异步上报（在事件循环中运行）
            asyncio.run(report(self, type, text, audio_data, report_time))
            asyncio.run(report(self, type, text, audio_data, report_time, task_instance_id, step_index))
        except Exception as e:
            self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
        finally:
@@ -1552,6 +1613,15 @@
    async def close(self, ws=None):
        """资源清理方法"""
        try:
            if self.server is not None:
                await self.server.unregister_device_handler(self.device_id, self)
            try:
                from core.app_runtime.session import end_session

                end_session(self, "disconnected")
            except Exception as session_error:
                self.logger.bind(tag=TAG).warning(f"断线 flush TaskSession 失败: {session_error}")

            # 清理 VAD 连接资源
            if (
                    hasattr(self, "vad")