Merge branch 'main' into py_test_tts

# Conflicts:
#	main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

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

@@ -1,15 +1,18 @@
 import os
 import wave
-import copy
 import uuid
 import queue
 import asyncio
 import traceback
 import threading
 import opuslib_next
+import json
+import io
+import time
+import concurrent.futures
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from typing import Optional, Tuple, List
+from typing import Optional, Tuple, List, Dict, Any
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.utils.util import remove_punctuation_and_length
@@ -24,10 +27,7 @@ class ASRProviderBase(ABC):
         pass
 
     # 打开音频通道
-    # 这里默认是非流式的处理方式
-    # 流式处理方式请在子类中重写
     async def open_audio_channels(self, conn):
-        # tts 消化线程
         conn.asr_priority_thread = threading.Thread(
             target=self.asr_text_priority_thread, args=(conn,), daemon=True
         )
@@ -52,41 +52,171 @@ class ASRProviderBase(ABC):
                 continue
 
     # 接收音频
-    # 这里默认是非流式的处理方式
-    # 流式处理方式请在子类中重写
     async def receive_audio(self, conn, audio, audio_have_voice):
         if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
             have_voice = audio_have_voice
         else:
             have_voice = conn.client_have_voice
-        # 如果本次没有声音，本段也没声音，就把声音丢弃了
+        
         conn.asr_audio.append(audio)
-        if have_voice == False and conn.client_have_voice == False:
+        if not have_voice and not conn.client_have_voice:
             conn.asr_audio = conn.asr_audio[-10:]
             return
 
-        # 如果本段有声音，且已经停止了
         if conn.client_voice_stop:
-            asr_audio_task = copy.deepcopy(conn.asr_audio)
+            asr_audio_task = conn.asr_audio.copy()
             conn.asr_audio.clear()
-
-            # 音频太短了，无法识别
             conn.reset_vad_states()
+
             if len(asr_audio_task) > 15:
                 await self.handle_voice_stop(conn, asr_audio_task)
 
     # 处理语音停止
-    async def handle_voice_stop(self, conn, asr_audio_task):
-        raw_text, _ = await self.speech_to_text(
-            asr_audio_task, conn.session_id, conn.audio_format
-        )  # 确保ASR模块返回原始文本
-        conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
-        text_len, _ = remove_punctuation_and_length(raw_text)
-        self.stop_ws_connection()
-        if text_len > 0:
-            # 使用自定义模块进行上报
-            await startToChat(conn, raw_text)
-            enqueue_asr_report(conn, raw_text, asr_audio_task)
+    async def handle_voice_stop(self, conn, 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)
+            
+            combined_pcm_data = b"".join(pcm_data)
+            
+            # 预先准备WAV数据
+            wav_data = None
+            # 使用连接的声纹识别提供者
+            if conn.voiceprint_provider and combined_pcm_data:
+                wav_data = self._pcm_to_wav(combined_pcm_data)
+            
+            
+            # 定义ASR任务
+            def run_asr():
+                start_time = time.monotonic()
+                try:
+                    loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(loop)
+                    try:
+                        result = loop.run_until_complete(
+                            self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
+                        )
+                        end_time = time.monotonic()
+                        logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
+                        return result
+                    finally:
+                        loop.close()
+                except Exception as e:
+                    end_time = time.monotonic()
+                    logger.bind(tag=TAG).error(f"ASR失败: {e}")
+                    return ("", None)
+            
+            # 定义声纹识别任务
+            def run_voiceprint():
+                if not wav_data:
+                    return None
+                try:
+                    loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(loop)
+                    try:
+                        # 使用连接的声纹识别提供者
+                        result = loop.run_until_complete(
+                            conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
+                        )
+                        return result
+                    finally:
+                        loop.close()
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
+                    return None
+            
+            # 使用线程池执行器并行运行
+            parallel_start_time = time.monotonic()
+            
+            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
+                asr_future = thread_executor.submit(run_asr)
+                
+                if conn.voiceprint_provider and wav_data:
+                    voiceprint_future = thread_executor.submit(run_voiceprint)
+                    
+                    # 等待两个线程都完成
+                    asr_result = asr_future.result(timeout=15)
+                    voiceprint_result = voiceprint_future.result(timeout=15)
+                    
+                    results = {"asr": asr_result, "voiceprint": voiceprint_result}
+                else:
+                    asr_result = asr_future.result(timeout=15)
+                    results = {"asr": asr_result, "voiceprint": None}
+            
+            
+            # 处理结果
+            raw_text, file_path = results.get("asr", ("", None))
+            speaker_name = results.get("voiceprint", None)
+            
+            # 记录识别结果
+            if raw_text:
+                logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
+            if speaker_name:
+                logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
+            
+            # 性能监控
+            total_time = time.monotonic() - total_start_time
+            logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
+            
+            # 检查文本长度
+            text_len, _ = remove_punctuation_and_length(raw_text)
+            self.stop_ws_connection()
+            
+            if text_len > 0:
+                # 构建包含说话人信息的JSON字符串
+                enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
+                
+                # 使用自定义模块进行上报
+                await startToChat(conn, enhanced_text)
+                enqueue_asr_report(conn, enhanced_text, asr_audio_task)
+                
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
+            import traceback
+            logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}")
+
+    def _build_enhanced_text(self, text: str, speaker_name: Optional[str]) -> str:
+        """构建包含说话人信息的文本"""
+        if speaker_name and speaker_name.strip():
+            return json.dumps({
+                "speaker": speaker_name,
+                "content": text
+            }, ensure_ascii=False)
+        else:
+            return text
+
+    def _pcm_to_wav(self, pcm_data: bytes) -> bytes:
+        """将PCM数据转换为WAV格式"""
+        if len(pcm_data) == 0:
+            logger.bind(tag=TAG).warning("PCM数据为空，无法转换WAV")
+            return b""
+        
+        # 确保数据长度是偶数（16位音频）
+        if len(pcm_data) % 2 != 0:
+            pcm_data = pcm_data[:-1]
+        
+        # 创建WAV文件头
+        wav_buffer = io.BytesIO()
+        try:
+            with wave.open(wav_buffer, 'wb') as wav_file:
+                wav_file.setnchannels(1)      # 单声道
+                wav_file.setsampwidth(2)      # 16位
+                wav_file.setframerate(16000)  # 16kHz采样率
+                wav_file.writeframes(pcm_data)
+            
+            wav_buffer.seek(0)
+            wav_data = wav_buffer.read()
+            
+            return wav_data
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"WAV转换失败: {e}")
+            return b""
 
     def stop_ws_connection(self):
         pass
@@ -113,27 +243,29 @@ class ASRProviderBase(ABC):
         pass
 
     @staticmethod
-    def decode_opus(opus_data: List[bytes]) -> bytes:
+    def decode_opus(opus_data: List[bytes]) -> List[bytes]:
         """将Opus音频数据解码为PCM数据"""
         try:
-            decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
+            decoder = opuslib_next.Decoder(16000, 1)
             pcm_data = []
-            buffer_size = 960  # 每次处理960个采样点
-
-            for opus_packet in opus_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:
+                    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解码错误，跳过当前数据包: {e}")
-                    continue
+                    logger.bind(tag=TAG).warning(f"Opus解码错误，跳过数据包 {i}: {e}")
                 except Exception as e:
-                    logger.bind(tag=TAG).error(f"音频处理错误: {e}", exc_info=True)
-                    continue
-
+                    logger.bind(tag=TAG).error(f"音频处理错误，数据包 {i}: {e}")
+            
             return pcm_data
+            
         except Exception as e:
-            logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}", exc_info=True)
+            logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}")
             return []

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

@@ -56,6 +56,16 @@ class ASRProvider(ASRProviderBase):
     async def receive_audio(self, conn, audio, audio_have_voice):
         conn.asr_audio.append(audio)
         conn.asr_audio = conn.asr_audio[-10:]
+        
+        # 存储音频数据
+        if not hasattr(conn, 'asr_audio_for_voiceprint'):
+            conn.asr_audio_for_voiceprint = []
+        conn.asr_audio_for_voiceprint.append(audio)
+        
+        # 当没有音频数据时处理完整语音片段
+        if not audio and len(conn.asr_audio_for_voiceprint) > 0:
+            await self.handle_voice_stop(conn, conn.asr_audio_for_voiceprint)
+            conn.asr_audio_for_voiceprint = []
 
         # 如果本次有声音，且之前没有建立连接
         if audio_have_voice and self.asr_ws is None and not self.is_processing:
