mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-28 10:03:54 +08:00
refactor: 优化相关打断处理
This commit is contained in:
Sakura-RanChen
@@ -39,6 +39,7 @@ class TTSProvider(TTSProviderBase):
|
||||
self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
|
||||
self.ws = None
|
||||
self._monitor_task = None
|
||||
self.activate_session = False
|
||||
self.last_active_time = None
|
||||
|
||||
# 模型和音色配置
|
||||
@@ -75,9 +76,9 @@ class TTSProvider(TTSProviderBase):
|
||||
current_time = time.time()
|
||||
if self.ws and current_time - self.last_active_time < 60:
|
||||
# 一分钟内才可以复用链接进行连续对话
|
||||
logger.bind(tag=TAG).info(f"使用已有链接...")
|
||||
logger.bind(tag=TAG).debug(f"使用已有链接...")
|
||||
return self.ws
|
||||
logger.bind(tag=TAG).info("开始建立新连接...")
|
||||
logger.bind(tag=TAG).debug("开始建立新连接...")
|
||||
|
||||
self.ws = await websockets.connect(
|
||||
self.ws_url,
|
||||
@@ -87,7 +88,7 @@ class TTSProvider(TTSProviderBase):
|
||||
close_timeout=10,
|
||||
)
|
||||
|
||||
logger.bind(tag=TAG).info("WebSocket连接建立成功")
|
||||
logger.bind(tag=TAG).debug("WebSocket连接建立成功")
|
||||
self.last_active_time = current_time
|
||||
return self.ws
|
||||
except Exception as e:
|
||||
@@ -101,16 +102,22 @@ class TTSProvider(TTSProviderBase):
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
message = self.tts_text_queue.get(timeout=1)
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
|
||||
)
|
||||
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
self.conn.client_abort = False
|
||||
# 过滤旧消息：检查sentence_id是否匹配
|
||||
if message.sentence_id != self.conn.sentence_id:
|
||||
continue
|
||||
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}"
|
||||
)
|
||||
|
||||
if self.conn.client_abort:
|
||||
try:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.finish_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}")
|
||||
@@ -121,16 +128,16 @@ class TTSProvider(TTSProviderBase):
|
||||
try:
|
||||
if not getattr(self.conn, "sentence_id", None):
|
||||
self.conn.sentence_id = uuid.uuid4().hex
|
||||
logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}")
|
||||
logger.bind(tag=TAG).debug(f"自动生成新的 会话ID: {self.conn.sentence_id}")
|
||||
|
||||
logger.bind(tag=TAG).info("开始启动TTS会话...")
|
||||
logger.bind(tag=TAG).debug("开始启动TTS会话...")
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
self.start_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
self.before_stop_play_files.clear()
|
||||
logger.bind(tag=TAG).info("TTS会话启动成功")
|
||||
logger.bind(tag=TAG).debug("TTS会话启动成功")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
|
||||
continue
|
||||
@@ -146,7 +153,6 @@ class TTSProvider(TTSProviderBase):
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
logger.bind(tag=TAG).debug("TTS文本发送成功")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
||||
continue
|
||||
@@ -161,12 +167,12 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
if message.sentence_type == SentenceType.LAST:
|
||||
try:
|
||||
logger.bind(tag=TAG).info("开始结束TTS会话...")
|
||||
logger.bind(tag=TAG).debug("开始结束TTS会话...")
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
self.finish_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
future.result()
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
|
||||
continue
|
||||
@@ -202,7 +208,6 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
await self.ws.send(json.dumps(continue_task_message))
|
||||
self.last_active_time = time.time()
|
||||
logger.bind(tag=TAG).debug(f"已发送文本: {filtered_text}")
|
||||
return
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
||||
@@ -216,22 +221,22 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def start_session(self, session_id):
|
||||
"""启动TTS会话"""
|
||||
logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
|
||||
logger.bind(tag=TAG).debug(f"开始会话～～{session_id}")
|
||||
try:
|
||||
# 检查并清理上一个会话的监听任务
|
||||
if (
|
||||
self._monitor_task is not None
|
||||
and isinstance(self._monitor_task, Task)
|
||||
and not self._monitor_task.done()
|
||||
):
|
||||
logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务...")
|
||||
# 上个会话处于激活状态时关闭上个连接新建链接
|
||||
if self.activate_session:
|
||||
await self.close()
|
||||
|
||||
# 设置会话激活标志
|
||||
self.activate_session = True
|
||||
|
||||
# 确保连接可用
|
||||
await self._ensure_connection()
|
||||
|
||||
# 启动监听任务
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
if self._monitor_task is None or self._monitor_task.done():
|
||||
logger.bind(tag=TAG).debug("启动监听任务...")
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
|
||||
# 发送run-task消息启动会话
|
||||
run_task_message = {
|
||||
@@ -260,7 +265,7 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
await self.ws.send(json.dumps(run_task_message))
|
||||
self.last_active_time = time.time()
|
||||
logger.bind(tag=TAG).info("会话启动请求已发送")
|
||||
logger.bind(tag=TAG).debug("会话启动请求已发送")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
|
||||
await self.close()
|
||||
@@ -268,7 +273,7 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def finish_session(self, session_id):
|
||||
"""结束TTS会话"""
|
||||
logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
|
||||
logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}")
|
||||
try:
|
||||
if self.ws and session_id:
|
||||
# 发送finish-task消息
|
||||
@@ -285,17 +290,6 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
await self.ws.send(json.dumps(finish_task_message))
|
||||
self.last_active_time = time.time()
|
||||
logger.bind(tag=TAG).info("会话结束请求已发送")
|
||||
# 等待监听任务完成
|
||||
if self._monitor_task:
|
||||
try:
|
||||
await self._monitor_task
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(
|
||||
f"等待监听任务完成时发生错误: {str(e)}"
|
||||
)
|
||||
finally:
|
||||
self._monitor_task = None
|
||||
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
|
||||
@@ -304,6 +298,8 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def close(self):
|
||||
"""清理资源"""
|
||||
await super().close()
|
||||
self.activate_session = False
|
||||
# 取消监听任务
|
||||
if self._monitor_task:
|
||||
try:
|
||||
@@ -325,19 +321,13 @@ class TTSProvider(TTSProviderBase):
|
||||
self.last_active_time = None
|
||||
|
||||
async def _start_monitor_tts_response(self):
|
||||
"""监听TTS响应"""
|
||||
"""监听TTS响应 - 长期运行"""
|
||||
try:
|
||||
session_finished = False
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
msg = await self.ws.recv()
|
||||
self.last_active_time = time.time()
|
||||
|
||||
# 检查客户端是否中止
|
||||
if self.conn.client_abort:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应")
|
||||
break
|
||||
|
||||
if isinstance(msg, str): # JSON控制消息
|
||||
try:
|
||||
data = json.loads(msg)
|
||||
@@ -348,19 +338,19 @@ class TTSProvider(TTSProviderBase):
|
||||
self.tts_audio_queue.put((SentenceType.FIRST, [], None))
|
||||
elif event == "result-generated":
|
||||
# 发送缓存的数据
|
||||
if self.conn.tts_MessageText:
|
||||
tts_text = self.get_tts_text(self.conn.sentence_id)
|
||||
if tts_text:
|
||||
logger.bind(tag=TAG).info(
|
||||
f"句子语音生成成功： {self.conn.tts_MessageText}"
|
||||
f"句子语音生成成功： {tts_text}"
|
||||
)
|
||||
self.tts_audio_queue.put(
|
||||
(SentenceType.FIRST, [], self.conn.tts_MessageText)
|
||||
(SentenceType.FIRST, [], tts_text)
|
||||
)
|
||||
self.conn.tts_MessageText = None
|
||||
self.clear_tts_text(self.conn.sentence_id)
|
||||
elif event == "task-finished":
|
||||
logger.bind(tag=TAG).debug("TTS任务完成~")
|
||||
self.activate_session = False
|
||||
self._process_before_stop_play_files()
|
||||
session_finished = True
|
||||
break
|
||||
elif event == "task-failed":
|
||||
error_code = data["header"].get("error_code", "unknown")
|
||||
error_message = data["header"].get("error_message", "未知错误")
|
||||
@@ -383,8 +373,8 @@ class TTSProvider(TTSProviderBase):
|
||||
)
|
||||
break
|
||||
|
||||
# 仅在连接异常且非正常结束时才关闭连接
|
||||
if not session_finished and self.ws:
|
||||
# 连接异常时关闭WebSocket
|
||||
if self.ws:
|
||||
try:
|
||||
await self.ws.close()
|
||||
except:
|
||||
@@ -392,6 +382,7 @@ class TTSProvider(TTSProviderBase):
|
||||
self.ws = None
|
||||
# 监听任务退出时清理引用
|
||||
finally:
|
||||
self.activate_session = False
|
||||
self._monitor_task = None
|
||||
|
||||
def audio_to_opus_data_stream(
|
||||
|
||||
@@ -135,6 +135,7 @@ class TTSProvider(TTSProviderBase):
|
||||
self.ws_url = f"wss://{self.host}/ws/v1"
|
||||
self.ws = None
|
||||
self._monitor_task = None
|
||||
self.activate_session = False
|
||||
self.last_active_time = None
|
||||
|
||||
# 专属tts设置
|
||||
@@ -188,7 +189,7 @@ class TTSProvider(TTSProviderBase):
|
||||
if self.ws and current_time - self.last_active_time < 10:
|
||||
# 10秒内才可以复用链接进行连续对话
|
||||
self.task_id = uuid.uuid4().hex
|
||||
logger.bind(tag=TAG).info(f"使用已有链接..., task_id: {self.task_id}")
|
||||
logger.bind(tag=TAG).debug(f"使用已有链接..., task_id: {self.task_id}")
|
||||
return self.ws
|
||||
logger.bind(tag=TAG).debug("开始建立新连接...")
|
||||
|
||||
@@ -214,15 +215,21 @@ class TTSProvider(TTSProviderBase):
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
message = self.tts_text_queue.get(timeout=1)
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
|
||||
)
|
||||
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
self.conn.client_abort = False
|
||||
# 过滤旧消息：检查sentence_id是否匹配
|
||||
if message.sentence_id != self.conn.sentence_id:
|
||||
continue
|
||||
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}"
|
||||
)
|
||||
|
||||
if self.conn.client_abort:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.finish_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
continue
|
||||
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
@@ -252,7 +259,6 @@ class TTSProvider(TTSProviderBase):
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
logger.bind(tag=TAG).debug("TTS文本发送成功")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
||||
continue
|
||||
@@ -317,22 +323,20 @@ class TTSProvider(TTSProviderBase):
|
||||
async def start_session(self, task_id):
|
||||
logger.bind(tag=TAG).debug("开始会话～～")
|
||||
try:
|
||||
# 会话开始时检测上个会话的监听状态
|
||||
if (
|
||||
self._monitor_task is not None
|
||||
and isinstance(self._monitor_task, Task)
|
||||
and not self._monitor_task.done()
|
||||
):
|
||||
logger.bind(tag=TAG).info(
|
||||
"检测到未完成的上个会话，关闭监听任务和连接..."
|
||||
)
|
||||
# 上个会话处于激活状态时关闭上个连接新建链接
|
||||
if self.activate_session:
|
||||
await self.close()
|
||||
|
||||
# 设置会话激活标志
|
||||
self.activate_session = True
|
||||
|
||||
# 建立新连接
|
||||
await self._ensure_connection()
|
||||
|
||||
# 启动监听任务
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
if self._monitor_task is None or self._monitor_task.done():
|
||||
logger.bind(tag=TAG).debug("启动监听任务...")
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
|
||||
start_request = {
|
||||
"header": {
|
||||
@@ -377,15 +381,7 @@ class TTSProvider(TTSProviderBase):
|
||||
await self.ws.send(json.dumps(stop_request))
|
||||
logger.bind(tag=TAG).debug("会话结束请求已发送")
|
||||
self.last_active_time = time.time()
|
||||
if self._monitor_task:
|
||||
try:
|
||||
await self._monitor_task
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(
|
||||
f"等待监听任务完成时发生错误: {str(e)}"
|
||||
)
|
||||
finally:
|
||||
self._monitor_task = None
|
||||
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
|
||||
# 确保清理资源
|
||||
@@ -394,6 +390,8 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def close(self):
|
||||
"""资源清理"""
|
||||
await super().close()
|
||||
self.activate_session = False
|
||||
if self._monitor_task:
|
||||
try:
|
||||
self._monitor_task.cancel()
|
||||
@@ -413,17 +411,13 @@ class TTSProvider(TTSProviderBase):
|
||||
self.last_active_time = None
|
||||
|
||||
async def _start_monitor_tts_response(self):
|
||||
"""监听TTS响应"""
|
||||
"""监听TTS响应 - 长期运行"""
|
||||
try:
|
||||
session_finished = False # 标记会话是否正常结束
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
msg = await self.ws.recv()
|
||||
self.last_active_time = time.time()
|
||||
# 检查客户端是否中止
|
||||
if self.conn.client_abort:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应")
|
||||
break
|
||||
|
||||
if isinstance(msg, str): # 文本控制消息
|
||||
try:
|
||||
data = json.loads(msg)
|
||||
@@ -436,19 +430,19 @@ class TTSProvider(TTSProviderBase):
|
||||
)
|
||||
elif event_name == "SentenceEnd":
|
||||
# 发送缓存的数据
|
||||
if self.conn.tts_MessageText:
|
||||
tts_text = self.get_tts_text(self.conn.sentence_id)
|
||||
if tts_text:
|
||||
logger.bind(tag=TAG).info(
|
||||
f"句子语音生成成功： {self.conn.tts_MessageText}"
|
||||
f"句子语音生成成功： {tts_text}"
|
||||
)
|
||||
self.tts_audio_queue.put(
|
||||
(SentenceType.FIRST, [], self.conn.tts_MessageText)
|
||||
(SentenceType.FIRST, [], tts_text)
|
||||
)
|
||||
self.conn.tts_MessageText = None
|
||||
self.clear_tts_text(self.conn.sentence_id)
|
||||
elif event_name == "SynthesisCompleted":
|
||||
logger.bind(tag=TAG).debug(f"会话结束～～")
|
||||
self.activate_session = False
|
||||
self._process_before_stop_play_files()
|
||||
session_finished = True
|
||||
break
|
||||
except json.JSONDecodeError:
|
||||
logger.bind(tag=TAG).warning("收到无效的JSON消息")
|
||||
# 二进制消息（音频数据）
|
||||
@@ -462,8 +456,8 @@ class TTSProvider(TTSProviderBase):
|
||||
f"处理TTS响应时出错: {e}\n{traceback.format_exc()}"
|
||||
)
|
||||
break
|
||||
# 仅在连接异常时才关闭
|
||||
if not session_finished and self.ws:
|
||||
# 连接异常时关闭WebSocket
|
||||
if self.ws:
|
||||
try:
|
||||
await self.ws.close()
|
||||
except:
|
||||
@@ -471,6 +465,7 @@ class TTSProvider(TTSProviderBase):
|
||||
self.ws = None
|
||||
# 监听任务退出时清理引用
|
||||
finally:
|
||||
self.activate_session = False
|
||||
self._monitor_task = None
|
||||
|
||||
def audio_to_opus_data_stream(
|
||||
|
||||
@@ -5,6 +5,7 @@ import queue
|
||||
import asyncio
|
||||
import threading
|
||||
import traceback
|
||||
import concurrent.futures
|
||||
|
||||
from core.utils import p3
|
||||
from datetime import datetime
|
||||
@@ -42,6 +43,8 @@ class TTSProviderBase(ABC):
|
||||
self.tts_audio_first_sentence = True
|
||||
self.before_stop_play_files = []
|
||||
self.report_on_last = False
|
||||
# sentence_id 到文本的映射，用于流式TTS获取正确的字幕文本
|
||||
self._sentence_text_map = {}
|
||||
|
||||
self.tts_text_buff = []
|
||||
self.punctuations = (
|
||||
@@ -277,14 +280,49 @@ class TTSProviderBase(ABC):
|
||||
)
|
||||
self.audio_play_priority_thread.start()
|
||||
|
||||
def store_tts_text(self, sentence_id, text):
|
||||
"""存储指定 sentence_id 对应的文本，用于流式TTS获取正确的字幕文本
|
||||
|
||||
Args:
|
||||
sentence_id: 会话ID
|
||||
text: 要存储的文本
|
||||
"""
|
||||
if sentence_id and text:
|
||||
self._sentence_text_map[sentence_id] = text
|
||||
# 只保留最近 5 个，防止内存泄漏
|
||||
if len(self._sentence_text_map) > 5:
|
||||
oldest = next(iter(self._sentence_text_map))
|
||||
del self._sentence_text_map[oldest]
|
||||
|
||||
def get_tts_text(self, sentence_id):
|
||||
"""获取指定 sentence_id 对应的文本
|
||||
|
||||
Args:
|
||||
sentence_id: 会话ID
|
||||
|
||||
Returns:
|
||||
str: 对应的文本，如果不存在返回 None
|
||||
"""
|
||||
return self._sentence_text_map.get(sentence_id)
|
||||
|
||||
def clear_tts_text(self, sentence_id):
|
||||
"""清除指定 sentence_id 的文本
|
||||
|
||||
Args:
|
||||
sentence_id: 会话ID
|
||||
"""
|
||||
if sentence_id in self._sentence_text_map:
|
||||
del self._sentence_text_map[sentence_id]
|
||||
|
||||
# 这里默认是非流式的处理方式
|
||||
# 流式处理方式请在子类中重写
|
||||
def tts_text_priority_thread(self):
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
message = self.tts_text_queue.get(timeout=1)
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
self.conn.client_abort = False
|
||||
# 过滤旧消息：检查sentence_id是否匹配
|
||||
if message.sentence_id != self.conn.sentence_id:
|
||||
continue
|
||||
if self.conn.client_abort:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
|
||||
continue
|
||||
@@ -386,6 +424,7 @@ class TTSProviderBase(ABC):
|
||||
|
||||
async def close(self):
|
||||
"""资源清理方法"""
|
||||
self._sentence_text_map.clear()
|
||||
if hasattr(self, "ws") and self.ws:
|
||||
await self.ws.close()
|
||||
|
||||
|
||||
@@ -221,7 +221,7 @@ class TTSProvider(TTSProviderBase):
|
||||
try:
|
||||
if self.ws:
|
||||
if self.enable_ws_reuse:
|
||||
logger.bind(tag=TAG).info(f"使用已有链接...")
|
||||
logger.bind(tag=TAG).debug(f"使用已有链接...")
|
||||
return self.ws
|
||||
else:
|
||||
try:
|
||||
@@ -272,12 +272,14 @@ class TTSProvider(TTSProviderBase):
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
message = self.tts_text_queue.get(timeout=1)
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
|
||||
)
|
||||
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
self.conn.client_abort = False
|
||||
# 过滤旧消息：检查sentence_id是否匹配
|
||||
if message.sentence_id != self.conn.sentence_id:
|
||||
continue
|
||||
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}"
|
||||
)
|
||||
|
||||
if self.conn.client_abort:
|
||||
try:
|
||||
@@ -327,7 +329,6 @@ class TTSProvider(TTSProviderBase):
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
logger.bind(tag=TAG).debug("TTS文本发送成功")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
||||
continue
|
||||
@@ -387,15 +388,8 @@ class TTSProvider(TTSProviderBase):
|
||||
async def start_session(self, session_id):
|
||||
logger.bind(tag=TAG).debug(f"开始会话～～{session_id}")
|
||||
try:
|
||||
# 等待上一个会话结束，最多等待3次
|
||||
for _ in range(3):
|
||||
if not self.activate_session:
|
||||
break
|
||||
logger.bind(tag=TAG).debug(f"等待上一个会话结束...")
|
||||
await asyncio.sleep(0.1)
|
||||
else:
|
||||
# 等待超时，强制清除连接状态
|
||||
logger.bind(tag=TAG).debug("等待上一个会话超时，清除连接状态...")
|
||||
# 上个会话处于激活状态时关闭上个连接新建链接
|
||||
if self.activate_session:
|
||||
await self.close()
|
||||
|
||||
# 设置会话激活标志
|
||||
@@ -468,6 +462,7 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def close(self):
|
||||
"""资源清理方法"""
|
||||
await super().close()
|
||||
self.activate_session = False
|
||||
# 取消监听任务
|
||||
if self._monitor_task:
|
||||
@@ -525,14 +520,16 @@ class TTSProvider(TTSProviderBase):
|
||||
and res.header.message_type == AUDIO_ONLY_RESPONSE
|
||||
):
|
||||
# 处理seed-tts-2.0文本字幕
|
||||
if self.resource_type and 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 self.resource_type:
|
||||
tts_text = self.get_tts_text(self.conn.sentence_id)
|
||||
if tts_text:
|
||||
logger.bind(tag=TAG).info(
|
||||
f"句子语音生成成功： {tts_text}"
|
||||
)
|
||||
self.tts_audio_queue.put(
|
||||
(SentenceType.FIRST, [], tts_text)
|
||||
)
|
||||
self.clear_tts_text(self.conn.sentence_id)
|
||||
self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
|
||||
elif not self.resource_type and res.optional.event == EVENT_TTSSentenceEnd:
|
||||
logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
|
||||
|
||||
@@ -117,6 +117,7 @@ class TTSProvider(TTSProviderBase):
|
||||
# WebSocket配置
|
||||
self.ws = None
|
||||
self._monitor_task = None
|
||||
self.activate_session = False
|
||||
|
||||
# 序列号管理
|
||||
self.text_seq = 0
|
||||
@@ -128,7 +129,7 @@ class TTSProvider(TTSProviderBase):
|
||||
async def _ensure_connection(self):
|
||||
"""确保WebSocket连接可用"""
|
||||
try:
|
||||
logger.bind(tag=TAG).info("开始建立新连接...")
|
||||
logger.bind(tag=TAG).debug("开始建立新连接...")
|
||||
|
||||
# 生成认证URL
|
||||
auth_url = XunfeiWSAuth.create_auth_url(
|
||||
@@ -141,7 +142,7 @@ class TTSProvider(TTSProviderBase):
|
||||
ping_timeout=10,
|
||||
close_timeout=10,
|
||||
)
|
||||
logger.bind(tag=TAG).info("WebSocket连接建立成功")
|
||||
logger.bind(tag=TAG).debug("WebSocket连接建立成功")
|
||||
return self.ws
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
|
||||
@@ -153,16 +154,21 @@ class TTSProvider(TTSProviderBase):
|
||||
while not self.conn.stop_event.is_set():
|
||||
try:
|
||||
message = self.tts_text_queue.get(timeout=1)
|
||||
|
||||
# 过滤旧消息：检查sentence_id是否匹配
|
||||
if message.sentence_id != self.conn.sentence_id:
|
||||
continue
|
||||
|
||||
logger.bind(tag=TAG).debug(
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
|
||||
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}"
|
||||
)
|
||||
|
||||
if message.sentence_type == SentenceType.FIRST:
|
||||
# 重置序列号
|
||||
self.text_seq = 0
|
||||
self.conn.client_abort = False
|
||||
# 增加序列号
|
||||
self.text_seq += 1
|
||||
|
||||
if self.conn.client_abort:
|
||||
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
|
||||
continue
|
||||
@@ -172,16 +178,16 @@ class TTSProvider(TTSProviderBase):
|
||||
try:
|
||||
if not getattr(self.conn, "sentence_id", None):
|
||||
self.conn.sentence_id = uuid.uuid4().hex
|
||||
logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}")
|
||||
logger.bind(tag=TAG).debug(f"自动生成新的 会话ID: {self.conn.sentence_id}")
|
||||
|
||||
logger.bind(tag=TAG).info("开始启动TTS会话...")
|
||||
logger.bind(tag=TAG).debug("开始启动TTS会话...")
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
self.start_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
self.before_stop_play_files.clear()
|
||||
logger.bind(tag=TAG).info("TTS会话启动成功")
|
||||
logger.bind(tag=TAG).debug("TTS会话启动成功")
|
||||
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
|
||||
@@ -199,7 +205,6 @@ class TTSProvider(TTSProviderBase):
|
||||
loop=self.conn.loop,
|
||||
)
|
||||
future.result(timeout=self.tts_timeout)
|
||||
logger.bind(tag=TAG).debug("TTS文本发送成功")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
||||
# 不使用continue，确保后续处理不被中断
|
||||
@@ -216,7 +221,7 @@ class TTSProvider(TTSProviderBase):
|
||||
# 处理会话结束
|
||||
if message.sentence_type == SentenceType.LAST:
|
||||
try:
|
||||
logger.bind(tag=TAG).info("开始结束TTS会话...")
|
||||
logger.bind(tag=TAG).debug("开始结束TTS会话...")
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.finish_session(self.conn.sentence_id),
|
||||
loop=self.conn.loop,
|
||||
@@ -257,30 +262,28 @@ class TTSProvider(TTSProviderBase):
|
||||
raise
|
||||
|
||||
async def start_session(self, session_id):
|
||||
logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
|
||||
logger.bind(tag=TAG).debug(f"开始会话～～{session_id}")
|
||||
try:
|
||||
# 会话开始时检测上个会话的监听状态
|
||||
if (
|
||||
self._monitor_task is not None
|
||||
and isinstance(self._monitor_task, Task)
|
||||
and not self._monitor_task.done()
|
||||
):
|
||||
logger.bind(tag=TAG).info(
|
||||
"检测到未完成的上个会话，关闭监听任务和连接..."
|
||||
)
|
||||
# 上个会话处于激活状态时关闭上个连接新建链接
|
||||
if self.activate_session:
|
||||
await self.close()
|
||||
|
||||
# 设置会话激活标志
|
||||
self.activate_session = True
|
||||
|
||||
# 建立新连接
|
||||
await self._ensure_connection()
|
||||
|
||||
# 启动监听任务
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
if self._monitor_task is None or self._monitor_task.done():
|
||||
logger.bind(tag=TAG).debug("启动监听任务...")
|
||||
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
||||
|
||||
# 发送会话启动请求
|
||||
start_request = self._build_base_request(status=0)
|
||||
|
||||
await self.ws.send(json.dumps(start_request))
|
||||
logger.bind(tag=TAG).info("会话启动请求已发送")
|
||||
logger.bind(tag=TAG).debug("会话启动请求已发送")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
|
||||
# 确保清理资源
|
||||
@@ -288,13 +291,13 @@ class TTSProvider(TTSProviderBase):
|
||||
raise
|
||||
|
||||
async def finish_session(self, session_id):
|
||||
logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
|
||||
logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}")
|
||||
try:
|
||||
if self.ws:
|
||||
# 发送会话结束请求
|
||||
stop_request = self._build_base_request(status=2)
|
||||
await self.ws.send(json.dumps(stop_request))
|
||||
logger.bind(tag=TAG).info("会话结束请求已发送")
|
||||
logger.bind(tag=TAG).debug("会话结束请求已发送")
|
||||
|
||||
if self._monitor_task:
|
||||
try:
|
||||
@@ -310,6 +313,8 @@ class TTSProvider(TTSProviderBase):
|
||||
|
||||
async def close(self):
|
||||
"""资源清理"""
|
||||
await super().close()
|
||||
self.activate_session = False
|
||||
if self._monitor_task:
|
||||
try:
|
||||
self._monitor_task.cancel()
|
||||
@@ -358,17 +363,19 @@ class TTSProvider(TTSProviderBase):
|
||||
)
|
||||
elif status == 2:
|
||||
logger.bind(tag=TAG).debug("收到结束状态的音频数据，TTS合成完成")
|
||||
self.activate_session = False
|
||||
self._process_before_stop_play_files()
|
||||
break
|
||||
else:
|
||||
if self.conn.tts_MessageText:
|
||||
tts_text = self.get_tts_text(self.conn.sentence_id)
|
||||
if tts_text:
|
||||
logger.bind(tag=TAG).info(
|
||||
f"句子语音生成成功： {self.conn.tts_MessageText}"
|
||||
f"句子语音生成成功： {tts_text}"
|
||||
)
|
||||
self.tts_audio_queue.put(
|
||||
(SentenceType.FIRST, [], self.conn.tts_MessageText)
|
||||
(SentenceType.FIRST, [], tts_text)
|
||||
)
|
||||
self.conn.tts_MessageText = None
|
||||
self.clear_tts_text(self.conn.sentence_id)
|
||||
try:
|
||||
audio_bytes = base64.b64decode(audio_data)
|
||||
self.opus_encoder.encode_pcm_to_opus_stream(
|
||||
@@ -405,6 +412,7 @@ class TTSProvider(TTSProviderBase):
|
||||
self.ws = None
|
||||
# 监听任务退出时清理引用
|
||||
finally:
|
||||
self.activate_session = False
|
||||
self._monitor_task = None
|
||||
|
||||
def to_tts(self, text: str) -> list:
|
||||
|
||||
Reference in New Issue
Block a user