流式输出提升的是用户感知速度而非模型吞吐量;应追踪三个指标:TTFT、中位Token间隔、尾延迟,单追TTFT会漏掉后半段真实瓶颈。
你的聊天界面感觉很快。你的批处理任务却很慢。同一个模型,同一个服务器。
区别通常在于流式传输。流式传输改变了客户端最先看到什么,但没有改变模型思考的速度。
大多数团队只计时一个数字:第一个 token。这个数字可能只解释了总等待时间的三分之一。剩下的隐藏在 token 之间的空隙里。
这是一个破除迷思的 FAQ。先讲论断,再讲证据,最后给出修正后的心智模型。最后可以用探针脚本验证。
论断:"设置 stream=True,整个调用就会加速。"
证据:总时间基本不变。流式传输只是把第一个字节提前了。生成的工作量完全相同。客户端感受到的是速度,而不是秒表。
修正后的模型:流式传输是一种感知工具,不是吞吐量工具。用在有人等待的地方,丢弃用在衡量管道的场景。
论断:"我的 SLA 是 p95 TTFT,其他都是噪音。"
证据:总响应时间有两个部分。首先是首个 token,然后是之后的流。20 个 token 的回答?TTFT 占据预算。500 个 token 的回答?间隔占据它。只追踪一个数字会掩盖后半段。
修正后的模型:追踪三个数字。TTFT、token 间间隔中位数、总时间。优化哪个取决于用户在哪里等待。
论断:"我们每个 token 刷新一个 chunk,UI 感觉很活跃。"
证据:每个 chunk 都要付出一次网络请求、一次 JSON 解析和一次 UI 更新的代价。小 chunk 在负载下序列化很差。很多免费服务器本来就会对 token 进行批处理。chunk 数量和吞吐量是两个不同的概念。
修正后的模型:衡量 tokens per second,不是 chunks per second。有时候缓冲两到三个 token 比逐个喷发感觉更顺畅。
论断:"我跳过超时处理,流会保持连接活跃。"
证据:流超时是基于空闲的。如果服务器每 25 秒输出一个 token,就会杀死一个 15 秒空闲超时的连接。更糟的是:长流会占用一个连接池槽位。其他请求会排在你这个免费流后面,造成安静的拥堵。
修正后的模型:基于间隔超时,而不是连接超时。如果提供商发送 keep-alive ping,就监测它们。把连接空闲和生成活跃分开。
论断:"用户点击停止,我们关闭 socket,钱省了。"
证据:关闭你的 socket 不会停止服务器的循环。有些提供商在生成完毕后仍会向孤立的尾部计费。其他的在 chunk 之间检查中止信号,而不是在 chunk 内部。你的客户端停了,服务器没有。
修正后的模型:把中止当作 UX,而不是成本控制。在服务端用 max_tokens 限制生成。然后对比 API 报告的用量。不要假设计么。
这个脚本用两种模式运行同一个提示词。它打印中位数 TTFT、中位数总时间和中位数 token 间隔。你需要 Python 3 和 requests 包。每个端点大约十分钟。
import statistics
import time
import requests
ENDPOINT = "https://your-endpoint/v1/chat/completions"
MODEL = "your-model"
HEADERS = {"Authorization": "Bearer $OPENAI_API_KEY"}
PROMPT = "Explain, in 150 words, why free-tier LLM latency is noisy."
TRIALS = 5
def measure(stream: bool) -> dict:
payload = {
"model": MODEL,
"messages": [{"role": "user", "content": PROMPT}],
"max_tokens": 300,
"stream": stream,
}
started = time.perf_counter()
first = None
gaps = []
last = started
resp = requests.post(ENDPOINT, json=payload, headers=HEADERS, stream=True)
if not stream:
# Time until the first byte, then read the body.
for _ in resp.iter_content(chunk_size=1):
first = time.perf_counter() - started
break
resp.json()
total = time.perf_counter() - started
return {"ttft": first, "total": total, "gap_median": None}
for line in resp.iter_lines():
if not line or not line.startswith(b"data:"):
continue
if line[5:].strip() == b"[DONE]":
break
now = time.perf_counter()
if first is None:
first = now - started
else:
gaps.append(now - last)
last = now
total = time.perf_counter() - started
return {
"ttft": first,
"total": total,
"gap_median": statistics.median(gaps) if gaps else None,
}
for stream in (False, True):
rows = [measure(stream) for _ in range(TRIALS)]
label = "stream" if stream else "plain"
print(f"--- {label} ({TRIALS} trials) ---")
print(f"median TTFT : {statistics.median(r['ttft'] for r in rows):.2f}s")
print(f"median total: {statistics.median(r['total'] for r in rows):.2f}s")
gaps = [r["gap_median"] for r in rows if r["gap_median"]]
if gaps:
print(f"median gap : {statistics.median(gaps):.2f}s")
如果提供商在一个 chunk 里批处理多个 token,gap_median 测量的就是 chunk 而不是 token。这仍然是你 UI 看到的数字,所以这是诚实的指标。
输出形状像这样。仅作说明,不是基准。先在你的端点上运行。
注意这个模式。首个 token 变快了,总时间保持不变。这就是整篇文章的核心论点,浓缩在一张表里。
我的经验法则:管道保持非流式,只在面向 UI 的路径上流式传输。
我先把这个探针指向 MonkeyCode 通过免费服务器选项提供的免费模型访问。免费层也值得在信任之前先测量。披露:本文是 MonkeyCode 产品推广的一部分。该脚本与提供商无关,任何 OpenAI 兼容的端点都可以工作。
你是配额饥渴型。10 次调用可能吃掉一天的免费层配额。设置 TRIALS = 2。
你的提供商不支持 SSE。流式路径会挂住。先测试一次调用。
你需要服务端指标。这是客户端的尺子。用提供商仪表盘看另一半。
五次试验是样本,不是普查。免费服务器设计上就很吵。当你的预算允许时,把 TRIALS 提高到二十。结果不在区域、模型或时间之间迁移。在你自己的端点上重新运行。不要把上面的表当作基准引用。它是一个形状,不是证明。
下次有新的免费模型上线时,在接入之前运行这个探针。首个 token 和最后一个 token 之间的间隔会让你惊讶。
你的用户等待的是最后一个 token。你的仪表盘也应该如此。