Merge pull request #1988 from myifeng/async-handler

调整OPUS函数式处理，不再以数组方式收集再遍历

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

@@ -83,7 +83,7 @@ async def checkWakeupWords(conn, text):
 
     # 播放唤醒词回复
     conn.client_abort = False
-    opus_packets, _ = audio_to_data(response.get("file_path"))
+    opus_packets = audio_to_data(response.get("file_path"))
 
     conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
     await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))

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

@@ -140,7 +140,7 @@ async def check_bind_device(conn):
 
         # 播放提示音
         music_path = "config/assets/bind_code.wav"
-        opus_packets, _ = audio_to_data(music_path)
+        opus_packets = audio_to_data(music_path)
         conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text))
 
         # 逐个播放数字
@@ -148,7 +148,7 @@ async def check_bind_device(conn):
             try:
                 digit = conn.bind_code[i]
                 num_path = f"config/assets/bind_code/{digit}.wav"
-                num_packets, _ = audio_to_data(num_path)
+                num_packets = audio_to_data(num_path)
                 conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None))
             except Exception as e:
                 conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
@@ -158,5 +158,5 @@ async def check_bind_device(conn):
         text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
         await send_stt_message(conn, text)
         music_path = "config/assets/bind_not_found.wav"
-        opus_packets, _ = audio_to_data(music_path)
+        opus_packets = audio_to_data(music_path)
         conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text))

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

@@ -16,8 +16,10 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
         conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
         conn.tts.tts_audio_first_sentence = False
         pre_buffer = True
+        await send_tts_message(conn, "start", None)
 
-    await send_tts_message(conn, "sentence_start", text)
+    if sentenceType == SentenceType.FIRST:
+        await send_tts_message(conn, "sentence_start", text)
 
     await sendAudio(conn, audios, pre_buffer)
 
@@ -31,44 +33,51 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
 
 # 播放音频
 async def sendAudio(conn, audios, pre_buffer=True):
-    if audios is None or len(audios) == 0:
+    if audios is None:
         return
-    # 流控参数优化
+    # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出
-    frame_duration = 60  # 帧时长（毫秒），匹配 Opus 编码
+    if isinstance(audios ,bytes):
-    start_time = time.perf_counter()
+        await conn.websocket.send(audios)
-    play_position = 0
-
-    # 仅当第一句话时执行预缓冲
-    if pre_buffer:
-        pre_buffer_frames = min(3, len(audios))
-        for i in range(pre_buffer_frames):
-            await conn.websocket.send(audios[i])
-        remaining_audios = audios[pre_buffer_frames:]
     else:
-        remaining_audios = audios
+        if audios is None or len(audios) == 0:
+            return
+        # 流控参数优化
+        frame_duration = 60  # 帧时长（毫秒），匹配 Opus 编码
+        start_time = time.perf_counter()
+        play_position = 0
+        # 仅当第一句话时执行预缓冲
+        if pre_buffer:
+            pre_buffer_frames = min(3, len(audios))
+            for i in range(pre_buffer_frames):
+                await conn.websocket.send(audios[i])
+            remaining_audios = audios[pre_buffer_frames:]
+        else:
+            remaining_audios = audios
 
-    # 播放剩余音频帧
+        # 播放剩余音频帧
-    for opus_packet in remaining_audios:
+        for opus_packet in remaining_audios:
-        if conn.client_abort:
+            if conn.client_abort:
-            break
+                break
 
-        # 重置没有声音的状态
+            # 重置没有声音的状态
-        conn.last_activity_time = time.time() * 1000
+            conn.last_activity_time = time.time() * 1000
 
-        # 计算预期发送时间
+            # 计算预期发送时间
-        expected_time = start_time + (play_position / 1000)
+            expected_time = start_time + (play_position / 1000)
-        current_time = time.perf_counter()
+            current_time = time.perf_counter()
-        delay = expected_time - current_time
+            delay = expected_time - current_time
-        if delay > 0:
+            if delay > 0:
-            await asyncio.sleep(delay)
+                await asyncio.sleep(delay)
 
