从 3.2s 到 380ms：首 token 延迟拆解与流式渲染实战

问题背景：用户感知的“秒开”到底卡在哪

在 AI 编程助手、智能体对话框这类产品中，用户对响应速度的感知并不取决于整段回答生成完毕的总时长，而是第一个字出现在屏幕上的时间，即 TTFT（Time To First Token）。一个典型的坏体验是：用户点击发送，界面转圈 3 秒毫无反馈，即便后续内容生成得再快，主观感受也是“卡了”。

很多团队一上来就怀疑模型太慢，盲目切换更小的模型，结果成本和质量双双受损。实际上，从用户点击发送到首 token 上屏，链路中至少经过以下环节：

前端事件处理与请求序列化
网络建连与网关路由（含鉴权、限流、配额校验）
服务端提示词拼装（系统提示、历史对话、RAG 检索结果）
模型服务排队与 Prefill（处理输入 token 的阶段）
首个 token 经网络回到前端并完成首次渲染

只有把每一段耗时测量出来，才能判断瓶颈究竟在检索、网关还是模型本身。

延迟拆解：用一张图看清 TTFT 的构成

下图把首 token 之前的关键路径标出来，注意 Prefill 阶段的耗时与输入长度强相关，而 Decode 阶段产出首 token 后，后续 token 是以流式方式持续到达的。

方案对比：全量返回、SSE 与 WebSocket

方案	首字延迟	实现复杂度	适用场景	主要问题
普通 JSON 全量返回	高（等于整段生成时间）	最低	短文本、批处理	界面长时间无反馈
SSE（Server-Sent Events）	低，token 逐个推送	低，基于 HTTP	单向对话流式输出	需正确处理代理缓冲
WebSocket	低，且支持双向	较高	语音、多轮中断、协同编辑	鉴权与重连需自行实现
fetch + ReadableStream	低，前端控制力强	中	需要自定义协议、错误码	需手动解析分块

对绝大多数 LLM 对话场景，SSE 是性价比最高的选择：它天然单向、自动重连、走标准 HTTP，网关和监控体系无需特殊改造。WebSocket 更适合需要“用户中途打断生成”这类双向控制的场景。

实操：FastAPI 流式接口 + 前端增量渲染

下面给出一套可直接运行的最小实现。后端使用 FastAPI 的 StreamingResponse 推送 SSE，并在每个阶段打点，方便定位 TTFT；前端用原生 fetch 读取流，不依赖任何框架。

后端：带阶段耗时打点的 SSE 接口

python

# server.py  安装：pip install fastapi uvicorn httpx
import asyncio
import time
import json
import httpx
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

# 用阶段事件把耗时透传给前端，便于拆解 TTFT
async def chat_stream(question: str):
    t0 = time.perf_counter()

    # 阶段1：模拟鉴权 / 提示词拼装 / RAG 检索（此处应尽量并行化）
    await asyncio.sleep(0.05)
    t_ready = time.perf_counter()

    # 阶段2：调用上游模型的流式接口（示例用伪地址，替换为真实网关）
    url = "https://your-llm-gateway/v1/chat/completions"
    headers = {"Authorization": "Bearer <token>"}
    payload = {
        "model": "your-model",
        "stream": True,  # 关键：必须开启流式
        "messages": [{"role": "user", "content": question}],
    }

    first_token_at = None
    try:
        async with httpx.AsyncClient(timeout=60) as client:
            async with client.stream("POST", url, headers=headers, json=payload) as resp:
                async for line in resp.aiter_lines():
                    if not line or not line.startswith("data: "):
                        continue
                    data = line[6:]
                    if data.strip() == "[DONE]":
                        break
                    chunk = json.loads(data)
                    delta = chunk["choices"][0]["delta"].get("content", "")
                    if not delta:
                        continue
                    if first_token_at is None:
                        first_token_at = time.perf_counter()
                        # 先把 TTFT 拆解指标发给前端
                        meta = {
                            "prepare_ms": round((t_ready - t0) * 1000),
                            "ttft_ms": round((first_token_at - t0) * 1000),
                        }
                        yield f"event: meta\ndata: {json.dumps(meta)}\n\n"
                    yield f"data: {json.dumps({'text': delta}, ensure_ascii=False)}\n\n"
    except Exception as e:
        # 错误也要以 SSE 事件下发，不能让前端无限转圈
        yield f"event: error\ndata: {json.dumps({'message': str(e)})}\n\n"

@app.get("/chat")
async def chat(q: str):
    return StreamingResponse(
        chat_stream(q),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # 关键：关闭 Nginx 缓冲
        },
    )

启动：uvicorn server:app --port 8000。

前端：fetch 读取分块并逐字上屏

javascript

// 前端：读取 SSE 流，收到首个事件即停止 loading
async function ask(question) {
  const box = document.getElementById("answer");
  box.textContent = "";
  const t0 = performance.now();

  const resp = await fetch(`/chat?q=${encodeURIComponent(question)}`);
  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    // SSE 以空行分隔事件，按 \n\n 切分
    const parts = buffer.split("\n\n");
    buffer = parts.pop(); // 最后一段可能不完整，留到下次

    for (const part of parts) {
      const lines = part.split("\n");
      const event = (lines.find(l => l.startsWith("event: ") ) || "").slice(7) || "message";
      const dataLine = lines.find(l => l.startsWith("data: "));
      if (!dataLine) continue;
      const data = JSON.parse(dataLine.slice(6));

      if (event === "meta") {
        console.log("服务端 TTFT(ms):", data.ttft_ms,
                    "前端实测(ms):", Math.round(performance.now() - t0));
      } else if (event === "error") {
        box.textContent = "出错了：" + data.message;
      } else {
        box.textContent += data.text; // 增量追加，立即上屏
      }
    }
  }
}

对比服务端打点的 ttft_ms 与前端实测值，二者之差主要是网络回传与浏览器渲染开销；如果服务端 TTFT 已经很低而前端仍旧慢，问题就出在缓冲或渲染层。

进一步压低 TTFT 的四个工程手段

检索与拼装并行化：多路 RAG 召回使用 asyncio.gather 并发，系统提示词等静态内容在进程启动时预构建，避免每个请求重复拼接长模板。
控制 Prefill 输入长度：输入 token 越多，Prefill 越慢。对历史对话做摘要或滑动窗口，把无关上下文挡在模型之外，往往比换模型收益更直接。
网关关闭响应缓冲：Nginx 默认会缓冲响应，导致 token 在网关攒成一批才下发，必须设置 X-Accel-Buffering: no 或 proxy_buffering off。
连接复用与就近接入：使用 HTTP/1.1 keep-alive 或 HTTP/2 复用上游连接，并让网关与模型服务同地域部署，减少建连与跨区 RTT。

避坑总结

坑一：开了 stream 却仍然整段出现。 现象是后端日志显示 token 已陆续生成，但前端隔了一两秒才一次性刷出一大段。根因通常是 Nginx、CDN 或某些 Serverless 平台缓冲了响应。修正方法是响应头加 X-Accel-Buffering: no、确认 CDN 关闭流式响应缓冲，并避免在业务代码里用 "".join(tokens) 把结果攒完再返回。

坑二：SSE 分块解析时 JSON 解析报错。 直接对每个 read() 到的 chunk 做 JSON.parse，会因为一个 UTF-8 字符或一条事件被 TCP 拆包而抛 Unexpected end of JSON input。修正方法是用缓冲区按 \n\n 切分事件，把不完整的尾部留到下一轮，并用 TextDecoder 的 stream: true 正确处理多字节字符。

坑三：上游报错后前端无限转圈。 流式响应一旦返回 200，HTTP 状态码就无法再表达中途错误，若异常未被捕获并下发，前端会一直停在 loading。修正方法是在后端 try/except 中通过 event: error 主动推送错误事件，前端监听后终止 loading 并提示，同时上报监控。

坑四：只优化模型，忽视检索与历史长度。 盲目换小模型后回答质量下降，TTFT 却没明显改善，说明瓶颈本就在 RAG 的串行 IO 或过长的对话历史。正确做法是先靠分阶段打点定位，再决定是并行化检索、裁剪上下文，还是调整模型，把优化精力花在实测的最长环节上。

把首 token 延迟当作一条可测量、可拆解的链路来治理，配合流式渲染让用户“立刻看到字”，往往无需更换模型，就能把主观秒开体验从 3 秒压缩到几百毫秒，这正是 LLM 应用架构层面最具性价比的性能投入。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

从 3.2s 到 380ms：首 token 延迟拆解与流式渲染实战 ​

问题背景：用户感知的“秒开”到底卡在哪 ​

延迟拆解：用一张图看清 TTFT 的构成 ​

方案对比：全量返回、SSE 与 WebSocket ​

实操：FastAPI 流式接口 + 前端增量渲染 ​

后端：带阶段耗时打点的 SSE 接口 ​

前端：fetch 读取分块并逐字上屏 ​

进一步压低 TTFT 的四个工程手段 ​

避坑总结 ​