流式传输 LLM 响应：为何对用户体验和成本至关重要

让我们的聊天界面显得“坏了”的那 900 毫秒

几个月前，我在调试一个客服助手。从数据上看，它的端到端延迟完全可以接受：大多数响应在 4–7 秒内完成。模型会选择工具、检索账户上下文，并生成有用的回答。

但用户很讨厌它。

原因很简单：界面在最初的 3.8 秒里一直空白。

没有加载提示文案，没有部分响应，也没有任何迹象表明模型已经理解了问题。只有一个空白的助手消息气泡。当我们将同一个端点改为通过服务器发送事件流式传输词元时，平均总完成时间几乎没有变化。但对于许多请求，首次可见输出的等待时间降到了约 700–900 毫秒，产品立刻让人感觉更快了。

这才是流式传输重要的真正原因。

LLM 响应的流式传输并不是魔法。它不会让模型思考得更快，不会减少词元总量，也不会自动降低你的账单。但它改变了交互过程，用户能切实感受到这种变化，同时也让应用能够更好地控制取消、渲染、内容审核、用量核算和背压。

本文将介绍 SSE 流式传输的实际工作方式、需要测量的指标、计费中容易忽略的细节，以及我在生产系统中使用的客户端模式。

使用流式响应的主要好处是什么？

流式响应的主要好处在于感知延迟：用户只需几百毫秒就能看到最初的词元，而不必在整个生成过程中一直盯着加载动画。总生成时间几乎不变，变化的是首词元延迟，而这才是用户实际感受到的指标。一个需要 9 秒才能完成、但在 400 毫秒后就开始显示的回复，会让人觉得很快；同样的回复如果在 9 秒后一次性显示出来，就会让人觉得系统出了问题。

它还有其他好处：流式传输可以让你逐步渲染部分输出，在用户已经看到足够内容时提前取消生成（节省输出词元），并避免长响应触发代理读取超时。但如果只记住一点，那就是：流式传输不会让模型更快，而是让用户体验中的等待感消失。

流式传输究竟意味着什么

当你发起非流式 LLM 请求时，服务器通常会等到模型生成完整响应后，再返回一个 JSON 载荷：

{
  "id": "resp_123",
  "model": "some-model",
  "output_text": "Sure — here is the answer...",
  "usage": {
    "input_tokens": 812,
    "output_tokens": 436
  }
}

采用流式传输时，服务器会在输出可用时逐步发送事件。最常见的传输方式是 服务器发送事件，通常缩写为 SSE.

SSE 本质上就是一个使用长连接和特殊内容类型的 HTTP 响应：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

服务器发送的不是一个 JSON 对象，而是一系列小事件：

event: message.delta
data: {"delta":"Sure"}

event: message.delta
data: {"delta":" — here"}

event: message.delta
data: {"delta":" is"}

event: message.delta
data: {"delta":" the answer..."}

event: message.done
data: {"finish_reason":"stop","usage":{"input_tokens":812,"output_tokens":436}}

不同 API 使用不同的事件名称和模式，但整体流程通常相同：

1. 建立 HTTP 连接。
2. 发送请求元数据和提示词。
3. 接收增量数据块。
4. 将数据块追加到当前助手消息中。
5. 接收包含停止原因的最终事件，通常还包含用量元数据。
6. 关闭流。

实际中，大多数 LLM“流式传输”并不是每个事件只包含一个词元。提供商可能会为了效率将多个词元合并为数据块。你可能收到一个字符、一个单词片段、几个单词，或一个结构化的 JSON 增量。构建客户端时，绝不要假设每个数据块都是完整的单词或句子。

延迟：总耗时与首词元延迟

最重要的是区分以下两项 总延迟 和 首词元延迟.

对于聊天用户体验，TTFT 通常比总耗时更重要。一个在 800 毫秒后开始显示、并在 8 秒内完成的响应，往往比一个在 4 秒后一次性显示的响应体验更好。

对于机器之间的工作流，流式传输通常没那么重要。如果下一步必须等完整的 JSON 对象生成完毕后才能执行，流式传输可能只会增加实现复杂度。

