add: 语音流式输出

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_audio:
-    reference_text: ["你弄来这些吟词宴曲来看，还是这些混话来欺负我。",]
+      - "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_key: "xxx"
-    api_url: "http://127.0.0.1:8080/v1/tts"
+    api_url: "http://192.168.101.242:10000/tts-fishspeech/v1/tts"
   GPT_SOVITS_V2:
     # 定义TTS API类型
     #启动tts方法：

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)
+                    if self.tts_stream:
-                    self.tts_queue.put(future)
+                        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)
+                if self.tts_stream:
-                self.tts_queue.put(future)
+                    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()
+                if self.tts_stream:
-                future = asyncio.run_coroutine_threadsafe(sendAudioMessage(self, opus_datas, text, text_index),
+                    chunk_queque, text, text_index = self.audio_play_queue.get()
-                                                          self.loop)
+                    future = asyncio.run_coroutine_threadsafe(
-                future.result()
+                        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:

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:

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

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