Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Sakura-RanChen
2025-08-22 17:10:36 +08:00
parent 60f8dbf418
commit 1988bced60
8 changed files with 67 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/handle/sendAudioHandle.py
+52 -4
View File
@@ -1,5 +1,6 @@
import json
import asyncio
import time
from core.providers.tts.dto.dto import SentenceType
from core.utils import textUtils
@@ -29,13 +30,60 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
# 播放音频
async def sendAudio(conn, audios):
async def sendAudio(conn, audios, pre_buffer=False):
"""
发送单个opus包，支持流控
Args:
conn: 连接对象
opus_packet: 单个opus数据包
pre_buffer: 快速发送音频
"""
if audios is None:
return
# 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出
if isinstance(audios, bytes):
if conn.client_abort:
return
# 短音频直接发送（例如：提示音）
if pre_buffer:
await conn.websocket.send(audios)
return
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 流控逻辑：确保按60ms的帧时长间隔发送
frame_duration = 60 # 毫秒
# 获取或初始化流控状态
if not hasattr(conn, "audio_flow_control"):
conn.audio_flow_control = {
"last_send_time": 0,
"packet_count": 0,
"start_time": time.perf_counter(),
}
flow_control = conn.audio_flow_control
current_time = time.perf_counter()
# 计算期望的发送时间
expected_time = flow_control["start_time"] + (
flow_control["packet_count"] * frame_duration / 1000
)
# 流控延迟
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
# 发送数据包
await conn.websocket.send(audios)
# 更新流控状态
flow_control["packet_count"] += 1
flow_control["last_send_time"] = time.perf_counter()
async def send_tts_message(conn, state, text=None):
"""发送 TTS 状态消息"""
@@ -56,7 +104,7 @@ async def send_tts_message(conn, state, text=None):
conn.tts.audio_to_opus_data_stream(
stop_tts_notify_voice,
callback=lambda audio_data: asyncio.run_coroutine_threadsafe(
sendAudio(conn, audio_data), conn.loop
sendAudio(conn, audio_data, True), conn.loop
),
)
# 清除服务端讲话状态
@@ -77,7 +125,7 @@ async def send_stt_message(conn, text):
display_text = text
try:
# 尝试解析JSON格式
if text.strip().startswith('{') and text.strip().endswith('}'):
if text.strip().startswith("{") and text.strip().endswith("}"):
parsed_data = json.loads(text)
if isinstance(parsed_data, dict) and "content" in parsed_data:
# 如果是包含说话人信息的JSON格式，只显示content部分
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
-1
View File
@@ -216,7 +216,6 @@ class TTSProvider(TTSProviderBase):
if message.sentence_type == SentenceType.FIRST:
self.conn.client_abort = False
self.reset_flow_controller()
if self.conn.client_abort:
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
main/xiaozhi-server/core/providers/tts/base.py
+15 -77
View File
@@ -11,7 +11,6 @@ from datetime import datetime
from core.utils import textUtils
from abc import ABC, abstractmethod
from config.logger import setup_logging
from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
from core.utils.tts import MarkdownCleaner
from core.utils.output_counter import add_device_output
@@ -71,7 +70,6 @@ class TTSProviderBase(ABC):
self.tts_stop_request = False
self.processed_chars = 0
self.is_first_sentence = True
self.flow_controller = FlowControlConfig.create_flow_controller()
def generate_filename(self, extension=".wav"):
return os.path.join(
@@ -225,7 +223,6 @@ class TTSProviderBase(ABC):
self.tts_text_buff = []
self.is_first_sentence = True
self.tts_audio_first_sentence = True
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
@@ -270,12 +267,19 @@ class TTSProviderBase(ABC):
if self.conn.client_abort:
logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
# 打断时丢弃未上报的音频数据
enqueue_text, enqueue_audio = None, []
continue
# 收到下一个文本开始或会话结束时进行上报
if sentence_type is not SentenceType.MIDDLE:
# 重置音频流控状态（新句子开始或者结束）
if hasattr(self.conn, 'audio_flow_control'):
self.conn.audio_flow_control = {
'last_send_time': 0,
'packet_count': 0,
'start_time': time.perf_counter()
}
# 上报TTS数据
if enqueue_text is not None and enqueue_audio is not None:
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
@@ -284,88 +288,22 @@ class TTSProviderBase(ABC):
# 计算音频数据的帧数
if isinstance(audio_datas, bytes):
frame_count = 1 # 单个字节流作为一帧
enqueue_audio.append(audio_datas)
else:
frame_count = 0
# 发送音频
future = asyncio.run_coroutine_threadsafe(
sendAudioMessage(self.conn, sentence_type, audio_datas, text),
self.conn.loop,
)
future.result()
# 记录输出和报告
if self.conn.max_output_size > 0 and text:
add_device_output(self.conn.headers.get("device-id"), len(text))
# 流控检查
if frame_count > 0:
max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME
wait_start_time = time.time()
retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL
while not self.flow_controller.can_send_frames(frame_count):
# 检查是否超时或需要停止
if (
time.time() - wait_start_time > max_wait_time
or self.conn.stop_event.is_set()
or self.conn.client_abort
):
logger.bind(tag=TAG).debug(
"流控等待超时或收到停止信号，跳过音频发送"
)
break
# 短暂等待后重试
time.sleep(retry_interval)
else:
# 可以发送，记录发送的帧数
self.flow_controller.record_sent_frames(frame_count)
# 发送音频
future = asyncio.run_coroutine_threadsafe(
self._send_audio_with_flow_control(
sentence_type, audio_datas, text
),
self.conn.loop,
)
future.result()
# 输出流控状态（调试用）
# status = self.flow_controller.get_status()
# logger.bind(tag=TAG).debug(
# f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, "
# f"可用令牌={status['available_tokens']}..."
# f"发送帧数={status['sent_frames']}..."
# f"消费帧数={status['consumed_frames']}..."
# f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..."
# )
else:
# 没有音频数据，直接发送
future = asyncio.run_coroutine_threadsafe(
self._send_audio_with_flow_control(
sentence_type, audio_datas, text
),
self.conn.loop,
)
future.result()
except Exception as e:
logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}")
async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text):
"""
带流控的音频发送方法 模拟设备消费音频帧的过程
实际应用中应该根据设备反馈来更新消费情况
"""
await sendAudioMessage(self.conn, sentence_type, audio_datas, text)
# 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步
if isinstance(audio_datas, bytes):
frame_count = 1
asyncio.create_task(simulate_device_consumption(self.flow_controller, frame_count))
# 在类中添加流控制器重置方法
def reset_flow_controller(self):
"""重置流控制器状态，通常在新会话开始时调用"""
if hasattr(self, "flow_controller"):
self.flow_controller.reset()
logger.bind(tag=TAG).info("流控制器状态已重置")
async def start_session(self, session_id):
pass
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
-1
View File
@@ -213,7 +213,6 @@ class TTSProvider(TTSProviderBase):
if message.sentence_type == SentenceType.FIRST:
self.conn.client_abort = False
self.reset_flow_controller()
if self.conn.client_abort:
try:
main/xiaozhi-server/core/providers/tts/index_stream.py
-1
View File
@@ -45,7 +45,6 @@ class TTSProvider(TTSProviderBase):
self.processed_chars = 0
self.tts_text_buff = []
self.before_stop_play_files.clear()
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
main/xiaozhi-server/core/providers/tts/linkerai.py
-1
View File
@@ -42,7 +42,6 @@ class TTSProvider(TTSProviderBase):
self.processed_chars = 0
self.tts_text_buff = []
self.before_stop_play_files.clear()
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
main/xiaozhi-server/core/utils/audio_flow_control.py
-201
View File
@@ -1,201 +0,0 @@
"""
音频流控模块
包含令牌桶算法和音频流控制器的实现
"""
import asyncio
import time
import threading
from collections import deque
from typing import Optional, Dict, Any
class TokenBucket:
"""令牌桶实现，用于限流控制"""
def __init__(self, capacity: int, refill_rate: float, initial_tokens: Optional[int] = None):
"""
初始化令牌桶
Args:
capacity: 桶容量（最大令牌数）
refill_rate: 令牌补充速率（每秒补充的令牌数）
initial_tokens: 初始令牌数，默认为桶容量
"""
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = initial_tokens if initial_tokens is not None else capacity
self.last_refill_time = time.time()
self.lock = threading.Lock()
def get_tokens(self, requested_tokens: int = 1) -> bool:
"""
获取指定数量的令牌
Args:
requested_tokens: 请求的令牌数量
Returns:
bool: 是否成功获取到令牌
"""
with self.lock:
self._refill_tokens()
if self.tokens >= requested_tokens:
self.tokens -= requested_tokens
return True
else:
return False
def get_available_tokens(self) -> int:
"""获取当前可用令牌数"""
with self.lock:
self._refill_tokens()
return int(self.tokens)
def _refill_tokens(self):
"""内部方法：补充令牌"""
current_time = time.time()
time_passed = current_time - self.last_refill_time
tokens_to_add = time_passed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_refill_time = current_time
class AudioFlowController:
"""音频流控制器，基于令牌桶算法控制音频数据发送"""
def __init__(self, max_device_buffer: int = 3000, refill_rate: float = 20):
"""
初始化音频流控制器
Args:
max_device_buffer: 设备端最大缓冲区大小（Opus帧数）
refill_rate: 令牌补充速率（每秒允许发送的帧数）
"""
self.max_device_buffer = max_device_buffer
self.token_bucket = TokenBucket(
capacity=max_device_buffer,
refill_rate=refill_rate,
initial_tokens=max_device_buffer // 2 # 初始令牌为容量的一半
)
self.sent_frames_count = 0 # 已发送帧数计数
self.device_consumed_frames = 0 # 设备端已消费帧数
self.pending_queue = deque() # 等待发送的数据队列
self._lock = threading.Lock()
def can_send_frames(self, frame_count: int) -> bool:
"""
检查是否可以发送指定数量的帧
Args:
frame_count: 要发送的帧数
Returns:
bool: 是否可以发送
"""
with self._lock:
# 检查设备端缓冲区是否会溢出
estimated_device_buffer = self.sent_frames_count - self.device_consumed_frames
if estimated_device_buffer + frame_count > self.max_device_buffer:
return False
# 检查令牌桶是否有足够令牌
return self.token_bucket.get_tokens(frame_count)
def update_device_consumption(self, consumed_frames: int):
"""
更新设备端消费的帧数
Args:
consumed_frames: 设备端消费的帧数
"""
with self._lock:
self.device_consumed_frames += consumed_frames
def record_sent_frames(self, frame_count: int):
"""
记录已发送的帧数
Args:
frame_count: 发送的帧数
"""
with self._lock:
self.sent_frames_count += frame_count
def get_status(self) -> Dict[str, Any]:
"""获取流控状态信息"""
with self._lock:
estimated_buffer = self.sent_frames_count - self.device_consumed_frames
return {
"sent_frames": self.sent_frames_count,
"consumed_frames": self.device_consumed_frames,
"estimated_device_buffer": estimated_buffer,
"available_tokens": self.token_bucket.get_available_tokens(),
"pending_queue_size": len(self.pending_queue),
"buffer_usage_percent": (estimated_buffer / self.max_device_buffer) * 100
}
def reset(self):
"""重置流控状态"""
with self._lock:
self.sent_frames_count = 0
self.device_consumed_frames = 0
self.pending_queue.clear()
# 重新初始化令牌桶
self.token_bucket = TokenBucket(
capacity=self.max_device_buffer,
refill_rate=self.token_bucket.refill_rate,
initial_tokens=self.max_device_buffer // 2
)
async def simulate_device_consumption(
flow_controller: AudioFlowController, frame_count: int
):
"""
模拟设备消费音频帧的过程
实际应用中应该根据设备反馈来更新消费情况
Args:
flow_controller: 流控制器实例
frame_count: 消费的帧数
"""
# 模拟设备播放延迟（60ms per frame）
await asyncio.sleep(frame_count * 0.06)
flow_controller.update_device_consumption(frame_count)
# 流控配置常量
class FlowControlConfig:
"""流控配置常量"""
# Opus 编码参数
OPUS_FRAME_DURATION_MS = 60 # Opus帧时长（毫秒）
OPUS_FRAMES_PER_SECOND = 1000 / OPUS_FRAME_DURATION_MS # 每秒帧数
# 默认流控参数
DEFAULT_MAX_DEVICE_BUFFER = 40 # 设备端最大缓冲帧数
DEFAULT_REFILL_RATE = OPUS_FRAMES_PER_SECOND # 默认令牌补充速率（帧/秒）
DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒）
DEFAULT_RETRY_INTERVAL = 0.06 # 流控重试间隔（秒）
# 预缓冲参数
PRE_BUFFER_FRAMES = 3 # 预缓冲帧数
@classmethod
def create_flow_controller(cls, max_buffer: Optional[int] = None,
refill_rate: Optional[float] = None) -> AudioFlowController:
"""
创建流控制器的工厂方法
Args:
max_buffer: 最大缓冲区大小，使用默认值如果为None
refill_rate: 令牌补充速率，使用默认值如果为None
Returns:
AudioFlowController: 配置好的流控制器实例
"""
return AudioFlowController(
max_device_buffer=max_buffer or cls.DEFAULT_MAX_DEVICE_BUFFER,
refill_rate=refill_rate or cls.DEFAULT_REFILL_RATE
)
main/xiaozhi-server/plugins_func/functions/play_music.py
-1
View File
@@ -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":