测试免费模型 API 真实并发能力的方法

免费模型 API 返回 200 状态码，只能说明服务器接收了一个请求，并不能证明你请求的并发量是真实存在的。

如果你在还没有测量出有效并发量之前就贸然添加 Worker、重试机制或队列，那就是在用一个想象中的数字去做调优。

为什么 HTTP 状态码会掩盖真实的并发上限

200 状态码是针对每个请求的，而不是一个吞吐量信号。

一个网关可以同时接收很多个 Socket，却每次只执行其中一个。

p50 延迟可能一直保持低位，而 p95 延迟在排队压力下却会暴涨。

免费层级可能根本不公布并发数或队列限制。

很多失败看起来像是模型问题，实际上却是并发问题。一个请求超时，只是因为它在其他请求后面排队等着。模型压根没看到什么坏 Prompt。测量有效并发量，可以把模型层面的失败和队列层面的失败区分开来。

用固定的小 Prompt 做一次并发扫描

记录完成的请求数、墙上时间（wall time）、p50、p95 和最大延迟。将第一次运行的结果与每次增加并发后的结果做对比。

用环境变量来配置端点，这样脚本就能适配任何 OpenAI 风格的模型 API。

import asyncio
import os
import time

import httpx

BASE_URL = os.getenv("MODEL_URL", "https://free-model.example/v1/chat/completions")
TOKEN = os.getenv("MODEL_TOKEN", "")
MODEL = os.getenv("MODEL_NAME", "model")
CONCURRENCIES = [1, 2, 4, 8]
PROMPT = "Return the single word ok."

async def one_request(client, _):
    started = time.perf_counter()
    try:
        response = await client.post(
            BASE_URL,
            headers={"Authorization": f"Bearer {TOKEN}"},
            json={
                "model": MODEL,
                "messages": [{"role": "user", "content": PROMPT}],
                "max_tokens": 8,
            },
        )
        return {
            "ok": response.status_code == 200,
            "status": response.status_code,
            "latency": time.perf_counter() - started,
        }
    except Exception as exc:
        return {
            "ok": False,
            "status": type(exc).__name__,
            "latency": time.perf_counter() - started,
        }

def percentile(values, p):
    ordered = sorted(values)
    index = int((len(ordered) - 1) * p / 100)
    return ordered[index]

async def run_sweep(concurrency):
    async with httpx.AsyncClient(timeout=60) as client:
        started = time.perf_counter()
        results = await asyncio.gather(
            *(one_request(client, i) for i in range(concurrency))
        )
        wall = time.perf_counter() - started
    return results, wall

async def main():
    baseline = None
    for concurrency in CONCURRENCIES:
        results, wall = await run_sweep(concurrency)
        latencies = sorted(result["latency"] for result in results)
        ok = sum(1 for result in results if result["ok"])
        if baseline is None:
            baseline = latencies[len(latencies) // 2] or wall
        completed = len(results)
        effective = round(baseline * completed / wall, 2) if wall else completed
        print(
            f"c={concurrency} ok={ok}/{completed} wall={wall:.2f}s "
            f"p50={percentile(latencies, 50):.2f}s "
            f"p95={percentile(latencies, 95):.2f}s "
            f"max={latencies[-1]:.2f}s effective={effective}"
        )
        await asyncio.sleep(2)

asyncio.run(main())

用 c=1 的墙上时间作为基准 T。

有效并行度：T * completed / wall。数值接近 1，说明请求被串行执行了。数值接近并发数，说明端点真正在并行运行它们。

接近 1，说明请求被串行执行了。

接近并发数，说明端点真正在并行运行它们。

当有效并行度不再增长时，继续增加 Worker 只会增加队列压力。

ok 数量下降或出现 429，就说明你已经越过了限制。

p95 远高于 p50，说明即使 p50 看起来还行，排队现象已经发生了。

示例解读，不是基准测试

这个端点在两个 Worker 之后就无法提供真正的并行度了。c=8 看起来更差，是因为有两个请求失败了，而且 p95 飙升过了七秒。答案不是增加更多 Worker，而是把信号量设为 2。

应用测量结果

把 Worker 池的规模锁定在有效并发量接近并发数的那个最大并发档位。

在 Worker 池前面加一个队列，而不是直接扩大池子。

把重试机制放在有界路径之外。只对在测量过的、有界的 Worker 数量下失败的请求进行重试。

每次配额变更、区域变更或模型变更之后，都要重新跑一次扫描。

用一个小信号量来做限流：

sem = asyncio.Semaphore(2)

async def bounded_request(client, i):
    async with sem:
        return await one_request(client, i)

MonkeyCode 的位置

免责声明：这篇文章是 MonkeyCode 产品推广的一部分。

运营商提供的可用性：MonkeyCode 提供免费模型访问和免费服务器选项。如果你用这个工具对准 MonkeyCode 的免费模型端点，请查阅最新文档获取准确的 URL、令牌格式和模型名称。不要从一个通用的 OpenAI 示例中假设这些细节。

无论哪种情况，这个探测工具的价值都是一样的：它把"我应该用多少个 Worker？"这个问题，从猜测变成了一个有测量依据的数字。

这是一次容量探测，不是质量基准测试。

免费端点的配额、区域和队列行为都可能发生变化。结果只是一个时间点的快照。

Prompt 长度、令牌限制和输出长度都会影响延迟。请使用一个固定的、有代表性的Payload。

单一客户端所在位置无法观察到施加在其他地方的速率限制。

不需要运行扫描的情况

脚本会发起真实的请求。请保持扫描规模小一些，并且尊重已公布的限制。

你已经有服务端的指标来观察队列深度、并发量和 p95。

端点已经公布了准确的并发数或速率限制。

端点是付费的或昂贵的，所以额外的探测流量会消耗真金白银。

你在评估的是输出质量，而不是容量。

结论

从一个 Worker 开始，跑一遍扫描，然后把 Worker 数量锁定在测量得出的有效并发量上，再去构建重试层。