← 返回信息流

dev.to #ai新闻

不要相信第一个 token:免费模型服务器的流式延迟剖析

dev.to作者:Jordan Huang教程评测AI评分:50/100

本文通过实测发现,在免费模型服务器上,流式请求的首 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

在不同时段运行它。免费服务器共享容量。早晨和晚上看起来不一样。

我测量到的结果

下面是一次示例运行。你的数字会有所不同。形态更重要。

指标非流式流式
首字节 / 首个Token1.1 秒2.4 秒
总耗时3.2 秒4.8 秒
Token间延迟 p95不适用180 毫秒
完成运行次数10/109/10

第一个Token到达的时间比整个非流式请求还要慢。这让我很惊讶。流式传输增加了开销,而不是消除了开销。

为什么?免费服务器通常会缓冲。它们在内部生成完整响应。然后将其作为流重放。无论哪种方式,你都要付出延迟的代价。

三个重要的数字

首个Token时间。这是用户看到的。慢的首个Token感觉像坏了。快的首个Token感觉是活的。

Token间延迟p95。这是节奏。如果Token突发到达,UI会卡顿。如果它们稳定到达,UI会感觉流畅。

完成率。流式请求可能会中途死掉。没有状态码。没有错误。只有沉默。你的代码必须检测到这一点。

断连陷阱

非流式请求要么成功返回,要么失败。而流式请求可能两者兼有:先返回 200 状态码,然后在输出 20 个 token 后连接中断。

我的探针会检查 [DONE] 标记。如果该标记始终未出现,则本次运行不完整,应视为失败,重试整个请求,不要使用部分输出。

以下是我使用的判定规则:

  • 收到 [DONE],无错误:成功。
  • 未收到 [DONE],无错误:不完整,重试一次。
  • HTTP 错误:失败,退避重试。
  • 流中途异常:失败,退避重试。

谁应该使用流式

流式输出适合聊天界面。用户期望逐 token 输出,即使模型速度较慢,也能带来响应即时的体验。

流式输出对 CI 和批处理任务则不太合适。等待同样数量的 token 需要更长时间,增加了失败模式,还需要额外的解析工作。普通的 JSON 响应更简单。

免费模型服务器让这种权衡更加复杂。首个 token 更慢,完成率更低。在做出决定之前,先测量数据。

局限性

这个探针不是负载测试。它一次只发送一个请求,不测量并发,也不测试吞吐量。

它还假设使用标准的 SSE 格式。有些端点使用不同的分隔符,请查阅你的服务商文档。

我只针对一个免费端点进行了测试。你的情况可能有所不同。一天中的时段、模型大小、提示词长度都会影响结果。

结论

流式并不自动意味着更快。在免费服务器上,它可能更慢。首个 token 具有欺骗性,要测量整个流的完成情况。

运行探针,保留输出。它会告诉你流式是否值得,该设置多长的超时时间,以及何时应回退到非流式。

如果你运行了探针,请分享你的数据。我想看看其他免费端点的表现如何。

阅读原文