update: 增加唤醒时声纹处理，1秒内发送至大模型【需优化唤醒锁机制，中途会遭受打断(偶发)，考虑忽略检测】

2026-07-22 07:03:53 +08:00 · 2025-08-28 17:58:23 +08:00
parent 91cd843cfe
commit 41b8fac3aa
10 changed files with 126 additions and 519 deletions
@@ -61,8 +61,6 @@ delete_audio: true
 close_connection_no_voice_time: 120
 # TTS请求超时时间(秒)
 tts_timeout: 10
-# 开启唤醒词加速
-enable_wakeup_words_response_cache: true
 # 开场是否回复唤醒词
 enable_greeting: true
 # 说完话是否开启提示音
@@ -1,16 +1,5 @@
-import time
 import json
-import random
 import asyncio
-from core.utils.dialogue import Message
-from core.providers.tts.dto.dto import SentenceType
-from core.utils.wakeup_word import WakeupWordsConfig
-from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message
-from core.utils.util import (
-    audio_to_data_stream,
-    remove_punctuation_and_length, 
-    opus_datas_to_wav_bytes
-)
 from core.providers.tools.device_mcp import (
    MCPClient,
    send_mcp_initialize_message,
@@ -19,18 +8,6 @@ from core.providers.tools.device_mcp import (

 TAG = __name__

-WAKEUP_CONFIG = {
-    "refresh_time": 5,
-    "words": ["你好", "你好啊", "嘿，你好", "嗨"],
-}
-
-# 创建全局的唤醒词配置管理器
-wakeup_words_config = WakeupWordsConfig()
-
-# 用于防止并发调用wakeupWordsResponse的锁
-_wakeup_response_lock = asyncio.Lock()
-
-
 async def handleHelloMessage(conn, msg_json):
    """处理hello消息"""
    audio_params = msg_json.get("audio_params")
@@ -51,100 +28,4 @@ async def handleHelloMessage(conn, msg_json):
            # 发送mcp消息，获取tools列表
            asyncio.create_task(send_mcp_tools_list_request(conn))

-    await conn.websocket.send(json.dumps(conn.welcome_msg))
-
-
-async def checkWakeupWords(conn, text):
-    enable_wakeup_words_response_cache = conn.config[
-        "enable_wakeup_words_response_cache"
-    ]
-
-    if not enable_wakeup_words_response_cache or not conn.tts:
-        return False
-
-    _, filtered_text = remove_punctuation_and_length(text)
-    if filtered_text not in conn.config.get("wakeup_words"):
-        return False
-
-    conn.just_woken_up = True
-    await send_stt_message(conn, text)
-
-    # 获取当前音色
-    voice = getattr(conn.tts, "voice", "default")
-    if not voice:
-        voice = "default"
-
-    # 获取唤醒词回复配置
-    response = wakeup_words_config.get_wakeup_response(voice)
-    if not response or not response.get("file_path"):
-        response = {
-            "voice": "default",
-            "file_path": "config/assets/wakeup_words.wav",
-            "time": 0,
-            "text": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦",
-        }
-    
-    # 获取音频数据
-    opus_packets = []
-    def handle_audio_frame(frame_data):
-        opus_packets.append(frame_data)
-    
-    audio_to_data_stream(response.get("file_path"), is_opus=True, callback=handle_audio_frame)
-
-    # 播放唤醒词回复
-    conn.client_abort = False
-
-    conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
-    await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))
-    await sendAudioMessage(conn, SentenceType.LAST, [], None)
-
-    # 补充对话
-    conn.dialogue.put(Message(role="assistant", content=response.get("text")))
-
-    # 检查是否需要更新唤醒词回复
-    if time.time() - response.get("time", 0) > WAKEUP_CONFIG["refresh_time"]:
-        if not _wakeup_response_lock.locked():
-            asyncio.create_task(wakeupWordsResponse(conn))
-    return True
-
-
-async def wakeupWordsResponse(conn):
-    if not conn.tts or not conn.llm or not conn.llm.response_no_stream:
-        return
-
-    try:
-        # 尝试获取锁，如果获取不到就返回
-        if not await _wakeup_response_lock.acquire():
-            return
-
-        # 生成唤醒词回复
-        wakeup_word = random.choice(WAKEUP_CONFIG["words"])
-        question = (
-            "此刻用户正在和你说```"
-            + wakeup_word
-            + "```。\n请你根据以上用户的内容进行20-30字回复。要符合系统设置的角色情感和态度，不要像机器人一样说话。\n"
-            + "请勿对这条内容本身进行任何解释和回应，请勿返回表情符号，仅返回对用户的内容的回复。"
-        )
-
-        result = conn.llm.response_no_stream(conn.config["prompt"], question)
-        if not result or len(result) == 0:
-            return
-
-        # 生成TTS音频
-        tts_result = await asyncio.to_thread(conn.tts.to_tts, result)
-        if not tts_result:
-            return
-
-        # 获取当前音色
-        voice = getattr(conn.tts, "voice", "default")
-
-        wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000)
-        file_path = wakeup_words_config.generate_file_path(voice)
-        with open(file_path, "wb") as f:
-            f.write(wav_bytes)
-        # 更新配置
-        wakeup_words_config.update_wakeup_response(voice, file_path, result)
-    finally:
-        # 确保在任何情况下都释放锁
-        if _wakeup_response_lock.locked():
-            _wakeup_response_lock.release()
+    await conn.websocket.send(json.dumps(conn.welcome_msg))
@@ -1,12 +1,11 @@
 import json
