@[toc]

> **看了很多相关博客,但是都没有本地客户端和服务器后端的完整代码示例,有的也只说了如何流式获取后端结果,基本没有讲两端如何同时实现流式输入输出,特此整理总结,给大家交流学习和使用!**

## 本地客户端

- `requests.post`得到流式回复的重要参数:

- `stream`:需要设置为True;

- `response.iter_content`:使用该函数接收返回的流式数据。

```python

import requests

import time

import json

def generate_stream_data():

# 假设这是要发送的文本列表

is_end = False

lines = [“Hello”, “world”, “this”, “is”, “a”, “stream”, “of”, “text”]

for line in lines:

print(line)

if lines.index(line) == len(lines) - 1:

is_end = True

yield json.dumps({‘line’: line, ‘is_end’: is_end}) + ‘\n’

time.sleep(0.5)

# 模拟数据处理时间

def get_stream_response(response):

# 流式接收response

rec_data_list = []

temp_data = ‘’

for chunk in response.iter_content(chunk_size=1):

temp_data += chunk.decode(‘utf-8’)

if temp_data.endswith(‘\n’):

temp_json = json.loads(temp_data)

rec_data_list.append(temp_json)

print(temp_data)

temp_data = ‘’

if temp_json[‘is_end’]:

break

print(rec_data_list)

print(“----------------------------”)

print(temp_data)

return rec_data_list

def stream_upload(url):

# 流式接收response

response = requests.post(url, data=generate_stream_data(), stream=True)

final_response = get_stream_response(response)

return final_response

url = ‘http://127.0.0.1:5000/stream

response = stream_upload(url)

```

## Flask服务器后端

- `flask.request`流式获取数据::

- 使用`request.stream.read`读取数据,而不是`get_data()`等一次性函数。

```python

from flask import Flask, Response, request

import time

import json

import requests

app = Flask(__name__)

def process_stream_data(stream_data):

# 假设这是要发送的数据

print(“开始生成新的数据流”)

is_end = False

print(stream_data)

for idx, line in enumerate(stream_data):

if idx == len(stream_data)-1:

is_end = True

print(line)

yield json.dumps(line)+“\n”

time.sleep(0.5)

# 模拟数据处理时间

def get_stream_request(chunk_size=1):

req_data_list = []

temp_data = ‘’

while True:

chunk = request.stream.read(chunk_size)

temp_data += chunk.decode(‘utf-8’)

if temp_data.endswith(‘\n’):

temp_json = json.loads(temp_data)

req_data_list.append(temp_json)

print(temp_data)

temp_data = ‘’

if temp_json[‘is_end’]:

return req_data_list

@app.route(‘/stream’, methods=[‘POST’])

def stream_text():

data = get_stream_request()

print(“----------------------------”)

return Response(process_stream_data(data))

if __name__ == “__main__”:

app.run(host=‘0.0.0.0’, port=5000, debug=True)

```

## 客户端/服务器端流式接收[打字机]效果

以上就是“Flask——基于python完整实现客户端和服务器后端流式请求及响应(适用ChatGPT等语言模型后端服务场景)”的全部内容,希望对你有所帮助。

关于Python技术储备

学好 Python 不论是就业还是做副业赚钱都不错,但要学会 Python 还是要有一个学习规划。最后大家分享一份全套的 Python 学习资料,给那些想学习 Python 的小伙伴们一点帮助!

一、Python所有方向的学习路线

Python所有方向的技术点做的整理,形成各个领域的知识点汇总,它的用处就在于,你可以按照上面的知识点去找对应的学习资源,保证自己学得较为全面。

在这里插入图片描述

二、Python必备开发工具

img

三、Python视频合集

观看零基础学习视频,看视频学习是最快捷也是最有效果的方式,跟着视频中老师的思路,从基础到深入,还是很容易入门的。

img

四、实战案例

光学理论是没用的,要学会跟着一起敲,要动手实操,才能将自己的所学运用到实际当中去,这时候可以搞点实战案例来学习。

img

五、Python练习题

检查学习结果。

img

六、面试资料

我们学习Python必然是为了找到高薪的工作,下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料,并且有阿里大佬给出了权威的解答,刷完这一套面试资料相信大家都能找到满意的工作。

img

最后祝大家天天进步!!

上面这份完整版的Python全套学习资料已经上传至CSDN官方,朋友如果需要可以直接微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