update:兼容豆包流式ASR

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

@@ -2,7 +2,6 @@ import http.client
 import json
 import asyncio
 from typing import Optional, Tuple, List
-import opuslib_next
 import os
 import uuid
 import hmac
@@ -14,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()
@@ -90,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")

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

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

@@ -1,10 +1,15 @@
 import os
+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()
@@ -13,6 +18,54 @@ 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
+        # 如果本次没有声音，本段也没声音，就把声音丢弃了
+        if have_voice == False and self.conn.client_have_voice == False:
+            self.conn.asr_audio.append(audio)
+            self.conn.asr_audio = self.conn.asr_audio[-10:]
+            return
+
+        # 如果本段有声音，且已经停止了
+        if self.conn.client_voice_stop:
+            self.conn.client_abort = False
+            # 音频太短了，无法识别
+            if len(self.conn.asr_audio) < 15:
+                self.conn.asr_audio.clear()
+                self.conn.reset_vad_states()
+            else:
+                await self.handle_voice_stop()
+
+    # 处理语音停止
+    async def handle_voice_stop(self):
+        raw_text, _ = await self.speech_to_text(
+            self.conn.asr_audio, 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, copy.deepcopy(self.conn.asr_audio))
+        self.conn.asr_audio.clear()
+        self.conn.reset_vad_states()
 
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""

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

@@ -1,265 +1,431 @@
-import time
-import os
-import uuid
 import json
 import gzip
+import uuid
+import asyncio
 import websockets
-from config.logger import setup_logging
-from typing import Optional, Tuple, List
+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_SERIALIZATION = 0b0000
-JSON = 0b0001
-THRIFT = 0b0011
-CUSTOM_TYPE = 0b1111
-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
+NO_SEQUENCE = 0b0000
+NEG_SEQUENCE = 0b0010
+JSON_SERIALIZATION = 0b0001
+GZIP_COMPRESSION = 0b0001
+PROTOCOL_VERSION = 0b0001
 
 
 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 = f"wss://{self.host}/api/v2/asr"
-        self.success_code = 1000
-        self.seg_duration = 15000
+        self.ws_url = "wss://openspeech.bytedance.com/api/v2/asr"
+        self.uid = config.get("uid", "streaming_asr_service")
+        self.workflow = config.get(
+            "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)
 
-    @staticmethod
-    def _generate_header(
-        message_type=CLIENT_FULL_REQUEST, message_type_specific_flags=NO_SEQUENCE
-    ) -> bytearray:
-        """Generate protocol header."""
-        header = bytearray()
-        header_size = 1
-        header.append((0b0001 << 4) | header_size)  # Protocol version
-        header.append((message_type << 4) | message_type_specific_flags)
-        header.append((0b0001 << 4) | 0b0001)  # JSON serialization & GZIP compression
-        header.append(0x00)  # reserved
-        return header
+        retry_count = 0
+        while retry_count < self.max_retries:
+            try:
+                # 确保关闭旧的连接
+                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
 
-    def _construct_request(self, reqid) -> dict:
-        """Construct the request payload."""
-        return {
+                headers = self.token_auth() if self.auth_method == "token" else None
+                self.asr_ws = await websockets.connect(
+                    self.ws_url,
+                    additional_headers=headers,
+                    max_size=1000000000,
+                    ping_interval=None,  # 禁用ping，因为服务器可能不支持
+                    ping_timeout=None,
+                    close_timeout=10,
+                )
+
+                # 发送初始化请求
+                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
+
+                # 等待初始化响应
+                try:
+                    init_res = await self.asr_ws.recv()
+                    logger.bind(tag=TAG).debug(f"收到原始响应: {init_res}")
+                    result = self.parse_response(init_res)
+                    logger.bind(tag=TAG).info(f"ASR服务初始化响应: {result}")
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
+                    raise e
+
+                # 启动接收ASR结果的异步任务
+                asr_priority = threading.Thread(
+                    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": {
-                "uid": str(uuid.uuid4()),
-            },
+            "user": {"uid": self.uid},
             "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",
-                "rate": 16000,
-                "language": "zh-CN",
-                "bits": 16,
-                "channel": 1,
-                "codec": "raw",
+                "format": self.format,
+                "codec": self.codec,
+                "rate": self.rate,
+                "language": self.language,
+                "bits": self.bits,
+                "channel": self.channel,
             },
         }
+        return req
 
-    async def _send_request(
-        self, audio_data: List[bytes], segment_size: int
-    ) -> Optional[str]:
-        """Send request to Volcano ASR service."""
+    def token_auth(self):
+        return {"Authorization": f"Bearer; {self.access_token}"}
+
+    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)}
-            async with websockets.connect(
-                self.ws_url, additional_headers=auth_header
-            ) as websocket:
-                # Prepare request data
-                request_params = self._construct_request(str(uuid.uuid4()))
-                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
+            while not self.conn.stop_event.is_set():
+                try:
+                    if self.asr_ws is None:
+                        logger.bind(tag=TAG).info("尝试重新连接ASR服务...")
+                        await self.open_audio_channels(self.conn)
+                        continue
 
