Merge branch 'xinnan-tech:main' into dev

README.md

@@ -141,11 +141,10 @@
 本项目提供两种部署方式，请根据您的具体需求选择：
 
 #### 🚀 部署方式选择
-
+| 部署方式 | 特点 | 适用场景 | 部署文档 | 配置要求 | 视频教程 | 
-| 部署方式 | 特点 | 适用场景 | Docker部署文档 | 源码部署文档 | 视频教程 | 
 |---------|------|---------|---------|---------|---------|
-| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) | [本地源码只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| - | 
+| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker版](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) / [源码部署](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| 如果使用`FunASR`要2核4G，如果全API，要2核2G | - | 
-| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) | 
+| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker版](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) / [源码部署](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | 如果使用`FunASR`要4核8G，如果全API，要2核4G| [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) | 
 
 
 > 💡 提示：以下是按最新代码部署后的测试平台，有需要可烧录测试，并发为6个，每天会清空数据
@@ -158,9 +157,23 @@ OTA接口地址: https://2662r3426b.vicp.fun/xiaozhi/ota/
 Websocket接口地址: wss://2662r3426b.vicp.fun/xiaozhi/v1/
 ```
 
+#### 🚩 配置说明和推荐
+> [!Note]
+> 本项目默认的配置是`入门全免费`设置，如果想效果更优，推荐使用`全流式配置`。
+> 
+> 本项目自`0.5.2`版本，已支持整个生命周期全流式，相比`0.5`版本以前，响应速度提升约`2.5秒`
+
+
+| 模块名称 | 入门全免费设置 | 全流式配置 |
+|---------|---------|------|
+| ASR(语音识别) | FunASR(本地) | ✅DoubaoASR(火山流式语音识别) |
+| LLM(大模型) | ChatGLMLLM(智谱glm-4-flash) | ✅DoubaoLLM(火山doubao-1-5-pro-32k-250115) |
+| TTS(语音合成) | EdgeTTS(微软语音) | ✅HuoshanDoubleStreamTTS(火山双流式语音合成) |
+| Intent(意图识别) | function_call(函数调用) | ✅function_call(函数调用) |
+| Memory(记忆功能) | mem_local_short(本地短期记忆） | ✅mem_local_short（本地短期记忆） |
+
 ---
 ## 功能清单 ✨
-
 ### 已实现 ✅
 
 | 功能模块 | 描述 |

main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java

@@ -227,7 +227,7 @@ public interface Constant {
     /**
      * 版本号
      */
-    public static final String VERSION = "0.5.1";
+    public static final String VERSION = "0.5.2";
 
     /**
      * 无效固件URL

main/xiaozhi-server/app.py

@@ -1,13 +1,13 @@
-import asyncio
 import sys
 import signal
+import asyncio
+from aioconsole import ainput
 from config.settings import load_config
-from core.websocket_server import WebSocketServer
-from core.ota_server import SimpleOtaServer
-from core.utils.util import check_ffmpeg_installed
 from config.logger import setup_logging
 from core.utils.util import get_local_ip
-from aioconsole import ainput
+from core.ota_server import SimpleOtaServer
+from core.websocket_server import WebSocketServer
+from core.utils.util import check_ffmpeg_installed
 
 TAG = __name__
 logger = setup_logging()

main/xiaozhi-server/config/logger.py

@@ -4,7 +4,7 @@ from loguru import logger
 from config.config_loader import load_config
 from config.settings import check_config_file
 
-SERVER_VERSION = "0.5.1"
+SERVER_VERSION = "0.5.2"
 _logger_initialized = False
 
 
@@ -48,7 +48,11 @@ def setup_logging():
 
     # 第一次初始化时配置日志
     if not _logger_initialized:
-        logger.configure(extra={"selected_module": log_config.get("selected_module", "00000000000000")})  # 新增配置
+        logger.configure(
+            extra={
+                "selected_module": log_config.get("selected_module", "00000000000000")
+            }
+        )  # 新增配置
         log_format = log_config.get(
             "log_format",
             "<green>{time:YYMMDD HH:mm:ss}</green>[{version}_{extra[selected_module]}][<light-blue>{extra[tag]}</light-blue>]-<level>{level}</level>-<light-green>{message}</light-green>",
@@ -62,7 +66,9 @@ def setup_logging():
         log_format = log_format.replace("{version}", SERVER_VERSION)
         log_format = log_format.replace("{selected_module}", selected_module_str)
         log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
-        log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+        log_format_file = log_format_file.replace(
+            "{selected_module}", selected_module_str
+        )
 
         log_level = log_config.get("log_level", "INFO")
         log_dir = log_config.get("log_dir", "tmp")
@@ -116,15 +122,25 @@ def update_module_string(selected_module_str):
         log_format = log_format.replace("{version}", SERVER_VERSION)
         log_format = log_format.replace("{selected_module}", selected_module_str)
         log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
-        log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+        log_format_file = log_format_file.replace(
+            "{selected_module}", selected_module_str
+        )
 
         logger.remove()
-        logger.add(sys.stdout, format=log_format, level=log_config.get("log_level", "INFO"), filter=formatter)
         logger.add(
-            os.path.join(log_config.get("log_dir", "tmp"), log_config.get("log_file", "server.log")),
+            sys.stdout,
+            format=log_format,
+            level=log_config.get("log_level", "INFO"),
+            filter=formatter,
+        )
+        logger.add(
+            os.path.join(
+                log_config.get("log_dir", "tmp"),
+                log_config.get("log_file", "server.log"),
+            ),
             format=log_format_file,
             level=log_config.get("log_level", "INFO"),
-            filter=formatter
+            filter=formatter,
         )
 
     except Exception as e:

main/xiaozhi-server/core/connection.py

@@ -13,18 +13,22 @@ import websockets
 from core.handle.mcpHandle import call_mcp_tool
 from core.utils.util import (
     extract_json_from_string,
-    initialize_modules,
     check_vad_update,
     check_asr_update,
     filter_sensitive_info,
-    initialize_tts,
 )
 from typing import Dict, Any
 from core.mcp.manager import MCPManager
+from core.utils.modules_initialize import (
+    initialize_modules,
+    initialize_tts,
+    initialize_asr,
+)
 from core.handle.reportHandle import report
 from core.providers.tts.default import DefaultTTS
 from concurrent.futures import ThreadPoolExecutor
 from core.utils.dialogue import Message, Dialogue
+from core.providers.asr.dto.dto import InterfaceType
 from core.handle.textHandle import handleTextMessage
 from core.handle.functionHandler import FunctionHandler
 from plugins_func.loadplugins import auto_import_modules
@@ -113,7 +117,6 @@ class ConnectionHandler:
 
         # asr相关变量
         self.asr_audio = []
-        self.asr_server_receive = True
 
         # llm相关变量
         self.llm_finish_task = True
@@ -318,10 +321,14 @@ class ConnectionHandler:
             if self.vad is None:
                 self.vad = self._vad
             if self.asr is None:
-                self.asr = self._asr
+                self.asr = self._initialize_asr()
+            # 打开语音识别通道
+            asyncio.run_coroutine_threadsafe(
+                self.asr.open_audio_channels(self), self.loop
+            )
             if self.tts is None:
                 self.tts = self._initialize_tts()
-            # 使用事件循环运行异步方法
+            # 打开语音合成通道
             asyncio.run_coroutine_threadsafe(
                 self.tts.open_audio_channels(self), self.loop
             )
@@ -359,6 +366,19 @@ class ConnectionHandler:
 
         return tts
 
+    def _initialize_asr(self):
+        """初始化ASR"""
+        if self._asr.interface_type == InterfaceType.LOCAL:
+            # 如果公共ASR是本地服务，则直接返回
+            # 因为本地一个实例ASR，可以被多个连接共享
+            asr = self._asr
+        else:
+            # 如果公共ASR是远程服务，则初始化一个新实例
+            # 因为远程ASR，涉及到websocket连接和接收线程，需要每个连接一个实例
+            asr = initialize_asr(self.config)
+
+        return asr
+
     def _initialize_private_config(self):
         """如果是从配置文件获取，则进行二次实例化"""
         if not self.read_config_from_api:
@@ -606,6 +626,8 @@ class ConnectionHandler:
         text_index = 0
 
         for response in llm_responses:
+            if self.client_abort:
+                break
             if functions is not None:
                 content, tools_call = response
                 if "content" in response:
@@ -631,9 +653,6 @@ class ConnectionHandler:
             if content is not None and len(content) > 0:
                 if not tool_call_flag:
                     response_message.append(content)
-                    if self.client_abort:
-                        break
-
                     if text_index == 0:
                         self.tts.tts_text_queue.put(
                             TTSMessageDTO(
@@ -818,9 +837,7 @@ class ConnectionHandler:
                 item = self.report_queue.get(timeout=1)
                 if item is None:  # 检测毒丸对象
                     break
-
                 type, text, audio_data, report_time = item
-
                 try:
                     # 检查线程池状态
                     if self.executor is None:
@@ -851,7 +868,6 @@ class ConnectionHandler:
 
     def clearSpeakStatus(self):
         self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态")
-        self.asr_server_receive = True
 
     async def close(self, ws=None):
         """资源清理方法"""

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

@@ -1,4 +1,3 @@
-from config.logger import setup_logging
 import json
 import uuid
 from core.handle.sendAudioHandle import send_stt_message
@@ -146,4 +145,3 @@ async def process_intent_result(conn, intent_result, original_text):
 def speak_txt(conn, text):
     conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text)
     conn.dialogue.put(Message(role="assistant", content=text))
- 

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

@@ -1,10 +1,7 @@
-import time
-import copy
-from core.utils.util import remove_punctuation_and_length
 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.handle.reportHandle import enqueue_asr_report
+import time
 from core.handle.sendAudioHandle import SentenceType
 from core.utils.util import audio_to_data
 
@@ -13,46 +10,17 @@ TAG = __name__
 
 async def handleAudioMessage(conn, audio):
     if conn.vad is None:
+        conn.logger.bind(tag=TAG).warning("VAD模块未初始化，继续等待")
         return
-    if not conn.asr_server_receive:
+    if conn.asr is None:
-        conn.logger.bind(tag=TAG).debug(f"前期数据处理中，暂停接收")
+        conn.logger.bind(tag=TAG).warning("ASR模块未初始化，继续等待")
         return
-    if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
+    # 当前片段是否有人说话
-        have_voice = conn.vad.is_vad(conn, audio)
+    have_voice = conn.vad.is_vad(conn, audio)
-    else:
+    # 设备长时间空闲检测，用于say goodbye
-        have_voice = conn.client_have_voice
+    await no_voice_close_connect(conn, have_voice)
-
+    # 接收音频
-    # 如果本次没有声音，本段也没声音，就把声音丢弃了
+    await conn.asr.receive_audio(audio, have_voice)
-    if have_voice == False and conn.client_have_voice == False:
-        await no_voice_close_connect(conn)
-        conn.asr_audio.append(audio)
-        conn.asr_audio = conn.asr_audio[
-            -10:
-        ]  # 保留最新的10帧音频内容，解决ASR句首丢字问题
-        return
-    conn.client_no_voice_last_time = 0.0
-    conn.asr_audio.append(audio)
-    # 如果本段有声音，且已经停止了
-    if conn.client_voice_stop:
-        conn.client_abort = False
-        conn.asr_server_receive = False
-        # 音频太短了，无法识别
-        if len(conn.asr_audio) < 15:
-            conn.asr_server_receive = True
-        else:
-            raw_text, _ = await conn.asr.speech_to_text(
-                conn.asr_audio, conn.session_id
-            )  # 确保ASR模块返回原始文本
-            conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
-            text_len, _ = remove_punctuation_and_length(raw_text)
-            if text_len > 0:
-                # 使用自定义模块进行上报
-                await startToChat(conn, raw_text)
-                enqueue_asr_report(conn, raw_text, copy.deepcopy(conn.asr_audio))
-            else:
-                conn.asr_server_receive = True
-        conn.asr_audio.clear()
-        conn.reset_vad_states()
 
 
 async def startToChat(conn, text):
@@ -73,7 +41,6 @@ async def startToChat(conn, text):
 
     if intent_handled:
         # 如果意图已被处理，不再进行聊天
-        conn.asr_server_receive = True
         return
 
     # 意图未被处理，继续常规聊天流程
@@ -81,7 +48,10 @@ async def startToChat(conn, text):
     conn.executor.submit(conn.chat, text)
 
 
-async def no_voice_close_connect(conn):
+async def no_voice_close_connect(conn, have_voice):
+    if have_voice:
+        conn.client_no_voice_last_time = 0.0
+        return
     if conn.client_no_voice_last_time == 0.0:
         conn.client_no_voice_last_time = time.time() * 1000
     else:
@@ -95,7 +65,6 @@ async def no_voice_close_connect(conn):
         ):
             conn.close_after_chat = True
             conn.client_abort = False
-            conn.asr_server_receive = False
             end_prompt = conn.config.get("end_prompt", {})
             if end_prompt and end_prompt.get("enable", True) is False:
                 conn.logger.bind(tag=TAG).info("结束对话，无需发送结束提示语")

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

@@ -8,6 +8,7 @@ TTS上报功能已集成到ConnectionHandler类中。
 
 具体实现请参考core/connection.py中的相关代码。
 """
+
 import time
 
 import opuslib_next
@@ -145,4 +146,4 @@ def enqueue_asr_report(conn, text, opus_data):
                 f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频"
             )
     except Exception as e:
-        conn.logger.bind(tag=TAG).error(f"加入ASR上报队列失败: {text}, {e}")
+        conn.logger.bind(tag=TAG).debug(f"加入ASR上报队列失败: {text}, {e}")

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

@@ -131,6 +131,11 @@ async def send_tts_message(conn, state, text=None):
 
 
 async def send_stt_message(conn, text):
+    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 状态消息"""
     stt_text = get_string_no_punctuation_or_emoji(text)
     await conn.websocket.send(

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

@@ -42,12 +42,13 @@ async def handleTextMessage(conn, message):
                 if len(conn.asr_audio) > 0:
                     await handleAudioMessage(conn, b"")
             elif msg_json["state"] == "detect":
-                conn.asr_server_receive = False
                 conn.client_have_voice = False
                 conn.asr_audio.clear()
                 if "text" in msg_json:
                     original_text = msg_json["text"]  # 保留原始文本
-                    filtered_len, filtered_text = remove_punctuation_and_length(original_text)
+                    filtered_len, filtered_text = remove_punctuation_and_length(
+                        original_text
+                    )
 
                     # 识别是否是唤醒词
                     is_wakeup_words = filtered_text in conn.config.get("wakeup_words")

main/xiaozhi-server/core/ota_server.py

@@ -3,8 +3,8 @@ import time
 import asyncio
 from aiohttp import web
 from config.logger import setup_logging
-from core.connection import ConnectionHandler
+from core.utils.util import get_local_ip
-from core.utils.util import get_local_ip, initialize_modules
+from core.utils.modules_initialize import initialize_modules
 
 TAG = __name__
 

main/xiaozhi-server/core/providers/asr/aliyun.py

@@ -2,9 +2,6 @@ import http.client
 import json
 import asyncio
 from typing import Optional, Tuple, List
-import opuslib_next
-import wave
-import io
 import os
 import uuid
 import hmac
@@ -16,6 +13,7 @@ import time
 from datetime import datetime
 from config.logger import setup_logging
 from core.providers.asr.base import ASRProviderBase
+from core.providers.asr.dto.dto import InterfaceType
 
 TAG = __name__
 logger = setup_logging()
@@ -92,6 +90,7 @@ class AccessToken:
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
         super().__init__()
+        self.interface_type = InterfaceType.NON_STREAM
         """阿里云ASR初始化"""
         # 新增空值判断逻辑
         self.access_key_id = config.get("access_key_id")
@@ -165,21 +164,6 @@ class ASRProvider(ASRProviderBase):
         request += "&enable_voice_detection=false"
         return request
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)  # 单声道
-            wf.setsampwidth(2)  # 16-bit
-            wf.setframerate(self.sample_rate)
-            wf.writeframes(b"".join(pcm_data))
-
-        logger.bind(tag=TAG).debug(f"音频文件已保存至: {file_path}")
-        return file_path
-
     async def _send_request(self, pcm_data: bytes) -> Optional[str]:
         """发送请求到阿里云ASR服务"""
         try:

main/xiaozhi-server/core/providers/asr/baidu.py

@@ -1,18 +1,10 @@
-import base64
-import hashlib
-import hmac
-import json
 import time
-from datetime import datetime, timezone
 import os
-import uuid
 from typing import Optional, Tuple, List
-import wave
-import opuslib_next
-
 from aip import AipSpeech
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
+from core.providers.asr.dto.dto import InterfaceType
 
 TAG = __name__
 logger = setup_logging()
@@ -21,6 +13,7 @@ logger = setup_logging()
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool = True):
         super().__init__()
+        self.interface_type = InterfaceType.NON_STREAM
         self.app_id = config.get("app_id")
         self.api_key = config.get("api_key")
         self.secret_key = config.get("secret_key")
@@ -36,20 +29,6 @@ class ASRProvider(ASRProviderBase):
         # 确保输出目录存在
         os.makedirs(self.output_dir, exist_ok=True)
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
-            wf.setframerate(16000)
-            wf.writeframes(b"".join(pcm_data))
-
-        return file_path
-
     async def speech_to_text(
         self, opus_data: List[bytes], session_id: str
     ) -> Tuple[Optional[str], Optional[str]]:

main/xiaozhi-server/core/providers/asr/base.py

@@ -1,7 +1,15 @@
-from abc import ABC, abstractmethod
+import os
-from typing import Optional, Tuple, List
+import time
+import copy
+import uuid
+import wave
 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
 
 TAG = __name__
 logger = setup_logging()
@@ -10,11 +18,66 @@ logger = setup_logging()
 class ASRProviderBase(ABC):
     def __init__(self):
         self.audio_format = "opus"
+        self.conn = None
+
+    # 打开音频通道
+    # 这里默认是非流式的处理方式
+    # 流式处理方式请在子类中重写
+    async def open_audio_channels(self, conn):
+        self.conn = conn
+
+    # 接收音频
+    # 这里默认是非流式的处理方式
+    # 流式处理方式请在子类中重写
+    async def receive_audio(self, audio, audio_have_voice):
+        if (
+            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
+        # 如果本次没有声音，本段也没声音，就把声音丢弃了
+        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:]
+            return
+
+        # 如果本段有声音，且已经停止了
+        if self.conn.client_voice_stop:
+            asr_audio_task = copy.deepcopy(self.conn.asr_audio)
+            self.conn.asr_audio.clear()
+            self.conn.client_abort = False
+            # 音频太短了，无法识别
+            self.conn.reset_vad_states()
+            if len(asr_audio_task) > 15:
+                await self.handle_voice_stop(asr_audio_task)
+
+    # 处理语音停止
+    async def handle_voice_stop(self, asr_audio_task):
+        raw_text, _ = await self.speech_to_text(
+            asr_audio_task, self.conn.session_id
+        )  # 确保ASR模块返回原始文本
+        self.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)
 
-    @abstractmethod
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""
-        pass
+        module_name = __name__.split(".")[-1]
+        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
+        file_path = os.path.join(self.output_dir, file_name)
+
+        with wave.open(file_path, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)  # 2 bytes = 16-bit
+            wf.setframerate(16000)
+            wf.writeframes(b"".join(pcm_data))
+
+        return file_path
 
     @abstractmethod
     async def speech_to_text(

main/xiaozhi-server/core/providers/asr/doubao.py

@@ -1,283 +1,429 @@
-import time
-import io
-import wave
-import os
-from typing import Optional, Tuple, List
-import uuid
-import websockets
 import json
 import gzip
-
+import uuid
+import asyncio
+import websockets
 import opuslib_next
 from core.providers.asr.base import ASRProviderBase
-
 from config.logger import setup_logging
+from core.providers.asr.dto.dto import InterfaceType
+import threading
 
 TAG = __name__
 logger = setup_logging()
 
 CLIENT_FULL_REQUEST = 0b0001
 CLIENT_AUDIO_ONLY_REQUEST = 0b0010
-
-NO_SEQUENCE = 0b0000
-NEG_SEQUENCE = 0b0010
-
 SERVER_FULL_RESPONSE = 0b1001
 SERVER_ACK = 0b1011
 SERVER_ERROR_RESPONSE = 0b1111
-
+NO_SEQUENCE = 0b0000
-NO_SERIALIZATION = 0b0000
+NEG_SEQUENCE = 0b0010
-JSON = 0b0001
+JSON_SERIALIZATION = 0b0001
-THRIFT = 0b0011
+GZIP_COMPRESSION = 0b0001
-CUSTOM_TYPE = 0b1111
+PROTOCOL_VERSION = 0b0001
-NO_COMPRESSION = 0b0000
-GZIP = 0b0001
-CUSTOM_COMPRESSION = 0b1111
-
-
-def parse_response(res):
-    """
-    protocol_version(4 bits), header_size(4 bits),
-    message_type(4 bits), message_type_specific_flags(4 bits)
-    serialization_method(4 bits) message_compression(4 bits)
-    reserved （8bits) 保留字段
-    header_extensions 扩展头(大小等于 8 * 4 * (header_size - 1) )
-    payload 类似与http 请求体
-    """
-    protocol_version = res[0] >> 4
-    header_size = res[0] & 0x0F
-    message_type = res[1] >> 4
-    message_type_specific_flags = res[1] & 0x0F
-    serialization_method = res[2] >> 4
-    message_compression = res[2] & 0x0F
-    reserved = res[3]
-    header_extensions = res[4 : header_size * 4]
-    payload = res[header_size * 4 :]
-    result = {}
-    payload_msg = None
-    payload_size = 0
-    if message_type == SERVER_FULL_RESPONSE:
-        payload_size = int.from_bytes(payload[:4], "big", signed=True)
-        payload_msg = payload[4:]
-    elif message_type == SERVER_ACK:
-        seq = int.from_bytes(payload[:4], "big", signed=True)
-        result["seq"] = seq
-        if len(payload) >= 8:
-            payload_size = int.from_bytes(payload[4:8], "big", signed=False)
-            payload_msg = payload[8:]
-    elif message_type == SERVER_ERROR_RESPONSE:
-        code = int.from_bytes(payload[:4], "big", signed=False)
-        result["code"] = code
-        payload_size = int.from_bytes(payload[4:8], "big", signed=False)
-        payload_msg = payload[8:]
-    if payload_msg is None:
-        return result
-    if message_compression == GZIP:
-        payload_msg = gzip.decompress(payload_msg)
-    if serialization_method == JSON:
-        payload_msg = json.loads(str(payload_msg, "utf-8"))
-    elif serialization_method != NO_SERIALIZATION:
-        payload_msg = str(payload_msg, "utf-8")
-    result["payload_msg"] = payload_msg
-    result["payload_size"] = payload_size
-    return result
 
 
 class ASRProvider(ASRProviderBase):
-    def __init__(self, config: dict, delete_audio_file: bool):
+    def __init__(self, config, delete_audio_file):
         super().__init__()
-        self.appid = config.get("appid")
+        self.interface_type = InterfaceType.STREAM
+        self.config = config
+        self.text = ""
+        self.max_retries = 3
+        self.retry_delay = 2  # 重试延迟秒数
+        self.recv_lock = asyncio.Lock()  # 添加接收锁
+
+        self.appid = str(config.get("appid"))
         self.cluster = config.get("cluster")
         self.access_token = config.get("access_token")
         self.boosting_table_name = config.get("boosting_table_name", "")
         self.correct_table_name = config.get("correct_table_name", "")
-        self.output_dir = config.get("output_dir")
+        self.output_dir = config.get("output_dir", "temp/")
         self.delete_audio_file = delete_audio_file
 
-        self.host = "openspeech.bytedance.com"
+        self.ws_url = "wss://openspeech.bytedance.com/api/v2/asr"
-        self.ws_url = f"wss://{self.host}/api/v2/asr"
+        self.uid = config.get("uid", "streaming_asr_service")
-        self.success_code = 1000
+        self.workflow = config.get(
-        self.seg_duration = 15000
+            "workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate"
+        )
+        self.result_type = config.get("result_type", "single")
+        self.format = config.get("format", "raw")
+        self.codec = config.get("codec", "pcm")
+        self.rate = config.get("sample_rate", 16000)
+        self.language = config.get("language", "zh-CN")
+        self.bits = config.get("bits", 16)
+        self.channel = config.get("channel", 1)
+        self.auth_method = config.get("auth_method", "token")
+        self.secret = config.get("secret", "access_secret")
+        self.decoder = opuslib_next.Decoder(16000, 1)
+        self.asr_ws = None
+        self.forward_task = None
+        self.conn = None
+        self.asr_thread = None
 
-        # 确保输出目录存在
+    ###################################################################################
-        os.makedirs(self.output_dir, exist_ok=True)
+    # 豆包流式ASR重写父类的方法--开始
+    ###################################################################################
+    async def open_audio_channels(self, conn):
+        await super().open_audio_channels(conn)
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
+        retry_count = 0
-        """PCM数据保存为WAV文件"""
+        while retry_count < self.max_retries:
-        module_name = __name__.split(".")[-1]
+            try:
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
+                # 确保关闭旧的连接
-        file_path = os.path.join(self.output_dir, file_name)
+                if self.asr_ws is not None:
+                    try:
+                        await self.asr_ws.close()
+                    except Exception as e:
+                        logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}")
+                    finally:
+                        self.asr_ws = None
 
-        with wave.open(file_path, "wb") as wf:
+                headers = self.token_auth() if self.auth_method == "token" else None
-            wf.setnchannels(1)
+                self.asr_ws = await websockets.connect(
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
+                    self.ws_url,
-            wf.setframerate(16000)
+                    additional_headers=headers,
-            wf.writeframes(b"".join(pcm_data))
+                    max_size=1000000000,
+                    ping_interval=None,  # 禁用ping，因为服务器可能不支持
+                    ping_timeout=None,
+                    close_timeout=10,
+                )
 
-        return file_path
+                # 发送初始化请求
+                request_params = self.construct_request(str(uuid.uuid4()))
+                try:
+                    payload_bytes = str.encode(json.dumps(request_params))
+                    payload_bytes = gzip.compress(payload_bytes)
+                    full_client_request = self.generate_header()
+                    full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
+                    full_client_request.extend(payload_bytes)
+                    await self.asr_ws.send(full_client_request)
+                    logger.bind(tag=TAG).debug(f"发送初始化请求: {request_params}")
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}")
+                    raise e
 
-    @staticmethod
+                # 等待初始化响应
-    def _generate_header(
+                try:
-        message_type=CLIENT_FULL_REQUEST, message_type_specific_flags=NO_SEQUENCE
+                    init_res = await self.asr_ws.recv()
-    ) -> bytearray:
+                    result = self.parse_response(init_res)
-        """Generate protocol header."""
+                    logger.bind(tag=TAG).info(f"ASR服务初始化响应: {result}")
-        header = bytearray()
+                except Exception as e:
-        header_size = 1
+                    logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
-        header.append((0b0001 << 4) | header_size)  # Protocol version
+                    raise e
-        header.append((message_type << 4) | message_type_specific_flags)
-        header.append((0b0001 << 4) | 0b0001)  # JSON serialization & GZIP compression
-        header.append(0x00)  # reserved
-        return header
 
-    def _construct_request(self, reqid) -> dict:
+                # 启动接收ASR结果的异步任务
-        """Construct the request payload."""
+                asr_priority = threading.Thread(
-        return {
+                    target=self._start_monitor_asr_response_thread, daemon=True
+                )
+                asr_priority.start()
+                return
+
+            except websockets.exceptions.WebSocketException as e:
+                retry_count += 1
+                if retry_count < self.max_retries:
+                    logger.bind(tag=TAG).warning(
+                        f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}"
+                    )
+                    await asyncio.sleep(self.retry_delay)
+                else:
+                    logger.bind(tag=TAG).error(
+                        f"WebSocket连接失败，已达到最大重试次数: {e}"
+                    )
+                    raise
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}")
+                raise
+
+    async def receive_audio(self, audio, _):
+        if not isinstance(audio, bytes):
+            return
+
+        try:
+            # 解码opus得到PCM数据
+            pcm_frame = self.decoder.decode(audio, 960)
+            payload = gzip.compress(pcm_frame)
+            audio_request = bytearray(self.generate_audio_default_header())
+            audio_request.extend(len(payload).to_bytes(4, "big"))
+            audio_request.extend(payload)
+            if self.asr_ws:
+                await self.asr_ws.send(audio_request)
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"发送音频数据时发生错误: {e}")
+
+    ###################################################################################
+    # 豆包流式ASR重写父类的方法--结束
+    ###################################################################################
+
+    def construct_request(self, reqid):
+        req = {
             "app": {
-                "appid": f"{self.appid}",
+                "appid": self.appid,
                 "cluster": self.cluster,
                 "token": self.access_token,
             },
-            "user": {
+            "user": {"uid": self.uid},
-                "uid": str(uuid.uuid4()),
-            },
             "request": {
                 "reqid": reqid,
-                "show_utterances": False,
+                "workflow": self.workflow,
+                "show_utterances": True,
+                "result_type": self.result_type,
                 "sequence": 1,
                 "boosting_table_name": self.boosting_table_name,
                 "correct_table_name": self.correct_table_name,
             },
             "audio": {
-                "format": "raw",
+                "format": self.format,
-                "rate": 16000,
+                "codec": self.codec,
-                "language": "zh-CN",
+                "rate": self.rate,
-                "bits": 16,
+                "language": self.language,
-                "channel": 1,
+                "bits": self.bits,
-                "codec": "raw",
+                "channel": self.channel,
             },
         }
+        return req
 
-    async def _send_request(
+    def token_auth(self):
-        self, audio_data: List[bytes], segment_size: int
+        return {"Authorization": f"Bearer; {self.access_token}"}
-    ) -> Optional[str]:
+
-        """Send request to Volcano ASR service."""
+    def generate_header(
+        self,
+        version=PROTOCOL_VERSION,
+        message_type=CLIENT_FULL_REQUEST,
+        message_type_specific_flags=NO_SEQUENCE,
+        serial_method=JSON_SERIALIZATION,
+        compression_type=GZIP_COMPRESSION,
+        reserved_data=0x00,
+        extension_header: bytes = b"",
+    ):
+        """
+        生成协议头：
+         - 第1字节：高4位：协议版本，低4位：头部大小（单位 4 字节）
+         - 第2字节：高4位：消息类型，低4位：消息类型特定标志
+         - 第3字节：高4位：序列化方式，低4位：压缩方式
+         - 第4字节：保留字段
+         - 后续：扩展头（如果有）
+        """
+        header = bytearray()
+        header_size = int(len(extension_header) / 4) + 1
+        header.append((version << 4) | header_size)
+        header.append((message_type << 4) | message_type_specific_flags)
+        header.append((serial_method << 4) | compression_type)
+        header.append(reserved_data)
+        header.extend(extension_header)
+        return header
+
+    def generate_full_default_header(self):
+        # full client request 默认头
+        return self.generate_header(
+            version=PROTOCOL_VERSION,
+            message_type=CLIENT_FULL_REQUEST,
+            message_type_specific_flags=NO_SEQUENCE,
+            serial_method=JSON_SERIALIZATION,
+            compression_type=GZIP_COMPRESSION,
+        )
+
+    def generate_audio_default_header(self):
+        # 普通音频片段请求
+        return self.generate_header(
+            version=PROTOCOL_VERSION,
+            message_type=CLIENT_AUDIO_ONLY_REQUEST,
+            message_type_specific_flags=NO_SEQUENCE,
+            serial_method=JSON_SERIALIZATION,
+            compression_type=GZIP_COMPRESSION,
+        )
+
+    def generate_last_audio_default_header(self):
+        # 最后一个音频片段标志
+        return self.generate_header(
+            version=PROTOCOL_VERSION,
+            message_type=CLIENT_AUDIO_ONLY_REQUEST,
+            message_type_specific_flags=NEG_SEQUENCE,  # 用 NEG_SEQUENCE 表示结束
+            serial_method=JSON_SERIALIZATION,
+            compression_type=GZIP_COMPRESSION,
+        )
+
+    def _start_monitor_asr_response_thread(self):
+        # 初始化链接
+        asyncio.run_coroutine_threadsafe(
+            self._forward_asr_results(), loop=self.conn.loop
+        )
+
+    async def _forward_asr_results(self):
         try:
-            auth_header = {"Authorization": "Bearer; {}".format(self.access_token)}
+            while not self.conn.stop_event.is_set():
-            async with websockets.connect(
+                try:
-                self.ws_url, additional_headers=auth_header
+                    if self.asr_ws is None:
-            ) as websocket:
+                        logger.bind(tag=TAG).info("尝试重新连接ASR服务...")
-                # Prepare request data
+                        await self.open_audio_channels(self.conn)
-                request_params = self._construct_request(str(uuid.uuid4()))
+                        continue
-                payload_bytes = str.encode(json.dumps(request_params))
-                payload_bytes = gzip.compress(payload_bytes)
-                full_client_request = self._generate_header()
-                full_client_request.extend(
-                    (len(payload_bytes)).to_bytes(4, "big")
-                )  # payload size(4 bytes)
-                full_client_request.extend(payload_bytes)  # payload
 
-                # Send header and metadata
+                    # 使用锁来确保同一时间只有一个协程在接收数据
-                # full_client_request
+                    async with self.recv_lock:
-                await websocket.send(full_client_request)
+                        response = await self.asr_ws.recv()
-                res = await websocket.recv()
+                        result = self.parse_response(response)
-                result = parse_response(res)
+                    # 检查是否需要重连
-                if (
+                    if result.get("need_reconnect", False):
-                    "payload_msg" in result
+                        logger.bind(tag=TAG).info(
-                    and result["payload_msg"]["code"] != self.success_code
+                            "检测到需要重连的错误，准备重新连接..."
-                ):
-                    logger.bind(tag=TAG).error(f"ASR error: {result}")
-                    return None
-
-                for seq, (chunk, last) in enumerate(
-                    self.slice_data(audio_data, segment_size), 1
-                ):
-                    if last:
-                        audio_only_request = self._generate_header(
-                            message_type=CLIENT_AUDIO_ONLY_REQUEST,
-                            message_type_specific_flags=NEG_SEQUENCE,
                         )
-                    else:
+                        if self.asr_ws is not None:
-                        audio_only_request = self._generate_header(
+                            try:
-                            message_type=CLIENT_AUDIO_ONLY_REQUEST
+                                await self.asr_ws.close()
-                        )
+                            except Exception as e:
-                    payload_bytes = gzip.compress(chunk)
+                                logger.bind(tag=TAG).warning(
-                    audio_only_request.extend(
+                                    f"关闭旧连接时发生错误: {e}"
-                        (len(payload_bytes)).to_bytes(4, "big")
+                                )
-                    )  # payload size(4 bytes)
+                            finally:
-                    audio_only_request.extend(payload_bytes)  # payload
+                                self.asr_ws = None
-                    # Send audio data
+                        continue
-                    await websocket.send(audio_only_request)
 
-                # Receive response
+                    if "payload_msg" in result:
-                response = await websocket.recv()
+                        if "result" in result["payload_msg"]:
-                result = parse_response(response)
+                            # 检查是否有utterances并且definite为True
-
+                            utterances = result["payload_msg"]["result"][0].get(
-                if (
+                                "utterances", []
-                    "payload_msg" in result
+                            )
-                    and result["payload_msg"]["code"] == self.success_code
+                            for utterance in utterances:
-                ):
+                                if utterance.get("definite", False):
-                    if len(result["payload_msg"]["result"]) > 0:
+                                    self.text = utterance["text"]
-                        return result["payload_msg"]["result"][0]["text"]
+                                    await self.handle_voice_stop(None)
-                    return None
+                                    break
-                else:
+                except websockets.ConnectionClosed:
-                    logger.bind(tag=TAG).error(f"ASR error: {result}")
+                    logger.bind(tag=TAG).debug("ASR服务连接已关闭，准备重连...")
-                    return None
+                    # 确保关闭旧连接
+                    if self.asr_ws is not None:
+                        try:
+                            await self.asr_ws.close()
+                        except Exception as e:
+                            logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}")
+                        finally:
+                            self.asr_ws = None
 
+                    retry_count = 0
+                    while (
+                        retry_count < self.max_retries
+                        and not self.conn.stop_event.is_set()
+                    ):
+                        try:
+                            logger.bind(tag=TAG).info(
+                                f"正在进行第{retry_count + 1}次重连尝试..."
+                            )
+                            await self.open_audio_channels(self.conn)
+                            break
+                        except Exception as e:
+                            retry_count += 1
+                            if retry_count < self.max_retries:
+                                logger.bind(tag=TAG).warning(
+                                    f"重连失败，等待{self.retry_delay}秒后重试: {e}"
+                                )
+                                await asyncio.sleep(self.retry_delay)
+                            else:
+                                logger.bind(tag=TAG).error(
+                                    f"重连失败，已达到最大重试次数: {e}"
+                                )
+                                await asyncio.sleep(
+                                    self.retry_delay
+                                )  # 继续等待，以便后续重试
+                except Exception as e:
+                    if not self.conn.stop_event.is_set():
+                        await asyncio.sleep(2)  # 增加重试延迟
         except Exception as e:
-            logger.bind(tag=TAG).error(f"ASR request failed: {e}", exc_info=True)
+            logger.bind(tag=TAG).error(f"ASR监听线程发生错误: {e}")
-            return None
+            # 确保在发生严重错误时也能继续尝试重连
+            if not self.conn.stop_event.is_set():
+                await asyncio.sleep(self.retry_delay)
+                await self._forward_asr_results()  # 递归重试
 
-    @staticmethod
+    async def speech_to_text(self, opus_data, session_id):
-    def slice_data(data: bytes, chunk_size: int) -> (list, bool):
+        result = self.text
+        self.text = ""  # 清空text
+        return result, None
+
+    def parse_response(self, res: bytes) -> dict:
         """
-        slice data
+        解析 ASR 服务返回的二进制响应。
-        :param data: wav data
+        根据协议格式解析头部和 payload，若采用 GZIP 压缩则先解压，再根据 JSON 反序列化。
-        :param chunk_size: the segment size in one request
-        :return: segment data, last flag
         """
-        data_len = len(data)
+        protocol_version = res[0] >> 4
-        offset = 0
+        header_size = res[0] & 0x0F
-        while offset + chunk_size < data_len:
+        message_type = res[1] >> 4
-            yield data[offset : offset + chunk_size], False
+        serialization_method = res[2] >> 4
-            offset += chunk_size
+        message_compression = res[2] & 0x0F
+        payload = res[header_size * 4 :]
+        result = {}
+        payload_msg = None
+        payload_size = 0
+
+        if message_type == SERVER_FULL_RESPONSE:
+            payload_size = int.from_bytes(payload[:4], "big", signed=True)
+            payload_msg = payload[4:]
+        elif message_type == SERVER_ACK:
+            seq = int.from_bytes(payload[:4], "big", signed=True)
+            result["seq"] = seq
+            if len(payload) >= 8:
+                payload_size = int.from_bytes(payload[4:8], "big", signed=False)
+                payload_msg = payload[8:]
+        elif message_type == SERVER_ERROR_RESPONSE:
+            code = int.from_bytes(payload[:4], "big", signed=False)
+            result["code"] = code
+            payload_size = int.from_bytes(payload[4:8], "big", signed=False)
+            payload_msg = payload[8:]
+
+        if payload_msg is None:
+            return result
+        if message_compression == GZIP_COMPRESSION:
+            payload_msg = gzip.decompress(payload_msg)
+        if serialization_method == JSON_SERIALIZATION:
+            payload_msg = json.loads(payload_msg.decode("utf-8"))
         else:
-            yield data[offset:data_len], True
+            payload_msg = payload_msg.decode("utf-8")
+        result["payload_msg"] = payload_msg
+        result["payload_size"] = payload_size
 
-    async def speech_to_text(
+        # 错误码处理
-        self, opus_data: List[bytes], session_id: str
+        if "code" in result:
-    ) -> Tuple[Optional[str], Optional[str]]:
+            error_code = result["code"]
-        """将语音数据转换为文本"""
+            error_message = ""
 
-        file_path = None
+            if error_code == 1000:
-        try:
+                error_message = "成功"
-            # 合并所有opus数据包
+            elif error_code == 1001:
-            if self.audio_format == "pcm":
+                error_message = "请求参数无效：请求参数缺失必需字段/字段值无效/重复请求"
-                pcm_data = opus_data
+            elif error_code == 1002:
+                error_message = "无访问权限：token无效/过期/无权访问指定服务"
+            elif error_code == 1003:
+                error_message = "访问超频：当前appid访问QPS超出设定阈值"
+            elif error_code == 1004:
+                error_message = "访问超额：当前appid访问次数超出限制"
+            elif error_code == 1005:
+                error_message = "服务器繁忙：服务过载，无法处理当前请求"
+            elif error_code == 1010:
+                error_message = "音频过长：音频数据时长超出阈值"
+            elif error_code == 1011:
+                error_message = "音频过大：音频数据大小超出阈值"
+            elif error_code == 1012:
+                error_message = "音频格式无效：音频header有误/无法进行音频解码"
+            elif error_code == 1013:
+                error_message = "音频静音：音频未识别出任何文本结果"
+            elif error_code >= 1020 and error_code <= 1022:
+                error_message = "识别相关错误：需要重连"
+                if error_code == 1020:
+                    error_message = "识别等待超时：等待下一包就绪超时"
+                elif error_code == 1021:
+                    error_message = "识别处理超时：识别处理过程超时"
+                elif error_code == 1022:
+                    error_message = "识别错误：识别过程中发生错误"
             else:
-                pcm_data = self.decode_opus(opus_data)
+                error_message = "未知错误：未归类错误"
-            combined_pcm_data = b"".join(pcm_data)
 
-            # 判断是否保存为WAV文件
+            logger.bind(tag=TAG).debug(
-            if self.delete_audio_file:
+                f"ASR错误: {error_message} (错误码: {error_code})"
-                pass
+            )
-            else:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
 
-            # 直接使用PCM数据
+            # 如果是识别相关错误（>=1020），标记需要重连
-            # 计算分段大小 (单声道, 16bit, 16kHz采样率)
+            if error_code >= 1020:
-            size_per_sec = 1 * 2 * 16000  # nchannels * sampwidth * framerate
+                result["need_reconnect"] = True
-            segment_size = int(size_per_sec * self.seg_duration / 1000)
 
-            # 语音识别
+        return result
-            start_time = time.time()
-            text = await self._send_request(combined_pcm_data, segment_size)
-            if text:
-                logger.bind(tag=TAG).debug(
-                    f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
-                )
-                return text, file_path
-            return "", file_path
-
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-            return "", file_path

main/xiaozhi-server/core/providers/asr/dto/dto.py

@@ -0,0 +1,9 @@
+from enum import Enum
+from typing import Union, Optional
+
+
+class InterfaceType(Enum):
+    # 接口类型
+    STREAM = "STREAM"  # 流式接口
+    NON_STREAM = "NON_STREAM"  # 非流式接口
+    LOCAL = "LOCAL"  # 本地服务

main/xiaozhi-server/core/providers/asr/fun_local.py

@@ -1,15 +1,14 @@
 import time
-import wave
 import os
 import sys
 import io
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
-import uuid
 from core.providers.asr.base import ASRProviderBase
 from funasr import AutoModel
 from funasr.utils.postprocess_utils import rich_transcription_postprocess
 import shutil
+from core.providers.asr.dto.dto import InterfaceType
 
 TAG = __name__
 logger = setup_logging()
@@ -38,6 +37,7 @@ class CaptureOutput:
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
         super().__init__()
+        self.interface_type = InterfaceType.LOCAL
         self.model_dir = config.get("model_dir")
         self.output_dir = config.get("output_dir")  # 修正配置键名
         self.delete_audio_file = delete_audio_file
@@ -53,20 +53,6 @@ class ASRProvider(ASRProviderBase):
                 # device="cuda:0",  # 启用GPU加速
             )
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
-            wf.setframerate(16000)
-            wf.writeframes(b"".join(pcm_data))
-
-        return file_path
-
     async def speech_to_text(
         self, opus_data: List[bytes], session_id: str
     ) -> Tuple[Optional[str], Optional[str]]:

main/xiaozhi-server/core/providers/asr/fun_server.py

@@ -1,11 +1,8 @@
 from typing import Optional, Tuple, List
-import opuslib_next
 from core.providers.asr.base import ASRProviderBase
-import os
+from core.providers.asr.dto.dto import InterfaceType
 import ssl
 import json
-import uuid
-import wave
 import websockets
 from config.logger import setup_logging
 import asyncio
@@ -23,6 +20,7 @@ class ASRProvider(ASRProviderBase):
         :param delete_audio_file: Boolean to indicate whether to delete audio files after processing.
         """
         super().__init__()
+        self.interface_type = InterfaceType.NON_STREAM
         self.host = config.get("host", "localhost")
         self.port = config.get("port", 10095)
         self.api_key = config.get("api_key", "none")
@@ -43,20 +41,6 @@ class ASRProvider(ASRProviderBase):
             self.ssl_context.check_hostname = False
             self.ssl_context.verify_mode = ssl.CERT_NONE
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
-            wf.setframerate(16000)
-            wf.writeframes(b"".join(pcm_data))
-
-        return file_path
-
     async def _receive_responses(self, ws) -> None:
         """
         Asynchronous generator to receive messages from the WebSocket.

main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py

@@ -5,8 +5,7 @@ import sys
 import io
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
-import uuid
+from core.providers.asr.dto.dto import InterfaceType
-import opuslib_next
 from core.providers.asr.base import ASRProviderBase
 
 import numpy as np
@@ -38,6 +37,7 @@ class CaptureOutput:
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
         super().__init__()
+        self.interface_type = InterfaceType.LOCAL
         self.model_dir = config.get("model_dir")
         self.output_dir = config.get("output_dir")
         self.delete_audio_file = delete_audio_file
@@ -84,20 +84,6 @@ class ASRProvider(ASRProviderBase):
                 use_itn=True,
             )
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
-            wf.setframerate(16000)
-            wf.writeframes(b"".join(pcm_data))
-
-        return file_path
-
     def read_wave(self, wave_filename: str) -> Tuple[np.ndarray, int]:
         """
         Args:

main/xiaozhi-server/core/providers/asr/tencent.py

@@ -5,11 +5,8 @@ import json
 import time
 from datetime import datetime, timezone
 import os
-import uuid
 from typing import Optional, Tuple, List
-import wave
+from core.providers.asr.dto.dto import InterfaceType
-import opuslib_next
-
 import requests
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
@@ -25,6 +22,7 @@ class ASRProvider(ASRProviderBase):
 
     def __init__(self, config: dict, delete_audio_file: bool = True):
         super().__init__()
+        self.interface_type = InterfaceType.NON_STREAM
         self.secret_id = config.get("secret_id")
         self.secret_key = config.get("secret_key")
         self.output_dir = config.get("output_dir")
@@ -33,20 +31,6 @@ class ASRProvider(ASRProviderBase):
         # 确保输出目录存在
         os.makedirs(self.output_dir, exist_ok=True)
 
-    def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
-        """PCM数据保存为WAV文件"""
-        module_name = __name__.split(".")[-1]
-        file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav"
-        file_path = os.path.join(self.output_dir, file_name)
-
-        with wave.open(file_path, "wb") as wf:
-            wf.setnchannels(1)
-            wf.setsampwidth(2)  # 2 bytes = 16-bit
-            wf.setframerate(16000)
-            wf.writeframes(b"".join(pcm_data))
-
-        return file_path
-
     async def speech_to_text(
         self, opus_data: List[bytes], session_id: str
     ) -> Tuple[Optional[str], Optional[str]]:

main/xiaozhi-server/core/utils/modules_initialize.py

@@ -0,0 +1,128 @@
+from typing import Dict, Any
+from config.logger import setup_logging
+from core.utils import tts, llm, intent, memory, vad, asr
+
+TAG = __name__
+logger = setup_logging()
+
+
+def initialize_modules(
+    logger,
+    config: Dict[str, Any],
+    init_vad=False,
+    init_asr=False,
+    init_llm=False,
+    init_tts=False,
+    init_memory=False,
+    init_intent=False,
+) -> Dict[str, Any]:
+    """
+    初始化所有模块组件
+
+    Args:
+        config: 配置字典
+
+    Returns:
+        Dict[str, Any]: 包含所有初始化后的模块的字典
+    """
+    modules = {}
+
+    # 初始化TTS模块
+    if init_tts:
+        select_tts_module = config["selected_module"]["TTS"]
+        modules["tts"] = initialize_tts(config)
+        logger.bind(tag=TAG).info(f"初始化组件: tts成功 {select_tts_module}")
+
+    # 初始化LLM模块
+    if init_llm:
+        select_llm_module = config["selected_module"]["LLM"]
+        llm_type = (
+            select_llm_module
+            if "type" not in config["LLM"][select_llm_module]
+            else config["LLM"][select_llm_module]["type"]
+        )
+        modules["llm"] = llm.create_instance(
+            llm_type,
+            config["LLM"][select_llm_module],
+        )
+        logger.bind(tag=TAG).info(f"初始化组件: llm成功 {select_llm_module}")
+
+    # 初始化Intent模块
+    if init_intent:
+        select_intent_module = config["selected_module"]["Intent"]
+        intent_type = (
+            select_intent_module
+            if "type" not in config["Intent"][select_intent_module]
+            else config["Intent"][select_intent_module]["type"]
+        )
+        modules["intent"] = intent.create_instance(
+            intent_type,
+            config["Intent"][select_intent_module],
+        )
+        logger.bind(tag=TAG).info(f"初始化组件: intent成功 {select_intent_module}")
+
+    # 初始化Memory模块
+    if init_memory:
+        select_memory_module = config["selected_module"]["Memory"]
+        memory_type = (
+            select_memory_module
+            if "type" not in config["Memory"][select_memory_module]
+            else config["Memory"][select_memory_module]["type"]
+        )
+        modules["memory"] = memory.create_instance(
+            memory_type,
+            config["Memory"][select_memory_module],
+            config.get("summaryMemory", None),
+        )
+        logger.bind(tag=TAG).info(f"初始化组件: memory成功 {select_memory_module}")
+
+    # 初始化VAD模块
+    if init_vad:
+        select_vad_module = config["selected_module"]["VAD"]
+        vad_type = (
+            select_vad_module
+            if "type" not in config["VAD"][select_vad_module]
+            else config["VAD"][select_vad_module]["type"]
+        )
+        modules["vad"] = vad.create_instance(
+            vad_type,
+            config["VAD"][select_vad_module],
+        )
+        logger.bind(tag=TAG).info(f"初始化组件: vad成功 {select_vad_module}")
+
+    # 初始化ASR模块
+    if init_asr:
+        select_asr_module = config["selected_module"]["ASR"]
+        modules["asr"] = initialize_asr(config)
+        logger.bind(tag=TAG).info(f"初始化组件: asr成功 {select_asr_module}")
+    return modules
+
+
+def initialize_tts(config):
+    select_tts_module = config["selected_module"]["TTS"]
+    tts_type = (
+        select_tts_module
+        if "type" not in config["TTS"][select_tts_module]
+        else config["TTS"][select_tts_module]["type"]
+    )
+    new_tts = tts.create_instance(
+        tts_type,
+        config["TTS"][select_tts_module],
+        str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"),
+    )
+    return new_tts
+
+
+def initialize_asr(config):
+    select_asr_module = config["selected_module"]["ASR"]
+    asr_type = (
+        select_asr_module
+        if "type" not in config["ASR"][select_asr_module]
+        else config["ASR"][select_asr_module]["type"]
+    )
+    new_asr = asr.create_instance(
+        asr_type,
+        config["ASR"][select_asr_module],
+        str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"),
+    )
+    return new_asr

main/xiaozhi-server/core/utils/util.py

@@ -7,8 +7,6 @@ import numpy as np
 import requests
 import opuslib_next
 from pydub import AudioSegment
-from typing import Dict, Any
-from core.utils import tts, llm, intent, memory, vad, asr
 import copy
 
 TAG = __name__
@@ -245,122 +243,6 @@ def extract_json_from_string(input_string):
     return None
 
 
-def initialize_modules(
-    logger,
-    config: Dict[str, Any],
-    init_vad=False,
-    init_asr=False,
-    init_llm=False,
-    init_tts=False,
-    init_memory=False,
-    init_intent=False,
-) -> Dict[str, Any]:
-    """
-    初始化所有模块组件
-
-    Args:
-        config: 配置字典
-
-    Returns:
-        Dict[str, Any]: 包含所有初始化后的模块的字典
-    """
-    modules = {}
-
-    # 初始化TTS模块
-    if init_tts:
-        select_tts_module = config["selected_module"]["TTS"]
-        modules["tts"] = initialize_tts(config)
-        logger.bind(tag=TAG).info(f"初始化组件: tts成功 {select_tts_module}")
-
-    # 初始化LLM模块
-    if init_llm:
-        select_llm_module = config["selected_module"]["LLM"]
-        llm_type = (
-            select_llm_module
-            if "type" not in config["LLM"][select_llm_module]
-            else config["LLM"][select_llm_module]["type"]
-        )
-        modules["llm"] = llm.create_instance(
-            llm_type,
-            config["LLM"][select_llm_module],
-        )
-        logger.bind(tag=TAG).info(f"初始化组件: llm成功 {select_llm_module}")
-
-    # 初始化Intent模块
-    if init_intent:
-        select_intent_module = config["selected_module"]["Intent"]
-        intent_type = (
-            select_intent_module
-            if "type" not in config["Intent"][select_intent_module]
-            else config["Intent"][select_intent_module]["type"]
-        )
-        modules["intent"] = intent.create_instance(
-            intent_type,
-            config["Intent"][select_intent_module],
-        )
-        logger.bind(tag=TAG).info(f"初始化组件: intent成功 {select_intent_module}")
-
-    # 初始化Memory模块
-    if init_memory:
-        select_memory_module = config["selected_module"]["Memory"]
-        memory_type = (
-            select_memory_module
-            if "type" not in config["Memory"][select_memory_module]
-            else config["Memory"][select_memory_module]["type"]
-        )
-        modules["memory"] = memory.create_instance(
-            memory_type,
-            config["Memory"][select_memory_module],
-            config.get("summaryMemory", None),
-        )
-        logger.bind(tag=TAG).info(f"初始化组件: memory成功 {select_memory_module}")
-
-    # 初始化VAD模块
-    if init_vad:
-        select_vad_module = config["selected_module"]["VAD"]
-        vad_type = (
-            select_vad_module
-            if "type" not in config["VAD"][select_vad_module]
-            else config["VAD"][select_vad_module]["type"]
-        )
-        modules["vad"] = vad.create_instance(
-            vad_type,
-            config["VAD"][select_vad_module],
-        )
-        logger.bind(tag=TAG).info(f"初始化组件: vad成功 {select_vad_module}")
-
-    # 初始化ASR模块
-    if init_asr:
-        select_asr_module = config["selected_module"]["ASR"]
-        asr_type = (
-            select_asr_module
-            if "type" not in config["ASR"][select_asr_module]
-            else config["ASR"][select_asr_module]["type"]
-        )
-        modules["asr"] = asr.create_instance(
-            asr_type,
-            config["ASR"][select_asr_module],
-            str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"),
-        )
-        logger.bind(tag=TAG).info(f"初始化组件: asr成功 {select_asr_module}")
-    return modules
-
-
-def initialize_tts(config):
-    select_tts_module = config["selected_module"]["TTS"]
-    tts_type = (
-        select_tts_module
-        if "type" not in config["TTS"][select_tts_module]
-        else config["TTS"][select_tts_module]["type"]
-    )
-    new_tts = tts.create_instance(
-        tts_type,
-        config["TTS"][select_tts_module],
-        str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"),
-    )
-    return new_tts
-
-
 def analyze_emotion(text):
     """
     分析文本情感并返回对应的emoji名称（支持中英文）

main/xiaozhi-server/core/websocket_server.py

@@ -2,8 +2,9 @@ import asyncio
 import websockets
 from config.logger import setup_logging
 from core.connection import ConnectionHandler
-from core.utils.util import initialize_modules, check_vad_update, check_asr_update
 from config.config_loader import get_config_from_api
+from core.utils.modules_initialize import initialize_modules
+from core.utils.util import check_vad_update, check_asr_update
 
 TAG = __name__