diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index a615791c..fcf5183b 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -11,6 +11,8 @@ import threading import traceback import subprocess import websockets +import opuslib_next +import numpy as np from core.utils.util import ( extract_json_from_string, @@ -114,6 +116,7 @@ class ConnectionHandler: self.client_abort = False self.client_is_speaking = False self.client_listen_mode = "auto" + self.client_aec = False # 是否启用了服务端AEC # 线程任务相关 self.loop = None # 在 handle_connection 中获取运行中的事件循环 @@ -153,7 +156,7 @@ class ConnectionHandler: # asr相关变量 # 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR # 所以涉及到ASR的变量,需要在这里定义,属于connection的私有变量 - self.asr_audio = [] + self.asr_audio = [] # 存储PCM帧列表,供VAD和ASR共享 self.asr_audio_queue = queue.Queue() self.current_speaker = None # 存储当前说话人 @@ -231,6 +234,9 @@ class ConnectionHandler: # 启动超时检查任务 self.timeout_task = asyncio.create_task(self._check_timeout()) + # 启动AEC缓存清理任务 + self._aec_cache_cleanup_task = asyncio.create_task(self._check_aec_cache_expiry()) + self.welcome_msg = self.config["xiaozhi"] self.welcome_msg["session_id"] = self.session_id @@ -366,12 +372,14 @@ class ConnectionHandler: if handled: return - # 不需要头部处理或没有头部时,直接处理原始消息 - self.asr_audio_queue.put(message) + # 入口处直接解码PCM,避免VAD和ASR重复解码 + pcm_frame = self._decode_opus_packet(message) + if pcm_frame: + self.asr_audio_queue.put(pcm_frame) async def _process_mqtt_audio_message(self, message): """ - 处理来自MQTT网关的音频消息,解析16字节头部并提取音频数据 + 处理来自MQTT网关的音频消息,解析16字节头部并提取音频数据,在入队前进行AEC处理 Args: message: 包含头部的音频消息 @@ -380,12 +388,20 @@ class ConnectionHandler: bool: 是否成功处理了消息 """ try: - # 提取头部信息 + # 解析timestamp timestamp = int.from_bytes(message[8:12], "big") - # 提取音频数据 audio_data = message[16:] - self.asr_audio_queue.put(audio_data) + # 入口直接解码PCM + pcm_frame = self._decode_opus_packet(audio_data) + if not pcm_frame: + return True + + # AEC处理:如果timestamp>0且启用了AEC + if timestamp > 0 and self.client_aec: + pcm_frame = self._apply_aec(timestamp, pcm_frame) + + self.asr_audio_queue.put(pcm_frame) return True except Exception as e: self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}") @@ -393,6 +409,172 @@ class ConnectionHandler: # 处理失败,返回False表示需要继续处理 return False + def _apply_aec(self, timestamp: int, pcm_frame: bytes) -> bytes: + """应用AEC处理 - 综合算法:互相关延迟估计 + Wiener滤波 + 频谱减法""" + try: + import numpy as np + + if not pcm_frame or len(pcm_frame) == 0: + return pcm_frame + + if not hasattr(self, "aec_audio_cache") or not self.aec_audio_cache: + return pcm_frame + + mic_audio = np.frombuffer(pcm_frame, dtype=np.int16).astype(np.float32) + mic_rms = np.sqrt(np.mean(mic_audio ** 2)) + + if mic_rms < 100: + return pcm_frame + + # 初始化 + if not hasattr(self, "_aec_filter_state"): + self._aec_filter_state = np.zeros(512) # 滤波器状态 + self._aec_delay_ms = 200 + self._aec_last_delay = 200 + self._aec_coherence_history = [] + + sorted_timestamps = sorted(self.aec_audio_cache.keys()) + if len(sorted_timestamps) < 2: + return pcm_frame + + # ========== 匹配参考帧(对数功率谱匹配) ========== + n = len(mic_audio) + sorted_timestamps = sorted(self.aec_audio_cache.keys()) + + # 找最接近的timestamp作为起点 + closest_idx = min(range(len(sorted_timestamps)), key=lambda i: abs(sorted_timestamps[i] - timestamp)) + + # 对数功率谱相关性计算 + def log_powerSpectrum_corr(audio1, audio2): + window = np.hanning(len(audio1)) + fft1 = np.fft.rfft(audio1 * window) + fft2 = np.fft.rfft(audio2 * window) + psd1 = np.abs(fft1) ** 2 + psd2 = np.abs(fft2) ** 2 + log1 = 10 * np.log10(psd1 + 1e-8) + log2 = 10 * np.log10(psd2 + 1e-8) + P_xy = np.dot(log1, log2) + P_xx = np.dot(log1, log1) + P_yy = np.dot(log2, log2) + return abs(P_xy) / (np.sqrt(P_xx) * np.sqrt(P_yy) + 1e-8) + + # 用对数功率谱匹配找最佳帧:前后各找2帧 + best_corr = -1 + best_ref_idx = closest_idx + + for offset in range(-2, 3): # T-2, T-1, T, T+1, T+2 + test_idx = closest_idx + offset + if test_idx < 0 or test_idx >= len(sorted_timestamps): + continue + test_ts = sorted_timestamps[test_idx] + test_ref = np.frombuffer(self.aec_audio_cache[test_ts], dtype=np.int16).astype(np.float32) + test_ref_rms = np.sqrt(np.mean(test_ref ** 2)) + if test_ref_rms < 50: + continue + + # 对数功率谱相关性 + corr = log_powerSpectrum_corr(mic_audio, test_ref) + + if corr > best_corr: + best_corr = corr + best_ref_idx = test_idx + + best_ts = sorted_timestamps[best_ref_idx] + + best_ref = np.frombuffer(self.aec_audio_cache[best_ts], dtype=np.int16).astype(np.float32) + ref_rms = np.sqrt(np.mean(best_ref ** 2)) + + if ref_rms < 50: + return pcm_frame + + # 对齐参考信号(直接截取相同长度) + aligned_ref = best_ref[:n] + if len(aligned_ref) < n: + aligned_ref = np.pad(aligned_ref, (0, n - len(aligned_ref))) + + # ========== 频域 AEC 处理(谱减法) ========== + # 时域信号经过声学路径后相位失真,导致时域相关性低且P_xy正负不定 + # 频域幅度谱不受相位影响,对数功率谱相关性稳定在0.97+ + # 公式:result_mag = max(|mic_fft| - |ref_fft| * scale * coef, 0) + + window = np.hanning(n) + mic_fft = np.fft.rfft(mic_audio * window) + ref_fft = np.fft.rfft(aligned_ref * window) + + mic_mag = np.abs(mic_fft) + ref_mag = np.abs(ref_fft) + mic_phase = np.angle(mic_fft) + + # 频域计算回声比例 scale + scale = np.sum(mic_mag * ref_mag) / (np.dot(ref_mag, ref_mag) + 1e-8) + + # 使用对数功率谱相关性作为coh + coh = best_corr + ref_power = ref_rms + + # 自适应系数:根据scale和coh动态调整 + # scale大(回声强)-> coef大;coh高(匹配准)-> coef大 + raw_coef = 1.0 + scale * 3 + (coh - 0.97) * 30 + coef = max(0.5, min(3.0, raw_coef)) + + # 谱减法(过减 + 半波整流) + if ref_power < 50: + output = mic_audio + coef = 0 + gain = 0 + else: + echo_mag = ref_mag * scale * coef + # 过减:使用较大系数抑制回声 + over_subtract = 1.5 + result_mag = np.maximum(mic_mag - echo_mag * over_subtract, mic_mag * 0.1) + + # 保留相位重建信号 + result_fft = result_mag * np.exp(1j * mic_phase) + output = np.fft.irfft(result_fft, n) + gain = scale * coef + + # 高置信度是纯回声时,再压一下确保VAD检测不到 + if coh >= 0.97 and ref_power > 500: + output = output * 0.3 + + # 后处理:限幅 + output = np.clip(output, -32768, 32767) + + # 转换为bytes + result = output.astype(np.int16).tobytes() + + return result + + except Exception as e: + self.logger.bind(tag=TAG).warning(f"[AEC] 处理失败: {e}") + return pcm_frame + + def _decode_opus_packet(self, opus_packet: bytes) -> bytes: + """ + 解码Opus数据包为PCM数据 + + Args: + opus_packet: Opus编码的音频数据 + + Returns: + bytes: 解码后的PCM数据,失败返回None + """ + try: + if not opus_packet or len(opus_packet) == 0: + return None + + self._init_connection_state(self) + pcm_frame = self._connection_opus_decoder.decode(opus_packet, 960) + return pcm_frame + except Exception as e: + self.logger.bind(tag=TAG).debug(f"Opus解码失败: {e}") + return None + + def _init_connection_state(self, conn): + """为连接初始化独立的Opus解码器""" + if not hasattr(conn, "_connection_opus_decoder"): + conn._connection_opus_decoder = opuslib_next.Decoder(16000, 1) + async def handle_restart(self, message): """处理服务器重启请求""" try: @@ -1363,6 +1545,13 @@ class ConnectionHandler: ): self.vad.release_conn_resources(self) + # 清理opus解码器 + if hasattr(self, "_connection_opus_decoder"): + try: + delattr(self, "_connection_opus_decoder") + except Exception: + pass + # 清理音频缓冲区 if hasattr(self, "audio_buffer"): self.audio_buffer.clear() @@ -1376,6 +1565,20 @@ class ConnectionHandler: pass self.timeout_task = None + # 取消AEC缓存清理任务 + if hasattr(self, "_aec_cache_cleanup_task") and self._aec_cache_cleanup_task and not self._aec_cache_cleanup_task.done(): + self._aec_cache_cleanup_task.cancel() + try: + await self._aec_cache_cleanup_task + except asyncio.CancelledError: + pass + self._aec_cache_cleanup_task = None + + # 清理AEC缓存 + if hasattr(self, "aec_audio_cache"): + self.aec_audio_cache.clear() + self.aec_audio_cache_time.clear() + # 清理工具处理器资源 if hasattr(self, "func_handler") and self.func_handler: try: @@ -1539,6 +1742,26 @@ class ConnectionHandler: finally: self.logger.bind(tag=TAG).info("超时检查任务已退出") + async def _check_aec_cache_expiry(self): + """定期清理过期的AEC缓存""" + try: + while not self.stop_event.is_set(): + if hasattr(self, "aec_audio_cache") and self.aec_audio_cache: + current_time = time.time() + expired_keys = [ + ts for ts, cache_time in list(self.aec_audio_cache_time.items()) + if current_time - cache_time > 120 # 2分钟过期 + ] + for ts in expired_keys: + self.aec_audio_cache.pop(ts, None) + self.aec_audio_cache_time.pop(ts, None) + if expired_keys: + self.logger.bind(tag=TAG).debug(f"[AEC] 清理过期缓存 {len(expired_keys)} 条") + # 每30秒检查一次 + await asyncio.sleep(30) + except Exception as e: + self.logger.bind(tag=TAG).error(f"AEC缓存清理任务出错: {e}") + @staticmethod def _extract_direct_answer_response(arguments_str): """从 direct_answer 的参数中提取 response 值。 diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 6bad8380..2f84a7b1 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -56,6 +56,9 @@ async def handleHelloMessage(conn: "ConnectionHandler", msg_json): conn.mcp_client = MCPClient() # 发送初始化 asyncio.create_task(send_mcp_initialize_message(conn)) + if features.get("aec"): + conn.logger.bind(tag=TAG).debug("客户端启用了服务端AEC") + conn.client_aec = True await conn.websocket.send(json.dumps(conn.welcome_msg)) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index ae4673a0..63a9f610 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -14,9 +14,9 @@ from core.handle.sendAudioHandle import send_stt_message, SentenceType TAG = __name__ -async def handleAudioMessage(conn: "ConnectionHandler", audio): +async def handleAudioMessage(conn: "ConnectionHandler", pcm_frame): # 当前片段是否有人说话 - have_voice = conn.vad.is_vad(conn, audio) + have_voice = conn.vad.is_vad(conn, pcm_frame) # 如果设备刚刚被唤醒,短暂忽略VAD检测 if hasattr(conn, "just_woken_up") and conn.just_woken_up: have_voice = False @@ -24,10 +24,14 @@ async def handleAudioMessage(conn: "ConnectionHandler", audio): if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done(): conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn)) return + # 服务端AEC功能需要实时触发打断 + if conn.client_aec and have_voice: + if conn.client_is_speaking and conn.client_listen_mode != "manual": + await handleAbortMessage(conn) # 设备长时间空闲检测,用于say goodbye await no_voice_close_connect(conn, have_voice) # 接收音频 - await conn.asr.receive_audio(conn, audio, have_voice) + await conn.asr.receive_audio(conn, pcm_frame, have_voice) async def resume_vad_detection(conn: "ConnectionHandler"): diff --git a/main/xiaozhi-server/core/handle/reportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py index 97fe8b47..ea621f74 100644 --- a/main/xiaozhi-server/core/handle/reportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -50,33 +50,27 @@ async def report(conn: "ConnectionHandler", type, text, opus_data, report_time): conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}") -def opus_to_wav(conn: "ConnectionHandler", opus_data): - """将Opus数据转换为WAV格式的字节流 +def opus_to_wav(conn: "ConnectionHandler", pcm_data): + """将PCM数据转换为WAV格式的字节流 Args: output_dir: 输出目录(保留参数以保持接口兼容) - opus_data: opus音频数据 + pcm_data: PCM音频数据(可能是列表或bytes) Returns: bytes: WAV格式的音频数据 """ - decoder = None try: - decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道 - pcm_data = [] + # 处理可能是列表或bytes的PCM数据 + if isinstance(pcm_data, list): + pcm_data_bytes = b"".join(pcm_data) + else: + pcm_data_bytes = pcm_data - for opus_packet in opus_data: - try: - pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms - pcm_data.append(pcm_frame) - except opuslib_next.OpusError as e: - conn.logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True) - - if not pcm_data: + if not pcm_data_bytes: raise ValueError("没有有效的PCM数据") # 创建WAV文件头 - pcm_data_bytes = b"".join(pcm_data) num_samples = len(pcm_data_bytes) // 2 # 16-bit samples # WAV文件头 @@ -97,12 +91,9 @@ def opus_to_wav(conn: "ConnectionHandler", opus_data): # 返回完整的WAV数据 return bytes(wav_header) + pcm_data_bytes - finally: - if decoder is not None: - try: - del decoder - except Exception as e: - conn.logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}") + except Exception as e: + conn.logger.bind(tag=TAG).error(f"PCM转WAV失败: {e}", exc_info=True) + raise def enqueue_tts_report(conn: "ConnectionHandler", text, opus_data): diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 17de11ae..421554d6 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,6 +1,7 @@ import json import time import asyncio +import opuslib_next from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -81,13 +82,24 @@ async def _send_to_mqtt_gateway( conn: "ConnectionHandler", opus_packet, timestamp, sequence ): """ - 发送带16字节头部的opus数据包给mqtt_gateway + 发送带16字节头部的opus数据包给mqtt_gateway,同时缓存音频用于AEC处理 Args: conn: 连接对象 opus_packet: opus数据包 timestamp: 时间戳 sequence: 序列号 """ + # 如果启用了服务端AEC,缓存PCM数据用于后续AEC处理 + if conn.client_aec and timestamp > 0: + if not hasattr(conn, "aec_audio_cache"): + conn.aec_audio_cache = {} + conn.aec_audio_cache_time = {} + conn._send_opus_decoder = opuslib_next.Decoder(16000, 1) + # 解码opus为PCM后缓存 + pcm_data = conn._send_opus_decoder.decode(bytes(opus_packet), 960) + conn.aec_audio_cache[timestamp] = bytes(pcm_data) + conn.aec_audio_cache_time[timestamp] = time.time() + # 为opus数据包添加16字节头部 header = bytearray(16) header[0] = 1 # type diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index 8824d17b..ce9f4d15 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -213,7 +213,7 @@ class ASRProvider(ASRProviderBase): return None async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if self._is_token_expired(): diff --git a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py index 6c0bf093..74bf6137 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py @@ -7,7 +7,6 @@ import hashlib import asyncio import requests import websockets -import opuslib_next from urllib import parse from datetime import datetime from config.logger import setup_logging @@ -74,7 +73,6 @@ class ASRProvider(ASRProviderBase): self.interface_type = InterfaceType.STREAM self.config = config self.text = "" - self.decoder = opuslib_next.Decoder(16000, 1) self.asr_ws = None self.forward_task = None self.is_processing = False @@ -129,9 +127,9 @@ class ASRProvider(ASRProviderBase): async def open_audio_channels(self, conn): await super().open_audio_channels(conn) - async def receive_audio(self, conn, audio, audio_have_voice): + async def receive_audio(self, conn, pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 - await super().receive_audio(conn, audio, audio_have_voice) + await super().receive_audio(conn, pcm_frame, audio_have_voice) # 只在有声音且没有连接时建立连接(排除正在停止的情况) if audio_have_voice and not self.is_processing and not self.asr_ws: @@ -144,7 +142,6 @@ class ASRProvider(ASRProviderBase): if self.asr_ws and self.is_processing and self.server_ready: try: - pcm_frame = self.decoder.decode(audio, 960) await self.asr_ws.send(pcm_frame) except Exception as e: logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}") @@ -232,10 +229,9 @@ class ASRProvider(ASRProviderBase): # 发送缓存音频 if conn.asr_audio: - for cached_audio in conn.asr_audio[-10:]: + for cached_pcm in conn.asr_audio[-10:]: try: - pcm_frame = self.decoder.decode(cached_audio, 960) - await self.asr_ws.send(pcm_frame) + await self.asr_ws.send(cached_pcm) except Exception as e: logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}") break @@ -328,7 +324,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).debug("ASR会话清理完成") - async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None): + async def speech_to_text(self, opus_data, session_id, artifacts=None): """获取识别结果""" result = self.text self.text = "" @@ -337,10 +333,3 @@ class ASRProvider(ASRProviderBase): async def close(self): """关闭资源""" await self._cleanup() - if hasattr(self, 'decoder') and self.decoder is not None: - try: - del self.decoder - self.decoder = None - logger.bind(tag=TAG).debug("Aliyun decoder resources released") - except Exception as e: - logger.bind(tag=TAG).debug(f"释放Aliyun decoder资源时出错: {e}") diff --git a/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py index 83b79655..a7c2953a 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py +++ b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py @@ -2,7 +2,6 @@ import json import uuid import asyncio import websockets -import opuslib_next from typing import List, TYPE_CHECKING if TYPE_CHECKING: @@ -22,7 +21,6 @@ class ASRProvider(ASRProviderBase): self.interface_type = InterfaceType.STREAM self.config = config self.text = "" - self.decoder = opuslib_next.Decoder(16000, 1) self.asr_ws = None self.forward_task = None self.is_processing = False @@ -55,9 +53,9 @@ class ASRProvider(ASRProviderBase): async def open_audio_channels(self, conn): await super().open_audio_channels(conn) - async def receive_audio(self, conn, audio, audio_have_voice): + async def receive_audio(self, conn, pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 - await super().receive_audio(conn, audio, audio_have_voice) + await super().receive_audio(conn, pcm_frame, audio_have_voice) # 只在有声音且没有连接时建立连接 if audio_have_voice and not self.is_processing and not self.asr_ws: @@ -71,8 +69,6 @@ class ASRProvider(ASRProviderBase): # 发送音频数据 if self.asr_ws and self.is_processing and self.server_ready: try: - pcm_frame = self.decoder.decode(audio, 960) - # 直接发送PCM音频数据(二进制) await self.asr_ws.send(pcm_frame) except Exception as e: logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}") @@ -179,10 +175,9 @@ class ASRProvider(ASRProviderBase): # 发送缓存音频 if conn.asr_audio: - for cached_audio in conn.asr_audio[-10:]: + for cached_pcm in conn.asr_audio[-10:]: try: - pcm_frame = self.decoder.decode(cached_audio, 960) - await self.asr_ws.send(pcm_frame) + await self.asr_ws.send(cached_pcm) except Exception as e: logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}") break @@ -312,7 +307,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).debug("ASR会话清理完成") - async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None): + async def speech_to_text(self, opus_data, session_id, artifacts=None): """获取识别结果""" result = self.text self.text = "" @@ -320,11 +315,4 @@ class ASRProvider(ASRProviderBase): async def close(self): """关闭资源""" - await self._cleanup() - if hasattr(self, 'decoder') and self.decoder is not None: - try: - del self.decoder - self.decoder = None - logger.bind(tag=TAG).debug("Aliyun BL decoder resources released") - except Exception as e: - logger.bind(tag=TAG).debug(f"释放Aliyun BL decoder资源时出错: {e}") \ No newline at end of file + await self._cleanup() \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/asr/baidu.py b/main/xiaozhi-server/core/providers/asr/baidu.py index aa14f443..541bb010 100644 --- a/main/xiaozhi-server/core/providers/asr/baidu.py +++ b/main/xiaozhi-server/core/providers/asr/baidu.py @@ -30,7 +30,7 @@ class ASRProvider(ASRProviderBase): os.makedirs(self.output_dir, exist_ok=True) async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if not opus_data: diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 986dd275..dcfaf3a7 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -10,7 +10,6 @@ import asyncio import tempfile import traceback import threading -import opuslib_next from abc import ABC, abstractmethod from config.logger import setup_logging @@ -59,13 +58,13 @@ class ASRProviderBase(ABC): continue # 接收音频 - async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice): + async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice): if conn.client_listen_mode == "manual": # 手动模式:缓存音频用于ASR识别 - conn.asr_audio.append(audio) + conn.asr_audio.append(pcm_frame) else: # 自动/实时模式:使用VAD检测 - conn.asr_audio.append(audio) + conn.asr_audio.append(pcm_frame) # 如果没有语音,且之前也没有声音,缓存部分音频 if not audio_have_voice and not conn.client_have_voice: @@ -74,24 +73,21 @@ class ASRProviderBase(ABC): # 自动模式下通过VAD检测到语音停止时触发识别 if conn.asr.interface_type != InterfaceType.STREAM and conn.client_voice_stop: - asr_audio_task = conn.asr_audio.copy() + # 直接使用asr_audio中的PCM数据 + pcm_bytes = b"".join(conn.asr_audio) + # 检查是否有足够的音频数据(每帧1920字节,15帧约28800字节) + if len(pcm_bytes) > 1920 * 15: + await self.handle_voice_stop(conn, [pcm_bytes]) conn.reset_audio_states() - if len(asr_audio_task) > 15: - await self.handle_voice_stop(conn, asr_audio_task) - # 处理语音停止 async def handle_voice_stop(self, conn: "ConnectionHandler", asr_audio_task: List[bytes]): """并行处理ASR和声纹识别""" try: total_start_time = time.monotonic() - # 准备音频数据 - if conn.audio_format == "pcm": - pcm_data = asr_audio_task - else: - pcm_data = self.decode_opus(asr_audio_task) - + # 数据已经是PCM直接使用 + pcm_data = asr_audio_task combined_pcm_data = b"".join(pcm_data) # 预先准备WAV数据 @@ -101,7 +97,7 @@ class ASRProviderBase(ABC): # 定义ASR任务 asr_task = self.speech_to_text_wrapper( - asr_audio_task, conn.session_id, conn.audio_format + asr_audio_task, conn.session_id ) if conn.voiceprint_provider and wav_data: @@ -270,15 +266,11 @@ class ASRProviderBase(ABC): return file_path async def speech_to_text_wrapper( - self, opus_data: List[bytes], session_id: str, audio_format="opus" + self, pcm_data: List[bytes], session_id: str ) -> Tuple[Optional[str], Optional[str]]: file_path = None temp_path = None try: - if audio_format == "pcm": - pcm_data = opus_data - else: - pcm_data = self.decode_opus(opus_data) combined_pcm_data = b"".join(pcm_data) free_space = shutil.disk_usage(self.output_dir).free @@ -304,7 +296,7 @@ class ASRProviderBase(ABC): ) text, _ = await self.speech_to_text( - opus_data, session_id, audio_format, artifacts + pcm_data, session_id, artifacts ) return text, file_path except OSError as e: @@ -332,50 +324,13 @@ class ASRProviderBase(ABC): self, opus_data: List[bytes], session_id: str, - audio_format="opus", artifacts: Optional[AudioArtifacts] = None, ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本 :param opus_data: 输入的Opus音频数据 :param session_id: 会话ID - :param audio_format: 音频格式,默认"opus" :param artifacts: 音频工件,包含PCM数据、文件路径等 :return: 识别结果文本和文件路径(如果有) """ pass - - @staticmethod - def decode_opus(opus_data: List[bytes]) -> List[bytes]: - """将Opus音频数据解码为PCM数据""" - decoder = None - try: - decoder = opuslib_next.Decoder(16000, 1) - pcm_data = [] - buffer_size = 960 # 每次处理960个采样点 (60ms at 16kHz) - - for i, opus_packet in enumerate(opus_data): - try: - if not opus_packet or len(opus_packet) == 0: - continue - - pcm_frame = decoder.decode(opus_packet, buffer_size) - if pcm_frame and len(pcm_frame) > 0: - pcm_data.append(pcm_frame) - - except opuslib_next.OpusError as e: - logger.bind(tag=TAG).warning(f"Opus解码错误,跳过数据包 {i}: {e}") - except Exception as e: - logger.bind(tag=TAG).error(f"音频处理错误,数据包 {i}: {e}") - - return pcm_data - - except Exception as e: - logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}") - return [] - finally: - if decoder is not None: - try: - del decoder - except Exception as e: - logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}") diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index 016ac0fd..ee8b7cc8 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/main/xiaozhi-server/core/providers/asr/doubao.py @@ -232,7 +232,7 @@ class ASRProvider(ASRProviderBase): yield data[offset:data_len], True async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" diff --git a/main/xiaozhi-server/core/providers/asr/doubao_stream.py b/main/xiaozhi-server/core/providers/asr/doubao_stream.py index 5c34aa77..0e5e6c32 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao_stream.py +++ b/main/xiaozhi-server/core/providers/asr/doubao_stream.py @@ -3,7 +3,6 @@ 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 @@ -22,7 +21,6 @@ class ASRProvider(ASRProviderBase): self.interface_type = InterfaceType.STREAM self.config = config self.text = "" - self.decoder = opuslib_next.Decoder(16000, 1) self.asr_ws = None self.forward_task = None self.is_processing = False # 添加处理状态标志 @@ -68,10 +66,10 @@ class ASRProvider(ASRProviderBase): async def open_audio_channels(self, conn): await super().open_audio_channels(conn) - async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice): + async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 - await super().receive_audio(conn, audio, audio_have_voice) - + await super().receive_audio(conn, pcm_frame, audio_have_voice) + # 如果本次有声音,且之前没有建立连接 if audio_have_voice and self.asr_ws is None and not self.is_processing: try: @@ -123,10 +121,9 @@ class ASRProvider(ASRProviderBase): # 发送缓存的音频数据 if conn.asr_audio and len(conn.asr_audio) > 0: - for cached_audio in conn.asr_audio[-10:]: + for cached_pcm in conn.asr_audio[-10:]: try: - pcm_frame = self.decoder.decode(cached_audio, 960) - payload = gzip.compress(pcm_frame) + payload = gzip.compress(cached_pcm) audio_request = bytearray( self.generate_audio_default_header() ) @@ -151,7 +148,6 @@ class ASRProvider(ASRProviderBase): # 发送当前音频数据 if self.asr_ws and self.is_processing and not self._is_stopping: try: - 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")) @@ -414,7 +410,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}") raise - async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None): + async def speech_to_text(self, opus_data, session_id, artifacts=None): result = self.text self.text = "" # 清空text return result, None @@ -433,11 +429,3 @@ class ASRProvider(ASRProviderBase): self.forward_task = None self.is_processing = False - # 显式释放decoder资源 - if hasattr(self, "decoder") and self.decoder is not None: - try: - del self.decoder - self.decoder = None - logger.bind(tag=TAG).debug("Doubao decoder resources released") - except Exception as e: - logger.bind(tag=TAG).debug(f"释放Doubao decoder资源时出错: {e}") diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index ef11ae02..5547104d 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/main/xiaozhi-server/core/providers/asr/fun_local.py @@ -69,7 +69,7 @@ class ASRProvider(ASRProviderBase): ) async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """语音转文本主处理逻辑""" retry_count = 0 diff --git a/main/xiaozhi-server/core/providers/asr/fun_server.py b/main/xiaozhi-server/core/providers/asr/fun_server.py index 19b5022d..d577a18e 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_server.py +++ b/main/xiaozhi-server/core/providers/asr/fun_server.py @@ -101,7 +101,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).debug(f"Sent end message: {end_message}") async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """ Convert speech data to text using FunASR. diff --git a/main/xiaozhi-server/core/providers/asr/openai.py b/main/xiaozhi-server/core/providers/asr/openai.py index 7b215589..c917cce5 100644 --- a/main/xiaozhi-server/core/providers/asr/openai.py +++ b/main/xiaozhi-server/core/providers/asr/openai.py @@ -24,7 +24,7 @@ class ASRProvider(ASRProviderBase): def requires_file(self) -> bool: return True - async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None) -> Tuple[Optional[str], Optional[str]]: + async def speech_to_text(self, opus_data: List[bytes], session_id: str, artifacts=None) -> Tuple[Optional[str], Optional[str]]: file_path = None try: if artifacts is None: diff --git a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py index 21438c1c..6444a0a4 100644 --- a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py +++ b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py @@ -41,7 +41,7 @@ class ASRProvider(ASRProviderBase): return True async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" temp_file_path = 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 742c3567..54c981a6 100644 --- a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py +++ b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py @@ -124,7 +124,7 @@ class ASRProvider(ASRProviderBase): return True async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """语音转文本主处理逻辑""" file_path = None diff --git a/main/xiaozhi-server/core/providers/asr/tencent.py b/main/xiaozhi-server/core/providers/asr/tencent.py index d873bd19..76a8867c 100644 --- a/main/xiaozhi-server/core/providers/asr/tencent.py +++ b/main/xiaozhi-server/core/providers/asr/tencent.py @@ -32,7 +32,7 @@ class ASRProvider(ASRProviderBase): os.makedirs(self.output_dir, exist_ok=True) async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if not opus_data: diff --git a/main/xiaozhi-server/core/providers/asr/vosk.py b/main/xiaozhi-server/core/providers/asr/vosk.py index 77cbf986..0871c085 100644 --- a/main/xiaozhi-server/core/providers/asr/vosk.py +++ b/main/xiaozhi-server/core/providers/asr/vosk.py @@ -44,7 +44,7 @@ class ASRProvider(ASRProviderBase): raise async def speech_to_text( - self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None + self, opus_data: List[bytes], session_id: str, artifacts=None ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" try: diff --git a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py index 5667106d..fa59ca34 100644 --- a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py @@ -4,7 +4,6 @@ import base64 import hashlib import asyncio import websockets -import opuslib_next import gc from time import mktime from datetime import datetime @@ -33,7 +32,6 @@ class ASRProvider(ASRProviderBase): self.interface_type = InterfaceType.STREAM self.config = config self.text = "" - self.decoder = opuslib_next.Decoder(16000, 1) self.asr_ws = None self.forward_task = None self.is_processing = False @@ -100,9 +98,9 @@ class ASRProvider(ASRProviderBase): async def open_audio_channels(self, conn: "ConnectionHandler"): await super().open_audio_channels(conn) - async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice): + async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 - await super().receive_audio(conn, audio, audio_have_voice) + await super().receive_audio(conn, pcm_frame, audio_have_voice) # 如果本次有声音,且之前没有建立连接 if audio_have_voice and self.asr_ws is None and not self.is_processing: @@ -116,7 +114,6 @@ class ASRProvider(ASRProviderBase): # 发送当前音频数据 if self.asr_ws and self.is_processing and self.server_ready: try: - pcm_frame = self.decoder.decode(audio, 960) await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME) except Exception as e: logger.bind(tag=TAG).warning(f"发送音频数据时发生错误: {e}") @@ -148,19 +145,15 @@ class ASRProvider(ASRProviderBase): # 发送首帧音频 if conn.asr_audio and len(conn.asr_audio) > 0: - first_audio = conn.asr_audio[-1] if conn.asr_audio else b"" - pcm_frame = ( - self.decoder.decode(first_audio, 960) if first_audio else b"" - ) - await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME) + first_pcm = conn.asr_audio[-1] if conn.asr_audio else b"" + await self._send_audio_frame(first_pcm, STATUS_FIRST_FRAME) self.server_ready = True logger.bind(tag=TAG).info("已发送首帧,开始识别") # 发送缓存的音频数据 - for cached_audio in conn.asr_audio[-10:]: + for cached_pcm in conn.asr_audio[-10:]: try: - pcm_frame = self.decoder.decode(cached_audio, 960) - await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME) + await self._send_audio_frame(cached_pcm, STATUS_CONTINUE_FRAME) except Exception as e: logger.bind(tag=TAG).info(f"发送缓存音频数据时发生错误: {e}") break @@ -322,7 +315,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).debug("ASR会话清理完成") - async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None): + async def speech_to_text(self, opus_data, session_id, artifacts=None): """获取识别结果""" result = self.text self.text = "" @@ -342,12 +335,3 @@ class ASRProvider(ASRProviderBase): self.forward_task = None self.is_processing = False - # 显式释放decoder资源 - if hasattr(self, "decoder") and self.decoder is not None: - try: - del self.decoder - self.decoder = None - logger.bind(tag=TAG).debug("Xunfei decoder resources released") - except Exception as e: - logger.bind(tag=TAG).debug(f"释放Xunfei decoder资源时出错: {e}") - diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 3682f3b8..d134f25d 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -1,7 +1,6 @@ import time import os import numpy as np -import opuslib_next import onnxruntime from config.logger import setup_logging from core.providers.vad.base import VADProviderBase @@ -39,8 +38,6 @@ class VADProvider(VADProviderBase): def _init_connection_state(self, conn): """为连接初始化独立的 VAD 状态""" - if not hasattr(conn, "_vad_opus_decoder"): - conn._vad_opus_decoder = opuslib_next.Decoder(16000, 1) if not hasattr(conn, "_vad_state"): conn._vad_state = np.zeros((2, 1, 128), dtype=np.float32) if not hasattr(conn, "_vad_context"): @@ -48,14 +45,14 @@ class VADProvider(VADProviderBase): def release_conn_resources(self, conn): """释放连接的 VAD 资源(连接关闭时调用)""" - for attr in ("_vad_opus_decoder", "_vad_state", "_vad_context"): + for attr in ("_vad_state", "_vad_context"): if hasattr(conn, attr): try: delattr(conn, attr) except Exception: pass - def is_vad(self, conn, opus_packet): + def is_vad(self, conn, pcm_frame): # 手动模式:直接返回True,不进行实时VAD检测,所有音频都缓存 if conn.client_listen_mode == "manual": return True @@ -63,7 +60,7 @@ class VADProvider(VADProviderBase): try: self._init_connection_state(conn) - pcm_frame = conn._vad_opus_decoder.decode(opus_packet, 960) + # pcm_frame已经是处理后的PCM数据 conn.client_audio_buffer.extend(pcm_frame) client_have_voice = False @@ -115,7 +112,5 @@ class VADProvider(VADProviderBase): conn.vad_last_voice_time = time.time() * 1000 return client_have_voice - except opuslib_next.OpusError as e: - logger.bind(tag=TAG).info(f"解码错误: {e}") except Exception as e: logger.bind(tag=TAG).error(f"Error processing audio packet: {e}")