用probe对MonkeyCode免费端点进行重复调用与负载测试,量化429错误率与p99延迟;提供了评估免费AI服务的可复现方法论。
每周都有新的免费模型端点出现。AI 新闻周期放出一个模型。免费层几天后就跟上了。
所有人都急着基于它开发。几乎没人先测试一下。
我见过这种失败模式。Demo 能跑。原型能跑。
然后真实流量一来。端点就垮了。
免费层是有约束的。约束是可以测试的。所以我写了一个探针。
本周,我把它对准了 MonkeyCode 的免费层。它提供 10M token 和一台免费服务器。项目本身是开源的。
披露:本文是 MonkeyCode 产品推广合作的一部分。
一个核心问题驱动了这次测试。我应该基于这个构建吗?
Demo 什么都告诉不了你。它用一条 prompt、一次调用、零负载。它是一张截图,而不是一次测量。
探针不一样。它重复调用。它变化负载。它记录失败。它给你可以比较的数字。
跳过这一步的代价是隐性的。你看不到你从未触发的 429。你看不到你从未测量过的 p99。你只会在生产环境中、在最糟糕的时刻看到它们。
目标是做决策,不是凭感觉。Pass、warn 或 fail。就这样。
免费端点的失败是可以预测的。认证会坏。延迟会抖。错误会在负载下出现。JSON 会乱码回来。
我把这些失败模式变成了六个检查项。每一个都很便宜。每一个几秒钟就能跑完。
每个检查项 30 次调用。一个脚本。最后一张表。
脚本之前有一条规则。从你的代码实际会使用的网络环境运行探针。如果生产环境运行在其他地方,笔记本的延迟是假的。
MonkeyCode 的免费层包含一台服务器。这是运行探针的自然位置。干净的环境,干净的数字。
脚本使用 httpx 和 asyncio。不用框架。不用包装器。
"""
preflight_probe.py — six checks for any free model server.
Usage:
python preflight_probe.py --base-url https://.../v1 --api-key KEY --model MODEL
"""
import argparse
import asyncio
import json
import statistics
import time
import httpx
PROMPT = (
"Write a haiku about distributed systems. "
'Return only JSON: {"haiku": "your haiku here"}'
)
async def one_call(client, base_url, model):
payload = {
"model": model,
"messages": [{"role": "user", "content": PROMPT}],
"max_tokens": 150,
}
start = time.perf_counter()
try:
response = await client.post(
f"{base_url}/chat/completions", json=payload, timeout=60.0
)
return response.status_code, time.perf_counter() - start, response.text
except Exception as exc:
return None, time.perf_counter() - start, str(exc)
async def check_auth(client, base_url, api_key):
response = await client.get(
f"{base_url}/models", headers={"Authorization": f"Bearer {api_key}"}
)
return response.status_code, response.text[:120]
async def check_throughput(client, base_url, model):
status, elapsed, body = await one_call(client, base_url, model)
if status != 200:
return status, 0.0
content = json.loads(body)["choices"][0]["message"]["content"]
words = len(content.split())
return status, round(words / elapsed, 2)
async def check_latency(client, base_url, model, n=30):
results = await asyncio.gather(
*(one_call(client, base_url, model) for _ in range(n))
)
ok = sorted(elapsed for status, elapsed, _ in results if status == 200)
if not ok:
return {}, sum(1 for status, _, _ in results if status != 200)
p50 = statistics.median(ok)
p95 = ok[min(len(ok) - 1, int(len(ok) * 0.95))]
p99 = ok[min(len(ok) - 1, int(len(ok) * 0.99))]
return {"p50": round(p50, 2), "p95": round(p95, 2), "p99": round(p99, 2)}, 0
async def check_errors(client, base_url, model, n=30):
codes = {}
recovered = 0
for _ in range(n):
status, _, _ = await one_call(client, base_url, model)
codes[status] = codes.get(status, 0) + 1
if status in (429, 500, 502, 503):
await asyncio.sleep(2.0)
retry_status, _, _ = await one_call(client, base_url, model)
if retry_status == 200:
recovered += 1
return codes, recovered
async def check_concurrency(client, base_url, model, levels=(1, 4, 8, 16)):
report = {}
for level in levels:
results = await asyncio.gather(
*(one_call(client, base_url, model) for _ in range(level))
)
ok = [elapsed for status, elapsed, _ in results if status == 200]
errors = sum(1 for status, _, _ in results if status != 200)
report[level] = {
"p50": round(statistics.median(ok), 2) if ok else None,
"errors": errors,
}
return report
async def check_structured(client, base_url, model, n=30):
valid = 0
for _ in range(n):
status, _, body = await one_call(client, base_url, model)
if status != 200:
continue
try:
content = json.loads(body)["choices"][0]["message"]["content"]
json.loads(content)
valid += 1
except (json.JSONDecodeError, KeyError, TypeError):
pass
return valid, n
async def main():
parser = argparse.ArgumentParser(
description="Six-check probe for free model servers"
)
parser.add_argument("--base-url", required=True, help="API base URL")
parser.add_argument("--api-key", required=True)
parser.add_argument("--model", required=True)
args = parser.parse_args()
headers = {"Authorization": f"Bearer {args.api_key}"}
async with httpx.AsyncClient(headers=headers, timeout=60.0) as client:
print("== 1. auth ==")
print(await check_auth(client, args.base_url, args.api_key))
print("== 2. throughput (to
把它保存为 preflight_probe.py。这样运行:
python preflight_probe.py \
--base-url https://your-endpoint.example/v1 \
--api-key "$YOUR_KEY" \
--model your-model
给它大约十分钟。去冲杯咖啡。
这是我的运行结果表。一个下午、一个区域、一个 API key。
数字只有放在上下文中才有价值。所以这里给出上下文。端点是一个共享的免费层。模型是默认路由选择的。我没有调优 prompt 或重试。
四轮 Pass。两个 warn。没有硬失败。
Auth 很无聊。这是褒义。key 能用,我就过了。
Throughput 保持在每秒 18 token。不快。但对于聊天和批处理任务足够了。
Structured output 是惊喜。30 次完成中有 27 次解析为干净的 JSON。三次失败是可预测的。两个有尾部逗号。一个把 JSON 包在了 markdown 围栏里。
Errors 出现在第 4 轮检查。30 次调用中出现两个 429 和一个 503。每次重试都在两秒退避后恢复。
这就是共享免费层的模式。速率限制存在。执行得很客气。
Concurrency 说的也是同样的故事。在 8 个并发调用时,p50 翻倍了。在 16 时,两个调用返回 503。两个都在重试时恢复了。
服务器会降级。不会崩溃。这是一个有意义的区别。
这张表是我复用的部分。它把原始数字变成决策。
判决规则很简单。两个 warn?构建一个原型。任何 fail?走人。
让我把它用到 MonkeyCode 的免费层。两个 warn。那它就在原型区了。我会信任它用于批处理任务、内部工具和非关键路径。
我不会把它放在有不可预测负载的客户-facing API 后面。除非加上重试层和 fallback 模型。
我会从两个区域运行探针。免费层根据你所在位置不同路由也不同。
我还会测试流式输出。流式输出完全改变了延迟计算方式。首 token 比总时间更重要。
我会每周重新跑一次。免费层会漂移。上个月的 Pass 可能变成这个月的 Fail。
这个探针是一个快照,不是保证。我跑了一次、一个区域、一个下午。
免费层每周都在变。配额会动。模型会被换掉。每次发布前重新跑探针。
token 计数是一个代理。我用的是词数,不是真正的 tokenizer。用于比较没问题,用于计费不行。
我只测试了端点暴露的部分。我没有跑我看不到的硬件基准测试。
还有,你的 mileage 会不同。字面意思。不同区域、不同时间段、共享池不同负载。把我的数字当作样本,不是规格。
如果你需要硬 SLA,跳过这个方法。如果你承受不起重试,跳过它。如果用户期待个位数延迟,跳过它。
免费层是用来实验的,不是用来做承诺的。这样对待它们。
MonkeyCode 是开源的。免费层给你 10M token 加一台免费服务器。如果你好奇,用这个探针跑一下,比较数字。
免费层隐藏它们的限制。探针把猜测变成数字。这就是全部意义。