From d06e297c2d980f10d8e2cb3ed4054d03784ab08e Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Wed, 4 Jun 2025 11:41:04 +0800 Subject: [PATCH] =?UTF-8?q?update:ASR=E5=8A=A0=E5=85=A5=E9=98=9F=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 12 ++- .../xiaozhi-server/core/handle/abortHandle.py | 2 - .../xiaozhi-server/core/handle/helloHandle.py | 2 - .../core/handle/intentHandler.py | 1 + .../core/handle/receiveAudioHandle.py | 6 +- .../core/handle/sendAudioHandle.py | 3 +- .../core/providers/asr/aliyun.py | 4 +- .../core/providers/asr/baidu.py | 4 +- .../xiaozhi-server/core/providers/asr/base.py | 80 ++++++++++++------- .../core/providers/asr/doubao.py | 4 +- .../core/providers/asr/doubao_stream.py | 4 +- .../core/providers/asr/fun_local.py | 4 +- .../core/providers/asr/fun_server.py | 4 +- .../core/providers/asr/sherpa_onnx_local.py | 4 +- .../core/providers/asr/tencent.py | 4 +- .../providers/intent/intent_llm/intent_llm.py | 7 +- main/xiaozhi-server/performance_tester.py | 6 +- 17 files changed, 85 insertions(+), 66 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 238eb562..aa1aaf42 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -35,7 +35,6 @@ from plugins_func.loadplugins import auto_import_modules from plugins_func.register import Action, ActionResponse from core.auth import AuthMiddleware, AuthenticationError from config.config_loader import get_private_config_from_api -from core.handle.receiveAudioHandle import handleAudioMessage from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType from config.logger import setup_logging, build_module_string, update_module_string from config.manage_api_client import DeviceNotFoundException, DeviceBindException @@ -81,6 +80,7 @@ class ConnectionHandler: self.welcome_msg = None self.max_output_size = 0 self.chat_history_conf = 0 + self.audio_format = "opus" # 客户端状态相关 self.client_abort = False @@ -117,7 +117,10 @@ class ConnectionHandler: self.client_voice_stop = False # asr相关变量 + # 因为实际部署时可能会用到公共的本地ASR，不能把变量暴露给公共ASR + # 所以涉及到ASR的变量，需要在这里定义，属于connection的私有变量 self.asr_audio = [] + self.asr_audio_queue = queue.Queue() # llm相关变量 self.llm_finish_task = True @@ -146,7 +149,6 @@ class ConnectionHandler: int(self.config.get("close_connection_no_voice_time", 120)) + 60 ) # 在原来第一道关闭的基础上加60秒，进行二道关闭 - self.audio_format = "opus" # {"mcp":true} 表示启用MCP功能 self.features = None @@ -254,7 +256,11 @@ class ConnectionHandler: if isinstance(message, str): await handleTextMessage(self, message) elif isinstance(message, bytes): - await handleAudioMessage(self, message) + if self.vad is None: + return + if self.asr is None: + return + self.asr_audio_queue.put(message) async def handle_restart(self, message): """处理服务器重启请求""" diff --git a/main/xiaozhi-server/core/handle/abortHandle.py b/main/xiaozhi-server/core/handle/abortHandle.py index fe271511..ce6f22b6 100644 --- a/main/xiaozhi-server/core/handle/abortHandle.py +++ b/main/xiaozhi-server/core/handle/abortHandle.py @@ -1,6 +1,4 @@ import json -import queue -from config.logger import setup_logging TAG = __name__ diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 6b8cb4fd..8434cf22 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -33,8 +33,6 @@ async def handleHelloMessage(conn, msg_json): format = audio_params.get("format") conn.logger.bind(tag=TAG).info(f"客户端音频格式: {format}") conn.audio_format = format - if conn.asr is not None: - conn.asr.set_audio_format(format) conn.welcome_msg["audio_params"] = audio_params features = msg_json.get("features") if features: diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index cb2e3c69..35660bdf 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -96,6 +96,7 @@ async def process_intent_result(conn, intent_result, original_text): } await send_stt_message(conn, original_text) + conn.client_abort = False # 使用executor执行函数调用和结果处理 def process_function_call(): diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index a5947e5a..ddce9d7a 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -10,10 +10,6 @@ TAG = __name__ async def handleAudioMessage(conn, audio): - if conn.vad is None: - return - if conn.asr is None or not hasattr(conn.asr, "conn") or conn.asr.conn is None: - return # 当前片段是否有人说话 have_voice = conn.vad.is_vad(conn, audio) if have_voice: @@ -22,7 +18,7 @@ async def handleAudioMessage(conn, audio): # 设备长时间空闲检测，用于say goodbye await no_voice_close_connect(conn, have_voice) # 接收音频 - await conn.asr.receive_audio(audio, have_voice) + await conn.asr.receive_audio(conn, audio, have_voice) async def startToChat(conn, text): diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 85a71c8e..ea81b7ed 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -89,8 +89,7 @@ async def sendAudio(conn, audios, pre_buffer=True): # 播放剩余音频帧 for opus_packet in remaining_audios: if conn.client_abort: - conn.client_abort = False - return + break # 每分钟重置一次计时器 if time.perf_counter() - last_reset_time > 60: diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index 7ed24c00..36f41745 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -213,7 +213,7 @@ class ASRProvider(ASRProviderBase): return None async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if self._is_token_expired(): @@ -223,7 +223,7 @@ class ASRProvider(ASRProviderBase): file_path = None try: # 解码Opus为PCM - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/baidu.py b/main/xiaozhi-server/core/providers/asr/baidu.py index 7df94ff3..aa8ea901 100644 --- a/main/xiaozhi-server/core/providers/asr/baidu.py +++ b/main/xiaozhi-server/core/providers/asr/baidu.py @@ -30,7 +30,7 @@ class ASRProvider(ASRProviderBase): os.makedirs(self.output_dir, exist_ok=True) async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if not opus_data: @@ -45,7 +45,7 @@ class ASRProvider(ASRProviderBase): return None, file_path # 将Opus音频数据解码为PCM - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 707f80bb..4267c2ff 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,15 +1,19 @@ import os -import time +import wave import copy import uuid -import wave +import queue +import asyncio +import traceback +import threading import opuslib_next from abc import ABC, abstractmethod from config.logger import setup_logging from typing import Optional, Tuple, List -from core.utils.util import remove_punctuation_and_length -from core.handle.reportHandle import enqueue_asr_report from core.handle.receiveAudioHandle import startToChat +from core.handle.reportHandle import enqueue_asr_report +from core.utils.util import remove_punctuation_and_length +from core.handle.receiveAudioHandle import handleAudioMessage TAG = __name__ logger = setup_logging() @@ -17,53 +21,71 @@ logger = setup_logging() class ASRProviderBase(ABC): def __init__(self): - self.audio_format = "opus" - self.conn = None + pass # 打开音频通道 # 这里默认是非流式的处理方式 # 流式处理方式请在子类中重写 async def open_audio_channels(self, conn): - self.conn = conn + # tts 消化线程 + conn.asr_priority_thread = threading.Thread( + target=self.asr_text_priority_thread, args=(conn,), daemon=True + ) + conn.asr_priority_thread.start() + + # 有序处理ASR音频 + def asr_text_priority_thread(self, conn): + while not conn.stop_event.is_set(): + try: + message = conn.asr_audio_queue.get(timeout=1) + future = asyncio.run_coroutine_threadsafe( + handleAudioMessage(conn, message), + conn.loop, + ) + future.result() + except queue.Empty: + continue + except Exception as e: + logger.bind(tag=TAG).error( + f"处理ASR文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) + continue # 接收音频 # 这里默认是非流式的处理方式 # 流式处理方式请在子类中重写 - async def receive_audio(self, audio, audio_have_voice): - if ( - self.conn.client_listen_mode == "auto" - or self.conn.client_listen_mode == "realtime" - ): + async def receive_audio(self, conn, audio, audio_have_voice): + if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice else: - have_voice = self.conn.client_have_voice + have_voice = conn.client_have_voice # 如果本次没有声音，本段也没声音，就把声音丢弃了 - self.conn.asr_audio.append(audio) - if have_voice == False and self.conn.client_have_voice == False: - self.conn.asr_audio = self.conn.asr_audio[-10:] + conn.asr_audio.append(audio) + if have_voice == False and conn.client_have_voice == False: + conn.asr_audio = conn.asr_audio[-10:] return # 如果本段有声音，且已经停止了 - if self.conn.client_voice_stop: - asr_audio_task = copy.deepcopy(self.conn.asr_audio) - self.conn.asr_audio.clear() + if conn.client_voice_stop: + asr_audio_task = copy.deepcopy(conn.asr_audio) + conn.asr_audio.clear() # 音频太短了，无法识别 - self.conn.reset_vad_states() + conn.reset_vad_states() if len(asr_audio_task) > 15: - await self.handle_voice_stop(asr_audio_task) + await self.handle_voice_stop(conn, asr_audio_task) # 处理语音停止 - async def handle_voice_stop(self, asr_audio_task): + async def handle_voice_stop(self, conn, asr_audio_task): raw_text, _ = await self.speech_to_text( - asr_audio_task, self.conn.session_id + asr_audio_task, conn.session_id, conn.audio_format ) # 确保ASR模块返回原始文本 - self.conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}") + conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}") text_len, _ = remove_punctuation_and_length(raw_text) if text_len > 0: # 使用自定义模块进行上报 - await startToChat(self.conn, raw_text) - enqueue_asr_report(self.conn, raw_text, asr_audio_task) + await startToChat(conn, raw_text) + enqueue_asr_report(conn, raw_text, asr_audio_task) def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str: """PCM数据保存为WAV文件""" @@ -81,15 +103,11 @@ class ASRProviderBase(ABC): @abstractmethod async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" pass - def set_audio_format(self, format: str) -> None: - """设置音频格式""" - self.audio_format = format - @staticmethod def decode_opus(opus_data: List[bytes]) -> bytes: """将Opus音频数据解码为PCM数据""" diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index 8a647508..ea590aad 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/main/xiaozhi-server/core/providers/asr/doubao.py @@ -228,14 +228,14 @@ class ASRProvider(ASRProviderBase): yield data[offset:data_len], True async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" file_path = None try: # 合并所有opus数据包 - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/doubao_stream.py b/main/xiaozhi-server/core/providers/asr/doubao_stream.py index da4098ad..da94e2fb 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao_stream.py +++ b/main/xiaozhi-server/core/providers/asr/doubao_stream.py @@ -190,7 +190,7 @@ class ASRProvider(ASRProviderBase): self.is_reconnecting = False self._session_close_event.set() - async def receive_audio(self, audio, _): + async def receive_audio(self, conn, audio, _): if not isinstance(audio, bytes): return @@ -411,7 +411,7 @@ class ASRProvider(ASRProviderBase): await asyncio.sleep(self.retry_delay) await self._forward_asr_results() # 递归重试 - async def speech_to_text(self, opus_data, session_id): + async def speech_to_text(self, opus_data, session_id, audio_format): result = self.text self.text = "" # 清空text return result, None diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index 8c7305c6..89a42b08 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/main/xiaozhi-server/core/providers/asr/fun_local.py @@ -54,7 +54,7 @@ class ASRProvider(ASRProviderBase): ) async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """语音转文本主处理逻辑""" file_path = None @@ -63,7 +63,7 @@ class ASRProvider(ASRProviderBase): while retry_count < MAX_RETRIES: try: # 合并所有opus数据包 - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/fun_server.py b/main/xiaozhi-server/core/providers/asr/fun_server.py index 60d5e425..7eb9e784 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_server.py +++ b/main/xiaozhi-server/core/providers/asr/fun_server.py @@ -100,7 +100,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).debug(f"Sent end message: {end_message}") async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """ Convert speech data to text using FunASR. @@ -109,7 +109,7 @@ class ASRProvider(ASRProviderBase): :return: Tuple containing recognized text and optional timestamp. """ file_path = None - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py index a27640dd..9e5ade80 100644 --- a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py +++ b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py @@ -109,14 +109,14 @@ class ASRProvider(ASRProviderBase): return samples_float32, f.getframerate() async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """语音转文本主处理逻辑""" file_path = None try: # 保存音频文件 start_time = time.time() - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/asr/tencent.py b/main/xiaozhi-server/core/providers/asr/tencent.py index 03301568..1fd1c6f4 100644 --- a/main/xiaozhi-server/core/providers/asr/tencent.py +++ b/main/xiaozhi-server/core/providers/asr/tencent.py @@ -32,7 +32,7 @@ class ASRProvider(ASRProviderBase): os.makedirs(self.output_dir, exist_ok=True) async def speech_to_text( - self, opus_data: List[bytes], session_id: str + self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if not opus_data: @@ -47,7 +47,7 @@ class ASRProvider(ASRProviderBase): return None, file_path # 将Opus音频数据解码为PCM - if self.audio_format == "pcm": + if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) diff --git a/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py b/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py index 51696bb7..29605393 100644 --- a/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py +++ b/main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py @@ -122,6 +122,8 @@ class IntentProvider(IntentProviderBase): async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str: if not self.llm: raise ValueError("LLM provider not set") + if conn.func_handler is None: + return '{"function_call": {"name": "continue_chat"}}' # 记录整体开始时间 total_start_time = time.time() @@ -148,9 +150,8 @@ class IntentProvider(IntentProviderBase): self.clean_cache() if self.promot == "": - if hasattr(conn, "func_handler"): - functions = conn.func_handler.get_functions() - self.promot = self.get_intent_system_prompt(functions) + functions = conn.func_handler.get_functions() + self.promot = self.get_intent_system_prompt(functions) music_config = initialize_music_handler(conn) music_file_names = music_config["music_file_names"] diff --git a/main/xiaozhi-server/performance_tester.py b/main/xiaozhi-server/performance_tester.py index 80503166..a5fc5221 100644 --- a/main/xiaozhi-server/performance_tester.py +++ b/main/xiaozhi-server/performance_tester.py @@ -140,7 +140,9 @@ class AsyncPerformanceTester: print(f"🎵 测试 STT: {stt_name}") - text, _ = await stt.speech_to_text([self.test_wav_list[0]], "1") + text, _ = await stt.speech_to_text( + [self.test_wav_list[0]], "1", stt.audio_format + ) if text is None: print(f"❌ {stt_name} 连接失败") @@ -151,7 +153,7 @@ class AsyncPerformanceTester: for i, sentence in enumerate(self.test_wav_list, 1): start = time.time() - text, _ = await stt.speech_to_text([sentence], "1") + text, _ = await stt.speech_to_text([sentence], "1", stt.audio_format) duration = time.time() - start total_time += duration