dev.to #ai新闻
不要相信第一个 token:免费模型服务器的流式延迟剖析
本文通过实测发现,在免费模型服务器上,流式请求的首 token 延迟可能高于非流式请求的总延迟,且流式请求可能中途断开。作者提供了一个简单的 Python 探针脚本,用于测量首 token 时间、token 间延迟和完成率,并建议在聊天 UI 中使用流式,而在 CI 和批处理任务中避免使用。
不要相信第一个Token:免费模型服务器上的流式延迟剖析
流式传输改变一切。我原本是这么以为的。然后我测量了一下。第一个Token是个谎言。
非流式请求掩盖了真实情况。它们一次性返回一大块数据。流式传输则返回涓涓细流。这股细流有它自己的延迟。免费服务器让这些延迟变得更糟。
我构建了一个探针。它发送一个流式请求。它记录每一个Token的到达时间。然后我把它指向了MonkeyCode的免费服务器选项。声明:本文是作为MonkeyCode产品推广的一部分准备的。
探针
脚本很小。它使用requests和iter_lines。没有框架。没有魔法。
#!/usr/bin/env python3
"""用于chat-completions端点的流式延迟探针。"""
import argparse
import json
import statistics
import time
import requests
def p95(values):
if not values:
return None
return sorted(values)[max(0, int(len(values) * 0.95) - 1)]
def run_probe(url, api_key, model, prompt="ping", max_tokens=128):
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": max_tokens,
"temperature": 0,
}
start = time.perf_counter()
first_token_ms = None
token_arrivals = []
error = None
done = False
try:
with requests.post(url, headers=headers, json=payload,
stream=True, timeout=60) as resp:
if resp.status_code != 200:
error = f"HTTP {resp.status_code}"
else:
for line in resp.iter_lines():
if not line or not line.startswith(b"data: "):
continue
data = line[6:]
if data == b"[DONE]":
done = True
break
json.loads(data) # 格式错误时抛出异常
now = time.perf_counter() - start
if first_token_ms is None:
first_token_ms = now * 1000
token_arrivals.append(now * 1000)
except Exception as exc:
error = f"{type(exc).__name__}: {exc}"
total_ms = (time.perf_counter() - start) * 1000
return {
"first_token_ms": first_token_ms,
"total_ms": total_ms,
"tokens": len(token_arrivals),
"inter_token_p95_ms": p95(
[b - a for a, b in zip(token_arrivals, token_arrivals[1:])]
),
"completed": done,
"error": error,
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--url", required=True)
parser.add_argument("--api-key", required=True)
parser.add_argument("--model", required=True)
parser.add_argument("--runs", type=int, default=10)
args = parser.parse_args()
results = []
for i in range(args.runs):
result = run_probe(args.url, args.api_key, args.model)
results.append(result)
print(f"run {i + 1}: {result}")
completed = [r for r in results if r["error"] is None and r["completed"]]
if completed:
print("\nSummary (completed runs only):")
print(f" first_token_ms median: "
f"{statistics.median(r['first_token_ms'] for r in completed):.0f}")
print(f" total_ms median: "
f"{statistics.median(r['total_ms'] for r in completed):.0f}")
print(f" inter_token_p95 median: "
f"{statistics.median(r['inter_token_p95_ms'] for r in completed):.0f}")
failed = [r for r in results if r["error"] or not r["completed"]]
if failed:
print(f"\nFailed/incomplete runs: {len(failed)}/{args.runs}")
if __name__ == "__main__":
main()python3 stream_probe.py \
--url https://your-endpoint/v1/chat/completions \
--api-key "$KEY" \
--model your-model \
--runs 10在不同时段运行它。免费服务器共享容量。早晨和晚上看起来不一样。
我测量到的结果
下面是一次示例运行。你的数字会有所不同。形态更重要。
| 指标 | 非流式 | 流式 |
|---|---|---|
| 首字节 / 首个Token | 1.1 秒 | 2.4 秒 |
| 总耗时 | 3.2 秒 | 4.8 秒 |
| Token间延迟 p95 | 不适用 | 180 毫秒 |
| 完成运行次数 | 10/10 | 9/10 |
第一个Token到达的时间比整个非流式请求还要慢。这让我很惊讶。流式传输增加了开销,而不是消除了开销。
为什么?免费服务器通常会缓冲。它们在内部生成完整响应。然后将其作为流重放。无论哪种方式,你都要付出延迟的代价。
三个重要的数字
首个Token时间。这是用户看到的。慢的首个Token感觉像坏了。快的首个Token感觉是活的。
Token间延迟p95。这是节奏。如果Token突发到达,UI会卡顿。如果它们稳定到达,UI会感觉流畅。
完成率。流式请求可能会中途死掉。没有状态码。没有错误。只有沉默。你的代码必须检测到这一点。
断连陷阱
非流式请求要么成功返回,要么失败。而流式请求可能两者兼有:先返回 200 状态码,然后在输出 20 个 token 后连接中断。
我的探针会检查 [DONE] 标记。如果该标记始终未出现,则本次运行不完整,应视为失败,重试整个请求,不要使用部分输出。
以下是我使用的判定规则:
- 收到 [DONE],无错误:成功。
- 未收到 [DONE],无错误:不完整,重试一次。
- HTTP 错误:失败,退避重试。
- 流中途异常:失败,退避重试。
谁应该使用流式
流式输出适合聊天界面。用户期望逐 token 输出,即使模型速度较慢,也能带来响应即时的体验。
流式输出对 CI 和批处理任务则不太合适。等待同样数量的 token 需要更长时间,增加了失败模式,还需要额外的解析工作。普通的 JSON 响应更简单。
免费模型服务器让这种权衡更加复杂。首个 token 更慢,完成率更低。在做出决定之前,先测量数据。
局限性
这个探针不是负载测试。它一次只发送一个请求,不测量并发,也不测试吞吐量。
它还假设使用标准的 SSE 格式。有些端点使用不同的分隔符,请查阅你的服务商文档。
我只针对一个免费端点进行了测试。你的情况可能有所不同。一天中的时段、模型大小、提示词长度都会影响结果。
结论
流式并不自动意味着更快。在免费服务器上,它可能更慢。首个 token 具有欺骗性,要测量整个流的完成情况。
运行探针,保留输出。它会告诉你流式是否值得,该设置多长的超时时间,以及何时应回退到非流式。
如果你运行了探针,请分享你的数据。我想看看其他免费端点的表现如何。