add: ASR测试任务

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
+                    field in config
-                and any(x in config[field] for x in ["你的", "placeholder"])
+                    and any(x in config[field] for x in ["你的", "placeholder"])
-                for field in token_fields
+                    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:
-                # 计算相对性能分数（越小越好）
+                for stt in valid_stt:
-                llm_score = (
+                    # 计算相对性能分数（越小越好）
-                    self.results["llm"][llm]["avg_first_token"] / min_first_token
+                    llm_score = (
-                )
+                            self.results["llm"][llm]["avg_first_token"] / min_first_token
-                tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time
+                    )
+                    tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time
+                    stt_score = self.results["stt"][stt]["avg_time"] / min_stt_time
 
-                # 计算稳定性分数（标准差/平均值，越小越稳定）
+                    # 计算稳定性分数（标准差/平均值，越小越稳定）
-                llm_stability = (
+                    llm_stability = (
-                    self.results["llm"][llm]["std_first_token"]
+                            self.results["llm"][llm]["std_first_token"]
-                    / self.results["llm"][llm]["avg_first_token"]
+                            / self.results["llm"][llm]["avg_first_token"]
-                )
+                    )
 
-                # 综合得分（考虑性能和稳定性）
+                    # 综合得分（考虑性能和稳定性）
-                # 性能权重0.7，稳定性权重0.3
+                    # LLM得分： 性能权重(70%) + 稳定性权重(30%)
-                llm_final_score = llm_score * 0.7 + llm_stability * 0.3
+                    llm_final_score = llm_score * 0.7 + llm_stability * 0.3
 
-                # 总分 = LLM得分(70%) + TTS得分(30%)
+                    # 总分 = LLM得分(70%) + TTS得分(30%) + STT得分(30%)
-                total_score = llm_final_score * 0.7 + tts_score * 0.3
+                    total_score = llm_final_score * 0.7 + tts_score * 0.3 + stt_score * 0.3
 
-                self.results["combinations"].append(
+                    self.results["combinations"].append(
-                    {
+                        {
-                        "llm": llm,
+                            "llm": llm,
-                        "tts": tts,
+                            "tts": tts,
-                        "score": total_score,
+                            "stt": stt,
-                        "details": {
+                            "score": total_score,
-                            "llm_first_token": self.results["llm"][llm][
+                            "details": {
-                                "avg_first_token"
+                                "llm_first_token": self.results["llm"][llm][
-                            ],
+                                    "avg_first_token"
-                            "llm_stability": llm_stability,
+                                ],
-                            "tts_time": self.results["tts"][tts]["avg_time"],
+                                "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 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(
+                if llm_name == "CozeLLM":
-                    x in config.get("user_id", "") for x in ["你的"]
+                    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，已跳过")
+                    print(f"⏭️  LLM {llm_name} 未配置api_key，已跳过")
-                    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")
                     continue
 
-                if not await self._check_ollama_service(base_url, model_name):
+                # 对于Ollama，先检查服务状态
-                    continue
+                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}")
+                    if not await self._check_ollama_service(base_url, model_name):
-            module_type = config.get("type", llm_name)
+                        continue
-            llm = create_llm_instance(module_type, config)
 
-            # 为每个句子创建独立任务
+                print(f"📋 添加LLM测试任务: {llm_name}")
-            for sentence in self.test_sentences:
+                module_type = config.get("type", llm_name)
-                sentence = sentence.encode("utf-8").decode("utf-8")
+                llm = create_llm_instance(module_type, config)
-                all_tasks.append(self._test_single_sentence(llm_name, llm, sentence))
+
+                # 为每个句子创建独立任务
+                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():
+        if self.config.get("TTS") is not None:
-            token_fields = ["access_token", "api_key", "token"]
+            for tts_name, config in self.config.get("TTS", {}).items():
-            if any(
+                token_fields = ["access_token", "api_key", "token"]
-                field in config
+                if any(
-                and any(x in config[field] for x in ["你的", "placeholder"])
+                        field in config
-                for field in token_fields
+                        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} 未配置access_token/api_key，已跳过")
-            print(f"🎵 添加TTS测试任务: {tts_name}")
+                    continue
-            all_tasks.append(self._test_tts(tts_name, config))
+                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()