diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index e4510286..93532734 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -23,6 +23,7 @@ from core.utils.util import ( check_vad_update, check_asr_update, filter_sensitive_info, + initialize_tts, ) from concurrent.futures import ThreadPoolExecutor from core.handle.receiveAudioHandle import handleAudioMessage @@ -51,7 +52,6 @@ class ConnectionHandler: _vad, _asr, _llm, - _tts, _memory, _intent, server=None, @@ -96,9 +96,9 @@ class ConnectionHandler: # 依赖的组件 self.vad = None self.asr = None + self.tts = None self._asr = _asr self._vad = _vad - self.tts = _tts self.llm = _llm self.memory = _memory self.intent = _intent @@ -312,7 +312,7 @@ class ConnectionHandler: if self.asr is None: self.asr = self._asr if self.tts is None: - self.tts = DefaultTTS(self.config, delete_audio_file=True) + self.tts = self._initialize_tts() # 使用事件循环运行异步方法 asyncio.run_coroutine_threadsafe(self.tts.open_audio_channels(self), self.loop) @@ -336,6 +336,17 @@ class ConnectionHandler: self.report_thread.start() self.logger.bind(tag=TAG).info("TTS上报线程已启动") + def _initialize_tts(self): + """初始化TTS""" + tts = None + if not self.need_bind: + tts = initialize_tts(self.config) + + if tts is None: + tts = DefaultTTS(self.config, delete_audio_file=True) + + return tts + def _initialize_private_config(self): """如果是从配置文件获取,则进行二次实例化""" if not self.read_config_from_api: @@ -497,8 +508,8 @@ class ConnectionHandler: self.dialogue.update_system_message(self.prompt) def chat(self, query, tool_call=False): + self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") self.llm_finish_task = False - self.logger.bind(tag=TAG).debug(f"Chat: {query}") if not tool_call: self.dialogue.put(Message(role="user", content=query)) diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index 66fe4d7d..c26eec3a 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -4,11 +4,11 @@ import traceback import uuid import json import websockets +from core.utils import opus_encoder_utils import queue from config.logger import setup_logging from core.providers.tts.base import TTSProviderBase from core.providers.tts.dto.dto import SentenceType, ContentType -from core.utils.util import pcm_to_data TAG = __name__ logger = setup_logging() @@ -150,6 +150,9 @@ class TTSProvider(TTSProviderBase): self.tts_text = "" # 合成文字语音后,播放的音频文件列表 self.tts_audio_files = [] + self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( + sample_rate=16000, channels=1, frame_size_ms=60 + ) ################################################################################### # 火山双流式TTS重写父类的方法--开始 @@ -239,7 +242,7 @@ class TTSProvider(TTSProviderBase): and res.header.message_type == AUDIO_ONLY_RESPONSE ): logger.bind(tag=TAG).debug(f"推送数据到队列里面~~") - opus_datas = pcm_to_data(res.payload) + opus_datas = self.wav_to_opus_data_audio_raw(res.payload) logger.bind(tag=TAG).debug( f"推送数据到队列里面帧数~~{len(opus_datas)}" ) @@ -436,3 +439,7 @@ class TTSProvider(TTSProviderBase): """资源清理方法""" await self.finish_connection() await self.ws.close() + + 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 diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py new file mode 100644 index 00000000..23d26c39 --- /dev/null +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -0,0 +1,136 @@ +""" +Opus编码工具类 +将PCM音频数据编码为Opus格式 +""" + +import logging +import traceback + +import numpy as np +from typing import List, Optional +from opuslib_next import Encoder +from opuslib_next import constants + + +class OpusEncoderUtils: + """PCM到Opus的编码器""" + + def __init__(self, sample_rate: int, channels: int, frame_size_ms: int): + """ + 初始化Opus编码器 + + Args: + sample_rate: 采样率 (Hz) + channels: 通道数 (1=单声道, 2=立体声) + frame_size_ms: 帧大小 (毫秒) + """ + self.sample_rate = sample_rate + self.channels = channels + self.frame_size_ms = frame_size_ms + # 计算每帧样本数 = 采样率 * 帧大小(毫秒) / 1000 + self.frame_size = (sample_rate * frame_size_ms) // 1000 + # 总帧大小 = 每帧样本数 * 通道数 + self.total_frame_size = self.frame_size * channels + + # 比特率和复杂度设置 + self.bitrate = 24000 # bps + self.complexity = 10 # 最高质量 + + # 缓冲区初始化为空 + self.buffer = np.array([], dtype=np.int16) + + try: + # 创建Opus编码器 + self.encoder = Encoder( + sample_rate, channels, constants.APPLICATION_AUDIO # 音频优化模式 + ) + self.encoder.bitrate = self.bitrate + self.encoder.complexity = self.complexity + self.encoder.signal = constants.SIGNAL_VOICE # 语音信号优化 + except Exception as e: + logging.error(f"初始化Opus编码器失败: {e}") + raise RuntimeError("初始化失败") from e + + def reset_state(self): + """重置编码器状态""" + self.encoder.reset_state() + self.buffer = np.array([], dtype=np.int16) + + def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: + """ + 将PCM数据编码为Opus格式 + + Args: + pcm_data: PCM字节数据 + end_of_stream: 是否为流的结束 + + Returns: + Opus数据包列表 + """ + # 将字节数据转换为short数组 + new_samples = self._convert_bytes_to_shorts(pcm_data) + + # 校验PCM数据 + self._validate_pcm_data(new_samples) + + # 将新数据追加到缓冲区 + self.buffer = np.append(self.buffer, new_samples) + + opus_packets = [] + offset = 0 + + # 处理所有完整帧 + while offset <= len(self.buffer) - self.total_frame_size: + frame = self.buffer[offset : offset + self.total_frame_size] + output = self._encode(frame) + if output: + opus_packets.append(output) + offset += self.total_frame_size + + # 保留未处理的样本 + self.buffer = self.buffer[offset:] + + # 流结束时处理剩余数据 + if end_of_stream and len(self.buffer) > 0: + # 创建最后一帧并用0填充 + last_frame = np.zeros(self.total_frame_size, dtype=np.int16) + last_frame[: len(self.buffer)] = self.buffer + + output = self._encode(last_frame) + if output: + opus_packets.append(output) + self.buffer = np.array([], dtype=np.int16) + + return opus_packets + + def _encode(self, frame: np.ndarray) -> Optional[bytes]: + """编码一帧音频数据""" + try: + # 将numpy数组转换为bytes + frame_bytes = frame.tobytes() + # opuslib要求输入字节数必须是channels*2的倍数 + encoded = self.encoder.encode(frame_bytes, self.frame_size) + return encoded + except Exception as e: + logging.error(f"Opus编码失败: {e}") + traceback.print_exc() + return None + + def _convert_bytes_to_shorts(self, bytes_data: bytes) -> np.ndarray: + """将字节数组转换为short数组 (16位PCM)""" + # 假设输入是小端字节序的16位PCM + return np.frombuffer(bytes_data, dtype=np.int16) + + def _validate_pcm_data(self, pcm_shorts: np.ndarray) -> None: + """验证PCM数据是否有效""" + # 16位PCM数据范围是 -32768 到 32767 + if np.any((pcm_shorts < -32768) | (pcm_shorts > 32767)): + invalid_samples = pcm_shorts[(pcm_shorts < -32768) | (pcm_shorts > 32767)] + logging.warning(f"发现无效PCM样本: {invalid_samples[:5]}...") + # 在实际应用中可以选择裁剪而不是抛出异常 + # np.clip(pcm_shorts, -32768, 32767, out=pcm_shorts) + + def close(self): + """关闭编码器并释放资源""" + # opuslib没有明确的关闭方法,Python的垃圾回收会处理 + pass diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 4c026b12..8f5ded4b 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -269,16 +269,7 @@ def initialize_modules( # 初始化TTS模块 if init_tts: select_tts_module = config["selected_module"]["TTS"] - tts_type = ( - select_tts_module - if "type" not in config["TTS"][select_tts_module] - else config["TTS"][select_tts_module]["type"] - ) - modules["tts"] = tts.create_instance( - tts_type, - config["TTS"][select_tts_module], - str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), - ) + modules["tts"] = initialize_tts(config) logger.bind(tag=TAG).info(f"初始化组件: tts成功 {select_tts_module}") # 初始化LLM模块 @@ -355,6 +346,21 @@ def initialize_modules( return modules +def initialize_tts(config): + select_tts_module = config["selected_module"]["TTS"] + tts_type = ( + select_tts_module + if "type" not in config["TTS"][select_tts_module] + else config["TTS"][select_tts_module]["type"] + ) + new_tts = tts.create_instance( + tts_type, + config["TTS"][select_tts_module], + str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), + ) + return new_tts + + def analyze_emotion(text): """ 分析文本情感并返回对应的emoji名称(支持中英文) diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 81974b61..e60f0c87 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -19,13 +19,12 @@ class WebSocketServer: "VAD" in self.config["selected_module"], "ASR" in self.config["selected_module"], "LLM" in self.config["selected_module"], - "TTS" in self.config["selected_module"], + False, "Memory" in self.config["selected_module"], "Intent" in self.config["selected_module"], ) self._vad = modules["vad"] if "vad" in modules else None self._asr = modules["asr"] if "asr" in modules else None - self._tts = modules["tts"] if "tts" in modules else None self._llm = modules["llm"] if "llm" in modules else None self._intent = modules["intent"] if "intent" in modules else None self._memory = modules["memory"] if "memory" in modules else None @@ -50,7 +49,6 @@ class WebSocketServer: self._vad, self._asr, self._llm, - self._tts, self._memory, self._intent, self, # 传入server实例 @@ -99,7 +97,7 @@ class WebSocketServer: update_vad, update_asr, "LLM" in new_config["selected_module"], - "TTS" in new_config["selected_module"], + False, "Memory" in new_config["selected_module"], "Intent" in new_config["selected_module"], ) @@ -109,8 +107,6 @@ class WebSocketServer: self._vad = modules["vad"] if "asr" in modules: self._asr = modules["asr"] - if "tts" in modules: - self._tts = modules["tts"] if "llm" in modules: self._llm = modules["llm"] if "intent" in modules: