从压测视角窥见vllm推理与模型运行时细节

本文内容由Authropic Claude Opus 4-8模型生成。信息仅供参考，不构成专业建议。

尽管AI尽力确保信息的准确性，但并不能保证其完全无误或适用于所有场景，请结合实际情况审慎使用。

前言

本篇是《从零开始的大模型（Transformer）学习笔记》的补充。此前我们把视角停在”模型是什么、前向传播、训练显存”等基础性基础。

这一篇换一个入口——从压测调优往下钻，看看一个请求打进 vLLM 之后，GPU 内部到底在搬什么、算什么，以及 TTFT / TPOT / max_num_seqs / KV cache / preempt 参数在其运行时流程中承担着什么样的作用。

目标只有一个：把”一深挖就露馅”的那些点，从物理原理往下推通。

压测的本质，是”加压 → 读一堆指标（化验单）→ 定位瓶颈 → 动对应旋钮”。而要读懂这些指标，得回到推理时 GPU 内部那条 搬权重 → 算 token 的流水线。

一、前向传播

压测一个模型，本质就是在反复触发”前向传播”这一个动作。 搞懂前向里 GPU 在干嘛，压测的一切指标就都有根了。

• 训练：前向算预测 → 算 loss → 反向传播算梯度 → 更新权重。四步。
• 推理：只有前向传播这一步。输入过一遍网络，吐出”下一个 token 的概率”，挑一个。权重全程不变，没有反向、没有梯度、没有更新。

常见混淆：分布式训练里学的 All-Reduce 梯度同步是训练专属（N 张卡各算梯度要同步）。

压测推理服务根本用不上那套通信——推理不产生梯度。把训练的通信原语套到推理调优上，是典型露馅点。

二、GPU 的地图：HBM、SM、和那条”搬运带”

后面所有故事都在这张图上跑：

┌────────────────────── 一张 GPU ──────────────────────┐
│  【HBM 显存 48GB】= 仓库（大、相对慢）                 │
│  ┌──────────────────────────────────────────┐        │
│  │ 模型权重 ~30GB │ KV cache 池 │ 激活值临时区 │        │
│  └──────────────────────────────────────────┘        │
│           ▲                                           │
│           │  ← 这条"搬运带"的带宽 = 瓶颈根源           │
│           ▼                                           │
│  【SM 计算核心】= 工厂（小、极快）                      │
│  真正做矩阵乘法的地方，自带一点点超快片上缓存，        │
│  但装不下 30GB 权重                                    │
└───────────────────────────────────────────────────────┘

• HBM（High Bandwidth Memory，高带宽显存）：就是 nvidia-smi 里那几十 GB。相对 SM 它”慢”，但相对内存/硬盘已经飞快。
• SM（Streaming Multiprocessor，流式多处理器）：真正算矩阵乘法的单元，快到大部分时间在等仓库把数据搬过来。

关键就一句：权重只能躺在 HBM，SM 缓存太小装不下，所以每次算都得把权重从 HBM 搬进 SM。 这条”搬运带”的速度（显存带宽 GB/s）是后面所有故事的主角。

三、量化为什么让推理变快

量化是把权重从 FP16 压到 INT4。它加速推理的原因，常被归结成”算得少了”，但真正的关键在别处：

权重从 30GB 变成 ~8GB，那条 HBM 搬运带每次要搬的字节数少了 4 倍 → 搬得快 → SM 等待时间短 → 推理快。

量化的性能收益，主要不来自算力，来自减少搬运量。这也解释了下一节的 decode 为什么慢。

四、前向传播的本质：一连串矩阵乘法

“输入过每一层得到预测”这句话太抽象，具体到底层：

前向传播 = 从头到尾一连串矩阵乘法。 权重是矩阵，那前向传播就是把输入向量，一个接一个乘过这几百个权重矩阵。

以输入 “猪八戒” 为例，模型内部真实发生的：

"猪八戒" → 切成 token [猪, 八, 戒]
   ↓ Embedding(查表): 每个 token 变成一个向量（如 3584 维）
── 进入第 1 层 Transformer Block ──────────────────
   ↓ Attention:
   │   向量 × W_Q → Q      (矩阵乘法!)
   │   向量 × W_K → K      (矩阵乘法!)
   │   向量 × W_V → V      (矩阵乘法!)
   │   Q·K 点积算相关性 → softmax → 加权求和 V
   ↓ FFN: 结果 × W1 × W2   (更大的矩阵乘法!)
   ↓ 残差 + LayerNorm
── 第 2 层、第 3 层 … 几十层，每层都重复 ──
   ↓
最后一层输出 → × 大矩阵 → 词表里每个 token 的概率 → 挑最高 → 下一个 token

从头到尾，W_Q / W_K / W_V / W1 / W2 … 这几十层、几百个矩阵，就是那 30GB。每过一个 token，就要把这堆矩阵全部乘一遍——这就是”权重每次都要从 HBM 搬进 SM”的原因：前向的每一步都是”输入 × 权重矩阵”，SM 要算就得把权重调进来，30GB 装不进 SM，只能算一层搬一层。

五、同一个前向，跑出 prefill 和 decode 两个阶段

prefill 和 decode 不是两种技术，是每个请求必经的先后两个阶段：一个请求先 prefill 一次，再 decode 很多次，不是”选一个跑法”。

一个请求的一生（时间线）

以请求 “帮我写诗” 为例，到达推理模型时：

t0 ┃【PREFILL】只发生一次
   ┃  [帮,我,写,诗] 4 个 token 一次性并行过完所有层
   ┃  ├─ 副产品: 把这 4 个 token 每层的 K、V 存进 KV cache ★写入
   ┃  └─ 吐出第 1 个新 token: "春"
t1 ┃【DECODE】开始反复发生
   ┃  输入 "春" 这 1 个 token 过完所有层
   ┃  ├─ 复用刚存的 [帮我写诗] 的 KV ★读取（不重算!）
   ┃  ├─ 把 "春" 自己的 K、V 追加进 KV cache ★写入(+1)
   ┃  └─ 吐出 "风"
t2 ┃  输入 "风" → 复用 [帮我写诗春] 的 KV → 吐 "又"
   ┃  … 直到 EOS 或 max_tokens
   ┗━ 请求结束，这条序列的 KV cache 释放

它俩为什么能配合？ 因为是同一条请求里的前后手：prefill 先把 KV 铺好，decode 后面才有得复用。

同一个前向，两种模式

网络的层、权重完全一样，跑的是同一套代码。唯一区别是一次喂几个 token：

	一次喂几个 token	对 KV 的动作	瓶颈
prefill	整个 prompt（几百个）	写入（第一次算，存下来）	算力 compute-bound
decode	1 个	读历史 + 追加自己 1 个	带宽 memory-bound

关键澄清：decode 为什么”读”还要”+1”

decode 每步既读又写。新 token “春” 过每一层时，也要乘 W_Q/W_K/W_V 生成自己的 Q、K、V：拿 Q 去和历史 K 做注意力（读），同时把自己的 K、V 追加进缓存（写，+1）。

今天生成的 token，就是明天的历史。 它现在是新 token，下一步就变成”历史 token”要被后面的人查，所以必须把自己的 K/V 留下。这就是每 decode 一步 KV cache 线性 +1 的原因，也是”序列越聊越胖”的机制。

系统层面：它俩”同时”发生

单个请求内是先后；但 vLLM 同时服务很多请求，某一瞬间：请求 A 刚进来在 prefill，请求 B、C 已在 decode。A 的 prefill（吃算力大户）一插进来，可能挤了 B、C 的 decode，导致它们那一下变慢——这就是延迟长尾的来源之一。--enable-chunked-prefill 就是把大 prefill 剁成小块插空进行，别让它一口气把别人饿死。

六、decode 太亏怎么救：batching 与 max_num_seqs

几个容易混的名词，摆在一起对比：

