diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index a75a607f..efcca44f 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -59,6 +59,8 @@ log: delete_audio: true # 没有语音输入多久后断开连接(秒)，默认2分钟，即120秒 close_connection_no_voice_time: 120 +# 开场是否回复唤醒词 +enable_greeting: true # 说完话是否开启提示音 enable_stop_tts_notify: false # 说完话是否开启提示音，音效地址 diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index de9587b5..8147db1a 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -8,7 +8,6 @@ from core.providers.tools.device_mcp import ( TAG = __name__ - async def handleHelloMessage(conn, msg_json): """处理hello消息""" audio_params = msg_json.get("audio_params") @@ -29,4 +28,4 @@ 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)) + await conn.websocket.send(json.dumps(conn.welcome_msg)) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 58b2cee7..2ddf03d6 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -1,11 +1,11 @@ import json -import asyncio 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 from core.handle.sendAudioHandle import send_stt_message 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,7 +23,7 @@ async def handle_user_intent(conn, text): pass # 检查是否有明确的退出命令 - filtered_text = remove_punctuation_and_length(text)[1] + _, filtered_text = remove_punctuation_and_length(text) if await check_direct_exit(conn, filtered_text): return True diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index e5b96be2..15a9e434 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,19 +1,21 @@ import time import json -from core.handle.sendAudioHandle import send_stt_message +from core.handle.abortHandle import handleAbortMessage from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -from core.handle.abortHandle import handleAbortMessage -from core.handle.sendAudioHandle import SentenceType -from core.utils.util import audio_to_data_stream +from core.utils.util import play_audio_frames, play_audio_response +from core.handle.sendAudioHandle import send_stt_message, SentenceType 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) - if have_voice: if conn.client_is_speaking: await handleAbortMessage(conn) @@ -105,9 +107,7 @@ async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) file_path = "config/assets/max_output_size.wav" - conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) - play_audio_frames(conn, file_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) + play_audio_response(conn, {"text": text, "file_path": file_path}) conn.close_after_chat = True @@ -142,18 +142,4 @@ async def check_bind_device(conn): text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) music_path = "config/assets/bind_not_found.wav" - conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) - play_audio_frames(conn, music_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) - - -def play_audio_frames(conn, file_path): - """播放音频文件并处理发送帧数据""" - def handle_audio_frame(frame_data): - conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None)) - - audio_to_data_stream( - file_path, - is_opus=True, - callback=handle_audio_frame - ) + play_audio_response(conn, {"text": text, "file_path": music_path}) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 8604279a..c61b0d0d 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,8 +1,8 @@ import json -import asyncio import time -from core.providers.tts.dto.dto import SentenceType +import asyncio from core.utils import textUtils +from core.providers.tts.dto.dto import SentenceType TAG = __name__ @@ -38,7 +38,7 @@ async def sendAudio(conn, audios, pre_buffer=False): opus_packet: 单个opus数据包 pre_buffer: 快速发送音频 """ - if audios is None: + if audios is None or len(audios) == 0: return if isinstance(audios, bytes): @@ -50,12 +50,8 @@ async def sendAudio(conn, audios, pre_buffer=False): await conn.websocket.send(audios) return - # 重置没有声音的状态 conn.last_activity_time = time.time() * 1000 - # 流控逻辑：确保按60ms的帧时长间隔发送 - frame_duration = 60 # 毫秒 - # 获取或初始化流控状态 if not hasattr(conn, "audio_flow_control"): conn.audio_flow_control = { @@ -64,15 +60,13 @@ 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() - - # 计算期望的发送时间 + # 计算预期发送时间 expected_time = flow_control["start_time"] + ( flow_control["packet_count"] * frame_duration / 1000 ) - - # 流控延迟 delay = expected_time - current_time if delay > 0: await asyncio.sleep(delay) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 86363217..c1e8f6be 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,13 +1,12 @@ import json -import time +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.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__ @@ -45,18 +44,21 @@ async def handleTextMessage(conn, message): conn.client_have_voice = False conn.asr_audio.clear() if "text" in msg_json: - conn.last_activity_time = time.time() * 1000 - original_text = msg_json["text"] # 保留原始文本 - filtered_len, filtered_text = remove_punctuation_and_length( - original_text - ) + original_text = msg_json["text"] # 保留设备上传的文本 # 识别是否是唤醒词 - is_wakeup_words = filtered_text in conn.config.get("wakeup_words") - if not is_wakeup_words: - # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, original_text, []) - # 否则需要LLM对文字内容进行答复 - await startToChat(conn, original_text) + is_wakeup_words = original_text in conn.config.get("wakeup_words") + # 是否开启唤醒词回复 + enable_greeting = conn.config.get("enable_greeting", True) + + if is_wakeup_words and not enable_greeting: + # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 + await send_stt_message(conn, original_text) + await send_tts_message(conn, "stop", None) + conn.client_is_speaking = False + else: + # 检测到唤醒词，开始等待后续进行声纹识别 + conn.wakeup_mode = True + conn.logger.bind(tag=TAG).info(f"检测到唤醒词~") 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 f4803d31..7885261d 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,14 +1,14 @@ 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 @@ -53,6 +53,10 @@ 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: @@ -62,6 +66,14 @@ 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() @@ -87,32 +99,13 @@ 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 @@ -131,48 +124,83 @@ class ASRProviderBase(ABC): logger.bind(tag=TAG).error(f"声纹识别失败: {e}") return None - # 使用线程池执行器并行运行 - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: - asr_future = thread_executor.submit(run_asr) + if wakeup_mode and conn.voiceprint_provider and wav_data: + conn.wakeup_mode = False + # 设置处理锁，防止后续音频片段重复处理 + conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期 - if conn.voiceprint_provider and wav_data: + # 唤醒模式：只执行声纹识别 + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor: 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, 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: + + speaker_name = voiceprint_result 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: - # 构建包含说话人信息的JSON字符串 - enhanced_text = self._build_enhanced_text(raw_text, 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") + 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/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 14301205..a65fe194 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -284,8 +284,8 @@ class TTSProviderBase(ABC): enqueue_audio = [] enqueue_text = text - # 计算音频数据的帧数 - if isinstance(audio_datas, bytes): + # 收集上报音频数据 + if isinstance(audio_datas, bytes) and enqueue_audio is not None: enqueue_audio.append(audio_datas) # 发送音频 diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 875ed0b1..04e83148 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,8 +1,8 @@ import os import queue +import aiohttp import asyncio import traceback -import aiohttp from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -109,10 +109,6 @@ 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) diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 1eec7327..a608049c 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -1,16 +1,17 @@ -import json -import socket -import subprocess import re import os -from io import BytesIO -from typing import Callable, Any -from core.utils import p3 -import numpy as np -import requests -import opuslib_next -from pydub import AudioSegment +import json import copy +import socket +import requests +import subprocess +import numpy as np +import opuslib_next +from io import BytesIO +from core.utils import p3 +from pydub import AudioSegment +from typing import Callable, Any +from core.providers.tts.dto.dto import SentenceType TAG = __name__ emoji_map = { @@ -273,6 +274,23 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = 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)) + + +def play_audio_frames(conn, file_path): + """播放音频文件并处理发送帧数据""" + def handle_audio_frame(frame_data): + conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None)) + + audio_to_data_stream( + file_path, + is_opus=True, + callback=handle_audio_frame + ) def check_vad_update(before_config, new_config): if (