用有效吞吐压测确定模型接口并发上限

大多数团队为 LLM endpoint 设定并发数时，凭的都是感觉，直到发生事故才发现真正的上限。没有测过的数字就是猜测，而当重试开始放大负载时，猜测的代价会很高。这次工作坊用一套可重复的逐级加压测试取代猜测，绘制 goodput 随施加并发数变化的曲线，让你能够给出有依据的并发上限和请求丢弃规则。下面是一份完整的教学提纲，包括练习、时间安排、可运行的测试程序，以及可以针对自有 gateway 重新执行的示例。

这次工作坊要验证的核心观点

Throughput（吞吐量）和 goodput（有效吞吐量）不是同一个指标，而在免费或共享 endpoint 上，混淆两者很快就会付出代价。Throughput 统计客户端发出的请求所得到的全部响应，包括调用方已经放弃等待后才到达的响应。Goodput 只统计在截止时间内到达、且携带可用内容的响应，这才是产品实际感受到的处理能力。同时跟踪这两个指标的逐级加压测试，通常会看到 goodput 先上升，再趋于平缓，最后下降，而 throughput 仍在继续攀升。这个转折点，就是你要找的拐点。

计时开始前需要准备什么

• 一个由你控制、位于客户端与模型之间的 gateway，哪怕只是一个 30 行的 HTTP 封装。
• 一台独立于被测机器的客户端机器，避免客户端的 CPU 争用扭曲 p95。
• 提前写下截止时间，例如短回答工作负载可以设为 12 秒。
• 一个 CSV 输出位置，以及写入小文件的权限，因为最终产物是一张数据表，不能只凭感觉。

在这里，免费模型 endpoint 和免费服务器选项确实能为这套方法提供帮助，而不只是营销卖点。把测试程序指向免费模型服务，就能在没有计费账户的情况下演练逐级加压测试；把负载生成器运行在免费服务器上，则可以让笔记本电脑退出关键路径。披露：本文是 MonkeyCode 产品推广活动的一部分。

练习 1——编写测试程序（15 分钟）

下面的测试程序是一份参考实现；我没有针对你的 endpoint 执行过它，因此请把它当作需要亲自运行验证的起点。它会逐级增加受限的并发请求数，为每个结果记录是否满足截止时间，并如实保留失败结果。

# ramp.py — unexecuted reference implementation
import asyncio, csv, json, statistics, time
import httpx

URL = "http://127.0.0.1:8080/v1/chat"   # your gateway, not the vendor URL
DEADLINE_S = 12.0
STEPS = [1, 2, 4, 8, 12, 16, 24, 32]     # offered concurrency, one step at a time
REQS_PER_STEP = 40

async def one(client, sem, idx, out):
    async with sem:
        t0 = time.perf_counter()
        code, ok = None, False
        try:
            r = await client.post(URL,
                json={"prompt": "ping", "max_tokens": 16},
                timeout=DEADLINE_S)
            code = r.status_code
            ok = (r.status_code == 200 and bool(r.text))
        except (httpx.TimeoutException, httpx.TransportError) as e:
            code = type(e).__name__
        out.append({"step": None, "idx": idx, "ok": ok, "code": code,
                    "ms": (time.perf_counter() - t0) * 1000})

async def ramp():
    rows = []
    async with httpx.AsyncClient(http2=False) as client:
        for c in STEPS:
            sem = asyncio.Semaphore(c)
            out = []
            t0 = time.perf_counter()
            await asyncio.gather(*[one(client, sem, i, out) for i in range(REQS_PER_STEP)])
            wall = time.perf_counter() - t0
            for r in out:
                r["step"] = c
            for r in out:
                r["wall_s"] = round(wall, 3)
            rows.extend(out)
            print(c, "concurrency ->", round(wall, 2), "s wall")
            await asyncio.sleep(2)   # let shared queues drain between steps
    return rows

if __name__ == "__main__":
    rows = asyncio.run(ramp())
    with open("ramp.csv", "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader(); w.writerows(rows)

练习 2——加入准入控制和截止时间传递（15 分钟）

对没有防护的 gateway 进行逐级加压，测到的是客户端，而不是整个系统。加入有界队列和截止时间请求头，让服务器能够尽早丢弃请求，避免把工作做完时，已经没有人在等待结果。

# gateway_guard.py — sketch, wire into your own framework
import time, asyncio

class Guard:
    def __init__(self, limit=8, max_wait_s=1.5):
        self.sem = asyncio.Semaphore(limit)
        self.max_wait_s = max_wait_s

    async def run(self, deadline_ms, fn, *a, **kw):
        budget = min(deadline_ms / 1000.0, self.max_wait_s)
        try:
            await asyncio.wait_for(self.sem.acquire(), timeout=budget)
        except asyncio.TimeoutError:
            return {"error": "shed", "reason": "queue_wait_exceeded"}
        try:
            return await fn(*a, **kw)
        finally:
            self.sem.release()

每个请求都携带 X-Deadline-Ms，并让 gateway 在调用下游前扣除自身已经消耗的时间。当剩余时间预算低于观测到的最慢调用耗时，就立即丢弃请求，返回 429，并附上 Retry-After 值。在 50 ms 内返回请求丢弃响应，比等待 12 秒后超时，对所有参与方的成本都更低，也能让逐级加压曲线如实反映系统表现。

练习 3——执行逐级加压测试并计算拐点（20 分钟）

分析只需要四行 pandas，公式比工具更重要。每一级的 goodput，等于在截止时间内成功返回的响应数，除以该级测试实际经过的秒数；拐点则是相较前一级，goodput 的提升幅度仍达到事先约定标准的最后一级。

import pandas as pd
df = pd.read_csv("ramp.csv")
df["in_deadline"] = df["ok"] & (df["ms"] <= 12000)
agg = df.groupby("step").agg(
    goodput=("in_deadline", "sum"),
    p95_ms=("ms", lambda s: s.quantile(0.95)),
    wall_s=("wall_s", "first"),
    errors=("ok", lambda s: int((~s).sum())),
)
agg["goodput_rps"] = agg["goodput"] / agg["wall_s"]
agg["gain_pct"] = agg["goodput_rps"].pct_change() * 100
print(agg.round(2))

使用书面规则，不要凭直觉：选择 goodput 增幅至少为 10%，且 p95 仍低于截止时间 60% 的最后一级作为停止点。人们经常跳过第二个条件，而恰恰是这个条件，能防止漂亮的吞吐量数字掩盖糟糕的尾延迟。

示例演练（待填写模板，并非实测数据）

下表特意没有填入真实数值；请将它复制到你的记录卡中，再根据自己的 CSV 填写。在这里展示编造的数字，会违背这次工作坊的初衷。

练习 4——写下丢弃与排队的判断规则（10 分钟）

把曲线转化为一张决策表，让值班工程师无需阅读分析过程，就能直接使用。

局限性，以及哪些人不适合使用这套方法

逐级加压测试测量的是共享 endpoint 当天的运行边界，结果不能直接套用到其他提供商、区域或一天中的其他时段。免费和共享容量可能因客户端无法观测的原因发生变化，因此提供商一侧发生任何变更后，都应重新测试，并把旧曲线当作历史记录。在没有阅读 endpoint 的服务条款或速率限制之前，不要进行激进的逐级加压测试；一旦持续出现 429 响应，就应立即停止。如果你需要合同约定的容量保证、多区域故障切换或租户级隔离，这次工作坊只能提供诊断，不能替代相应的工程建设。

75 分钟安排，以及你最终带走的成果

记录卡就是交付物：截止时间、拐点并发数、拐点处的 p95、请求丢弃规则，以及测试日期。一张标有日期的记录，比无人添加说明的 dashboard 更有价值，因为它能告诉下一位工程师，这个数字是否仍有参考意义。如果你有免费模型服务和免费服务器可用，先在那里完整执行一遍练习，把生产环境预算留给曲线确实证明有必要的改动。

如果要采取进一步行动，可以考虑屏蔽此人和／或举报滥用行为。