update:更新流式ASR首词延迟测试工具

main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py

@@ -5,104 +5,116 @@ import uuid
 import os
 import websockets
 import gzip
-import hmac
-import base64
-import hashlib
 import random
 from urllib import parse
 from tabulate import tabulate
 from config.settings import load_config
-description = "流式ASR首词耗时测试"
+import tempfile
+import wave
+description = "流式ASR首词延迟测试"
+try:
+    import dashscope
+except ImportError:
+    dashscope = None
 
-class AccessToken:
+class BaseASRTester:
-    @staticmethod
+    def __init__(self, config_key: str):
-    def _encode_text(text):
-        encoded_text = parse.quote_plus(text)
-        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
-
-    @staticmethod
-    def _encode_dict(dic):
-        keys = dic.keys()
-        dic_sorted = [(key, dic[key]) for key in sorted(keys)]
-        encoded_text = parse.urlencode(dic_sorted)
-        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
-
-    @staticmethod
-    def create_token(access_key_id, access_key_secret):
-        parameters = {
-            "AccessKeyId": access_key_id,
-            "Action": "CreateToken",
-            "Format": "JSON",
-            "RegionId": "cn-shanghai",
-            "SignatureMethod": "HMAC-SHA1",
-            "SignatureNonce": str(uuid.uuid1()),
-            "SignatureVersion": "1.0",
-            "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-            "Version": "2019-02-28",
-        }
-        query_string = AccessToken._encode_dict(parameters)
-        string_to_sign = (
-            "GET" + "&" + AccessToken._encode_text("/") + "&" + AccessToken._encode_text(query_string)
-        )
-        secreted_string = hmac.new(
-            bytes(access_key_secret + "&", encoding="utf-8"),
-            bytes(string_to_sign, encoding="utf-8"),
-            hashlib.sha1,
-        ).digest()
-        signature = base64.b64encode(secreted_string)
-        signature = AccessToken._encode_text(signature)
-        full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (signature, query_string)
-        response = requests.get(full_url)
-        if response.ok:
-            root_obj = response.json()
-            if "Token" in root_obj:
-                return root_obj["Token"]["Id"], root_obj["Token"]["ExpireTime"]
-        return None, None
-
-
-class DoubaoStreamASRPerformanceTester:
-    def __init__(self):
         self.config = load_config()
+        self.config_key = config_key
+        self.asr_config = self.config.get("ASR", {}).get(config_key, {})
         self.test_audio_files = self._load_test_audio_files()
         self.results = []
-        
+
     def _load_test_audio_files(self):
-        """加载测试用的音频文件"""
         audio_root = os.path.join(os.getcwd(), "config", "assets")
         test_files = []
-        
         if os.path.exists(audio_root):
             for file_name in os.listdir(audio_root):
-                if file_name.endswith('.wav') or file_name.endswith('.pcm'):
+                if file_name.endswith(('.wav', '.pcm')):
-                    with open(os.path.join(audio_root, file_name), 'rb') as f:
+                    file_path = os.path.join(audio_root, file_name)
-                        test_files.append(f.read())
+                    with open(file_path, 'rb') as f:
+                        test_files.append({
+                            'data': f.read(),
+                            'path': file_path,
+                            'name': file_name
+                        })
         return test_files
