diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index efcca44f..c0aa3f93 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -59,6 +59,10 @@ log: delete_audio: true # 没有语音输入多久后断开连接(秒),默认2分钟,即120秒 close_connection_no_voice_time: 120 +# TTS请求超时时间(秒) +tts_timeout: 10 +# 开启唤醒词加速 +enable_wakeup_words_response_cache: true # 开场是否回复唤醒词 enable_greeting: true # 说完话是否开启提示音 diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 8147db1a..bb307b87 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,5 +1,13 @@ +import time import json +import random import asyncio +from core.utils.dialogue import Message +from core.utils.util import audio_to_data_stream +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 remove_punctuation_and_length, opus_datas_to_wav_bytes from core.providers.tools.device_mcp import ( MCPClient, send_mcp_initialize_message, @@ -8,6 +16,18 @@ 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消息""" audio_params = msg_json.get("audio_params") @@ -28,4 +48,100 @@ async def handleHelloMessage(conn, msg_json): # 发送mcp消息,获取tools列表 asyncio.create_task(send_mcp_tools_list_request(conn)) - await conn.websocket.send(json.dumps(conn.welcome_msg)) \ No newline at end of file + 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 2ddf03d6..72424968 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -3,6 +3,7 @@ import uuid import asyncio from core.utils.dialogue import Message from core.providers.tts.dto.dto import ContentType +from core.handle.helloHandle import checkWakeupWords from plugins_func.register import Action, ActionResponse from core.handle.sendAudioHandle import send_stt_message from core.utils.util import remove_punctuation_and_length @@ -27,6 +28,10 @@ async def handle_user_intent(conn, text): 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的聊天方法,不再进行意图分析 return False diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 15a9e434..afe60084 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,5 +1,6 @@ import time import json +import asyncio from core.handle.abortHandle import handleAbortMessage from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit @@ -10,12 +11,16 @@ TAG = __name__ async def handleAudioMessage(conn, audio): - # 检查是否在唤醒处理锁定期内 - if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic(): - return - # 当前片段是否有人说话 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) @@ -24,6 +29,11 @@ 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 diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index c61b0d0d..f482a1ef 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -30,13 +30,14 @@ 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: conn: 连接对象 opus_packet: 单个opus数据包 pre_buffer: 快速发送音频 + frame_duration: 帧时长(毫秒),匹配 Opus 编码 """ if audios is None or len(audios) == 0: return @@ -60,7 +61,6 @@ async def sendAudio(conn, audios, pre_buffer=False): "start_time": time.perf_counter(), } - frame_duration=60 flow_control = conn.audio_flow_control current_time = time.perf_counter() # 计算预期发送时间 @@ -77,6 +77,35 @@ async def sendAudio(conn, audios, pre_buffer=False): # 更新流控状态 flow_control["packet_count"] += 1 flow_control["last_send_time"] = time.perf_counter() + else: + # 流控参数优化 + 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 c1e8f6be..f12b974a 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -3,8 +3,9 @@ import asyncio from core.utils.util import filter_sensitive_info from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage -from core.handle.receiveAudioHandle import handleAudioMessage +from core.handle.reportHandle import enqueue_asr_report 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.providers.tools.device_iot import handleIotDescriptors, handleIotStatus @@ -55,10 +56,16 @@ async def handleTextMessage(conn, message): await send_stt_message(conn, original_text) await send_tts_message(conn, "stop", None) conn.client_is_speaking = False + elif is_wakeup_words and enable_greeting: + conn.just_woken_up = True + # 上报纯文字数据(复用ASR上报功能,但不提供音频数据) + enqueue_asr_report(conn, "嘿,你好呀", []) + await startToChat(conn, "嘿,你好呀") else: - # 检测到唤醒词,开始等待后续进行声纹识别 - conn.wakeup_mode = True - conn.logger.bind(tag=TAG).info(f"检测到唤醒词~") + # 上报纯文字数据(复用ASR上报功能,但不提供音频数据) + enqueue_asr_report(conn, original_text, []) + # 否则需要LLM对文字内容进行答复 + await startToChat(conn, original_text) elif msg_json["type"] == "iot": conn.logger.bind(tag=TAG).info(f"收到iot消息:{message}") if "descriptors" in msg_json: diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 7885261d..250a25f2 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -53,10 +53,6 @@ class ASRProviderBase(ABC): # 接收音频 async def receive_audio(self, conn, audio, audio_have_voice): - # 检查是否在唤醒处理锁定期内 - if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic(): - return - if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice else: @@ -66,14 +62,6 @@ class ASRProviderBase(ABC): if not have_voice and not conn.client_have_voice: conn.asr_audio = conn.asr_audio[-10:] return - - # 检查是否处于唤醒模式 - if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10: - asr_audio_task = conn.asr_audio.copy() - conn.reset_vad_states() - conn.asr_audio.clear() - - await self.handle_voice_stop(conn, asr_audio_task) if conn.client_voice_stop: asr_audio_task = conn.asr_audio.copy() @@ -102,10 +90,27 @@ class ASRProviderBase(ABC): if conn.voiceprint_provider and combined_pcm_data: wav_data = self._pcm_to_wav(combined_pcm_data) - # 检查是否处于唤醒模式 - wakeup_mode = getattr(conn, 'wakeup_mode', False) + # 定义ASR任务 + def run_asr(): + start_time = time.monotonic() + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete( + self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) + ) + end_time = time.monotonic() + logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") + return result + finally: + loop.close() + except Exception as e: + end_time = time.monotonic() + logger.bind(tag=TAG).error(f"ASR失败: {e}") + return ("", None) - # 声纹识别任务(公共逻辑) + # 定义声纹识别任务 def run_voiceprint(): if not wav_data: return None @@ -124,83 +129,48 @@ class ASRProviderBase(ABC): logger.bind(tag=TAG).error(f"声纹识别失败: {e}") return None - if wakeup_mode and conn.voiceprint_provider and wav_data: - conn.wakeup_mode = False - # 设置处理锁,防止后续音频片段重复处理 - conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期 + # 使用线程池执行器并行运行 + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: + asr_future = thread_executor.submit(run_asr) - # 唤醒模式:只执行声纹识别 - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor: + if conn.voiceprint_provider and wav_data: voiceprint_future = thread_executor.submit(run_voiceprint) + + # 等待两个线程都完成 + asr_result = asr_future.result(timeout=15) voiceprint_result = voiceprint_future.result(timeout=15) - - speaker_name = voiceprint_result + + results = {"asr": asr_result, "voiceprint": voiceprint_result} + else: + asr_result = asr_future.result(timeout=15) + results = {"asr": asr_result, "voiceprint": None} + + + # 处理结果 + raw_text, _ = results.get("asr", ("", None)) + speaker_name = results.get("voiceprint", None) + + # 记录识别结果 + if raw_text: + logger.bind(tag=TAG).info(f"识别文本: {raw_text}") + if speaker_name: logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") - - fixed_text = "嘿,你好啊" - enhanced_text = self._build_enhanced_text(fixed_text, speaker_name) + + # 性能监控 + total_time = time.monotonic() - total_start_time + logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") + + # 检查文本长度 + text_len, _ = remove_punctuation_and_length(raw_text) + self.stop_ws_connection() + + if text_len > 0: + # 构建包含说话人信息的JSON字符串 + enhanced_text = self._build_enhanced_text(raw_text, speaker_name) - # 性能监控 - total_time = time.monotonic() - total_start_time - logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s") - + # 使用自定义模块进行上报 await startToChat(conn, enhanced_text) enqueue_asr_report(conn, enhanced_text, asr_audio_task) - else: - # 正常模式:执行声纹识别和ASR - def run_asr(): - start_time = time.monotonic() - try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - result = loop.run_until_complete( - self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) - ) - end_time = time.monotonic() - logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") - return result - finally: - loop.close() - except Exception as e: - end_time = time.monotonic() - logger.bind(tag=TAG).error(f"ASR失败: {e}") - return ("", None) - - # 使用线程池执行器并行运行 - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: - asr_future = thread_executor.submit(run_asr) - - if conn.voiceprint_provider and wav_data: - voiceprint_future = thread_executor.submit(run_voiceprint) - asr_result = asr_future.result(timeout=15) - voiceprint_result = voiceprint_future.result(timeout=15) - results = {"asr": asr_result, "voiceprint": voiceprint_result} - else: - asr_result = asr_future.result(timeout=15) - results = {"asr": asr_result, "voiceprint": None} - - # 处理结果 - raw_text, _ = results.get("asr", ("", None)) - speaker_name = results.get("voiceprint", None) - - if raw_text: - logger.bind(tag=TAG).info(f"识别文本: {raw_text}") - if speaker_name: - logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") - - # 性能监控 - total_time = time.monotonic() - total_start_time - logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") - - # 检查文本长度 - text_len, _ = remove_punctuation_and_length(raw_text) - self.stop_ws_connection() - - if text_len > 0: - enhanced_text = self._build_enhanced_text(raw_text, speaker_name) - await startToChat(conn, enhanced_text) - enqueue_asr_report(conn, enhanced_text, asr_audio_task) except Exception as e: logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 734681e2..38e70927 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -478,3 +478,142 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None + def to_tts(self, text: str) -> list: + """非流式TTS处理,用于测试及保存音频文件的场景""" + try: + # 创建新的事件循环 + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + # 生成会话ID + session_id = uuid.uuid4().hex + # 存储音频数据 + audio_data = [] + + async def _generate_audio(): + # 刷新Token(如果需要) + if self._is_token_expired(): + self._refresh_token() + + # 建立WebSocket连接 + ws = await websockets.connect( + self.ws_url, + additional_headers={"X-NLS-Token": self.token}, + ping_interval=30, + ping_timeout=10, + close_timeout=10, + ) + try: + # 发送StartSynthesis请求 + start_message_id = str(uuid.uuid4().hex) + start_request = { + "header": { + "message_id": start_message_id, + "task_id": session_id, + "namespace": "FlowingSpeechSynthesizer", + "name": "StartSynthesis", + "appkey": self.appkey, + }, + "payload": { + "voice": self.voice, + "format": self.format, + "sample_rate": self.sample_rate, + "volume": self.volume, + "speech_rate": self.speech_rate, + "pitch_rate": self.pitch_rate, + "enable_subtitle": True, + }, + } + await ws.send(json.dumps(start_request)) + + # 等待SynthesisStarted响应 + synthesis_started = False + while not synthesis_started: + msg = await ws.recv() + if isinstance(msg, str): + data = json.loads(msg) + header = data.get("header", {}) + if header.get("name") == "SynthesisStarted": + synthesis_started = True + logger.bind(tag=TAG).debug("TTS合成已启动") + elif header.get("name") == "TaskFailed": + error_info = data.get("payload", {}).get( + "error_info", {} + ) + error_code = error_info.get("error_code") + error_message = error_info.get( + "error_message", "未知错误" + ) + raise Exception( + f"启动合成失败: {error_code} - {error_message}" + ) + + # 发送文本合成请求 + filtered_text = MarkdownCleaner.clean_markdown(text) + run_message_id = str(uuid.uuid4().hex) + run_request = { + "header": { + "message_id": run_message_id, + "task_id": session_id, + "namespace": "FlowingSpeechSynthesizer", + "name": "RunSynthesis", + "appkey": self.appkey, + }, + "payload": {"text": filtered_text}, + } + await ws.send(json.dumps(run_request)) + + # 发送停止合成请求 + stop_message_id = str(uuid.uuid4().hex) + stop_request = { + "header": { + "message_id": stop_message_id, + "task_id": session_id, + "namespace": "FlowingSpeechSynthesizer", + "name": "StopSynthesis", + "appkey": self.appkey, + } + } + await ws.send(json.dumps(stop_request)) + + # 接收音频数据 + synthesis_completed = False + while not synthesis_completed: + msg = await ws.recv() + if isinstance(msg, (bytes, bytearray)): + self.opus_encoder.encode_pcm_to_opus_stream( + msg, + end_of_stream=False, + callback=lambda opus: audio_data.append(opus) + ) + elif isinstance(msg, str): + data = json.loads(msg) + header = data.get("header", {}) + event_name = header.get("name") + if event_name == "SynthesisCompleted": + synthesis_completed = True + logger.bind(tag=TAG).debug("TTS合成完成") + elif event_name == "TaskFailed": + error_info = data.get("payload", {}).get( + "error_info", {} + ) + error_code = error_info.get("error_code") + error_message = error_info.get( + "error_message", "未知错误" + ) + raise Exception( + f"合成失败: {error_code} - {error_message}" + ) + finally: + try: + await ws.close() + except: + pass + + loop.run_until_complete(_generate_audio()) + loop.close() + + return audio_data + except Exception as e: + logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") + return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index a65fe194..04a7fa36 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -1,21 +1,22 @@ import os import re -import queue +import time import uuid +import queue import asyncio import threading -from typing import Callable, Any +import traceback from core.utils import p3 -import time from datetime import datetime from core.utils import textUtils +from typing import Callable, Any from abc import ABC, abstractmethod from config.logger import setup_logging -from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report from core.handle.sendAudioHandle import sendAudioMessage +from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream from core.providers.tts.dto.dto import ( TTSMessageDTO, SentenceType, @@ -23,8 +24,6 @@ from core.providers.tts.dto.dto import ( InterfaceType, ) -import traceback - TAG = __name__ logger = setup_logging() @@ -144,6 +143,68 @@ class TTSProviderBase(ABC): except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") return None + + def to_tts(self, text): + text = MarkdownCleaner.clean_markdown(text) + max_repeat_time = 5 + if self.delete_audio_file: + # 需要删除文件的直接转为音频数据 + while max_repeat_time > 0: + try: + audio_bytes = asyncio.run(self.text_to_speak(text, None)) + if audio_bytes: + audio_datas = [] + audio_bytes_to_data_stream( + audio_bytes, + file_type=self.audio_file_type, + is_opus=True, + callback=lambda data: audio_datas.append(data) + ) + return audio_datas + else: + max_repeat_time -= 1 + except Exception as e: + logger.bind(tag=TAG).warning( + f"语音生成失败{5 - max_repeat_time + 1}次: {text},错误: {e}" + ) + max_repeat_time -= 1 + if max_repeat_time > 0: + logger.bind(tag=TAG).info( + f"语音生成成功: {text},重试{5 - max_repeat_time}次" + ) + else: + logger.bind(tag=TAG).error( + f"语音生成失败: {text},请检查网络或服务是否正常" + ) + return None + else: + tmp_file = self.generate_filename() + try: + while not os.path.exists(tmp_file) and max_repeat_time > 0: + try: + asyncio.run(self.text_to_speak(text, tmp_file)) + except Exception as e: + logger.bind(tag=TAG).warning( + f"语音生成失败{5 - max_repeat_time + 1}次: {text},错误: {e}" + ) + # 未执行成功,删除文件 + if os.path.exists(tmp_file): + os.remove(tmp_file) + max_repeat_time -= 1 + + if max_repeat_time > 0: + logger.bind(tag=TAG).info( + f"语音生成成功: {text}:{tmp_file},重试{5 - max_repeat_time}次" + ) + else: + logger.bind(tag=TAG).error( + f"语音生成失败: {text},请检查网络或服务是否正常" + ) + + return tmp_file + except Exception as e: + logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") + return None @abstractmethod async def text_to_speak(self, text, output_file): 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 463c16fc..ff295581 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -628,3 +628,104 @@ class TTSProvider(TTSProviderBase): def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) + + def to_tts(self, text: str) -> list: + """非流式生成音频数据,用于生成音频及测试场景 + Args: + text: 要转换的文本 + Returns: + list: 音频数据列表 + """ + try: + # 创建事件循环 + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + # 生成会话ID + session_id = uuid.uuid4().__str__().replace("-", "") + + # 存储音频数据 + audio_data = [] + + async def _generate_audio(): + # 创建新的WebSocket连接 + ws_header = { + "X-Api-App-Key": self.appId, + "X-Api-Access-Key": self.access_token, + "X-Api-Resource-Id": self.resource_id, + "X-Api-Connect-Id": uuid.uuid4(), + } + ws = await websockets.connect( + self.ws_url, additional_headers=ws_header, max_size=1000000000 + ) + + try: + # 启动会话 + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_StartSession, sessionId=session_id + ).as_bytes() + payload = self.get_payload_bytes( + event=EVENT_StartSession, speaker=self.voice + ) + await self.send_event(ws, header, optional, payload) + + # 发送文本 + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_TaskRequest, sessionId=session_id + ).as_bytes() + payload = self.get_payload_bytes( + event=EVENT_TaskRequest, text=text, speaker=self.voice + ) + await self.send_event(ws, header, optional, payload) + + # 发送结束会话请求 + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_FinishSession, sessionId=session_id + ).as_bytes() + payload = str.encode("{}") + await self.send_event(ws, header, optional, payload) + + # 接收音频数据 + while True: + msg = await ws.recv() + res = self.parser_response(msg) + + if ( + res.optional.event == EVENT_TTSResponse + and res.header.message_type == AUDIO_ONLY_RESPONSE + ): + self.wav_to_opus_data_audio_raw_stream(res.payload, callback=lambda opus_frame: audio_data.append(opus_frame)) + elif res.optional.event == EVENT_SessionFinished: + break + + finally: + # 清理资源 + try: + await ws.close() + except: + pass + + # 运行异步任务 + loop.run_until_complete(_generate_audio()) + loop.close() + + return audio_data + + except Exception as e: + logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") + return [] diff --git a/main/xiaozhi-server/core/providers/tts/index_stream.py b/main/xiaozhi-server/core/providers/tts/index_stream.py index 6f6b829c..8abb06f2 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/main/xiaozhi-server/core/providers/tts/index_stream.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 @@ -177,3 +179,57 @@ class TTSProvider(TTSProviderBase): await super().close() if hasattr(self, "opus_encoder"): self.opus_encoder.close() + + def to_tts(self, text: str) -> list: + """非流式TTS处理,用于测试及保存音频文件的场景 + Args: + text: 要转换的文本 + Returns: + list: 返回opus编码后的音频数据列表 + """ + start_time = time.time() + text = MarkdownCleaner.clean_markdown(text) + + payload = {"text": text, "character": self.character} + + try: + with requests.post(self.api_url, json=payload, 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/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 04e83148..9b5e946f 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,7 +1,9 @@ import os +import time import queue import aiohttp import asyncio +import requests import traceback from config.logger import setup_logging from core.utils.tts import MarkdownCleaner @@ -195,3 +197,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/providers/tts/paddle_speech.py b/main/xiaozhi-server/core/providers/tts/paddle_speech.py index 4b987b71..864cf505 100644 --- a/main/xiaozhi-server/core/providers/tts/paddle_speech.py +++ b/main/xiaozhi-server/core/providers/tts/paddle_speech.py @@ -1,14 +1,15 @@ -import asyncio -import json -import base64 -import aiohttp -import numpy as np import io import wave +import json +import base64 +import asyncio import websockets -from core.providers.tts.base import TTSProviderBase -from config.logger import setup_logging +import numpy as np from datetime import datetime +from config.logger import setup_logging +from core.providers.tts.base import TTSProviderBase + + TAG = __name__ logger = setup_logging() @@ -74,43 +75,9 @@ class TTSProvider(TTSProviderBase): async def text_to_speak(self, text, output_file): if self.protocol == "websocket": return await self.text_streaming(text, output_file) - elif self.protocol == "http": - return await self.text(text, output_file) else: raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.") - async def text(self, text, output_file): - request_json = { - "text": text, - "spk_id": self.spk_id, - "speed": self.speed, - "volume": self.volume, - "sample_rate": self.sample_rate, - "save_path": self.save_path - } - - try: - async with aiohttp.ClientSession() as session: - async with session.post(self.url, json=request_json) as resp: - if resp.status == 200: - resp_json = await resp.json() - if resp_json.get("success"): - data = resp_json["result"] - audio_bytes = base64.b64decode(data["audio"]) - if output_file: - with open(output_file, "wb") as file_to_save: - file_to_save.write(audio_bytes) - else: - return audio_bytes - else: - raise Exception( - f"Error: {resp_json.get('message', 'Unknown error')} while processing text: {text}") - else: - raise Exception( - f"HTTP Error: {resp.status} - {await resp.text()} while processing text: {text}") - except Exception as e: - raise Exception(f"Error during TTS HTTP request: {e} while processing text: {text}") - async def text_streaming(self, text, output_file): try: # 使用 websockets 异步连接到 WebSocket 服务器 diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index a608049c..49782bfb 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -2,6 +2,7 @@ import re import os import json import copy +import wave import socket import requests import subprocess @@ -292,6 +293,33 @@ def play_audio_frames(conn, file_path): callback=handle_audio_frame ) + +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..d2f4fb32 --- /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 \ No newline at end of file