-import asyncio
 import uuid
-from core.handle.sendAudioHandle import send_stt_message
-from core.handle.helloHandle import checkWakeupWords
-from core.utils.util import remove_punctuation_and_length
-from core.providers.tts.dto.dto import ContentType
+import asyncio
 from core.utils.dialogue import Message
+from core.providers.tts.dto.dto import ContentType
 from plugins_func.register import Action, ActionResponse
+from core.handle.sendAudioHandle import send_stt_message
+from core.utils.util import remove_punctuation_and_length
 from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType

 TAG = __name__
@@ -24,13 +23,9 @@ async def handle_user_intent(conn, text):
        pass

    # 检查是否有明确的退出命令
-    filtered_text = remove_punctuation_and_length(text)[1]
+    _, filtered_text = remove_punctuation_and_length(text)
    if await check_direct_exit(conn, filtered_text):
        return True
-    
-    # 检查是否是唤醒词
-    if await checkWakeupWords(conn, filtered_text):
-        return True

    if conn.intent_type == "function_call":
        # 使用支持function calling的聊天方法,不再进行意图分析
@@ -1,26 +1,21 @@
 import time
 import json
-import asyncio
-from core.utils.util import audio_to_data_stream
 from core.handle.abortHandle import handleAbortMessage
 from core.handle.intentHandler import handle_user_intent
 from core.utils.output_counter import check_device_output_limit
+from core.utils.util import play_audio_frames, play_audio_response
 from core.handle.sendAudioHandle import send_stt_message, SentenceType

 TAG = __name__


 async def handleAudioMessage(conn, audio):
+    # 检查是否在唤醒处理锁定期内
+    if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
+        return
+    
    # 当前片段是否有人说话
    have_voice = conn.vad.is_vad(conn, audio)
-    # 如果设备刚刚被唤醒，短暂忽略VAD检测
-    if have_voice and hasattr(conn, "just_woken_up") and conn.just_woken_up:
-        have_voice = False
-        # 设置一个短暂延迟后恢复VAD检测
-        conn.asr_audio.clear()
-        if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
-            conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
-        return
    if have_voice:
        if conn.client_is_speaking:
            await handleAbortMessage(conn)
@@ -29,13 +24,6 @@ async def handleAudioMessage(conn, audio):
    # 接收音频
    await conn.asr.receive_audio(conn, audio, have_voice)

-
-async def resume_vad_detection(conn):
-    # 等待2秒后恢复VAD检测
-    await asyncio.sleep(1)
-    conn.just_woken_up = False
-
-
 async def startToChat(conn, text):
    # 检查输入是否是JSON格式（包含说话人信息）
    speaker_name = None
@@ -119,9 +107,7 @@ async def max_out_size(conn):
    text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！"
    await send_stt_message(conn, text)
    file_path = "config/assets/max_output_size.wav"
