From 1988bced6068137e4179ac87b7a273eb4296cbdf Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 22 Aug 2025 17:10:36 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=98=E5=8C=96=E6=B5=81=E6=8E=A7?= =?UTF-8?q?=E9=9F=B3=E9=A2=91=E6=92=AD=E6=94=BE=20wechat=E8=81=8A=E5=A4=A9?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F=E9=94=99=E8=AF=AFSTT=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 56 ++++- .../core/providers/tts/aliyun_stream.py | 1 - .../xiaozhi-server/core/providers/tts/base.py | 92 ++------ .../providers/tts/huoshan_double_stream.py | 1 - .../core/providers/tts/index_stream.py | 1 - .../core/providers/tts/linkerai.py | 1 - .../core/utils/audio_flow_control.py | 201 ------------------ .../plugins_func/functions/play_music.py | 1 - 8 files changed, 67 insertions(+), 287 deletions(-) delete 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 bb85c007..8604279a 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/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部分 diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 32e4d50f..734681e2 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/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文本处理线程") diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 35d80fb6..6a29f1f5 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/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 diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index ac6301f9..463c16fc 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/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: diff --git a/main/xiaozhi-server/core/providers/tts/index_stream.py b/main/xiaozhi-server/core/providers/tts/index_stream.py index d562e802..6f6b829c 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/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() diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index a855e8ed..875ed0b1 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/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() diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py deleted file mode 100644 index 958e68a7..00000000 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ /dev/null @@ -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 - ) diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 2cbc4018..dd967d0a 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/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":