词	指什么	比方
请求 request	一次用户调用	一位客人
序列 sequence	请求对应的那串 token，在 GPU 里活着的一条	客人的整张账单
batch（一批）	某个 decode step 被凑到一起同时算的那几条序列	这一轮同时服务的一桌客人

关系：1 请求 = 1 序列；max_num_seqs 条序列 = 一批的大小上限。

decode 逐 token 很亏：拉一遍 30GB 权重只喂 1 个 token。怎么救？——别只喂 1 个，凑一批一起喂：

把 32 条不同请求的 decode 凑进同一次前向：拉一遍权重，同时给 32 条各生成 1 个 token。带宽成本还是拉一遍，产出 ×32。

这就是 max_num_seqs=32 的物理意义，也是”decode 天生爱 batch”的原因。注意用词——不是”节约带宽”（总量一样），是摊薄（amortization）：同一次搬运服务了更多 token，每 token 的搬运成本被摊平。

而 prefill 不爱 batch：单条就已经喂几百 token 把算力吃满了，再堆只是排队。

七、天花板：KV cache 才是 max_num_seqs 的上限

既然 batch 越大吞吐越高，为什么不设成 1000？拦着你的是KV cache 吃显存。

每一条活着的序列，都拖着一坨自己的 KV cache 躺在显存里。序列越长（聊得越久），这坨越大；序列越多（batch 越大），总占用越高。

一条序列的 KV 占用 = 2(K和V) × 层数 × KV头数 × head_dim × 当前token数 × 每数字节

要塞 1000 条序列同时 decode → 1000 坨 KV 同时躺显存。KV 池就那么大，池满 → 新序列进不来（排队），或正在跑的被踢出去（抢占）。所以：

max_num_seqs 的真实天花板不是算力，是 KV 显存。你敢开多大的 batch，取决于显存能同时躺下多少坨 KV。

PagedAttention 与一个高频误区

PagedAttention 把 KV 切成 16-token 定长块，逻辑→物理映射像操作系统分页，消灭碎片、能塞更多并发，还让相同前缀的块共享（= prefix caching）。

关键误区：调小 --max-model-len 不能直接换来更多并发。 因为 PagedAttention 是按实际用到的 token 动态分块分配 KV，不是一上来按 max-model-len 预留满。max-model-len 只是”单条最长能到多少”的上限，真正决定占用的是序列实际长到多少。

松绑 KV 天花板的正确手段：--kv-cache-dtype fp8（每坨 KV 从 16bit 压到 8bit，同池躺下两倍序列）、调大 --gpu-memory-utilization（单卡吃更多显存）、少开序列（调小 max_num_seqs，牺牲吞吐）。

八、抢占（preemption）：过载时的应急动作

KV 池住满了、还有序列要追加 K/V 时，调度器被迫踢掉某条正在跑的序列腾地方，这就是抢占。两种处理：

• Recompute（重算，默认）：丢掉这条序列已存的 KV，打回等待队列，等有空位再从头重新 prefill。之前的 prefill 白算，等于返工。
• Swap（换出）：把 KV 搬到 CPU 内存暂存，等空位再搬回。走 PCIe，慢，但比重算省。

不管哪种，抢占都是纯亏。所以 num_preemptions_total > 0 = 这个负载/配置已经过载，是调优时最明确的”踩过线”信号。被抢占的序列会卡住、（若重算）重新 prefill 才恢复吐字，直接制造 TPOT 长尾。

九、压测客户端的指标

p50 / p95 / p99：为什么不用平均值

把 100 条请求的延迟从小到大排队：p50 = 第 50 位（中位数），p95 = 第 95 位，p99 = 第 99 位。

平均值会被”大多数正常请求”稀释，掩盖少数惨案。100 条里 99 条 0.1s、1 条 10s，平均才 0.2s 看着很美——但那个用户等了 10s，体验崩了。p99 专门盯这种长尾。

p50 看典型体验，p99/p95 看最惨那批有多惨。p50 和 p99 差距越大 = 长尾越严重 = 系统越不稳。生产 SLA 通常写 “p99 < 500ms”，不写平均值。

TTFT 与 TPOT

• TTFT（Time To First Token） = 请求发出 → 收到第 1 个 token。物理来源：排队 + prefill 跑完整个 prompt。用户感知的”反应快不快”。prompt 越长 TTFT 越大。
• TPOT（Time Per Output Token） = 第一个字之后，平均每蹦一个新字的间隔。算法上是 (总生成时间 - TTFT) / (token数 - 1)——减 1 是因为第 1 个 token 已算进 TTFT。物理来源：decode 阶段每步搬权重的时间。用户感知的”说话语速顺不顺”。
• 只吐 1 个 token 时 TPOT 算不出（没间隔），所以压测常开 ignore_eos 强制吐满输出长度。

TTFT = 等它开口（prefill 主导）。TPOT = 它说话的语速（decode 主导）。 两个阶段，两个瓶颈，必须分开看。

吞吐与服务端指标

• throughput_tok_s：系统每秒吐多少 token（整体产能）。
• 从 vLLM /metrics 抓的四张”化验单”：

指标	含义	怎么读
kv_cache_usage	KV 池填充率 0~1	越接近 1 越吃紧
num_requests_running	同时 decode 的序列数	应 ≈ max_num_seqs
num_requests_waiting	排队等准入数	>0 = 调度打满，开始排队
num_preemptions_total	累计抢占次数	>0 = 过载

长尾是”病症”，preempt 是一张”化验单”：化验单阴性（=0），就把”抢占”这个病因划掉，往别处（prefill 插队、共享卡算力争抢）查。压测盯一堆指标，就是用每张专项化验单，把模糊的”变慢了”拆成能定位的具体病因。

别忘了先看 fail：一旦 >0，说明服务已扛不住，此时延迟数据要打折看（最惨的请求可能直接失败没进 p99 统计，存在幸存者偏差）。

十、调优方法论：一切都在找拐点（knee）

负载从低往高加，系统走过三个区：

吞吐↑                         ┌── 饱和甚至倒退（过载区）
    │                    ╭────┘
    │              ╭─────╯ ← knee 拐点（甜点区在附近）
    │      ╭───────╯
    │  ╭───╯  （欠载区）
    └──┴────────────────────────→ 负载

区	仪表盘特征
欠载	waiting=0、kv_usage 低、preempt=0、延迟平、吞吐还在涨
拐点	waiting 刚 >0、吞吐涨势变缓、p99 开始翘、preempt 还=0
过载	waiting 堆积、kv_usage→1、preempt>0、p99 爆炸、吞吐饱和/倒退、fail>0

实操：一档档加压

方法论一句话：固定一切，只动一个变量，一路加到过载，回头取拐点前那一档。 两根轴分开扫：

• 轴 A — 扫客户端并发（定配置、测容量）：保持模型参数不变，并发 32→64→128→256→512，每档读仪表盘。waiting 刚 >0、吞吐涨势缓、preempt=0 那档就是拐点。
• 轴 B — 扫 max_num_seqs（定负载、测最优配置）：它是启动参数不能热改，8→16→32→64→128 每值重启、同负载压。判据：吞吐增益 <5% 但 ttft_p99 增幅 >20% 那档，就是过了拐点。

两条铁律：短/长 prompt 各扫一遍（长 prompt prefill 重，拐点更早）；按 SLA 选工作点（延迟敏感取拐点偏左留余量，吞吐优先取拐点处甚至偏右）。

共享卡的边界

在共享 GPU 上压测要记住：调度层现象（waiting/kv/preempt/拐点位置）照测有效，因为一个模型的 KV 池是它独占的，别人抢不走；但延迟绝对值（tpot_p99 几毫秒）测不准，别的进程抢 SM/带宽会污染。要落 SLA 数字，必须清场独占再测一遍。

收束：一条打通的因果链

【模型】权重=几百个浮点矩阵 30GB，躺在 HBM
   ↓
【前向传播】输入 × 权重矩阵 → 吐"下个 token 概率"；推理=纯前向，无反向
   ↓ 同一套前向，两种跑法：
 PREFILL：整 prompt 并行 → 吃算力 → 写 KV → 决定 TTFT
 DECODE ：每次 1 token → 等带宽 → 读历史 KV + 写自己(+1) → 决定 TPOT
   ↓ decode 太亏
【batching】凑一批一起算，搬一次权重喂多个 token → 摊薄 = max_num_seqs
   ↓ 为什么不能开到 1000
【KV cache 天花板】每条活序列拖一坨 KV，越聊越胖；池满 → 抢占 preempt
   松绑：fp8 压 KV / 调 gpu-mem-util /（不是调 max-model-len!）
   ↓ 压到哪了
【四张化验单】waiting / kv_usage / preempt / running + p50/p99 长尾 + fail
   ↓ 调优动作
【找拐点】固定一切扫一个变量，加到过载，取"waiting 刚动 / preempt 还没冒"那档

从”模型权重是什么”到”压测该看哪个数、动哪个旋钮”，是一条连续的因果链，不是散点。decode 为什么慢、KV 为什么是天花板、p99 为什么比平均值重要、max_num_seqs 与 max_num_batched_tokens 差在哪、抢占是怎么回事——这些点都能顺着 搬权重 → 算 token 这条主线往下推。

理论到这里告一段落，剩下的交给数据：拿压测工具扫一轮，等拐点真的出现，再把这套方法论对到自己的实测上。

附：压测客户端代码

文中所有指标都来自下面这个 async 流式压测客户端——逐块记录 token 到达时间算出 TTFT/TPOT，同时后台采样 vLLM /metrics（KV 用量、队列、抢占、prefix 命中率）。用法：

python3 loadgen.py --url http://127.0.0.1:9202 --model gemma4 \
    -c 32 -d 30 --in-tokens 128

#!/usr/bin/env python3
"""async 流式压测客户端: 精确测量 vLLM 的 TTFT / TPOT。

指标定义(行业标准):
  TTFT = 发出请求 -> 收到第 1 个 token 的时间   (由 prefill 阶段主导)
  TPOT = (总生成时间 - TTFT) / (输出 token 数 - 1)  (由 decode 阶段主导)
"""
import argparse
import asyncio
import json
import re
import statistics
import time
from dataclasses import dataclass, field

import httpx

# vLLM /metrics 里反映显存/调度压力的指标
_M = {
    # vLLM 0.23+ 把 gpu_cache_usage_perc 改名为 kv_cache_usage_perc,两种都匹配
    "cache": re.compile(r"^vllm:(?:gpu|kv)_cache_usage_perc\S*\s+([0-9.eE+-]+)", re.M),
    "running": re.compile(r"^vllm:num_requests_running\S*\s+([0-9.eE+-]+)", re.M),
    "waiting": re.compile(r"^vllm:num_requests_waiting\S*\s+([0-9.eE+-]+)", re.M),
    "preempt": re.compile(r"^vllm:num_preemptions_total\S*\s+([0-9.eE+-]+)", re.M),
    # prefix cache 命中率(累计计数, 用运行窗口内的增量算命中率)
    "pfx_q": re.compile(r"^vllm:(?:gpu_)?prefix_cache_queries(?:_total)?\S*\s+([0-9.eE+-]+)", re.M),
    "pfx_h": re.compile(r"^vllm:(?:gpu_)?prefix_cache_hits(?:_total)?\S*\s+([0-9.eE+-]+)", re.M),
}


@dataclass
class ReqResult:
    ttft: float = 0.0            # 秒
    tpot: float = 0.0            # 秒/token
    total: float = 0.0           # 秒
    out_tokens: int = 0
    ok: bool = False
    err: str = ""


@dataclass
class Stats:
    results: list = field(default_factory=list)

    def add(self, r: ReqResult):
        self.results.append(r)

    def summary(self) -> dict:
        good = [r for r in self.results if r.ok]
        if not good:
            return {"ok": 0, "fail": len(self.results)}

        def pct(vals, p):
            vals = sorted(vals)
            k = int(round((len(vals) - 1) * p / 100))
            return vals[k]

        ttfts = [r.ttft for r in good]
        tpots = [r.tpot for r in good if r.out_tokens > 1]
        return {
            "ok": len(good),
            "fail": len(self.results) - len(good),
            "ttft_p50": round(statistics.median(ttfts), 4),
            "ttft_p99": round(pct(ttfts, 99), 4),
            "tpot_p50": round(statistics.median(tpots), 4) if tpots else None,
            "tpot_p99": round(pct(tpots, 99), 4) if tpots else None,
            "throughput_tok_s": round(sum(r.out_tokens for r in good) /
                                      max(sum(r.total for r in good) / len(good), 1e-9), 1),
        }


# 用重复词粗略凑够目标输入 token 数(避免依赖 tokenizer;1 词≈1 token 的近似)。
# prefix_shared=True 时所有请求用同一段前缀 -> 专门用来观察 prefix caching 效果。
_PREFIX = "You are a helpful assistant. " * 8


def make_prompt(in_tokens: int, prefix_shared: bool) -> str:
    head = _PREFIX if prefix_shared else f"session-{time.time_ns()} "
    filler = "hello world " * max(1, (in_tokens - len(head.split())) // 2)
    return head + filler


async def one_request(client: httpx.AsyncClient, url: str, model: str,
                      prompt: str, max_out: int, ignore_eos: bool = True) -> ReqResult:
    """发一条流式请求, 逐块记录 token 到达时间, 算出 TTFT/TPOT。"""
    r = ReqResult()
    payload = {
        "model": model,
        "prompt": prompt,
        "max_tokens": max_out,
        "temperature": 0.0,        # 固定输出长度, 让 TPOT 可比
        "ignore_eos": ignore_eos,  # 忽略 EOS 强制生成满 max_out, 否则模型早停导致 TPOT 无数据
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    t0 = time.perf_counter()
    first_t = None
    n = 0
    try:
        async with client.stream("POST", url, json=payload,
                                 timeout=httpx.Timeout(300.0, connect=10.0)) as resp:
            if resp.status_code != 200:
                r.err = f"HTTP {resp.status_code}"
                await resp.aread()
                return r
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data.strip() == "[DONE]":
                    break
                chunk = json.loads(data)
                choices = chunk.get("choices") or []
                if choices and choices[0].get("text"):
                    if first_t is None:
                        first_t = time.perf_counter()   # 第一个 token 落地 = TTFT
                    n += 1
        now = time.perf_counter()
        if first_t is None:
            r.err = "无 token 返回"
            return r
        r.ttft = first_t - t0
        r.total = now - t0
        r.out_tokens = n
        r.tpot = (now - first_t) / (n - 1) if n > 1 else 0.0
        r.ok = True
    except Exception as e:                              # 压测客户端要吞异常继续
        r.err = f"{type(e).__name__}: {e}"
    return r


async def _sample_metrics(client: httpx.AsyncClient, base_url: str,
                          stop: asyncio.Event, samples: list, interval: float):
    """压测期间周期性抓 /metrics, 记录 KV 用量/队列/抢占, 反映显存与调度压力。"""
    url = base_url.rstrip("/") + "/metrics"
    while not stop.is_set():
        try:
            resp = await client.get(url, timeout=5)
            text = resp.text
            row = {}
            for k, rx in _M.items():
                m = rx.search(text)
                row[k] = float(m.group(1)) if m else None
            samples.append(row)
        except Exception:
            pass
        try:
            await asyncio.wait_for(stop.wait(), timeout=interval)
        except asyncio.TimeoutError:
            pass


def _mem_summary(samples: list) -> dict:
    """把采样点汇总成峰值/均值, 便于看 max_num_seqs 对显存的影响。"""
    if not samples:
        return {"metrics": "无(该端点无 /metrics 或未采到)"}
    def peak(k):
        vals = [s[k] for s in samples if s.get(k) is not None]
        return round(max(vals), 3) if vals else None
    def avg(k):
        vals = [s[k] for s in samples if s.get(k) is not None]
        return round(sum(vals) / len(vals), 3) if vals else None
    # prefix cache 命中率: 用窗口内(末采样-首采样)的增量算, 排除历史累计干扰
    def window_delta(k):
        vals = [s[k] for s in samples if s.get(k) is not None]
        return (vals[-1] - vals[0]) if len(vals) >= 2 else None
    q = window_delta("pfx_q")
    h = window_delta("pfx_h")
    hit_rate = round(h / q, 3) if (q and q > 0 and h is not None) else None
    return {
        "kv_cache_usage_peak": peak("cache"),   # KV 池用量峰值(0~1), 越接近1越吃紧
        "kv_cache_usage_avg": avg("cache"),
        "running_peak": peak("running"),        # 同时在跑序列数峰值(应≈max_num_seqs上限)
        "waiting_peak": peak("waiting"),         # 排队数峰值, >0 说明并发打满了调度
        "preemptions_total": peak("preempt"),   # 抢占次数, >0 说明 KV 不够被迫换出重算
        "prefix_cache_hit_rate": hit_rate,      # 本轮 prefix cache 命中率, 随机前缀应≈0
    }


async def run_load(base_url: str, model: str, concurrency: int, duration: float,
                   in_tokens: int, max_out: int, prefix_shared: bool,
                   sample_metrics: bool = True, ignore_eos: bool = True):
    """固定并发数持续压测 duration 秒, 边跑边补新请求(模拟持续负载)。
    同时后台采样 /metrics, 返回 (Stats, 显存汇总dict)。"""
    url = base_url.rstrip("/") + "/v1/completions"
    stats = Stats()
    deadline = time.perf_counter() + duration
    limits = httpx.Limits(max_connections=concurrency + 8,
                          max_keepalive_connections=concurrency + 8)
    samples: list = []

    async with httpx.AsyncClient(limits=limits) as client:
        stop = asyncio.Event()
        sampler = None
        if sample_metrics:
            sampler = asyncio.create_task(
                _sample_metrics(client, base_url, stop, samples, 2.0))

        async def worker():
            while time.perf_counter() < deadline:
                prompt = make_prompt(in_tokens, prefix_shared)
                stats.add(await one_request(client, url, model, prompt, max_out, ignore_eos))

        await asyncio.gather(*[worker() for _ in range(concurrency)])
        stop.set()
        if sampler:
            await sampler
    return stats, _mem_summary(samples)


def main():
    ap = argparse.ArgumentParser(description="vLLM 流式压测客户端 (TTFT/TPOT)")
    ap.add_argument("--url", default="http://127.0.0.1:8000", help="vLLM base url")
    ap.add_argument("--model", required=True, help="模型名(vLLM --served-model-name)")
    ap.add_argument("-c", "--concurrency", type=int, default=16)
    ap.add_argument("-d", "--duration", type=float, default=30.0, help="压测秒数")
    ap.add_argument("--in-tokens", type=int, default=128, help="输入长度: 128短 / 4096长")
    ap.add_argument("--max-out", type=int, default=128, help="输出 token 数(固定以可比)")
    ap.add_argument("--prefix-shared", action="store_true", help="所有请求共享前缀(测prefix cache)")
    ap.add_argument("--no-metrics", action="store_true", help="不采集 /metrics 显存指标")
    ap.add_argument("--respect-eos", action="store_true",
                    help="尊重模型 EOS 早停(默认 ignore_eos 强制生成满 max_out 以测 TPOT)")
    ap.add_argument("--json", action="store_true", help="只输出 JSON")
    args = ap.parse_args()

    stats, mem = asyncio.run(run_load(
        args.url, args.model, args.concurrency, args.duration,
        args.in_tokens, args.max_out, args.prefix_shared,
        sample_metrics=not args.no_metrics, ignore_eos=not args.respect_eos))
    s = stats.summary()
    s.update(mem)                       # 延迟 + 显存合并输出
    if args.json:
        print(json.dumps(s))
    else:
        print(f"并发={args.concurrency} 输入={args.in_tokens}tok "
              f"-> {json.dumps(s, indent=2, ensure_ascii=False)}")


if __name__ == "__main__":
    main()