update: 音频后台队列稳定发送

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/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/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

@@ -103,6 +103,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的倍数