Flask流式传输模拟ChatGPT实时响应的实现原理
ChatGPT的实时响应本质上是服务端在生成内容的过程中，逐步将已生成的部分推送到客户端，客户端每接收到一部分数据就更新展示内容，从而呈现出逐字或逐段输出的效果。在Flask中实现这种效果，核心是利用流式响应机制，让视图函数返回一个可迭代的生成器对象，Flask会持续从这个生成器中获取内容并发送给客户端，而不是等待所有内容生成完成后再返回。
核心技术选型
实现流式传输有两种常见的技术方案，分别是分块传输编码和Server Sent Events（SSE），两者各有适用场景：
- 分块传输编码：是HTTP协议的标准特性，服务端将响应内容分成多个块依次发送，不需要提前声明内容总长度，适合简单的流式场景。
- Server Sent Events：是HTML5规范中的服务端推送技术，基于HTTP协议，服务端可以向客户端持续发送事件流，客户端通过EventSource接口监听接收，支持自动重连，更适合需要稳定推送的场景。
基于SSE的Flask流式响应实现
SSE是模拟ChatGPT实时响应的更优选择，因为它有标准化的事件格式，客户端处理起来更方便。下面是实现的具体步骤和代码。
1. 服务端实现
首先需要安装Flask库，然后编写支持SSE的流式响应视图函数，注意响应头的Content-Type要设置为text/event-stream，同时关闭缓存，避免浏览器缓存流式内容。
from flask import Flask, Response
import time
import json
app = Flask(__name__)
def generate_stream_response():
# 模拟ChatGPT逐字生成内容的过程，实际场景中这里可以替换为模型推理的逻辑
content_list = ["你", "好", "，", "我", "是", "AI", "助", "手", "，", "有", "什", "么", "可", "以", "帮", "你", "？"]
for char in content_list:
# SSE格式要求每个事件以data:开头，以两个换行符结尾
yield f"data: {json.dumps({'content': char})}\n\n"
# 模拟生成间隔，避免发送速度过快
time.sleep(0.1)
# 发送结束标志
yield f"data: {json.dumps({'done': True})}\n\n"
@app.route('/stream_chat', methods=['GET'])
def stream_chat():
return Response(
generate_stream_response(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no' # 关闭Nginx等代理的缓冲，确保流式内容实时发送
}
)
if __name__ == '__main__':
app.run(debug=True, threaded=True)2. 前端接收实现
前端使用EventSource接口监听服务端的事件流，每接收到一条数据就更新页面展示内容，直到接收到结束标志。
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>流式响应演示</title>
</head>
<body>
<div id="chat-content"></div>
<script>
const chatContent = document.getElementById('chat-content');
// 创建EventSource连接，指向服务端的流式接口
const eventSource = new EventSource('/stream_chat');
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
// 如果接收到结束标志，关闭连接
if (data.done) {
eventSource.close();
return;
}
// 将接收到的内容追加到页面
chatContent.innerHTML += data.content;
};
eventSource.onerror = function() {
console.log('连接出错，已关闭');
eventSource.close();
};
</script>
</body>
</html>基于分块传输的简化实现
如果不需要SSE的事件格式支持，也可以使用更简单的分块传输方式，直接返回生成器对象，Flask会自动使用分块传输编码发送内容。
from flask import Flask, Response
import time
app = Flask(__name__)
def generate_chunk_response():
content_list = ["你", "好", "，", "这", "是", "分", "块", "传", "输", "的", "测", "试", "。"]
for char in content_list:
yield char
time.sleep(0.1)
@app.route('/chunk_chat', methods=['GET'])
def chunk_chat():
return Response(generate_chunk_response(), mimetype='text/plain')
if __name__ == '__main__':
app.run(debug=True)这种方式的客户端处理更简单，直接监听响应进度即可：
const chatDiv = document.getElementById('chat-content');
fetch('/chunk_chat')
.then(response => {
const reader = response.body.getReader();
const decoder = new TextDecoder();
function readChunk() {
reader.read().then(({done, value}) => {
if (done) {
return;
}
const chunk = decoder.decode(value, {stream: true});
chatDiv.innerHTML += chunk;
readChunk();
});
}
readChunk();
});注意事项
- 流式响应适合处理耗时较长的生成任务，比如大语言模型的推理过程，不要在简单接口中使用，避免占用连接资源。
- 如果应用部署在Nginx等反向代理后面，需要关闭代理的响应缓冲，否则流式内容会被代理缓存到完整后才转发给客户端，无法达到实时效果。
- 生产环境中建议给流式接口添加合理的超时时间和错误重试机制，提升稳定性。
- 如果需要携带用户认证信息，SSE的EventSource默认不支持自定义请求头，可以通过在URL中携带令牌参数或者使用分块传输+fetch的方式实现。
Flaskstreaming_responseServer_Sent_Events实时响应流式传输修改时间：2026-06-03 23:58:32