LLM 流式输出实战：SSE 推流与前端渲染

大模型是逐 token 生成的，LLM 流式输出让用户在整段回答写完之前就开始看到文字，体验上从“转圈等待”变成“边想边说”。本文用 SSE（Server-Sent Events）把后端推流到前端渲染的完整链路讲透，并给出可运行的 Python 与 JavaScript 代码。

一、为什么需要流式输出

调用大模型时，模型从左往右一个个生成 token，整段回答可能要几秒甚至几十秒。如果等全部生成完再一次性返回，用户面对的是一段漫长的空白，流失率很高。流式输出的本质是“边生成边下发”：后端每拿到一小块 token，就立刻通过一条长连接推给前端，前端逐字渲染，用户几乎实时看到思考过程。

它带来的不只是体感提升。流式能让前端提前展示“已生成”的内容、更早触发停止、更早暴露异常；对长文生成、代码续写、Agent 多步推理这类场景，流式几乎是标配。理解了这一点，再看协议与代码就顺理成章。

从工程角度看，流式还改变了错误处理的节奏。传统同步接口是“要么全成功、要么全失败”，而流式下后端可能已经吐出前半段、却在中途因模型超时或上游抖动而中断。前端需要能优雅处理这种“半截响应”：丢弃不完整的 JSON、向用户提示“生成中断，可重试”，而不是把残缺内容当作最终结果。这是流式上线前必须想清楚的第一件事。

二、SSE 协议基础

SSE 是 HTTP 上一种单向、文本、长连接的推送协议，天然适合“服务器主动推送、客户端只接收”的流式场景。相比 WebSocket 的双向全双工，SSE 更简单：它就是一个 Content-Type: text/event-stream 的 HTTP 响应，服务端按 字段: 值 的格式不断写入事件，事件之间用空行分隔。

和轮询（polling）相比，SSE 不需要客户端反复发起请求，服务端有数据才推送，省去大量空请求；和 WebSocket 相比，SSE 是单向的、基于普通 HTTP，调试更直观，也更容易被现有网关与 CDN 兼容。正因如此，几乎所有大模型厂商的流式接口都选择 SSE 作为传输格式，理解它几乎等于理解了行业的事实标准。

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"choices":[{"delta":{"content":"大"}}]}

data: {"choices":[{"delta":{"content":"模型"}}]}

data: [DONE]

大模型厂商的流式接口通常就返回这种格式：每个 data: 行是一段增量 JSON，末尾以 data: [DONE] 收尾。我们只要在服务端“透传”，就能把厂商的流原样送给浏览器。

三、后端：用 httpx 转发大模型流式响应

实战中，我们通常不会让前端直连模型厂商（密钥会暴露），而是用后端做一层代理：后端以 stream: true 调用模型，再把厂商返回的 SSE 字节流原样转发给前端。下面用 FastAPI + httpx 演示一个最小可用的流式代理。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, os

app = FastAPI()
AGNES_CHAT = "https://api.agnes-ai.cn/v1/chat/completions"
API_KEY = os.getenv("AGNES_KEY")

async def proxy_stream(prompt: str):
    payload = {
        "model": "agnes-2.5-flash",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    headers = {"Authorization": f"Bearer {API_KEY}",
               "Content-Type": "application/json"}
    async with httpx.AsyncClient(timeout=60) as client:
        async with client.stream("POST", AGNES_CHAT,
                                 json=payload, headers=headers) as resp:
            async for line in resp.aiter_lines():
                if not line or not line.startswith("data:"):
                    continue
                data = line[len("data:"):].strip()
                if data == "[DONE]":
                    yield "data: [DONE]\n\n"
                    break
                yield f"data: {data}\n\n"

@app.post("/chat/stream")
async def chat_stream(req: dict):
    prompt = req.get("prompt", "")
    return StreamingResponse(proxy_stream(prompt),
                             media_type="text/event-stream")

关键点在于 client.stream(...) 拿到的是异步字节流，aiter_lines() 按行迭代，我们只做“裁剪前缀 + 透传”两件事。注意这里没有把整段响应读进内存，因此即使回答很长也不会撑爆服务端。

还有一个易踩的坑是“缓冲不刷新”。某些网关层会等缓冲区写满才真正发给客户端，导致流式变成“攒一批发一批”。在 FastAPI 这类 ASGI 框架下，只要正确返回 StreamingResponse 并关闭代理缓冲，字节就会随产随发。生产环境务必在网关层关掉对这条路径的缓冲，否则用户看到的就是“几十秒后一次性蹦出全文”，流式形同虚设。

四、前端：浏览器原生接收 SSE

前端不只可以用 EventSource，更灵活的做法是用 fetch + ReadableStream 手动解析，便于携带自定义请求体（POST）和鉴权头。下面这段 JavaScript 把后端推来的增量 JSON 解析成文本并逐字渲染。

async function streamChat(prompt) {
  const resp = await fetch("/chat/stream", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ prompt })
  });
  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 });
    const events = buffer.split("\n\n");
    buffer = events.pop();
    for (const evt of events) {
      const line = evt.split("\n").find(l => l.startsWith("data:"));
      if (!line || line.includes("[DONE]")) continue;
      const json = JSON.parse(line.slice(5).trim());
      const delta = json.choices?.[0]?.delta?.content || "";
      render(delta);
    }
  }
}

