update: 恢复唤醒播放机制,待优化和改造(目前只有linkerai拥有正常保存唤醒音频to_tts)

This commit is contained in:
Sakura-RanChen
2025-08-27 10:04:27 +08:00
parent 5a08a41cc8
commit 91cd843cfe
10 changed files with 455 additions and 37 deletions
+4
View File
@@ -61,6 +61,10 @@ delete_audio: true
close_connection_no_voice_time: 120
# TTS请求超时时间(秒)
tts_timeout: 10
# 开启唤醒词加速
enable_wakeup_words_response_cache: true
# 开场是否回复唤醒词
enable_greeting: true
# 说完话是否开启提示音
enable_stop_tts_notify: false
# 说完话是否开启提示音,音效地址
@@ -1,5 +1,16 @@
import time
import json
import random
import asyncio
from core.utils.dialogue import Message
from core.providers.tts.dto.dto import SentenceType
from core.utils.wakeup_word import WakeupWordsConfig
from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message
from core.utils.util import (
audio_to_data_stream,
remove_punctuation_and_length,
opus_datas_to_wav_bytes
)
from core.providers.tools.device_mcp import (
MCPClient,
send_mcp_initialize_message,
@@ -8,6 +19,17 @@ from core.providers.tools.device_mcp import (
TAG = __name__
WAKEUP_CONFIG = {
"refresh_time": 5,
"words": ["你好", "你好啊", "嘿,你好", ""],
}
# 创建全局的唤醒词配置管理器
wakeup_words_config = WakeupWordsConfig()
# 用于防止并发调用wakeupWordsResponse的锁
_wakeup_response_lock = asyncio.Lock()
async def handleHelloMessage(conn, msg_json):
"""处理hello消息"""
@@ -30,3 +52,99 @@ async def handleHelloMessage(conn, msg_json):
asyncio.create_task(send_mcp_tools_list_request(conn))
await conn.websocket.send(json.dumps(conn.welcome_msg))
async def checkWakeupWords(conn, text):
enable_wakeup_words_response_cache = conn.config[
"enable_wakeup_words_response_cache"
]
if not enable_wakeup_words_response_cache or not conn.tts:
return False
_, filtered_text = remove_punctuation_and_length(text)
if filtered_text not in conn.config.get("wakeup_words"):
return False
conn.just_woken_up = True
await send_stt_message(conn, text)
# 获取当前音色
voice = getattr(conn.tts, "voice", "default")
if not voice:
voice = "default"
# 获取唤醒词回复配置
response = wakeup_words_config.get_wakeup_response(voice)
if not response or not response.get("file_path"):
response = {
"voice": "default",
"file_path": "config/assets/wakeup_words.wav",
"time": 0,
"text": "哈啰啊,我是小智啦,声音好听的台湾女孩一枚,超开心认识你耶,最近在忙啥,别忘了给我来点有趣的料哦,我超爱听八卦的啦",
}
# 获取音频数据
opus_packets = []
def handle_audio_frame(frame_data):
opus_packets.append(frame_data)
audio_to_data_stream(response.get("file_path"), is_opus=True, callback=handle_audio_frame)
# 播放唤醒词回复
conn.client_abort = False
conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))
await sendAudioMessage(conn, SentenceType.LAST, [], None)
# 补充对话
conn.dialogue.put(Message(role="assistant", content=response.get("text")))
# 检查是否需要更新唤醒词回复
if time.time() - response.get("time", 0) > WAKEUP_CONFIG["refresh_time"]:
if not _wakeup_response_lock.locked():
asyncio.create_task(wakeupWordsResponse(conn))
return True
async def wakeupWordsResponse(conn):
if not conn.tts or not conn.llm or not conn.llm.response_no_stream:
return
try:
# 尝试获取锁,如果获取不到就返回
if not await _wakeup_response_lock.acquire():
return
# 生成唤醒词回复
wakeup_word = random.choice(WAKEUP_CONFIG["words"])
question = (
"此刻用户正在和你说```"
+ wakeup_word
+ "```。\n请你根据以上用户的内容进行20-30字回复。要符合系统设置的角色情感和态度,不要像机器人一样说话。\n"
+ "请勿对这条内容本身进行任何解释和回应,请勿返回表情符号,仅返回对用户的内容的回复。"
)
result = conn.llm.response_no_stream(conn.config["prompt"], question)
if not result or len(result) == 0:
return
# 生成TTS音频
tts_result = await asyncio.to_thread(conn.tts.to_tts, result)
if not tts_result:
return
# 获取当前音色
voice = getattr(conn.tts, "voice", "default")
wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000)
file_path = wakeup_words_config.generate_file_path(voice)
with open(file_path, "wb") as f:
f.write(wav_bytes)
# 更新配置
wakeup_words_config.update_wakeup_response(voice, file_path, result)
finally:
# 确保在任何情况下都释放锁
if _wakeup_response_lock.locked():
_wakeup_response_lock.release()
@@ -2,6 +2,7 @@ import json
import asyncio
import uuid
from core.handle.sendAudioHandle import send_stt_message
from core.handle.helloHandle import checkWakeupWords
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import ContentType
from core.utils.dialogue import Message
@@ -26,6 +27,10 @@ async def handle_user_intent(conn, text):
filtered_text = remove_punctuation_and_length(text)[1]
if await check_direct_exit(conn, filtered_text):
return True
# 检查是否是唤醒词
if await checkWakeupWords(conn, filtered_text):
return True
if conn.intent_type == "function_call":
# 使用支持function calling的聊天方法,不再进行意图分析
@@ -1,11 +1,11 @@
import time
import json
from core.handle.sendAudioHandle import send_stt_message
import asyncio
from core.utils.util import audio_to_data_stream
from core.handle.abortHandle import handleAbortMessage
from core.handle.intentHandler import handle_user_intent
from core.utils.output_counter import check_device_output_limit
from core.handle.abortHandle import handleAbortMessage
from core.handle.sendAudioHandle import SentenceType
from core.utils.util import audio_to_data_stream
from core.handle.sendAudioHandle import send_stt_message, SentenceType
TAG = __name__
@@ -13,7 +13,14 @@ TAG = __name__
async def handleAudioMessage(conn, audio):
# 当前片段是否有人说话
have_voice = conn.vad.is_vad(conn, audio)
# 如果设备刚刚被唤醒,短暂忽略VAD检测
if have_voice and hasattr(conn, "just_woken_up") and conn.just_woken_up:
have_voice = False
# 设置一个短暂延迟后恢复VAD检测
conn.asr_audio.clear()
if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
return
if have_voice:
if conn.client_is_speaking:
await handleAbortMessage(conn)
@@ -22,6 +29,13 @@ async def handleAudioMessage(conn, audio):
# 接收音频
await conn.asr.receive_audio(conn, audio, have_voice)
async def resume_vad_detection(conn):
# 等待2秒后恢复VAD检测
await asyncio.sleep(1)
conn.just_woken_up = False
async def startToChat(conn, text):
# 检查输入是否是JSON格式(包含说话人信息)
speaker_name = None
@@ -146,7 +160,6 @@ async def check_bind_device(conn):
play_audio_frames(conn, music_path)
conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
def play_audio_frames(conn, file_path):
"""播放音频文件并处理发送帧数据"""
def handle_audio_frame(frame_data):
@@ -1,8 +1,8 @@
import json
import asyncio
import time
from core.providers.tts.dto.dto import SentenceType
import asyncio
from core.utils import textUtils
from core.providers.tts.dto.dto import SentenceType
TAG = __name__
@@ -30,7 +30,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
# 播放音频
async def sendAudio(conn, audios, pre_buffer=False):
async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60):
"""
发送单个opus包,支持流控
Args:
@@ -50,12 +50,8 @@ async def sendAudio(conn, audios, pre_buffer=False):
await conn.websocket.send(audios)
return
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 流控逻辑:确保按60ms的帧时长间隔发送
frame_duration = 60 # 毫秒
# 获取或初始化流控状态
if not hasattr(conn, "audio_flow_control"):
conn.audio_flow_control = {
@@ -66,13 +62,10 @@ async def sendAudio(conn, audios, pre_buffer=False):
flow_control = conn.audio_flow_control
current_time = time.perf_counter()
# 计算期望的发送时间
# 计算预期发送时间
expected_time = flow_control["start_time"] + (
flow_control["packet_count"] * frame_duration / 1000
)
# 流控延迟
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
@@ -83,6 +76,38 @@ async def sendAudio(conn, audios, pre_buffer=False):
# 更新流控状态
flow_control["packet_count"] += 1
flow_control["last_send_time"] = time.perf_counter()
else:
if audios is None or len(audios) == 0:
return
# 流控参数优化
frame_duration = 60 # 帧时长(毫秒),匹配 Opus 编码
start_time = time.perf_counter()
play_position = 0
# 执行预缓冲
pre_buffer_frames = min(3, len(audios))
for i in range(pre_buffer_frames):
await conn.websocket.send(audios[i])
remaining_audios = audios[pre_buffer_frames:]
# 播放剩余音频帧
for opus_packet in remaining_audios:
if conn.client_abort:
break
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 计算预期发送时间
expected_time = start_time + (play_position / 1000)
current_time = time.perf_counter()
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
await conn.websocket.send(opus_packet)
play_position += frame_duration
async def send_tts_message(conn, state, text=None):
+21 -6
View File
@@ -1,13 +1,14 @@
import json
import time
import asyncio
from core.handle.abortHandle import handleAbortMessage
from core.handle.helloHandle import handleHelloMessage
from core.providers.tools.device_mcp import handle_mcp_message
from core.utils.util import remove_punctuation_and_length, filter_sensitive_info
from core.handle.receiveAudioHandle import startToChat, handleAudioMessage
from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus
from core.handle.reportHandle import enqueue_asr_report
import asyncio
from core.providers.tools.device_mcp import handle_mcp_message
from core.handle.receiveAudioHandle import startToChat, handleAudioMessage
from core.handle.sendAudioHandle import send_stt_message, send_tts_message
from core.utils.util import remove_punctuation_and_length, filter_sensitive_info
from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus
TAG = __name__
@@ -50,9 +51,23 @@ async def handleTextMessage(conn, message):
filtered_len, filtered_text = remove_punctuation_and_length(
original_text
)
# 识别是否是唤醒词
is_wakeup_words = filtered_text in conn.config.get("wakeup_words")
if not is_wakeup_words:
# 是否开启唤醒词回复
enable_greeting = conn.config.get("enable_greeting", True)
if is_wakeup_words and not enable_greeting:
# 如果是唤醒词,且关闭了唤醒词回复,就不用回答
await send_stt_message(conn, original_text)
await send_tts_message(conn, "stop", None)
conn.client_is_speaking = False
elif is_wakeup_words:
conn.just_woken_up = True
# 上报纯文字数据(复用ASR上报功能,但不提供音频数据)
enqueue_asr_report(conn, "嘿,你好呀", [])
await startToChat(conn, "嘿,你好呀")
else:
# 上报纯文字数据(复用ASR上报功能,但不提供音频数据)
enqueue_asr_report(conn, original_text, [])
# 否则需要LLM对文字内容进行答复
@@ -286,8 +286,8 @@ class TTSProviderBase(ABC):
enqueue_audio = []
enqueue_text = text
# 计算音频数据的帧数
if isinstance(audio_datas, bytes):
# 收集上报音频数据
if isinstance(audio_datas, bytes) and enqueue_audio is not None:
enqueue_audio.append(audio_datas)
# 发送音频
@@ -1,8 +1,10 @@
import os
import time
import queue
import asyncio
import traceback
import aiohttp
import asyncio
import requests
import traceback
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -199,3 +201,71 @@ class TTSProvider(TTSProviderBase):
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def to_tts(self, text: str) -> list:
"""非流式TTS处理,用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
start_time = time.time()
text = MarkdownCleaner.clean_markdown(text)
params = {
"tts_text": text,
"spk_id": self.voice,
"frame_duration": 60,
"stream": False,
"target_sr": 16000,
"audio_format": self.audio_format,
"instruct_text": "请生成一段自然流畅的语音",
}
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
}
try:
with requests.get(
self.api_url, params=params, headers=headers, timeout=5
) as response:
if response.status_code != 200:
logger.bind(tag=TAG).error(
f"TTS请求失败: {response.status_code}, {response.text}"
)
return []
logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}")
# 使用opus编码器处理PCM数据
opus_datas = []
pcm_data = response.content
# 计算每帧的字节数
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels
* self.opus_encoder.frame_size_ms
/ 1000
* 2
)
# 分帧处理PCM数据
for i in range(0, len(pcm_data), frame_bytes):
frame = pcm_data[i : i + frame_bytes]
if len(frame) < frame_bytes:
# 最后一帧可能不足,用0填充
frame = frame + b"\x00" * (frame_bytes - len(frame))
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=(i + frame_bytes >= len(pcm_data)),
callback=lambda opus: opus_datas.append(opus)
)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
+38 -10
View File
@@ -1,16 +1,17 @@
import json
import socket
import subprocess
import re
import os
from io import BytesIO
from typing import Callable, Any
from core.utils import p3
import numpy as np
import requests
import opuslib_next
from pydub import AudioSegment
import json
import wave
import copy
import socket
import requests
import subprocess
import numpy as np
import opuslib_next
from io import BytesIO
from core.utils import p3
from pydub import AudioSegment
from typing import Callable, Any
TAG = __name__
emoji_map = {
@@ -274,6 +275,33 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] =
callback(frame_data)
def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
"""
将opus帧列表解码为wav字节流
"""
decoder = opuslib_next.Decoder(sample_rate, channels)
pcm_datas = []
frame_duration = 60 # ms
frame_size = int(sample_rate * frame_duration / 1000) # 960
for opus_frame in opus_datas:
# 解码为PCM(返回bytes,2字节/采样点)
pcm = decoder.decode(opus_frame, frame_size)
pcm_datas.append(pcm)
pcm_bytes = b"".join(pcm_datas)
# 写入wav字节流
wav_buffer = BytesIO()
with wave.open(wav_buffer, "wb") as wf:
wf.setnchannels(channels)
wf.setsampwidth(2) # 16bit
wf.setframerate(sample_rate)
wf.writeframes(pcm_bytes)
return wav_buffer.getvalue()
def check_vad_update(before_config, new_config):
if (
new_config.get("selected_module") is None
@@ -0,0 +1,140 @@
import os
import re
import yaml
import time
import hashlib
import portalocker
from typing import Dict
class FileLock:
def __init__(self, file, timeout=5):
self.file = file
self.timeout = timeout
self.start_time = None
def __enter__(self):
self.start_time = time.time()
while True:
try:
portalocker.lock(self.file, portalocker.LOCK_EX | portalocker.LOCK_NB)
return self.file
except portalocker.LockException:
if time.time() - self.start_time > self.timeout:
raise TimeoutError("获取文件锁超时")
time.sleep(0.1)
def __exit__(self, exc_type, exc_val, exc_tb):
portalocker.unlock(self.file)
class WakeupWordsConfig:
def __init__(self):
self.config_file = "data/.wakeup_words.yaml"
self.assets_dir = "config/assets/wakeup_words"
self._ensure_directories()
self._config_cache = None
self._last_load_time = 0
self._cache_ttl = 1 # 缓存有效期(秒)
self._lock_timeout = 5 # 文件锁超时时间(秒)
def _ensure_directories(self):
"""确保必要的目录存在"""
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
os.makedirs(self.assets_dir, exist_ok=True)
def _load_config(self) -> Dict:
"""加载配置文件,使用缓存机制"""
current_time = time.time()
# 如果缓存有效,直接返回缓存
if (
self._config_cache is not None
and current_time - self._last_load_time < self._cache_ttl
):
return self._config_cache
try:
with open(self.config_file, "a+") as f:
with FileLock(f, timeout=self._lock_timeout):
f.seek(0)
content = f.read()
config = yaml.safe_load(content) if content else {}
self._config_cache = config
self._last_load_time = current_time
return config
except (TimeoutError, IOError) as e:
print(f"加载配置文件失败: {e}")
return {}
except Exception as e:
print(f"加载配置文件时发生未知错误: {e}")
return {}
def _save_config(self, config: Dict):
"""保存配置到文件,使用文件锁保护"""
try:
with open(self.config_file, "w") as f:
with FileLock(f, timeout=self._lock_timeout):
yaml.dump(config, f, allow_unicode=True)
self._config_cache = config
self._last_load_time = time.time()
except (TimeoutError, IOError) as e:
print(f"保存配置文件失败: {e}")
raise
except Exception as e:
print(f"保存配置文件时发生未知错误: {e}")
raise
def get_wakeup_response(self, voice: str) -> Dict:
voice = hashlib.md5(voice.encode()).hexdigest()
"""获取唤醒词回复配置"""
config = self._load_config()
if not config or voice not in config:
return None
# 检查文件大小
file_path = config[voice]["file_path"]
if not os.path.exists(file_path) or os.stat(file_path).st_size < (15 * 1024):
return None
return config[voice]
def update_wakeup_response(self, voice: str, file_path: str, text: str):
"""更新唤醒词回复配置"""
try:
# 过滤表情符号
filtered_text = re.sub(r'[\U0001F600-\U0001F64F\U0001F900-\U0001F9FF]', '', text)
config = self._load_config()
voice_hash = hashlib.md5(voice.encode()).hexdigest()
config[voice_hash] = {
"voice": voice,
"file_path": file_path,
"time": time.time(),
"text": filtered_text,
}
self._save_config(config)
except Exception as e:
print(f"更新唤醒词回复配置失败: {e}")
raise
def generate_file_path(self, voice: str) -> str:
"""生成音频文件路径,使用voice的哈希值作为文件名"""
try:
# 生成voice的哈希值
voice_hash = hashlib.md5(voice.encode()).hexdigest()
file_path = os.path.join(self.assets_dir, f"{voice_hash}.wav")
# 如果文件已存在,先删除
if os.path.exists(file_path):
try:
os.remove(file_path)
except Exception as e:
print(f"删除已存在的音频文件失败: {e}")
raise
return file_path
except Exception as e:
print(f"生成音频文件路径失败: {e}")
raise