-    
+
-    async def test_doubao_stream_asr(self, test_count=5):
+    async def test(self, test_count=5):
-        """测试豆包流式ASR首词响应时间"""
+        raise NotImplementedError
+
+    def _calculate_result(self, service_name, latencies, test_count):
+        valid_latencies = [l for l in latencies if l > 0]
+        if valid_latencies:
+            avg_latency = sum(valid_latencies) / len(valid_latencies)
+            status = f"成功（{len(valid_latencies)}/{test_count}次有效）"
+        else:
+            avg_latency = 0
+            status = "失败: 所有测试均失败"
+        return {"name": service_name, "latency": avg_latency, "status": status}
+
+
+class DoubaoStreamASRTester(BaseASRTester):
+    def __init__(self):
+        super().__init__("DoubaoStreamASR")
+
+    def _generate_header(self):
+        header = bytearray()
+        header.append((0x01 << 4) | 0x01)
+        header.append((0x01 << 4) | 0x00)
+        header.append((0x01 << 4) | 0x01)
+        header.append(0x00)
+        return header
+
+    def _generate_audio_default_header(self):
+        return self._generate_header()
+
+    def _parse_response(self, res: bytes) -> dict:
+        try:
+            if len(res) < 4:
+                return {"error": "响应数据长度不足"}
+            header = res[:4]
+            message_type = header[1] >> 4
+            if message_type == 0x0F:
+                code = int.from_bytes(res[4:8], "big", signed=False)
+                msg_length = int.from_bytes(res[8:12], "big", signed=False)
+                error_msg = json.loads(res[12:].decode("utf-8"))
+                return {
+                    "code": code,
+                    "msg_length": msg_length,
+                    "payload_msg": error_msg
+                }
+            try:
+                json_data = res[12:].decode("utf-8")
+                return {"payload_msg": json.loads(json_data)}
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                return {"error": "JSON解析失败"}
+        except Exception:
+            return {"error": "解析响应失败"}
+
+    async def test(self, test_count=5):
         if not self.test_audio_files:
-            print("没有找到测试音频文件")
+            return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未找到测试音频"}
-            return
+        if not self.asr_config:
-            
+            return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未配置"}
-        asr_config = self.config["ASR"]["DoubaoStreamASR"]
+
         latencies = []
-        
         for i in range(test_count):
             try:
                 ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel"
-                appid = asr_config["appid"]
+                appid = self.asr_config["appid"]
-                access_token = asr_config["access_token"]
+                access_token = self.asr_config["access_token"]
-                uid = asr_config.get("uid", "streaming_asr_service")
+                uid = self.asr_config.get("uid", "streaming_asr_service")
-                
+
                 start_time = time.time()
-                
+
                 headers = {
                     "X-Api-App-Key": appid,
                     "X-Api-Access-Key": access_token,
                     "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
                     "X-Api-Connect-Id": str(uuid.uuid4())
                 }
-                
+
                 async with websockets.connect(
                     ws_url,
                     additional_headers=headers,
@@ -111,12 +123,8 @@ class DoubaoStreamASRPerformanceTester:
                     ping_timeout=None,
                     close_timeout=10
                 ) as ws:
-                    # 发送初始化请求
                     request_params = {
-                        "app": {
+                        "app": {"appid": appid, "token": access_token},
-                            "appid": appid,
-                            "token": access_token
-                        },
                         "user": {"uid": uid},
                         "request": {
                             "reqid": str(uuid.uuid4()),
@@ -135,203 +143,141 @@ class DoubaoStreamASRPerformanceTester:
                             "sample_rate": 16000
                         }
                     }
-                    
+
                     payload_bytes = str.encode(json.dumps(request_params))
                     payload_bytes = gzip.compress(payload_bytes)
                     full_client_request = self._generate_header()
                     full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
                     full_client_request.extend(payload_bytes)
-                    
                     await ws.send(full_client_request)
-                    
+
                     init_res = await ws.recv()
                     result = self._parse_response(init_res)
-                    
                     if "code" in result and result["code"] != 1000:
-                        raise Exception(f"ASR服务初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}")
+                        raise Exception(f"初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}")
-                    
+
-                    # 发送音频数据
+                    audio_data = self.test_audio_files[0]['data']
-                    audio_data = self.test_audio_files[0]
                     if audio_data.startswith(b'RIFF'):
                         audio_data = audio_data[44:]
-                        
+
-                    # 直接发送原始音频数据，不进行opus解码
                     payload = gzip.compress(audio_data)
                     audio_request = bytearray(self._generate_audio_default_header())
                     audio_request.extend(len(payload).to_bytes(4, "big"))
                     audio_request.extend(payload)
                     await ws.send(audio_request)
-                    
+
-                    # 等待第一个数据块
                     first_chunk = await ws.recv()
                     latency = time.time() - start_time
                     latencies.append(latency)
                     await ws.close()
-                    
+
             except Exception as e:
-                print(f"第{i+1}次测试: {str(e)}")
+                print(f"[豆包ASR] 第{i+1}次测试失败: {str(e)}")
                 latencies.append(0)
-        
+
         return self._calculate_result("豆包流式ASR", latencies, test_count)
-        
+
-    async def test_aliyun_stream_asr(self, test_count=5):
+
-        """测试阿里云流式ASR首词响应时间"""
+class QwenASRFlashTester(BaseASRTester):
+    def __init__(self):
+        super().__init__("Qwen3ASRFlash")
+
+    async def _test_single(self, audio_file_info):
+        start_time = time.time()
+        temp_file_path = None
+
+        try:
+            audio_data = audio_file_info['data']
+            with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
+                temp_file_path = f.name
+
+            with wave.open(temp_file_path, 'wb') as wav_file:
+                wav_file.setnchannels(1)
+                wav_file.setsampwidth(2)
+                wav_file.setframerate(16000)
+                wav_file.writeframes(audio_data)
+
+            messages = [
+                {
+                    "role": "user",
+                    "content": [
+                        {"audio": temp_file_path}
+                    ]
+                }
+            ]
+
+            api_key = self.asr_config.get("api_key") or os.getenv("DASHSCOPE_API_KEY")
+            if not api_key:
+                raise ValueError("未配置 api_key")
+
+            if dashscope is None:
+                raise RuntimeError("未安装 dashscope 库")
+
+            dashscope.api_key = api_key
+
+            response = dashscope.MultiModalConversation.call(
+                model="qwen3-asr-flash",
+                messages=messages,
+                result_format="message",
+                asr_options={"enable_lid": True, "enable_itn": False},
+                stream=True
+            )
+
+            for chunk in response:
+                latency = time.time() - start_time
+                return latency
+
+            raise Exception("流式结束，未收到任何响应")
+
+        except Exception as e:
+            raise Exception(f"通义ASR流式失败: {str(e)}")
+
+        finally:
+            if temp_file_path and os.path.exists(temp_file_path):
+                try:
+                    os.unlink(temp_file_path)
+                except:
+                    pass
+
+    async def test(self, test_count=5):
         if not self.test_audio_files:
-            print("没有找到测试音频文件")
+            return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未找到测试音频"}
-            return
+        if not self.asr_config and not os.getenv("DASHSCOPE_API_KEY"):
-            
+            return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未配置 api_key"}
-        asr_config = self.config["ASR"]["AliyunStreamASR"]
+
         latencies = []
-        
         for i in range(test_count):
             try:
-                access_key_id = asr_config["access_key_id"]
+                # print(f"\n[通义ASR] 开始第 {i+1} 次测试...")
-                access_key_secret = asr_config["access_key_secret"]
+                latency = await self._test_single(self.test_audio_files[0])
-                appkey = asr_config["appkey"]
+                latencies.append(latency)
-                host = asr_config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com")
+                # print(f"[通义ASR] 第{i+1}次成功 延迟: {latency:.3f}s")
-                
-                # 获取Token
-                token, _ = AccessToken.create_token(access_key_id, access_key_secret)
-                if not token:
-                    raise Exception("无法获取阿里云ASR Token")
-                
-                # 确定WebSocket URL
-                if "-internal." in host:
-                    ws_url = f"ws://{host}/ws/v1"
-                else:
-                    ws_url = f"wss://{host}/ws/v1"
-                
-                start_time = time.time()
-                async with websockets.connect(
-                    ws_url,
-                    additional_headers={"X-NLS-Token": token},
-                    max_size=1000000000,
-                    ping_interval=None,
-                    ping_timeout=None,
-                    close_timeout=10
-                ) as ws:
-                    # 发送开始请求
-                    start_request = {
-                        "header": {
-                            "namespace": "SpeechTranscriber",
-                            "name": "StartTranscription",
-                            "status": 20000000,
-                            "message_id": ''.join(random.choices('0123456789abcdef', k=32)),
-                            "task_id": ''.join(random.choices('0123456789abcdef', k=32)),
-                            "status_text": "Gateway:SUCCESS:Success.",
-                            "appkey": appkey
-                        },
-                        "payload": {
-                            "format": "pcm",
-                            "sample_rate": 16000,
-                            "enable_intermediate_result": True,
-                            "enable_punctuation_prediction": True,
-                            "enable_inverse_text_normalization": True,
-                            "max_sentence_silence": asr_config.get("max_sentence_silence", 8000),
-                            "enable_voice_detection": False,
-                        }
-                    }
-                    await ws.send(json.dumps(start_request, ensure_ascii=False))
-                    
-                    # 等待服务器准备
-                    start_response = await ws.recv()
-                    response_data = json.loads(start_response)
-                    if response_data["header"]["name"] != "TranscriptionStarted":
-                        raise Exception("阿里云ASR服务初始化失败")
-                    
-                    # 发送音频数据
-                    audio_data = self.test_audio_files[0]
-                    if audio_data.startswith(b'RIFF'):
-                        audio_data = audio_data[44:]  # 去掉WAV头
-                    
-                    await ws.send(audio_data)
-                    
-                    # 等待第一个结果
-                    while True:
-                        response = await ws.recv()
-                        if isinstance(response, str):
-                            result = json.loads(response)
-                            if result["header"]["name"] == "TranscriptionResultChanged":
-                                latency = time.time() - start_time
-                                latencies.append(latency)
-                                break
-                            elif result["header"]["name"] == "TaskFailed":
-                                raise Exception(f"阿里云ASR识别失败: {result.get('payload', {}).get('error_info', '未知错误')}")
-                    
-                    # 发送停止请求
-                    stop_msg = {
-                        "header": {
-                            "namespace": "SpeechTranscriber",
-                            "name": "StopTranscription",
-                            "status": 20000000,
-                            "message_id": ''.join(random.choices('0123456789abcdef', k=32)),
-                            "status_text": "Client:Stop",
-                            "appkey": appkey
-                        }
-                    }
-                    await ws.send(json.dumps(stop_msg, ensure_ascii=False))
-                    await ws.close()
-                    
             except Exception as e:
-                print(f"第{i+1}次测试: {str(e)}")
+                # print(f"[通义ASR] 第{i+1}次测试失败: {str(e)}")
                 latencies.append(0)
-        
+
-        return self._calculate_result("阿里云流式ASR", latencies, test_count)
+        return self._calculate_result("通义千问ASR", latencies, test_count)
-    
+
-    def _generate_header(self):
+
-        """生成请求头"""
+class ASRPerformanceSuite:
-        header = bytearray()
+    def __init__(self):
-        header.append((0x01 << 4) | 0x01)
+        self.testers = []
-        header.append((0x01 << 4) | 0x00)
+        self.results = []
-        header.append((0x01 << 4) | 0x01)
+
-        header.append(0x00)
+    def register_tester(self, tester_class):
-        return header
-    
-    def _generate_audio_default_header(self):
-        """生成音频请求头"""
-        return self._generate_header()
-    
-    def _parse_response(self, res: bytes) -> dict:
-        """解析响应"""
         try:
-            if len(res) < 4:
+            tester = tester_class()
-                return {"error": "响应数据长度不足"}
+            self.testers.append(tester)
-                
+            print(f"已注册测试器: {tester.config_key}")
-            header = res[:4]
+        except Exception as e:
-            message_type = header[1] >> 4
+            name_map = {
-            
+                "DoubaoStreamASRTester": "豆包流式ASR",
-            if message_type == 0x0F:
+                "QwenASRFlashTester": "通义千问ASR"
-                code = int.from_bytes(res[4:8], "big", signed=False)
+            }
-                msg_length = int.from_bytes(res[8:12], "big", signed=False)
+            name = name_map.get(tester_class.__name__, tester_class.__name__)
-                error_msg = json.loads(res[12:].decode("utf-8"))
+            print(f"跳过 {name}: {str(e)}")
-                return {
+
-                    "code": code,
-                    "msg_length": msg_length,
-                    "payload_msg": error_msg
-                }
-                
-            try:
-                json_data = res[12:].decode("utf-8")
-                return {"payload_msg": json.loads(json_data)}
-            except (UnicodeDecodeError, json.JSONDecodeError):
-                return {"error": "JSON解析失败"}
-                
-        except Exception:
-            return {"error": "解析响应失败"}
-    
-    def _calculate_result(self, service_name, latencies, test_count):
-        """计算结果"""
-        valid_latencies = [l for l in latencies if l > 0]
-        if valid_latencies:
-            avg_latency = sum(valid_latencies) / len(valid_latencies)
-            status = f"成功（{len(valid_latencies)}/{test_count}次有效）"
-        else:
-            avg_latency = 0
-            status = "失败: 所有测试均失败"
-        return {"name": service_name, "latency": avg_latency, "status": status}
-    
     def _print_results(self, test_count):
