python flask框架 SSE流式返回、跨域、报错

from flask import Flask, Blueprint, request, Response, stream_with_context, abort from flask_cors import CORSreturn Response(xxx) 是手动构造响应对象自由度最高可以实现流式输出 Response( response, # 【最重要】响应内容可以是字符串、迭代器、生成器 content_type, # 响应数据类型 headers # 自定义http响应头 ) search_sse_v1它不是普通函数是生成器函数内部有 yield stream_with_context(生成器) 作用保证生成器迭代时flask请求上下文不会销毁sse必须加 Flask 工作机制 请求进来 → 进入视图函数 search_sse_v1() 视图函数执行到 return Response(...) → 视图函数执行结束 默认请求上下文request、current_app 等直接销毁 但是 生成器是延后执行的视图函数search_sse_v1 return 之后框架才会慢慢迭代生成器、读取 yield 的数据。 上下文销毁后生成器里面一旦用到 request/app直接报错 stream_with_context(迭代器) 作用把请求上下文 “保存” 住迭代生成器的时候上下文依然有效。只要做流式 SSE必须套这个! content_typetext/event-stream 标准SSE响应类型告诉客户端这是流式长连接 本次返回不是普通 json/html是 SSE 流式长连接 如果不写这个 浏览器会当成普通文本一次性接收不会开启流式解析SSE 直接失效 常见对比 application/json普通接口一次性返回 json text/html网页 text/event-streamSSE 服务端推送流 额外响应头 Cache-Control: no-cache 禁止浏览器缓存流数据 Connection: keep-alive 保持长连接 SSE 是长连接不加这两个头部分浏览器 / 代理会提前切断流。 1. 浏览器POST请求 /v1/search/sse 2. 进入视图函数 search_sse_v1() 3. 执行参数解析、鉴权、各种参数校验 4. 调用 search_sse_v1(参数) → 创建生成器⚠️ 此时生成器内部代码还没运行 5. 用 stream_with_context 包裹生成器 6. 构造 Response 对象 return 返回 【视图函数到此执行完毕】 分割线 7. Flask框架发现Response内部是一个生成器 8. 框架开始循环迭代生成器执行生成器内部代码 9. 生成器 yield 一段sse消息 10. Flask立刻把这段数据通过网络发给浏览器 11. 暂停继续循环等待下一次yield 12. 所有yield完成连接关闭 视图函数代码一次性跑完生成器代码是 return 之后框架逐步慢慢执行 Q1为什么不能直接 return 生成器 Flask 不识别直接返回的生成器会报错。必须套 Response。 Q2stream_with_context 可以省略吗 如果你的生成器里面完全不使用 request /current_app/logger勉强能跑 一旦需要打印日志、拿请求信息必崩。开发规范SSE 一律加上。 Q3Response 是把所有 yield 数据收集起来再发送 ❌ 不是 Response 收到生成器之后迭代一次 yield就立刻往外推送一次不会缓存全部内容。这就是流式的本质。 极简总结背诵版 return Response()手动构造 HTTP 响应对象支持传入生成器实现分段输出 第一个参数放被 stream_with_context 包裹的生成器 stream_with_context保留请求上下文防止生成器运行时环境失效 content_typetext/event-stream标记这是 SSE 流让客户端启用流式解析 视图函数先执行完所有参数校验return Response 之后框架才逐步执行生成器里面的 yield 逻辑。# Response 、 stream_with_context、abort from flask import Flask, Blueprint, request, Response, stream_with_context, abort # 跨域 from flask_cors import CORS import json import time # 初始化flask应用 app Flask(__name__) CORS(app) # 注册蓝图和你代码保持一致 search_sse Blueprint(search_sse, __name__) def search_sse_v1(sid): ✅ SSE核心生成器函数依靠 yield 分段推送数据 不能一次性return循环不断yield分片 # 模拟大模型/搜索流式分片结果 answer_chunks [ Flask SSE流式返回讲解开始。, SSE全称Server-Sent Events服务端单向推送。, 依靠 yield 持续输出文本分片。, stream_with_context 用来保留flask请求上下文。, 消息格式必须遵循 data:xxx\\n\\n 规范。 ] # 循环推送每一段内容 for chunk in answer_chunks: resp_data { sid: sid, content: chunk, finished: False # 是否结束标记 } # SSE标准格式重点 # data: 内容 两个换行 \n\n # ensure_asciiFalse,不转义中文等非 ASCII 字符直接输出原始文字 # ensure_asciiTrue,前端收到的是 \uxxxx 转义字符串 yield fdata: {json.dumps(resp_data, ensure_asciiFalse)}\n\n time.sleep(0.6) # 模拟接口处理耗时方便观察流式效果 # 推送结束消息通知前端流完成 end_data { sid: sid, content: , finished: True } yield fdata: {json.dumps(end_data, ensure_asciiFalse)}\n\n # ---------------- SSE接口路由 核心代码 ---------------- search_sse.route(/v1/search/sse, methods[POST]) def search_sse_controller(): try: # 1、获取原始请求body 二进制流 data request.get_data() # 2、解析json参数 try: data json.loads(data) except Exception as e: # json解析失败参数错误 raise Exception(parameter_error) app.logger.info(fsearch_sse_v1, data: {data}) # 流式返回核心 return Response(xxx) 是手动构造响应对象自由度最高可以实现流式输出 Response( response, # 【最重要】响应内容可以是字符串、迭代器、生成器 content_type, # 响应数据类型 headers # 自定义http响应头 ) search_sse_v1它不是普通函数是生成器函数内部有 yield stream_with_context(生成器) 作用保证生成器迭代时flask请求上下文不会销毁sse必须加 Flask 工作机制 请求进来 → 进入视图函数 search_sse_v1() 视图函数执行到 return Response(...) → 视图函数执行结束 默认请求上下文request、current_app 等直接销毁 但是 生成器是延后执行的视图函数search_sse_v1 return 之后框架才会慢慢迭代生成器、读取 yield 的数据。 上下文销毁后生成器里面一旦用到 request/app直接报错 stream_with_context(迭代器) 作用把请求上下文 “保存” 住迭代生成器的时候上下文依然有效。只要做流式 SSE必须套这个! content_typetext/event-stream 标准SSE响应类型告诉客户端这是流式长连接 本次返回不是普通 json/html是 SSE 流式长连接 如果不写这个 浏览器会当成普通文本一次性接收不会开启流式解析SSE 直接失效 常见对比 application/json普通接口一次性返回 json text/html网页 text/event-streamSSE 服务端推送流 额外响应头 Cache-Control: no-cache 禁止浏览器缓存流数据 Connection: keep-alive 保持长连接 SSE 是长连接不加这两个头部分浏览器 / 代理会提前切断流。 1. 浏览器POST请求 /v1/search/sse 2. 进入视图函数 search_sse_v1() 3. 执行参数解析、鉴权、各种参数校验 4. 调用 search_sse_v1(参数) → 创建生成器⚠️ 此时生成器内部代码还没运行 5. 用 stream_with_context 包裹生成器 6. 构造 Response 对象 return 返回 【视图函数到此执行完毕】 分割线 7. Flask框架发现Response内部是一个生成器 8. 框架开始循环迭代生成器执行生成器内部代码 9. 生成器 yield 一段sse消息 10. Flask立刻把这段数据通过网络发给浏览器 11. 暂停继续循环等待下一次yield 12. 所有yield完成连接关闭 视图函数代码一次性跑完生成器代码是 return 之后框架逐步慢慢执行 Q1为什么不能直接 return 生成器 Flask 不识别直接返回的生成器会报错。必须套 Response。 Q2stream_with_context 可以省略吗 如果你的生成器里面完全不使用 request /current_app/logger勉强能跑 一旦需要打印日志、拿请求信息必崩。开发规范SSE 一律加上。 Q3Response 是把所有 yield 数据收集起来再发送 ❌ 不是 Response 收到生成器之后迭代一次 yield就立刻往外推送一次不会缓存全部内容。这就是流式的本质。 极简总结背诵版 return Response()手动构造 HTTP 响应对象支持传入生成器实现分段输出 第一个参数放被 stream_with_context 包裹的生成器 stream_with_context保留请求上下文防止生成器运行时环境失效 content_typetext/event-stream标记这是 SSE 流让客户端启用流式解析 视图函数先执行完所有参数校验return Response 之后框架才逐步执行生成器里面的 yield 逻辑。 return Response( responsestream_with_context(search_sse_v1(data.get(sid))), content_typetext/event-stream, headers{ Cache-Control: no-cache, # 禁止浏览器缓存流式数据 Connection: keep-alive # 保持TCP长连接不要传输完立刻断开 } ) except Exception as e: app.logger.error(fSSE接口异常: {str(e)}) # abort() 是 Flask 内置函数作用立刻终止当前请求直接给前端返回指定 HTTP 错误码后面代码不再执行。 拆解 abort(500, e.args[0]) 第一个参数 500 HTTP 500 服务器内部错误。 代表后端代码 / 业务逻辑出现异常。 第二个参数 e.args[0] e上方 except Exception as e 捕获到的异常对象 e.args元组存储异常提示文字 e.args[0]取出异常里自定义的错误文本比如你抛的 Exception(参数错误:search) 整体效果 接口报错时前端收到 HTTP 状态码 500同时携带错误描述字符串。 abort(500, e.args[0]) # 注册蓝图到app app.register_blueprint(search_sse) if __name__ __main__: # ✅ threadedTrue 必须开启单线程模式流式会阻塞 app.run(host127.0.0.1, port5000, debugTrue, threadedTrue)!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSSE POST流式测试/title /head body h3流式输出结果/h3 div idresult stylewhite-space: pre-wrap;font-size:16px;/div script async function startSSE() { const resultDom document.getElementById(result); // 请求参数和后端接口要求匹配 const reqBody { search: 学习Flask SSE流式接口, search_type: 1, sid: null, hid: null, regenerate: null, tab_no: all, code: }; try { const response await fetch(http://127.0.0.1:5000/v1/search/sse, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(reqBody) }); // 获取流读取器 const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const {done, value} await reader.read(); if (done) { resultDom.innerHTML brbr✅ 流式传输全部结束; break; } // 二进制转字符串放入缓冲区 buffer decoder.decode(value); // SSE消息分隔符两条换行 \n\n const chunks buffer.split(\n\n); buffer chunks.pop(); // 剩余不完整内容留在buffer下次继续解析 for (const msg of chunks) { if (!msg.startsWith(data: )) continue; // 截取data: 后面的json字符串 const jsonStr msg.replace(data: , ); const data JSON.parse(jsonStr); console.log(收到分片, data); resultDom.innerHTML data.content; if (data.finished) { reader.cancel(); // 主动关闭流 } } } } catch (err) { resultDom.innerHTML br❌ 请求异常${err.message}; console.error(err); } } // 执行 startSSE(); /script /body /html