在开发对话类应用时，类似ChatGPT的流式响应能极大提升用户体验，用户不需要等待所有内容生成完毕就能看到逐步输出的结果。Flask作为轻量级的Python Web框架，本身就支持流式响应能力，结合合适的通信技术就能实现这个效果。
实现流式响应的核心原理
传统的HTTP响应是一次性返回所有内容，而流式响应要求服务端在处理过程中逐步向客户端发送数据，客户端实时接收并渲染。Flask中可以通过stream_with_context装饰器配合生成器函数实现响应流，而更适合文本流式场景的通信方式是Server-Sent Events（简称SSE），它允许服务端单向向客户端持续推送数据，无需客户端反复请求。
后端Flask实现步骤
1. 基础环境准备
首先需要安装Flask框架，执行以下命令即可：
pip install flask
2. 编写流式响应接口
我们需要定义一个返回SSE格式数据的接口，使用生成器逐步产出内容，同时使用stream_with_context确保请求上下文在流传输过程中可用。以下是完整示例代码：
from flask import Flask, Response, stream_with_context, request
import time
app = Flask(__name__)
# 模拟逐步生成内容的函数，类似大模型逐字输出
def generate_stream_content(prompt):
# 模拟大模型返回的拆分内容片段
content_parts = [
"你好",
"，我是",
"基于Flask实现的",
"流式响应",
"示例。",
"当前你输入的内容是：",
prompt
]
for part in content_parts:
# 模拟生成每个片段的延迟，模拟真实处理的耗时
time.sleep(0.5)
# SSE格式要求每个消息以data:开头，两个换行结束
yield f"data: {part}\n\n"
@app.route("/stream_chat", methods=["POST"])
def stream_chat():
prompt = request.form.get("prompt", "")
# 返回流式响应，设置Content-Type为SSE要求的类型
return Response(
stream_with_context(generate_stream_content(prompt)),
mimetype="text/event-stream"
)
if __name__ == "__main__":
app.run(debug=True, port=5000)3. 代码说明
上面的代码中，generate_stream_content是生成器函数，负责逐步产出内容，每个产出内容都按照SSE的格式包装，即data: 内容\n\n，这样前端才能正确解析。Response对象指定mimetype为text/event-stream，告诉浏览器这是SSE流数据，不要等待全部接收完成再处理。
前端接收流式数据实现
前端可以使用原生的EventSource对象或者普通的fetch API来接收SSE流，以下是使用fetch的实现示例，兼容性更好：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>Flask流式响应示例</title>
</head>
<body>
<input type="text" id="promptInput" placeholder="请输入内容" style="width: 300px;">
<button onclick="sendRequest()">发送</button>
<div id="result" style="margin-top: 20px; padding: 10px; border: 1px solid #ccc;"></div>
<script>
function sendRequest() {
const prompt = document.getElementById("promptInput").value;
const resultDiv = document.getElementById("result");
resultDiv.innerHTML = ""; // 清空之前的结果
fetch("http://127.0.0.1:5000/stream_chat", {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded"
},
body: `prompt=${encodeURIComponent(prompt)}`
}).then(response => {
const reader = response.body.getReader();
const decoder = new TextDecoder();
// 循环读取流数据
function readStream() {
reader.read().then(({ done, value }) => {
if (done) {
console.log("流传输结束");
return;
}
// 解码二进制数据为文本
const text = decoder.decode(value);
// 解析SSE格式的数据，跳过心跳等空消息
const lines = text.split("\n").filter(line => line.startsWith("data:"));
lines.forEach(line => {
const content = line.replace("data: ", "");
resultDiv.innerHTML += content;
});
// 继续读取下一段数据
readStream();
});
}
readStream();
});
}
</script>
</body>
</html>注意事项
- 生产环境中需要关闭Flask的debug模式，避免出现流传输异常。
- 如果使用了反向代理（如Nginx），需要配置代理支持流式传输，避免代理缓冲数据导致流式效果失效，例如Nginx中可以设置
proxy_buffering off;。 - SSE是单向通信，只适合服务端向客户端推送数据的场景，如果需要双向通信可以考虑WebSocket，但流式文本输出场景SSE更简单轻量。
提示：如果后端需要对接真实的AI大模型API，只需要把generate_stream_content函数中的模拟内容替换为调用大模型流式接口返回的内容即可，核心的流传输逻辑不需要修改。
Flask流式响应Server-Sent_Events实时通信stream_with_context修改时间：2026-05-31 23:50:27