diff --git a/main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py b/main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py index e0309685..e5ccdaa6 100644 --- a/main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py +++ b/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: - @staticmethod - 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): +class BaseASRTester: + def __init__(self, config_key: str): 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'): - with open(os.path.join(audio_root, file_name), 'rb') as f: - test_files.append(f.read()) + if file_name.endswith(('.wav', '.pcm')): + file_path = os.path.join(audio_root, file_name) + 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): - """测试豆包流式ASR首词响应时间""" + + async def test(self, test_count=5): + 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 - - asr_config = self.config["ASR"]["DoubaoStreamASR"] + return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未找到测试音频"} + if not self.asr_config: + return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未配置"} + latencies = [] - for i in range(test_count): try: ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel" - appid = asr_config["appid"] - access_token = asr_config["access_token"] - uid = asr_config.get("uid", "streaming_asr_service") - + appid = self.asr_config["appid"] + access_token = self.asr_config["access_token"] + 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": { - "appid": appid, - "token": access_token - }, + "app": {"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', '未知错误')}") - - # 发送音频数据 - audio_data = self.test_audio_files[0] + raise Exception(f"初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}") + + audio_data = self.test_audio_files[0]['data'] 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 - - asr_config = self.config["ASR"]["AliyunStreamASR"] + return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未找到测试音频"} + if not self.asr_config and not os.getenv("DASHSCOPE_API_KEY"): + return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未配置 api_key"} + latencies = [] - for i in range(test_count): try: - access_key_id = asr_config["access_key_id"] - access_key_secret = asr_config["access_key_secret"] - appkey = asr_config["appkey"] - host = asr_config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com") - - # 获取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() - + # print(f"\n[通义ASR] 开始第 {i+1} 次测试...") + latency = await self._test_single(self.test_audio_files[0]) + latencies.append(latency) + # print(f"[通义ASR] 第{i+1}次成功 延迟: {latency:.3f}s") 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) - - 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: - """解析响应""" + + return self._calculate_result("通义千问ASR", latencies, test_count) + + +class ASRPerformanceSuite: + def __init__(self): + self.testers = [] + self.results = [] + + def register_tester(self, tester_class): 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": "解析响应失败"} - - 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} - + tester = tester_class() + self.testers.append(tester) + print(f"已注册测试器: {tester.config_key}") + except Exception as e: + name_map = { + "DoubaoStreamASRTester": "豆包流式ASR", + "QwenASRFlashTester": "通义千问ASR" + } + name = name_map.get(tester_class.__name__, tester_class.__name__) + print(f"跳过 {name}: {str(e)}") + 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("\n测试说明：测量从发送请求到接收第一个识别结果的时间，取多次测试平均值") - print("- 超时控制: 单个请求最大等待时间为10秒") - print("- 错误处理: 无法连接和超时的列为网络错误") - print("- 排序规则: 按平均耗时从快到慢排序") - + print(tabulate(table_data, headers=["ASR服务", "首词延迟", "状态"], tablefmt="grid")) + print("\n测试说明：") + print("- 测量从发送请求到接收第一个有效识别文本的时间") + print("- 超时控制: DashScope 默认超时，豆包 WebSocket 超时10秒") + print("- 排序规则: 成功的按延迟升序，失败的排在后面") + async def run(self, test_count=5): - """执行测试""" print(f"开始流式ASR首词响应时间测试...") - print(f"每个ASR服务测试次数: {test_count}次") - - if not self.config.get("ASR"): - print("配置文件中未找到ASR配置") - return - - # 测试每种ASR服务 + print(f"每个ASR服务测试次数: {test_count}次\n") + self.results = [] - - # 测试豆包ASR - if self.config["ASR"].get("DoubaoStreamASR"): - result = await self.test_doubao_stream_asr(test_count) + for tester in self.testers: + print(f"\n--- 测试 {tester.config_key} ---") + result = await tester.test(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()) \ No newline at end of file