-                # Send header and metadata
-                # full_client_request
-                await websocket.send(full_client_request)
-                res = await websocket.recv()
-                result = parse_response(res)
-                if (
-                    "payload_msg" in result
-                    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,
+                    # 使用锁来确保同一时间只有一个协程在接收数据
+                    async with self.recv_lock:
+                        response = await self.asr_ws.recv()
+                        result = self.parse_response(response)
+                    # 检查是否需要重连
+                    if result.get("need_reconnect", False):
+                        logger.bind(tag=TAG).info(
+                            "检测到需要重连的错误，准备重新连接..."
                         )
-                    else:
-                        audio_only_request = self._generate_header(
-                            message_type=CLIENT_AUDIO_ONLY_REQUEST
-                        )
-                    payload_bytes = gzip.compress(chunk)
-                    audio_only_request.extend(
-                        (len(payload_bytes)).to_bytes(4, "big")
-                    )  # payload size(4 bytes)
-                    audio_only_request.extend(payload_bytes)  # payload
-                    # Send audio data
-                    await websocket.send(audio_only_request)
+                        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
+                        continue
 
-                # Receive response
-                response = await websocket.recv()
-                result = parse_response(response)
-
-                if (
-                    "payload_msg" in result
-                    and result["payload_msg"]["code"] == self.success_code
-                ):
-                    if len(result["payload_msg"]["result"]) > 0:
-                        return result["payload_msg"]["result"][0]["text"]
-                    return None
-                else:
-                    logger.bind(tag=TAG).error(f"ASR error: {result}")
-                    return None
+                    if "payload_msg" in result:
+                        if "result" in result["payload_msg"]:
+                            # 检查是否有utterances并且definite为True
+                            utterances = result["payload_msg"]["result"][0].get(
+                                "utterances", []
+                            )
+                            for utterance in utterances:
+                                if utterance.get("definite", False):
+                                    self.text = utterance["text"]
+                                    await self.handle_voice_stop()
+                                    break
+                except websockets.ConnectionClosed:
+                    logger.bind(tag=TAG).debug("ASR服务连接已关闭，准备重连...")
+                    # 确保关闭旧连接
+                    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:
+                    logger.bind(tag=TAG).error(f"处理ASR结果时发生错误: {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)
-            return None
+            logger.bind(tag=TAG).error(f"ASR监听线程发生错误: {e}")
+            # 确保在发生严重错误时也能继续尝试重连
+            if not self.conn.stop_event.is_set():
+                await asyncio.sleep(self.retry_delay)
+                await self._forward_asr_results()  # 递归重试
 
-    @staticmethod
-    def slice_data(data: bytes, chunk_size: int) -> (list, bool):
+    async def speech_to_text(self, opus_data, session_id):
+        result = self.text
+        self.text = ""  # 清空text
+        return result, None
+
+    def parse_response(self, res: bytes) -> dict:
         """
-        slice data
-        :param data: wav data
-        :param chunk_size: the segment size in one request
-        :return: segment data, last flag
+        解析 ASR 服务返回的二进制响应。
+        根据协议格式解析头部和 payload，若采用 GZIP 压缩则先解压，再根据 JSON 反序列化。
         """
-        data_len = len(data)
-        offset = 0
-        while offset + chunk_size < data_len:
-            yield data[offset : offset + chunk_size], False
-            offset += chunk_size
+        protocol_version = res[0] >> 4
+        header_size = res[0] & 0x0F
+        message_type = res[1] >> 4
+        serialization_method = res[2] >> 4
+        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
-    ) -> Tuple[Optional[str], Optional[str]]:
-        """将语音数据转换为文本"""
+        # 错误码处理
+        if "code" in result:
+            error_code = result["code"]
+            error_message = ""
 
-        file_path = None
-        try:
-            # 合并所有opus数据包
-            if self.audio_format == "pcm":
-                pcm_data = opus_data
+            if error_code == 1000:
+                error_message = "成功"
+            elif error_code == 1001:
+                error_message = "请求参数无效：请求参数缺失必需字段/字段值无效/重复请求"
+            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)
-            combined_pcm_data = b"".join(pcm_data)
+                error_message = "未知错误：未归类错误"
 
-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
+            logger.bind(tag=TAG).debug(
+                f"ASR错误: {error_message} (错误码: {error_code})"
+            )
 
-            # 直接使用PCM数据
-            # 计算分段大小 (单声道, 16bit, 16kHz采样率)
-            size_per_sec = 1 * 2 * 16000  # nchannels * sampwidth * framerate
-            segment_size = int(size_per_sec * self.seg_duration / 1000)
+            # 如果是识别相关错误（>=1020），标记需要重连
+            if error_code >= 1020:
+                result["need_reconnect"] = True
 
-            # 语音识别
-            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
+        return result

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

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

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
-import opuslib_next
+from core.providers.asr.dto.dto import InterfaceType
 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

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

@@ -5,9 +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 requests
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
@@ -23,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")