diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 04c352bf..b8f06f70 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -12,18 +12,22 @@ import subprocess import websockets 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 @@ -112,7 +116,6 @@ class ConnectionHandler: # asr相关变量 self.asr_audio = [] - self.asr_server_receive = True # llm相关变量 self.llm_finish_task = True @@ -315,10 +318,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 ) @@ -356,6 +363,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: @@ -597,6 +617,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: @@ -622,9 +644,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( @@ -801,9 +820,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: @@ -834,7 +851,6 @@ class ConnectionHandler: def clearSpeakStatus(self): self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") - self.asr_server_receive = True async def close(self, ws=None): """资源清理方法""" diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 4c876550..cb2e3c69 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/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)) - diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 18482a04..6776dd0f 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/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: - conn.logger.bind(tag=TAG).debug(f"前期数据处理中，暂停接收") + if conn.asr is None: + 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) - else: - have_voice = conn.client_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() + # 当前片段是否有人说话 + have_voice = conn.vad.is_vad(conn, audio) + # 设备长时间空闲检测，用于say goodbye + await no_voice_close_connect(conn, have_voice) + # 接收音频 + await conn.asr.receive_audio(audio, have_voice) 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("结束对话，无需发送结束提示语") diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 5694c618..6625838c 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/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( diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index da6ede47..602f72aa 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -41,12 +41,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") diff --git a/main/xiaozhi-server/core/ota_server.py b/main/xiaozhi-server/core/ota_server.py index 89e36869..5f89a503 100644 --- a/main/xiaozhi-server/core/ota_server.py +++ b/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, initialize_modules +from core.utils.util import get_local_ip +from core.utils.modules_initialize import initialize_modules TAG = __name__ diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index 5fa18253..7ed24c00 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/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") diff --git a/main/xiaozhi-server/core/providers/asr/baidu.py b/main/xiaozhi-server/core/providers/asr/baidu.py index 87cb6a71..7df94ff3 100644 --- a/main/xiaozhi-server/core/providers/asr/baidu.py +++ b/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") diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 153f40e2..d3ce5ee0 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/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文件""" diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index bd33af40..d738f508 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/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 diff --git a/main/xiaozhi-server/core/providers/asr/dto/dto.py b/main/xiaozhi-server/core/providers/asr/dto/dto.py new file mode 100644 index 00000000..85b2d7b2 --- /dev/null +++ b/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" # 本地服务 diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index e3b7fc83..8c7305c6 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/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 diff --git a/main/xiaozhi-server/core/providers/asr/fun_server.py b/main/xiaozhi-server/core/providers/asr/fun_server.py index ff411340..60d5e425 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_server.py +++ b/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") diff --git a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py index 24522d83..a27640dd 100644 --- a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py +++ b/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 diff --git a/main/xiaozhi-server/core/providers/asr/tencent.py b/main/xiaozhi-server/core/providers/asr/tencent.py index 22a7ed94..03301568 100644 --- a/main/xiaozhi-server/core/providers/asr/tencent.py +++ b/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") diff --git a/main/xiaozhi-server/core/utils/modules_initialize.py b/main/xiaozhi-server/core/utils/modules_initialize.py new file mode 100644 index 00000000..f2e3968e --- /dev/null +++ b/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 diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 8f5ded4b..ff861aa9 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/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名称（支持中英文） diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index e60f0c87..1a1ba4c0 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/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__