mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-22 07:03:53 +08:00
Merge pull request #2210 from xinnan-tech/pertext
update:更新流式ASR首词延迟测试工具
This commit is contained in:
hrz
@@ -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())
|
||||
Reference in New Issue
Block a user