update:ASR加入队列

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):
         """处理服务器重启请求"""

main/xiaozhi-server/core/handle/abortHandle.py

@@ -1,6 +1,4 @@
 import json
-import queue
-from config.logger import setup_logging
 
 TAG = __name__
 

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:

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():

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):

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
+            break
-            return
 
         # 每分钟重置一次计时器
         if time.perf_counter() - last_reset_time > 60:

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)

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)

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"
+        pass
-        self.conn = None
 
     # 打开音频通道
     # 这里默认是非流式的处理方式
     # 流式处理方式请在子类中重写
     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):
+    async def receive_audio(self, conn, audio, audio_have_voice):
-        if (
+        if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
-            self.conn.client_listen_mode == "auto"
-            or self.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)
+        conn.asr_audio.append(audio)
-        if have_voice == False and self.conn.client_have_voice == False:
+        if have_voice == False and conn.client_have_voice == False:
-            self.conn.asr_audio = self.conn.asr_audio[-10:]
+            conn.asr_audio = conn.asr_audio[-10:]
             return
 
         # 如果本段有声音，且已经停止了
-        if self.conn.client_voice_stop:
+        if conn.client_voice_stop:
-            asr_audio_task = copy.deepcopy(self.conn.asr_audio)
+            asr_audio_task = copy.deepcopy(conn.asr_audio)
-            self.conn.asr_audio.clear()
+            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)
+            await startToChat(conn, raw_text)
-            enqueue_asr_report(self.conn, raw_text, asr_audio_task)
+            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数据"""

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)

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

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)

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)

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)

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)

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()
-                functions = conn.func_handler.get_functions()
+            self.promot = self.get_intent_system_prompt(functions)
-                self.promot = self.get_intent_system_prompt(functions)
 
         music_config = initialize_music_handler(conn)
         music_file_names = music_config["music_file_names"]

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