Merge pull request #2094 from xinnan-tech/py_fix_audio

fix: 优化流控音频播放 wechat聊天模式错误STT消息发送

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

@@ -1,5 +1,6 @@
 import json
 import asyncio
+import time
 from core.providers.tts.dto.dto import SentenceType
 from core.utils import textUtils
 
@@ -29,13 +30,60 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
 
 
 # 播放音频
-async def sendAudio(conn, audios):
+async def sendAudio(conn, audios, pre_buffer=False):
+    """
+    发送单个opus包，支持流控
+    Args:
+        conn: 连接对象
+        opus_packet: 单个opus数据包
+        pre_buffer: 快速发送音频
+    """
     if audios is None:
         return
-    # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出
+
     if isinstance(audios, bytes):
+        if conn.client_abort:
+            return
+
+        # 短音频直接发送（例如：提示音）
+        if pre_buffer:
+            await conn.websocket.send(audios)
+            return
+
+        # 重置没有声音的状态
+        conn.last_activity_time = time.time() * 1000
+
+        # 流控逻辑：确保按60ms的帧时长间隔发送
+        frame_duration = 60  # 毫秒
+
+        # 获取或初始化流控状态
+        if not hasattr(conn, "audio_flow_control"):
+            conn.audio_flow_control = {
+                "last_send_time": 0,
+                "packet_count": 0,
+                "start_time": time.perf_counter(),
+            }
+
+        flow_control = conn.audio_flow_control
+        current_time = time.perf_counter()
+
+        # 计算期望的发送时间
+        expected_time = flow_control["start_time"] + (
+            flow_control["packet_count"] * frame_duration / 1000
+        )
+
+        # 流控延迟
+        delay = expected_time - current_time
+        if delay > 0:
+            await asyncio.sleep(delay)
+
+        # 发送数据包
         await conn.websocket.send(audios)
 
+        # 更新流控状态
+        flow_control["packet_count"] += 1
+        flow_control["last_send_time"] = time.perf_counter()
+
 
 async def send_tts_message(conn, state, text=None):
     """发送 TTS 状态消息"""
@@ -56,7 +104,7 @@ async def send_tts_message(conn, state, text=None):
             conn.tts.audio_to_opus_data_stream(
                 stop_tts_notify_voice,
                 callback=lambda audio_data: asyncio.run_coroutine_threadsafe(
-                    sendAudio(conn, audio_data), conn.loop
+                    sendAudio(conn, audio_data, True), conn.loop
                 ),
             )
         # 清除服务端讲话状态
@@ -77,7 +125,7 @@ async def send_stt_message(conn, text):
     display_text = text
     try:
         # 尝试解析JSON格式
-        if text.strip().startswith('{') and text.strip().endswith('}'):
+        if text.strip().startswith("{") and text.strip().endswith("}"):
             parsed_data = json.loads(text)
             if isinstance(parsed_data, dict) and "content" in parsed_data:
                 # 如果是包含说话人信息的JSON格式，只显示content部分

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

@@ -216,7 +216,6 @@ class TTSProvider(TTSProviderBase):
 
                 if message.sentence_type == SentenceType.FIRST:
                     self.conn.client_abort = False
-                    self.reset_flow_controller()
 
                 if self.conn.client_abort:
                     logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")

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

@@ -11,7 +11,6 @@ from datetime import datetime
 from core.utils import textUtils
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption
 from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
 from core.utils.tts import MarkdownCleaner
 from core.utils.output_counter import add_device_output
@@ -71,7 +70,6 @@ class TTSProviderBase(ABC):
         self.tts_stop_request = False
         self.processed_chars = 0
         self.is_first_sentence = True