-        await conn.websocket.send(opus_packet)
+            await conn.websocket.send(opus_packet)
 
-        play_position += frame_duration
+            play_position += frame_duration
 
 
 async def send_tts_message(conn, state, text=None):
     """发送 TTS 状态消息"""
+    if text is None and state == "sentence_start":
+        return
     message = {"type": "tts", "state": state, "session_id": conn.session_id}
     if text is not None:
         message["text"] = textUtils.check_emoji(text)

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

@@ -268,11 +268,7 @@ class TTSProvider(TTSProviderBase):
                     )
                     if message.content_file and os.path.exists(message.content_file):
                         # 先处理文件音频数据
-                        file_audio = self._process_audio_file(message.content_file)
+                        self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
-                        self.before_stop_play_files.append(
-                            (file_audio, message.content_detail)
-                        )
-
                 if message.sentence_type == SentenceType.LAST:
                     try:
                         logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -422,7 +418,6 @@ class TTSProvider(TTSProviderBase):
 
     async def _start_monitor_tts_response(self):
         """监听TTS响应"""
-        opus_datas_cache = []
         is_first_sentence = True
         first_sentence_segment_count = 0  # 添加计数器
         try:
@@ -448,25 +443,15 @@ class TTSProvider(TTSProviderBase):
                             elif event_name == "SentenceBegin":
                                 opus_datas_cache = []
                             elif event_name == "SentenceEnd":
-                                if (
+                                # 发送缓存的数据
-                                    not is_first_sentence
+                                if self.conn.tts_MessageText:
-                                    or first_sentence_segment_count > 10
+                                    logger.bind(tag=TAG).info(
-                                ):
+                                        f"句子语音生成成功： {self.conn.tts_MessageText}"
-                                    # 发送缓存的数据
+                                    )
-                                    if self.conn.tts_MessageText:
+                                    self.tts_audio_queue.put(
-                                        logger.bind(tag=TAG).info(
+                                        (SentenceType.FIRST, [], self.conn.tts_MessageText)
-                                            f"句子语音生成成功： {self.conn.tts_MessageText}"
+                                    )
-                                        )
+                                    self.conn.tts_MessageText = None
-                                        self.tts_audio_queue.put(
-                                            (SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText)
-                                        )
-                                        self.conn.tts_MessageText = None
-                                    else:
-                                        self.tts_audio_queue.put(
-                                            (SentenceType.MIDDLE, opus_datas_cache, None)
-                                        )
-                                # 第一句话结束后，将标志设置为False
-                                is_first_sentence = False
                             elif event_name == "SynthesisCompleted":
                                 logger.bind(tag=TAG).debug(f"会话结束～～")
                                 self._process_before_stop_play_files()
@@ -477,22 +462,7 @@ class TTSProvider(TTSProviderBase):
                     # 二进制消息（音频数据）
                     elif isinstance(msg, (bytes, bytearray)):
                         logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
-                        opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False)
+                        self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus)
-                        logger.bind(tag=TAG).debug(
-                            f"推送数据到队列里面帧数～～{len(opus_datas)}"
-                        )
-                        if is_first_sentence:
-                            first_sentence_segment_count += 1
-                            if first_sentence_segment_count <= 6:
-                                self.tts_audio_queue.put(
-                                    (SentenceType.MIDDLE, opus_datas, None)
-                                )
-                            else:
-                                opus_datas_cache.extend(opus_datas)
-                        else:
-                            # 后续句子缓存
-                            opus_datas_cache.extend(opus_datas)
-
                 except websockets.ConnectionClosed:
                     logger.bind(tag=TAG).warning("WebSocket连接已关闭")
                     break
@@ -512,8 +482,10 @@ class TTSProvider(TTSProviderBase):
         finally:
             self._monitor_task = None
 
-    def to_tts(self, text: str) -> list:
+    def to_tts_stream(self, text: str, opus_handler=None) -> None:
         """非流式TTS处理，用于测试及保存音频文件的场景"""
+        if opus_handler is None:
+            opus_handler = self.handle_opus
         try:
             # 创建新的事件循环
             loop = asyncio.new_event_loop()