@@ -148,6 +158,8 @@ class ASRProvider(ASRProviderBase):
     async def _forward_asr_results(self, conn):
         try:
             while self.asr_ws and not conn.stop_event.is_set():
+                # 获取当前连接的音频数据
+                audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
                 try:
                     response = await self.asr_ws.recv()
                     result = self.parse_response(response)
@@ -171,7 +183,8 @@ class ASRProvider(ASRProviderBase):
                                 logger.bind(tag=TAG).error(f"识别文本：空")
                                 self.text = ""
                                 conn.reset_vad_states()
-                                await self.handle_voice_stop(conn, None)
+                                if len(audio_data) > 15:  # 确保有足够音频数据
+                                    await self.handle_voice_stop(conn, audio_data)
                                 break
 
                             for utterance in utterances:
@@ -181,7 +194,8 @@ class ASRProvider(ASRProviderBase):
                                         f"识别到文本: {self.text}"
                                     )
                                     conn.reset_vad_states()
-                                    await self.handle_voice_stop(conn, None)
+                                    if len(audio_data) > 15:  # 确保有足够音频数据
+                                        await self.handle_voice_stop(conn, audio_data)
                                     break
                         elif "error" in payload:
                             error_msg = payload.get("error", "未知错误")
@@ -208,6 +222,13 @@ class ASRProvider(ASRProviderBase):
                 await self.asr_ws.close()
                 self.asr_ws = None
             self.is_processing = False
+            if conn:
+                if hasattr(conn, 'asr_audio_for_voiceprint'):
+                    conn.asr_audio_for_voiceprint = []
+                if hasattr(conn, 'asr_audio'):
+                    conn.asr_audio = []
+                if hasattr(conn, 'has_valid_voice'):
+                    conn.has_valid_voice = False
 
     def stop_ws_connection(self):
         if self.asr_ws:
@@ -349,3 +370,12 @@ class ASRProvider(ASRProviderBase):
                 pass
             self.forward_task = None
         self.is_processing = False
+        # 清理所有连接的音频缓冲区
+        if hasattr(self, '_connections'):
+            for conn in self._connections.values():
+                if hasattr(conn, 'asr_audio_for_voiceprint'):
+                    conn.asr_audio_for_voiceprint = []
+                if hasattr(conn, 'asr_audio'):
+                    conn.asr_audio = []
+                if hasattr(conn, 'has_valid_voice'):
+                    conn.has_valid_voice = False

main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py

@@ -16,10 +16,11 @@ class IntentProvider(IntentProviderBase):
         super().__init__(config)
         self.llm = None
         self.promot = ""
-        # 添加缓存管理
-        self.intent_cache = {}  # 缓存意图识别结果
-        self.cache_expiry = 600  # 缓存有效期10分钟
-        self.cache_max_size = 100  # 最多缓存100个意图
+        # 导入全局缓存管理器
+        from core.utils.cache.manager import cache_manager, CacheType
+
+        self.cache_manager = cache_manager
+        self.CacheType = CacheType
         self.history_count = 4  # 默认使用最近4条对话记录
 
     def get_intent_system_prompt(self, functions_list: str) -> str:
@@ -102,27 +103,6 @@ class IntentProvider(IntentProviderBase):
         )
         return prompt
 
