From 6ffa325b73e6b5ac5cd773969127c61e6f327dae Mon Sep 17 00:00:00 2001 From: FAN-yeB <1442100690@qq.com> Date: Mon, 8 Dec 2025 15:51:04 +0800 Subject: [PATCH] =?UTF-8?q?update:=E7=BB=9F=E4=B8=80=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E6=B5=8B=E9=80=9F=E5=B7=A5=E5=85=B7=E7=BB=9F=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E5=8C=BA=E9=97=B4=EF=BC=8C=E6=96=B0=E5=A2=9E=E7=99=BE?= =?UTF-8?q?=E7=82=BC=E5=B9=B3=E5=8F=B0=E6=B5=81=E5=BC=8FTTS=E6=B5=8B?= =?UTF-8?q?=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../performance_tester_stream_asr.py | 256 ++++++++++-------- .../performance_tester_stream_tts.py | 235 ++++++++++++---- 2 files changed, 323 insertions(+), 168 deletions(-) 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 f4508ef2..f81f6317 100644 --- a/main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py +++ b/main/xiaozhi-server/performance_tester/performance_tester_stream_asr.py @@ -50,7 +50,9 @@ class BaseASRTester: raise NotImplementedError def _calculate_result(self, service_name, latencies, test_count): - valid_latencies = [l for l in latencies if l > 0] + """计算测试结果（修复：正确处理None值，剔除失败测试）""" + # 剔除None值（失败的测试）和无效延迟，只统计有效延迟 + valid_latencies = [l for l in latencies if l is not None and l > 0] if valid_latencies: avg_latency = sum(valid_latencies) / len(valid_latencies) status = f"成功（{len(valid_latencies)}/{test_count}次有效）" @@ -64,16 +66,45 @@ class DoubaoStreamASRTester(BaseASRTester): def __init__(self): super().__init__("DoubaoStreamASR") - def _generate_header(self): + def _generate_header( + self, + version=0x01, + message_type=0x01, + message_type_specific_flags=0x00, + serial_method=0x01, + compression_type=0x01, + reserved_data=0x00, + extension_header: bytes = b"", + ): + """生成协议头（修复：使用正确的Header格式）""" header = bytearray() - header.append((0x01 << 4) | 0x01) - header.append((0x01 << 4) | 0x00) - header.append((0x01 << 4) | 0x01) - header.append(0x00) + header_size = int(len(extension_header) / 4) + 1 + header.append((version << 4) | header_size) + header.append((message_type << 4) | message_type_specific_flags) + header.append((serial_method << 4) | compression_type) + header.append(reserved_data) + header.extend(extension_header) return header def _generate_audio_default_header(self): - return self._generate_header() + """生成音频数据Header""" + return self._generate_header( + version=0x01, + message_type=0x02, + message_type_specific_flags=0x00, # 普通音频帧 + serial_method=0x01, + compression_type=0x01, + ) + + def _generate_last_audio_header(self): + """生成最后一帧音频的Header（标记音频结束）""" + return self._generate_header( + version=0x01, + message_type=0x02, + message_type_specific_flags=0x02, # 0x02表示这是最后一帧 + serial_method=0x01, + compression_type=0x01, + ) def _parse_response(self, res: bytes) -> dict: try: @@ -110,6 +141,7 @@ class DoubaoStreamASRTester(BaseASRTester): ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel" appid = self.asr_config["appid"] access_token = self.asr_config["access_token"] + cluster = self.asr_config.get("cluster", "volcengine_input_common") uid = self.asr_config.get("uid", "streaming_asr_service") start_time = time.time() @@ -130,7 +162,7 @@ class DoubaoStreamASRTester(BaseASRTester): close_timeout=10 ) as ws: request_params = { - "app": {"appid": appid, "token": access_token}, + "app": {"appid": appid, "cluster": cluster, "token": access_token}, "user": {"uid": uid}, "request": { "reqid": str(uuid.uuid4()), @@ -166,8 +198,9 @@ class DoubaoStreamASRTester(BaseASRTester): if audio_data.startswith(b'RIFF'): audio_data = audio_data[44:] + # 发送音频数据（使用最后一帧标记，告诉服务端音频已结束） payload = gzip.compress(audio_data) - audio_request = bytearray(self._generate_audio_default_header()) + audio_request = bytearray(self._generate_last_audio_header()) # 修复：使用最后一帧Header audio_request.extend(len(payload).to_bytes(4, "big")) audio_request.extend(payload) await ws.send(audio_request) @@ -175,11 +208,12 @@ class DoubaoStreamASRTester(BaseASRTester): first_chunk = await ws.recv() latency = time.time() - start_time latencies.append(latency) + print(f"[豆包ASR] 第{i+1}次 首词延迟: {latency:.3f}s") await ws.close() except Exception as e: print(f"[豆包ASR] 第{i+1}次测试失败: {str(e)}") - latencies.append(0) + latencies.append(None) return self._calculate_result("豆包流式ASR", latencies, test_count) @@ -189,11 +223,12 @@ class QwenASRFlashTester(BaseASRTester): 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'] + + # 优化：将临时文件准备工作移到计时前，减少磁盘IO对性能测试的影响 with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: temp_file_path = f.name @@ -221,6 +256,9 @@ class QwenASRFlashTester(BaseASRTester): dashscope.api_key = api_key + # 统一计时起点：在API调用前开始计时（但文件准备已完成） + start_time = time.time() + response = dashscope.MultiModalConversation.call( model="qwen3-asr-flash", messages=messages, @@ -257,10 +295,10 @@ class QwenASRFlashTester(BaseASRTester): # 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") + print(f"[通义ASR] 第{i+1}次 首词延迟: {latency:.3f}s") except Exception as e: # print(f"[通义ASR] 第{i+1}次测试失败: {str(e)}") - latencies.append(0) + latencies.append(None) return self._calculate_result("通义千问ASR", latencies, test_count) @@ -268,134 +306,115 @@ class QwenASRFlashTester(BaseASRTester): class XunfeiStreamASRTester(BaseASRTester): def __init__(self): super().__init__("XunfeiStreamASR") - + def _create_url(self): - """生成讯飞ASR认证URL""" - url = 'ws://iat.cn-huabei-1.xf-yun.com/v1' - # 生成RFC1123格式的时间戳 + url = "wss://iat-api.xfyun.cn/v2/iat" now = datetime.now() date = format_date_time(mktime(now.timetuple())) - # 拼接字符串 - signature_origin = "host: " + "iat.cn-huabei-1.xf-yun.com" + "\n" - signature_origin += "date: " + date + "\n" - signature_origin += "GET " + "/v1 " + "HTTP/1.1" + signature_origin = f"host: iat-api.xfyun.cn\ndate: {date}\nGET /v2/iat HTTP/1.1" + signature_sha = hmac.new( + self.asr_config["api_secret"].encode('utf-8'), + signature_origin.encode('utf-8'), + hashlib.sha256 + ).digest() + signature_sha = base64.b64encode(signature_sha).decode() - # 进行hmac-sha256进行加密 - signature_sha = hmac.new(self.asr_config["api_secret"].encode('utf-8'), signature_origin.encode('utf-8'), - digestmod=hashlib.sha256).digest() - signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8') + authorization_origin = f'api_key="{self.asr_config["api_key"]}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha}"' + authorization = base64.b64encode(authorization_origin.encode()).decode() - authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % ( - self.asr_config["api_key"], "hmac-sha256", "host date request-line", signature_sha) - authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') + v = {"authorization": authorization, "date": date, "host": "iat-api.xfyun.cn"} + return url + "?" + parse.urlencode(v) - # 将请求的鉴权参数组合为字典 - v = { - "authorization": authorization, - "date": date, - "host": "iat.cn-huabei-1.xf-yun.com" - } - - # 拼接鉴权参数，生成url - url = url + '?' + parse.urlencode(v) - return url - - async def test(self, test_count=5): + async def test(self, test_count: int = 5): if not self.test_audio_files: return {"name": "讯飞流式ASR", "latency": 0, "status": "失败: 未找到测试音频"} if not self.asr_config: return {"name": "讯飞流式ASR", "latency": 0, "status": "失败: 未配置"} - - # 检查必要的配置参数 - required_keys = ["app_id", "api_key", "api_secret"] - for key in required_keys: - if key not in self.asr_config: - return {"name": "讯飞流式ASR", "latency": 0, "status": f"失败: 缺少配置项 {key}"} - + + required = ["app_id", "api_key", "api_secret"] + for k in required: + if k not in self.asr_config: + return {"name": "讯飞流式ASR", "latency": 0, "status": f"失败: 缺少配置 {k}"} + latencies = [] - STATUS_FIRST_FRAME = 0 - + frame_size = 1280 + audio_raw = self.test_audio_files[0]['data'] + if audio_raw.startswith(b'RIFF'): + audio_raw = audio_raw[44:] + for i in range(test_count): try: - # 生成认证URL - ws_url = self._create_url() - - # 获取音频数据 - audio_data = self.test_audio_files[0]['data'] - if audio_data.startswith(b'RIFF'): - audio_data = audio_data[44:] # 跳过WAV文件头 - - # 识别参数 - iat_params = { - "domain": self.asr_config.get("domain", "slm"), - "language": self.asr_config.get("language", "zh_cn"), - "accent": self.asr_config.get("accent", "mandarin"), - "dwa": self.asr_config.get("dwa", "wpgs"), - "result": { - "encoding": "utf8", - "compress": "raw", - "format": "plain" - } - } - - # 准备首帧数据 - first_frame_data = { - "header": { - "status": STATUS_FIRST_FRAME, - "app_id": self.asr_config["app_id"] - }, - "parameter": { - "iat": iat_params - }, - "payload": { - "audio": { - "audio": base64.b64encode(audio_data[:960]).decode('utf-8'), - "sample_rate": 16000, - "encoding": "raw" - } - } - } - - # 启动连接并测量时间 start_time = time.time() - + ws_url = self._create_url() + async with websockets.connect( ws_url, - max_size=1000000000, + additional_headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}, + max_size=1 << 30, ping_interval=None, ping_timeout=None, close_timeout=30, ) as ws: - # 发送首帧数据 - await ws.send(json.dumps(first_frame_data, ensure_ascii=False)) - print(f"[讯飞ASR] 第{i+1}次测试：已发送首帧，等待响应...") - - # 直接等待第一个响应并计算延迟 - # 参考豆包和通义千问的实现方式，简化逻辑 - response_received = False - while not response_received: - try: - # 设置较大的超时时间 - response = await asyncio.wait_for(ws.recv(), timeout=30.0) - - # 收到响应立即计算延迟，不管内容是什么 - # 这样可以准确测量首包到达时间 - latency = time.time() - start_time - latencies.append(latency) - response_received = True - - print(f"[讯飞ASR] 第{i+1}次测试：收到首包响应，延迟: {latency:.3f}s") + + # 第一帧：移除 punc 字段，避免未知参数错误 + await ws.send(json.dumps({ + "common": {"app_id": self.asr_config["app_id"]}, + "business": { + "domain": "iat", + "language": "zh_cn", + "accent": "mandarin", + "dwa": "wpgs", + "vad_eos": 5000 + # 已移除 "punc": True + }, + "data": { + "status": 0, + "format": "audio/L16;rate=16000", + "encoding": "raw", + "audio": base64.b64encode(audio_raw[:frame_size]).decode() + } + }, ensure_ascii=False)) + + # 后续所有帧 + pos = frame_size + while pos < len(audio_raw): + chunk = audio_raw[pos:pos + frame_size] + status = 2 if (pos + frame_size >= len(audio_raw)) else 1 + await ws.send(json.dumps({ + "data": { + "status": status, + "format": "audio/L16;rate=16000", + "encoding": "raw", + "audio": base64.b64encode(chunk).decode() + } + }, ensure_ascii=False)) + if status == 2: break - except asyncio.TimeoutError: - print(f"[讯飞ASR] 第{i+1}次测试：响应超时") - raise Exception("获取响应超时") + pos += frame_size + + # 接收首词 + first_token = True + async for message in ws: + data = json.loads(message) + if data.get("code") != 0: + raise Exception(f"讯飞错误: {data.get('message')}") + + ws_result = data.get("data", {}).get("result", {}).get("ws") + if ws_result: + text = "".join(cw.get("w", "") for seg in ws_result for cw in seg.get("cw", [])) + if text.strip() and first_token: + latency = time.time() - start_time + latencies.append(latency) + print(f"[讯飞ASR] 第{i+1}次 首词延迟: {latency:.3f}s") + first_token = False + break + except Exception as e: print(f"[讯飞ASR] 第{i+1}次测试失败: {str(e)}") - latencies.append(0) - - return self._calculate_result("讯飞流式ASR", latencies, test_count) + latencies.append(None) + return self._calculate_result("讯飞流式ASR", latencies, test_count) class ASRPerformanceSuite: def __init__(self): self.testers = [] @@ -438,8 +457,9 @@ class ASRPerformanceSuite: print(tabulate(table_data, headers=["ASR服务", "首词延迟", "状态"], tablefmt="grid")) print("\n测试说明：") - print("- 测量从发送请求到接收第一个有效识别文本的时间") - print("- 超时控制: DashScope 默认超时，豆包 WebSocket 超时10秒") + print("- 计时起点: 建立连接前（包含握手、发送音频、接收首个识别结果全流程）") + print("- 通义千问优化: 临时文件准备在计时前完成，减少磁盘IO对测试的影响") + print("- 错误处理: 失败的测试不计入平均值，只统计成功测试的延迟") print("- 排序规则: 成功的按延迟升序，失败的排在后面") async def run(self, test_count=5): diff --git a/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py b/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py index ce53fea5..31d49a3e 100644 --- a/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py +++ b/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py @@ -35,11 +35,12 @@ class StreamTTSPerformanceTester: host = tts_config["host"] ws_url = f"wss://{host}/ws/v1" + # 统一计时起点：在建立连接前开始计时 start_time = time.time() async with websockets.connect(ws_url, extra_headers={"X-NLS-Token": token}) as ws: task_id = str(uuid.uuid4()) message_id = str(uuid.uuid4()) - + start_request = { "header": { "message_id": message_id, @@ -55,14 +56,15 @@ class StreamTTSPerformanceTester: "volume": 50, "speech_rate": 0, "pitch_rate": 0, + "enable_subtitle": True, } } await ws.send(json.dumps(start_request)) - + start_response = json.loads(await ws.recv()) if start_response["header"]["name"] != "SynthesisStarted": raise Exception("启动合成失败") - + run_request = { "header": { "message_id": str(uuid.uuid4()), @@ -74,23 +76,142 @@ class StreamTTSPerformanceTester: "payload": {"text": text} } await ws.send(json.dumps(run_request)) - + while True: response = await ws.recv() if isinstance(response, bytes): latency = time.time() - start_time latencies.append(latency) + print(f"[阿里云TTS] 第{i+1}次 首词延迟: {latency:.3f}s") break elif isinstance(response, str): data = json.loads(response) if data["header"]["name"] == "TaskFailed": raise Exception(f"合成失败: {data['payload']['error_info']}") - + except Exception as e: - latencies.append(0) + print(f"[阿里云TTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("阿里云TTS", latencies, test_count) + async def test_alibl_tts(self, text=None, test_count=5): + """测试阿里云百炼CosyVoice流式TTS首词延迟""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["AliBLTTS"] + api_key = tts_config["api_key"] + model = tts_config.get("model", "cosyvoice-v2") + voice = tts_config.get("voice", "longxiaochun_v2") + format_type = tts_config.get("format", "pcm") + sample_rate = int(tts_config.get("sample_rate", "24000")) + + ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/" + headers = { + "Authorization": f"Bearer {api_key}", + "X-DashScope-DataInspection": "enable", + } + + start_time = time.time() + + async with websockets.connect( + ws_url, + additional_headers=headers, + ping_interval=30, + ping_timeout=10, + close_timeout=10, + max_size=10 * 1024 * 1024, + ) as ws: + session_id = uuid.uuid4().hex + + # 1. 发送 run-task（启动任务） + run_task_message = { + "header": { + "action": "run-task", + "task_id": session_id, + "streaming": "duplex", + }, + "payload": { + "task_group": "audio", + "task": "tts", + "function": "SpeechSynthesizer", + "model": model, + "parameters": { + "text_type": "PlainText", + "voice": voice, + "format": format_type, + "sample_rate": sample_rate, + "volume": 50, + "rate": 1.0, + "pitch": 1.0, + }, + "input": {} + }, + } + await ws.send(json.dumps(run_task_message)) + + # 2. 等待 task-started 事件（关键！必须等这个再发文本） + task_started = False + while not task_started: + msg = await ws.recv() + if isinstance(msg, str): + data = json.loads(msg) + header = data.get("header", {}) + event = header.get("event") + if event == "task-started": + task_started = True + print(f"[阿里云百炼TTS] 第{i+1}次 任务启动成功") + elif event == "task-failed": + raise Exception(f"启动失败: {header.get('error_message', '未知错误')}") + + # 3. 发送 continue-task（发送文本！这是正确动作） + continue_task_message = { + "header": { + "action": "continue-task", # 改回 continue-task + "task_id": session_id, + "streaming": "duplex", + }, + "payload": {"input": {"text": text}}, + } + await ws.send(json.dumps(continue_task_message)) + + # 4. 发送 finish-task（结束任务） + finish_task_message = { + "header": { + "action": "finish-task", + "task_id": session_id, + "streaming": "duplex", + }, + "payload": {"input": {}} + } + await ws.send(json.dumps(finish_task_message)) + + # 5. 等待第一个音频数据块 + while True: + msg = await asyncio.wait_for(ws.recv(), timeout=15.0) + if isinstance(msg, (bytes, bytearray)) and len(msg) > 0: + latency = time.time() - start_time + print(f"[阿里云百炼TTS] 第{i+1}次 首词延迟: {latency:.3f}s") + latencies.append(latency) + break + elif isinstance(msg, str): + data = json.loads(msg) + event = data.get("header", {}).get("event") + if event == "task-failed": + raise Exception(f"合成失败: {data}") + elif event == "task-finished": + if not latencies or latencies[-1] is None: + raise Exception("任务结束但未收到音频") + + except Exception as e: + print(f"[阿里云百炼TTS] 第{i+1}次失败: {str(e)}") + latencies.append(None) + + return self._calculate_result("阿里云百炼TTS", latencies, test_count) + async def test_doubao_tts(self, text=None, test_count=5): """测试火山引擎流式TTS首词延迟（测试多次取平均）""" text = text or self.test_texts[0] @@ -114,13 +235,12 @@ class StreamTTSPerformanceTester: } async with websockets.connect(ws_url, additional_headers=ws_header, max_size=1000000000) as ws: session_id = uuid.uuid4().hex - + # 发送会话启动请求 header = bytes([ - (0b0001 << 4) | 0b0001, - 0b0001 << 4 | 0b100, - 0b0001 << 4 | 0b0000, - 0 + (0b0001 << 4) | 0b0001, + 0b0001 << 4 | 0b1011, + 0b0001 << 4 | 0b0000, ]) optional = bytearray() optional.extend((1).to_bytes(4, "big", signed=True)) @@ -129,13 +249,13 @@ class StreamTTSPerformanceTester: optional.extend(session_id_bytes) payload = json.dumps({"speaker": speaker}).encode() await ws.send(header + optional + len(payload).to_bytes(4, "big", signed=True) + payload) - + # 发送文本 header = bytes([ - (0b0001 << 4) | 0b0001, - 0b0001 << 4 | 0b100, - 0b0001 << 4 | 0b0000, - 0 + (0b0001 << 4) | 0b0001, + 0b0001 << 4 | 0b1011, + 0b0001 << 4 | 0b0000, + 0 ]) optional = bytearray() optional.extend((200).to_bytes(4, "big", signed=True)) @@ -144,13 +264,15 @@ class StreamTTSPerformanceTester: optional.extend(session_id_bytes) payload = json.dumps({"text": text, "speaker": speaker}).encode() await ws.send(header + optional + len(payload).to_bytes(4, "big", signed=True) + payload) - + first_chunk = await ws.recv() latency = time.time() - start_time latencies.append(latency) - + print(f"[火山引擎TTS] 第{i+1}次 首词延迟: {latency:.3f}s") + except Exception as e: - latencies.append(0) + print(f"[火山引擎TTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("火山引擎TTS", latencies, test_count) @@ -191,22 +313,24 @@ class StreamTTSPerformanceTester: first_chunk = await ws.recv() latency = time.time() - start_time latencies.append(latency) - + print(f"[PaddleSpeechTTS] 第{i+1}次 首词延迟: {latency:.3f}s") + # 发送结束请求 end_request = { "task": "tts", "signal": "end" } await ws.send(json.dumps(end_request)) - + # 确保连接正常关闭 try: await ws.recv() except websockets.exceptions.ConnectionClosedOK: pass - + except Exception as e: - latencies.append(0) + print(f"[PaddleSpeechTTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("PaddleSpeechTTS", latencies, test_count) @@ -220,29 +344,32 @@ class StreamTTSPerformanceTester: tts_config = self.config["TTS"]["IndexStreamTTS"] api_url = tts_config.get("api_url") voice = tts_config.get("voice") - + + # 统一计时起点：在建立连接前开始计时 start_time = time.time() - + async with aiohttp.ClientSession() as session: payload = {"text": text, "character": voice} async with session.post(api_url, json=payload, timeout=10) as resp: if resp.status != 200: raise Exception(f"请求失败: {resp.status}, {await resp.text()}") - + async for chunk in resp.content.iter_any(): data = chunk[0] if isinstance(chunk, (list, tuple)) else chunk if not data: continue - + latency = time.time() - start_time latencies.append(latency) + print(f"[IndexStreamTTS] 第{i+1}次 首词延迟: {latency:.3f}s") resp.close() break else: - latencies.append(0) - + latencies.append(None) + except Exception as e: - latencies.append(0) + print(f"[IndexStreamTTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("IndexStreamTTS", latencies, test_count) @@ -257,7 +384,8 @@ class StreamTTSPerformanceTester: api_url = tts_config["api_url"] access_token = tts_config["access_token"] voice = tts_config["voice"] - + + # 统一计时起点：在建立连接前开始计时 start_time = time.time() async with aiohttp.ClientSession() as session: params = { @@ -273,21 +401,23 @@ class StreamTTSPerformanceTester: "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } - + async with session.get(api_url, params=params, headers=headers, timeout=10) as resp: if resp.status != 200: raise Exception(f"请求失败: {resp.status}, {await resp.text()}") - + # 接收第一个数据块 async for _ in resp.content.iter_any(): latency = time.time() - start_time latencies.append(latency) + print(f"[LinkeraiTTS] 第{i+1}次 首词延迟: {latency:.3f}s") break else: - latencies.append(0) - + latencies.append(None) + except Exception as e: - latencies.append(0) + print(f"[LinkeraiTTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("LinkeraiTTS", latencies, test_count) @@ -305,10 +435,9 @@ class StreamTTSPerformanceTester: api_secret = tts_config["api_secret"] api_url = tts_config.get("api_url", "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6") voice = tts_config.get("voice", "x5_lingxiaoxuan_flow") - # 生成认证URL auth_url = self._create_xunfei_auth_url(api_key, api_secret, api_url) - + start_time = time.time() async with websockets.connect( auth_url, ping_interval=30, @@ -318,10 +447,7 @@ class StreamTTSPerformanceTester: ) as ws: # 构造请求 request = self._build_xunfei_request(app_id, text, voice) - # 发送请求后立即计时，确保准确测量从发送文本到接收首块的时间 await ws.send(json.dumps(request)) - start_time = time.time() - # 等待第一个音频数据块 first_audio_received = False while not first_audio_received: @@ -329,14 +455,14 @@ class StreamTTSPerformanceTester: data = json.loads(msg) header = data.get("header", {}) code = header.get("code") - + if code != 0: message = header.get("message", "未知错误") raise Exception(f"合成失败: {code} - {message}") - + payload = data.get("payload", {}) audio_payload = payload.get("audio", {}) - + if audio_payload: status = audio_payload.get("status", 0) audio_data = audio_payload.get("audio", "") @@ -344,10 +470,12 @@ class StreamTTSPerformanceTester: # 收到第一个音频数据块 latency = time.time() - start_time latencies.append(latency) + print(f"[讯飞TTS] 第{i+1}次 首词延迟: {latency:.3f}s") first_audio_received = True break except Exception as e: - latencies.append(0) + print(f"[讯飞TTS] 第{i+1}次测试失败: {str(e)}") + latencies.append(None) return self._calculate_result("讯飞TTS", latencies, test_count) @@ -431,8 +559,9 @@ class StreamTTSPerformanceTester: def _calculate_result(self, service_name, latencies, test_count): - """计算测试结果""" - valid_latencies = [l for l in latencies if l > 0] + """计算测试结果（正确处理None值，剔除失败测试）""" + # 剔除失败的测试（None值和<=0延迟），只统计有效延迟 + valid_latencies = [l for l in latencies if l is not None and l > 0] if valid_latencies: avg_latency = sum(valid_latencies) / len(valid_latencies) status = f"成功（{len(valid_latencies)}/{test_count}次有效）" @@ -466,9 +595,10 @@ class StreamTTSPerformanceTester: ] print(tabulate(table_data, headers=["TTS服务", "首词延迟(秒)", "状态"], tablefmt="grid")) - print("\n测试说明：测量从发送请求到接收第一个音频数据块的时间，取多次测试平均值") + print("\n测试说明：测量从建立连接到接收第一个音频数据块的时间（包含握手、鉴权、发送文本），取多次测试平均值") + print("- 计时起点: 建立WebSocket/HTTP连接前（统一包含网络建连、握手、发送文本全流程）") print("- 超时控制: 单个请求最大等待时间为10秒") - print("- 错误处理: 无法连接和超时的列为网络错误") + print("- 错误处理: 失败的测试不计入平均值，只统计成功测试的延迟") print("- 排序规则: 按平均耗时从快到慢排序") @@ -494,7 +624,12 @@ class StreamTTSPerformanceTester: # 测试阿里云TTS result = await self.test_aliyun_tts(test_text, test_count) self.results.append(result) - + + # 测试阿里云百炼TTS + if self.config.get("TTS", {}).get("AliBLTTS"): + result = await self.test_alibl_tts(test_text, test_count) + self.results.append(result) + # 测试火山引擎TTS result = await self.test_doubao_tts(test_text, test_count) self.results.append(result)