diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 473c095b..4230715a 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -718,6 +718,7 @@ TTS: #各参数意义见开发文档：https://tts.linkerai.top/docs#/default/text_to_speech_tts_get type: linkerai api_url: https://tts.linkerai.top/tts + audio_format: "pcm" access_token: "test" voice: "OUeAo1mhq6IBExi" output_dir: tmp/ \ 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 0b17eef5..6bc77dff 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -331,10 +331,11 @@ class TTSProviderBase(ABC): segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: 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) + if tts_file: + 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 diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 8ad069bd..cdb7296e 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,3 +1,8 @@ +import asyncio +import traceback +import queue +import requests +from pathlib import Path from core.providers.tts.base import TTSProviderBase from core.providers.tts.dto.dto import ( TTSMessageDTO, @@ -6,8 +11,7 @@ from core.providers.tts.dto.dto import ( InterfaceType ) from config.logger import setup_logging -from core.utils import opus_encoder_utils -import requests +from core.utils import opus_encoder_utils, textUtils TAG = __name__ logger = setup_logging() @@ -20,8 +24,6 @@ class TTSProvider(TTSProviderBase): self.access_token = config.get("access_token") self.voice = config.get("voice") self.api_url = config.get("api_url") - - # 根据配置选择音频格式，优先使用opus self.audio_format = config.get("audio_format", "opus") # 创建Opus编码器 @@ -31,15 +33,135 @@ class TTSProvider(TTSProviderBase): frame_size_ms=60 ) - async def text_to_speak(self, text, _): - """将文本转换为语音（流式）""" - await self.send_text(self.voice, text) + # 添加文本缓冲区 + self.text_buffer = "" + # 句子结束标点集合 + self.sentence_endings = ("。", "？", "！", "；", ":", ".", "?", "!", ";","……") + # 逗号类标点（用于第一句话分割） + self.comma_endings = ("，", "~", "、", ",", "。", ".", "？", "?", "！", "!", "；", ";", "：",) + + # PCM缓冲区 + self.pcm_buffer = bytearray() + + ################################################################################### + # linkerai单流式TTS重写父类的方法--开始 + ################################################################################### + + def tts_text_priority_thread(self): + """流式文本处理线程""" + 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}" + ) + + if message.sentence_type == SentenceType.FIRST: + # 初始化流式状态 + self.tts_audio_first_sentence = True + self.pcm_buffer = bytearray() + self.text_buffer = "" # 重置文本缓冲区 + + elif ContentType.TEXT == message.content_type: + # 将文本添加到缓冲区 + self.text_buffer += message.content_detail + # 尝试分割并发送完整句子 + self._process_text_buffer() + + elif ContentType.FILE == message.content_type: + # 先处理缓冲区中的剩余文本 + self._flush_text_buffer() + # 处理文件类型 + if message.content_file and Path(message.content_file).exists(): + audio_datas = self._process_audio_file(message.content_file) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, audio_datas, message.content_detail) + ) + + if message.sentence_type == SentenceType.LAST: + # 处理缓冲区中的剩余文本 + self._flush_text_buffer() + # 发送结束帧 + if self.pcm_buffer: + opus_datas = self.wav_to_opus_data_audio_raw(self.pcm_buffer, is_end=True) + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, "")) + self.pcm_buffer = bytearray() + self.tts_audio_queue.put((SentenceType.LAST, [], None)) + self.text_buffer = "" # 重置文本缓冲区 + + except queue.Empty: + continue + except Exception as e: + logger.bind(tag=TAG).error( + f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) + + def _process_text_buffer(self): + """处理文本缓冲区，分割并发送完整句子""" + while True: + # 查找最近的句子结束位置 + end_pos = -1 + for punct in self.sentence_endings: + pos = self.text_buffer.find(punct) + if pos != -1 and (end_pos == -1 or pos < end_pos): + end_pos = pos + + # 如果是第一句话，也允许在逗号处分隔 + if self.tts_audio_first_sentence and end_pos == -1: + for punct in self.comma_endings: + pos = self.text_buffer.find(punct) + if pos != -1 and (end_pos == -1 or pos < end_pos): + end_pos = pos + + # 找到分割点 + if end_pos != -1: + # 提取完整句子 + sentence = self.text_buffer[:end_pos + 1] + sentence = textUtils.get_string_no_punctuation_or_emoji(sentence) + + if not sentence.strip(): # 检查是否为空文本 + self.text_buffer = self.text_buffer[end_pos + 1:] + continue + + self.text_buffer = self.text_buffer[end_pos + 1:] + + # 发送句子 + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(sentence), + loop=self.conn.loop + ) + future.result() + + # 更新第一句话标志 + if self.tts_audio_first_sentence: + self.tts_audio_first_sentence = False + else: + break + + def _flush_text_buffer(self): + """处理缓冲区中剩余的文本""" + if self.text_buffer: + clean_text = textUtils.get_string_no_punctuation_or_emoji(self.text_buffer) + if clean_text.strip(): # 检查是否为空文本 + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(clean_text), + loop=self.conn.loop + ) + future.result() + self.text_buffer = "" + + async def text_to_speak(self, text): + # 发送文本 + await self.send_text(text) return - async def send_text(self, speaker: str, text: str): - """向 TTS 服务发送文本并获取音频流""" + ################################################################################### + # linkerai单流式TTS重写父类的方法--结束 + ################################################################################### + + async def send_text(self, text: str): + """流式处理TTS音频""" try: - # 构造请求参数 params = { "tts_text": text, "spk_id": self.voice, @@ -48,48 +170,38 @@ class TTSProvider(TTSProviderBase): "target_sr": 16000, "audio_format": self.audio_format, } + headers = {"Authorization": f"Bearer {self.access_token}"} - # 构造请求头 - headers = { - "Authorization": f"Bearer {self.access_token}", - } + with requests.get(self.api_url, params=params, headers=headers, stream=True) as response: + if response.status_code != 200: + logger.error(f"TTS请求失败: {response.status_code}, {response.text}") + return - # 发送流式请求 - response = requests.get( - self.api_url, - params=params, - headers=headers, - stream=True - ) + logger.debug(f"处理TTS文本: {text}") - # 检查响应状态 - if response.status_code != 200: - logger.bind(tag=TAG).error(f"TTS 请求失败: {response.status_code}, {response.text}") - return + # 流式处理音频数据 + for chunk in response.iter_content(chunk_size=1024): + if chunk: + # 实时编码并发送音频帧 + self.pcm_buffer.extend(chunk) - # 处理音频流 - audio_frames = [] - for chunk in response.iter_content(chunk_size=1024): - if chunk: - if self.audio_format == "pcm": - # 将PCM转换为Opus - opus_frames = self.opus_encoder.encode_pcm_to_opus(chunk, False) - audio_frames.extend(opus_frames) - else: - # 直接使用Opus帧 - audio_frames.append(chunk) - - # 将音频帧放入队列 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, audio_frames, text) - ) + # 处理剩余缓冲区数据 + if self.pcm_buffer: + opus_datas = self.wav_to_opus_data_audio_raw(self.pcm_buffer, is_end=True) + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, text)) + self.pcm_buffer = bytearray() except Exception as e: - logger.bind(tag=TAG).error(f"TTS 流式处理异常：{str(e)}") + logger.error(f"TTS流式处理异常：{str(e)}") raise + # 保持原有方法 + 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 + async def close(self): - """资源清理方法""" + """资源清理""" await super().close() if hasattr(self, "opus_encoder"): self.opus_encoder.close()