From 8b81b918cc33b20ff9d5266096bf87cd9e5c5091 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 13:49:12 +0800 Subject: [PATCH 01/10] =?UTF-8?q?=E7=A7=BB=E9=99=A4=E6=9C=AA=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E7=9A=84total=5Fduration=E8=BF=94=E5=9B=9E=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/handle/helloHandle.py | 2 +- .../core/handle/receiveAudioHandle.py | 6 +++--- main/xiaozhi-server/core/providers/tts/base.py | 8 ++++---- main/xiaozhi-server/core/utils/p3.py | 17 ++--------------- main/xiaozhi-server/core/utils/util.py | 8 ++------ 5 files changed, 12 insertions(+), 29 deletions(-) diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index e4e836b4..c4f55841 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/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")) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index e6f39632..174c8b2b 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/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)) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 40570326..87bb64f6 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -86,7 +86,7 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - audio_datas, _ = audio_bytes_to_data( + audio_datas = audio_bytes_to_data( audio_bytes, file_type=self.audio_file_type, is_opus=True ) return audio_datas @@ -334,11 +334,11 @@ class TTSProviderBase(ABC): tuple: (sentence_type, audio_datas, content_detail) """ if tts_file.endswith(".p3"): - audio_datas, _ = p3.decode_opus_from_file(tts_file) + audio_datas = p3.decode_opus_from_file(tts_file) elif self.conn.audio_format == "pcm": - audio_datas, _ = self.audio_to_pcm_data(tts_file) + audio_datas = self.audio_to_pcm_data(tts_file) else: - audio_datas, _ = self.audio_to_opus_data(tts_file) + audio_datas = self.audio_to_opus_data(tts_file) if ( self.delete_audio_file diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index c75b968e..f196da0b 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -5,10 +5,6 @@ def decode_opus_from_file(input_file): 从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,11 +22,8 @@ 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 - # 计算总时长 - total_duration = (total_frames * frame_duration_ms) / 1000.0 - return opus_datas, total_duration + return opus_datas def decode_opus_from_bytes(input_bytes): """ @@ -38,10 +31,6 @@ def decode_opus_from_bytes(input_bytes): """ 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 +42,5 @@ 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, total_duration \ No newline at end of file + return opus_datas \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index ce6f20ea..641e2655 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -224,12 +224,9 @@ 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) - # 音频时长(秒) - duration = len(audio) / 1000.0 - # 获取原始PCM数据（16位小端） 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(audio_bytes, file_type, is_opus=True): @@ -245,9 +242,8 @@ 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 pcm_to_data(raw_data, is_opus=True): From 75f077324704ff5ccf89fd882f1bb109239ec291 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 14:00:41 +0800 Subject: [PATCH 02/10] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86Opus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/utils/opus_encoder_utils.py | 47 ++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index 23d26c39..f9154da1 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/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: From faf2890695d1adbfedf2b0e0bfa7b8be0fbfc24b Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 17:04:46 +0800 Subject: [PATCH 03/10] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/utils/p3.py | 44 ++++++++++++++++++++++++++-- 1 file changed, 41 insertions(+), 3 deletions(-) diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index f196da0b..f2ee3239 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -1,8 +1,9 @@ import struct +from typing import Callable, Any def decode_opus_from_file(input_file): """ - 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 + 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表。 """ opus_datas = [] @@ -25,9 +26,30 @@ def decode_opus_from_file(input_file): return opus_datas +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 = [] @@ -43,4 +65,20 @@ def decode_opus_from_bytes(input_bytes): raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") opus_datas.append(opus_data) - return opus_datas \ No newline at end of file + return opus_datas + +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) From ac1e19f621be1cb7a83a580780c6bb44b249d13c Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 14:10:55 +0800 Subject: [PATCH 04/10] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=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 From 1b6cd01eafb59cb495bcb4a0b642f859dfcd83c1 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 14:53:52 +0800 Subject: [PATCH 05/10] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=96=B9=E6=B3=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/utils/util.py | 45 ++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 641e2655..088d8962 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/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 @@ -245,6 +247,23 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): raw_data = audio.raw_data return pcm_to_data(raw_data, is_opus) +def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]): + """ + 直接用音频二进制数据转为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) + return None + def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 @@ -276,6 +295,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): """ From 095fe72db083cd0e4ce0987fbfdf5bda52e38bbd Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 18:35:37 +0800 Subject: [PATCH 06/10] =?UTF-8?q?=E5=BE=85=E5=AE=8C=E5=96=84=E6=B5=81?= =?UTF-8?q?=E6=8E=A7=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 6 +- .../core/providers/tts/aliyun_stream.py | 10 +-- .../xiaozhi-server/core/providers/tts/base.py | 79 ++++++++++--------- .../core/utils/audio_flow_control.py | 4 +- main/xiaozhi-server/core/utils/util.py | 20 ++++- 5 files changed, 65 insertions(+), 54 deletions(-) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 09c54b2d..bf155ee9 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/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) @@ -74,7 +76,7 @@ async def sendAudio(conn, audios, pre_buffer=True): async def send_tts_message(conn, state, text=None): """发送 TTS 状态消息""" - if text is None: + if text is None and state == "sentence_start": return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index d7f16140..5b3938e9 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -630,12 +630,4 @@ class TTSProvider(TTSProviderBase): return audio_data 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 + return [] \ 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 55ce989b..d105a892 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -4,13 +4,15 @@ import queue import uuid import asyncio import threading +from typing import Callable, Any + from core.utils import p3 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.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 @@ -51,11 +53,9 @@ class TTSProviderBase(ABC): "；", ";", "：", - "~", ) self.first_sentence_punctuations = ( "，", - "～", "~", "、", ",", @@ -80,7 +80,15 @@ 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 to_tts(self, text, opus_handler=handle_opus) -> None: text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -89,10 +97,13 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - audio_datas = audio_bytes_to_data( - audio_bytes, file_type=self.audio_file_type, is_opus=True + self.tts_audio_queue.put( + (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: @@ -132,8 +143,10 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).error( f"语音生成失败: {text}，请检查网络或服务是否正常" ) - - return tmp_file + self.tts_audio_queue.put( + (SentenceType.FIRST, None, text) + ) + self._process_audio_file(tmp_file, callback=opus_handler) except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") return None @@ -142,13 +155,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, @@ -215,28 +228,12 @@ 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: - 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) - ) + self.to_tts(segment_text, opus_handler=self.handle_opus) elif ContentType.FILE == message.content_type: self._process_remaining_text() tts_file = message.content_file if tts_file and os.path.exists(tts_file): - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, message.content_detail) - ) - + self._process_audio_file(tts_file, callback=self.handle_opus) if message.sentence_type == SentenceType.LAST: self._process_remaining_text() self.tts_audio_queue.put( @@ -295,6 +292,11 @@ class TTSProviderBase(ABC): # 实际应用中可能需要从设备端获取真实的消费情况 estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) self.flow_controller.update_device_consumption(estimated_consumption) + status = self.flow_controller.get_status() + logger.bind(tag=TAG).debug( + f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + f"可用令牌={status['available_tokens']}..." + ) else: # 可以发送，记录发送的帧数 self.flow_controller.record_sent_frames(frame_count) @@ -312,11 +314,11 @@ class TTSProviderBase(ABC): enqueue_tts_report(self.conn, text, audio_datas) # 输出流控状态（调试用） - if frame_count > 10: # 只在较大的音频块时输出状态 + 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']}, 发送文本: {text[:20]}..." + f"可用令牌={status['available_tokens']}..." ) else: # 没有音频数据，直接发送 @@ -415,7 +417,7 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file): + def _process_audio_file(self, tts_file, callback: Callable[[Any], Any]): """处理音频文件并转换为指定格式 Args: @@ -426,11 +428,11 @@ class TTSProviderBase(ABC): 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 @@ -439,7 +441,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: @@ -466,7 +467,7 @@ class TTSProviderBase(ABC): ) else: tts_file = self.to_tts(segment_text) - audio_datas = self._process_audio_file(tts_file) + audio_datas = self._process_audio_file(tts_file, callback=self.handle_opus) self.tts_audio_queue.put( (SentenceType.MIDDLE, audio_datas, segment_text) ) diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index 74a5585f..ecdb5570 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -173,8 +173,8 @@ class FlowControlConfig: OPUS_FRAMES_PER_SECOND = 1000 / OPUS_FRAME_DURATION_MS # 每秒帧数 # 默认流控参数 - DEFAULT_MAX_DEVICE_BUFFER = 1000 # 设备端最大缓冲帧数 - DEFAULT_REFILL_RATE = 20 # 默认令牌补充速率（帧/秒） + DEFAULT_MAX_DEVICE_BUFFER = 40 # 设备端最大缓冲帧数 + DEFAULT_REFILL_RATE = 5 # 默认令牌补充速率（帧/秒） DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 088d8962..8aa2d712 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -230,6 +230,23 @@ def audio_to_data(audio_file_path, is_opus=True): 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 + pcm_to_data_stream(raw_data, is_opus,callback) + def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): """ @@ -247,7 +264,7 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): raw_data = audio.raw_data return pcm_to_data(raw_data, is_opus) -def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]): +def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: """ 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 """ @@ -262,7 +279,6 @@ def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callab 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) - return None def pcm_to_data(raw_data, is_opus=True): From ce6a711c8e5d619aa535a4d64159d88369a08091 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 10:09:28 +0800 Subject: [PATCH 07/10] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=B5=81=E6=8E=A7?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 24 +++++-------- .../xiaozhi-server/core/providers/tts/base.py | 36 ++++++++++--------- .../core/utils/audio_flow_control.py | 4 +-- 3 files changed, 30 insertions(+), 34 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 5b3938e9..94806924 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -447,21 +447,15 @@ class TTSProvider(TTSProviderBase): elif event_name == "SentenceBegin": opus_datas_cache = [] elif event_name == "SentenceEnd": - if ( - not is_first_sentence - or first_sentence_segment_count > 10 - ): - # 发送缓存的数据 - if self.conn.tts_MessageText: - logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" - ) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, [], self.conn.tts_MessageText) - ) - self.conn.tts_MessageText = None - # 第一句话结束后，将标志设置为False - is_first_sentence = False + # 发送缓存的数据 + if self.conn.tts_MessageText: + logger.bind(tag=TAG).info( + f"句子语音生成成功： {self.conn.tts_MessageText}" + ) + self.tts_audio_queue.put( + (SentenceType.FIRST, [], self.conn.tts_MessageText) + ) + self.conn.tts_MessageText = None elif event_name == "SynthesisCompleted": logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index d105a892..2e99557b 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -253,7 +253,7 @@ 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=0.1) except queue.Empty: if self.conn.stop_event.is_set(): break @@ -288,15 +288,14 @@ class TTSProviderBase(ABC): # 短暂等待后重试 time.sleep(retry_interval) - # 更新设备消费估计（这里假设设备以恒定速率消费） - # 实际应用中可能需要从设备端获取真实的消费情况 - estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) - self.flow_controller.update_device_consumption(estimated_consumption) - status = self.flow_controller.get_status() - logger.bind(tag=TAG).debug( - f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - f"可用令牌={status['available_tokens']}..." - ) + # 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) @@ -314,16 +313,19 @@ class TTSProviderBase(ABC): 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']}..." - ) + # 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( - sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self._send_audio_with_flow_control(sentence_type, audio_datas, text), self.conn.loop, ) future.result() diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index ecdb5570..102914c5 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -174,9 +174,9 @@ class FlowControlConfig: # 默认流控参数 DEFAULT_MAX_DEVICE_BUFFER = 40 # 设备端最大缓冲帧数 - DEFAULT_REFILL_RATE = 5 # 默认令牌补充速率（帧/秒） + DEFAULT_REFILL_RATE = OPUS_FRAMES_PER_SECOND # 默认令牌补充速率（帧/秒） DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） - DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） + DEFAULT_RETRY_INTERVAL = 0.06 # 流控重试间隔（秒） # 预缓冲参数 PRE_BUFFER_FRAMES = 3 # 预缓冲帧数 From 903c0e51eeb1f789337e40444e806f3aea674ae4 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 10:13:42 +0800 Subject: [PATCH 08/10] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E5=89=A9=E4=BD=99?= =?UTF-8?q?=E6=96=87=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/providers/tts/base.py | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 2e99557b..575e1325 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -461,18 +461,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: - 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, callback=self.handle_opus) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, audio_datas, segment_text) - ) + self.to_tts(segment_text, opus_handler=self.handle_opus) self.processed_chars += len(full_text) return True return False From abf1fa0ef4b72411f4059d623718f48b09277f16 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 11:34:57 +0800 Subject: [PATCH 09/10] =?UTF-8?q?=E8=B0=83=E6=95=B4=E5=AD=90=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 12 ++---- .../xiaozhi-server/core/providers/tts/base.py | 28 +++++++------ .../providers/tts/huoshan_double_stream.py | 40 +++++-------------- .../core/providers/tts/linkerai.py | 12 ++---- 4 files changed, 31 insertions(+), 61 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 94806924..52451cba 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/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.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) if message.sentence_type == SentenceType.LAST: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -486,7 +482,7 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式TTS处理，用于测试及保存音频文件的场景""" try: # 创建新的事件循环 @@ -591,7 +587,7 @@ class TTSProvider(TTSProviderBase): if isinstance(msg, (bytes, bytearray)): # 编码为Opus并收集 self.opus_encoder.encode_pcm_to_opus_stream( - msg, False, self.handle_opus + msg, False, opus_handler ) # audio_data.extend(opus_frames) elif isinstance(msg, str): @@ -621,7 +617,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 [] \ 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 575e1325..d55f434e 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -88,7 +88,12 @@ class TTSProviderBase(ABC): (SentenceType.MIDDLE, opus_data, None) ) - def to_tts(self, text, opus_handler=handle_opus) -> 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: @@ -146,7 +151,7 @@ class TTSProviderBase(ABC): self.tts_audio_queue.put( (SentenceType.FIRST, None, text) ) - self._process_audio_file(tmp_file, callback=opus_handler) + 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 @@ -228,14 +233,14 @@ class TTSProviderBase(ABC): self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + self.to_tts_stream(segment_text, opus_handler=self.handle_opus) 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): - self._process_audio_file(tts_file, callback=self.handle_opus) + self._process_audio_file_stream(tts_file, callback=self.handle_opus) 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) ) @@ -419,15 +424,12 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file, callback: Callable[[Any], Any]): + def _process_audio_file_stream(self, tts_file, callback: Callable[[Any], Any]) -> None: """处理音频文件并转换为指定格式 Args: tts_file: 音频文件路径 - content_detail: 内容详情 - - Returns: - tuple: (sentence_type, audio_datas, content_detail) + callback: 文件处理函数 """ if tts_file.endswith(".p3"): p3.decode_opus_from_file_stream(tts_file, callback=callback) @@ -450,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: @@ -461,7 +463,7 @@ class TTSProviderBase(ABC): if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + self.to_tts_stream(segment_text, opus_handler=opus_handler) self.processed_chars += len(full_text) return True return False 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 ea7f18fd..f45b9a96 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/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 @@ -260,11 +262,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.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) if message.sentence_type == SentenceType.LAST: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -452,21 +450,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) - 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.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) 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: @@ -642,15 +626,15 @@ class TTSProvider(TTSProviderBase): ) ) - def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False): - opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end) - return opus_datas + def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): + return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式生成音频数据，用于生成音频及测试场景 Args: text: 要转换的文本 + opus_handler: opus数据处理方法 Returns: list: 音频数据列表 @@ -663,9 +647,6 @@ class TTSProvider(TTSProviderBase): # 生成会话ID session_id = uuid.uuid4().__str__().replace("-", "") - # 存储音频数据 - audio_data = [] - async def _generate_audio(): # 创建新的WebSocket连接 ws_header = { @@ -728,8 +709,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) - audio_data.extend(opus_datas) + self.wav_to_opus_data_audio_raw_stream(res.payload, callback=opus_handler) elif res.optional.event == EVENT_SessionFinished: break @@ -744,8 +724,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 [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index d1be586c..a9877d43 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/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.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, 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,7 +233,7 @@ 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) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: From 4d50eeb8303fb4d9514404a303363319c4c72174 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 14:53:59 +0800 Subject: [PATCH 10/10] =?UTF-8?q?=E4=BF=AE=E6=AD=A3self=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 4 +++- .../core/providers/tts/huoshan_double_stream.py | 4 +++- .../core/providers/tts/linkerai.py | 17 ++++++----------- 3 files changed, 12 insertions(+), 13 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 52451cba..4929491c 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -482,8 +482,10 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: + 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() 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 f45b9a96..118cb4fe 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -629,7 +629,7 @@ class TTSProvider(TTSProviderBase): def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) - def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: + def to_tts_stream(self, text: str, opus_handler=None) -> None: """非流式生成音频数据，用于生成音频及测试场景 Args: @@ -639,6 +639,8 @@ class TTSProvider(TTSProviderBase): Returns: list: 音频数据列表 """ + if opus_handler is None: + opus_handler = self.handle_opus try: # 创建事件循环 loop = asyncio.new_event_loop() diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index a9877d43..09d3d82a 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -233,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_stream(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=None) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: text: 要转换的文本 - - Returns: - list: 返回opus编码后的音频数据列表 + opus_handler: opus数据处理方法 """ + if opus_handler is None: + opus_handler = self.handle_opus start_time = time.time() text = MarkdownCleaner.clean_markdown(text) @@ -291,14 +291,9 @@ class TTSProvider(TTSProviderBase): # 最后一帧可能不足，用0填充 frame = frame + b"\x00" * (frame_bytes - len(frame)) - opus = self.opus_encoder.encode_pcm_to_opus( - frame, end_of_stream=(i + frame_bytes >= len(pcm_data)) + self.opus_encoder.encode_pcm_to_opus_stream( + 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 []