From ac1e19f621be1cb7a83a580780c6bb44b249d13c Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 14:10:55 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=B5=81=E5=BC=8F=E5=A4=84?= =?UTF-8?q?=E7=90=86opus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 61 +++--- .../core/providers/tts/aliyun_stream.py | 38 ++-- .../xiaozhi-server/core/providers/tts/base.py | 118 +++++++++-- .../core/utils/audio_flow_control.py | 200 ++++++++++++++++++ 4 files changed, 352 insertions(+), 65 deletions(-) create mode 100644 main/xiaozhi-server/core/utils/audio_flow_control.py diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 0ca36e1e..09c54b2d 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -31,44 +31,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 - # 流控参数优化 - 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:] + # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出 + if isinstance(audios ,bytes): + await conn.websocket.send(audios) 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: - if conn.client_abort: - break + # 播放剩余音频帧 + for opus_packet in remaining_audios: + if conn.client_abort: + break - # 重置没有声音的状态 - conn.last_activity_time = time.time() * 1000 + # 重置没有声音的状态 + conn.last_activity_time = time.time() * 1000 - # 计算预期发送时间 - expected_time = start_time + (play_position / 1000) - current_time = time.perf_counter() - delay = expected_time - current_time - if delay > 0: - await asyncio.sleep(delay) + # 计算预期发送时间 + expected_time = start_time + (play_position / 1000) + current_time = time.perf_counter() + delay = expected_time - current_time + if delay > 0: + await asyncio.sleep(delay) - await conn.websocket.send(opus_packet) + 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: + return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: message["text"] = textUtils.check_emoji(text) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 0bfb367b..d7f16140 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -422,7 +422,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: @@ -458,13 +457,9 @@ class TTSProvider(TTSProviderBase): f"句子语音生成成功： {self.conn.tts_MessageText}" ) self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText) + (SentenceType.MIDDLE, [], 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": @@ -477,22 +472,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) - 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) - + self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus) except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") break @@ -616,10 +596,10 @@ class TTSProvider(TTSProviderBase): msg = await ws.recv() if isinstance(msg, (bytes, bytearray)): # 编码为Opus并收集 - opus_frames = self.opus_encoder.encode_pcm_to_opus( - msg, False + self.opus_encoder.encode_pcm_to_opus_stream( + msg, False, self.handle_opus ) - audio_data.extend(opus_frames) + # audio_data.extend(opus_frames) elif isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) @@ -651,3 +631,11 @@ class TTSProvider(TTSProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") return [] + + def handle_opus(self, opus_data: bytes): + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～" + ) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_data, None) + ) \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 87bb64f6..55ce989b 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -5,10 +5,11 @@ import uuid import asyncio import threading 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.audio_flow_control import FlowControlConfig, simulate_device_consumption from core.utils.util import audio_to_data, audio_bytes_to_data from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output @@ -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( @@ -253,26 +256,115 @@ class TTSProviderBase(ABC): text = None try: try: - sentence_type, audio_datas, text = self.tts_audio_queue.get( - timeout=1 - ) + sentence_type, audio_datas, text = self.tts_audio_queue.get(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, - ) - 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 self.flow_control_enabled: + # 计算音频数据的帧数 + 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: + 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) + + # 更新设备消费估计（这里假设设备以恒定速率消费） + # 实际应用中可能需要从设备端获取真实的消费情况 + estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) + self.flow_controller.update_device_consumption(estimated_consumption) + 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 > 10: # 只在较大的音频块时输出状态 + status = self.flow_controller.get_status() + logger.bind(tag=TAG).debug( + f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + f"可用令牌={status['available_tokens']}, 发送文本: {text[:20]}..." + ) + else: + # 没有音频数据，直接发送 + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self.conn, 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 diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py new file mode 100644 index 00000000..74a5585f --- /dev/null +++ b/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 = 1000 # 设备端最大缓冲帧数 + DEFAULT_REFILL_RATE = 20 # 默认令牌补充速率（帧/秒） + DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） + DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） + + # 预缓冲参数 + 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 + ) \ No newline at end of file