-    conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text))
-    play_audio_frames(conn, file_path)
-    conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
+    play_audio_response(conn, {"text": text, "file_path": file_path})
    conn.close_after_chat = True


@@ -156,17 +142,4 @@ async def check_bind_device(conn):
        text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
        await send_stt_message(conn, text)
        music_path = "config/assets/bind_not_found.wav"
-        conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text))
-        play_audio_frames(conn, music_path)
-        conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
-
-def play_audio_frames(conn, file_path):
-    """播放音频文件并处理发送帧数据"""
-    def handle_audio_frame(frame_data):
-        conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None))
-
-    audio_to_data_stream(
-        file_path,
-        is_opus=True,
-        callback=handle_audio_frame
-    )
+        play_audio_response(conn, {"text": text, "file_path": music_path})
@@ -30,7 +30,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text):


 # 播放音频
-async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60):
+async def sendAudio(conn, audios, pre_buffer=False):
    """
    发送单个opus包，支持流控
    Args:
@@ -38,7 +38,7 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60):
        opus_packet: 单个opus数据包
        pre_buffer: 快速发送音频
    """
-    if audios is None:
+    if audios is None or len(audios) == 0:
        return

    if isinstance(audios, bytes):
@@ -60,6 +60,7 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60):
                "start_time": time.perf_counter(),
            }

+        frame_duration=60
        flow_control = conn.audio_flow_control
        current_time = time.perf_counter()
        # 计算预期发送时间
@@ -76,38 +77,6 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60):
        # 更新流控状态
        flow_control["packet_count"] += 1
        flow_control["last_send_time"] = time.perf_counter()
-    else:
-        if audios is None or len(audios) == 0:
-            return
-        # 流控参数优化
-        frame_duration = 60  # 帧时长（毫秒），匹配 Opus 编码
-        start_time = time.perf_counter()
-        play_position = 0
-
-        # 执行预缓冲
-        pre_buffer_frames = min(3, len(audios))
-        for i in range(pre_buffer_frames):
-            await conn.websocket.send(audios[i])
-        remaining_audios = audios[pre_buffer_frames:]
-
-        # 播放剩余音频帧
-        for opus_packet in remaining_audios:
-            if conn.client_abort:
-                break
-
-            # 重置没有声音的状态
-            conn.last_activity_time = time.time() * 1000
-
-            # 计算预期发送时间
-            expected_time = start_time + (play_position / 1000)
-            current_time = time.perf_counter()
-            delay = expected_time - current_time
-            if delay > 0:
-                await asyncio.sleep(delay)
-
-            await conn.websocket.send(opus_packet)
-
-            play_position += frame_duration


 async def send_tts_message(conn, state, text=None):
@@ -1,13 +1,11 @@
 import json
-import time
 import asyncio
+from core.utils.util import filter_sensitive_info
 from core.handle.abortHandle import handleAbortMessage
 from core.handle.helloHandle import handleHelloMessage
-from core.handle.reportHandle import enqueue_asr_report
+from core.handle.receiveAudioHandle import handleAudioMessage
 from core.providers.tools.device_mcp import handle_mcp_message
-from core.handle.receiveAudioHandle import startToChat, handleAudioMessage
 from core.handle.sendAudioHandle import send_stt_message, send_tts_message
-from core.utils.util import remove_punctuation_and_length, filter_sensitive_info
 from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus

 TAG = __name__
@@ -46,14 +44,9 @@ async def handleTextMessage(conn, message):
                conn.client_have_voice = False
                conn.asr_audio.clear()
                if "text" in msg_json:
-                    conn.last_activity_time = time.time() * 1000
-                    original_text = msg_json["text"]  # 保留原始文本
-                    filtered_len, filtered_text = remove_punctuation_and_length(
-                        original_text
-                    )
-
+                    original_text = msg_json["text"]  # 保留设备上传的文本
                    # 识别是否是唤醒词
-                    is_wakeup_words = filtered_text in conn.config.get("wakeup_words")
+                    is_wakeup_words = original_text in conn.config.get("wakeup_words")
                    # 是否开启唤醒词回复
                    enable_greeting = conn.config.get("enable_greeting", True)

@@ -62,16 +55,10 @@ async def handleTextMessage(conn, message):
                        await send_stt_message(conn, original_text)
                        await send_tts_message(conn, "stop", None)
                        conn.client_is_speaking = False
-                    elif is_wakeup_words:
-                        conn.just_woken_up = True
-                        # 上报纯文字数据（复用ASR上报功能，但不提供音频数据）
-                        enqueue_asr_report(conn, "嘿，你好呀", [])
-                        await startToChat(conn, "嘿，你好呀")
                    else:
-                        # 上报纯文字数据（复用ASR上报功能，但不提供音频数据）
-                        enqueue_asr_report(conn, original_text, [])
-                        # 否则需要LLM对文字内容进行答复
-                        await startToChat(conn, original_text)
+                        # 检测到唤醒词，开始等待后续进行声纹识别
+                        conn.wakeup_mode = True
+                        conn.logger.bind(tag=TAG).info(f"检测到唤醒词~")
        elif msg_json["type"] == "iot":
            conn.logger.bind(tag=TAG).info(f"收到iot消息：{message}")
            if "descriptors" in msg_json:
@@ -1,14 +1,14 @@
 import os
+import io
 import wave
 import uuid
+import json
+import time
 import queue
 import asyncio
 import traceback
 import threading
 import opuslib_next
-import json
-import io
-import time
 import concurrent.futures
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
@@ -53,6 +53,10 @@ class ASRProviderBase(ABC):

    # 接收音频
    async def receive_audio(self, conn, audio, audio_have_voice):
+        # 检查是否在唤醒处理锁定期内
+        if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
+            return
+            
        if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
            have_voice = audio_have_voice
        else:
@@ -62,6 +66,14 @@ class ASRProviderBase(ABC):
        if not have_voice and not conn.client_have_voice:
            conn.asr_audio = conn.asr_audio[-10:]
            return
+        
+        # 检查是否处于唤醒模式
+        if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10:
+            asr_audio_task = conn.asr_audio.copy()
+            conn.reset_vad_states()
+            conn.asr_audio.clear()
+                
+            await self.handle_voice_stop(conn, asr_audio_task)

        if conn.client_voice_stop:
            asr_audio_task = conn.asr_audio.copy()
@@ -87,32 +99,13 @@ class ASRProviderBase(ABC):
            
            # 预先准备WAV数据
            wav_data = None
-            # 使用连接的声纹识别提供者
            if conn.voiceprint_provider and combined_pcm_data:
                wav_data = self._pcm_to_wav(combined_pcm_data)
            
+            # 检查是否处于唤醒模式
+            wakeup_mode = getattr(conn, 'wakeup_mode', False)
            
-            # 定义ASR任务
-            def run_asr():
-                start_time = time.monotonic()
-                try:
-                    loop = asyncio.new_event_loop()
-                    asyncio.set_event_loop(loop)
-                    try:
-                        result = loop.run_until_complete(
-                            self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
-                        )
-                        end_time = time.monotonic()
-                        logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
-                        return result
-                    finally:
-                        loop.close()
-                except Exception as e:
-                    end_time = time.monotonic()
-                    logger.bind(tag=TAG).error(f"ASR失败: {e}")
-                    return ("", None)
-            
-            # 定义声纹识别任务
+            # 声纹识别任务（公共逻辑）
            def run_voiceprint():
                if not wav_data:
                    return None
@@ -131,48 +124,83 @@ class ASRProviderBase(ABC):
                    logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
                    return None
            
-            # 使用线程池执行器并行运行
-            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
-                asr_future = thread_executor.submit(run_asr)
+            if wakeup_mode and conn.voiceprint_provider and wav_data:
+                conn.wakeup_mode = False
+                # 设置处理锁，防止后续音频片段重复处理
+                conn.wakeup_processing_lock = time.monotonic() + 3  # 3秒锁定期
                
-                if conn.voiceprint_provider and wav_data:
+                # 唤醒模式：只执行声纹识别
+                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor:
                    voiceprint_future = thread_executor.submit(run_voiceprint)
-                    
-                    # 等待两个线程都完成
-                    asr_result = asr_future.result(timeout=15)
                    voiceprint_result = voiceprint_future.result(timeout=15)
-                    
-                    results = {"asr": asr_result, "voiceprint": voiceprint_result}
-                else:
-                    asr_result = asr_future.result(timeout=15)
-                    results = {"asr": asr_result, "voiceprint": None}
-            
-            
-            # 处理结果
-            raw_text, file_path = results.get("asr", ("", None))
-            speaker_name = results.get("voiceprint", None)
-            
-            # 记录识别结果
-            if raw_text:
-                logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
-            if speaker_name:
+
+                speaker_name = voiceprint_result
                logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
-            
-            # 性能监控
-            total_time = time.monotonic() - total_start_time
-            logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
-            
-            # 检查文本长度
-            text_len, _ = remove_punctuation_and_length(raw_text)
-            self.stop_ws_connection()
-            
-            if text_len > 0:
-                # 构建包含说话人信息的JSON字符串
-                enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
+
+                fixed_text = "嘿，你好啊"
+                enhanced_text = self._build_enhanced_text(fixed_text, speaker_name)
                
-                # 使用自定义模块进行上报
+                # 性能监控
+                total_time = time.monotonic() - total_start_time
+                logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s")
+
                await startToChat(conn, enhanced_text)
                enqueue_asr_report(conn, enhanced_text, asr_audio_task)
+            else:
+                # 正常模式：执行声纹识别和ASR
+                def run_asr():
+                    start_time = time.monotonic()
+                    try:
+                        loop = asyncio.new_event_loop()
+                        asyncio.set_event_loop(loop)
+                        try:
+                            result = loop.run_until_complete(
+                                self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
+                            )
+                            end_time = time.monotonic()
+                            logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
+                            return result
+                        finally:
+                            loop.close()
+                    except Exception as e:
+                        end_time = time.monotonic()
+                        logger.bind(tag=TAG).error(f"ASR失败: {e}")
+                        return ("", None)
+                
+                # 使用线程池执行器并行运行
+                with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
+                    asr_future = thread_executor.submit(run_asr)
+                    
+                    if conn.voiceprint_provider and wav_data:
+                        voiceprint_future = thread_executor.submit(run_voiceprint)
+                        asr_result = asr_future.result(timeout=15)
+                        voiceprint_result = voiceprint_future.result(timeout=15)
+                        results = {"asr": asr_result, "voiceprint": voiceprint_result}
+                    else:
+                        asr_result = asr_future.result(timeout=15)
+                        results = {"asr": asr_result, "voiceprint": None}
+                
+                # 处理结果
+                raw_text, _ = results.get("asr", ("", None))
+                speaker_name = results.get("voiceprint", None)
+                
+                if raw_text:
+                    logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
+                if speaker_name:
+                    logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
+                
+                # 性能监控
+                total_time = time.monotonic() - total_start_time
+                logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
+                
+                # 检查文本长度
+                text_len, _ = remove_punctuation_and_length(raw_text)
+                self.stop_ws_connection()
+                
+                if text_len > 0:
+                    enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
+                    await startToChat(conn, enhanced_text)
+                    enqueue_asr_report(conn, enhanced_text, asr_audio_task)
                
        except Exception as e:
            logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
@@ -1,9 +1,7 @@
 import os
-import time
 import queue
 import aiohttp
 import asyncio
-import requests
 import traceback
 from config.logger import setup_logging
 from core.utils.tts import MarkdownCleaner
@@ -111,10 +109,6 @@ class TTSProvider(TTSProviderBase):
        finally:
            return None

-    ###################################################################################
-    # linkerai单流式TTS重写父类的方法--结束
-    ###################################################################################
-
    async def text_to_speak(self, text, is_last):
        """流式处理TTS音频，每句只推送一次音频列表"""
        await self._tts_request(text, is_last)
@@ -201,71 +195,3 @@ class TTSProvider(TTSProviderBase):
        except Exception as e:
            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
            self.tts_audio_queue.put((SentenceType.LAST, [], None))
-
-    def to_tts(self, text: str) -> list:
-        """非流式TTS处理，用于测试及保存音频文件的场景
-        Args:
-            text: 要转换的文本
-        Returns:
-            list: 返回opus编码后的音频数据列表
-        """
-        start_time = time.time()
-        text = MarkdownCleaner.clean_markdown(text)
-
-        params = {
-            "tts_text": text,
-            "spk_id": self.voice,
-            "frame_duration": 60,
-            "stream": False,
-            "target_sr": 16000,
-            "audio_format": self.audio_format,
-            "instruct_text": "请生成一段自然流畅的语音",
-        }
-        headers = {
-            "Authorization": f"Bearer {self.access_token}",
-            "Content-Type": "application/json",
-        }
-
-        try:
-            with requests.get(
-                self.api_url, params=params, headers=headers, timeout=5
-            ) as response:
-                if response.status_code != 200:
-                    logger.bind(tag=TAG).error(
-                        f"TTS请求失败: {response.status_code}, {response.text}"
-                    )
-                    return []
-
-                logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
-
-                # 使用opus编码器处理PCM数据
-                opus_datas = []
-                pcm_data = response.content
-
-                # 计算每帧的字节数
-                frame_bytes = int(
-                    self.opus_encoder.sample_rate
-                    * self.opus_encoder.channels
-                    * self.opus_encoder.frame_size_ms
-                    / 1000
-                    * 2
-                )
-
-                # 分帧处理PCM数据
-                for i in range(0, len(pcm_data), frame_bytes):
-                    frame = pcm_data[i : i + frame_bytes]
-                    if len(frame) < frame_bytes:
-                        # 最后一帧可能不足，用0填充
-                        frame = frame + b"\x00" * (frame_bytes - len(frame))
-
-                    self.opus_encoder.encode_pcm_to_opus_stream(
-                        frame,
-                        end_of_stream=(i + frame_bytes >= len(pcm_data)),
-                        callback=lambda opus: opus_datas.append(opus)
-                    )
-
-                return opus_datas
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
-            return []
@@ -1,7 +1,6 @@
 import re
 import os
 import json
-import wave
 import copy
 import socket
 import requests
@@ -12,6 +11,7 @@ from io import BytesIO
 from core.utils import p3
 from pydub import AudioSegment
 from typing import Callable, Any
+from core.providers.tts.dto.dto import SentenceType

 TAG = __name__
 emoji_map = {
@@ -274,33 +274,23 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] =
            frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
            callback(frame_data)

+def play_audio_response(conn, response):
+    """音频响应处理"""
+    conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], response.get("text")))
+    play_audio_frames(conn, response.get("file_path"))
+    conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))

-def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
-    """
-    将opus帧列表解码为wav字节流
-    """
-    decoder = opuslib_next.Decoder(sample_rate, channels)
-    pcm_datas = []

-    frame_duration = 60  # ms
-    frame_size = int(sample_rate * frame_duration / 1000)  # 960
-
-    for opus_frame in opus_datas:
-        # 解码为PCM（返回bytes，2字节/采样点）
-        pcm = decoder.decode(opus_frame, frame_size)
-        pcm_datas.append(pcm)
-
-    pcm_bytes = b"".join(pcm_datas)
-
-    # 写入wav字节流
-    wav_buffer = BytesIO()
-    with wave.open(wav_buffer, "wb") as wf:
-        wf.setnchannels(channels)
-        wf.setsampwidth(2)  # 16bit
-        wf.setframerate(sample_rate)
-        wf.writeframes(pcm_bytes)
-    return wav_buffer.getvalue()
+def play_audio_frames(conn, file_path):
+    """播放音频文件并处理发送帧数据"""
+    def handle_audio_frame(frame_data):
+        conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None))