-    def clean_cache(self):
-        """清理过期缓存"""
-        now = time.time()
-        # 找出过期键
-        expired_keys = [
-            k
-            for k, v in self.intent_cache.items()
-            if now - v["timestamp"] > self.cache_expiry
-        ]
-        for key in expired_keys:
-            del self.intent_cache[key]
-
-        # 如果缓存太大，移除最旧的条目
-        if len(self.intent_cache) > self.cache_max_size:
-            # 按时间戳排序并保留最新的条目
-            sorted_items = sorted(
-                self.intent_cache.items(), key=lambda x: x[1]["timestamp"]
-            )
-            for key, _ in sorted_items[: len(sorted_items) - self.cache_max_size]:
-                del self.intent_cache[key]
-
     def replyResult(self, text: str, original_text: str):
         llm_result = self.llm.response_no_stream(
             system_prompt=text,
@@ -145,21 +125,16 @@ class IntentProvider(IntentProviderBase):
         logger.bind(tag=TAG).debug(f"使用意图识别模型: {model_info}")
 
         # 计算缓存键
-        cache_key = hashlib.md5(text.encode()).hexdigest()
+        cache_key = hashlib.md5((conn.device_id + text).encode()).hexdigest()
 
         # 检查缓存
-        if cache_key in self.intent_cache:
-            cache_entry = self.intent_cache[cache_key]
-            # 检查缓存是否过期
-            if time.time() - cache_entry["timestamp"] <= self.cache_expiry:
-                cache_time = time.time() - total_start_time
-                logger.bind(tag=TAG).debug(
-                    f"使用缓存的意图: {cache_key} -> {cache_entry['intent']}, 耗时: {cache_time:.4f}秒"
-                )
-                return cache_entry["intent"]
-
-        # 清理缓存
-        self.clean_cache()
+        cached_intent = self.cache_manager.get(self.CacheType.INTENT, cache_key)
+        if cached_intent is not None:
+            cache_time = time.time() - total_start_time
+            logger.bind(tag=TAG).debug(
+                f"使用缓存的意图: {cache_key} -> {cached_intent}, 耗时: {cache_time:.4f}秒"
+            )
+            return cached_intent
 
         if self.promot == "":
             functions = conn.func_handler.get_functions()
@@ -259,10 +234,7 @@ class IntentProvider(IntentProviderBase):
                     conn.dialogue.dialogue = clean_history
 
                 # 添加到缓存
-                self.intent_cache[cache_key] = {
-                    "intent": intent,
-                    "timestamp": time.time(),
-                }
+                self.cache_manager.set(self.CacheType.INTENT, cache_key, intent)
 
                 # 后处理时间
                 postprocess_time = time.time() - postprocess_start_time
@@ -272,10 +244,7 @@ class IntentProvider(IntentProviderBase):
                 return intent
             else:
                 # 添加到缓存
-                self.intent_cache[cache_key] = {
-                    "intent": intent,
-                    "timestamp": time.time(),
-                }
+                self.cache_manager.set(self.CacheType.INTENT, cache_key, intent)
 
                 # 后处理时间
                 postprocess_time = time.time() - postprocess_start_time

main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py

@@ -79,10 +79,11 @@ short_term_memory_prompt_only_content = """
 1、总结user的重要信息，以便在未来的对话中提供更个性化的服务
 2、不要重复总结，不要遗忘之前记忆，除非原来的记忆超过了1800字内，否则不要遗忘、不要压缩用户的历史记忆
 3、用户操控的设备音量、播放音乐、天气、退出、不想对话等和用户本身无关的内容，这些信息不需要加入到总结中
-4、不要把设备操控的成果结果和失败结果加入到总结中，也不要把用户的一些废话加入到总结中
-5、不要为了总结而总结，如果用户的聊天没有意义，请返回原来的历史记录也是可以的
-6、只需要返回总结摘要，严格控制在1800字内
-7、不要包含代码、xml，不需要解释、注释和说明，保存记忆时仅从对话提取信息，不要混入示例内容
+4、聊天内容中的今天的日期时间、今天的天气情况与用户事件无关的数据，这些信息如果当成记忆存储会影响后序对话，这些信息不需要加入到总结中
+5、不要把设备操控的成果结果和失败结果加入到总结中，也不要把用户的一些废话加入到总结中
+6、不要为了总结而总结，如果用户的聊天没有意义，请返回原来的历史记录也是可以的
+7、只需要返回总结摘要，严格控制在1800字内
+8、不要包含代码、xml，不需要解释、注释和说明，保存记忆时仅从对话提取信息，不要混入示例内容
 """
 
 

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -51,13 +51,13 @@ class ServerPluginExecutor(ToolExecutor):
         tools = {}
 
         # 获取必要的函数
-        necessary_functions = ["handle_exit_intent", "get_time", "get_lunar"]
+        necessary_functions = ["handle_exit_intent", "get_lunar"]
 
         # 获取配置中的函数
         config_functions = self.config["Intent"][
             self.config["selected_module"]["Intent"]
         ].get("functions", [])
-        
+
         # 转换为列表
         if not isinstance(config_functions, list):
             try: