ASR 性能压测与容量规划 —— 3 个实例能接多少路电话?

暂停课程路线,处理一个直接的工作问题:「显存只够起 3 个 ASR 实例,这 3 个实例最多能接多少路并发通话?怎么科学地测出来?」本报告给你:指标体系 → 话务量数学 → GPU 供给模型 → 可运行的压测代码 → 容量判定流程。

已决策 3 实例拓扑?配套的可执行压测方案(SLO、11 档阶梯、6 个测试用例、记录模板、时间表)见 loadtest-plan-a40-3inst.md,照着跑即可。

一、指标体系:测什么

二、需求侧:你的话务量(Erlang 数学)

电信业的百年标准:话务量单位 Erlang = 平均并发呼叫数(1 Erlang = 一路呼叫占满一小时)。提供话务量 = 呼叫到达率 × 平均通话时长;由它可用 Erlang B(无排队,呼损概率)与 Erlang C(无限排队,等待概率)算出需要多少条"线路"。

手算示例:每小时 50 通、平均 3 分钟 → 话务量 = 50 × 3/60 = 2.5 Erlang,即平均 2.5 路并发。下面是可运行的计算器(改参数重跑):

三、供给侧:官方基准告诉我们什么

Qwen3-ASR 技术报告 §2.4 Table 2(vLLM v0.14.0、BF16、约 2 分钟音频/请求、GPU 型号未公开)——这是唯一官方数据,完整抄录:

读这张表的三个要点(都是能从数字直接看出来的):

1. RTF 随并发增长很慢(批处理摊销:并发 ×128,RTF 只 ×6.8)——0.6B 的算力根本不是瓶颈;
2. TTFT 随并发增长极快(×128 并发,p95 从 105ms 涨到 6195ms,×59)——排队延迟先撞线;
3. 吞吐上界 ≈ 2000 秒音频/秒(0.6B 单实例)——这是「延迟不管的话」的理论天花板:2000 路通话 × 1x 实时。
4. 顺带一个口径关系(防读混):「吞吐」列就是聚合 RTFx;「RTF」列 = 并发数 ÷ 吞吐(验证:128 ÷ 2000 = 0.064,正是表中 RTF@128)——前者聚合、后者每路。

四、A40 上的最大实例数(0.6B)—— 算给你看

48GB 显存能塞几个 0.6B 实例?先立一个反直觉的事实:实例不是越多越好——每个实例都要完整复制一份权重和固定开销,KV 池总量随之被蚕食,显存侧容量随实例数单调下降。所以「最大实例数」有三档答案,由三笔账决定(权重为推导值,进程开销无公开数字、须实测):

账一:权重。0.6B 版总参数约 0.9B(含编码器),BF16 下 ≈ 1.8GB。

账二:每实例固定开销。CUDA context + CUDA graph 等进程级开销,vLLM 无公开数字(已核实;vLLM 启动时自行探测)——按 ~2GB 估,以 nvidia-smi 差分实测为准。

账三:每路通话的 KV 开销(推导)。0.6B 文本配置(28 层、GQA 8 KV 头、head_dim 128、bf16)下,每个 token 的 KV = 2×28×8×128×2B = 112 KB;一路通话占的 token = 音频嵌入(12.5/秒)+ 生成文本(约 5/秒):

拓扑对比。vLLM 官方文档明确 gpu_memory_utilization 是每实例独立限额(默认已改为 0.92),并给出"同卡两实例各设 0.5"的官方示例。把 0.92 按实例数均分:

读表得出两条规则:

• 呼叫短(≤5 分钟):瓶颈是 TTFT,不是 KV——单实例 KV 能容约 69 路 5 分钟通话,而 TTFT 在 15–32 路就先撞线(报告 Table 2 的形状)。拆分多实例让每实例负载降为 1/N、TTFT 大幅改善,总容量接近线性扩容(3 实例 ≈ 3 × 单实例 TTFT 容量)。
• 呼叫长(≥10 分钟):KV 池与 TTFT 汇合——10 分钟通话 1.18GB/路,单实例 KV 容量 ~34 路,与 TTFT 约束同量级。拆分让权重与固定开销重复占用,总 KV 容量反而缩水(1 实例 40GB 池 vs 3 实例合计 33GB),单实例大池更优。

五、压测协议:业界标准六步法

业界事实上的参考基准是 NVIDIA Riva 的流式压测方法(非官方认证,业内普遍沿用其口径;官方开放基准 MLPerf Inference 测离线吞吐、不建模流式延迟 SLO):客户端 --simulate_realtime=true(每路流按 1x 实时回放音频)、--num_parallel_requests=N、固定块长(100/160/800ms),统计中间/最终延迟与 RTFX。FunASR/WeNet 同样以「并发 vs RTF 表」呈现。我们复刻这套方法(Qwen3-ASR 官方仓库没有自带压测脚本,已核实),协议如下:

六、代码实例 ①:并发压测负载生成器

把 Riva 的「N 路并发 × 1x 实时回放」模式翻译成针对 Qwen3-ASR 服务(WebSocket/OpenAI 兼容端点均可适配)的可运行代码。发送/接收的消息格式按你的服务协议调整(注释已标出):

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""asr_loadgen.py — 流式 ASR 并发压测器(通话场景)

准备音频:  sox call_8k.wav -r 16000 -c 1 call_16k.wav
运行:      python3 asr_loadgen.py --url ws://gpu-host:9000/asr \
               --wav call_16k.wav --streams 16 --ramp 1.0
阶梯加压:  外层循环改 --streams(1,2,4,8,16,32,64,128),每档 5-10 分钟。
依赖:      pip install websockets numpy
"""
import asyncio, argparse, json, statistics, time, wave

import numpy as np
import websockets


def load_pcm16(path, sr=16000):
    """读 16kHz 单声道 s16le wav → 按 chunk_ms 切成字节块列表"""
    w = wave.open(path, "rb")
    assert w.getframerate() == sr and w.getnchannels() == 1, \
        "请先重采样: sox in.wav -r 16000 -c 1 out.wav"
    data = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16)
    n = int(sr * 0.1)                      # 100ms 一块
    return [data[i:i + n].tobytes() for i in range(0, len(data), n)]


async def one_stream(idx, args, chunks, results):
    await asyncio.sleep(idx * args.ramp)   # 错峰爬坡,避免建连冲击
    ttft = None
    audio_sec = len(chunks) * args.chunk_ms / 1000.0
    t0 = time.perf_counter()
    try:
        async with websockets.connect(args.url, max_size=None) as ws:
            for i, ch in enumerate(chunks):
                # 1x 实时节拍:第 i 块应在 t0 + i*0.1s 发出
                await asyncio.sleep(max(0.0, t0 + i * args.chunk_ms / 1000.0
                                        - time.perf_counter()))
                await ws.send(ch)
                try:
                    r = await asyncio.wait_for(ws.recv(), timeout=args.timeout)
                    if ttft is None:
                        ttft = time.perf_counter() - t0   # 首个部分结果
                except asyncio.TimeoutError:
                    pass                                  # 该块无返回(端点未到)
            await ws.send(json.dumps({"type": "end"}))    # 结束帧,按协议调整
            final = await asyncio.wait_for(ws.recv(), timeout=args.timeout)
        wall = time.perf_counter() - t0
        results.append({
            "stream": idx,
            "ttft_s": round(ttft, 3) if ttft else None,
            "rtf": round(wall / audio_sec, 3),            # >1 = 这路流跟不上实时
            "final_len": len(str(final)),
        })
    except Exception as e:
        results.append({"stream": idx, "error": repr(e)})


async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", required=True)
    ap.add_argument("--wav", required=True)
    ap.add_argument("--streams", type=int, default=16)
    ap.add_argument("--ramp", type=float, default=1.0, help="每流错峰秒数")
    ap.add_argument("--chunk-ms", type=int, default=100)
    ap.add_argument("--timeout", type=float, default=5.0)
    args = ap.parse_args()

    chunks = load_pcm16(args.wav)
    results = []
    t0 = time.perf_counter()
    await asyncio.gather(*[one_stream(i, args, chunks, results)
                           for i in range(args.streams)])
    wall = time.perf_counter() - t0

    ok = [r for r in results if "error" not in r]
    bad = [r for r in results if "error" in r]
    ttfts = sorted(r["ttft_s"] for r in ok if r["ttft_s"] is not None)
    rtfs = sorted(r["rtf"] for r in ok)
    p = lambda a, q: a[min(len(a) - 1, int(len(a) * q))] if a else float("nan")

    print(f"\n=== 目标 {args.url} | {args.streams} 路并发 | 音频 {len(chunks)*0.1:.0f}s ===")
    print(f"成功 {len(ok)} / 失败 {len(bad)} | 整轮墙钟 {wall:.0f}s")
    if ttfts:
        print(f"TTFT  p50={p(ttfts,0.5):.2f}s  p95={p(ttfts,0.95):.2f}s  max={ttfts[-1]:.2f}s")
        print(f"RTF   p50={p(rtfs,0.5):.2f}  p95={p(rtfs,0.95):.2f}  max={rtfs[-1]:.2f}"
              f"  | RTF>1 的流: {sum(r > 1 for r in rtfs)}(超实时=延迟堆积)")
    for r in bad[:5]:
        print("错误样例:", r)
    # 判定:本档通过 = 全部流 RTF ≤ SLO 且 TTFT p95 ≤ 阈值


if __name__ == "__main__":
    asyncio.run(main())

七、代码实例 ②:压测期间的服务器侧监控

# GPU 侧:每 5 秒采样利用率/显存/功耗(压测同时开)
nvidia-smi --query-gpu=timestamp,utilization.gpu,memory.used,power.draw \
           --format=csv -l 5 > gpu_$(date +%s).csv

# 显存增量法测"每路流开销":同一实例分别在 1/8/32 路稳态下取 memory.used 差分

# 音频准备(通话 8k → 16k):
sox call_8k.wav -r 16000 -c 1 call_16k.wav

# 阶梯加压驱动(bash 伪循环,每档 300s,档间冷却 60s)
for N in 1 2 4 8 16 32 64 128; do
  python3 asr_loadgen.py --url ws://host:9000/asr --wav call_16k.wav \
      --streams $N --ramp 0.5 | tee -a ladder.log
  sleep 60
done

八、容量判定:把两条边装回你的场景

压测做完,你会得到两个数:硬上限(RTF→1 的吞吐天花板:0.6B 报告口径 ≈ 2000 路——延迟不管时的物理极限)和工程上限(TTFT SLO 撞线处,按报告数据推算 0.6B 单实例约 16–64 路,取决于你的 SLO 松紧)。真正的答案在两者之间、由你的压测收敛:

检索练习

来源汇总(2026-09 核实)