diff --git a/config.yaml b/config.yaml index 6d9cd989..76ad130b 100644 --- a/config.yaml +++ b/config.yaml @@ -129,7 +129,7 @@ LLM: # 可在这里找到你的api key https://bigmodel.cn/usercenter/proj-mgmt/apikeys model_name: glm-4-flash url: https://open.bigmodel.cn/api/paas/v4/ - api_key: 你的chat-glm api key + api_key: f04500d80a234c9085be2b4020dce25f.gnnnWr3BRLuqeDHr OllamaLLM: # 定义LLM API类型 type: ollama @@ -168,6 +168,11 @@ LLM: base_url: http://homeassistant.local:8123 agent_id: conversation.chatgpt api_key: 你的home assistant api访问令牌 + +TTS_SET: + TTS_STREAM: false #是否启动流响应 true/false，默认false + MAX_WORKERS: 4 #并发请求tts的数量，这个调太大，本地tts会变慢 + TTS: # 当前支持的type为edge、doubao，可自行适配 EdgeTTS: @@ -219,21 +224,29 @@ TTS: output_file: tmp/ response_format: wav reference_id: null - reference_audio: ["/tmp/test.wav",] - reference_text: ["你弄来这些吟词宴曲来看，还是这些混话来欺负我。",] + reference_audio: + - "audio_ref/shinchan.wav" + - "audio_ref/shinchan2.wav" + - "audio_ref/shinchan3.wav" + - "audio_ref/shinchan4.wav" + reference_text: + - "难过的事讨厌的事丢脸的事全都集中起来，在这里，让水冲走所有的烦恼不就好了吗。" + - "我还是觉得船到桥头自然直这句话最棒了。" + - "我不是小鬼，我是野原新之助，小新这次的目标是成为圆梦之星哦，在努力一下好了，是我的脚自己要走这么快的喔，竞争就是这么激烈。小白，我们走了，成为圆梦之星是不是可以打败怪兽呢。" + - "工钱，今天一起算，有磨能使鬼推钱。哎呀，你难倒我了，像我这么乖的小孩，怎么更乖。" normalize: true max_new_tokens: 1024 chunk_length: 200 top_p: 0.7 repetition_penalty: 1.2 temperature: 0.7 - streaming: false + streaming: true use_memory_cache: "on" seed: null channels: 1 rate: 44100 - api_key: "你的api_key" - api_url: "http://127.0.0.1:8080/v1/tts" + api_key: "xxx" + api_url: "http://192.168.101.242:10000/tts-fishspeech/v1/tts" GPT_SOVITS_V2: # 定义TTS API类型 #启动tts方法： diff --git a/core/connection.py b/core/connection.py index 588a3a2c..619ad922 100644 --- a/core/connection.py +++ b/core/connection.py @@ -13,7 +13,7 @@ from core.utils.dialogue import Message, Dialogue from core.handle.textHandle import handleTextMessage from core.utils.util import get_string_no_punctuation_or_emoji from concurrent.futures import ThreadPoolExecutor, TimeoutError -from core.handle.sendAudioHandle import sendAudioMessage +from core.handle.sendAudioHandle import sendAudioMessage, sendAudioMessageStream from core.handle.receiveAudioHandle import handleAudioMessage from config.private_config import PrivateConfig from core.auth import AuthMiddleware, AuthenticationError @@ -32,6 +32,8 @@ class ConnectionHandler: self.logger = setup_logging() self.auth = AuthMiddleware(config) + self.tts_stream = self.config.get("TTS_SET", {}).get("TTS_STREAM", False) + self.websocket = None self.headers = None self.session_id = None @@ -46,8 +48,10 @@ class ConnectionHandler: self.loop = asyncio.get_event_loop() self.stop_event = threading.Event() self.tts_queue = queue.Queue() + self.tts_queue_stream = queue.Queue() self.audio_play_queue = queue.Queue() - self.executor = ThreadPoolExecutor(max_workers=10) + max_workers = self.config.get("TTS_SET", {}).get("MAX_WORKERS", 10) + self.executor = ThreadPoolExecutor(max_workers=max_workers) # 依赖的组件 self.vad = _vad @@ -74,6 +78,7 @@ class ConnectionHandler: # tts相关变量 self.tts_first_text_index = -1 self.tts_last_text_index = -1 + self.tts_duration = 0 # iot相关变量 self.iot_descriptors = {} @@ -262,8 +267,17 @@ class ConnectionHandler: # segment_text = " " text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) - self.tts_queue.put(future) + if self.tts_stream: + stream_queue = queue.Queue() + self.executor.submit(self.speak_and_play_stream, segment_text, stream_queue) + self.tts_queue_stream.put({ + "text": segment_text, + "chunk_queque": stream_queue, + "text_index": text_index + }) + else: + future = self.executor.submit(self.speak_and_play, segment_text, text_index) + self.tts_queue.put(future) processed_chars += len(segment_text_raw) # 更新已处理字符位置 # 处理最后剩余的文本 @@ -274,8 +288,17 @@ class ConnectionHandler: if segment_text: text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) - self.tts_queue.put(future) + if self.tts_stream: + stream_queue = queue.Queue() + self.executor.submit(self.speak_and_play_stream, segment_text, stream_queue, text_index) + self.tts_queue_stream.put({ + "text": segment_text, + "chunk_queque": stream_queue, + "text_index": text_index + }) + else: + future = self.executor.submit(self.speak_and_play, segment_text, text_index) + self.tts_queue.put(future) self.llm_finish_task = True self.dialogue.put(Message(role="assistant", content="".join(response_message))) @@ -283,6 +306,12 @@ class ConnectionHandler: return True def _tts_priority_thread(self): + if self.tts_stream: + self._tts_priority_thread_stream() + else: + self._tts_priority_thread_no_stream() + + def _tts_priority_thread_no_stream(self): while not self.stop_event.is_set(): text = None try: @@ -322,14 +351,47 @@ class ConnectionHandler: ) self.logger.bind(tag=TAG).error(f"tts_priority priority_thread: {text} {e}") + def _tts_priority_thread_stream(self): + while not self.stop_event.is_set(): + text = None + try: + tts_stream_queue_msg = self.tts_queue_stream.get() + try: + text = tts_stream_queue_msg["text"] + chunk_queque = tts_stream_queue_msg["chunk_queque"] + text_index = tts_stream_queue_msg["text_index"] + except TimeoutError: + self.logger.error("TTS 任务超时") + continue + except Exception as e: + self.logger.error(f"TTS 任务出错: {e}") + continue + if not self.client_abort: + # 如果没有中途打断就发送语音 + self.audio_play_queue.put((chunk_queque, text, text_index)) + except Exception as e: + self.clearSpeakStatus() + asyncio.run_coroutine_threadsafe( + self.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": self.session_id})), + self.loop + ) + self.logger.error(f"tts_priority priority_thread: {text}{e}") + def _audio_play_priority_thread(self): while not self.stop_event.is_set(): text = None try: - opus_datas, text, text_index = self.audio_play_queue.get() - future = asyncio.run_coroutine_threadsafe(sendAudioMessage(self, opus_datas, text, text_index), - self.loop) - future.result() + if self.tts_stream: + chunk_queque, text, text_index = self.audio_play_queue.get() + future = asyncio.run_coroutine_threadsafe( + sendAudioMessageStream(self, chunk_queque, text, text_index), + self.loop) + future.result() + else: + opus_datas, text, text_index = self.audio_play_queue.get() + future = asyncio.run_coroutine_threadsafe(sendAudioMessage(self, opus_datas, text, text_index), + self.loop) + future.result() except Exception as e: self.logger.bind(tag=TAG).error(f"audio_play_priority priority_thread: {text} {e}") @@ -344,11 +406,18 @@ class ConnectionHandler: self.logger.bind(tag=TAG).debug(f"TTS 文件生成完毕: {tts_file}") return tts_file, text, text_index + def speak_and_play_stream(self, text, queue: queue.Queue, text_index=0): + if text is None or len(text) <= 0: + self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{text}") + return None, text + self.tts.to_tts_stream(text, queue, text_index) + def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") self.asr_server_receive = True self.tts_last_text_index = -1 self.tts_first_text_index = -1 + self.tts_duration = 0 def recode_first_last_text(self, text, text_index=0): if self.tts_first_text_index == -1: diff --git a/core/handle/sendAudioHandle.py b/core/handle/sendAudioHandle.py index d0dcb82f..230bf269 100644 --- a/core/handle/sendAudioHandle.py +++ b/core/handle/sendAudioHandle.py @@ -1,3 +1,5 @@ +import traceback + from config.logger import setup_logging import json import asyncio @@ -15,6 +17,65 @@ async def isLLMWantToFinish(last_text): return False +async def sendAudioMessageStream(conn, audios_queue, text, text_index=0): + # 发送句子开始消息 + if text_index == conn.tts_first_text_index: + logger.bind(tag=TAG).info(f"发送第一段语音: {text}") + await send_tts_message(conn, "sentence_start", text) + + # 初始化流控参数 + frame_duration = 60 # 毫秒 + start_time = time.time() # 使用高精度计时器 + play_position = 0 # 已播放的时长（毫秒） + time_out_stop = False + while True: + try: + start_get_queue = time.time() + # 尝试获取数据，如果没有数据，则等待一小段时间再试 + audio_data_chunke = None + try: + audio_data_chunke = audios_queue.get(timeout=5) # 设置超时为1秒 + except Exception as e: + # 如果超时，继续等待 + logger.bind(tag=TAG).error(f"获取队列超时～{e}") + + audio_opus_datas = audio_data_chunke.get('data') if audio_data_chunke else None + duration = audio_data_chunke.get('duration') if audio_data_chunke else 0 + + if audio_data_chunke: + start_time = time.time() + # 检查是否超过 5 秒没有数据 + if time.time() - start_time > 15: + logger.bind(tag=TAG).error("超过15秒没有数据，退出。") + break + + if audio_data_chunke and audio_data_chunke.get("end", True): + break + + if audio_opus_datas: + queue_duration = time.time() - start_get_queue + last_duration = conn.tts_duration - queue_duration + if last_duration <= 0: + last_duration = 0 + conn.tts_duration = duration + last_duration + for opus_packet in audio_opus_datas: + await conn.websocket.send(opus_packet) + start_time = time.time() # 更新获取数据的时间 + except Exception as e: + logger.bind(tag=TAG).error(f"发生错误: {e}") + traceback.print_exc() # 打印错误堆栈 + await send_tts_message(conn, "sentence_end", text) + + print(f'{text_index}-{conn.tts_last_text_index}') + # 发送结束消息（如果是最后一个文本） + if conn.llm_finish_task and text_index == conn.tts_last_text_index: + if conn.tts_duration and conn.tts_duration > 0: + await asyncio.sleep(conn.tts_duration) + await send_tts_message(conn, 'stop', None) + if await isLLMWantToFinish(text): + await conn.close() + + async def sendAudioMessage(conn, audios, text, text_index=0): # 发送句子开始消息 if text_index == conn.tts_first_text_index: diff --git a/core/providers/tts/base.py b/core/providers/tts/base.py index 84737608..989e8382 100644 --- a/core/providers/tts/base.py +++ b/core/providers/tts/base.py @@ -5,6 +5,7 @@ import numpy as np import opuslib_next from pydub import AudioSegment from abc import ABC, abstractmethod +import queue TAG = __name__ logger = setup_logging() @@ -33,6 +34,13 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).info(f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次") return tmp_file + except Exception as e: + logger.bind(tag=TAG).info(f": {e}") + return None + + def to_tts_stream(self, text, queue: queue.Queue, text_index=0): + try: + asyncio.run(self.text_to_speak_stream(text, queue, text_index)) except Exception as e: logger.bind(tag=TAG).info(f"Failed to generate TTS file: {e}") return None @@ -41,6 +49,9 @@ class TTSProviderBase(ABC): async def text_to_speak(self, text, output_file): pass + async def text_to_speak_stream(self, text, queue: queue.Queue, text_index=0): + raise Exception("该TTS还没有实现stream模式") + def wav_to_opus_data(self, wav_file_path): # 使用pydub加载PCM文件 # 获取文件后缀名 @@ -82,3 +93,31 @@ class TTSProviderBase(ABC): opus_datas.append(opus_data) return opus_datas, duration + + def wav_to_opus_data_audio_raw(self, raw_data): + # 初始化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 + + opus_datas = [] + # 按帧处理所有音频数据（包括最后一帧可能补零） + 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: + # logger.bind(tag=TAG).info("开始补0") + chunk += b'\x00' * (frame_size * 2 - len(chunk)) + + # 转换为numpy数组处理 + np_frame = np.frombuffer(chunk, dtype=np.int16) + + # 编码Opus数据 + opus_data = encoder.encode(np_frame.tobytes(), frame_size) + opus_datas.append(opus_data) + + return opus_datas diff --git a/core/providers/tts/fishspeech.py b/core/providers/tts/fishspeech.py index 5794a2e3..c96db551 100644 --- a/core/providers/tts/fishspeech.py +++ b/core/providers/tts/fishspeech.py @@ -1,11 +1,20 @@ import base64 import os +import traceback import uuid +import queue +import io + +import numpy as np import requests import ormsgpack from pathlib import Path + +import torch +import torchaudio from pydantic import BaseModel, Field, conint, model_validator +from pydub import AudioSegment from typing_extensions import Annotated from datetime import datetime from typing import Literal @@ -155,4 +164,102 @@ class TTSProvider(TTSProviderBase): print(f"Request failed with status code {response.status_code}") print(response.json()) + def _get_audio_from_tts(self, data_bytes): + tts_speech = torch.from_numpy(np.array(np.frombuffer(data_bytes, dtype=np.int16))).unsqueeze(dim=0) + with io.BytesIO() as bf: + torchaudio.save(bf, tts_speech, 44100, format="wav") + audio = AudioSegment.from_file(bf, format="wav") + audio = audio.set_channels(1).set_frame_rate(16000) + return audio + async def text_to_speak_stream(self, text, queue: queue.Queue, text_index=0): + try: + # Prepare reference data + byte_audios = [audio_to_bytes(ref_audio) for ref_audio in self.reference_audio] + ref_texts = [read_ref_text(ref_text) for ref_text in self.reference_text] + + data = { + "text": text, + "references": [ + ServeReferenceAudio( + audio=audio if audio else b"", text=text + ) + for text, audio in zip(ref_texts, byte_audios) + ], + "reference_id": self.reference_id, + "normalize": self.normalize, + "format": self.format, + "max_new_tokens": self.max_new_tokens, + "chunk_length": self.chunk_length, + "top_p": self.top_p, + "repetition_penalty": self.repetition_penalty, + "temperature": self.temperature, + "streaming": self.streaming, + "use_memory_cache": self.use_memory_cache, + "seed": self.seed, + } + + pydantic_data = ServeTTSRequest(**data) + audio_buff = None + chunk_total = b'' + last_raw = b'' + audio_raw = b'' + print("请求tts") + with requests.post( + self.api_url, + data=ormsgpack.packb(pydantic_data, option=ormsgpack.OPT_SERIALIZE_PYDANTIC), + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/msgpack", + }, + ) as response: + if response.status_code == 200: + for chunk in response.iter_content(): + # 拼接当前块和上一块数据 + chunk_total += chunk + # 最后一个是静音，说明是一个完整的音频 + if len(chunk_total) % 2 == 0 and chunk_total[-2:] == b'\x00\x00': + audio = self._get_audio_from_tts(chunk_total) + audio_raw = audio_raw + audio.raw_data + #长度凑够2贞开始发送，60ms*4=240ms + if len(audio_raw) >= 7680: + duration = 60 * len(audio_raw) // 1920 + if (len(audio_raw) % 1920) > 0: + duration += 60 + duration = duration / 1000.0 + logger.bind(tag=TAG).info(f'发送数据长度：{len(audio_raw)}') + opus_datas = self.wav_to_opus_data_audio_raw(audio_raw) + queue.put({ + "data": opus_datas, + "duration": duration, + "end": False, + "text_index": text_index + }) + audio_raw = b'' + chunk_total = b'' + if len(chunk_total) > 0: + audio = self._get_audio_from_tts(chunk_total) + audio_raw = audio_raw + audio.raw_data + duration = 60 * len(audio_raw) // 1920 + if (len(audio_raw) % 1920) > 0: + duration += 60 + duration = duration / 1000.0 + # 把 audio 转成 opus + logger.bind(tag=TAG).info(f'发送数据长度：{len(audio_raw)}') + opus_datas = self.wav_to_opus_data_audio_raw(audio_raw) + queue.put({ + "data": opus_datas, + "duration": duration, + "end": False + }) + + else: + print('请求失败:', response.status_code, response.text) + queue.put({ + "data": None, + "end": True + }) + except Exception as e: + logger.bind(tag=TAG).error("tts发生错误") + traceback.print_exc() + raise e