-        self.flow_controller = FlowControlConfig.create_flow_controller()
 
     def generate_filename(self, extension=".wav"):
         return os.path.join(
@@ -225,7 +223,6 @@ class TTSProviderBase(ABC):
                     self.tts_text_buff = []
                     self.is_first_sentence = True
                     self.tts_audio_first_sentence = True
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()
@@ -270,12 +267,19 @@ class TTSProviderBase(ABC):
 
                 if self.conn.client_abort:
                     logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
-                    # 打断时丢弃未上报的音频数据
                     enqueue_text, enqueue_audio = None, []
                     continue
 
                 # 收到下一个文本开始或会话结束时进行上报
                 if sentence_type is not SentenceType.MIDDLE:
+                    # 重置音频流控状态（新句子开始或者结束）
+                    if hasattr(self.conn, 'audio_flow_control'):
+                        self.conn.audio_flow_control = {
+                            'last_send_time': 0,
+                            'packet_count': 0,
+                            'start_time': time.perf_counter()
+                        }
+                    
                     # 上报TTS数据
                     if enqueue_text is not None and enqueue_audio is not None:
                         enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
@@ -284,88 +288,22 @@ class TTSProviderBase(ABC):
 
                 # 计算音频数据的帧数
                 if isinstance(audio_datas, bytes):
-                    frame_count = 1  # 单个字节流作为一帧
                     enqueue_audio.append(audio_datas)
-                else:
+
-                    frame_count = 0
+                # 发送音频
+                future = asyncio.run_coroutine_threadsafe(
+                    sendAudioMessage(self.conn, sentence_type, audio_datas, text),
+                    self.conn.loop,
+                )
+                future.result()
 
                 # 记录输出和报告
                 if self.conn.max_output_size > 0 and text:
                     add_device_output(self.conn.headers.get("device-id"), len(text))
 
-                # 流控检查
-                if frame_count > 0:
-                    max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME
-                    wait_start_time = time.time()
-                    retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL
-
-                    while not self.flow_controller.can_send_frames(frame_count):
-                        # 检查是否超时或需要停止
-                        if (
-                            time.time() - wait_start_time > max_wait_time
-                            or self.conn.stop_event.is_set()
-                            or self.conn.client_abort
-                        ):
-                            logger.bind(tag=TAG).debug(
-                                "流控等待超时或收到停止信号，跳过音频发送"
-                            )
-                            break
-                        # 短暂等待后重试
-                        time.sleep(retry_interval)
-                    else:
-                        # 可以发送，记录发送的帧数
-                        self.flow_controller.record_sent_frames(frame_count)
-
-                        # 发送音频
-                        future = asyncio.run_coroutine_threadsafe(
-                            self._send_audio_with_flow_control(
-                                sentence_type, audio_datas, text
-                            ),
-                            self.conn.loop,
-                        )
-                        future.result()
-
-                        # 输出流控状态（调试用）
-                        # status = self.flow_controller.get_status()
-                        # logger.bind(tag=TAG).debug(
-                        #     f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, "
-                        #     f"可用令牌={status['available_tokens']}..."
-                        #     f"发送帧数={status['sent_frames']}..."
-                        #     f"消费帧数={status['consumed_frames']}..."
-                        #     f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..."
-                        # )
-                else:
-                    # 没有音频数据，直接发送
-                    future = asyncio.run_coroutine_threadsafe(
-                        self._send_audio_with_flow_control(
-                            sentence_type, audio_datas, text
-                        ),
-                        self.conn.loop,
-                    )
-                    future.result()
-
             except Exception as e:
                 logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}")
 
-    async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text):
-        """
-        带流控的音频发送方法 模拟设备消费音频帧的过程
-        实际应用中应该根据设备反馈来更新消费情况
-        """
-        await sendAudioMessage(self.conn, sentence_type, audio_datas, text)
-
-        # 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步
-        if isinstance(audio_datas, bytes):
-            frame_count = 1
-            asyncio.create_task(simulate_device_consumption(self.flow_controller, frame_count))
-
-    # 在类中添加流控制器重置方法
-    def reset_flow_controller(self):
-        """重置流控制器状态，通常在新会话开始时调用"""
-        if hasattr(self, "flow_controller"):
-            self.flow_controller.reset()
-            logger.bind(tag=TAG).info("流控制器状态已重置")
-
     async def start_session(self, session_id):
         pass
 

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -213,7 +213,6 @@ class TTSProvider(TTSProviderBase):
 
                 if message.sentence_type == SentenceType.FIRST:
                     self.conn.client_abort = False
