diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index b3c1b1a8..ab662798 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -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 # 说完话是否开启提示音，音效地址 diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index de9587b5..9374303d 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -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() \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 58b2cee7..66102a97 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -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的聊天方法,不再进行意图分析 diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index e5b96be2..b0166fe3 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -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): diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 8604279a..fb62bd32 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -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): diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 86363217..d5dd1338 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -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对文字内容进行答复 diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 6a29f1f5..3de3ac27 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -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) # 发送音频 diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 875ed0b1..3306942b 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -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 [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 1eec7327..26484b41 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -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 diff --git a/main/xiaozhi-server/core/utils/wakeup_word.py b/main/xiaozhi-server/core/utils/wakeup_word.py new file mode 100644 index 00000000..70274dd4 --- /dev/null +++ b/main/xiaozhi-server/core/utils/wakeup_word.py @@ -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