背景：
最近在做大模型开发的时候，要实现流式输出，应小伙伴要求，需要把python的流式输出代码也暴露出来，因此做个总结，跟之前的成为系列。
java的请看这篇文章：
SpringBoot手动实现流式输出方案整理以及SSE规范输出详解-CSDN博客
Https流式输出问题处理
Https流式输出一次输出一大段，一卡一卡的-解决方案-CSDN博客
代码实现：
前提条件说明：
文中采用Flask框架，确保你已经安装了 Flask ：pip install flask
大模型采用deepseek
接口层调用：
from flask import Flask, Response, render_template_string
from flask import requestapp = Flask(__name__)#Flask测试
@app.route('/test', methods=['GET'])
def test():return "欢迎使用DeepSeek流式输出示例!"@app.route('/stream/post', methods=['POST'])
def stream3():text = '详解一下flask框架'return Response(deepseek.event_stream(text), mimetype='text/event-stream')# 后端接口为post请求示例
@app.route('/index1')
def index2():return render_template_string('''<!DOCTYPE html><html><head><title>Flask流式输出示例</title></head><body><h1>Flask流式输出演示</h1><div id="output" style="border: 1px solid #ccc; padding: 10px; margin: 10px;"></div><button onclick="startStream()">开始流式请求</button><script>const output = document.getElementById('output');function startStream() {const prompt = '测试输入'; // 你可以从输入框获取fetch('/stream/post', {method: 'POST',headers: {'Content-Type': 'application/json',},body: JSON.stringify({ prompt: prompt }),}).then(response => {const reader = response.body.getReader();const decoder = new TextDecoder('utf-8');function read() {reader.read().then(({ done, value }) => {if (done) {alert('处理完成！结果已保存');return;}const chunk = decoder.decode(value);output.textContent += chunk;read(); // 继续读取下一个块});}read();});}</script></body></html>''')@app.route('/stream/get', methods=['GET'])
def stream():# text = request.args.get('text', 'Hello, this is a stream example from DeepSeek!')text = '详解一下flask框架'return Response(deepseek.event_stream(text), mimetype='text/event-stream')# 后端接口为get请求示例
@app.route('/')
def index():return """<!DOCTYPE html><html><body><h1>Stream Test</h1><button onclick="startStream()">Start Stream</button><div id="output" style="margin-top: 20px; border: 1px solid #ccc; padding: 10px;"></div><script>function startStream() {const eventSource = new EventSource('/stream/get?text=详解一下flask框架');const outputDiv = document.getElementById('output');eventSource.onmessage = function(e) {outputDiv.innerHTML += e.data + ' ';};eventSource.onerror = function(e) {console.error('EventSource failed:', e);eventSource.close();};}</script></body></html>"""if __name__ == '__main__':app.run(port=8888, host='0.0.0.0')封装的方法内容
from openai import OpenAIclient = OpenAI(api_key="YOUR_API_KEY",base_url="http://192.168.0.0:8080/v1"
)# 流式输出示例
def process(content: str):response = client.chat.completions.create(model="deepseek-r1:14b",messages=[{"role": "user", "content": content}],stream=True)return responsedef event_stream(content: str):try:# "你是一名Java工程师,用Java实现快速排序"result = ''resMap = {}content = process(content)for data in content: # process()返回流式数据# print(data.choices[0].delta.content, end="", flush=True)result += data.choices[0].delta.contentresMap['content'] = data.choices[0].delta.contentresMap['request_id'] = '123456789'resMap['userid'] = '789465'# yield f"data: {data.choices[0].delta.content}\n\n".encode('utf-8')yield f"data: {resMap}\n\n".encode('utf-8')# 后置处理after_event(result)resMap['status'] = 'completed'yield f"data: {resMap}\n\n"except Exception as e:raise f"data: 错误发生：{str(e)}\n\n"
def after_event(result):# 保存数据库内容print('保存数据库\n')print("========:"+result)上面的内部测试代码为：
response = process("你是一名Java工程师,用Java实现快速排序")
for chunk in response:print(chunk.choices[0].delta.content, end="", flush=True)
web运行效果：
内容为博主空闲时间做的demo示例，修改扩展之后可以用正式代码，仅供参考，如有不当之处请多多指教！
![[特殊字符] Next.js Turbo 模式不支持 @svgr/webpack 的原因与解决方案](http://pic.xiahunao.cn/nshx/[特殊字符] Next.js Turbo 模式不支持 @svgr/webpack 的原因与解决方案)