From eaade698fb2c5c1b3d41bccdf8d2d30e5f6253b0 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Sat, 30 Aug 2025 22:59:21 +0800 Subject: [PATCH 1/2] =?UTF-8?q?update:=E6=B5=81=E5=BC=8F=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E6=9A=82=E4=B8=8D=E7=A8=B3=E5=AE=9A=EF=BC=8C=E5=85=88=E5=9B=9E?= =?UTF-8?q?=E6=BB=9A=E5=88=B0=E6=97=A9=E6=9C=9F=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 4 + .../xiaozhi-server/core/handle/helloHandle.py | 112 +++++++++- .../core/handle/intentHandler.py | 14 +- .../core/handle/receiveAudioHandle.py | 51 +++-- .../core/handle/sendAudioHandle.py | 89 ++++---- main/xiaozhi-server/core/handle/textHandle.py | 29 ++- .../xiaozhi-server/core/providers/asr/base.py | 152 ++++++-------- .../core/providers/tts/aliyun_stream.py | 195 +++++++++++++++++- .../xiaozhi-server/core/providers/tts/base.py | 140 ++++++------- .../providers/tts/huoshan_double_stream.py | 144 ++++++++++++- .../core/providers/tts/index_stream.py | 111 ++++++++-- .../core/providers/tts/linkerai.py | 135 ++++++++++-- .../core/providers/tts/paddle_speech.py | 30 +-- .../core/providers/vad/silero.py | 6 +- .../core/utils/opus_encoder_utils.py | 17 +- main/xiaozhi-server/core/utils/p3.py | 37 +++- main/xiaozhi-server/core/utils/util.py | 78 ++++--- main/xiaozhi-server/core/utils/wakeup_word.py | 140 +++++++++++++ .../plugins_func/functions/play_music.py | 1 + 19 files changed, 1116 insertions(+), 369 deletions(-) create mode 100644 main/xiaozhi-server/core/utils/wakeup_word.py 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..e4e836b4 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,13 +1,33 @@ +import time import json +import random import asyncio +from core.utils.dialogue import Message +from core.utils.util import audio_to_data +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.tts.dto.dto import ContentType, SentenceType from core.providers.tools.device_mcp import ( MCPClient, send_mcp_initialize_message, send_mcp_tools_list_request, ) +from core.utils.wakeup_word import WakeupWordsConfig 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,94 @@ 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": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦", + } + + # 播放唤醒词回复 + conn.client_abort = False + opus_packets, _ = audio_to_data(response.get("file_path")) + + 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() diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 2ddf03d6..81911f55 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -1,11 +1,12 @@ import json -import uuid import asyncio -from core.utils.dialogue import Message -from core.providers.tts.dto.dto import ContentType -from plugins_func.register import Action, ActionResponse +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 +from plugins_func.register import Action, ActionResponse from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType TAG = __name__ @@ -23,9 +24,12 @@ async def handle_user_intent(conn, text): pass # 检查是否有明确的退出命令 - _, filtered_text = remove_punctuation_and_length(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 15a9e434..e6f39632 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,21 +1,28 @@ -import time -import json -from core.handle.abortHandle import handleAbortMessage +from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -from core.utils.util import play_audio_frames, play_audio_response -from core.handle.sendAudioHandle import send_stt_message, SentenceType +from core.handle.abortHandle import handleAbortMessage +import time +import asyncio +import json +from core.handle.sendAudioHandle import SentenceType +from core.utils.util import audio_to_data 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,11 +31,18 @@ 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 actual_text = text - + try: # 尝试解析JSON格式的输入 if text.strip().startswith('{') and text.strip().endswith('}'): @@ -37,13 +51,13 @@ async def startToChat(conn, text): speaker_name = data['speaker'] actual_text = data['content'] conn.logger.bind(tag=TAG).info(f"解析到说话人信息: {speaker_name}") - + # 直接使用JSON格式的文本，不解析 actual_text = text except (json.JSONDecodeError, KeyError): # 如果解析失败，继续使用原始文本 pass - + # 保存说话人信息到连接对象 if speaker_name: conn.current_speaker = speaker_name @@ -107,7 +121,8 @@ async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) file_path = "config/assets/max_output_size.wav" - play_audio_response(conn, {"text": text, "file_path": file_path}) + opus_packets, _ = audio_to_data(file_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) conn.close_after_chat = True @@ -125,15 +140,16 @@ async def check_bind_device(conn): # 播放提示音 music_path = "config/assets/bind_code.wav" - conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) - play_audio_frames(conn, music_path) + opus_packets, _ = audio_to_data(music_path) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 try: digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" - play_audio_frames(conn, num_path) + num_packets, _ = audio_to_data(num_path) + conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None)) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue @@ -142,4 +158,5 @@ async def check_bind_device(conn): text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) music_path = "config/assets/bind_not_found.wav" - play_audio_response(conn, {"text": text, "file_path": music_path}) + opus_packets, _ = audio_to_data(music_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index c61b0d0d..0ca36e1e 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,25 +1,25 @@ import json -import time import asyncio -from core.utils import textUtils +import time from core.providers.tts.dto.dto import SentenceType +from core.utils import textUtils TAG = __name__ async def sendAudioMessage(conn, sentenceType, audios, text): + # 发送句子开始消息 + conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}") + + pre_buffer = False if conn.tts.tts_audio_first_sentence: conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") conn.tts.tts_audio_first_sentence = False - await send_tts_message(conn, "start", None) + pre_buffer = True - if sentenceType == SentenceType.FIRST: - await send_tts_message(conn, "sentence_start", text) + await send_tts_message(conn, "sentence_start", text) - await sendAudio(conn, audios) - # 发送句子开始消息 - if sentenceType is not SentenceType.MIDDLE: - conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}") + await sendAudio(conn, audios, pre_buffer) # 发送结束消息（如果是最后一个文本） if conn.llm_finish_task and sentenceType == SentenceType.LAST: @@ -30,59 +30,45 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 -async def sendAudio(conn, audios, pre_buffer=False): - """ - 发送单个opus包，支持流控 - Args: - conn: 连接对象 - opus_packet: 单个opus数据包 - pre_buffer: 快速发送音频 - """ +async def sendAudio(conn, audios, pre_buffer=True): if audios is None or len(audios) == 0: return + # 流控参数优化 + frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码 + start_time = time.perf_counter() + play_position = 0 - if isinstance(audios, bytes): + # 仅当第一句话时执行预缓冲 + if pre_buffer: + 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:] + else: + remaining_audios = audios + + # 播放剩余音频帧 + for opus_packet in remaining_audios: if conn.client_abort: - return - - # 短音频直接发送（例如：提示音） - if pre_buffer: - await conn.websocket.send(audios) - return + break + # 重置没有声音的状态 conn.last_activity_time = time.time() * 1000 - # 获取或初始化流控状态 - if not hasattr(conn, "audio_flow_control"): - conn.audio_flow_control = { - "last_send_time": 0, - "packet_count": 0, - "start_time": time.perf_counter(), - } - - frame_duration=60 - flow_control = conn.audio_flow_control - current_time = time.perf_counter() # 计算预期发送时间 - expected_time = flow_control["start_time"] + ( - flow_control["packet_count"] * frame_duration / 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(audios) + await conn.websocket.send(opus_packet) - # 更新流控状态 - flow_control["packet_count"] += 1 - flow_control["last_send_time"] = time.perf_counter() + play_position += frame_duration async def send_tts_message(conn, state, text=None): """发送 TTS 状态消息""" - if text is None and state == "sentence_start": - return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: message["text"] = textUtils.check_emoji(text) @@ -95,12 +81,8 @@ async def send_tts_message(conn, state, text=None): stop_tts_notify_voice = conn.config.get( "stop_tts_notify_voice", "config/assets/tts_notify.mp3" ) - conn.tts.audio_to_opus_data_stream( - stop_tts_notify_voice, - callback=lambda audio_data: asyncio.run_coroutine_threadsafe( - sendAudio(conn, audio_data, True), conn.loop - ), - ) + audios, _ = conn.tts.audio_to_opus_data(stop_tts_notify_voice) + await sendAudio(conn, audios) # 清除服务端讲话状态 conn.clearSpeakStatus() @@ -109,17 +91,18 @@ async def send_tts_message(conn, state, text=None): async def send_stt_message(conn, text): - """发送 STT 状态消息""" end_prompt_str = conn.config.get("end_prompt", {}).get("prompt") if end_prompt_str and end_prompt_str == text: await send_tts_message(conn, "start") return + """发送 STT 状态消息""" + # 解析JSON格式，提取实际的用户说话内容 display_text = text try: # 尝试解析JSON格式 - if text.strip().startswith("{") and text.strip().endswith("}"): + if text.strip().startswith('{') and text.strip().endswith('}'): parsed_data = json.loads(text) if isinstance(parsed_data, dict) and "content" in parsed_data: # 如果是包含说话人信息的JSON格式，只显示content部分 diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index c1e8f6be..1bcf4545 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,12 +1,14 @@ import json -import asyncio -from core.utils.util import filter_sensitive_info +import time from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage -from core.handle.receiveAudioHandle import handleAudioMessage 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.handle.sendAudioHandle import send_stt_message, send_tts_message from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus +from core.handle.reportHandle import enqueue_asr_report +import asyncio TAG = __name__ @@ -44,9 +46,14 @@ async def handleTextMessage(conn, message): conn.client_have_voice = False conn.asr_audio.clear() if "text" in msg_json: - original_text = msg_json["text"] # 保留设备上传的文本 + conn.last_activity_time = time.time() * 1000 + original_text = msg_json["text"] # 保留原始文本 + filtered_len, filtered_text = remove_punctuation_and_length( + original_text + ) + # 识别是否是唤醒词 - is_wakeup_words = original_text in conn.config.get("wakeup_words") + is_wakeup_words = filtered_text in conn.config.get("wakeup_words") # 是否开启唤醒词回复 enable_greeting = conn.config.get("enable_greeting", True) @@ -55,10 +62,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: + 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..972818d4 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,18 +1,18 @@ import os -import io import wave import uuid -import json -import time import queue import asyncio import traceback import threading import opuslib_next +import json +import io +import time import concurrent.futures from abc import ABC, abstractmethod from config.logger import setup_logging -from typing import Optional, Tuple, List +from typing import Optional, Tuple, List, Dict, Any from core.handle.receiveAudioHandle import startToChat from core.handle.reportHandle import enqueue_asr_report from core.utils.util import remove_punctuation_and_length @@ -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() @@ -99,13 +87,32 @@ class ASRProviderBase(ABC): # 预先准备WAV数据 wav_data = None + # 使用连接的声纹识别提供者 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 +131,50 @@ 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秒锁定期 + # 使用线程池执行器并行运行 + parallel_start_time = time.monotonic() + + 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, file_path = 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..0bfb367b 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -268,7 +268,11 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) + if message.sentence_type == SentenceType.LAST: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -418,6 +422,9 @@ class TTSProvider(TTSProviderBase): async def _start_monitor_tts_response(self): """监听TTS响应""" + opus_datas_cache = [] + is_first_sentence = True + first_sentence_segment_count = 0 # 添加计数器 try: session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): @@ -438,16 +445,28 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], None) ) + elif event_name == "SentenceBegin": + opus_datas_cache = [] elif event_name == "SentenceEnd": - # 发送缓存的数据 - if self.conn.tts_MessageText: - logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" - ) - self.tts_audio_queue.put( - (SentenceType.FIRST, [], self.conn.tts_MessageText) - ) - self.conn.tts_MessageText = None + if ( + not is_first_sentence + or first_sentence_segment_count > 10 + ): + # 发送缓存的数据 + if self.conn.tts_MessageText: + logger.bind(tag=TAG).info( + f"句子语音生成成功： {self.conn.tts_MessageText}" + ) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText) + ) + self.conn.tts_MessageText = None + else: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, None) + ) + # 第一句话结束后，将标志设置为False + is_first_sentence = False elif event_name == "SynthesisCompleted": logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() @@ -458,7 +477,22 @@ class TTSProvider(TTSProviderBase): # 二进制消息（音频数据） elif isinstance(msg, (bytes, bytearray)): logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") - self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus) + opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False) + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～{len(opus_datas)}" + ) + if is_first_sentence: + first_sentence_segment_count += 1 + if first_sentence_segment_count <= 6: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas, None) + ) + else: + opus_datas_cache.extend(opus_datas) + else: + # 后续句子缓存 + opus_datas_cache.extend(opus_datas) + except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") break @@ -478,3 +512,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)): + # 编码为Opus并收集 + opus_frames = self.opus_encoder.encode_pcm_to_opus( + msg, False + ) + audio_data.extend(opus_frames) + 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 [] diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index a65fe194..40570326 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -4,14 +4,12 @@ import queue import uuid import asyncio import threading -from typing import Callable, Any from core.utils import p3 -import time from datetime import datetime from core.utils import textUtils 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.util import audio_to_data, audio_bytes_to_data from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report @@ -33,6 +31,7 @@ class TTSProviderBase(ABC): def __init__(self, config, delete_audio_file): self.interface_type = InterfaceType.NON_STREAM self.conn = None + self.tts_timeout = 10 self.delete_audio_file = delete_audio_file self.audio_file_type = "wav" self.output_file = config.get("output_dir", "tmp/") @@ -51,9 +50,11 @@ class TTSProviderBase(ABC): "；", ";", "：", + "~", ) self.first_sentence_punctuations = ( "，", + "～", "~", "、", ",", @@ -76,14 +77,7 @@ class TTSProviderBase(ABC): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - def handle_opus(self, opus_data: bytes): - logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}") - self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None)) - - def handle_audio_file(self, file_audio: bytes, text): - self.before_stop_play_files.append((file_audio, text)) - - def to_tts_stream(self, text, opus_handler: Callable[[bytes], None] = None) -> None: + def to_tts(self, text): text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -92,14 +86,10 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - self.tts_audio_queue.put((SentenceType.FIRST, None, text)) - audio_bytes_to_data_stream( - audio_bytes, - file_type=self.audio_file_type, - is_opus=True, - callback=opus_handler, + audio_datas, _ = audio_bytes_to_data( + audio_bytes, file_type=self.audio_file_type, is_opus=True ) - break + return audio_datas else: max_repeat_time -= 1 except Exception as e: @@ -139,8 +129,8 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).error( f"语音生成失败: {text}，请检查网络或服务是否正常" ) - self.tts_audio_queue.put((SentenceType.FIRST, None, text)) - self._process_audio_file_stream(tmp_file, callback=opus_handler) + + return tmp_file except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") return None @@ -149,17 +139,13 @@ class TTSProviderBase(ABC): async def text_to_speak(self, text, output_file): pass - def audio_to_pcm_data_stream( - self, audio_file_path, callback: Callable[[Any], Any] = None - ): + def audio_to_pcm_data(self, audio_file_path): """音频文件转换为PCM编码""" - return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback) + return audio_to_data(audio_file_path, is_opus=False) - def audio_to_opus_data_stream( - self, audio_file_path, callback: Callable[[Any], Any] = None - ): + def audio_to_opus_data(self, audio_file_path): """音频文件转换为Opus编码""" - return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback) + return audio_to_data(audio_file_path, is_opus=True) def tts_one_sentence( self, @@ -191,6 +177,7 @@ class TTSProviderBase(ABC): async def open_audio_channels(self, conn): self.conn = conn + self.tts_timeout = conn.config.get("tts_timeout", 10) # tts 消化线程 self.tts_priority_thread = threading.Thread( target=self.tts_text_priority_thread, daemon=True @@ -225,16 +212,30 @@ class TTSProviderBase(ABC): self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: - self.to_tts_stream(segment_text, opus_handler=self.handle_opus) + if self.delete_audio_file: + audio_datas = self.to_tts(segment_text) + if audio_datas: + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, segment_text) + ) + else: + tts_file = self.to_tts(segment_text) + if tts_file: + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, segment_text) + ) elif ContentType.FILE == message.content_type: - self._process_remaining_text_stream(opus_handler=self.handle_opus) + self._process_remaining_text() tts_file = message.content_file if tts_file and os.path.exists(tts_file): - self._process_audio_file_stream( - tts_file, callback=self.handle_opus + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (message.sentence_type, audio_datas, message.content_detail) ) + if message.sentence_type == SentenceType.LAST: - self._process_remaining_text_stream(opus_handler=self.handle_opus) + self._process_remaining_text() self.tts_audio_queue.put( (message.sentence_type, [], message.content_detail) ) @@ -248,59 +249,29 @@ class TTSProviderBase(ABC): continue def _audio_play_priority_thread(self): - # 需要上报的文本和音频列表 - enqueue_text = None - enqueue_audio = None while not self.conn.stop_event.is_set(): text = None try: try: sentence_type, audio_datas, text = self.tts_audio_queue.get( - timeout=0.1 + timeout=1 ) except queue.Empty: if self.conn.stop_event.is_set(): break continue - - if self.conn.client_abort: - logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据") - enqueue_text, enqueue_audio = None, [] - continue - - # 收到下一个文本开始或会话结束时进行上报 - if sentence_type is not SentenceType.MIDDLE: - # 重置音频流控状态（新句子开始或者结束） - if hasattr(self.conn, 'audio_flow_control'): - self.conn.audio_flow_control = { - 'last_send_time': 0, - 'packet_count': 0, - 'start_time': time.perf_counter() - } - - # 上报TTS数据 - if enqueue_text is not None and enqueue_audio is not None: - enqueue_tts_report(self.conn, enqueue_text, enqueue_audio) - enqueue_audio = [] - enqueue_text = text - - # 收集上报音频数据 - if isinstance(audio_datas, bytes) and enqueue_audio is not None: - enqueue_audio.append(audio_datas) - - # 发送音频 future = asyncio.run_coroutine_threadsafe( sendAudioMessage(self.conn, sentence_type, audio_datas, text), self.conn.loop, ) future.result() - - # 记录输出和报告 if self.conn.max_output_size > 0 and text: add_device_output(self.conn.headers.get("device-id"), len(text)) - + enqueue_tts_report(self.conn, text, audio_datas) except Exception as e: - logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}") + logger.bind(tag=TAG).error( + f"audio_play_priority priority_thread: {text} {e}" + ) async def start_session(self, session_id): pass @@ -352,21 +323,22 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file_stream( - self, tts_file, callback: Callable[[Any], Any] - ) -> None: + def _process_audio_file(self, tts_file): """处理音频文件并转换为指定格式 Args: tts_file: 音频文件路径 - callback: 文件处理函数 + content_detail: 内容详情 + + Returns: + tuple: (sentence_type, audio_datas, content_detail) """ if tts_file.endswith(".p3"): - p3.decode_opus_from_file_stream(tts_file, callback=callback) + audio_datas, _ = p3.decode_opus_from_file(tts_file) elif self.conn.audio_format == "pcm": - self.audio_to_pcm_data_stream(tts_file, callback=callback) + audio_datas, _ = self.audio_to_pcm_data(tts_file) else: - self.audio_to_opus_data_stream(tts_file, callback=callback) + audio_datas, _ = self.audio_to_opus_data(tts_file) if ( self.delete_audio_file @@ -375,6 +347,7 @@ class TTSProviderBase(ABC): and tts_file.startswith(self.output_file) ): os.remove(tts_file) + return audio_datas def _process_before_stop_play_files(self): for audio_datas, text in self.before_stop_play_files: @@ -382,9 +355,7 @@ class TTSProviderBase(ABC): self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def _process_remaining_text_stream( - self, opus_handler: Callable[[bytes], None] = None - ): + def _process_remaining_text(self): """处理剩余的文本并生成语音 Returns: @@ -395,7 +366,18 @@ class TTSProviderBase(ABC): if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: - self.to_tts_stream(segment_text, opus_handler=opus_handler) + if self.delete_audio_file: + audio_datas = self.to_tts(segment_text) + if audio_datas: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, audio_datas, segment_text) + ) + else: + tts_file = self.to_tts(segment_text) + audio_datas = self._process_audio_file(tts_file) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, audio_datas, segment_text) + ) self.processed_chars += len(full_text) return True return False 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..33e2750d 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -4,7 +4,6 @@ import json import queue import asyncio import traceback -from typing import Callable, Any import websockets from core.utils.tts import MarkdownCleaner from config.logger import setup_logging @@ -267,7 +266,11 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) + if message.sentence_type == SentenceType.LAST: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -425,6 +428,9 @@ class TTSProvider(TTSProviderBase): async def _start_monitor_tts_response(self): """监听TTS响应""" + opus_datas_cache = [] + is_first_sentence = True + first_sentence_segment_count = 0 # 添加计数器 try: session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): @@ -445,14 +451,37 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], self.tts_text) ) + opus_datas_cache = [] + first_sentence_segment_count = 0 # 重置计数器 elif ( res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE ): logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") - self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) + opus_datas = self.wav_to_opus_data_audio_raw(res.payload) + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～{len(opus_datas)}" + ) + if is_first_sentence: + first_sentence_segment_count += 1 + if first_sentence_segment_count <= 6: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas, None) + ) + else: + opus_datas_cache.extend(opus_datas) + else: + # 后续句子缓存 + opus_datas_cache.extend(opus_datas) elif res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") + if not is_first_sentence or first_sentence_segment_count > 10: + # 发送缓存的数据 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, None) + ) + # 第一句话结束后，将标志设置为False + is_first_sentence = False elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() @@ -626,5 +655,110 @@ 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 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 + + 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 + ): + opus_datas = self.wav_to_opus_data_audio_raw(res.payload) + audio_data.extend(opus_datas) + 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..5ea9eb36 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/main/xiaozhi-server/core/providers/tts/index_stream.py @@ -3,6 +3,8 @@ import queue import asyncio import traceback import aiohttp +import requests +import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -25,13 +27,15 @@ class TTSProvider(TTSProviderBase): self.api_url = config.get("api_url", "http://8.138.114.124:11996/tts") self.audio_format = "pcm" self.before_stop_play_files = [] + self.segment_count = 0 # 创建Opus编码器 需注意接口返回的采样率为24000 self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( sample_rate=24000, channels=1, frame_size_ms=60 ) - # PCM缓冲区 + # 文本缓冲区和PCM缓冲区 + self.text_buffer = "" self.pcm_buffer = bytearray() def tts_text_priority_thread(self): @@ -44,6 +48,7 @@ class TTSProvider(TTSProviderBase): self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] + self.segment_count = 0 self.before_stop_play_files.clear() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) @@ -57,11 +62,14 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) if message.sentence_type == SentenceType.LAST: # 处理剩余的文本 - self._process_remaining_text_stream(True) + self._process_remaining_text(True) except queue.Empty: continue @@ -70,7 +78,7 @@ class TTSProvider(TTSProviderBase): f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) - def _process_remaining_text_stream(self, is_last=False): + def _process_remaining_text(self, is_last=False): """处理剩余的文本并生成语音 Returns: bool: 是否成功处理了文本 @@ -135,6 +143,8 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() + opus_datas_cache = [] + self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 处理音频流数据 @@ -148,22 +158,41 @@ class TTSProvider(TTSProviderBase): while len(self.pcm_buffer) >= frame_bytes: frame = bytes(self.pcm_buffer[:frame_bytes]) del self.pcm_buffer[:frame_bytes] - - self.opus_encoder.encode_pcm_to_opus_stream( - frame, - end_of_stream=False, - callback=self.handle_opus + opus = self.opus_encoder.encode_pcm_to_opus( + frame, end_of_stream=False ) + if opus: + if self.segment_count < 10: # 前10个片段直接发送 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus, None) + ) + self.segment_count += 1 + else: + opus_datas_cache.extend(opus) # flush 剩余不足一帧的数据 if self.pcm_buffer: - self.opus_encoder.encode_pcm_to_opus_stream( - bytes(self.pcm_buffer), - end_of_stream=True, - callback=self.handle_opus + opus = self.opus_encoder.encode_pcm_to_opus( + bytes(self.pcm_buffer), end_of_stream=True ) + if opus: + if self.segment_count < 10: # 前10个片段直接发送 + # 直接发送 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus, None) + ) + self.segment_count += 1 + else: + # 后续片段缓存 + opus_datas_cache.extend(opus) self.pcm_buffer.clear() + # 如果不是前10个片段，发送缓存的数据 + if self.segment_count >= 10 and opus_datas_cache: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, None) + ) + # 如果是最后一段，输出音频获取完毕 if is_last: self._process_before_stop_play_files() @@ -177,3 +206,59 @@ 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)) + + opus = self.opus_encoder.encode_pcm_to_opus( + frame, end_of_stream=(i + frame_bytes >= len(pcm_data)) + ) + if opus: + opus_datas.extend(opus) + + return opus_datas + + except Exception as e: + logger.bind(tag=TAG).error(f"TTS请求异常: {e}") + return [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 04e83148..d1be586c 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 queue -import aiohttp import asyncio import traceback +import aiohttp +import requests +import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -22,15 +24,23 @@ class TTSProvider(TTSProviderBase): self.api_url = config.get("api_url") self.audio_format = "pcm" self.before_stop_play_files = [] + self.segment_count = 0 # 添加片段计数器 # 创建Opus编码器 self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( sample_rate=16000, channels=1, frame_size_ms=60 ) + # 添加文本缓冲区 + self.text_buffer = "" + # PCM缓冲区 self.pcm_buffer = bytearray() + ################################################################################### + # linkerai单流式TTS重写父类的方法--开始 + ################################################################################### + def tts_text_priority_thread(self): """流式文本处理线程""" while not self.conn.stop_event.is_set(): @@ -41,6 +51,7 @@ class TTSProvider(TTSProviderBase): self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] + self.segment_count = 0 self.before_stop_play_files.clear() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) @@ -54,10 +65,14 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) + file_audio = self._process_audio_file(message.content_file) + self.before_stop_play_files.append( + (file_audio, message.content_detail) + ) + if message.sentence_type == SentenceType.LAST: # 处理剩余的文本 - self._process_remaining_text_stream(True) + self._process_remaining_text(True) except queue.Empty: continue @@ -66,7 +81,7 @@ class TTSProvider(TTSProviderBase): f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) - def _process_remaining_text_stream(self, is_last=False): + def _process_remaining_text(self, is_last=False): """处理剩余的文本并生成语音 Returns: @@ -109,6 +124,10 @@ class TTSProvider(TTSProviderBase): finally: return None + ################################################################################### + # linkerai单流式TTS重写父类的方法--结束 + ################################################################################### + async def text_to_speak(self, text, is_last): """流式处理TTS音频，每句只推送一次音频列表""" await self._tts_request(text, is_last) @@ -157,6 +176,8 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() + opus_datas_cache = [] + self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 兼容 iter_chunked / iter_chunks / iter_any @@ -173,21 +194,41 @@ class TTSProvider(TTSProviderBase): frame = bytes(self.pcm_buffer[:frame_bytes]) del self.pcm_buffer[:frame_bytes] - self.opus_encoder.encode_pcm_to_opus_stream( - frame, - end_of_stream=False, - callback=self.handle_opus + opus = self.opus_encoder.encode_pcm_to_opus( + frame, end_of_stream=False ) + if opus: + if self.segment_count < 10: # 前10个片段直接发送 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus, None) + ) + self.segment_count += 1 + else: + opus_datas_cache.extend(opus) # flush 剩余不足一帧的数据 if self.pcm_buffer: - self.opus_encoder.encode_pcm_to_opus_stream( - bytes(self.pcm_buffer), - end_of_stream=True, - callback=self.handle_opus + opus = self.opus_encoder.encode_pcm_to_opus( + bytes(self.pcm_buffer), end_of_stream=True ) + if opus: + if self.segment_count < 10: # 前10个片段直接发送 + # 直接发送 + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus, None) + ) + self.segment_count += 1 + else: + # 后续片段缓存 + opus_datas_cache.extend(opus) self.pcm_buffer.clear() + # 如果不是前10个片段，发送缓存的数据 + if self.segment_count >= 10 and opus_datas_cache: + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_datas_cache, None) + ) + # 如果是最后一段，输出音频获取完毕 if is_last: self._process_before_stop_play_files() @@ -195,3 +236,73 @@ 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)) + + opus = self.opus_encoder.encode_pcm_to_opus( + frame, end_of_stream=(i + frame_bytes >= len(pcm_data)) + ) + if opus: + opus_datas.extend(opus) + + return opus_datas + + except Exception as e: + logger.bind(tag=TAG).error(f"TTS请求异常: {e}") + return [] diff --git a/main/xiaozhi-server/core/providers/tts/paddle_speech.py b/main/xiaozhi-server/core/providers/tts/paddle_speech.py index 4b987b71..7f1c6a23 100644 --- a/main/xiaozhi-server/core/providers/tts/paddle_speech.py +++ b/main/xiaozhi-server/core/providers/tts/paddle_speech.py @@ -8,7 +8,6 @@ import wave import websockets from core.providers.tts.base import TTSProviderBase from config.logger import setup_logging -from datetime import datetime TAG = __name__ logger = setup_logging() @@ -19,12 +18,11 @@ class TTSProvider(TTSProviderBase): super().__init__(config, delete_audio_file) self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming") self.protocol = config.get("protocol", "websocket") - if config.get("private_voice"): self.spk_id = int(config.get("private_voice")) else: - self.spk_id = int(config.get("spk_id", "0")) - + self.spk_id = int(config.get("spk_id", "0")) + sample_rate = config.get("sample_rate", 24000) self.sample_rate = float(sample_rate) if sample_rate else 24000 @@ -34,21 +32,7 @@ class TTSProvider(TTSProviderBase): volume = config.get("volume", 1.0) self.volume = float(volume) if volume else 1.0 - self.delete_audio_file = config.get("delete_audio", True) - if not self.delete_audio_file: - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - save_path = config.get("save_path") - if save_path: - if not save_path.endswith('.wav'): - save_path = f"{save_path}_{timestamp}.wav" - else: - other_path = save_path[:-4] - save_path = f"{other_path}_{timestamp}.wav" - self.save_path = save_path - else: - self.save_path = f"./streaming_tts_{timestamp}.wav" - else: - self.save_path = None + self.save_path = config.get("save_path", "./streaming_tts.wav") async def pcm_to_wav(self, pcm_data: bytes, sample_rate: int = 24000, num_channels: int = 1, bits_per_sample: int = 16) -> bytes: @@ -167,12 +151,6 @@ class TTSProvider(TTSProviderBase): # 接收结束响应避免服务抛出异常 await ws.recv() - # 根据配置决定是否保存文件 - if not self.delete_audio_file and self.save_path: - with open(self.save_path, "wb") as f: - f.write(wav_data) - logger.bind(tag=TAG).info(f"音频文件已保存到: {self.save_path}") - # 返回或保存音频数据 if output_file: with open(output_file, "wb") as file_to_save: @@ -181,4 +159,4 @@ class TTSProvider(TTSProviderBase): return wav_data except Exception as e: - raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}") \ No newline at end of file + raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}") diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 95ab8ff3..80e65b55 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -33,7 +33,7 @@ class VADProvider(VADProviderBase): int(min_silence_duration_ms) if min_silence_duration_ms else 1000 ) - # 至少要多少帧才算有语音,增加灵敏度 + # 至少要多少帧才算有语音 self.frame_window_threshold = 1 def is_vad(self, conn, opus_packet): @@ -70,7 +70,9 @@ class VADProvider(VADProviderBase): # 更新滑动窗口 conn.client_voice_window.append(is_voice) - client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold) + client_have_voice = ( + conn.client_voice_window.count(True) >= self.frame_window_threshold + ) # 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话 if conn.client_have_voice and not client_have_voice: diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index 8ca406d3..23d26c39 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -5,8 +5,9 @@ Opus编码工具类 import logging import traceback + import numpy as np -from typing import Optional, Callable, Any +from typing import List, Optional from opuslib_next import Encoder from opuslib_next import constants @@ -55,14 +56,13 @@ class OpusEncoderUtils: self.encoder.reset_state() self.buffer = np.array([], dtype=np.int16) - def encode_pcm_to_opus_stream(self, pcm_data: bytes, end_of_stream: bool, callback: Callable[[Any], Any]): + def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: """ - 将PCM数据编码为Opus格式，以流式方式进行处理 + 将PCM数据编码为Opus格式 Args: pcm_data: PCM字节数据 - end_of_stream: 是否为流的结束, - callback: opus处理方法 + end_of_stream: 是否为流的结束 Returns: Opus数据包列表 @@ -76,6 +76,7 @@ class OpusEncoderUtils: # 将新数据追加到缓冲区 self.buffer = np.append(self.buffer, new_samples) + opus_packets = [] offset = 0 # 处理所有完整帧 @@ -83,7 +84,7 @@ class OpusEncoderUtils: frame = self.buffer[offset : offset + self.total_frame_size] output = self._encode(frame) if output: - callback(output) + opus_packets.append(output) offset += self.total_frame_size # 保留未处理的样本 @@ -97,9 +98,11 @@ class OpusEncoderUtils: output = self._encode(last_frame) if output: - callback(output) + opus_packets.append(output) self.buffer = np.array([], dtype=np.int16) + return opus_packets + def _encode(self, frame: np.ndarray) -> Optional[bytes]: """编码一帧音频数据""" try: diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index 415e5366..c75b968e 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -1,12 +1,15 @@ -import io import struct -from typing import Callable, Any +def decode_opus_from_file(input_file): + """ + 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 + """ + opus_datas = [] + total_frames = 0 + sample_rate = 16000 # 文件采样率 + frame_duration_ms = 60 # 帧时长 + frame_size = int(sample_rate * frame_duration_ms / 1000) -def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): - """ - 从p3文件中解码 Opus 数据，由 callback 处理 Opus 数据包。 - """ with open(input_file, 'rb') as f: while True: # 读取头部（4字节）：[1字节类型，1字节保留，2字节长度] @@ -22,13 +25,23 @@ def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): if len(opus_data) != data_len: raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.") - callback(opus_data) + opus_datas.append(opus_data) + total_frames += 1 + # 计算总时长 + total_duration = (total_frames * frame_duration_ms) / 1000.0 + return opus_datas, total_duration -def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]): +def decode_opus_from_bytes(input_bytes): """ - 从p3二进制数据中解码 Opus 数据，由 callback 处理 Opus 数据包。 + 从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 """ + import io + opus_datas = [] + total_frames = 0 + sample_rate = 16000 # 文件采样率 + frame_duration_ms = 60 # 帧时长 + frame_size = int(sample_rate * frame_duration_ms / 1000) f = io.BytesIO(input_bytes) while True: @@ -39,4 +52,8 @@ def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]): opus_data = f.read(data_len) if len(opus_data) != data_len: raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") - callback(opus_data) + opus_datas.append(opus_data) + total_frames += 1 + + total_duration = (total_frames * frame_duration_ms) / 1000.0 + return opus_datas, total_duration \ 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 a608049c..ce6f20ea 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -1,17 +1,16 @@ +import json +import socket +import subprocess import re import os -import json -import copy -import socket -import requests -import subprocess -import numpy as np -import opuslib_next +import wave from io import BytesIO from core.utils import p3 +import numpy as np +import requests +import opuslib_next from pydub import AudioSegment -from typing import Callable, Any -from core.providers.tts.dto.dto import SentenceType +import copy TAG = __name__ emoji_map = { @@ -212,7 +211,7 @@ def extract_json_from_string(input_string): return None -def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any], Any]=None) -> None: +def audio_to_data(audio_file_path, is_opus=True): # 获取文件后缀名 file_type = os.path.splitext(audio_file_path)[1] if file_type: @@ -225,29 +224,33 @@ def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any] # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配） audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2) + # 音频时长(秒) + duration = len(audio) / 1000.0 + # 获取原始PCM数据（16位小端） raw_data = audio.raw_data - pcm_to_data_stream(raw_data, is_opus, callback) + return pcm_to_data(raw_data, is_opus), duration -def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: +def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): """ 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 """ if file_type == "p3": # 直接用p3解码 - return p3.decode_opus_from_bytes_stream(audio_bytes, callback) + return p3.decode_opus_from_bytes(audio_bytes) else: # 其他格式用pydub audio = AudioSegment.from_file( BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"] ) audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2) + duration = len(audio) / 1000.0 raw_data = audio.raw_data - pcm_to_data_stream(raw_data, is_opus, callback) + return pcm_to_data(raw_data, is_opus), duration -def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None): +def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) @@ -255,6 +258,7 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = frame_duration = 60 # 60ms per frame frame_size = int(16000 * frame_duration / 1000) # 960 samples/frame + datas = [] # 按帧处理所有音频数据（包括最后一帧可能补零） for i in range(0, len(raw_data), frame_size * 2): # 16bit=2bytes/sample # 获取当前帧的二进制数据 @@ -269,28 +273,40 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = np_frame = np.frombuffer(chunk, dtype=np.int16) # 编码Opus数据 frame_data = encoder.encode(np_frame.tobytes(), frame_size) - callback(frame_data) else: frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk) - callback(frame_data) -def play_audio_response(conn, response): - """音频响应处理""" - conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], response.get("text"))) - play_audio_frames(conn, response.get("file_path")) - conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) + datas.append(frame_data) + + return datas -def play_audio_frames(conn, file_path): - """播放音频文件并处理发送帧数据""" - def handle_audio_frame(frame_data): - conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None)) +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() - audio_to_data_stream( - file_path, - is_opus=True, - callback=handle_audio_frame - ) def check_vad_update(before_config, new_config): if ( 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..ab58f290 --- /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 diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index dd967d0a..2cbc4018 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/main/xiaozhi-server/plugins_func/functions/play_music.py @@ -212,6 +212,7 @@ async def play_local_music(conn, specific_file=None): conn.logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}") return text = _get_random_play_prompt(selected_music) + await send_stt_message(conn, text) conn.dialogue.put(Message(role="assistant", content=text)) if conn.intent_type == "intent_llm": From 152087443250703dd7d8a594da3357d4c8868200 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Sat, 30 Aug 2025 23:02:50 +0800 Subject: [PATCH 2/2] =?UTF-8?q?update:=E5=8D=87=E7=BA=A7=E7=89=88=E6=9C=AC?= =?UTF-8?q?=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/java/xiaozhi/common/constant/Constant.java | 2 +- main/manager-mobile/src/pages/settings/index.vue | 2 +- main/xiaozhi-server/config/logger.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java index bf747600..ba6567a4 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java +++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java @@ -237,7 +237,7 @@ public interface Constant { /** * 版本号 */ - public static final String VERSION = "0.7.5"; + public static final String VERSION = "0.7.6"; /** * 无效固件URL diff --git a/main/manager-mobile/src/pages/settings/index.vue b/main/manager-mobile/src/pages/settings/index.vue index 41d16c00..0cecd6bf 100644 --- a/main/manager-mobile/src/pages/settings/index.vue +++ b/main/manager-mobile/src/pages/settings/index.vue @@ -222,7 +222,7 @@ function showAbout() { title: `关于${import.meta.env.VITE_APP_TITLE}`, content: `${import.meta.env.VITE_APP_TITLE}\n\n基于 Vue.js 3 + uni-app 构建的跨平台移动端管理应用，为小智ESP32智能硬件提供设备管理、智能体配置等功能。\n\n© 2025 xiaozhi-esp32-server`, title: `关于小智智控台`, - content: `小智智控台\n\n基于 Vue.js 3 + uni-app 构建的跨平台移动端管理应用，为小智ESP32智能硬件提供设备管理、智能体配置等功能。\n\n© 2025 xiaozhi-esp32-server 0.7.5`, + content: `小智智控台\n\n基于 Vue.js 3 + uni-app 构建的跨平台移动端管理应用，为小智ESP32智能硬件提供设备管理、智能体配置等功能。\n\n© 2025 xiaozhi-esp32-server 0.7.6`, showCancel: false, confirmText: '确定', }) diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py index cd4d9965..5a815f06 100644 --- a/main/xiaozhi-server/config/logger.py +++ b/main/xiaozhi-server/config/logger.py @@ -5,7 +5,7 @@ from config.config_loader import load_config from config.settings import check_config_file from datetime import datetime -SERVER_VERSION = "0.7.5" +SERVER_VERSION = "0.7.6" _logger_initialized = False