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

update:流式优化暂不稳定，先回滚到早期代码

Browse Source
This commit is contained in:
hrz
2025-08-30 22:59:21 +08:00
parent b4f4995ff9
commit eaade698fb
19 changed files with 1116 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/providers/asr/base.py
+63 -89
View File
@@ -1,18 +1,18 @@
import os
import io
import wave
import uuid
import json
import time
import queue
import asyncio
import traceback
import threading
import opuslib_next
import json
import io
import time
import concurrent.futures
from abc import ABC, abstractmethod
from config.logger import setup_logging
from typing import Optional, Tuple, List
from typing import Optional, Tuple, List, Dict, Any
from core.handle.receiveAudioHandle import startToChat
from core.handle.reportHandle import enqueue_asr_report
from core.utils.util import remove_punctuation_and_length
@@ -53,10 +53,6 @@ class ASRProviderBase(ABC):
# 接收音频
async def receive_audio(self, conn, audio, audio_have_voice):
# 检查是否在唤醒处理锁定期内
if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
return
if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
have_voice = audio_have_voice
else:
@@ -66,14 +62,6 @@ class ASRProviderBase(ABC):
if not have_voice and not conn.client_have_voice:
conn.asr_audio = conn.asr_audio[-10:]
return
# 检查是否处于唤醒模式
if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10:
asr_audio_task = conn.asr_audio.copy()
conn.reset_vad_states()
conn.asr_audio.clear()
await self.handle_voice_stop(conn, asr_audio_task)
if conn.client_voice_stop:
asr_audio_task = conn.asr_audio.copy()
@@ -99,13 +87,32 @@ class ASRProviderBase(ABC):
# 预先准备WAV数据
wav_data = None
# 使用连接的声纹识别提供者
if conn.voiceprint_provider and combined_pcm_data:
wav_data = self._pcm_to_wav(combined_pcm_data)
# 检查是否处于唤醒模式
wakeup_mode = getattr(conn, 'wakeup_mode', False)
# 声纹识别任务（公共逻辑）
# 定义ASR任务
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 定义声纹识别任务
def run_voiceprint():
if not wav_data:
return None
@@ -124,83 +131,50 @@ class ASRProviderBase(ABC):
logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
return None
if wakeup_mode and conn.voiceprint_provider and wav_data:
conn.wakeup_mode = False
# 设置处理锁，防止后续音频片段重复处理
conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期
# 使用线程池执行器并行运行
parallel_start_time = time.monotonic()
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
# 唤醒模式：只执行声纹识别
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor:
if conn.voiceprint_provider and wav_data:
voiceprint_future = thread_executor.submit(run_voiceprint)
# 等待两个线程都完成
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
speaker_name = voiceprint_result
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, file_path = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
# 记录识别结果
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
fixed_text = "嘿，你好啊"
enhanced_text = self._build_enhanced_text(fixed_text, speaker_name)
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
# 构建包含说话人信息的JSON字符串
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s")
# 使用自定义模块进行上报
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
else:
# 正常模式：执行声纹识别和ASR
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 使用线程池执行器并行运行
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
if conn.voiceprint_provider and wav_data:
voiceprint_future = thread_executor.submit(run_voiceprint)
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, _ = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
+184 -11
View File
@@ -268,7 +268,11 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
try:
logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -418,6 +422,9 @@ class TTSProvider(TTSProviderBase):
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
opus_datas_cache = []
is_first_sentence = True
first_sentence_segment_count = 0 # 添加计数器
try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set():
@@ -438,16 +445,28 @@ class TTSProvider(TTSProviderBase):
self.tts_audio_queue.put(
(SentenceType.FIRST, [], None)
)
elif event_name == "SentenceBegin":
opus_datas_cache = []
elif event_name == "SentenceEnd":
# 发送缓存的数据
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
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, opus_datas_cache, self.conn.tts_MessageText)
)
self.conn.tts_MessageText = None
else:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 第一句话结束后，将标志设置为False
is_first_sentence = False
elif event_name == "SynthesisCompleted":
logger.bind(tag=TAG).debug(f"会话结束～～")
self._process_before_stop_play_files()
@@ -458,7 +477,22 @@ class TTSProvider(TTSProviderBase):
# 二进制消息（音频数据）
elif isinstance(msg, (bytes, bytearray)):
logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus)
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)
except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
break
@@ -478,3 +512,142 @@ class TTSProvider(TTSProviderBase):
finally:
self._monitor_task = None
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景"""
try:
# 创建新的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 生成会话ID
session_id = uuid.uuid4().hex
# 存储音频数据
audio_data = []
async def _generate_audio():
# 刷新Token（如果需要）
if self._is_token_expired():
self._refresh_token()
# 建立WebSocket连接
ws = await websockets.connect(
self.ws_url,
additional_headers={"X-NLS-Token": self.token},
ping_interval=30,
ping_timeout=10,
close_timeout=10,
)
try:
# 发送StartSynthesis请求
start_message_id = str(uuid.uuid4().hex)
start_request = {
"header": {
"message_id": start_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "StartSynthesis",
"appkey": self.appkey,
},
"payload": {
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"volume": self.volume,
"speech_rate": self.speech_rate,
"pitch_rate": self.pitch_rate,
"enable_subtitle": True,
},
}
await ws.send(json.dumps(start_request))
# 等待SynthesisStarted响应
synthesis_started = False
while not synthesis_started:
msg = await ws.recv()
if isinstance(msg, str):
data = json.loads(msg)
header = data.get("header", {})
if header.get("name") == "SynthesisStarted":
synthesis_started = True
logger.bind(tag=TAG).debug("TTS合成已启动")
elif header.get("name") == "TaskFailed":
error_info = data.get("payload", {}).get(
"error_info", {}
)
error_code = error_info.get("error_code")
error_message = error_info.get(
"error_message", "未知错误"
)
raise Exception(
f"启动合成失败: {error_code} - {error_message}"
)
# 发送文本合成请求
filtered_text = MarkdownCleaner.clean_markdown(text)
run_message_id = str(uuid.uuid4().hex)
run_request = {
"header": {
"message_id": run_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "RunSynthesis",
"appkey": self.appkey,
},
"payload": {"text": filtered_text},
}
await ws.send(json.dumps(run_request))
# 发送停止合成请求
stop_message_id = str(uuid.uuid4().hex)
stop_request = {
"header": {
"message_id": stop_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "StopSynthesis",
"appkey": self.appkey,
}
}
await ws.send(json.dumps(stop_request))
# 接收音频数据
synthesis_completed = False
while not synthesis_completed:
msg = await ws.recv()
if isinstance(msg, (bytes, bytearray)):
# 编码为Opus并收集
opus_frames = self.opus_encoder.encode_pcm_to_opus(
msg, False
)
audio_data.extend(opus_frames)
elif isinstance(msg, str):
data = json.loads(msg)
header = data.get("header", {})
event_name = header.get("name")
if event_name == "SynthesisCompleted":
synthesis_completed = True
logger.bind(tag=TAG).debug("TTS合成完成")
elif event_name == "TaskFailed":
error_info = data.get("payload", {}).get(
"error_info", {}
)
error_code = error_info.get("error_code")
error_message = error_info.get(
"error_message", "未知错误"
)
raise Exception(
f"合成失败: {error_code} - {error_message}"
)
finally:
try:
await ws.close()
except:
pass
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
main/xiaozhi-server/core/providers/tts/base.py
+61 -79
View File
@@ -4,14 +4,12 @@ import queue
import uuid
import asyncio
import threading
from typing import Callable, Any
from core.utils import p3
import time
from datetime import datetime
from core.utils import textUtils
from abc import ABC, abstractmethod
from config.logger import setup_logging
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
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
from core.handle.reportHandle import enqueue_tts_report
@@ -33,6 +31,7 @@ class TTSProviderBase(ABC):
def __init__(self, config, delete_audio_file):
self.interface_type = InterfaceType.NON_STREAM
self.conn = None
self.tts_timeout = 10
self.delete_audio_file = delete_audio_file
self.audio_file_type = "wav"
self.output_file = config.get("output_dir", "tmp/")
@@ -51,9 +50,11 @@ class TTSProviderBase(ABC):
"；",
";",
"：",
"~",
)
self.first_sentence_punctuations = (
"，",
"～",
"~",
"、",
",",
@@ -76,14 +77,7 @@ class TTSProviderBase(ABC):
f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
)
def handle_opus(self, opus_data: bytes):
logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}")
self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None))
def handle_audio_file(self, file_audio: bytes, text):
self.before_stop_play_files.append((file_audio, text))
def to_tts_stream(self, text, opus_handler: Callable[[bytes], None] = None) -> None:
def to_tts(self, text):
text = MarkdownCleaner.clean_markdown(text)
max_repeat_time = 5
if self.delete_audio_file:
@@ -92,14 +86,10 @@ class TTSProviderBase(ABC):
try:
audio_bytes = asyncio.run(self.text_to_speak(text, None))
if audio_bytes:
self.tts_audio_queue.put((SentenceType.FIRST, None, text))
audio_bytes_to_data_stream(
audio_bytes,
file_type=self.audio_file_type,
is_opus=True,
callback=opus_handler,
audio_datas, _ = audio_bytes_to_data(
audio_bytes, file_type=self.audio_file_type, is_opus=True
)
break
return audio_datas
else:
max_repeat_time -= 1
except Exception as e:
@@ -139,8 +129,8 @@ class TTSProviderBase(ABC):
logger.bind(tag=TAG).error(
f"语音生成失败: {text}，请检查网络或服务是否正常"
)
self.tts_audio_queue.put((SentenceType.FIRST, None, text))
self._process_audio_file_stream(tmp_file, callback=opus_handler)
return tmp_file
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
return None
@@ -149,17 +139,13 @@ class TTSProviderBase(ABC):
async def text_to_speak(self, text, output_file):
pass
def audio_to_pcm_data_stream(
self, audio_file_path, callback: Callable[[Any], Any] = None
):
def audio_to_pcm_data(self, audio_file_path):
"""音频文件转换为PCM编码"""
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback)
return audio_to_data(audio_file_path, is_opus=False)
def audio_to_opus_data_stream(
self, audio_file_path, callback: Callable[[Any], Any] = None
):
def audio_to_opus_data(self, audio_file_path):
"""音频文件转换为Opus编码"""
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback)
return audio_to_data(audio_file_path, is_opus=True)
def tts_one_sentence(
self,
@@ -191,6 +177,7 @@ class TTSProviderBase(ABC):
async def open_audio_channels(self, conn):
self.conn = conn
self.tts_timeout = conn.config.get("tts_timeout", 10)
# tts 消化线程
self.tts_priority_thread = threading.Thread(
target=self.tts_text_priority_thread, daemon=True
@@ -225,16 +212,30 @@ class TTSProviderBase(ABC):
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
if segment_text:
self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
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)
)
elif ContentType.FILE == message.content_type:
self._process_remaining_text_stream(opus_handler=self.handle_opus)
self._process_remaining_text()
tts_file = message.content_file
if tts_file and os.path.exists(tts_file):
self._process_audio_file_stream(
tts_file, callback=self.handle_opus
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put(
(message.sentence_type, audio_datas, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
self._process_remaining_text_stream(opus_handler=self.handle_opus)
self._process_remaining_text()
self.tts_audio_queue.put(
(message.sentence_type, [], message.content_detail)
)
@@ -248,59 +249,29 @@ class TTSProviderBase(ABC):
continue
def _audio_play_priority_thread(self):
# 需要上报的文本和音频列表
enqueue_text = None
enqueue_audio = None
while not self.conn.stop_event.is_set():
text = None
try:
try:
sentence_type, audio_datas, text = self.tts_audio_queue.get(
timeout=0.1
timeout=1
)
except queue.Empty:
if self.conn.stop_event.is_set():
break
continue
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)
enqueue_audio = []
enqueue_text = text
# 收集上报音频数据
if isinstance(audio_datas, bytes) and enqueue_audio is not None:
enqueue_audio.append(audio_datas)
# 发送音频
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_thread: {text} {e}")
logger.bind(tag=TAG).error(
f"audio_play_priority priority_thread: {text} {e}"
)
async def start_session(self, session_id):
pass
@@ -352,21 +323,22 @@ class TTSProviderBase(ABC):
else:
return None
def _process_audio_file_stream(
self, tts_file, callback: Callable[[Any], Any]
) -> None:
def _process_audio_file(self, tts_file):
"""处理音频文件并转换为指定格式
Args:
tts_file: 音频文件路径
callback: 文件处理函数
content_detail: 内容详情
Returns:
tuple: (sentence_type, audio_datas, content_detail)
"""
if tts_file.endswith(".p3"):
p3.decode_opus_from_file_stream(tts_file, callback=callback)
audio_datas, _ = p3.decode_opus_from_file(tts_file)
elif self.conn.audio_format == "pcm":
self.audio_to_pcm_data_stream(tts_file, callback=callback)
audio_datas, _ = self.audio_to_pcm_data(tts_file)
else:
self.audio_to_opus_data_stream(tts_file, callback=callback)
audio_datas, _ = self.audio_to_opus_data(tts_file)
if (
self.delete_audio_file
@@ -375,6 +347,7 @@ 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:
@@ -382,9 +355,7 @@ class TTSProviderBase(ABC):
self.before_stop_play_files.clear()
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def _process_remaining_text_stream(
self, opus_handler: Callable[[bytes], None] = None
):
def _process_remaining_text(self):
"""处理剩余的文本并生成语音
Returns:
@@ -395,7 +366,18 @@ class TTSProviderBase(ABC):
if remaining_text:
segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
if segment_text:
self.to_tts_stream(segment_text, opus_handler=opus_handler)
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)
self.tts_audio_queue.put(
(SentenceType.MIDDLE, audio_datas, segment_text)
)
self.processed_chars += len(full_text)
return True
return False
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+139 -5
View File
@@ -4,7 +4,6 @@ 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
@@ -267,7 +266,11 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
try:
logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -425,6 +428,9 @@ class TTSProvider(TTSProviderBase):
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
opus_datas_cache = []
is_first_sentence = True
first_sentence_segment_count = 0 # 添加计数器
try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set():
@@ -445,14 +451,37 @@ class TTSProvider(TTSProviderBase):
self.tts_audio_queue.put(
(SentenceType.FIRST, [], self.tts_text)
)
opus_datas_cache = []
first_sentence_segment_count = 0 # 重置计数器
elif (
res.optional.event == EVENT_TTSResponse
and res.header.message_type == AUDIO_ONLY_RESPONSE
):
logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
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)
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:
# 发送缓存的数据
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 第一句话结束后，将标志设置为False
is_first_sentence = False
elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束～～")
self._process_before_stop_play_files()
@@ -626,5 +655,110 @@ 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 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 to_tts(self, text: str) -> list:
"""非流式生成音频数据，用于生成音频及测试场景
Args:
text: 要转换的文本
Returns:
list: 音频数据列表
"""
try:
# 创建事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 生成会话ID
session_id = uuid.uuid4().__str__().replace("-", "")
# 存储音频数据
audio_data = []
async def _generate_audio():
# 创建新的WebSocket连接
ws_header = {
"X-Api-App-Key": self.appId,
"X-Api-Access-Key": self.access_token,
"X-Api-Resource-Id": self.resource_id,
"X-Api-Connect-Id": uuid.uuid4(),
}
ws = await websockets.connect(
self.ws_url, additional_headers=ws_header, max_size=1000000000
)
try:
# 启动会话
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_StartSession, sessionId=session_id
).as_bytes()
payload = self.get_payload_bytes(
event=EVENT_StartSession, speaker=self.voice
)
await self.send_event(ws, header, optional, payload)
# 发送文本
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_TaskRequest, sessionId=session_id
).as_bytes()
payload = self.get_payload_bytes(
event=EVENT_TaskRequest, text=text, speaker=self.voice
)
await self.send_event(ws, header, optional, payload)
# 发送结束会话请求
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_FinishSession, sessionId=session_id
).as_bytes()
payload = str.encode("{}")
await self.send_event(ws, header, optional, payload)
# 接收音频数据
while True:
msg = await ws.recv()
res = self.parser_response(msg)
if (
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)
elif res.optional.event == EVENT_SessionFinished:
break
finally:
# 清理资源
try:
await ws.close()
except:
pass
# 运行异步任务
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
main/xiaozhi-server/core/providers/tts/index_stream.py
+98 -13
View File
@@ -3,6 +3,8 @@ import queue
import asyncio
import traceback
import aiohttp
import requests
import time
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -25,13 +27,15 @@ class TTSProvider(TTSProviderBase):
self.api_url = config.get("api_url", "http://8.138.114.124:11996/tts")
self.audio_format = "pcm"
self.before_stop_play_files = []
self.segment_count = 0
# 创建Opus编码器 需注意接口返回的采样率为24000
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=24000, channels=1, frame_size_ms=60
)
# PCM缓冲区
# 文本缓冲区和PCM缓冲区
self.text_buffer = ""
self.pcm_buffer = bytearray()
def tts_text_priority_thread(self):
@@ -44,6 +48,7 @@ class TTSProvider(TTSProviderBase):
self.tts_stop_request = False
self.processed_chars = 0
self.tts_text_buff = []
self.segment_count = 0
self.before_stop_play_files.clear()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
@@ -57,11 +62,14 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
# 处理剩余的文本
self._process_remaining_text_stream(True)
self._process_remaining_text(True)
except queue.Empty:
continue
@@ -70,7 +78,7 @@ class TTSProvider(TTSProviderBase):
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
def _process_remaining_text_stream(self, is_last=False):
def _process_remaining_text(self, is_last=False):
"""处理剩余的文本并生成语音
Returns:
bool: 是否成功处理了文本
@@ -135,6 +143,8 @@ class TTSProvider(TTSProviderBase):
return
self.pcm_buffer.clear()
opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text))
# 处理音频流数据
@@ -148,22 +158,41 @@ class TTSProvider(TTSProviderBase):
while len(self.pcm_buffer) >= frame_bytes:
frame = bytes(self.pcm_buffer[:frame_bytes])
del self.pcm_buffer[:frame_bytes]
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=False,
callback=self.handle_opus
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=False
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
opus_datas_cache.extend(opus)
# flush 剩余不足一帧的数据
if self.pcm_buffer:
self.opus_encoder.encode_pcm_to_opus_stream(
bytes(self.pcm_buffer),
end_of_stream=True,
callback=self.handle_opus
opus = self.opus_encoder.encode_pcm_to_opus(
bytes(self.pcm_buffer), end_of_stream=True
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
# 直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
# 后续片段缓存
opus_datas_cache.extend(opus)
self.pcm_buffer.clear()
# 如果不是前10个片段，发送缓存的数据
if self.segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 如果是最后一段，输出音频获取完毕
if is_last:
self._process_before_stop_play_files()
@@ -177,3 +206,59 @@ class TTSProvider(TTSProviderBase):
await super().close()
if hasattr(self, "opus_encoder"):
self.opus_encoder.close()
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
start_time = time.time()
text = MarkdownCleaner.clean_markdown(text)
payload = {"text": text, "character": self.character}
try:
with requests.post(self.api_url, json=payload, timeout=5) as response:
if response.status_code != 200:
logger.bind(tag=TAG).error(
f"TTS请求失败: {response.status_code}, {response.text}"
)
return []
logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
# 使用opus编码器处理PCM数据
opus_datas = []
pcm_data = response.content
# 计算每帧的字节数
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels
* self.opus_encoder.frame_size_ms
/ 1000
* 2
)
# 分帧处理PCM数据
for i in range(0, len(pcm_data), frame_bytes):
frame = pcm_data[i : i + frame_bytes]
if len(frame) < frame_bytes:
# 最后一帧可能不足，用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))
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
main/xiaozhi-server/core/providers/tts/linkerai.py
+123 -12
View File
@@ -1,8 +1,10 @@
import os
import queue
import aiohttp
import asyncio
import traceback
import aiohttp
import requests
import time
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -22,15 +24,23 @@ class TTSProvider(TTSProviderBase):
self.api_url = config.get("api_url")
self.audio_format = "pcm"
self.before_stop_play_files = []
self.segment_count = 0 # 添加片段计数器
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
# 添加文本缓冲区
self.text_buffer = ""
# PCM缓冲区
self.pcm_buffer = bytearray()
###################################################################################
# linkerai单流式TTS重写父类的方法--开始
###################################################################################
def tts_text_priority_thread(self):
"""流式文本处理线程"""
while not self.conn.stop_event.is_set():
@@ -41,6 +51,7 @@ class TTSProvider(TTSProviderBase):
self.tts_stop_request = False
self.processed_chars = 0
self.tts_text_buff = []
self.segment_count = 0
self.before_stop_play_files.clear()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
@@ -54,10 +65,14 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
# 处理剩余的文本
self._process_remaining_text_stream(True)
self._process_remaining_text(True)
except queue.Empty:
continue
@@ -66,7 +81,7 @@ class TTSProvider(TTSProviderBase):
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
def _process_remaining_text_stream(self, is_last=False):
def _process_remaining_text(self, is_last=False):
"""处理剩余的文本并生成语音
Returns:
@@ -109,6 +124,10 @@ class TTSProvider(TTSProviderBase):
finally:
return None
###################################################################################
# linkerai单流式TTS重写父类的方法--结束
###################################################################################
async def text_to_speak(self, text, is_last):
"""流式处理TTS音频，每句只推送一次音频列表"""
await self._tts_request(text, is_last)
@@ -157,6 +176,8 @@ class TTSProvider(TTSProviderBase):
return
self.pcm_buffer.clear()
opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text))
# 兼容 iter_chunked / iter_chunks / iter_any
@@ -173,21 +194,41 @@ class TTSProvider(TTSProviderBase):
frame = bytes(self.pcm_buffer[:frame_bytes])
del self.pcm_buffer[:frame_bytes]
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=False,
callback=self.handle_opus
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=False
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
opus_datas_cache.extend(opus)
# flush 剩余不足一帧的数据
if self.pcm_buffer:
self.opus_encoder.encode_pcm_to_opus_stream(
bytes(self.pcm_buffer),
end_of_stream=True,
callback=self.handle_opus
opus = self.opus_encoder.encode_pcm_to_opus(
bytes(self.pcm_buffer), end_of_stream=True
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
# 直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
# 后续片段缓存
opus_datas_cache.extend(opus)
self.pcm_buffer.clear()
# 如果不是前10个片段，发送缓存的数据
if self.segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 如果是最后一段，输出音频获取完毕
if is_last:
self._process_before_stop_play_files()
@@ -195,3 +236,73 @@ class TTSProvider(TTSProviderBase):
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
start_time = time.time()
text = MarkdownCleaner.clean_markdown(text)
params = {
"tts_text": text,
"spk_id": self.voice,
"frame_duration": 60,
"stream": False,
"target_sr": 16000,
"audio_format": self.audio_format,
"instruct_text": "请生成一段自然流畅的语音",
}
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
}
try:
with requests.get(
self.api_url, params=params, headers=headers, timeout=5
) as response:
if response.status_code != 200:
logger.bind(tag=TAG).error(
f"TTS请求失败: {response.status_code}, {response.text}"
)
return []
logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
# 使用opus编码器处理PCM数据
opus_datas = []
pcm_data = response.content
# 计算每帧的字节数
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels
* self.opus_encoder.frame_size_ms
/ 1000
* 2
)
# 分帧处理PCM数据
for i in range(0, len(pcm_data), frame_bytes):
frame = pcm_data[i : i + frame_bytes]
if len(frame) < frame_bytes:
# 最后一帧可能不足，用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))
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
main/xiaozhi-server/core/providers/tts/paddle_speech.py
+4 -26
View File
@@ -8,7 +8,6 @@ import wave
import websockets
from core.providers.tts.base import TTSProviderBase
from config.logger import setup_logging
from datetime import datetime
TAG = __name__
logger = setup_logging()
@@ -19,12 +18,11 @@ class TTSProvider(TTSProviderBase):
super().__init__(config, delete_audio_file)
self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming")
self.protocol = config.get("protocol", "websocket")
if config.get("private_voice"):
self.spk_id = int(config.get("private_voice"))
else:
self.spk_id = int(config.get("spk_id", "0"))
self.spk_id = int(config.get("spk_id", "0"))
sample_rate = config.get("sample_rate", 24000)
self.sample_rate = float(sample_rate) if sample_rate else 24000
@@ -34,21 +32,7 @@ class TTSProvider(TTSProviderBase):
volume = config.get("volume", 1.0)
self.volume = float(volume) if volume else 1.0
self.delete_audio_file = config.get("delete_audio", True)
if not self.delete_audio_file:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
save_path = config.get("save_path")
if save_path:
if not save_path.endswith('.wav'):
save_path = f"{save_path}_{timestamp}.wav"
else:
other_path = save_path[:-4]
save_path = f"{other_path}_{timestamp}.wav"
self.save_path = save_path
else:
self.save_path = f"./streaming_tts_{timestamp}.wav"
else:
self.save_path = None
self.save_path = config.get("save_path", "./streaming_tts.wav")
async def pcm_to_wav(self, pcm_data: bytes, sample_rate: int = 24000, num_channels: int = 1,
bits_per_sample: int = 16) -> bytes:
@@ -167,12 +151,6 @@ class TTSProvider(TTSProviderBase):
# 接收结束响应避免服务抛出异常
await ws.recv()
# 根据配置决定是否保存文件
if not self.delete_audio_file and self.save_path:
with open(self.save_path, "wb") as f:
f.write(wav_data)
logger.bind(tag=TAG).info(f"音频文件已保存到: {self.save_path}")
# 返回或保存音频数据
if output_file:
with open(output_file, "wb") as file_to_save:
@@ -181,4 +159,4 @@ class TTSProvider(TTSProviderBase):
return wav_data
except Exception as e:
raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")
raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")
main/xiaozhi-server/core/providers/vad/silero.py
+4 -2
View File
@@ -33,7 +33,7 @@ class VADProvider(VADProviderBase):
int(min_silence_duration_ms) if min_silence_duration_ms else 1000
)
# 至少要多少帧才算有语音,增加灵敏度
# 至少要多少帧才算有语音
self.frame_window_threshold = 1
def is_vad(self, conn, opus_packet):
@@ -70,7 +70,9 @@ class VADProvider(VADProviderBase):
# 更新滑动窗口
conn.client_voice_window.append(is_voice)
client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold)
client_have_voice = (
conn.client_voice_window.count(True) >= self.frame_window_threshold
)
# 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话
if conn.client_have_voice and not client_have_voice: