Merge pull request #2584 from xinnan-tech/py_test_websocket

fix: decoder 引发的内存泄漏

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

@@ -10,7 +10,7 @@ TTS上报功能已集成到ConnectionHandler类中。
 """
 
 import time
-
+import gc
 import opuslib_next
 
 from config.manage_api_client import report as manage_report
@@ -56,6 +56,8 @@ def opus_to_wav(conn, opus_data):
     Returns:
         bytes: WAV格式的音频数据
     """
+    decoder = None
+    try:
         decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
         pcm_data = []
 
@@ -91,6 +93,13 @@ def opus_to_wav(conn, opus_data):
 
         # 返回完整的WAV数据
         return bytes(wav_header) + pcm_data_bytes
+    finally:
+        if decoder is not None:
+            try:
+                del decoder
+                gc.collect()
+            except Exception as e:
+                conn.logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")
 
 
 def enqueue_tts_report(conn, text, opus_data):

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

@@ -5,6 +5,7 @@ import hmac
 import base64
 import hashlib
 import asyncio
+import gc
 import requests
 import websockets
 import opuslib_next
@@ -347,4 +348,12 @@ class ASRProvider(ASRProviderBase):
 
     async def close(self):
         """关闭资源"""
-        await self._cleanup()
+        await self._cleanup(None)
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+                self.decoder = None
+                gc.collect()
+                logger.bind(tag=TAG).debug("Aliyun decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Aliyun decoder资源时出错: {e}")

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

@@ -10,6 +10,7 @@ import traceback
 import threading
 import opuslib_next
 import concurrent.futures
+import gc
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
@@ -241,6 +242,7 @@ class ASRProviderBase(ABC):
     @staticmethod
     def decode_opus(opus_data: List[bytes]) -> List[bytes]:
         """将Opus音频数据解码为PCM数据"""
+        decoder = None
         try:
             decoder = opuslib_next.Decoder(16000, 1)
             pcm_data = []
@@ -265,3 +267,10 @@ class ASRProviderBase(ABC):
         except Exception as e:
             logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}")
             return []
+        finally:
+            if decoder is not None:
+                try:
+                    del decoder
+                    gc.collect()
+                except Exception as e:
+                    logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")

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

@@ -4,6 +4,7 @@ import uuid
 import asyncio
 import websockets
 import opuslib_next
+import gc
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
 from core.providers.asr.dto.dto import InterfaceType
@@ -370,6 +371,17 @@ class ASRProvider(ASRProviderBase):
                 pass
             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
+                gc.collect()
+                logger.bind(tag=TAG).debug("Doubao decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Doubao decoder资源时出错: {e}")
+
         # 清理所有连接的音频缓冲区
         if hasattr(self, '_connections'):
             for conn in self._connections.values():

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

@@ -5,6 +5,7 @@ import hashlib
 import asyncio
 import websockets
 import opuslib_next
+import gc
 from time import mktime
 from datetime import datetime
 from urllib.parse import urlencode
@@ -512,6 +513,17 @@ class ASRProvider(ASRProviderBase):
                 pass
             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
+                gc.collect()
+                logger.bind(tag=TAG).debug("Xunfei decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Xunfei decoder资源时出错: {e}")
+
         # 清理所有连接的音频缓冲区
         if hasattr(self, "_connections"):
             for conn in self._connections.values():

main/xiaozhi-server/core/providers/vad/silero.py

@@ -2,6 +2,7 @@ import time
 import numpy as np
 import torch
 import opuslib_next
+import gc
 from config.logger import setup_logging
 from core.providers.vad.base import VADProviderBase
 
@@ -36,6 +37,14 @@ class VADProvider(VADProviderBase):
         # 至少要多少帧才算有语音
         self.frame_window_threshold = 3
 
+    def __del__(self):
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+                gc.collect()
+            except Exception:
+                pass
+
     def is_vad(self, conn, opus_packet):
         try:
             pcm_frame = self.decoder.decode(opus_packet, 960)

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

@@ -6,6 +6,7 @@ Opus编码工具类
 import logging
 import traceback
 import numpy as np
+import gc
 from opuslib_next import Encoder
 from opuslib_next import constants
 from typing import Optional, Callable, Any
@@ -128,5 +129,10 @@ class OpusEncoderUtils:
 
     def close(self):
         """关闭编码器并释放资源"""
-        # opuslib没有明确的关闭方法，Python的垃圾回收会处理
+        if hasattr(self, 'encoder') and self.encoder:
-        pass
+            try:
+                del self.encoder
+                self.encoder = None
+                gc.collect()
+            except Exception as e:
+                logging.error(f"Error releasing Opus encoder: {e}")

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

@@ -8,6 +8,7 @@ import requests
 import subprocess
 import numpy as np
 import opuslib_next
+import gc
 from io import BytesIO
 from core.utils import p3
 from pydub import AudioSegment
@@ -372,6 +373,7 @@ def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
     将opus帧列表解码为wav字节流
     """
     decoder = opuslib_next.Decoder(sample_rate, channels)
+    try:
         pcm_datas = []
 
         frame_duration = 60  # ms
@@ -392,6 +394,13 @@ def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
             wf.setframerate(sample_rate)
             wf.writeframes(pcm_bytes)
         return wav_buffer.getvalue()
+    finally:
+        if decoder is not None:
+            try:
+                del decoder
+                gc.collect()
+            except Exception:
+                pass
 
 
 def check_vad_update(before_config, new_config):