AI 对话界面里，文字像打字机一样一个个蹦出来，这个效果几乎成了所有大模型产品的标配。但很多人第一次接触时都会好奇：后端明明是一次性把整段回答算完的，前端凭什么能一个字一个字地显示？答案就藏在SSE（Server-Sent Events）这套机制里。它不是什么新东西，浏览器原生支持了十几年，却在 AI 时代重新翻红，成了流式输出的首选方案。这篇内容我会从 SSE 的底层原理讲起，一路带到 Vue 前端怎么接、FastAPI 后端怎么发、AbortController 怎么中断、idle timeout 怎么排查，把这条链路彻底打通。不管你是刚入门的前端，还是想搞懂流式接口的后端，看完都能自己动手复现一个"打字机"效果。

1. 为什么 AI 回答非要"一个字一个字"蹦

1.1 从用户体验倒推技术选型

先想一个问题：如果后端把整段回答算完再一次性返回，用户会看到什么？答案是——一个长时间的空白等待，然后"啪"地一下整段文字全冒出来。大模型生成一段几百字的回答，耗时可能三五秒甚至十几秒，这段时间里页面毫无反馈，用户会以为卡死了，直接刷新走人。

流式输出解决的正是这个"等待焦虑"。它让首字节到达时间（TTFT，Time To First Token）从"整段生成完"提前到"第一个 token 生成完"，通常几百毫秒就能看到第一个字。用户感知到的响应速度一下子快了一个数量级，哪怕总耗时没变，体验也是天壤之别。

这里有个容易被忽略的点：流式输出不只是"好看"。对于长回答，用户可以在文字还在生成时就开始阅读，边读边等，实际感知等待时间被大幅压缩。这是产品层面的刚需，不是锦上添花。

1.2 为什么是 SSE，而不是 WebSocket 或轮询

实现"服务端持续推数据"的方案有好几种，为什么 AI 场景几乎清一色选 SSE？我们横向对比一下。

关键差异在于：AI 对话是典型的"一问一答"单向流——客户端发一次请求，服务端持续推 token，客户端不需要在这条连接上再发消息。WebSocket 的全双工能力在这里是浪费，反而带来了握手升级、心跳保活、连接管理的额外复杂度。而 SSE 就是一条普通的 HTTP 长连接，服务端Content-Type: text/event-stream，然后不断往响应体里写数据，浏览器自动帮你解析。

还有一个现实因素：SSE 走的是标准 HTTP，天然兼容现有的鉴权、网关、负载均衡体系。WebSocket 的Upgrade握手在很多企业网关、代理层会被拦或者需要额外配置，运维成本高。这也是为什么 OpenAI、Anthropic 这些主流大模型 API 的流式接口，默认都用 SSE。

注意：SSE 是单向的，服务端到客户端。如果你的场景需要客户端在流式过程中频繁发指令（比如实时协作编辑），那 WebSocket 更合适。AI 对话不属于这类。

1.3 SSE 的数据格式长什么样

SSE 的协议格式简单到令人发指，它就是纯文本，靠特定前缀和空行来分帧。一个完整的事件帧长这样：

data: 你 data: 好 data: 呀

每一行以data:开头，后面跟内容，然后一个空行表示这一帧结束。浏览器收到后触发onmessage事件，event.data就是data:后面的内容。除了data，还有几个字段：

• event:自定义事件类型，前端可以用addEventListener监听特定类型
• id:事件 ID，断线重连时浏览器会通过Last-Event-ID请求头带上，方便服务端续传
• retry:指定重连间隔毫秒数
• 以:开头的行是注释，常用来做心跳保活

这里有个坑：SSE 规范里，一个data:字段后面必须跟一个空行才会触发事件。如果你服务端写完data: xxx忘了补空行，前端会一直等，什么都收不到。这是新手最常踩的坑之一。

2. 后端怎么把 token "挤"出来：FastAPI 实战

2.1 用 StreamingResponse 搭建流式接口

FastAPI 做流式输出非常顺手，核心就是StreamingResponse配合一个生成器（generator）。生成器yield一次，响应体就往外写一段。先看一个最小可运行的骨架：

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() async def fake_llm_stream(prompt: str): reply = f"你问的是：{prompt}。我来慢慢回答你。" for char in reply: yield f"data: {char}\n\n" await asyncio.sleep(0.05) yield "data: [DONE]\n\n" @app.get("/chat/stream") async def chat_stream(prompt: str): return StreamingResponse( fake_llm_stream(prompt), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, )

几个关键点必须说清楚。media_type必须是text/event-stream，这是 SSE 的标识，浏览器看到它才会按事件流解析。Cache-Control: no-cache防止中间层缓存流式响应。X-Accel-Buffering: no是给 Nginx 看的，告诉它别缓冲——这个头不加，Nginx 默认会把响应攒够一定大小才转发，你的"逐字"效果就变成了"攒一批蹦一批"。

生成器里每个yield的字符串，注意结尾是\n\n，两个换行。第一个\n结束data:这一行，第二个\n是帧结束的空行。少一个都不行。

2.2 接真实大模型：以 Ollama 为例

上面的fake_llm_stream只是演示。真实场景里，你要把大模型的流式输出转发出去。以本地 Ollama 为例，它本身支持流式返回，我们只需要把它的流"搬运"到 SSE 里：

import httpx import json async def ollama_stream(prompt: str): async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", "http://localhost:11434/api/generate", json={"model": "qwen2", "prompt": prompt, "stream": True}, ) as resp: async for line in resp.aiter_lines(): if not line: continue chunk = json.loads(line) token = chunk.get("response", "") if token: yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n" if chunk.get("done"): yield "data: [DONE]\n\n" break

这里有个细节值得展开：为什么要把 token 包成 JSON 再发，而不是直接发纯文本？因为纯文本里如果包含换行符，会破坏 SSE 的帧结构。SSE 规定data:后面的内容不能有裸换行，否则会被解析成多行。大模型输出的 token 里换行很常见，直接拼进去必然出问题。包一层 JSON，换行符被转义成\n，就安全了。前端拿到后再JSON.parse还原。这是生产环境必须做的处理。

另外httpx.AsyncClient(timeout=None)也很关键。流式请求可能持续几十秒，默认超时会把连接掐断。设成None表示不超时，但你要在业务层自己做超时控制，否则一个卡死的请求会一直挂着。

2.3 心跳保活与 idle timeout 的坑

热词里有个报错很典型：stream disconnected before completion: idle timeout waiting for sse。这个错误的本质是：连接建立后，一段时间内没有任何数据流动，中间层（网关、代理、负载均衡）判定连接空闲，主动断开。

大模型有个特点——首 token 延迟可能很长。用户发完问题，模型要先做 prefill（处理输入），可能好几秒才吐出第一个字。这几秒里连接上没有任何数据，如果网关的空闲超时设得短（比如 5 秒），连接就被掐了。

解决办法是发心跳。在等待首 token 期间，定期往流里写注释行：

async def stream_with_heartbeat(prompt: str): queue = asyncio.Queue() async def produce(): async for chunk in ollama_stream(prompt): await queue.put(chunk) await queue.put(None) task = asyncio.create_task(produce()) while True: try: item = await asyncio.wait_for(queue.get(), timeout=10) except asyncio.TimeoutError: yield ": heartbeat\n\n" continue if item is None: break yield item await task

: heartbeat\n\n是 SSE 注释行，前端不会触发onmessage，但连接上有数据流动，网关就不会判定空闲。10 秒发一次，比大多数网关的默认空闲超时（通常 30 到 60 秒）短，留足余量。

提示：心跳间隔不要设太短，否则无谓地占用带宽；也不要太长，要小于链路上最短的那个空闲超时。我一般取 10 到 15 秒，实测比较稳。

3. Vue 前端怎么接住这条流

3.1 原生 EventSource 的甜与苦

浏览器原生提供了EventSource，用起来极其简单：

const es = new EventSource('/chat/stream?prompt=你好') es.onmessage = (e) => { if (e.data === '[DONE]') { es.close() return } appendText(JSON.parse(e.data).token) } es.onerror = (err) => { console.error('SSE 出错', err) }

但EventSource有两个硬伤，在 AI 场景里很致命。