在 Python 中测量 TTFT

下面是测量首个数据块延迟和流式传输总耗时的简单模式。这里特意采用通用的 API 形式：请根据你的提供商调整 URL 和载荷。

import json
import time
import requests

url = "https://api.example.com/v1/chat/completions"

payload = {
    "model": "sonnet-4.6",
    "stream": True,
    "messages": [
        {"role": "user", "content": "Explain SSE streaming in 5 bullet points."}
    ]
}

headers = {
    "Authorization": "Bearer $API_KEY",
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
}

start = time.perf_counter()
first_chunk_at = None
chars = 0

with requests.post(url, headers=headers, json=payload, stream=True) as r:
    r.raise_for_status()

    for raw_line in r.iter_lines(decode_unicode=True):
        if not raw_line:
            continue

        if not raw_line.startswith("data: "):
            continue

        data = raw_line[len("data: "):]

        if data == "[DONE]":
            break

        event = json.loads(data)

        # Provider-specific: this may be event["delta"], event["choices"][0]["delta"], etc.
        delta = event.get("delta", "")

        if delta and first_chunk_at is None:
            first_chunk_at = time.perf_counter()

        chars += len(delta)
        print(delta, end="", flush=True)

end = time.perf_counter()

print("\n\n--- metrics ---")
print(f"ttft_ms={((first_chunk_at or end) - start) * 1000:.0f}")
print(f"total_ms={(end - start) * 1000:.0f}")
print(f"chars={chars}")

一个常见陷阱是：如果你通过会缓冲响应的本地代理、无服务器函数或反向代理测试流式传输，可能会误以为模型没有进行流式输出。模型可能正在正常流式输出，但你的基础设施会暂存数据块，直到响应完成后才发送。

我在以下情况中遇到过这个问题：

• 启用了 Nginx 缓冲。
• 云函数不支持流式响应。
• 压缩中间件等待缓冲区积累更多数据。
• Web 框架先构建完整响应，再将其返回。
• 企业代理将小数据块合并。

对于 Nginx，下面这项设置往往很重要：

location /api/llm/stream {
    proxy_pass http://app_backend;
    proxy_http_version 1.1;

    proxy_buffering off;
    proxy_cache off;

    chunked_transfer_encoding on;
    add_header X-Accel-Buffering no;
}

SSE 与 WebSocket

SSE 并不是唯一的流式传输方式。WebSocket 也很常见，尤其适用于双向应用。不过，对于大多数 LLM 聊天界面，SSE 是更简单的默认选择。

该 EventSource 浏览器 API 很方便，但只支持 GET 请求。大多数 LLM 调用需要 POST 请求体和认证标头，因此我通常使用以下方式进行流式传输 fetch() 和 ReadableStream.

浏览器客户端模式：Fetch + ReadableStream

一个最简浏览器实现如下：

async function streamChat({ messages, signal, onDelta, onDone }) {
  const res = await fetch("/api/chat/stream", {
    method: "POST",
    headers: {
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: "gpt-5.5",
      stream: true,
      messages
    }),
    signal
  });

  if (!res.ok) {
    throw new Error(`HTTP ${res.status}`);
  }

  const reader = res.body.getReader();
  const decoder = new TextDecoder("utf-8");

  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();

    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    const lines = buffer.split("\n");
    buffer = lines.pop() || "";

    for (const line of lines) {
      const trimmed = line.trim();

      if (!trimmed.startsWith("data:")) continue;

      const data = trimmed.slice(5).trim();

      if (data === "[DONE]") {
        onDone?.();
        return;
      }

      const event = JSON.parse(data);

      if (event.delta) {
        onDelta(event.delta);
      }
    }
  }

  onDone?.();
}

在 UI 中：

const controller = new AbortController();

let assistantText = "";

streamChat({
  messages,
  signal: controller.signal,
  onDelta(delta) {
    assistantText += delta;
    renderAssistantMessage(assistantText);
  },
  onDone() {
    markMessageComplete();
  }
});

// Cancel button
document.querySelector("#stop").onclick = () => {
  controller.abort();
};

不要直接渲染每个数据块

