Merge pull request #2686 from xinnan-tech/py_audio_await

Py audio 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.9";
+    public static final String VERSION = "0.8.10";
 
     /**
      * 无效固件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.9'
+      version: '0.8.10'
     }),
     showCancel: false,
     confirmText: t('common.confirm'),

main/xiaozhi-server/config.yaml

@@ -471,7 +471,6 @@ ASR:
     domain: slm # 识别领域，iat:日常用语，medical:医疗，finance:金融等
     language: zh_cn # 语言，zh_cn:中文，en_us:英文
     accent: mandarin # 方言，mandarin:普通话
-    dwa: wpgs # 动态修正，wpgs:实时返回中间结果
     # 调整音频处理参数以提高长语音识别质量
     output_dir: tmp/
   

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.9"
+SERVER_VERSION = "0.8.10"
 _logger_initialized = False
 
 

main/xiaozhi-server/core/connection.py

@@ -1171,6 +1171,11 @@ class ConnectionHandler:
                     except queue.Empty:
                         break
 
+            # 重置音频流控器（取消后台任务并清空队列）
+            if hasattr(self, "audio_rate_controller") and self.audio_rate_controller:
+                self.audio_rate_controller.reset()
+                self.logger.bind(tag=TAG).debug("已重置音频流控器")
+
             self.logger.bind(tag=TAG).debug(
                 f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}"
             )

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

@@ -10,7 +10,6 @@ TTS上报功能已集成到ConnectionHandler类中。
 """
 
 import time
-import gc
 import opuslib_next
 
 from config.manage_api_client import report as manage_report

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

@@ -16,7 +16,14 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
         await send_tts_message(conn, "start", None)
 
     if sentenceType == SentenceType.FIRST:
-        await send_tts_message(conn, "sentence_start", text)
+        # 同一句子的后续消息加入流控队列，其他情况立即发送
+        if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller and getattr(conn, "audio_flow_control", {}).get("sentence_id") == conn.sentence_id:
+            conn.audio_rate_controller.add_message(
+                lambda: send_tts_message(conn, "sentence_start", text)
+            )
+        else:
+            # 新句子或流控器未初始化，立即发送
+            await send_tts_message(conn, "sentence_start", text)
 
     await sendAudio(conn, audios)
     # 发送句子开始消息
@@ -31,6 +38,22 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
             await conn.close()
 
 
+async def _wait_for_audio_completion(conn):
+    """
+    等待音频队列清空
+
+    Args:
+        conn: 连接对象
+    """
+    if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller:
+        rate_controller = conn.audio_rate_controller
+        conn.logger.bind(tag=TAG).debug(
+            f"等待音频发送完成，队列中还有 {len(rate_controller.queue)} 个包"
+        )
+        await rate_controller.queue_empty_event.wait()
+        conn.logger.bind(tag=TAG).debug("音频发送完成")
+
+
 async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     """
     发送带16字节头部的opus数据包给mqtt_gateway
@@ -53,7 +76,6 @@ 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):
     """
     发送音频包，使用 AudioRateController 进行精确的流量控制
@@ -62,130 +84,121 @@ async def sendAudio(conn, audios, frame_duration=60):
         conn: 连接对象
         audios: 单个opus包(bytes) 或 opus包列表
         frame_duration: 帧时长（毫秒），默认60ms
-
-    改进点：
-    1. 使用单一时间基准，避免累积误差
-    2. 每次检查队列时重新计算 elapsed_ms，更精准
-    3. 支持高并发而不产生时间偏差
     """
     if audios is None or len(audios) == 0:
         return
 
-    # 获取发送延迟配置
     send_delay = conn.config.get("tts_audio_send_delay", -1) / 1000.0
+    is_single_packet = isinstance(audios, bytes)
 
-    if isinstance(audios, bytes):
+    # 初始化或获取 RateController
-        # 单个 opus 包处理
+    rate_controller, flow_control = _get_or_create_rate_controller(
-        await _sendAudio_single(conn, audios, send_delay, frame_duration)
+        conn, frame_duration, is_single_packet
-    else:
+    )
-        # 音频列表处理（如文件型音频）
+
-        await _sendAudio_list(conn, audios, send_delay, frame_duration)
+    # 统一转换为列表处理
+    audio_list = [audios] if is_single_packet else audios
+
+    # 发送音频包
+    await _send_audio_with_rate_control(
+        conn, audio_list, rate_controller, flow_control, send_delay
+    )
 
 
-async def _sendAudio_single(conn, opus_packet, send_delay, frame_duration=60):
+def _get_or_create_rate_controller(conn, frame_duration, is_single_packet):
     """
-    发送单个 opus 包
+    获取或创建 RateController 和 flow_control
-    使用 AudioRateController 进行流控
+
+    Args:
+        conn: 连接对象
+        frame_duration: 帧时长
+        is_single_packet: 是否单包模式（True: TTS流式单包, False: 批量包）
+
+    Returns:
+        (rate_controller, flow_control)
     """
-    # 重置流控状态，第一次读取和会话发生转变时
+    # 判断是否需要重置：单包模式且 sentence_id 变化，或者控制器不存在
-    if not hasattr(conn, "audio_rate_controller") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id:
+    need_reset = (
-        if hasattr(conn, "audio_rate_controller"):
+        is_single_packet
-            conn.audio_rate_controller.reset()
+        and getattr(conn, "audio_flow_control", {}).get("sentence_id") != conn.sentence_id
-        else:
+    ) or not hasattr(conn, "audio_rate_controller")
+
+    if need_reset:
+        # 创建或获取 rate_controller
+        if not hasattr(conn, "audio_rate_controller"):
             conn.audio_rate_controller = AudioRateController(frame_duration)
+        else:
             conn.audio_rate_controller.reset()
 
+        # 初始化 flow_control
         conn.audio_flow_control = {
             "packet_count": 0,
             "sequence": 0,
             "sentence_id": conn.sentence_id,
         }
 
-    if conn.client_abort:
+        # 启动后台发送循环
-        return
+        _start_background_sender(conn, conn.audio_rate_controller, conn.audio_flow_control)
 
-    conn.last_activity_time = time.time() * 1000
+    return conn.audio_rate_controller, conn.audio_flow_control
 
-    rate_controller = conn.audio_rate_controller
-    flow_control = conn.audio_flow_control
-    packet_count = flow_control["packet_count"]
 
-    # 预缓冲：前5个包直接发送，不做延迟
+def _start_background_sender(conn, rate_controller, flow_control):
+    """
+    启动后台发送循环任务
+
+    Args:
+        conn: 连接对象
+        rate_controller: 速率控制器
+        flow_control: 流控状态
+    """
+    async def send_callback(packet):
+        # 检查是否应该中止
+        if conn.client_abort:
+            raise asyncio.CancelledError("客户端已中止")
+
+        conn.last_activity_time = time.time() * 1000
+        await _do_send_audio(conn, packet, flow_control)
+        conn.client_is_speaking = True
+
+    # 使用 start_sending 启动后台循环
+    rate_controller.start_sending(send_callback)
+
+
+async def _send_audio_with_rate_control(conn, audio_list, rate_controller, flow_control, send_delay):
+    """
+    使用 rate_controller 发送音频包
+
+    Args:
+        conn: 连接对象
+        audio_list: 音频包列表
+        rate_controller: 速率控制器
+        flow_control: 流控状态
+        send_delay: 固定延迟（秒），-1表示使用动态流控
+    """
     pre_buffer_count = 5
 
-    if packet_count < pre_buffer_count or send_delay > 0:
+    for packet in audio_list:
-        # 预缓冲阶段或固定延迟模式，直接发送
-        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
-        conn.client_is_speaking = True
-
-        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)
-        conn.client_is_speaking = True
-
-    # 更新流控状态
-    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
 
-        if send_delay > 0:
+        # 预缓冲：前5个包直接发送
+        if flow_control["packet_count"] < pre_buffer_count:
+            await _do_send_audio(conn, packet, flow_control)
+            conn.client_is_speaking = True
+        elif send_delay > 0:
             # 固定延迟模式
             await asyncio.sleep(send_delay)
-        else:
+            await _do_send_audio(conn, packet, flow_control)
-            # 使用流控器进行精确延迟
-            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)
             conn.client_is_speaking = True
-            continue
+        else:
-
+            # 动态流控模式：仅添加到队列，由后台循环负责发送
-        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
+            rate_controller.add_audio(packet)
-        conn.client_is_speaking = True
 
 
-async def _do_send_audio(conn, opus_packet, flow_control, frame_duration=60):
+async def _do_send_audio(conn, opus_packet, flow_control):
     """
     执行实际的音频发送
     """
@@ -224,6 +237,8 @@ async def send_tts_message(conn, state, text=None):
             )
             audios = audio_to_data(stop_tts_notify_voice, is_opus=True)
             await sendAudio(conn, audios)
+        # 等待所有音频包发送完成
+        await _wait_for_audio_completion(conn)
         # 清除服务端讲话状态
         conn.clearSpeakStatus()
 

main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py

@@ -1,12 +1,14 @@
 import time
+import asyncio
 from typing import Dict, Any
 
-from core.handle.receiveAudioHandle import handleAudioMessage, startToChat
+from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.handle.sendAudioHandle import send_stt_message, send_tts_message
 from core.handle.textMessageHandler import TextMessageHandler
 from core.handle.textMessageType import TextMessageType
 from core.utils.util import remove_punctuation_and_length
+from core.providers.asr.dto.dto import InterfaceType
 
 TAG = __name__
 
@@ -29,8 +31,18 @@ class ListenTextMessageHandler(TextMessageHandler):
         elif msg_json["state"] == "stop":
             conn.client_have_voice = True
             conn.client_voice_stop = True
-            if len(conn.asr_audio) > 0:
+            if conn.asr.interface_type == InterfaceType.STREAM:
-                await handleAudioMessage(conn, b"")
+                # 流式模式下，发送结束请求
+                asyncio.create_task(conn.asr._send_stop_request())
+            else:
+                # 非流式模式：直接触发ASR识别
+                if len(conn.asr_audio) > 0:
+                    asr_audio_task = conn.asr_audio.copy()
+                    conn.asr_audio.clear()
+                    conn.reset_vad_states()
+
+                    if len(asr_audio_task) > 0:
+                        await conn.asr.handle_voice_stop(conn, asr_audio_task)
         elif msg_json["state"] == "detect":
             conn.client_have_voice = False
             conn.asr_audio.clear()

main/xiaozhi-server/core/providers/asr/aliyun_stream.py

@@ -5,12 +5,9 @@ import hmac
 import base64
 import hashlib
 import asyncio
-import gc
 import requests
 import websockets
 import opuslib_next
-import random
-from typing import Optional, Tuple, List
 from urllib import parse
 from datetime import datetime
 from config.logger import setup_logging
@@ -140,13 +137,13 @@ class ASRProvider(ASRProviderBase):
         conn.asr_audio.append(audio)
         conn.asr_audio = conn.asr_audio[-10:]
 
-        # 只在有声音且没有连接时建立连接
+        # 只在有声音且没有连接时建立连接（排除正在停止的情况）
-        if audio_have_voice and not self.is_processing:
+        if audio_have_voice and not self.is_processing and not self.asr_ws:
             try:
                 await self._start_recognition(conn)
             except Exception as e:
                 logger.bind(tag=TAG).error(f"开始识别失败: {str(e)}")
-                await self._cleanup(conn)
+                await self._cleanup()
                 return
 
         if self.asr_ws and self.is_processing and self.server_ready:
@@ -186,10 +183,8 @@ class ASRProvider(ASRProviderBase):
             "header": {
                 "namespace": "SpeechTranscriber",
                 "name": "StartTranscription",
-                "status": 20000000,
                 "message_id": uuid.uuid4().hex,
                 "task_id": self.task_id,
-                "status_text": "Gateway:SUCCESS:Success.",
                 "appkey": self.appkey
             },
             "payload": {
@@ -208,7 +203,7 @@ class ASRProvider(ASRProviderBase):
     async def _forward_results(self, conn):
         """转发识别结果"""
         try:
-            while self.asr_ws and not conn.stop_event.is_set():
+            while not conn.stop_event.is_set():
                 try:
                     response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
                     result = json.loads(response)
@@ -219,7 +214,10 @@ class ASRProvider(ASRProviderBase):
                     status = header.get("status", 0)
 
                     if status != 20000000:
-                        if status in [40000004, 40010004]:  # 连接超时或客户端断开
+                        if status == 40010004:
+                            logger.bind(tag=TAG).warning(f"请在服务端响应完成后再关闭链接，状态码: {status}")
+                            break
+                        if status in [40000004, 40010003]:  # 连接超时或客户端断开
                             logger.bind(tag=TAG).warning(f"连接问题，状态码: {status}")
                             break
                         elif status in [40270002, 40270003]:  # 音频问题
@@ -244,32 +242,43 @@ class ASRProvider(ASRProviderBase):
                                     logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
                                     break
                         continue
-                    
-                    if message_name == "TranscriptionResultChanged":
-                        # 中间结果
-                        text = payload.get("result", "")
-                        if text:
-                            self.text = text
                     elif message_name == "SentenceEnd":
-                        # 最终结果
+                        # 句子结束（每个句子都会触发）
                         text = payload.get("result", "")
                         if text:
-                            self.text = text
+                            logger.bind(tag=TAG).info(f"识别到文本: {text}")
-                            conn.reset_vad_states()
+
-                            # 传递缓存的音频数据
+                            # 手动模式下累积识别结果
-                            audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
+                            if conn.client_listen_mode == "manual":
-                            await self.handle_voice_stop(conn, audio_data)
+                                if self.text:
-                            # 清空缓存
+                                    self.text += text
-                            conn.asr_audio_for_voiceprint = []
+                                else:
-                            break
+                                    self.text = text
-                    elif message_name == "TranscriptionCompleted":
+
-                        # 识别完成
+                                # 手动模式下，只有在收到stop信号后才触发处理（仅处理一次）
-                        self.is_processing = False
+                                if conn.client_voice_stop:
-                        break
+                                    audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
+                                    if len(audio_data) > 0:
+                                        logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
+                                        await self.handle_voice_stop(conn, audio_data)
+                                        # 清理音频缓存
+                                        conn.asr_audio.clear()
+                                        conn.reset_vad_states()
+                                    break
+                            else:
+                                # 自动模式下直接覆盖
+                                self.text = text
+                                conn.reset_vad_states()
+                                audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
+                                await self.handle_voice_stop(conn, audio_data)
+                                break
 
                 except asyncio.TimeoutError:
-                    continue
+                    logger.bind(tag=TAG).error("接收结果超时")
-                except websockets.exceptions.ConnectionClosed:
+                    break
+                except websockets.ConnectionClosed:
+                    logger.bind(tag=TAG).info("ASR服务连接已关闭")
+                    self.is_processing = False
                     break
                 except Exception as e:
                     logger.bind(tag=TAG).error(f"处理结果失败: {str(e)}")
@@ -278,55 +287,44 @@ class ASRProvider(ASRProviderBase):
         except Exception as e:
             logger.bind(tag=TAG).error(f"结果转发失败: {str(e)}")
         finally:
-            await self._cleanup(conn)
+            # 清理连接的音频缓存
+            await self._cleanup()
+            if conn:
+                if hasattr(conn, 'asr_audio_for_voiceprint'):
+                    conn.asr_audio_for_voiceprint = []
+                if hasattr(conn, 'asr_audio'):
+                    conn.asr_audio = []
 
-    async def _cleanup(self, conn):
+    async def _send_stop_request(self):
-        """清理资源"""
+        """发送停止识别请求（不关闭连接）"""
-        logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}")
+        if self.asr_ws:
-        
-        # 清理连接的音频缓存
-        if conn and hasattr(conn, 'asr_audio_for_voiceprint'):
-            conn.asr_audio_for_voiceprint = []
-        
-        # 判断是否需要发送终止请求
-        should_stop = self.is_processing or self.server_ready
-        
-        # 发送停止识别请求
-        if self.asr_ws and should_stop:
             try:
+                # 先停止音频发送
+                self.is_processing = False
+
                 stop_msg = {
                     "header": {
                         "namespace": "SpeechTranscriber",
                         "name": "StopTranscription",
-                        "status": 20000000,
                         "message_id": uuid.uuid4().hex,
                         "task_id": self.task_id,
-                        "status_text": "Client:Stop",
                         "appkey": self.appkey
                     }
                 }
-                logger.bind(tag=TAG).debug("正在发送ASR终止请求")
+                logger.bind(tag=TAG).debug("停止识别请求已发送")
                 await self.asr_ws.send(json.dumps(stop_msg, ensure_ascii=False))
-                await asyncio.sleep(0.1)
-                logger.bind(tag=TAG).debug("ASR终止请求已发送")
             except Exception as e:
-                logger.bind(tag=TAG).error(f"ASR终止请求发送失败: {e}")
+                logger.bind(tag=TAG).error(f"发送停止识别请求失败: {e}")
 
-        # 状态重置（在终止请求发送后）
+    async def _cleanup(self):
+        """清理资源（关闭连接）"""
+        logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}")
+
+        # 状态重置
         self.is_processing = False
         self.server_ready = False
         logger.bind(tag=TAG).debug("ASR状态已重置")
 
-        # 清理任务
-        if self.forward_task and not self.forward_task.done():
-            self.forward_task.cancel()
-            try:
-                await asyncio.wait_for(self.forward_task, timeout=1.0)
-            except Exception as e:
-                logger.bind(tag=TAG).debug(f"forward_task取消异常: {e}")
-            finally:
-                self.forward_task = None
-        
         # 关闭连接
         if self.asr_ws:
             try:
@@ -338,6 +336,9 @@ class ASRProvider(ASRProviderBase):
             finally:
                 self.asr_ws = None
 
+        # 清理任务引用
+        self.forward_task = None
+
         logger.bind(tag=TAG).debug("ASR会话清理完成")
 
     async def speech_to_text(self, opus_data, session_id, audio_format):

main/xiaozhi-server/core/providers/asr/base.py

@@ -9,8 +9,6 @@ import asyncio
 import traceback
 import threading
 import opuslib_next
-import concurrent.futures
-import gc
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
@@ -54,23 +52,26 @@ class ASRProviderBase(ABC):
 
     # 接收音频
     async def receive_audio(self, conn, audio, audio_have_voice):
-        if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
+        if conn.client_listen_mode == "manual":
-            have_voice = audio_have_voice
+            # 手动模式：缓存音频用于ASR识别
+            conn.asr_audio.append(audio)
         else:
-            have_voice = conn.client_have_voice
+            # 自动/实时模式：使用VAD检测
+            have_voice = audio_have_voice
 
-        conn.asr_audio.append(audio)
+            conn.asr_audio.append(audio)
-        if not have_voice and not conn.client_have_voice:
+            if not have_voice and not conn.client_have_voice:
-            conn.asr_audio = conn.asr_audio[-10:]
+                conn.asr_audio = conn.asr_audio[-10:]
-            return
+                return
 
-        if conn.client_voice_stop:
+            # 自动模式下通过VAD检测到语音停止时触发识别
-            asr_audio_task = conn.asr_audio.copy()
+            if conn.client_voice_stop:
-            conn.asr_audio.clear()
+                asr_audio_task = conn.asr_audio.copy()
-            conn.reset_vad_states()
+                conn.asr_audio.clear()
+                conn.reset_vad_states()
 
-            if len(asr_audio_task) > 15:
+                if len(asr_audio_task) > 15:
-                await self.handle_voice_stop(conn, asr_audio_task)
+                    await self.handle_voice_stop(conn, asr_audio_task)
 
     # 处理语音停止
     async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]):
@@ -92,66 +93,31 @@ class ASRProviderBase(ABC):
                 wav_data = self._pcm_to_wav(combined_pcm_data)
 
             # 定义ASR任务
-            def run_asr():
+            asr_task = self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
-                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).debug(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)
 
-            # 定义声纹识别任务
+            if conn.voiceprint_provider and wav_data:
-            def run_voiceprint():
+                voiceprint_task = conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
-                if not wav_data:
+                # 并发等待两个结果
-                    return None
+                asr_result, voiceprint_result = await asyncio.gather(
-                try:
+                    asr_task, voiceprint_task, return_exceptions=True
-                    loop = asyncio.new_event_loop()
+                )
-                    asyncio.set_event_loop(loop)
+            else:
-                    try:
+                asr_result = await asr_task
-                        # 使用连接的声纹识别提供者
+                voiceprint_result = None
-                        result = loop.run_until_complete(
-                            conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
-                        )
-                        return result
-                    finally:
-                        loop.close()
-                except Exception as e:
-                    logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
-                    return None
 
-            # 使用线程池执行器并行运行
+            # 记录识别结果 - 检查是否为异常
-            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
+            if isinstance(asr_result, Exception):
-                asr_future = thread_executor.submit(run_asr)
+                logger.bind(tag=TAG).error(f"ASR识别失败: {asr_result}")
+                raw_text = ""
+            else:
+                raw_text, _ = asr_result
 
-                if conn.voiceprint_provider and wav_data:
+            if isinstance(voiceprint_result, Exception):
-                    voiceprint_future = thread_executor.submit(run_voiceprint)
+                logger.bind(tag=TAG).error(f"声纹识别失败: {voiceprint_result}")
+                speaker_name = ""
+            else:
+                speaker_name = voiceprint_result
 
-                    # 等待两个线程都完成
-                    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:

main/xiaozhi-server/core/providers/asr/doubao_stream.py

@@ -4,7 +4,6 @@ import uuid
 import asyncio
 import websockets
 import opuslib_next
-import gc
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
 from core.providers.asr.dto.dto import InterfaceType
@@ -19,8 +18,6 @@ class ASRProvider(ASRProviderBase):
         self.interface_type = InterfaceType.STREAM
         self.config = config
         self.text = ""
-        self.max_retries = 3
-        self.retry_delay = 2
         self.decoder = opuslib_next.Decoder(16000, 1)
         self.asr_ws = None
         self.forward_task = None
@@ -57,14 +54,13 @@ class ASRProvider(ASRProviderBase):
     async def receive_audio(self, conn, audio, audio_have_voice):
         conn.asr_audio.append(audio)
         conn.asr_audio = conn.asr_audio[-10:]
-        
         # 存储音频数据
         if not hasattr(conn, 'asr_audio_for_voiceprint'):
             conn.asr_audio_for_voiceprint = []
         conn.asr_audio_for_voiceprint.append(audio)
 
         # 当没有音频数据时处理完整语音片段
-        if not audio and len(conn.asr_audio_for_voiceprint) > 0:
+        if conn.client_listen_mode != "manual" and not audio and len(conn.asr_audio_for_voiceprint) > 0:
             await self.handle_voice_stop(conn, conn.asr_audio_for_voiceprint)
             conn.asr_audio_for_voiceprint = []
 
@@ -180,6 +176,7 @@ class ASRProvider(ASRProviderBase):
                                 payload.get("audio_info", {}).get("duration", 0) > 2000
                                 and not utterances
                                 and not payload["result"].get("text")
+                                and conn.client_listen_mode != "manual"
                             ):
                                 logger.bind(tag=TAG).error(f"识别文本：空")
                                 self.text = ""
@@ -188,15 +185,44 @@ class ASRProvider(ASRProviderBase):
                                     await self.handle_voice_stop(conn, audio_data)
                                 break
 
+                            # 专门处理没有文本的识别结果（手动模式下可能已经识别完成但是没松按键）
+                            elif not payload["result"].get("text") and not utterances:
+                                if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 0:
+                                    logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理")
+                                    await self.handle_voice_stop(conn, audio_data)
+                                    # 清理音频缓存
+                                    conn.asr_audio.clear()
+                                    conn.reset_vad_states()
+                                    break
+
                             for utterance in utterances:
                                 if utterance.get("definite", False):
-                                    self.text = utterance["text"]
+                                    current_text = utterance["text"]
                                     logger.bind(tag=TAG).info(
-                                        f"识别到文本: {self.text}"
+                                        f"识别到文本: {current_text}"
                                     )
-                                    conn.reset_vad_states()
+
-                                    if len(audio_data) > 15:  # 确保有足够音频数据
+                                    # 手动模式下累积识别结果
-                                        await self.handle_voice_stop(conn, audio_data)
+                                    if conn.client_listen_mode == "manual":
+                                        if self.text:
+                                            self.text += current_text
+                                        else:
+                                            self.text = current_text
+
+                                        # 在接收消息中途时收到停止信号
+                                        if conn.client_voice_stop and len(audio_data) > 0:
+                                            logger.bind(tag=TAG).debug("消息中途收到停止信号，触发处理")
+                                            await self.handle_voice_stop(conn, audio_data)
+                                            # 清理音频缓存
+                                            conn.asr_audio.clear()
+                                            conn.reset_vad_states()
+                                        break
+                                    else:
+                                        # 自动模式下直接覆盖
+                                        self.text = current_text
+                                        conn.reset_vad_states()
+                                        if len(audio_data) > 15:  # 确保有足够音频数据
+                                            await self.handle_voice_stop(conn, audio_data)
                                     break
                         elif "error" in payload:
                             error_msg = payload.get("error", "未知错误")
@@ -228,8 +254,6 @@ class ASRProvider(ASRProviderBase):
                     conn.asr_audio_for_voiceprint = []
                 if hasattr(conn, 'asr_audio'):
                     conn.asr_audio = []
-                if hasattr(conn, 'has_valid_voice'):
-                    conn.has_valid_voice = False
 
     def stop_ws_connection(self):
         if self.asr_ws:
@@ -237,6 +261,20 @@ class ASRProvider(ASRProviderBase):
             self.asr_ws = None
         self.is_processing = False
 
+    async def _send_stop_request(self):
+        """发送最后一个音频帧以通知服务器结束"""
+        if self.asr_ws:
+            try:
+                # 发送结束标记的音频帧（gzip压缩的空数据）
+                empty_payload = gzip.compress(b"")
+                last_audio_request = bytearray(self.generate_last_audio_default_header())
+                last_audio_request.extend(len(empty_payload).to_bytes(4, "big"))
+                last_audio_request.extend(empty_payload)
+                await self.asr_ws.send(last_audio_request)
+                logger.bind(tag=TAG).debug("已发送结束音频帧")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"发送结束音频帧时出错: {e}")
+
     def construct_request(self, reqid):
         req = {
             "app": {
@@ -388,5 +426,3 @@ class ASRProvider(ASRProviderBase):
                     conn.asr_audio_for_voiceprint = []
                 if hasattr(conn, 'asr_audio'):
                     conn.asr_audio = []
-                if hasattr(conn, 'has_valid_voice'):
-                    conn.has_valid_voice = False

main/xiaozhi-server/core/providers/asr/fun_local.py

@@ -1,14 +1,16 @@
-import time
 import os
-import sys
 import io
+import sys
+import time
+import shutil
 import psutil
+import asyncio
+
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
-from core.providers.asr.base import ASRProviderBase
 from funasr import AutoModel
 from funasr.utils.postprocess_utils import rich_transcription_postprocess
-import shutil
+from core.providers.asr.base import ASRProviderBase
 from core.providers.asr.dto.dto import InterfaceType
 
 TAG = __name__
@@ -90,16 +92,17 @@ class ASRProvider(ASRProviderBase):
                 else:
                     file_path = self.save_audio_to_file(pcm_data, session_id)
 
-                # 语音识别
+                # 语音识别 - 使用线程池避免阻塞事件循环
                 start_time = time.time()
-                result = self.model.generate(
+                result = await asyncio.to_thread(
+                    self.model.generate,
                     input=combined_pcm_data,
                     cache={},
                     language="auto",
                     use_itn=True,
                     batch_size_s=60,
                 )
-                text = rich_transcription_postprocess(result[0]["text"])
+                text = await asyncio.to_thread(rich_transcription_postprocess, result[0]["text"])
                 logger.bind(tag=TAG).debug(
                     f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
                 )

main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py

@@ -1,8 +1,5 @@
 import os
-import json
-import asyncio
 import tempfile
-import difflib
 from typing import Optional, Tuple, List
 import dashscope
 from config.logger import setup_logging
@@ -16,7 +13,8 @@ logger = setup_logging()
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
         super().__init__()
-        self.interface_type = InterfaceType.STREAM
+        # 音频文件上传类型，流式文本识别输出
+        self.interface_type = InterfaceType.NON_STREAM
         """Qwen3-ASR-Flash ASR初始化"""
         
         # 配置参数
@@ -130,27 +128,11 @@ class ASRProvider(ASRProviderBase):
             
             # 处理流式响应
             full_text = ""
-            last_text = ""  # 用于存储上一个文本片段
             for chunk in response:
                 try:
                     text = chunk["output"]["choices"][0]["message"].content[0]["text"]
-                    # 标准化文本片段（去除首尾空格）
+                    # 更新为最新的完整文本
-                    normalized_text = text.strip()
+                    full_text = text.strip()
-                    # 只有当新文本片段与上一个不同时才处理
-                    if normalized_text != last_text:
-                        # 提取新增的文本部分
-                        # 通过比较当前文本和上一个文本，找到新增的部分
-                        if normalized_text.startswith(last_text):
-                            # 如果当前文本以最后一个文本开头，则新增部分是两者的差集
-                            new_part = normalized_text[len(last_text):]
-                        else:
-                            # 如果不以最后一个文本开头，说明识别结果发生了较大变化，直接使用当前文本
-                            new_part = normalized_text
-                        
-                        # 将新增部分添加到完整文本中
-                        full_text += new_part
-                        last_text = normalized_text
-                    # 这里可以实时处理文本片段，例如通过回调函数
                 except:
                     pass
             

main/xiaozhi-server/core/providers/asr/xunfei_stream.py

@@ -35,9 +35,6 @@ class ASRProvider(ASRProviderBase):
         self.forward_task = None
         self.is_processing = False
         self.server_ready = False
-        self.last_frame_sent = False  # 标记是否已发送最终帧
-        self.best_text = ""  # 保存最佳识别结果
-        self.has_final_result = False  # 标记是否收到最终识别结果
 
         # 讯飞配置
         self.app_id = config.get("app_id")
@@ -52,7 +49,6 @@ class ASRProvider(ASRProviderBase):
             "domain": config.get("domain", "slm"),
             "language": config.get("language", "zh_cn"),
             "accent": config.get("accent", "mandarin"),
-            "dwa": config.get("dwa", "wpgs"),
             "result": {"encoding": "utf8", "compress": "raw", "format": "plain"},
         }
 
@@ -116,7 +112,7 @@ class ASRProvider(ASRProviderBase):
                 await self._start_recognition(conn)
             except Exception as e:
                 logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}")
-                await self._cleanup(conn)
+                await self._cleanup()
                 return
 
         # 发送当前音频数据
@@ -126,7 +122,7 @@ class ASRProvider(ASRProviderBase):
                 await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME)
             except Exception as e:
                 logger.bind(tag=TAG).warning(f"发送音频数据时发生错误: {e}")
-                await self._cleanup(conn)
+                await self._cleanup()
 
     async def _start_recognition(self, conn):
         """开始识别会话"""
@@ -136,6 +132,10 @@ class ASRProvider(ASRProviderBase):
             ws_url = self.create_url()
             logger.bind(tag=TAG).info(f"正在连接ASR服务: {ws_url[:50]}...")
 
+            # 如果为手动模式,设置超时时长为一分钟
+            if conn.client_listen_mode == "manual":
+                self.iat_params["eos"] = 60000
+
             self.asr_ws = await websockets.connect(
                 ws_url,
                 max_size=1000000000,
@@ -146,8 +146,6 @@ class ASRProvider(ASRProviderBase):
 
             logger.bind(tag=TAG).info("ASR WebSocket连接已建立")
             self.server_ready = False
-            self.last_frame_sent = False
-            self.best_text = ""
             self.forward_task = asyncio.create_task(self._forward_results(conn))
 
             # 发送首帧音频
@@ -196,23 +194,12 @@ class ASRProvider(ASRProviderBase):
 
         await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False))
 
-        # 标记是否发送了最终帧
-        if status == STATUS_LAST_FRAME:
-            self.last_frame_sent = True
-            logger.bind(tag=TAG).info("标记最终帧已发送")
-
     async def _forward_results(self, conn):
         """转发识别结果"""
         try:
-            while self.asr_ws and not conn.stop_event.is_set():
+            while not conn.stop_event.is_set():
-                # 获取当前连接的音频数据
-                audio_data = getattr(conn, "asr_audio_for_voiceprint", [])
                 try:
-                    # 如果已发送最终帧，增加超时时间等待完整结果
+                    response = await asyncio.wait_for(self.asr_ws.recv(), timeout=60)
-                    timeout = 3.0 if self.last_frame_sent else 30.0
-                    response = await asyncio.wait_for(
-                        self.asr_ws.recv(), timeout=timeout
-                    )
                     result = json.loads(response)
                     logger.bind(tag=TAG).debug(f"收到ASR结果: {result}")
 
@@ -236,144 +223,27 @@ class ASRProvider(ASRProviderBase):
                             # 解码base64文本
                             decoded_text = base64.b64decode(text_data).decode("utf-8")
                             text_json = json.loads(decoded_text)
-
                             # 提取文本内容
                             text_ws = text_json.get("ws", [])
-                            result_text = ""
                             for i in text_ws:
                                 for j in i.get("cw", []):
                                     w = j.get("w", "")
-                                    result_text += w
+                                    self.text += w
 
-                            # 更新识别文本 - 实时更新策略
-                            # 只检查是否为空字符串，不再过滤任何标点符号
-                            # 这样可以确保所有识别到的内容，包括标点符号都能被实时更新
-                            if result_text and result_text.strip():
-                                # 实时更新：正常情况下都更新，提高响应速度
-                                should_update = True
-
-                                # 保存最佳文本
-                                # 1. 如果是识别完成状态或最终帧后收到的结果，优先保存
-                                # 2. 否则保存最长的有意义文本
-                                # 取消对标点符号的过滤，只检查是否为空
-                                # 这样可以保留所有识别到的内容，包括各种标点符号
-                                is_valid_text = len(result_text.strip()) > 0
-
-                                if (
-                                    self.last_frame_sent or status == 2
-                                ) and is_valid_text:
-                                    self.best_text = result_text
-                                    self.has_final_result = True  # 标记已收到最终结果
-                                    logger.bind(tag=TAG).debug(
-                                        f"保存最终识别结果: {self.best_text}"
-                                    )
-                                elif (
-                                    len(result_text) > len(self.best_text)
-                                    and is_valid_text
-                                    and not self.has_final_result
-                                ):
-                                    self.best_text = result_text
-                                    logger.bind(tag=TAG).debug(
-                                        f"保存中间最佳文本: {self.best_text}"
-                                    )
-
-                                # 如果已发送最终帧，只过滤空文本
-                                if self.last_frame_sent:
-                                    # 只拒绝完全空的结果
-                                    if not result_text.strip():
-                                        should_update = False
-                                        logger.bind(tag=TAG).warning(
-                                            f"最终帧后拒绝空文本"
-                                        )
-
-                                if should_update:
-                                    # 处理流式识别结果，避免简单替换导致内容丢失
-                                    # 1. 如果是中间状态（非最终帧后），可能需要替换为更完整的识别
-                                    # 2. 如果是最终帧后收到的结果，可能是对前面文本的补充
-                                    if self.last_frame_sent:
-                                        # 最终帧后收到的结果可能是标点符号等补充内容
-                                        # 检查是否需要合并文本而不是替换
-                                        # 如果当前文本是纯标点而前面已有内容，应该追加而不是替换
-                                        if len(
-                                            self.text
-                                        ) > 0 and result_text.strip() in [
-                                            "。",
-                                            ".",
-                                            "?",
-                                            "？",
-                                            "!",
-                                            "！",
-                                            ",",
-                                            "，",
-                                            ";",
-                                            "；",
-                                        ]:
-                                            # 对于标点符号，追加到现有文本后
-                                            self.text = (
-                                                self.text.rstrip().rstrip("。.")
-                                                + result_text
-                                            )
-                                        else:
-                                            # 其他情况保持替换逻辑
-                                            self.text = result_text
-                                    else:
-                                        # 中间状态替换为新的识别结果
-                                        self.text = result_text
-
-                                    logger.bind(tag=TAG).info(
-                                        f"实时更新识别文本: {self.text} (最终帧已发送: {self.last_frame_sent})"
-                                    )
-
-                    # 识别完成，但如果还没发送最终帧，继续等待
                     if status == 2:
-                        logger.bind(tag=TAG).info(
+                        if conn.client_listen_mode == "manual":
-                            f"识别完成状态已到达，当前识别文本: {self.text}"
+                            audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
-                        )
+                            if len(audio_data) > 0:
-
+                                logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
-                        # 如果还没发送最终帧，继续等待
+                                await self.handle_voice_stop(conn, audio_data)
-                        if not self.last_frame_sent:
+                                # 清理音频缓存
-                            logger.bind(tag=TAG).info(
+                                conn.asr_audio.clear()
-                                "识别完成但最终帧未发送，继续等待..."
-                            )
-                            continue
-
-                        # 已发送最终帧且收到完成状态，使用最佳策略选择最终结果
-                        # 优先使用识别完成状态下的最新结果，而不是仅仅基于长度
-                        if self.best_text:
-                            # 如果当前文本是在最终帧发送后或识别完成状态下收到的，优先使用
-                            if (
-                                self.last_frame_sent or status == 2
-                            ) and self.text.strip():
-                                logger.bind(tag=TAG).info(
-                                    f"使用完成状态下的最新识别结果: {self.text}"
-                                )
-                            elif len(self.best_text) > len(self.text):
-                                logger.bind(tag=TAG).info(
-                                    f"使用更长的最佳文本作为最终结果: {self.text} -> {self.best_text}"
-                                )
-                                self.text = self.best_text
-
-                        logger.bind(tag=TAG).info(f"获取到最终完整文本: {self.text}")
                         conn.reset_vad_states()
-                        if len(audio_data) > 15:  # 确保有足够音频数据
-                            # 准备处理结果
-                            pass
                         break
 
                 except asyncio.TimeoutError:
-                    if self.last_frame_sent:
+                    logger.bind(tag=TAG).error("接收结果超时")
-                        # 超时时也使用最佳文本
+                    break
-                        if self.best_text and len(self.best_text) > len(self.text):
-                            logger.bind(tag=TAG).info(
-                                f"超时，使用最佳文本: {self.text} -> {self.best_text}"
-                            )
-                            self.text = self.best_text
-                        logger.bind(tag=TAG).info(
-                            f"最终帧后超时，使用结果: {self.text}"
-                        )
-                        break
-                    # 如果还没发送最终帧，继续等待
-                    continue
                 except websockets.ConnectionClosed:
                     logger.bind(tag=TAG).info("ASR服务连接已关闭")
                     self.is_processing = False
@@ -390,17 +260,15 @@ class ASRProvider(ASRProviderBase):
             if hasattr(e, "__cause__") and e.__cause__:
                 logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}")
         finally:
-            if self.asr_ws:
+            # 清理连接资源
-                await self.asr_ws.close()
+            await self._cleanup()
-                self.asr_ws = None
+
-            self.is_processing = False
+            # 清理连接的音频缓存
             if conn:
                 if hasattr(conn, "asr_audio_for_voiceprint"):
                     conn.asr_audio_for_voiceprint = []
                 if hasattr(conn, "asr_audio"):
                     conn.asr_audio = []
-                if hasattr(conn, "has_valid_voice"):
-                    conn.has_valid_voice = False
 
     async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]):
         """处理语音停止，发送最后一帧并处理识别结果"""
@@ -408,22 +276,13 @@ class ASRProvider(ASRProviderBase):
             # 先发送最后一帧表示音频结束
             if self.asr_ws and self.is_processing:
                 try:
-                    # 取最后一个有效的音频帧作为最后一帧数据
+                    await self._send_audio_frame(b"", STATUS_LAST_FRAME)
-                    last_frame = b""
+                    logger.bind(tag=TAG).debug(f"已发送停止请求")
-                    if asr_audio_task:
-                        last_audio = asr_audio_task[-1]
-                        last_frame = self.decoder.decode(last_audio, 960)
-                    await self._send_audio_frame(last_frame, STATUS_LAST_FRAME)
-                    logger.bind(tag=TAG).info("已发送最后一帧")
 
-                    # 发送最终帧后，给_forward_results适当时间处理最终结果
                     await asyncio.sleep(0.25)
-
-                    logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}")
                 except Exception as e:
-                    logger.bind(tag=TAG).error(f"发送最后一帧失败: {e}")
+                    logger.bind(tag=TAG).error(f"发送停止请求失败: {e}")
 
-            # 调用父类的handle_voice_stop方法处理识别结果
             await super().handle_voice_stop(conn, asr_audio_task)
         except Exception as e:
             logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
@@ -437,40 +296,27 @@ class ASRProvider(ASRProviderBase):
             self.asr_ws = None
         self.is_processing = False
 
-    async def _cleanup(self, conn):
+    async def _send_stop_request(self):
-        """清理资源"""
+        """发送停止识别请求（不关闭连接）"""
-        logger.bind(tag=TAG).info(
+        if self.asr_ws:
+            try:
+                # 先停止音频发送
+                self.is_processing = False
+                await self._send_audio_frame(b"", STATUS_LAST_FRAME)
+                logger.bind(tag=TAG).debug("已发送停止请求")
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"发送停止请求失败: {e}")
+
+    async def _cleanup(self):
+        """清理资源（关闭连接）"""
+        logger.bind(tag=TAG).debug(
             f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}"
         )
 
-        # 发送最后一帧
-        if self.asr_ws and self.is_processing:
-            try:
-                await self._send_audio_frame(b"", STATUS_LAST_FRAME)
-                await asyncio.sleep(0.1)
-                logger.bind(tag=TAG).info("已发送最后一帧")
-            except Exception as e:
-                logger.bind(tag=TAG).error(f"发送最后一帧失败: {e}")
-
         # 状态重置
         self.is_processing = False
         self.server_ready = False
-        self.last_frame_sent = False
+        logger.bind(tag=TAG).debug("ASR状态已重置")
-        self.best_text = ""
-        self.has_final_result = False
-        logger.bind(tag=TAG).info("ASR状态已重置")
-
-        # 清理任务
-        if self.forward_task and not self.forward_task.done():
-            self.forward_task.cancel()
-            try:
-                await asyncio.wait_for(self.forward_task, timeout=1.0)
-            except asyncio.CancelledError:
-                pass
-            except Exception as e:
-                logger.bind(tag=TAG).debug(f"forward_task取消异常: {e}")
-            finally:
-                self.forward_task = None
 
         # 关闭连接
         if self.asr_ws:
@@ -483,16 +329,10 @@ class ASRProvider(ASRProviderBase):
             finally:
                 self.asr_ws = None
 
-        # 清理连接的音频缓存
+        # 清理任务引用
-        if conn:
+        self.forward_task = None
-            if hasattr(conn, "asr_audio_for_voiceprint"):
-                conn.asr_audio_for_voiceprint = []
-            if hasattr(conn, "asr_audio"):
-                conn.asr_audio = []
-            if hasattr(conn, "has_valid_voice"):
-                conn.has_valid_voice = False
 
-        logger.bind(tag=TAG).info("ASR会话清理完成")
+        logger.bind(tag=TAG).debug("ASR会话清理完成")
 
     async def speech_to_text(self, opus_data, session_id, audio_format):
         """获取识别结果"""
@@ -530,5 +370,3 @@ class ASRProvider(ASRProviderBase):
                     conn.asr_audio_for_voiceprint = []
                 if hasattr(conn, "asr_audio"):
                     conn.asr_audio = []
-                if hasattr(conn, "has_valid_voice"):
-                    conn.has_valid_voice = False

main/xiaozhi-server/core/providers/vad/silero.py

@@ -2,7 +2,6 @@ import time
 import numpy as np
 import torch
 import opuslib_next
-import gc
 from config.logger import setup_logging
 from core.providers.vad.base import VADProviderBase
 
@@ -45,6 +44,10 @@ class VADProvider(VADProviderBase):
                 pass
 
     def is_vad(self, conn, opus_packet):
+        # 手动模式：直接返回True，不进行实时VAD检测，所有音频都缓存
+        if conn.client_listen_mode == "manual":
+            return True
+            
         try:
             pcm_frame = self.decoder.decode(opus_packet, 960)
             conn.client_audio_buffer.extend(pcm_frame)  # 将新数据加入缓冲区

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

@@ -10,12 +10,6 @@ class AudioRateController:
     """
     音频速率控制器 - 按照60ms帧时长精确控制音频发送
     解决高并发下的时间累积误差问题
-
-    关键改进：
-    1. 单一时间基准（start_timestamp 只初始化一次）
-    2. 每次检查队列时重新计算 elapsed_ms，避免累积误差
-    3. 分离"检查时间"和"发送"两个操作
-    4. 支持高并发而不产生延迟
     """
 
     def __init__(self, frame_duration=60):
@@ -29,24 +23,34 @@ class AudioRateController:
         self.start_timestamp = None  # 开始时间戳（只读，不修改）
         self.pending_send_task = None
         self.logger = logger
+        self.queue_empty_event = asyncio.Event()  # 队列清空事件
+        self.queue_empty_event.set()  # 初始为空状态
 
     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()
+        self.queue_empty_event.set()  # 队列已清空
 
     def add_audio(self, opus_packet):
         """添加音频包到队列"""
         self.queue.append(("audio", opus_packet))
+        self.queue_empty_event.clear()  # 队列非空，清除事件
+
+    def add_message(self, message_callback):
+        """
+        添加消息到队列（立即发送，不占用播放时间）
+
+        Args:
+            message_callback: 消息发送回调函数 async def()
+        """
+        self.queue.append(("message", message_callback))
+        self.queue_empty_event.clear()  # 队列非空，清除事件
 
     def _get_elapsed_ms(self):
         """获取已经过的时间（毫秒）"""
@@ -62,34 +66,47 @@ class AudioRateController:
             send_audio_callback: 发送音频的回调函数 async def(opus_packet)
         """
         if self.start_timestamp is None:
-            self.reset()
+            self.start_timestamp = time.time()
 
         while self.queue:
             item = self.queue[0]
             item_type = item[0]
 
-            if item_type == "audio":
+            if item_type == "message":
+                # 消息类型：立即发送，不占用播放时间
+                _, message_callback = item
+                self.queue.pop(0)
+                try:
+                    await message_callback()
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"发送消息失败: {e}")
+                    raise
+
+            elif item_type == "audio":
                 _, opus_packet = item
 
-                # 计算时间差
+                # 循环等待直到时间到达
-                elapsed_ms = self._get_elapsed_ms()
+                while True:
-                output_ms = self.play_position
+                    # 计算时间差
+                    elapsed_ms = self._get_elapsed_ms()
+                    output_ms = self.play_position
 
-                if elapsed_ms < output_ms:
+                    if elapsed_ms < output_ms:
-                    # 还不到发送时间，计算等待时长
+                        # 还不到发送时间，计算等待时长
-                    wait_ms = output_ms - elapsed_ms
+                        wait_ms = output_ms - elapsed_ms
 
-                    # 等待后继续检查（允许被中断）
+                        # 等待后继续检查（允许被中断）
-                    try:
+                        try:
-                        await asyncio.sleep(wait_ms / 1000)
+                            await asyncio.sleep(wait_ms / 1000)
-                    except asyncio.CancelledError:
+                        except asyncio.CancelledError:
-                        self.logger.bind(tag=TAG).debug("音频发送任务被取消")
+                            self.logger.bind(tag=TAG).debug("音频发送任务被取消")
-                        raise
+                            raise
+                        # 等待结束后重新检查时间（循环回到 while True）
+                    else:
+                        # 时间已到，跳出等待循环
+                        break
 
-                    # 继续循环检查（时间可能已到）
+                # 时间已到，从队列移除并发送
-                    continue
-
-                # 时间已到，发送音频
                 self.queue.pop(0)
                 self.play_position += self.frame_duration
 
@@ -99,14 +116,15 @@ class AudioRateController:
                     self.logger.bind(tag=TAG).error(f"发送音频失败: {e}")
                     raise
 
+        self.queue_empty_event.set()
 
-    async def start_sending(self, send_audio_callback, send_message_callback=None):
+
+    def start_sending(self, send_audio_callback):
         """
         启动异步发送任务
 
         Args:
             send_audio_callback: 发送音频的回调函数
-            send_message_callback: 发送消息的回调函数
 
         Returns:
             asyncio.Task: 发送任务
@@ -114,11 +132,11 @@ class AudioRateController:
         async def _send_loop():
             try:
                 while True:
-                    await self.check_queue(send_audio_callback, send_message_callback)
+                    await self.check_queue(send_audio_callback)
                     # 如果队列空了，短暂等待后再检查（避免 busy loop）
                     await asyncio.sleep(0.01)
             except asyncio.CancelledError:
-                self.logger.bind(tag=TAG).info("音频发送循环已停止")
+                self.logger.bind(tag=TAG).debug("音频发送循环已停止")
             except Exception as e:
                 self.logger.bind(tag=TAG).error(f"音频发送循环异常: {e}")
 

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

@@ -6,7 +6,6 @@ Opus编码工具类
 import logging
 import traceback
 import numpy as np
-import gc
 from opuslib_next import Encoder
 from opuslib_next import constants
 from typing import Optional, Callable, Any
@@ -103,6 +102,9 @@ class OpusEncoderUtils:
     def _encode(self, frame: np.ndarray) -> Optional[bytes]:
         """编码一帧音频数据"""
         try:
+            # 编码器已释放，跳过编码
+            if not hasattr(self, 'encoder') or self.encoder is None:
+                return None
             # 将numpy数组转换为bytes
             frame_bytes = frame.tobytes()
             # opuslib要求输入字节数必须是channels*2的倍数

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

@@ -8,7 +8,6 @@ import requests
 import subprocess
 import numpy as np
 import opuslib_next
-import gc
 from io import BytesIO
 from core.utils import p3
 from pydub import AudioSegment