@@ -616,10 +588,10 @@ class TTSProvider(TTSProviderBase):
                         msg = await ws.recv()
                         if isinstance(msg, (bytes, bytearray)):
                             # 编码为Opus并收集
-                            opus_frames = self.opus_encoder.encode_pcm_to_opus(
+                            self.opus_encoder.encode_pcm_to_opus_stream(
-                                msg, False
+                                msg, False, opus_handler
                             )
-                            audio_data.extend(opus_frames)
+                            # audio_data.extend(opus_frames)
                         elif isinstance(msg, str):
                             data = json.loads(msg)
                             header = data.get("header", {})
@@ -647,7 +619,5 @@ class TTSProvider(TTSProviderBase):
             loop.run_until_complete(_generate_audio())
             loop.close()
 
-            return audio_data
         except Exception as e:
             logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
-            return []

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

@@ -4,12 +4,15 @@ import queue
 import uuid
 import asyncio
 import threading
+from typing import Callable, Any
+
 from core.utils import p3
-from datetime import datetime
+import time
 from core.utils import textUtils
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from core.utils.util import audio_to_data, audio_bytes_to_data
+from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption
+from core.utils.util import audio_to_data, audio_bytes_to_data, audio_bytes_to_data_stream, audio_to_data_stream
 from core.utils.tts import MarkdownCleaner
 from core.utils.output_counter import add_device_output
 from core.handle.reportHandle import enqueue_tts_report
@@ -50,11 +53,9 @@ class TTSProviderBase(ABC):
             "；",
             ";",
             "：",
-            "~",
         )
         self.first_sentence_punctuations = (
             "，",
-            "～",
             "~",
             "、",
             ",",
@@ -70,6 +71,8 @@ class TTSProviderBase(ABC):
         self.tts_stop_request = False
         self.processed_chars = 0
         self.is_first_sentence = True
+        self.flow_controller = FlowControlConfig.create_flow_controller()
+        self.flow_control_enabled = config.get("enable_flow_control", True)
 
     def generate_filename(self, extension=".wav"):
         return os.path.join(
@@ -77,7 +80,20 @@ class TTSProviderBase(ABC):
             f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
         )
 
-    def to_tts(self, text):
+    def handle_opus(self, opus_data: bytes):
+        logger.bind(tag=TAG).debug(
+            f"推送数据到队列里面帧数～～ {len(opus_data)}"
+        )
+        self.tts_audio_queue.put(
+            (SentenceType.MIDDLE, opus_data, None)
+        )
+
+    def handle_audio_file(self, file_audio: bytes, text):
+        self.before_stop_play_files.append(
+            (file_audio, text)
+        )
+
+    def to_tts_stream(self, text, opus_handler=handle_opus) -> None:
         text = MarkdownCleaner.clean_markdown(text)
         max_repeat_time = 5
         if self.delete_audio_file:
@@ -86,10 +102,13 @@ class TTSProviderBase(ABC):
                 try:
                     audio_bytes = asyncio.run(self.text_to_speak(text, None))
                     if audio_bytes:
-                        audio_datas, _ = audio_bytes_to_data(
+                        self.tts_audio_queue.put(
-                            audio_bytes, file_type=self.audio_file_type, is_opus=True
+                            (SentenceType.FIRST, None, text)
                         )
-                        return audio_datas
+                        audio_bytes_to_data_stream(
+                            audio_bytes, file_type=self.audio_file_type, is_opus=True, callback=opus_handler
+                        )
+                        max_repeat_time = 0
                     else:
                         max_repeat_time -= 1
                 except Exception as e:
@@ -129,8 +148,10 @@ class TTSProviderBase(ABC):
                     logger.bind(tag=TAG).error(
                         f"语音生成失败: {text}，请检查网络或服务是否正常"
                     )
-
+                    self.tts_audio_queue.put(
-                return tmp_file
+                        (SentenceType.FIRST, None, text)
+                    )
+                self._process_audio_file_stream(tmp_file, callback=opus_handler)
             except Exception as e:
                 logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
                 return None
@@ -139,13 +160,13 @@ class TTSProviderBase(ABC):
     async def text_to_speak(self, text, output_file):
         pass
 
-    def audio_to_pcm_data(self, audio_file_path):
+    def audio_to_pcm_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=None):
         """音频文件转换为PCM编码"""
-        return audio_to_data(audio_file_path, is_opus=False)
+        return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback)
 
-    def audio_to_opus_data(self, audio_file_path):
+    def audio_to_opus_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=None):
         """音频文件转换为Opus编码"""
-        return audio_to_data(audio_file_path, is_opus=True)
+        return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback)
 
     def tts_one_sentence(
         self,
@@ -212,30 +233,14 @@ class TTSProviderBase(ABC):
                     self.tts_text_buff.append(message.content_detail)
                     segment_text = self._get_segment_text()
                     if segment_text:
-                        if self.delete_audio_file:
+                        self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
-                            audio_datas = self.to_tts(segment_text)
-                            if audio_datas:
-                                self.tts_audio_queue.put(
-                                    (message.sentence_type, audio_datas, segment_text)
-                                )
-                        else:
-                            tts_file = self.to_tts(segment_text)
-                            if tts_file:
-                                audio_datas = self._process_audio_file(tts_file)
-                                self.tts_audio_queue.put(
-                                    (message.sentence_type, audio_datas, segment_text)
-                                )
                 elif ContentType.FILE == message.content_type:
-                    self._process_remaining_text()
+                    self._process_remaining_text_stream(opus_handler=self.handle_opus)
                     tts_file = message.content_file
                     if tts_file and os.path.exists(tts_file):
-                        audio_datas = self._process_audio_file(tts_file)
+                        self._process_audio_file_stream(tts_file, callback=self.handle_opus)
-                        self.tts_audio_queue.put(
-                            (message.sentence_type, audio_datas, message.content_detail)
-                        )
-
                 if message.sentence_type == SentenceType.LAST:
-                    self._process_remaining_text()
+                    self._process_remaining_text_stream(opus_handler=self.handle_opus)
                     self.tts_audio_queue.put(
                         (message.sentence_type, [], message.content_detail)
                     )
@@ -253,26 +258,122 @@ class TTSProviderBase(ABC):
             text = None
             try:
                 try:
-                    sentence_type, audio_datas, text = self.tts_audio_queue.get(
+                    sentence_type, audio_datas, text = self.tts_audio_queue.get(timeout=0.1)
-                        timeout=1
-                    )
                 except queue.Empty:
                     if self.conn.stop_event.is_set():
                         break
                     continue
-                future = asyncio.run_coroutine_threadsafe(
+
-                    sendAudioMessage(self.conn, sentence_type, audio_datas, text),
+                # 如果启用了流控
-                    self.conn.loop,
+                if self.flow_control_enabled:
-                )
+                    # 计算音频数据的帧数
-                future.result()
+                    if isinstance(audio_datas, bytes):
-                if self.conn.max_output_size > 0 and text:
+                        frame_count = 1  # 单个字节流作为一帧
-                    add_device_output(self.conn.headers.get("device-id"), len(text))
+                    elif isinstance(audio_datas, (list, tuple)):
-                enqueue_tts_report(self.conn, text, audio_datas)
+                        frame_count = len(audio_datas)
+                    else:
+                        frame_count = 0
+
+                    # 流控检查
+                    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).warning(
+                                    f"流控等待超时或收到停止信号，跳过音频发送: {text}"
+                                )
+                                break
+
+                            # 短暂等待后重试
+                            time.sleep(retry_interval)
+
+                            # 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:
+                            # 可以发送，记录发送的帧数
+                            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()
+
+                            # 记录输出和报告
+                            if self.conn.max_output_size > 0 and text:
+                                add_device_output(self.conn.headers.get("device-id"), len(text))
+                            enqueue_tts_report(self.conn, text, audio_datas)
+
+                            # 输出流控状态（调试用）
+                            # if frame_count > 0:  # 只在较大的音频块时输出状态
+                            #     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()
+                else:
+                    # 未启用流控，直接发送
+                    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))
+                    enqueue_tts_report(self.conn, text, audio_datas)
+
             except Exception as e:
                 logger.bind(tag=TAG).error(
-                    f"audio_play_priority priority_thread: {text} {e}"
+                    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
+        elif isinstance(audio_datas, (list, tuple)):
+            frame_count = len(audio_datas)
+        else:
+            frame_count = 0
+
+        if frame_count > 0:
+            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
 
@@ -323,22 +424,19 @@ class TTSProviderBase(ABC):
         else:
             return None
 
-    def _process_audio_file(self, tts_file):
+    def _process_audio_file_stream(self, tts_file, callback: Callable[[Any], Any]) -> None:
         """处理音频文件并转换为指定格式
 
         Args:
             tts_file: 音频文件路径
-            content_detail: 内容详情
+            callback: 文件处理函数
-
-        Returns:
-            tuple: (sentence_type, audio_datas, content_detail)
         """
         if tts_file.endswith(".p3"):
-            audio_datas, _ = p3.decode_opus_from_file(tts_file)
+            p3.decode_opus_from_file_stream(tts_file, callback=callback)
         elif self.conn.audio_format == "pcm":
-            audio_datas, _ = self.audio_to_pcm_data(tts_file)
+            self.audio_to_pcm_data_stream(tts_file, callback=callback)
         else:
-            audio_datas, _ = self.audio_to_opus_data(tts_file)
+            self.audio_to_opus_data_stream(tts_file, callback=callback)
 
         if (
             self.delete_audio_file
@@ -347,7 +445,6 @@ class TTSProviderBase(ABC):
             and tts_file.startswith(self.output_file)
         ):
             os.remove(tts_file)
-        return audio_datas
 
     def _process_before_stop_play_files(self):
         for audio_datas, text in self.before_stop_play_files:
@@ -355,7 +452,7 @@ class TTSProviderBase(ABC):
         self.before_stop_play_files.clear()
         self.tts_audio_queue.put((SentenceType.LAST, [], None))
 
-    def _process_remaining_text(self):
+    def _process_remaining_text_stream(self, opus_handler=handle_opus):
         """处理剩余的文本并生成语音
 
         Returns:
@@ -366,18 +463,7 @@ class TTSProviderBase(ABC):
         if remaining_text:
             segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
             if segment_text:
-                if self.delete_audio_file:
+                self.to_tts_stream(segment_text, opus_handler=opus_handler)
-                    audio_datas = self.to_tts(segment_text)
-                    if audio_datas:
-                        self.tts_audio_queue.put(
-                            (SentenceType.MIDDLE, audio_datas, segment_text)
-                        )
-                else:
-                    tts_file = self.to_tts(segment_text)
-                    audio_datas = self._process_audio_file(tts_file)
-                    self.tts_audio_queue.put(
-                        (SentenceType.MIDDLE, audio_datas, segment_text)
-                    )
                 self.processed_chars += len(full_text)
                 return True
         return False

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

@@ -4,6 +4,8 @@ import json
 import queue
 import asyncio
 import traceback
+from typing import Callable, Any
+
 import websockets
 from core.utils.tts import MarkdownCleaner
 from config.logger import setup_logging
@@ -266,11 +268,7 @@ class TTSProvider(TTSProviderBase):
                     )
                     if message.content_file and os.path.exists(message.content_file):
                         # 先处理文件音频数据
-                        file_audio = self._process_audio_file(message.content_file)
+                        self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
-                        self.before_stop_play_files.append(
-                            (file_audio, message.content_detail)
-                        )
-
                 if message.sentence_type == SentenceType.LAST:
                     try:
                         logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -458,21 +456,7 @@ class TTSProvider(TTSProviderBase):
                         and res.header.message_type == AUDIO_ONLY_RESPONSE
                     ):
                         logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
-                        opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
+                        self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
-                        logger.bind(tag=TAG).debug(
-                            f"推送数据到队列里面帧数～～{len(opus_datas)}"
-                        )
-                        if is_first_sentence:
-                            first_sentence_segment_count += 1
-                            if first_sentence_segment_count <= 6:
-                                self.tts_audio_queue.put(
-                                    (SentenceType.MIDDLE, opus_datas, None)
-                                )
-                            else:
-                                opus_datas_cache.extend(opus_datas)
-                        else:
-                            # 后续句子缓存
-                            opus_datas_cache.extend(opus_datas)
                     elif res.optional.event == EVENT_TTSSentenceEnd:
                         logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
                         if not is_first_sentence or first_sentence_segment_count > 10:
@@ -655,19 +639,21 @@ class TTSProvider(TTSProviderBase):
             )
         )
 
-    def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False):
+    def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None):
-        opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end)
+        return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback)
-        return opus_datas
 
-    def to_tts(self, text: str) -> list:
+    def to_tts_stream(self, text: str, opus_handler=None) -> None:
         """非流式生成音频数据，用于生成音频及测试场景
 
         Args:
             text: 要转换的文本
+            opus_handler: opus数据处理方法
 
         Returns:
             list: 音频数据列表
         """
+        if opus_handler is None:
+            opus_handler = self.handle_opus
         try:
             # 创建事件循环
             loop = asyncio.new_event_loop()
@@ -676,9 +662,6 @@ class TTSProvider(TTSProviderBase):
             # 生成会话ID
             session_id = uuid.uuid4().__str__().replace("-", "")
 
-            # 存储音频数据
-            audio_data = []
-
             async def _generate_audio():
                 # 创建新的WebSocket连接
                 ws_header = {
@@ -741,8 +724,7 @@ class TTSProvider(TTSProviderBase):
                             res.optional.event == EVENT_TTSResponse
                             and res.header.message_type == AUDIO_ONLY_RESPONSE
                         ):
-                            opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
+                            self.wav_to_opus_data_audio_raw_stream(res.payload, callback=opus_handler)
-                            audio_data.extend(opus_datas)
                         elif res.optional.event == EVENT_SessionFinished:
                             break
 
@@ -757,8 +739,6 @@ class TTSProvider(TTSProviderBase):
             loop.run_until_complete(_generate_audio())
             loop.close()
 
-            return audio_data
-
         except Exception as e:
             logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
             return []

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

@@ -65,14 +65,10 @@ class TTSProvider(TTSProviderBase):
                     )
                     if message.content_file and os.path.exists(message.content_file):
                         # 先处理文件音频数据
-                        file_audio = self._process_audio_file(message.content_file)
+                        self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
-                        self.before_stop_play_files.append(
-                            (file_audio, message.content_detail)
-                        )
-
                 if message.sentence_type == SentenceType.LAST:
                     # 处理剩余的文本
-                    self._process_remaining_text(True)
+                    self._process_remaining_text_stream(True)
 
             except queue.Empty:
                 continue
@@ -81,7 +77,7 @@ class TTSProvider(TTSProviderBase):
                     f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
                 )
 
-    def _process_remaining_text(self, is_last=False):
+    def _process_remaining_text_stream(self, is_last=False):
         """处理剩余的文本并生成语音
 
         Returns:
@@ -237,15 +233,15 @@ class TTSProvider(TTSProviderBase):
             logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
             self.tts_audio_queue.put((SentenceType.LAST, [], None))
 
-    def to_tts(self, text: str) -> list:
+    def to_tts_stream(self, text: str, opus_handler=None) -> list:
         """非流式TTS处理，用于测试及保存音频文件的场景
 
         Args:
             text: 要转换的文本
-
+            opus_handler: opus数据处理方法
-        Returns:
-            list: 返回opus编码后的音频数据列表
         """
+        if opus_handler is None:
+            opus_handler = self.handle_opus
         start_time = time.time()
         text = MarkdownCleaner.clean_markdown(text)
 
@@ -295,14 +291,9 @@ class TTSProvider(TTSProviderBase):
                         # 最后一帧可能不足，用0填充
                         frame = frame + b"\x00" * (frame_bytes - len(frame))
 
-                    opus = self.opus_encoder.encode_pcm_to_opus(
+                    self.opus_encoder.encode_pcm_to_opus_stream(
-                        frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
+                        frame, end_of_stream=(i + frame_bytes >= len(pcm_data)), callback=opus_handler
                     )
-                    if opus:
-                        opus_datas.extend(opus)
-
-                return opus_datas
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
-            return []

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

@@ -0,0 +1,200 @@
+"""
+音频流控模块
+包含令牌桶算法和音频流控制器的实现
+"""
+
+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/core/utils/opus_encoder_utils.py

@@ -7,7 +7,7 @@ import logging
 import traceback
 
 import numpy as np
-from typing import List, Optional
+from typing import List, Optional, Callable, Any
 from opuslib_next import Encoder
 from opuslib_next import constants
 
@@ -103,6 +103,51 @@ class OpusEncoderUtils:
 
         return opus_packets
 
+    def encode_pcm_to_opus_stream(self, pcm_data: bytes, end_of_stream: bool, callback: Callable[[Any], Any]):
+        """
+        将PCM数据编码为Opus格式，以流式方式进行处理
+
+        Args:
+            pcm_data: PCM字节数据
+            end_of_stream: 是否为流的结束,
+            callback: opus处理方法
+
+        Returns:
+            Opus数据包列表
+        """
+        # 将字节数据转换为short数组
+        new_samples = self._convert_bytes_to_shorts(pcm_data)
+
+        # 校验PCM数据
+        self._validate_pcm_data(new_samples)
+
+        # 将新数据追加到缓冲区
+        self.buffer = np.append(self.buffer, new_samples)
+
+        offset = 0
+
+        # 处理所有完整帧
+        while offset <= len(self.buffer) - self.total_frame_size:
+            frame = self.buffer[offset : offset + self.total_frame_size]
+            output = self._encode(frame)
+            if output:
+                callback(output)
+            offset += self.total_frame_size
+
+        # 保留未处理的样本
+        self.buffer = self.buffer[offset:]
+
+        # 流结束时处理剩余数据
+        if end_of_stream and len(self.buffer) > 0:
+            # 创建最后一帧并用0填充
+            last_frame = np.zeros(self.total_frame_size, dtype=np.int16)
+            last_frame[: len(self.buffer)] = self.buffer
+
+            output = self._encode(last_frame)
+            if output:
+                callback(output)
+            self.buffer = np.array([], dtype=np.int16)
+
     def _encode(self, frame: np.ndarray) -> Optional[bytes]:
         """编码一帧音频数据"""
         try:

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

@@ -1,14 +1,11 @@
 import struct
+from typing import Callable, Any
 
 def decode_opus_from_file(input_file):
     """
-    从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。
+    从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表。
     """
     opus_datas = []
-    total_frames = 0
-    sample_rate = 16000  # 文件采样率
-    frame_duration_ms = 60  # 帧时长
-    frame_size = int(sample_rate * frame_duration_ms / 1000)
 
     with open(input_file, 'rb') as f:
         while True:
@@ -26,22 +23,36 @@ def decode_opus_from_file(input_file):
                 raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.")
 
             opus_datas.append(opus_data)
-            total_frames += 1
 
-    # 计算总时长
+    return opus_datas
-    total_duration = (total_frames * frame_duration_ms) / 1000.0
+
-    return opus_datas, total_duration
+def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]):
+    """
+    从p3文件中解码 Opus 数据，由 callback 处理 Opus 数据包。
+    """
+    with open(input_file, 'rb') as f:
+        while True:
+            # 读取头部（4字节）：[1字节类型，1字节保留，2字节长度]
+            header = f.read(4)
+            if not header:
+                break
+
+            # 解包头部信息
+            _, _, data_len = struct.unpack('>BBH', header)
+
+            # 根据头部指定的长度读取 Opus 数据
+            opus_data = f.read(data_len)
+            if len(opus_data) != data_len:
+                raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.")
+
+            callback(opus_data)
 
 def decode_opus_from_bytes(input_bytes):
     """
-    从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。
+    从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表。
     """
     import io
     opus_datas = []