如果每收到一个很小的数据块就更新 React 状态，可能会无意中造成客户端性能问题。有些流每秒会产生数十个数据块。每次都重新渲染庞大的聊天组件树，会浪费资源。

在实际开发中，我通常按动画帧批量更新 UI：

let pending = "";
let scheduled = false;

function onDelta(delta) {
  pending += delta;

  if (!scheduled) {
    scheduled = true;

    requestAnimationFrame(() => {
      assistantText += pending;
      pending = "";
      scheduled = false;
      setAssistantText(assistantText);
    });
  }
}

这样既能保持界面流畅，也能让用户感受到内容正在实时输出。

正确处理部分输出

流式传输意味着应用会收到不完整的文本。这听起来显而易见，但许多错误都源于将部分输出当作最终输出处理。

Markdown 暂时无效

在流式传输过程中，Markdown 的格式可能不完整：

Here is the query:

```sql
SELECT *
FROM users
WHERE

代码围栏可能还没有闭合。表格可能只有表头行。链接可能只输出了一半。渲染器需要能够容忍这些情况。

对于聊天 UI，我倾向于：

• 增量渲染 Markdown，同时做好容错处理。
• 禁用有风险的 HTML。
• 在消息完成之前，避免执行开销较大的语法高亮。
• 显示不显眼的“正在生成”状态。
• 流结束后，重新执行最终格式化。

JSON 流式传输更难

如果让模型生成 JSON 并以流式方式输出，中间状态通常是无效的 JSON。对于希望实时更新结构化 UI 的应用来说，这是个问题。

例如：

{
  "title": "Incident Summary",
  "severity": "high",
  "steps": [
    "Restart

此时还无法解析。

常见的处理方式有三种：

1. 等待输出完成，然后一次性解析。
2. 使用按行分隔的 JSON，其中每个完整行都可以独立解析。
3. 流式传输语义事件，而不是原始 JSON。

对于结构化流式传输，我喜欢采用事件形式的输出：

event: section
data: {"title":"Summary"}

event: bullet
data: {"text":"Database failover started at 09:42 UTC."}

event: bullet
data: {"text":"Read latency remained elevated for 6 minutes."}

这种模式需要更多编排，但更容易实现可靠渲染。

用户点击“停止”时，究竟发生了什么

取消生成是流式传输一个常被低估的优点。

对于非流式请求，用户可能已经离开页面，而模型仍在服务器上继续生成。根据提供商和后端的实现，这次生成可能仍会消耗输出 token。

采用流式传输时，客户端可以中止 HTTP 请求：

const controller = new AbortController();

fetch("/api/chat/stream", {
  method: "POST",
  body: JSON.stringify(payload),
  signal: controller.signal
});

// Later
controller.abort();

但取消操作并不总能在端到端链路上立即生效。

其中涉及多个层级：

• 浏览器中止 fetch。
• 后端检测到连接关闭。
• 后端取消向上游提供商发出的请求。
• 提供商停止生成。
• 根据已处理或已生成的 token 计算用量。

一个常见陷阱是：如果后端代理了流，却没有将取消操作传递到上游，用户的浏览器虽然断开了连接，服务器仍会继续读取模型输出，直到生成完成。

用 Node 风格的伪代码表示：

app.post("/api/chat/stream", async (req, res) => {
  const upstreamController = new AbortController();

  req.on("close", () => {
    upstreamController.abort();
  });

  const upstream = await fetch(PROVIDER_URL, {
    method: "POST",
    headers: providerHeaders,
    body: JSON.stringify({
      ...req.body,
      stream: true
    }),
    signal: upstreamController.signal
  });

  res.setHeader("Content-Type", "text/event-stream");

  for await (const chunk of upstream.body) {
    res.write(chunk);
  }

  res.end();
});

这就是“UI 停止了”和“模型停止了”的区别。

流式传输与成本：哪些会变，哪些不会

流式传输会改变你获知用量的时间，但通常不会改变基本计费单位。

大多数 LLM API 的收费依据包括：

• 输入 token。
• 输出 token。
• 缓存的输入 token（如果适用提示词缓存）。
• 工具调用或特殊模态，具体取决于 API。
• 对于某些模型系列，有时还包括推理 token 或内部按预算分配的计算资源。

流式传输本身通常不会带来折扣。如果模型生成了 700 个输出 token，无论它们是逐步到达，还是通过最终的一次性载荷到达，都应按 700 个输出 token 核算。

流式传输对成本的影响主要体现在实际运行中：

用量核算中的陷阱

许多流式 API 会在流结束时发送用量元数据：

{
  "type": "message.done",
  "usage": {
    "input_tokens": 1240,
    "output_tokens": 389
  },
  "finish_reason": "stop"
}

如果流在最终事件到达之前中断，应用可能收不到用量信息。这并不一定意味着用量为零。

对于内部核算，我通常存储三个字段：

{
  "request_id": "chat_01j...",
  "status": "client_aborted",
  "usage": null,
  "estimated_output_chars": 1842,
  "final_usage_received": false
}

如果提供商提供请求日志或用量事件，之后再进行核对。如果不提供，至少可以为产品分析做粗略估算，同时避免在计费时呈现虚假的精确度。

除非你的条款明确允许，否则不要根据字符数估算向最终用户计费。分词方式因模型和语言而异。

后端架构模式

生产环境中的流式传输链路通常如下：

Browser
  |
  | POST /api/chat/stream
  v
App Backend
  |
  | provider streaming request
  v
LLM API

我很少将提供商密钥直接暴露给浏览器。后端可以提供：

• 身份认证和授权。
• 按用户或工作区进行速率限制。
• 提示词组装和检索。
• 工具执行。
• 敏感信息脱敏和日志记录控制。
• 提供商故障切换。
• 取消操作传递。
• 用量记录。

Python FastAPI 后端的最简流式响应：

import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI()

PROVIDER_URL = "https://api.example.com/v1/chat/completions"

@app.post("/api/chat/stream")
async def chat_stream(request: Request):
    body = await request.json()

    async def event_generator():
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "POST",
                PROVIDER_URL,
                headers={
                    "Authorization": "Bearer YOUR_PROVIDER_KEY",
                    "Content-Type": "application/json",
                    "Accept": "text/event-stream",
                },
                json={
                    **body,
                    "stream": True,
                },
            ) as upstream:
                async for line in upstream.aiter_lines():
                    if await request.is_disconnected():
                        break

                    if not line:
                        continue

                    # Optionally transform provider-specific events here.
                    yield line + "\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

在实际开发中，我通常将提供商事件统一转换为一种内部模式：

{
  "type": "delta",
  "text": "partial text here"
}

然后：

{
  "type": "done",
  "finish_reason": "stop",
  "usage": {
    "input_tokens": 812,
    "output_tokens": 436
  }
}

如果你通过同一个界面支持 Claude、GPT、Gemini、Llama、Qwen 或 Kimi 等多个模型，这一点就很重要。各提供商的流语义略有不同。尽早统一格式，可以让前端保持简单，而这正是你想要的。

TTFT 不只取决于模型

当 TTFT 表现不佳时，人们往往首先归咎于模型。有时确实如此：更大的模型、更长的上下文以及更复杂的推理模式，都可能增加首个 token 的延迟。但在首个 token 到达之前，还需要经过许多处理环节。

影响因素包括：

• TLS 连接建立。
• 后端冷启动。
• 提示词构建和检索。
• 向量数据库延迟。
• 生成前的工具调用。
• 提供商侧排队。
• 长输入上下文的读取与处理。
• 安全检查或策略层。
• 代理缓冲。
• 客户端渲染延迟。

实用的链路追踪应将以下环节区分开来：

{
  "trace_id": "req_abc",
  "timings_ms": {
    "auth": 12,
    "retrieval": 84,
    "prompt_build": 9,
    "provider_connect": 141,
    "ttfb": 612,
    "first_delta": 804,
    "stream_complete": 5820
  }
}

如果检索耗时 600 毫秒，而模型的首个增量输出在 250 毫秒后才到达，那么切换模型可能无法改善用户体验。如果主要耗时来自与提供方建立连接，复用连接可能会有所帮助。如果前端要等到完整段落生成后才渲染，那么流式传输本身正常工作，只是你的 UI 策略隐藏了它。

何时不应使用流式传输

流式传输并不总是值得采用。

在以下情况下，我会避免使用流式传输：

• 响应非常短，例如分类或路由结果。
• 调用方是另一个需要完整 JSON 的服务。
• 输出必须通过完整的内容审核后才能显示。
• 请求在任务队列或批处理流水线中运行。
• 框架或托管层导致流式传输不可靠。
• 产品需要一次性呈现完整答案，而非逐步显示文本。

流式传输也有用户体验方面的缺点：它可能让用户看到模型在回答过程中改变思路。人们在聊天中可以容忍一定程度的这种情况，但在高风险工作流中，这会显得不够完善。对于法律、医疗、金融或合规要求严格的界面，你可以选择仅在通过防护检查后进行流式输出，或先将流式输出写入内部缓冲区，再显示安全的文本片段。

实用的客户端模式

以下是我为大多数聊天产品推荐的模式：

1. 在文本到达前显示状态

不要让助手的消息气泡空着。使用明确的状态：

• “正在读取上下文…”
• “正在搜索文档…”
• “正在起草回答…”
• “仍在处理中…”

当工具调用发生在文本生成之前时，这尤其有用。

2. 提供真正有效的停止按钮

停止按钮不只是用户体验上的装饰。如果实现正确，它可以节省输出 token。

确保：

• 浏览器中止请求。
• 后端检测到连接断开。
• 后端中止向提供方发出的请求。
• 将该请求记录为已取消。

3. 增量渲染，完成后再格式化

流式传输期间：

• 追加文本。
• 以轻量方式渲染 Markdown。
• 避免开销较大的语法高亮。
• 保持滚动行为可预期。

完成后：

• 重新解析 Markdown。
• 为代码添加高亮。
• 启用复制按钮。
• 如果系统支持，附上引用或参考资料。

4. 谨慎存储未完成的消息

如果流式传输中途失败，需要确定这在对话记录中如何体现。

可选方案：

• 保留助手的部分消息，并标记为“已中断”。
• 将其完全删除。
• 让用户使用同一提示词重试。
• 让模型根据已有的部分上下文继续生成。

我通常会保留部分消息，并加上醒目的标记。悄悄删除会让用户困惑，而假装消息已完成则更糟。

5. 统一事件格式

前端不应需要了解五种提供方各自特有的流式格式。应在后端统一格式：

{"type":"delta","text":"Hello"}
{"type":"tool_start","name":"search_docs"}
{"type":"tool_result","name":"search_docs","status":"ok"}
{"type":"done","finish_reason":"stop","usage":{"input_tokens":100,"output_tokens":50}}
{"type":"error","message":"upstream timeout"}

这也能让模型系列之间的切换或请求的动态路由更容易。如果你使用多模型 API 层，这种格式统一可能已经是网关的一部分；否则，值得自行构建。

实践要点

• 流式传输通过缩短首个可见输出的等待时间来改善感知延迟，而不是让整个生成过程变得免费或瞬间完成。
• 分别测量 TTFT、整体完成时间和每秒 token 数。
• SSE 通常是实现 LLM 文本流式传输最简单的传输方式；请使用 fetch() 流来处理需要 POST 正文的情况。
• 留意代理和框架的缓冲行为。许多“流式传输问题”实际上是基础设施缓冲问题。
• 将部分输出视为无效内容：Markdown 可能不完整，JSON 可能无法解析，代码围栏可能尚未闭合。
• 只有当中止操作从浏览器传递到后端，再传递到提供方时，取消请求才能节省成本。
• 用量元数据通常在流结束时才到达；对于中断的流，需要谨慎核算。
• 不要每收到一个微小片段就重新渲染整个 UI。使用动画帧批量更新。
• 在后端统一各提供方特有的流事件格式。
• 对于简短的分类任务、批处理任务、严格的结构化输出，或要求在显示前完成全面内容审核的工作流，不要使用流式传输。