From 86f5be23fe1ffa88a8c01201204393a5f224c538 Mon Sep 17 00:00:00 2001 From: QJH Date: Thu, 15 May 2025 17:01:58 +0800 Subject: [PATCH] =?UTF-8?q?add:=20ASR=E6=B5=8B=E8=AF=95=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/performance_tester.py | 309 +++++++++++++++------- 1 file changed, 218 insertions(+), 91 deletions(-) diff --git a/main/xiaozhi-server/performance_tester.py b/main/xiaozhi-server/performance_tester.py index 38daea00..bccbf7e6 100644 --- a/main/xiaozhi-server/performance_tester.py +++ b/main/xiaozhi-server/performance_tester.py @@ -1,15 +1,17 @@ -import time -import aiohttp import asyncio +import logging +import os +import statistics +import time +from typing import Dict + +import aiohttp from tabulate import tabulate -from typing import Dict, List + +from config.settings import load_config +from core.utils.asr import create_instance as create_stt_instance from core.utils.llm import create_instance as create_llm_instance from core.utils.tts import create_instance as create_tts_instance -import statistics -from config.settings import load_config -import inspect -import os -import logging # 设置全局日志级别为WARNING，抑制INFO级别日志 logging.basicConfig(level=logging.WARNING) @@ -26,7 +28,15 @@ class AsyncPerformanceTester: "请用100字概括量子计算的基本原理和应用前景", ], ) - self.results = {"llm": {}, "tts": {}, "combinations": []} + + self.test_wav_list = [] + self.wav_root = r".\config\assets\test_asr" + os.makedirs(self.wav_root, exist_ok=True) + for wav_name in os.listdir(self.wav_root): + with open(os.path.join(self.wav_root, wav_name), "rb") as f: + self.test_wav_list.append(f.read()) + + self.results = {"llm": {}, "tts": {}, "stt": {}, "combinations": []} async def _check_ollama_service(self, base_url: str, model_name: str) -> bool: """异步检查Ollama服务状态""" @@ -63,9 +73,9 @@ class AsyncPerformanceTester: token_fields = ["access_token", "api_key", "token"] if any( - field in config - and any(x in config[field] for x in ["你的", "placeholder"]) - for field in token_fields + field in config + and any(x in config[field] for x in ["你的", "placeholder"]) + for field in token_fields ): print(f"⏭️ TTS {tts_name} 未配置access_token/api_key，已跳过") return {"name": tts_name, "type": "tts", "errors": 1} @@ -109,6 +119,57 @@ class AsyncPerformanceTester: print(f"⚠️ {tts_name} 测试失败: {str(e)}") return {"name": tts_name, "type": "tts", "errors": 1} + async def _test_stt(self, stt_name: str, config: Dict) -> Dict: + """异步测试单个STT性能""" + try: + logging.getLogger("core.providers.asr.base").setLevel(logging.WARNING) + token_fields = ["access_token", "api_key", "token"] + if any( + field in config + and any(x in config[field] for x in ["你的", "placeholder"]) + for field in token_fields + ): + print(f"⏭️ STT {stt_name} 未配置access_token/api_key，已跳过") + return {"name": stt_name, "type": "stt", "errors": 1} + + module_type = config.get("type", stt_name) + stt = create_stt_instance(module_type, config, delete_audio_file=True) + stt.audio_format = "pcm" + + print(f"🎵 测试 STT: {stt_name}") + + text, _ = await stt.speech_to_text([self.test_wav_list[0]], "1") + + if text is None: + print(f"❌ {stt_name} 连接失败") + return {"name": stt_name, "type": "stt", "errors": 1} + + total_time = 0 + test_count = len(self.test_sentences[:2]) + + for i, sentence in enumerate(self.test_wav_list, 1): + start = time.time() + text, _ = await stt.speech_to_text([sentence], "1") + duration = time.time() - start + total_time += duration + + if text: + print(f"✓ {stt_name} [{i}/{test_count}]") + else: + print(f"✗ {stt_name} [{i}/{test_count}]") + return {"name": stt_name, "type": "stt", "errors": 1} + + return { + "name": stt_name, + "type": "stt", + "avg_time": total_time / test_count, + "errors": 0, + } + + except Exception as e: + print(f"⚠️ {stt_name} 测试失败: {str(e)}") + return {"name": stt_name, "type": "stt", "errors": 1} + async def _test_llm(self, llm_name: str, config: Dict) -> Dict: """异步测试单个LLM性能""" try: @@ -124,7 +185,7 @@ class AsyncPerformanceTester: return {"name": llm_name, "type": "llm", "errors": 1} else: if "api_key" in config and any( - x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"] + x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"] ): print(f"🚫 跳过未配置的LLM: {llm_name}") return {"name": llm_name, "type": "llm", "errors": 1} @@ -196,7 +257,7 @@ class AsyncPerformanceTester: async def process_response(): nonlocal first_token_received, first_token_time for chunk in llm.response( - "perf_test", [{"role": "user", "content": sentence}] + "perf_test", [{"role": "user", "content": sentence}] ): if not first_token_received and chunk.strip() != "": first_token_time = time.time() - sentence_start @@ -234,6 +295,7 @@ class AsyncPerformanceTester: if v["errors"] == 0 and v["avg_first_token"] >= 0.05 ] valid_tts = [k for k, v in self.results["tts"].items() if v["errors"] == 0] + valid_stt = [k for k, v in self.results["stt"].items() if v["errors"] == 0] # 找出基准值 min_first_token = ( @@ -246,42 +308,51 @@ class AsyncPerformanceTester: if valid_tts else 1 ) + min_stt_time = ( + min([self.results["stt"][stt]["avg_time"] for stt in valid_stt]) + if valid_stt + else 1 + ) for llm in valid_llms: for tts in valid_tts: - # 计算相对性能分数（越小越好） - llm_score = ( - self.results["llm"][llm]["avg_first_token"] / min_first_token - ) - tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time + for stt in valid_stt: + # 计算相对性能分数（越小越好） + llm_score = ( + self.results["llm"][llm]["avg_first_token"] / min_first_token + ) + tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time + stt_score = self.results["stt"][stt]["avg_time"] / min_stt_time - # 计算稳定性分数（标准差/平均值，越小越稳定） - llm_stability = ( - self.results["llm"][llm]["std_first_token"] - / self.results["llm"][llm]["avg_first_token"] - ) + # 计算稳定性分数（标准差/平均值，越小越稳定） + llm_stability = ( + self.results["llm"][llm]["std_first_token"] + / self.results["llm"][llm]["avg_first_token"] + ) - # 综合得分（考虑性能和稳定性） - # 性能权重0.7，稳定性权重0.3 - llm_final_score = llm_score * 0.7 + llm_stability * 0.3 + # 综合得分（考虑性能和稳定性） + # LLM得分： 性能权重(70%) + 稳定性权重(30%) + llm_final_score = llm_score * 0.7 + llm_stability * 0.3 - # 总分 = LLM得分(70%) + TTS得分(30%) - total_score = llm_final_score * 0.7 + tts_score * 0.3 + # 总分 = LLM得分(70%) + TTS得分(30%) + STT得分(30%) + total_score = llm_final_score * 0.7 + tts_score * 0.3 + stt_score * 0.3 - self.results["combinations"].append( - { - "llm": llm, - "tts": tts, - "score": total_score, - "details": { - "llm_first_token": self.results["llm"][llm][ - "avg_first_token" - ], - "llm_stability": llm_stability, - "tts_time": self.results["tts"][tts]["avg_time"], - }, - } - ) + self.results["combinations"].append( + { + "llm": llm, + "tts": tts, + "stt": stt, + "score": total_score, + "details": { + "llm_first_token": self.results["llm"][llm][ + "avg_first_token" + ], + "llm_stability": llm_stability, + "tts_time": self.results["tts"][tts]["avg_time"], + "stt_time": self.results["stt"][stt]["avg_time"], + }, + } + ) # 分数越小越好 self.results["combinations"].sort(key=lambda x: x["score"]) @@ -302,7 +373,7 @@ class AsyncPerformanceTester: ) if llm_table: - print("\nLLM 性能排行:") + print("\nLLM 性能排行:\n") print( tabulate( llm_table, @@ -321,7 +392,7 @@ class AsyncPerformanceTester: tts_table.append([name, f"{data['avg_time']:.3f}秒"]) # 不需要固定宽度 if tts_table: - print("\nTTS 性能排行:") + print("\nTTS 性能排行:\n") print( tabulate( tts_table, @@ -334,17 +405,37 @@ class AsyncPerformanceTester: else: print("\n⚠️ 没有可用的TTS模块进行测试。") + stt_table = [] + for name, data in self.results["stt"].items(): + if data["errors"] == 0: + stt_table.append([name, f"{data['avg_time']:.3f}秒"]) # 不需要固定宽度 + + if stt_table: + print("\nSTT 性能排行:\n") + print( + tabulate( + stt_table, + headers=["模型名称", "合成耗时"], + tablefmt="github", + colalign=("left", "right"), + disable_numparse=True, + ) + ) + else: + print("\n⚠️ 没有可用的STT模块进行测试。") + if self.results["combinations"]: - print("\n推荐配置组合 (得分越小越好):") + print("\n推荐配置组合 (得分越小越好):\n") combo_table = [] - for combo in self.results["combinations"][:5]: + for combo in self.results["combinations"][:]: combo_table.append( [ - f"{combo['llm']} + {combo['tts']}", # 不需要固定宽度 + f"{combo['llm']} + {combo['tts']} + {combo['stt']}", # 不需要固定宽度 f"{combo['score']:.3f}", f"{combo['details']['llm_first_token']:.3f}秒", f"{combo['details']['llm_stability']:.3f}", f"{combo['details']['tts_time']:.3f}秒", + f"{combo['details']['stt_time']:.3f}秒", ] ) @@ -357,9 +448,10 @@ class AsyncPerformanceTester: "LLM首字耗时", "稳定性", "TTS合成耗时", + "STT合成耗时", ], tablefmt="github", - colalign=("left", "right", "right", "right", "right"), + colalign=("left", "right", "right", "right", "right", "right"), disable_numparse=True, ) ) @@ -372,8 +464,12 @@ class AsyncPerformanceTester: if result["errors"] == 0: if result["type"] == "llm": self.results["llm"][result["name"]] = result - else: + elif result["type"] == "tts": self.results["tts"][result["name"]] = result + elif result["type"] == "stt": + self.results["stt"][result["name"]] = result + else: + pass async def run(self): """执行全量异步测试""" @@ -383,52 +479,71 @@ class AsyncPerformanceTester: all_tasks = [] # LLM测试任务 - for llm_name, config in self.config.get("LLM", {}).items(): - # 检查配置有效性 - if llm_name == "CozeLLM": - if any(x in config.get("bot_id", "") for x in ["你的"]) or any( - x in config.get("user_id", "") for x in ["你的"] + if self.config.get("LLM") is not None: + for llm_name, config in self.config.get("LLM", {}).items(): + # 检查配置有效性 + if llm_name == "CozeLLM": + if any(x in config.get("bot_id", "") for x in ["你的"]) or any( + x in config.get("user_id", "") for x in ["你的"] + ): + print(f"⏭️ LLM {llm_name} 未配置bot_id/user_id，已跳过") + continue + elif "api_key" in config and any( + x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"] ): - print(f"⏭️ LLM {llm_name} 未配置bot_id/user_id，已跳过") - continue - elif "api_key" in config and any( - x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"] - ): - print(f"⏭️ LLM {llm_name} 未配置api_key，已跳过") - continue - - # 对于Ollama，先检查服务状态 - if llm_name == "Ollama": - base_url = config.get("base_url", "http://localhost:11434") - model_name = config.get("model_name") - if not model_name: - print(f"🚫 Ollama未配置model_name") + print(f"⏭️ LLM {llm_name} 未配置api_key，已跳过") continue - if not await self._check_ollama_service(base_url, model_name): - continue + # 对于Ollama，先检查服务状态 + if llm_name == "Ollama": + base_url = config.get("base_url", "http://localhost:11434") + model_name = config.get("model_name") + if not model_name: + print(f"🚫 Ollama未配置model_name") + continue - print(f"📋 添加LLM测试任务: {llm_name}") - module_type = config.get("type", llm_name) - llm = create_llm_instance(module_type, config) + if not await self._check_ollama_service(base_url, model_name): + continue - # 为每个句子创建独立任务 - for sentence in self.test_sentences: - sentence = sentence.encode("utf-8").decode("utf-8") - all_tasks.append(self._test_single_sentence(llm_name, llm, sentence)) + print(f"📋 添加LLM测试任务: {llm_name}") + module_type = config.get("type", llm_name) + llm = create_llm_instance(module_type, config) + + # 为每个句子创建独立任务 + for sentence in self.test_sentences: + sentence = sentence.encode("utf-8").decode("utf-8") + all_tasks.append(self._test_single_sentence(llm_name, llm, sentence)) # TTS测试任务 - for tts_name, config in self.config.get("TTS", {}).items(): - token_fields = ["access_token", "api_key", "token"] - if any( - field in config - and any(x in config[field] for x in ["你的", "placeholder"]) - for field in token_fields - ): - print(f"⏭️ TTS {tts_name} 未配置access_token/api_key，已跳过") - continue - print(f"🎵 添加TTS测试任务: {tts_name}") - all_tasks.append(self._test_tts(tts_name, config)) + if self.config.get("TTS") is not None: + for tts_name, config in self.config.get("TTS", {}).items(): + token_fields = ["access_token", "api_key", "token"] + if any( + field in config + and any(x in config[field] for x in ["你的", "placeholder"]) + for field in token_fields + ): + print(f"⏭️ TTS {tts_name} 未配置access_token/api_key，已跳过") + continue + print(f"🎵 添加TTS测试任务: {tts_name}") + all_tasks.append(self._test_tts(tts_name, config)) + + # STT测试任务 + if len(self.test_wav_list) >= 1: + if self.config.get("ASR") is not None: + for stt_name, config in self.config.get("ASR", {}).items(): + token_fields = ["access_token", "api_key", "token"] + if any( + field in config + and any(x in config[field] for x in ["你的", "placeholder"]) + for field in token_fields + ): + print(f"⏭️ ASR {stt_name} 未配置access_token/api_key，已跳过") + continue + print(f"🎵 添加ASR测试任务: {stt_name}") + all_tasks.append(self._test_stt(stt_name, config)) + else: + print(f"\n⚠️ {self.wav_root} 路径下没有音频文件，已跳过STT测试任务") print( f"\n✅ 找到 {len([t for t in all_tasks if 'test_single_sentence' in str(t)]) / len(self.test_sentences):.0f} 个可用LLM模块" @@ -436,6 +551,9 @@ class AsyncPerformanceTester: print( f"✅ 找到 {len([t for t in all_tasks if '_test_tts' in str(t)])} 个可用TTS模块" ) + print( + f"✅ 找到 {len([t for t in all_tasks if '_test_stt' in str(t)])} 个可用STT模块" + ) print("\n⏳ 开始并发测试所有模块...\n") # 并发执行所有测试任务 @@ -469,9 +587,9 @@ class AsyncPerformanceTester: "name": llm_name, "type": "llm", "avg_response": sum(data["response_times"]) - / len(data["response_times"]), + / len(data["response_times"]), "avg_first_token": sum(data["first_token_times"]) - / len(data["first_token_times"]), + / len(data["first_token_times"]), "std_first_token": ( statistics.stdev(data["first_token_times"]) if len(data["first_token_times"]) > 1 @@ -494,6 +612,15 @@ class AsyncPerformanceTester: if result["errors"] == 0: self.results["tts"][result["name"]] = result + # 处理STT结果 + for result in [ + r + for r in all_results + if r and isinstance(r, dict) and r.get("type") == "stt" + ]: + if result["errors"] == 0: + self.results["stt"][result["name"]] = result + # 生成组合建议并打印结果 print("\n📊 生成测试报告...") self._generate_combinations()