第一，它只支持 GET 请求，不能自定义请求头。这意味着你没法在请求头里带Authorizationtoken 做鉴权，也没法发 POST 请求体。AI 对话的 prompt 往往很长，塞进 URL query 既不优雅也有长度限制。很多团队被迫把 token 放 query 参数里，安全性打折扣。

第二，它不能中断。EventSource只有close()，但close()之后这条连接就废了，没法复用。更麻烦的是，用户点"停止生成"时，你希望的是中断这次请求，而不是简单关掉连接——后端可能还在跑，白白消耗算力。

所以生产环境里，AI 流式对话几乎都用fetch+ReadableStream手动解析，而不是EventSource。

3.2 用 fetch 手动解析 SSE 流

fetch返回的response.body是一个ReadableStream，我们可以拿到 reader 逐块读取，然后按 SSE 格式解析。这是目前最灵活、最可控的方案：

async function streamChat(prompt, onToken, signal) { const resp = await fetch('/chat/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt }), signal, }) const reader = resp.body.getReader() const decoder = new TextDecoder('utf-8') let buffer = '' while (true) { const { done, value } = await reader.read() if (done) break buffer += decoder.decode(value, { stream: true }) const parts = buffer.split('\n\n') buffer = parts.pop() for (const part of parts) { const line = part.trim() if (!line || line.startsWith(':')) continue if (!line.startsWith('data:')) continue const data = line.slice(5).trim() if (data === '[DONE]') return try { onToken(JSON.parse(data).token) } catch { onToken(data) } } } }

这段代码有几个地方值得细抠。decoder.decode(value, { stream: true })里的stream: true很重要，它告诉解码器当前块可能不是完整的 UTF-8 字符，先别急着报错，等下一块拼上再解。中文是多字节字符，一个汉字可能被切在两个 chunk 之间，不加这个参数就会解出乱码。

buffer的作用是处理"半帧"。网络传输是分块的，一个 SSE 帧可能被切成两半到达。所以每次都要把上次剩下的buffer和这次的新数据拼起来，再按\n\n切分，最后一段（可能不完整）留回buffer等下次。这个模式是所有流式解析的通用套路，务必理解。

3.3 在 Vue 组件里落地打字机效果

把上面的逻辑包进 Vue 组件，配合响应式数据，就能实现打字机效果。用 Vue 3 的 Composition API 写：

<template> <div class="chat"> <div class="answer">{{ answer }}</div> <button v-if="loading" @click="stop">停止生成</button> </div> </template> <script setup> import { ref } from 'vue' const answer = ref('') const loading = ref(false) let controller = null async function send(prompt) { answer.value = '' loading.value = true controller = new AbortController() try { await streamChat( prompt, (token) => { answer.value += token }, controller.signal ) } catch (e) { if (e.name !== 'AbortError') { answer.value += '\n[出错了，请重试]' } } finally { loading.value = false controller = null } } function stop() { controller?.abort() loading.value = false } </script>

answer.value += token这一句就是打字机效果的全部秘密。每次收到一个 token 就追加到响应式字符串上，Vue 自动触发视图更新，用户就看到文字一个个冒出来。不需要什么定时器，不需要逐字动画，数据驱动视图，天然就是打字机。

这里有个性能细节：如果 token 来得非常密集（比如每秒上百个），频繁触发响应式更新会有性能压力。可以做一个小的批量合并——用一个缓冲区攒 token，每 16 毫秒（一帧）flush 一次。不过实测下来，主流模型的输出速度（每秒几十个 token）直接追加完全没问题，不用过早优化。

3.4 AbortController：让"停止生成"真正生效

AbortController是这套方案里最优雅的部分。它和fetch的signal配合，调用controller.abort()时，fetch会立即中断请求，reader.read()抛出AbortError，整个流被干净地切断。

关键在于：中断是双向的。前端 abort 后，HTTP 连接关闭，后端 FastAPI 那边会收到客户端断开的信号。如果你用的是StreamingResponse包生成器，生成器会在下一次yield时被取消（抛出asyncio.CancelledError）。这意味着后端也能及时停止大模型推理，省下算力。这一点很多人不知道，以为只是前端不显示了，其实后端也停了。

要正确处理这个取消，后端生成器里最好加try/finally：

async def ollama_stream(prompt: str): try: async with httpx.AsyncClient(timeout=None) as client: async with client.stream(...) as resp: async for line in resp.aiter_lines(): yield ... except asyncio.CancelledError: # 客户端断开，清理资源 raise finally: # 关闭上游连接等 pass

CancelledError必须重新raise，不能吞掉，否则 asyncio 的取消机制会出问题。这是 asyncio 编程的一个通用原则。

4. 那些让人抓狂的线上问题排查

4.1 文字"攒一批蹦一批"：缓冲惹的祸

最常见的现象：本地开发好好的，一部署到线上，打字机效果就变成了"卡几秒，蹦一大段，再卡几秒"。这几乎可以断定是中间层缓冲导致的。

排查顺序是这样的。先看 Nginx，默认proxy_buffering on，它会把上游响应攒起来。解决方法是针对这个 location 关掉缓冲：

location /chat/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; }

proxy_http_version 1.1和Connection ''是为了启用 chunked 传输，HTTP/1.0 不支持流式。chunked_transfer_encoding off在某些 Nginx 版本上反而必要，避免它自己再做一层分块。

如果 Nginx 关了还是不行，检查后端有没有加X-Accel-Buffering: no响应头，以及应用服务器（uvicorn）本身有没有缓冲。uvicorn 默认不缓冲，但如果前面套了 gunicorn，要注意 worker 配置。

再往上，如果用了云厂商的负载均衡或 CDN，它们也可能缓冲。这类中间层通常有"关闭缓冲"或"流式转发"的开关，需要单独确认。排查这类问题的通用思路是：从最靠近浏览器的环节开始，逐层往上游排除，每层都确认它是否在缓冲。

4.2 idle timeout 断流的完整排查链路

回到那个热词报错idle timeout waiting for sse。我遇到过好几次，排查下来原因各不相同，这里把完整链路梳理一遍。

第一步，确认是不是首 token 太慢。如果模型 prefill 就要 8 秒，而网关空闲超时是 5 秒，那必然断。验证方法是在后端打日志，记录请求开始到第一个 token 的时间。如果确实慢，加心跳是正解。

第二步，确认心跳有没有真正发出去。有时候代码里写了心跳，但被缓冲卡住了，心跳根本没到网关。这时候要抓包或者看网关日志，确认连接上有没有数据流动。

第三步，检查网关的空闲超时配置。不同厂商叫法不一样，有的叫idle_timeout，有的叫read_timeout，有的叫keepalive_timeout。要找到实际生效的那个值，让心跳间隔小于它。

第四步，考虑客户端侧。有些浏览器或运行环境对长时间无数据的连接也会主动断。这种情况心跳同样能解决。

把这四步走完，基本能定位。我的经验是，心跳是解决 idle timeout 的万能药，只要保证连接上每隔几秒有数据流动，绝大多数空闲超时都不会触发。

4.3 中文乱码与半帧问题

前面提过TextDecoder的stream: true，这里再展开一下为什么它这么重要。UTF-8 编码里，一个汉字占 3 个字节。网络传输按字节流分块，完全可能把某个汉字的 3 个字节切成 1+2 两块。如果第一块到达时你直接decode，解码器看到不完整的字节序列，会输出替换字符（那个问号方块）。

stream: true的作用是让解码器"记住"这个不完整的尾巴，等下一块数据来了拼上再解。这是流式文本处理的必备参数，忘了它中文必乱码。

半帧问题同理。SSE 帧以\n\n分隔，但网络分块不会照顾你的帧边界。所以必须用buffer缓存不完整的尾部。判断标准很简单：split('\n\n')之后，最后一段如果没以\n\n结尾，它就是不完整的，要留回 buffer。上面代码里parts.pop()拿走的正是这最后一段。

提示：这两个问题（乱码、半帧）在英文场景下不容易暴露，因为英文是单字节，且短帧被切分的概率低。一旦上中文、上长文本，问题立刻显现。做国际化产品时尤其要注意。

5. 从能跑到好用：几个进阶优化点

5.1 首 token 加速与预连接

用户感知最快的优化，是让第一个字尽快出现。除了模型本身的推理速度，前端可以做一件事：在用户还在打字时，就提前建立连接。比如用户聚焦输入框时，先发一个空的预热请求，把 TCP 连接、TLS 握手、鉴权都走完。等用户真正提交时，省掉这些握手时间，首 token 能快几百毫秒。

后端侧，可以优化 prompt 的处理，减少 prefill 时间。如果用的是本地模型，确保模型已经加载进显存，别等请求来了才加载。

5.2 断线重连与续传

SSE 规范支持Last-Event-ID，服务端给每个事件带上id:，客户端断线重连时浏览器自动带上最后收到的 ID。服务端据此可以从断点继续推。实现续传需要服务端维护每个会话的生成状态，复杂度不低。对于 AI 对话，更常见的做法是：断线后不续传，而是让用户重新发起，或者前端把已收到的内容保留，提示"连接中断，点击继续"。

如果要做续传，思路是把生成结果缓存起来，重连时根据Last-Event-ID找到位置，把后面的内容补发。这需要生成过程与连接解耦——生成在一个独立任务里跑，结果写进队列或缓存，连接只是消费者。这样即使连接断了，生成也不中断，重连后接着读。

5.3 多路复用与并发控制

一个页面可能同时有多个流（比如多个对话窗口）。每个流占一条 HTTP 连接，浏览器对同域名 HTTP/1.1 连接数有限制（通常 6 个）。如果并发流多了，会排队。解决办法是升级到 HTTP/2，它支持在一条连接上多路复用，不受这个限制。

另外要注意，用户快速切换对话时，旧流的AbortController要及时 abort，否则旧流还在后台跑，浪费资源还可能串数据。我的做法是每个对话维护自己的 controller，切换时 abort 掉上一个。

5.4 安全与防爬的边界

流式接口也是接口，同样面临鉴权、限流、防刷的问题。几个实践要点：token 放请求头而不是 URL，避免被日志记录；对单用户的并发流数量做限制，防止有人开一堆连接拖垮后端；对请求频率做限流。这些和普通接口的安全策略一致，只是流式接口的连接存活时间长，限流维度要按"并发连接数"而不只是"请求数"来算。

至于防止查看源码、防爬虫这类需求，本质上和 SSE 无关，属于前端安全的通用话题。要清楚一点：前端代码最终都要下发到浏览器，任何"防查看"都只是提高门槛，做不到绝对。真正敏感的逻辑必须放后端。

6. 一套可复用的最小实现清单

把前面所有内容收敛成一套可以直接抄的清单，方便你快速搭起自己的流式对话。

后端（FastAPI）要点：

• 用StreamingResponse包异步生成器，media_type="text/event-stream"
• 每个数据帧格式data: {json}\n\n，token 包 JSON 防换行破坏
• 响应头加Cache-Control: no-cache、X-Accel-Buffering: no
• 等待首 token 期间发: heartbeat\n\n心跳，间隔 10 到 15 秒
• 生成器用try/except asyncio.CancelledError处理客户端断开，记得重新 raise
• 上游大模型请求设timeout=None，业务层自己做超时

前端（Vue）要点：

• 用fetch+ReadableStream，不用EventSource（要 POST 和鉴权）
• TextDecoder必须带{ stream: true }，否则中文乱码
• 用buffer处理半帧，按\n\n切分，尾部留回 buffer
• 用AbortController实现停止生成，前后端双向中断
• 收到[DONE]主动结束，别等连接自然关闭

部署（Nginx）要点：

• 针对流式 location 关proxy_buffering和proxy_cache
• 启用proxy_http_version 1.1，Connection ''
• 确认网关空闲超时大于心跳间隔

这套组合我在几个项目里反复用过，从本地开发到线上部署，基本没再遇到断流和乱码。真正容易翻车的从来不是 SSE 协议本身——它简单得很——而是链路上那些默认开启缓冲、默认设置超时的中间层。把每一层都摸清楚，流式输出就是件很稳的事。

最后分享一个我踩过的坑：有次线上打字机效果时好时坏，排查半天发现是负载均衡有两台机器，其中一台的配置没同步，缓冲没关。所以改配置时一定要确认所有实例都生效，别只改了一台就以为万事大吉。这种"部分生效"的问题最难查，因为现象是间歇性的，很容易被误判成网络抖动。