Source: https://aiprimetech.io/zh/blog/streaming-api-guide/ 2026 年 22 日，六月 · 6 分钟 · API # 流式传输 LLM 响应：为何对用户体验和成本至关重要 A 作者 [AI Prime Tech](https://aiprimetech.io/zh/) · AI 辅助撰写 ## 让我们的聊天界面显得“坏了”的那 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 使用不同的事件名称和模式，但整体流程通常相同： - 建立 HTTP 连接。 - 发送请求元数据和提示词。 - 接收增量数据块。 - 将数据块追加到当前助手消息中。 - 接收包含停止原因的最终事件，通常还包含用量元数据。 - 关闭流。 实际中，大多数 LLM“流式传输”并不是每个事件只包含一个词元。提供商可能会为了效率将多个词元合并为数据块。你可能收到一个字符、一个单词片段、几个单词，或一个结构化的 JSON 增量。构建客户端时，绝不要假设每个数据块都是完整的单词或句子。 ## 延迟：总耗时与首词元延迟 最重要的是区分以下两项 总延迟 和 首词元延迟. | 指标 | 衡量的内容 | 重要性 | 首字节延迟，TTFB | 从服务器收到首个字节的时间 | 包含路由、身份验证、排队和模型调度的耗时 | 首词元延迟，TTFT | 首个生成的词元或数据块对用户可见的时间 | 决定用户感知响应速度的主要因素 | 每秒词元数 | 首个词元之后的输出生成速度 | 决定长回答完成的速度 | 总完成时间 | 收到完整响应的时间 | 对自动化、工具调用和批处理任务很重要 | 获得可用答案的时间 | 用户获得足够信息、可以采取行动的时间 | 通常早于完整响应完成的时间 对于聊天用户体验，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 是更简单的默认选择。 | 功能 | SSE | WebSocket | 方向 | 服务器到客户端 | 双向 | 协议 | 普通 HTTP 响应 | WebSocket 协议升级 | 浏览器支持 | 原生 EventSource；也支持使用 fetch 流 | 原生 WebSocket | 重连行为 | 内置重连机制，但仍需应用逻辑配合 | 需自行实现 | 负载均衡器兼容性 | 通常更容易 | 有时需要额外配置 | 最适合 | 词元流式传输、进度事件、简单聊天 | 实时协作、音频、多轮实时会话 | 请求体支持 | EventSource 仅支持 GET；使用 fetch 进行 POST 流式传输 | 全双向消息传输 该 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 ``` 此时还无法解析。 常见的处理方式有三种： - 等待输出完成，然后一次性解析。 - 使用按行分隔的 JSON，其中每个完整行都可以独立解析。 - 流式传输语义事件，而不是原始 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 核算。 流式传输对成本的影响主要体现在实际运行中： | 场景 | 成本影响 | 用户在生成 80 个 token 后取消 | 如果取消操作成功传递，就可能避免生成剩余输出 | UI 提前检测到答案已足够 | 应用可以主动停止生成 | 后端未能中止上游请求 | 即使客户端已断开连接，你也可能需要支付完整生成的费用 | 用量仅在最终事件中返回 | 对于已中止的流，可能需要依赖提供商侧的日志或估算 | 默认以流式方式输出长回答 | 由于体验更流畅，用户可能会消耗更多 | 在流式输出文本之前调用工具 | TTFT 可能增加，但输出 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。使用动画帧批量更新。 - 在后端统一各提供方特有的流事件格式。 - 对于简短的分类任务、批处理任务、严格的结构化输出，或要求在显示前完成全面内容审核的工作流，不要使用流式传输。 [推理](/zh/blog/tag/inference/)[API](/zh/blog/tag/api/) A [AI Prime Tech](https://aiprimetech.io/zh/) 本博客由 AI Prime Tech 发布。文章借助 AI 辅助起草；在据此采取行动前，请对照官方文档核实模型 ID、价格和限额等特定版本的细节。 以更低价格接入 Claude API 一个 API 密钥即可使用 Claude Opus 5.5、Sonnet 5、Haiku 4.5 和 Fable 5.1，以及 GPT-6 个模型。按量付费，无需订阅。 [获取 API 密钥 →](/register) AI Prime Tech 是独立的第三方 API 网关。Claude™ 和 Anthropic® 是 Anthropic, PBC 的商标。此处不暗示任何关联或背书。