-                    self.reset_flow_controller()
 
                 if self.conn.client_abort:
                     try:

main/xiaozhi-server/core/providers/tts/index_stream.py

@@ -45,7 +45,6 @@ class TTSProvider(TTSProviderBase):
                     self.processed_chars = 0
                     self.tts_text_buff = []
                     self.before_stop_play_files.clear()
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()

main/xiaozhi-server/core/providers/tts/linkerai.py

@@ -42,7 +42,6 @@ class TTSProvider(TTSProviderBase):
                     self.processed_chars = 0
                     self.tts_text_buff = []
                     self.before_stop_play_files.clear()
-                    self.reset_flow_controller()
                 elif ContentType.TEXT == message.content_type:
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()

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

@@ -1,201 +0,0 @@
-"""
-音频流控模块
-包含令牌桶算法和音频流控制器的实现
-"""
-
-import asyncio
-import time
-import threading
-from collections import deque
-from typing import Optional, Dict, Any
-
-
-class TokenBucket:
-    """令牌桶实现，用于限流控制"""
-
-    def __init__(self, capacity: int, refill_rate: float, initial_tokens: Optional[int] = None):
-        """
-        初始化令牌桶
-
-        Args:
-            capacity: 桶容量（最大令牌数）
-            refill_rate: 令牌补充速率（每秒补充的令牌数）
-            initial_tokens: 初始令牌数，默认为桶容量
-        """
-        self.capacity = capacity
-        self.refill_rate = refill_rate
-        self.tokens = initial_tokens if initial_tokens is not None else capacity
-        self.last_refill_time = time.time()
-        self.lock = threading.Lock()
-
-    def get_tokens(self, requested_tokens: int = 1) -> bool:
-        """
-        获取指定数量的令牌
-
-        Args:
-            requested_tokens: 请求的令牌数量
-
-        Returns:
-            bool: 是否成功获取到令牌
-        """
-        with self.lock:
-            self._refill_tokens()
-
-            if self.tokens >= requested_tokens:
-                self.tokens -= requested_tokens
-                return True
-            else:
-                return False
-
-    def get_available_tokens(self) -> int:
-        """获取当前可用令牌数"""
-        with self.lock:
-            self._refill_tokens()
-            return int(self.tokens)
-
-    def _refill_tokens(self):
-        """内部方法：补充令牌"""
-        current_time = time.time()
-        time_passed = current_time - self.last_refill_time
-        tokens_to_add = time_passed * self.refill_rate
-
-        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
-        self.last_refill_time = current_time
-
-
-class AudioFlowController:
-    """音频流控制器，基于令牌桶算法控制音频数据发送"""
-
-    def __init__(self, max_device_buffer: int = 3000, refill_rate: float = 20):
-        """
-        初始化音频流控制器
-
-        Args:
-            max_device_buffer: 设备端最大缓冲区大小（Opus帧数）
-            refill_rate: 令牌补充速率（每秒允许发送的帧数）
-        """
-        self.max_device_buffer = max_device_buffer
-        self.token_bucket = TokenBucket(
-            capacity=max_device_buffer,
-            refill_rate=refill_rate,
-            initial_tokens=max_device_buffer // 2  # 初始令牌为容量的一半
-        )
-        self.sent_frames_count = 0  # 已发送帧数计数
-        self.device_consumed_frames = 0  # 设备端已消费帧数
-        self.pending_queue = deque()  # 等待发送的数据队列
-        self._lock = threading.Lock()
-
-    def can_send_frames(self, frame_count: int) -> bool:
-        """
-        检查是否可以发送指定数量的帧
-
-        Args:
-            frame_count: 要发送的帧数
-
-        Returns:
-            bool: 是否可以发送
-        """
-        with self._lock:
-            # 检查设备端缓冲区是否会溢出
-            estimated_device_buffer = self.sent_frames_count - self.device_consumed_frames
-            if estimated_device_buffer + frame_count > self.max_device_buffer:
-                return False
-
-            # 检查令牌桶是否有足够令牌
-            return self.token_bucket.get_tokens(frame_count)
-
-    def update_device_consumption(self, consumed_frames: int):
-        """
-        更新设备端消费的帧数
-
-        Args:
-            consumed_frames: 设备端消费的帧数
-        """
-        with self._lock:
-            self.device_consumed_frames += consumed_frames
-
-    def record_sent_frames(self, frame_count: int):
-        """
-        记录已发送的帧数
-
-        Args:
-            frame_count: 发送的帧数
-        """
-        with self._lock:
-            self.sent_frames_count += frame_count
-
-    def get_status(self) -> Dict[str, Any]:
-        """获取流控状态信息"""
-        with self._lock:
-            estimated_buffer = self.sent_frames_count - self.device_consumed_frames
-            return {
-                "sent_frames": self.sent_frames_count,
-                "consumed_frames": self.device_consumed_frames,
-                "estimated_device_buffer": estimated_buffer,
-                "available_tokens": self.token_bucket.get_available_tokens(),
-                "pending_queue_size": len(self.pending_queue),
-                "buffer_usage_percent": (estimated_buffer / self.max_device_buffer) * 100
-            }
-
-    def reset(self):
-        """重置流控状态"""
-        with self._lock:
-            self.sent_frames_count = 0
-            self.device_consumed_frames = 0
-            self.pending_queue.clear()
-            # 重新初始化令牌桶
-            self.token_bucket = TokenBucket(
-                capacity=self.max_device_buffer,
-                refill_rate=self.token_bucket.refill_rate,
-                initial_tokens=self.max_device_buffer // 2
-            )
-
-
-async def simulate_device_consumption(
-    flow_controller: AudioFlowController, frame_count: int
-):
-    """
-    模拟设备消费音频帧的过程
-    实际应用中应该根据设备反馈来更新消费情况
-    Args:
-        flow_controller: 流控制器实例
-        frame_count: 消费的帧数
-    """
-    # 模拟设备播放延迟（60ms per frame）
-    await asyncio.sleep(frame_count * 0.06)
-    flow_controller.update_device_consumption(frame_count)
-
-
-# 流控配置常量
-class FlowControlConfig:
-    """流控配置常量"""
-    # Opus 编码参数
-    OPUS_FRAME_DURATION_MS = 60  # Opus帧时长（毫秒）
-    OPUS_FRAMES_PER_SECOND = 1000 / OPUS_FRAME_DURATION_MS  # 每秒帧数
-
-    # 默认流控参数
-    DEFAULT_MAX_DEVICE_BUFFER = 40  # 设备端最大缓冲帧数
-    DEFAULT_REFILL_RATE = OPUS_FRAMES_PER_SECOND  # 默认令牌补充速率（帧/秒）
-    DEFAULT_MAX_WAIT_TIME = 5.0  # 流控最大等待时间（秒）
-    DEFAULT_RETRY_INTERVAL = 0.06  # 流控重试间隔（秒）
-
-    # 预缓冲参数
-    PRE_BUFFER_FRAMES = 3  # 预缓冲帧数
-
-    @classmethod
-    def create_flow_controller(cls, max_buffer: Optional[int] = None,
-                               refill_rate: Optional[float] = None) -> AudioFlowController:
-        """
-        创建流控制器的工厂方法
-
-        Args:
-            max_buffer: 最大缓冲区大小，使用默认值如果为None
-            refill_rate: 令牌补充速率，使用默认值如果为None
-
-        Returns:
-            AudioFlowController: 配置好的流控制器实例
-        """
-        return AudioFlowController(
-            max_device_buffer=max_buffer or cls.DEFAULT_MAX_DEVICE_BUFFER,
-            refill_rate=refill_rate or cls.DEFAULT_REFILL_RATE
-        )

main/xiaozhi-server/plugins_func/functions/play_music.py

@@ -212,7 +212,6 @@ async def play_local_music(conn, specific_file=None):
             conn.logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
             return
         text = _get_random_play_prompt(selected_music)
-        await send_stt_message(conn, text)
         conn.dialogue.put(Message(role="assistant", content=text))
 
         if conn.intent_type == "intent_llm":