这里用 \n\n 切分 SSE 事件，再取 data: 行解析 JSON 取 delta.content。把每个 delta 追加进同一个 DOM 节点即可实现“打字机”效果。若用原生 EventSource，则只支持 GET，需要把参数放到 URL query 上。

如果后端支持 GET 且不需要自定义请求体，用浏览器原生的 EventSource 会更省心：它内置自动重连、自带 onmessage 回调，代码量更少。但一旦要 POST 鉴权信息或复杂参数，就必须回到 fetch + 手动解析的方案。无论哪种，前端渲染都建议做节流——把多次 delta 累积到一定量再一次性更新 DOM，避免每个字符都触发重排导致卡顿。

五、鉴权、安全与边界

流式接口同样是业务接口，鉴权不能省：后端代理层应校验用户会话，再决定是否去调模型，避免被刷。与要求“返回固定 JSON 结构”的 大模型结构化输出实战 不同，流式输出放弃的是“一次性完整结果”，换来的是实时性，因此更适合对话、续写这类开放生成；而需要严格字段回填的场景（如表单抽取）仍建议走结构化输出。

另外要控制单连接生命周期与并发：每个流式连接都会占用一个后端 worker，应在网关层设置合理的超时与最大并发，防止长连接把连接池耗尽。配合 AI Agent 工作流：Function Calling 这类多步推理，流式还能把每一步的中间思考实时回传给用户，体验更透明。

还有一个现实问题：流式接口更容易被恶意刷量，因为单个请求会持续占用 worker 数十秒。除了会话校验，还应配合限流（例如按用户维度限制并发流式连接数）与计费埋点，把“实时体验”和“成本控制”同时管起来。这一点在把能力开放给外部调用方时尤为关键。

六、常见问题与优化

流式最常被坑的地方在“中间层缓冲”和“连接管理”。下面这张表覆盖了高频问题和处理办法。

七、与其他能力的协同

流式输出不是孤立能力，它和站内多套 AI 工程化实践能自然组合：

• 和 RAG 进阶：混合检索与重排 结合，可先把检索过程以流式“思考链”展示，再给出最终答案；
• 和 vLLM 大模型部署实战 结合，自托管推理服务同样支持 stream: true，把流推送能力下沉到私有集群；
• 和 Dify + Ollama 私有知识库 结合，知识库问答的前端打字机效果正是流式输出的最佳落地场景。

把这几块串起来，你就拥有了一条从“用户提问 → 检索 → 推理 → 逐字呈现”的完整链路。

更进一步的玩法是“在流里夹带结构化信号”：比如先用几个 token 输出一个 [思考] 标记，再输出正文，前端据此区分“推理过程”与“最终答案”，实现可折叠的思考链。这正是把流式输出与可解释性结合的典型实践，也是很多推理类模型在对话产品里的标准呈现方式。

八、小结

LLM 流式输出的本质是用 SSE 把“边生成边下发”这件事标准化：后端用异步流式客户端透传模型字节流，前端用 fetch + ReadableStream 解析增量并渲染。它和结构化输出、Function Calling、RAG 等能力互补而非互斥。掌握这条链路后，你的对话产品就能从“转圈等待”升级为“边想边说”的实时体验。