在我们和 AI 聊天中，AI Chat 都采用了一种「打字机」式效果的实时响应方式，AI 的回答逐字逐句地呈现在我们眼前。
在实现这个功能的技术方案选择上，不管是 DeepSeek ，还是腾讯元宝都在这个对话逻辑中选择了使用 SSE，如下面 4 张图：
这是为啥，它有什么优势，以及如何实现的。
SSE 的优势
因为它在该场景下的优势非常明显，主要是以下 4 点：
1.场景的高度匹配。
AI 对话的核心交互模式是：
- 用户发起一次请求（发送问题）。
- AI 进行一次持续的、单向的响应输出（生成回答）。
SSE 的单向通信（服务器 → 客户端）模型与这个场景高度切尔西。它就像一条专门为服务器向客户端输送数据的「单行道」，不多一分功能，也不少一毫关键。
相比之下，WebSocket 提供的是全双工通信，即客户端和服务器可以随时互相发送消息，这是一条「双向八车道高速公路」。为了实现 AI 的流式回答，我们只需要其中一个方向的车道，而另一方向的车道（客户端 → 服务器）在 AI 回答期间是闲置的。在这种场景下使用 WebSocket，无异于「杀鸡用牛刀」，引入了不必要的复杂性。用户的提问完全可以通过另一个独立的、常规的 HTTP POST 请求来完成，这让整个系统的架构更加清晰和解耦。
2.HTTP 原生支持，与生俱来的优势
SSE 是建立在标准 HTTP 协议之上的。这意味着：
- 无需协议升级：SSE 连接的建立就是一个普通的 HTTP GET 请求，服务器以
Content-Type: text/event-stream响应。而 WebSocket 则需要一个特殊的「协议升级」（Upgrade）握手过程，从 HTTP 切换到ws://或wss://协议，过程相对复杂。 - 兼容性极佳：由于它就是 HTTP，所以它能天然地穿透现有的网络基础设施，包括防火墙、企业代理、负载均衡器等，几乎不会遇到兼容性问题。WebSocket 有时则会因为代理服务器不支持其协议升级而被阻断。并且云服务商对于 Websocket 的支持并不是很完善。
- 实现轻量：无论是前端还是后端，实现 SSE 都非常简单。前端一个
EventSourceAPI 即可搞定，后端也只需遵循简单的文本格式返回数据流。这大大降低了开发和维护的成本。
3.断网自动重连，原生容错
这是 SSE 的「王牌特性」，尤其在网络不稳定的移动端至关重要。
想象一下，当 AI 正在为我们生成一篇长文时，我们的手机网络突然从 Wi-Fi 切换到 5G，造成了瞬间的网络中断。
- 如果使用 WebSocket：连接会断开，我们需要手动编写复杂的 JavaScript 代码来监听断开事件、设置定时器、尝试重连、并在重连成功后告知服务器从哪里继续，实现起来非常繁琐。
- 如果使用 SSE：浏览器会自动处理这一切。
EventSourceAPI 在检测到连接中断后，会自动在几秒后（这个间隔可以通过retry字段由服务器建议）发起重连。更棒的是，它还会自动将最后收到的消息id通过Last-Event-ID请求头发送给服务器，让服务器可以从中断的地方继续推送数据，实现无缝的「断点续传」。当然，Last-Event-ID 的处理逻辑需要服务端来处理。
这种由浏览器原生提供的、可靠的容错机制，为我们省去了大量心力，并极大地提升了用户体验。
4. 易于调试
因为 SSE 的数据流是纯文本并通过标准 HTTP 传输，调试起来异常方便：
- 我们可以直接在浏览器地址栏输入 SSE 端点的 URL，就能在页面上看到服务器推送的实时文本流。
- 我偿可以使用任何 HTTP 调试工具，如
curl命令行或者 Chrome 开发者工具的「网络」面板，清晰地看到每一次数据推送的内容。
而 WebSocket 的数据传输基于帧，格式更复杂，通常需要专门的工具来调试和分析。
使用 SSE 实现「打字机」效果
1.后端——调用大模型并开启「流式」开关
当后端服务器收到用户的问题后，它并不等待大语言模型生成完整的答案。相反，它在调用 LLM 的 API 时，会传递一个关键参数：stream=True。
这个参数告诉 LLM：「请不要等全部内容生成完再给我，而是每生成一小部分（通常是一个或几个‘词元’/Token），就立刻通过数据流发给我。」
下面是一个使用 Python 和 OpenAI API 的后端伪代码示例：
python
from flask import Flask, Response, request
import openai
import json
app = Flask(__name__)
# 假设 OpenAI 的 API Key 已经配置好
# openai.api_key = "YOUR_API_KEY"
@app.route('/chat-stream')
def chat_stream():
prompt = request.args.get('prompt')
def generate_events():
try:
# 关键：设置 stream=True
response_stream = openai.ChatCompletion.create(
model="gpt-4", # 或其他模型
messages=[{"role": "user", "content": prompt}],
stream=True
)
# 遍历从大模型返回的数据流
for chunk in response_stream:
# 提取内容部分
content = chunk.choices[0].delta.get('content', '')
if content:
# 关键：将每个内容块封装成 SSE 格式并 yield 出去
# 使用 json.dumps 保证数据格式正确
sse_data = f"data: {json.dumps({'token': content})}\n\n"
yield sse_data
# (可选) 发送一个结束信号
yield "event: done\ndata: [STREAM_END]\n\n"
except Exception as e:
# 错误处理
error_message = f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"
yield error_message
# 返回一个流式响应，并设置正确的 MIME 类型
return Response(generate_events(), mimetype='text/event-stream')
if __name__ == '__main__':
app.run(threaded=True)
在这段代码中，有几个关键点：
- stream=True：向 LLM 请求流式数据。
- 生成器函数（
generate_events）：使用yield关键字，每从 LLM 收到一小块数据，就立即将其处理成 SSE 格式（data: ...\n\n）并发送出去。 Response(..., mimetype='text/event-stream')：告诉浏览器，这是一个 SSE 流，请保持连接并准备接收事件。
2.SSE 格式的约定
后端 yield 的每一条 data: 都像是一个个装着文字的信封，通过 HTTP 长连接这个管道持续不断地寄给前端。
前端收到的原始数据流看起来就像这样：
vbnet
data: {"token": "当"}
data: {"token": "然"}
data: {"token": "，"}
data: {"token": "很"}
data: {"token": "乐"}
data: {"token": "意"}
data: {"token": "为"}
data: {"token": "您"}
data: {"token": "解"}
data: {"token": "答"}
data: {"token": "。"}
event: done
data: [STREAM_END]
看 DeepSeek 和腾讯元宝的数据格式，略有不同，不过有一点，都是直接用的 JSON 格式，且元宝的返回值相对冗余一些。 且都没有 data: 的前缀。
3.前端监听并拼接成「打字机」
前端的工作就是接收这些「信封」，拆开并把里面的文字一个个地追加到聊天框里。
html
<!-- HTML 结构 -->
<div id="chat-box"></div>
<input id="user-input" type="text">
<button onclick="sendMessage()">发送</button>
<script>
let eventSource;
function sendMessage() {
const input = document.getElementById('user-input');
const prompt = input.value;
input.value = '';
const chatBox = document.getElementById('chat-box');
// 创建一个新的 p 标签来显示 AI 的回答
const aiMessageElement = document.createElement('p');
aiMessageElement.textContent = "AI: ";
chatBox.appendChild(aiMessageElement);
// 建立 SSE 连接
eventSource = new EventSource(`/chat-stream?prompt=${encodeURIComponent(prompt)}`);
// 监听 message 事件，这是接收所有 "data:" 字段的地方
eventSource.onmessage = function(event) {
// 解析 JSON 字符串
const data = JSON.parse(event.data);
const token = data.token;
if (token) {
// 将新收到的文字追加到 p 标签末尾
aiMessageElement.textContent += token;
}
};
// 监听自定义的 done 事件，表示数据流结束
eventSource.addEventListener('done', function(event) {
console.log('Stream finished:', event.data);
// 关闭连接，释放资源
eventSource.close();
});
// 监听错误
eventSource.onerror = function(err) {
console.error("EventSource failed:", err);
aiMessageElement.textContent += " [出现错误，连接已断开]";
eventSource.close();
};
}
</script>
这段代码主要有如下的点：
new EventSource(...)：发起连接。eventSource.onmessage：这是主要的处理函数。每当收到一条data:消息，它就会被触发。aiMessageElement.textContent += token;：这就是「打字机」效果的精髓所在——持续地在同一个 DOM 元素上追加内容，而不是创建新的元素。eventSource.close()：在接收到结束信号或发生错误后，务必关闭连接，以避免不必要的资源占用。
EventSource 的来源与发展
在 SSE 标准化之前，Web 的基础是 HTTP 的请求-响应模型：客户端发起请求，服务器给予响应，然后连接关闭。这种模式无法满足服务器主动向客户端推送信息的需求。为了突破这一限制，开发者们创造了多种「模拟」实时通信的技术。
- 短轮询：这是最简单直接的方法。客户端通过 JavaScript 定时（如每隔几秒）向服务器发送一次 HTTP 请求，询问是否有新数据。无论有无更新，服务器都会立即返回响应。这种方式实现简单，但缺点显而易见：存在大量无效请求，实时性差，并且对服务器造成了巨大的负载压力。
- 长轮询：为了改进短轮询，长轮询应运而生。客户端发送一个请求后，服务器并不会立即响应，而是会保持连接打开，直到有新数据产生或者连接超时。一旦服务器发送了数据并关闭了连接，客户端会立即发起一个新的长轮询请求。这大大减少了无效请求，提高了数据的实时性，但仍然存在 HTTP 连接的开销，并且实现起来相对复杂。
- Comet：一个时代的统称：在 HTML5 标准化之前，像长轮询和 HTTP 流（HTTP Streaming）这样的技术被统称为 Comet。 Comet 是一种设计模式，它描述了使用原生 HTTP 协议在服务器和浏览器之间实现持续、双向交互的多种技术集合。 它是对实现实时 Web 应用的早期探索，为后来更成熟的标准化技术（如 SSE 和 WebSockets）奠定了基础。
随着 Web 应用对实时性要求的日益增长，需要一种更高效、更标准的解决方案。
- WHATWG 的早期工作：SSE 机制最早由 Ian Hickson 作为「WHATWG Web Applications 1.0」提案的一部分，于 2004 年开始进行规范制定。
- Opera 的先行实践：2006 年 9 月，Opera 浏览器在一项名为“Server-Sent Events”的功能中，率先实验性地实现了这项技术，展示了其可行性。
- HTML5 标准化：最终，SSE 作为 HTML5 标准的一部分被正式确立。它通过定义一种名为
text/event-stream的 MIME 类型，让服务器可以通过一个持久化的 HTTP 连接向客户端发送事件流。 客户端一旦与服务器建立连接，就会保持该连接打开，持续接收服务器发送的数据。
SSE 的本质是利用了 HTTP 的流信息机制。服务器向客户端声明接下来要发送的是一个数据流，而不是一次性的数据包，从而实现了一种用时很长的「下载」过程，服务器得以在此期间不断推送新数据。
其返回内容标准大概如下：
event-source 必须编码成 utf8 的格式，消息的每个字段都是用”\n”来做分割，下面 4 个规范定义好的字段：
- Event: 事件类型
- Data: 发送的数据
- ID：每一条事件流的ID
- Retry: 告知浏览器在所有的连接丢失之后重新开启新的连接等待的事件，在自动重连连接的过程中，之前收到的最后一个事件流ID会被发送到服务器
在实际中，大概率不一定按这个标准来实现。对于一些重连的逻辑需要自行实现。
现在大部分的浏览器都兼容这个特性，如图：
参考资料：
- en.wikipedia.org/wiki/Server…
- learn.microsoft.com/zh-cn/azure…
- www.cnblogs.com/openmind-in…
- javascript.ruanyifeng.com/htmlapi/eve…
以上。