+    audio_to_data_stream(
+        file_path,
+        is_opus=True,
+        callback=handle_audio_frame
+    )

 def check_vad_update(before_config, new_config):
    if (
@@ -1,140 +0,0 @@
-import os
-import re
-import yaml
-import time
-import hashlib
-import portalocker
-from typing import Dict
-
-
-class FileLock:
-    def __init__(self, file, timeout=5):
-        self.file = file
-        self.timeout = timeout
-        self.start_time = None
-
-    def __enter__(self):
-        self.start_time = time.time()
-        while True:
-            try:
-                portalocker.lock(self.file, portalocker.LOCK_EX | portalocker.LOCK_NB)
-                return self.file
-            except portalocker.LockException:
-                if time.time() - self.start_time > self.timeout:
-                    raise TimeoutError("获取文件锁超时")
-                time.sleep(0.1)
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        portalocker.unlock(self.file)
-
-
-class WakeupWordsConfig:
-    def __init__(self):
-        self.config_file = "data/.wakeup_words.yaml"
-        self.assets_dir = "config/assets/wakeup_words"
-        self._ensure_directories()
-        self._config_cache = None
-        self._last_load_time = 0
-        self._cache_ttl = 1  # 缓存有效期（秒）
-        self._lock_timeout = 5  # 文件锁超时时间（秒）
-
-    def _ensure_directories(self):
-        """确保必要的目录存在"""
-        os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
-        os.makedirs(self.assets_dir, exist_ok=True)
-
-    def _load_config(self) -> Dict:
-        """加载配置文件，使用缓存机制"""
-        current_time = time.time()
-
-        # 如果缓存有效，直接返回缓存
-        if (
-            self._config_cache is not None
-            and current_time - self._last_load_time < self._cache_ttl
-        ):
-            return self._config_cache
-
-        try:
-            with open(self.config_file, "a+") as f:
-                with FileLock(f, timeout=self._lock_timeout):
-                    f.seek(0)
-                    content = f.read()
-                    config = yaml.safe_load(content) if content else {}
-                    self._config_cache = config
-                    self._last_load_time = current_time
-                    return config
-        except (TimeoutError, IOError) as e:
-            print(f"加载配置文件失败: {e}")
-            return {}
-        except Exception as e:
-            print(f"加载配置文件时发生未知错误: {e}")
-            return {}
-
-    def _save_config(self, config: Dict):
-        """保存配置到文件，使用文件锁保护"""
-        try:
-            with open(self.config_file, "w") as f:
-                with FileLock(f, timeout=self._lock_timeout):
-                    yaml.dump(config, f, allow_unicode=True)
-                    self._config_cache = config
-                    self._last_load_time = time.time()
-        except (TimeoutError, IOError) as e:
-            print(f"保存配置文件失败: {e}")
-            raise
-        except Exception as e:
-            print(f"保存配置文件时发生未知错误: {e}")
-            raise
-
-    def get_wakeup_response(self, voice: str) -> Dict:
-        voice = hashlib.md5(voice.encode()).hexdigest()
-        """获取唤醒词回复配置"""
-        config = self._load_config()
-
-        if not config or voice not in config:
-            return None
-
-        # 检查文件大小
-        file_path = config[voice]["file_path"]
-        if not os.path.exists(file_path) or os.stat(file_path).st_size < (15 * 1024):
-            return None
-
-        return config[voice]
-
-    def update_wakeup_response(self, voice: str, file_path: str, text: str):
-        """更新唤醒词回复配置"""
-        try:
-            # 过滤表情符号
-            filtered_text = re.sub(r'[\U0001F600-\U0001F64F\U0001F900-\U0001F9FF]', '', text)
-
-            config = self._load_config()
-            voice_hash = hashlib.md5(voice.encode()).hexdigest()
-            config[voice_hash] = {
-                "voice": voice,
-                "file_path": file_path,
-                "time": time.time(),
-                "text": filtered_text,
-            }
-            self._save_config(config)
-        except Exception as e:
-            print(f"更新唤醒词回复配置失败: {e}")
-            raise
-
-    def generate_file_path(self, voice: str) -> str:
-        """生成音频文件路径，使用voice的哈希值作为文件名"""
-        try:
-            # 生成voice的哈希值
-            voice_hash = hashlib.md5(voice.encode()).hexdigest()
-            file_path = os.path.join(self.assets_dir, f"{voice_hash}.wav")
-
-            # 如果文件已存在，先删除
-            if os.path.exists(file_path):
-                try:
-                    os.remove(file_path)
-                except Exception as e:
-                    print(f"删除已存在的音频文件失败: {e}")
-                    raise
-
-            return file_path
-        except Exception as e:
-            print(f"生成音频文件路径失败: {e}")
-            raise