-    total_frames = 0
-    sample_rate = 16000  # 文件采样率
-    frame_duration_ms = 60  # 帧时长
-    frame_size = int(sample_rate * frame_duration_ms / 1000)
 
     f = io.BytesIO(input_bytes)
     while True:
@@ -53,7 +64,21 @@ def decode_opus_from_bytes(input_bytes):
         if len(opus_data) != data_len:
             raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.")
         opus_datas.append(opus_data)
-        total_frames += 1
 
-    total_duration = (total_frames * frame_duration_ms) / 1000.0
+    return opus_datas
-    return opus_datas, total_duration
+
+def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]):
+    """
+    从p3二进制数据中解码 Opus 数据，由 callback 处理 Opus 数据包。
+    """
+
+    f = io.BytesIO(input_bytes)
+    while True:
+        header = f.read(4)
+        if not header:
+            break
+        _, _, data_len = struct.unpack('>BBH', header)
+        opus_data = f.read(data_len)
+        if len(opus_data) != data_len:
+            raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.")
+        callback(opus_data)

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

@@ -5,6 +5,8 @@ import re
 import os
 import wave
 from io import BytesIO
+from typing import Callable, Any
+
 from core.utils import p3
 import numpy as np
 import requests
@@ -224,12 +226,26 @@ def audio_to_data(audio_file_path, is_opus=True):
     # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
     audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
 
-    # 音频时长(秒)
+    # 获取原始PCM数据（16位小端）
-    duration = len(audio) / 1000.0
+    raw_data = audio.raw_data
+    return pcm_to_data(raw_data, is_opus)
+
+def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any], Any]=None) -> None:
+    # 获取文件后缀名
+    file_type = os.path.splitext(audio_file_path)[1]
+    if file_type:
+        file_type = file_type.lstrip(".")
+    # 读取音频文件，-nostdin 参数：不要从标准输入读取数据，否则FFmpeg会阻塞
+    audio = AudioSegment.from_file(
+        audio_file_path, format=file_type, parameters=["-nostdin"]
+    )
+
+    # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
+    audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
 
     # 获取原始PCM数据（16位小端）
     raw_data = audio.raw_data
-    return pcm_to_data(raw_data, is_opus), duration
+    pcm_to_data_stream(raw_data, is_opus,callback)
 
 
 def audio_bytes_to_data(audio_bytes, file_type, is_opus=True):
@@ -245,9 +261,24 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True):
             BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"]
         )
         audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
-        duration = len(audio) / 1000.0
         raw_data = audio.raw_data
-        return pcm_to_data(raw_data, is_opus), duration
+        return pcm_to_data(raw_data, is_opus)
+
+def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None:
+    """
+    直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3
+    """
+    if file_type == "p3":
+        # 直接用p3解码
+        return p3.decode_opus_from_bytes_stream(audio_bytes, callback)
+    else:
+        # 其他格式用pydub
+        audio = AudioSegment.from_file(
+            BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"]
+        )
+        audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
+        raw_data = audio.raw_data
+        pcm_to_data_stream(raw_data, is_opus, callback)
 
 
 def pcm_to_data(raw_data, is_opus=True):
@@ -280,6 +311,32 @@ def pcm_to_data(raw_data, is_opus=True):
 
     return datas
 
+def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any]=None):
+    # 初始化Opus编码器
+    encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO)
+
+    # 编码参数
+    frame_duration = 60  # 60ms per frame
+    frame_size = int(16000 * frame_duration / 1000)  # 960 samples/frame
+
+    # 按帧处理所有音频数据（包括最后一帧可能补零）
+    for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
+        # 获取当前帧的二进制数据
+        chunk = raw_data[i : i + frame_size * 2]
+
+        # 如果最后一帧不足，补零
+        if len(chunk) < frame_size * 2:
+            chunk += b"\x00" * (frame_size * 2 - len(chunk))
+
+        if is_opus:
+            # 转换为numpy数组处理
+            np_frame = np.frombuffer(chunk, dtype=np.int16)
+            # 编码Opus数据
+            frame_data = encoder.encode(np_frame.tobytes(), frame_size)
+            callback(frame_data)
+        else:
+            frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
+            callback(frame_data)
 
 def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
     """