-        """打印测试结果"""
         if not self.results:
             print("没有有效的ASR测试结果")
             return
@@ -341,7 +287,6 @@ class DoubaoStreamASRPerformanceTester:
         print(f"{'='*60}")
         print(f"测试次数: 每个ASR服务测试 {test_count} 次")
 
-        # 排序结果：成功优先，按延迟升序
         success_results = sorted(
             [r for r in self.results if "成功" in r["status"]],
             key=lambda x: x["latency"]
@@ -349,56 +294,41 @@ class DoubaoStreamASRPerformanceTester:
         failed_results = [r for r in self.results if "成功" not in r["status"]]
 
         table_data = [
-            [r["name"], f"{r['latency']:.3f}", r["status"]]
+            [r["name"], f"{r['latency']:.3f}s" if r['latency'] > 0 else "N/A", r["status"]]
             for r in success_results + failed_results
         ]
 
-        print(tabulate(table_data, headers=["ASR服务", "首词延迟(秒)", "状态"], tablefmt="grid"))
+        print(tabulate(table_data, headers=["ASR服务", "首词延迟", "状态"], tablefmt="grid"))
-        print("\n测试说明：测量从发送请求到接收第一个识别结果的时间，取多次测试平均值")
+        print("\n测试说明：")
-        print("- 超时控制: 单个请求最大等待时间为10秒")
+        print("- 测量从发送请求到接收第一个有效识别文本的时间")
-        print("- 错误处理: 无法连接和超时的列为网络错误")
+        print("- 超时控制: DashScope 默认超时，豆包 WebSocket 超时10秒")
-        print("- 排序规则: 按平均耗时从快到慢排序")
+        print("- 排序规则: 成功的按延迟升序，失败的排在后面")
-    
+
     async def run(self, test_count=5):
-        """执行测试"""
         print(f"开始流式ASR首词响应时间测试...")
-        print(f"每个ASR服务测试次数: {test_count}次")
+        print(f"每个ASR服务测试次数: {test_count}次\n")
-        
+
-        if not self.config.get("ASR"):
-            print("配置文件中未找到ASR配置")
-            return
-        
-        # 测试每种ASR服务
         self.results = []
-        
+        for tester in self.testers:
-        # 测试豆包ASR
+            print(f"\n--- 测试 {tester.config_key} ---")
-        if self.config["ASR"].get("DoubaoStreamASR"):
+            result = await tester.test(test_count)
-            result = await self.test_doubao_stream_asr(test_count)
             self.results.append(result)
-        else:
+
-            print("配置文件中未找到豆包流式ASR配置，跳过测试")
-        
-        # 测试阿里云ASR
-        if self.config["ASR"].get("AliyunStreamASR"):
-            result = await self.test_aliyun_stream_asr(test_count)
-            self.results.append(result)
-        else:
-            print("配置文件中未找到阿里云流式ASR配置，跳过测试")
-        
-        # 打印结果
         self._print_results(test_count)
 
+
 async def main():
     import argparse
-    
     parser = argparse.ArgumentParser(description="流式ASR首词响应时间测试工具")
     parser.add_argument("--count", type=int, default=5, help="测试次数")
-    
     args = parser.parse_args()
-    await DoubaoStreamASRPerformanceTester().run(args.count)
+
+    suite = ASRPerformanceSuite()
+    suite.register_tester(DoubaoStreamASRTester)
+    suite.register_tester(QwenASRFlashTester)
+
+    await suite.run(args.count)
+
 
 if __name__ == "__main__":
-    import os
-    import gzip
-    import opuslib_next
     asyncio.run(main())