Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

update:合并双流式

Browse Source
This commit is contained in:
hrz
2025-05-26 22:30:45 +08:00
parent 0c8e943d1b
commit b8349af3da
5 changed files with 178 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+15 -4
View File
@@ -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))
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+9 -2
View File
@@ -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
main/xiaozhi-server/core/utils/opus_encoder_utils.py
+136
View File
@@ -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
main/xiaozhi-server/core/utils/util.py
+16 -10
View File
@@ -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名称（支持中英文）
main/xiaozhi-server/core/websocket_server.py
+2 -6
View File
@@ -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: